jacob-mathison/query-orchestration
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Merged in bugfix/livetests (pull request #71)

Browse Source 
Live Tests

* testfixes

* lint

* vendor
This commit is contained in:
Michael McGuinness
2025-02-17 17:25:33 +00:00
parent e357555725
commit 0ea544926b
6 changed files with 139 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
database/migrations/00000000000102_job_views.up.sql
+1 -1
View File
@@ -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
deployments/compose.local.yaml
+15 -9
View File
@@ -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:
deployments/prometheus.allmetrics.yaml
+6 -6
View File
@@ -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:
devbox.lock
+31
View File
@@ -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",
internal/job/sync/sync.go
+1 -1
View File
@@ -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
}
}
internal/job/sync/sync_test.go
+85 -55
View File
@@ -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)
})
}