From 0ea544926b705c7211c66a8ff27f0533b4fdd4cc Mon Sep 17 00:00:00 2001 From: Michael McGuinness Date: Mon, 17 Feb 2025 17:25:33 +0000 Subject: [PATCH] Merged in bugfix/livetests (pull request #71) Live Tests * testfixes * lint * vendor --- .../00000000000102_job_views.up.sql | 2 +- deployments/compose.local.yaml | 24 +-- deployments/prometheus.allmetrics.yaml | 12 +- devbox.lock | 31 ++++ internal/job/sync/sync.go | 2 +- internal/job/sync/sync_test.go | 140 +++++++++++------- 6 files changed, 139 insertions(+), 72 deletions(-) diff --git a/database/migrations/00000000000102_job_views.up.sql b/database/migrations/00000000000102_job_views.up.sql index c027b028..1deb774a 100644 --- a/database/migrations/00000000000102_job_views.up.sql +++ b/database/migrations/00000000000102_job_views.up.sql @@ -7,7 +7,7 @@ RETURNS TABLE (id uuid, totalCount BIGINT) AS $$ DECLARE totalCount BIGINT := 0; BEGIN - SELECT COUNT(*)::BIGINT INTO totalCount FROM documents WHERE jobId = _jobId LIMIT 1; + SELECT COUNT(*)::BIGINT INTO totalCount FROM documents WHERE jobId = _jobId; RETURN QUERY SELECT d.id, totalCount diff --git a/deployments/compose.local.yaml b/deployments/compose.local.yaml index 0baf4142..8697e719 100644 --- a/deployments/compose.local.yaml +++ b/deployments/compose.local.yaml @@ -34,8 +34,14 @@ services: image: queryorchestration:latest command: ["./docSyncRunner"] depends_on: - - db - - localstack + localstack: + condition: service_healthy + db: + condition: service_healthy + ports: + - "8082:8080" + expose: + - 8080 environment: LOG_LEVEL: DEBUG QUEUE_URL: ${DOCUMENT_SYNC_URL} @@ -64,7 +70,7 @@ services: db: condition: service_healthy ports: - - "8082:8080" + - "8083:8080" expose: - 8080 environment: @@ -95,7 +101,7 @@ services: db: condition: service_healthy ports: - - "8083:8080" + - "8084:8080" expose: - 8080 environment: @@ -126,7 +132,7 @@ services: db: condition: service_healthy ports: - - "8084:8080" + - "8085:8080" expose: - 8080 environment: @@ -157,7 +163,7 @@ services: db: condition: service_healthy ports: - - "8085:8080" + - "8086:8080" expose: - 8080 environment: @@ -188,7 +194,7 @@ services: db: condition: service_healthy ports: - - "8086:8080" + - "8080:8080" expose: - 8080 environment: @@ -219,7 +225,7 @@ services: db: condition: service_healthy ports: - - "8087:8080" + - "8088:8080" expose: - 8080 environment: @@ -250,7 +256,7 @@ services: db: condition: service_healthy ports: - - "8088:8080" + - "8089:8080" expose: - 8080 environment: diff --git a/deployments/prometheus.allmetrics.yaml b/deployments/prometheus.allmetrics.yaml index 38fd3590..37424bc0 100644 --- a/deployments/prometheus.allmetrics.yaml +++ b/deployments/prometheus.allmetrics.yaml @@ -4,6 +4,12 @@ global: scrape_interval: 10s scrape_configs: + - job_name: 'query_service' + static_configs: + - targets: + - 'query_service:8080' + metrics_path: '/metrics' + - job_name: 'doc_init_runner' static_configs: - targets: @@ -34,12 +40,6 @@ scrape_configs: - 'query_runner:8080' metrics_path: '/metrics' - - job_name: 'query_service' - static_configs: - - targets: - - 'query_service:8080' - metrics_path: '/metrics' - - job_name: 'job_sync_runner' static_configs: - targets: diff --git a/devbox.lock b/devbox.lock index 7c636870..4a644d45 100644 --- a/devbox.lock +++ b/devbox.lock @@ -65,6 +65,37 @@ } } }, + "github:NixOS/nixpkgs/nixpkgs-unstable": { + "resolved": "github:NixOS/nixpkgs/b1b43d32be000928cc71250ed77f4a0a5f2bc23a?lastModified=1739698114&narHash=sha256-8S9n69Dnpg8DhfFlP0YvMGmSOY2X4kImGSPWXYNpaHM%3D" + }, + "glibcLocales@latest": { + "last_modified": "2025-02-07T11:26:36Z", + "resolved": "github:NixOS/nixpkgs/d98abf5cf5914e5e4e9d57205e3af55ca90ffc1d#glibcLocales", + "source": "devbox-search", + "version": "2.40-66", + "systems": { + "aarch64-linux": { + "outputs": [ + { + "name": "out", + "path": "/nix/store/msl6h7i0mgzayw4icb58hccfkh9672bl-glibc-locales-2.40-66", + "default": true + } + ], + "store_path": "/nix/store/msl6h7i0mgzayw4icb58hccfkh9672bl-glibc-locales-2.40-66" + }, + "x86_64-linux": { + "outputs": [ + { + "name": "out", + "path": "/nix/store/p1fn0y9zbq7fng8cn49cvinqqh47slax-glibc-locales-2.40-66", + "default": true + } + ], + "store_path": "/nix/store/p1fn0y9zbq7fng8cn49cvinqqh47slax-glibc-locales-2.40-66" + } + } + }, "go-migrate@4.18.1": { "last_modified": "2024-12-23T21:10:33Z", "resolved": "github:NixOS/nixpkgs/de1864217bfa9b5845f465e771e0ecb48b30e02d#go-migrate", diff --git a/internal/job/sync/sync.go b/internal/job/sync/sync.go index fe824e68..9a512fc5 100644 --- a/internal/job/sync/sync.go +++ b/internal/job/sync/sync.go @@ -45,7 +45,7 @@ func (s *Service) Sync(ctx context.Context, id uuid.UUID) error { } offset += s.batchSize - if ids == nil || int64(offset) >= *ids[0].Totalcount { + if len(ids) == 0 || ids[0].Totalcount == nil || int64(offset) >= *ids[0].Totalcount { hasMore = false } } diff --git a/internal/job/sync/sync_test.go b/internal/job/sync/sync_test.go index 2e5ad64c..94430ede 100644 --- a/internal/job/sync/sync_test.go +++ b/internal/job/sync/sync_test.go @@ -24,65 +24,95 @@ type JobSyncConfig struct { } func TestTrigger(t *testing.T) { - ctx := context.Background() + t.Run("two batches", func(t *testing.T) { + ctx := context.Background() - pool, err := pgxmock.NewPool() - if err != nil { - t.Fatalf("failed to open pgxmock database: %v", err) - } + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } - cfg := &JobSyncConfig{} - cfg.DBPool = pool - cfg.DBQueries = repository.New(pool) - mockSQS := queuemock.NewMockSQSClient(t) - cfg.QueueClient = mockSQS - cfg.DocumentSyncURL = "/i/am/here" + cfg := &JobSyncConfig{} + cfg.DBPool = pool + cfg.DBQueries = repository.New(pool) + mockSQS := queuemock.NewMockSQSClient(t) + cfg.QueueClient = mockSQS + cfg.DocumentSyncURL = "/i/am/here" - svc := New(cfg) + svc := New(cfg) - jobId := uuid.New() - docs := []document.Document{ - { - ID: uuid.New(), - JobID: jobId, - }, - { - ID: uuid.New(), - JobID: jobId, - }, - } + jobId := uuid.New() + docs := []document.Document{ + { + ID: uuid.New(), + JobID: jobId, + }, + { + ID: uuid.New(), + JobID: jobId, + }, + } - svc.batchSize = 1 - total := int64(2) - pool.ExpectQuery("name: ListJobDocumentIDsBatch :many").WithArgs(database.MustToDBUUID(jobId), svc.batchSize, int32(0)). - WillReturnRows( - pgxmock.NewRows([]string{"id", "totalCount"}). - AddRow(database.MustToDBUUID(docs[0].ID), &total), - ) - mockSQS.EXPECT(). - SendMessage( - mock.Anything, - mock.MatchedBy(func(in *sqs.SendMessageInput) bool { - return *in.QueueUrl == cfg.DocumentSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", docs[0].ID.String()) - }), - mock.Anything, - ). - Return(&sqs.SendMessageOutput{}, nil) - pool.ExpectQuery("name: ListJobDocumentIDsBatch :many").WithArgs(database.MustToDBUUID(jobId), svc.batchSize, int32(1)). - WillReturnRows( - pgxmock.NewRows([]string{"id", "totalCount"}). - AddRow(database.MustToDBUUID(docs[1].ID), &total), - ) - mockSQS.EXPECT(). - SendMessage( - mock.Anything, - mock.MatchedBy(func(in *sqs.SendMessageInput) bool { - return *in.QueueUrl == cfg.DocumentSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", docs[1].ID.String()) - }), - mock.Anything, - ). - Return(&sqs.SendMessageOutput{}, nil) + svc.batchSize = 1 + total := int64(2) + pool.ExpectQuery("name: ListJobDocumentIDsBatch :many").WithArgs(database.MustToDBUUID(jobId), svc.batchSize, int32(0)). + WillReturnRows( + pgxmock.NewRows([]string{"id", "totalCount"}). + AddRow(database.MustToDBUUID(docs[0].ID), &total), + ) + mockSQS.EXPECT(). + SendMessage( + mock.Anything, + mock.MatchedBy(func(in *sqs.SendMessageInput) bool { + return *in.QueueUrl == cfg.DocumentSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", docs[0].ID.String()) + }), + mock.Anything, + ). + Return(&sqs.SendMessageOutput{}, nil) + pool.ExpectQuery("name: ListJobDocumentIDsBatch :many").WithArgs(database.MustToDBUUID(jobId), svc.batchSize, int32(1)). + WillReturnRows( + pgxmock.NewRows([]string{"id", "totalCount"}). + AddRow(database.MustToDBUUID(docs[1].ID), &total), + ) + mockSQS.EXPECT(). + SendMessage( + mock.Anything, + mock.MatchedBy(func(in *sqs.SendMessageInput) bool { + return *in.QueueUrl == cfg.DocumentSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", docs[1].ID.String()) + }), + mock.Anything, + ). + Return(&sqs.SendMessageOutput{}, nil) - err = svc.Sync(ctx, jobId) - assert.NoError(t, err) + err = svc.Sync(ctx, jobId) + assert.NoError(t, err) + }) + t.Run("empty batch", func(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + + cfg := &JobSyncConfig{} + cfg.DBPool = pool + cfg.DBQueries = repository.New(pool) + mockSQS := queuemock.NewMockSQSClient(t) + cfg.QueueClient = mockSQS + cfg.DocumentSyncURL = "/i/am/here" + + svc := New(cfg) + + jobId := uuid.New() + + svc.batchSize = 1 + pool.ExpectQuery("name: ListJobDocumentIDsBatch :many").WithArgs(database.MustToDBUUID(jobId), svc.batchSize, int32(0)). + WillReturnRows( + pgxmock.NewRows([]string{"id", "totalCount"}), + ) + + err = svc.Sync(ctx, jobId) + assert.NoError(t, err) + }) }