package docsyncrunner_test import ( "fmt" "regexp" "testing" "time" docsyncrunner "queryorchestration/api/docSyncRunner" "queryorchestration/internal/client" "queryorchestration/internal/database/repository" "queryorchestration/internal/document" "queryorchestration/internal/document/batch/outcome" documentsync "queryorchestration/internal/document/sync" "queryorchestration/internal/server/runner" "queryorchestration/internal/serviceconfig/objectstore" "queryorchestration/internal/serviceconfig/queue" "queryorchestration/internal/serviceconfig/queue/documentclean" "queryorchestration/internal/test" "github.com/google/uuid" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) type DocSyncConfig struct { runner.BaseConfig[docsyncrunner.Body] documentclean.DocCleanConfig queue.QueueConfig objectstore.ObjectStoreConfig } func TestDocSyncRunner(t *testing.T) { cfg := &DocSyncConfig{} test.CreateDB(t, cfg) acfg := test.CreateAWSContainer(t, cfg) test.SetQueueClient(t, t.Context(), cfg, acfg.ExternalEndpoint) cfg.DocumentCleanURL = test.CreateQueue(t, cfg, test.DocCleanRunnerName) runner := docsyncrunner.New(&docsyncrunner.Services{ Document: documentsync.New(cfg, &documentsync.Services{ Document: document.New(cfg), Client: client.New(cfg), }), Outcome: outcome.New(cfg.GetDBQueries()), }) err := cfg.GetDBQueries().CreateClient(t.Context(), &repository.CreateClientParams{ Clientid: "clientid", Name: "client_name", }) require.NoError(t, err) err = cfg.GetDBQueries().AddClientCanSync(t.Context(), &repository.AddClientCanSyncParams{ Clientid: "clientid", Cansync: true, }) require.NoError(t, err) docId, err := cfg.GetDBQueries().CreateDocument(t.Context(), &repository.CreateDocumentParams{ Clientid: "clientid", Hash: "hash", BatchID: nil, }) require.NoError(t, err) part := uint16(1) filetype := "pdf" key := objectstore.BucketKey{ CreatedAt: time.Now().UTC(), ClientID: "clientid", EntityID: uuid.New(), Location: objectstore.Import, Part: &part, FileType: &filetype, } err = cfg.GetDBQueries().AddDocumentEntry(t.Context(), &repository.AddDocumentEntryParams{ Documentid: docId, Bucket: "bucket", Key: key.String(), }) require.NoError(t, err) doc := docsyncrunner.Body{ DocumentID: docId, } assert.True(t, runner.Process(t.Context(), doc)) test.AssertMessageBody(t, cfg, cfg.GetDocumentCleanURL(), regexp.MustCompile(fmt.Sprintf("{\"id\":\"%s\"}", docId))) } // setupDocSyncTest creates shared infrastructure and returns the config. func setupDocSyncTest(t *testing.T) *DocSyncConfig { t.Helper() cfg := &DocSyncConfig{} test.CreateDB(t, cfg) acfg := test.CreateAWSContainer(t, cfg) test.SetQueueClient(t, t.Context(), cfg, acfg.ExternalEndpoint) cfg.DocumentCleanURL = test.CreateQueue(t, cfg, test.DocCleanRunnerName) return cfg } // setupBatchSyncTest creates a client, batch, document, outcome row, and // returns all IDs needed for a sync runner test. func setupBatchSyncTest(t *testing.T, cfg *DocSyncConfig, canSync bool) (uuid.UUID, uuid.UUID) { t.Helper() ctx := t.Context() clientID := fmt.Sprintf("sync_%s_%s", fmt.Sprintf("%v", canSync), uuid.New().String()[:8]) // Create client and set sync flag err := cfg.GetDBQueries().CreateClient(ctx, &repository.CreateClientParams{ Clientid: clientID, Name: "Test " + clientID, }) require.NoError(t, err) err = cfg.GetDBQueries().AddClientCanSync(ctx, &repository.AddClientCanSyncParams{ Clientid: clientID, Cansync: canSync, }) require.NoError(t, err) // Create batch (client must exist first due to FK) batchID, err := cfg.GetDBQueries().CreateBatchUpload(ctx, &repository.CreateBatchUploadParams{ ClientID: clientID, OriginalFilename: "test.zip", TotalDocuments: 1, }) require.NoError(t, err) // Create document linked to batch docID, err := cfg.GetDBQueries().CreateDocument(ctx, &repository.CreateDocumentParams{ Clientid: clientID, Hash: "hash_" + uuid.New().String()[:8], BatchID: &batchID, }) require.NoError(t, err) part := uint16(1) filetype := "pdf" key := objectstore.BucketKey{ CreatedAt: time.Now().UTC(), ClientID: clientID, EntityID: uuid.New(), Location: objectstore.Import, Part: &part, FileType: &filetype, } err = cfg.GetDBQueries().AddDocumentEntry(ctx, &repository.AddDocumentEntryParams{ Documentid: docID, Bucket: "bucket", Key: key.String(), }) require.NoError(t, err) // Insert init_complete outcome (the state before sync) err = cfg.GetDBQueries().InsertBatchDocumentOutcome(ctx, &repository.InsertBatchDocumentOutcomeParams{ BatchID: batchID, Filename: "test.pdf", Column3: repository.BatchOutcomeStatusInitComplete, DocumentID: &docID, }) require.NoError(t, err) return batchID, docID } // TestDocSyncRunner_ReportsSyncComplete verifies that processing a batch document // with can_sync=true updates the outcome to sync_complete. func TestDocSyncRunner_ReportsSyncComplete(t *testing.T) { cfg := setupDocSyncTest(t) ctx := t.Context() batchID, docID := setupBatchSyncTest(t, cfg, true) syncRunner := docsyncrunner.New(&docsyncrunner.Services{ Document: documentsync.New(cfg, &documentsync.Services{ Document: document.New(cfg), Client: client.New(cfg), }), Outcome: outcome.New(cfg.GetDBQueries()), }) assert.True(t, syncRunner.Process(ctx, docsyncrunner.Body{DocumentID: docID})) // Verify outcome updated to sync_complete rows, err := cfg.GetDBQueries().ListBatchDocumentOutcomes(ctx, batchID) require.NoError(t, err) require.Len(t, rows, 1) assert.Equal(t, repository.BatchOutcomeStatusSyncComplete, rows[0].Outcome) } // TestDocSyncRunner_ReportsSyncSkipped verifies that processing a batch document // with can_sync=false updates the outcome to sync_skipped. func TestDocSyncRunner_ReportsSyncSkipped(t *testing.T) { cfg := setupDocSyncTest(t) ctx := t.Context() batchID, docID := setupBatchSyncTest(t, cfg, false) syncRunner := docsyncrunner.New(&docsyncrunner.Services{ Document: documentsync.New(cfg, &documentsync.Services{ Document: document.New(cfg), Client: client.New(cfg), }), Outcome: outcome.New(cfg.GetDBQueries()), }) assert.True(t, syncRunner.Process(ctx, docsyncrunner.Body{DocumentID: docID})) // Verify outcome updated to sync_skipped rows, err := cfg.GetDBQueries().ListBatchDocumentOutcomes(ctx, batchID) require.NoError(t, err) require.Len(t, rows, 1) assert.Equal(t, repository.BatchOutcomeStatusSyncSkipped, rows[0].Outcome) }