From c1e9d930377f14d3b2e6053356afe877f8df975d Mon Sep 17 00:00:00 2001 From: Michael McGuinness Date: Wed, 12 Feb 2025 19:00:25 +0000 Subject: [PATCH] Merged in feature/jobsync (pull request #63) Document Sync and Job Sync * docsyncfunc * startedQueryWork * morefixes * jobsyncsvcstart * save * jobsynctext * save * save * docsync * shorttest * jobsync * jobsyncupdateoncollectorupdate * passlivetest * collectortest * lint --- README.md | 4 +- api/docCleanRunner/runner_test.go | 3 +- api/docInitRunner/runner_test.go | 53 ++-- api/docSyncRunner/runner.go | 54 ++++ api/docSyncRunner/runner_test.go | 85 ++++++ api/docTextRunner/runner_test.go | 8 +- api/jobSyncRunner/runner.go | 54 ++++ api/jobSyncRunner/runner_test.go | 82 ++++++ api/queryService/controllers.go | 14 +- api/queryService/job_test.go | 6 +- api/queryService/jobcollector.go | 4 +- api/queryService/jobcollector_test.go | 45 ++- api/queryService/query_test.go | 11 +- cmd/docCleanRunner/main.go | 7 +- cmd/docInitRunner/main.go | 21 +- cmd/docSyncRunner/main.go | 61 ++++ cmd/docTextRunner/main.go | 7 +- cmd/jobSyncRunner/main.go | 42 +++ cmd/queryService/main.go | 46 ++- .../00000000000102_job_views.down.sql | 1 + .../00000000000102_job_views.up.sql | 20 ++ database/queries/job.sql | 5 +- deployments/compose.local.yaml | 49 ++++ deployments/compose.test.yaml | 2 +- devbox.json | 6 +- internal/database/repository/job.sql.go | 38 +++ internal/database/repository/job_test.go | 106 +++++++ internal/database/repository/result_test.go | 4 - internal/document/clean/clean.go | 2 +- internal/document/clean/clean_test.go | 3 +- internal/document/clean/create_test.go | 5 +- internal/document/clean/service.go | 6 +- internal/document/clean/service_test.go | 2 - internal/document/clean/version/service.go | 15 + .../document/clean/version/service_test.go | 13 + internal/document/clean/version/version.go | 20 ++ .../document/clean/version/version_test.go | 26 ++ internal/document/init/create.go | 44 +-- internal/document/init/create_test.go | 125 +++------ internal/document/init/service.go | 13 +- internal/document/init/service_test.go | 17 +- internal/document/sync/service.go | 30 ++ internal/document/sync/service_test.go | 20 ++ internal/document/sync/sync.go | 41 +++ internal/document/sync/sync_test.go | 80 ++++++ internal/document/text/create_test.go | 5 +- internal/document/text/extract.go | 2 +- internal/document/text/extract_test.go | 3 +- internal/document/text/service.go | 6 +- internal/document/text/service_test.go | 6 +- internal/document/text/version/service.go | 15 + .../document/text/version/service_test.go | 13 + internal/document/text/version/version.go | 20 ++ .../document/text/version/version_test.go | 26 ++ internal/document/version.go | 35 --- internal/document/version_test.go | 43 --- internal/job/collector/create.go | 8 +- internal/job/collector/createprivate_test.go | 4 +- internal/job/collector/service.go | 6 +- internal/job/collector/update/service.go | 32 +++ internal/job/collector/update/service_test.go | 24 ++ internal/job/collector/{ => update}/update.go | 59 ++-- .../job/collector/{ => update}/update_test.go | 46 ++- .../{ => update}/updateprivate_test.go | 265 +++++++++++++----- internal/job/sync/service.go | 23 ++ internal/job/sync/service_test.go | 13 + internal/job/sync/sync.go | 51 ++++ internal/job/sync/sync_test.go | 88 ++++++ internal/query/test/test_test.go | 9 +- internal/server/runner/listener.go | 5 - internal/server/server.go | 5 + .../queue/documentinit/config.go | 13 + .../queue/documentinit/config_test.go | 20 ++ .../queue/documentsync/config.go | 13 + .../queue/documentsync/config_test.go | 20 ++ .../serviceconfig/queue/jobsync/config.go | 13 + .../queue/jobsync/config_test.go | 20 ++ internal/test/ecosystem.go | 26 +- internal/test/ecosystem_test.go | 10 +- internal/test/runner.go | 4 + internal/test/service_test.go | 3 + scripts/Taskfile.yml | 2 +- scripts/database.yml | 6 +- scripts/local-deployments.yml | 2 + scripts/tests.yml | 15 +- sqlc.yml | 2 +- test/process_test.go | 19 +- test/queryService/client_test.go | 5 +- test/queryService/exportservice_test.go | 5 +- test/queryService/job_test.go | 5 +- test/queryService/jobcollectorservice_test.go | 80 ++++-- test/queryService/openapi_test.go | 5 +- test/queryService/queryservice_test.go | 5 +- test/queryService/testquery_test.go | 5 +- 94 files changed, 1898 insertions(+), 512 deletions(-) create mode 100644 api/docSyncRunner/runner.go create mode 100644 api/docSyncRunner/runner_test.go create mode 100644 api/jobSyncRunner/runner.go create mode 100644 api/jobSyncRunner/runner_test.go create mode 100644 cmd/docSyncRunner/main.go create mode 100644 cmd/jobSyncRunner/main.go create mode 100644 database/migrations/00000000000102_job_views.down.sql create mode 100644 database/migrations/00000000000102_job_views.up.sql create mode 100644 internal/document/clean/version/service.go create mode 100644 internal/document/clean/version/service_test.go create mode 100644 internal/document/clean/version/version.go create mode 100644 internal/document/clean/version/version_test.go create mode 100644 internal/document/sync/service.go create mode 100644 internal/document/sync/service_test.go create mode 100644 internal/document/sync/sync.go create mode 100644 internal/document/sync/sync_test.go create mode 100644 internal/document/text/version/service.go create mode 100644 internal/document/text/version/service_test.go create mode 100644 internal/document/text/version/version.go create mode 100644 internal/document/text/version/version_test.go delete mode 100644 internal/document/version.go delete mode 100644 internal/document/version_test.go create mode 100644 internal/job/collector/update/service.go create mode 100644 internal/job/collector/update/service_test.go rename internal/job/collector/{ => update}/update.go (82%) rename internal/job/collector/{ => update}/update_test.go (50%) rename internal/job/collector/{ => update}/updateprivate_test.go (67%) create mode 100644 internal/job/sync/service.go create mode 100644 internal/job/sync/service_test.go create mode 100644 internal/job/sync/sync.go create mode 100644 internal/job/sync/sync_test.go create mode 100644 internal/serviceconfig/queue/documentinit/config.go create mode 100644 internal/serviceconfig/queue/documentinit/config_test.go create mode 100644 internal/serviceconfig/queue/documentsync/config.go create mode 100644 internal/serviceconfig/queue/documentsync/config_test.go create mode 100644 internal/serviceconfig/queue/jobsync/config.go create mode 100644 internal/serviceconfig/queue/jobsync/config_test.go diff --git a/README.md b/README.md index 04ff4e26..7815b9ea 100644 --- a/README.md +++ b/README.md @@ -32,7 +32,7 @@ When using environment variables, there is a hiearchy that will be respected. (F ## Testing -For testing integrations such as queries against a database, or interactions with an SQS queue, this repository employs the use of docker testcontainers. +For testing endtoends such as queries against a database, or interactions with an SQS queue, this repository employs the use of docker testcontainers. `https://testcontainers.com/` @@ -47,7 +47,7 @@ This will: 1. Generate the latest version of the specs. 2. Run the linting commands. 3. Run the unit tests. -4. Run the integration tests. +4. Run the endtoend tests. ## Naming Conventions diff --git a/api/docCleanRunner/runner_test.go b/api/docCleanRunner/runner_test.go index f94535ce..54c53a4e 100644 --- a/api/docCleanRunner/runner_test.go +++ b/api/docCleanRunner/runner_test.go @@ -9,6 +9,7 @@ import ( "queryorchestration/internal/database/repository" "queryorchestration/internal/document" documentclean "queryorchestration/internal/document/clean" + cleanversion "queryorchestration/internal/document/clean/version" "queryorchestration/internal/serviceconfig" "queryorchestration/internal/serviceconfig/objectstore" "queryorchestration/internal/serviceconfig/queue/documenttext" @@ -50,7 +51,7 @@ func TestDocCleanRunner(t *testing.T) { runner := doccleanrunner.New(validator.New(), &doccleanrunner.Services{ Clean: documentclean.New(cfg, &documentclean.Services{ - Document: document.New(cfg), + Version: cleanversion.New(cfg), }), }) assert.NotNil(t, runner) diff --git a/api/docInitRunner/runner_test.go b/api/docInitRunner/runner_test.go index 53fde512..65d677ae 100644 --- a/api/docInitRunner/runner_test.go +++ b/api/docInitRunner/runner_test.go @@ -5,30 +5,28 @@ import ( "encoding/json" "fmt" docinitrunner "queryorchestration/api/docInitRunner" - "queryorchestration/internal/client" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" "queryorchestration/internal/document" documentinit "queryorchestration/internal/document/init" "queryorchestration/internal/job" - "queryorchestration/internal/job/collector" "queryorchestration/internal/serviceconfig" - "queryorchestration/internal/serviceconfig/objectstore" - "queryorchestration/internal/serviceconfig/queue/documentclean" - objectstoremock "queryorchestration/mocks/objectstore" + "queryorchestration/internal/serviceconfig/queue/documentsync" + queuemock "queryorchestration/mocks/queue" "testing" + "github.com/aws/aws-sdk-go-v2/service/sqs" "github.com/aws/aws-sdk-go-v2/service/sqs/types" "github.com/go-playground/validator/v10" "github.com/google/uuid" "github.com/pashagolub/pgxmock/v3" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" ) type DocInitConfig struct { serviceconfig.BaseConfig - documentclean.DocCleanConfig - objectstore.ObjectStoreConfig + documentsync.DocSyncConfig } func TestDocInitRunner(t *testing.T) { @@ -42,16 +40,11 @@ func TestDocInitRunner(t *testing.T) { cfg := &DocInitConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) - mockStore := objectstoremock.NewMockS3Client(t) - cfg.StoreClient = mockStore + mockSQS := queuemock.NewMockSQSClient(t) + cfg.QueueClient = mockSQS runner := docinitrunner.New(validator.New(), &docinitrunner.Services{ - Document: documentinit.New(cfg, &documentinit.Services{ - Job: job.New(cfg, &job.Services{ - Collector: collector.New(cfg, &collector.Services{}), - Client: client.New(cfg), - }), - }), + Document: documentinit.New(cfg), }) assert.NotNil(t, runner) @@ -60,10 +53,11 @@ func TestDocInitRunner(t *testing.T) { ClientID: uuid.New(), } bucketName := "bucketName" - location := fmt.Sprintf("aaa/%s/aaa", j.ID.String()) + location := fmt.Sprintf("%s/%s/aaa", j.ClientID.String(), j.ID.String()) docinfo := document.Document{ - ID: uuid.New(), - Hash: "example_hash", + ID: uuid.New(), + JobID: j.ID, + Hash: "example_hash", } doc := docinitrunner.S3EventNotification{ Records: []docinitrunner.S3EventRecord{ @@ -88,14 +82,6 @@ func TestDocInitRunner(t *testing.T) { Body: &body, } - pool.ExpectQuery("name: GetJob :one").WithArgs(database.MustToDBUUID(j.ID)).WillReturnRows( - pgxmock.NewRows([]string{"id", "clientId", "canSync"}). - AddRow(database.MustToDBUUID(j.ID), database.MustToDBUUID(j.ClientID), j.CanSync), - ) - pool.ExpectQuery("name: GetClient :one").WithArgs(database.MustToDBUUID(j.ClientID)).WillReturnRows( - pgxmock.NewRows([]string{"id", "name", "canSync"}). - AddRow(database.MustToDBUUID(j.ClientID), "client_name", true), - ) pool.ExpectQuery("name: GetDocumentIDByHash :one").WithArgs(docinfo.Hash, database.MustToDBUUID(j.ID)).WillReturnRows( pgxmock.NewRows([]string{"id"}), ) @@ -108,6 +94,21 @@ func TestDocInitRunner(t *testing.T) { pool.ExpectExec("name: AddDocumentEntry :exec").WithArgs(database.MustToDBUUID(docinfo.ID), bucketName, location). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectCommit() + pool.ExpectQuery("name: GetDocument :one").WithArgs(database.MustToDBUUID(docinfo.ID)). + WillReturnRows( + pgxmock.NewRows([]string{"id", "jobId", "hash"}). + AddRow(database.MustToDBUUID(docinfo.ID), database.MustToDBUUID(docinfo.JobID), "example"), + ) + + mockSQS.EXPECT(). + SendMessage( + mock.Anything, + mock.MatchedBy(func(in *sqs.SendMessageInput) bool { + return *in.QueueUrl == cfg.DocumentSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", docinfo.ID.String()) + }), + mock.Anything, + ). + Return(&sqs.SendMessageOutput{}, nil) err = runner.Process(ctx, msg) assert.NoError(t, err) diff --git a/api/docSyncRunner/runner.go b/api/docSyncRunner/runner.go new file mode 100644 index 00000000..0c3da893 --- /dev/null +++ b/api/docSyncRunner/runner.go @@ -0,0 +1,54 @@ +package docsyncrunner + +import ( + "context" + "encoding/json" + documentsync "queryorchestration/internal/document/sync" + + "github.com/go-playground/validator/v10" + "github.com/google/uuid" + + "github.com/aws/aws-sdk-go-v2/service/sqs/types" +) + +const Name = "docSyncRunner" + +type Services struct { + Document *documentsync.Service +} + +type Runner struct { + validator *validator.Validate + svc *Services +} + +func New(validator *validator.Validate, svc *Services) Runner { + return Runner{ + validator: validator, + svc: svc, + } +} + +type Body struct { + ID uuid.UUID `json:"id" validate:"required,uuid"` +} + +func (s Runner) Process(ctx context.Context, req *types.Message) error { + var body Body + err := json.Unmarshal([]byte(*req.Body), &body) + if err != nil { + return err + } + + err = s.validator.Struct(body) + if err != nil { + return err + } + + err = s.svc.Document.Sync(ctx, body.ID) + if err != nil { + return err + } + + return nil +} diff --git a/api/docSyncRunner/runner_test.go b/api/docSyncRunner/runner_test.go new file mode 100644 index 00000000..a9f5a5e3 --- /dev/null +++ b/api/docSyncRunner/runner_test.go @@ -0,0 +1,85 @@ +package docsyncrunner_test + +import ( + "context" + "encoding/json" + docsyncrunner "queryorchestration/api/docSyncRunner" + "queryorchestration/internal/client" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/document" + documentsync "queryorchestration/internal/document/sync" + "queryorchestration/internal/job" + "queryorchestration/internal/serviceconfig" + "queryorchestration/internal/serviceconfig/queue/documentclean" + "testing" + + "github.com/aws/aws-sdk-go-v2/service/sqs/types" + "github.com/go-playground/validator/v10" + "github.com/google/uuid" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +type DocSyncConfig struct { + serviceconfig.BaseConfig + documentclean.DocCleanConfig +} + +func TestDocInitRunner(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + + cfg := &DocSyncConfig{} + cfg.DBPool = pool + cfg.DBQueries = repository.New(pool) + + runner := docsyncrunner.New(validator.New(), &docsyncrunner.Services{ + Document: documentsync.New(cfg, &documentsync.Services{ + Document: document.New(cfg), + Job: job.New(cfg, &job.Services{ + Client: client.New(cfg), + }), + }), + }) + assert.NotNil(t, runner) + + j := job.Job{ + ID: uuid.New(), + } + docinfo := document.Document{ + ID: uuid.New(), + JobID: j.ID, + } + doc := docsyncrunner.Body{ + ID: docinfo.ID, + } + bodyBytes, err := json.Marshal(doc) + assert.NoError(t, err) + body := string(bodyBytes) + msg := &types.Message{ + Body: &body, + } + + pool.ExpectQuery("name: GetDocument :one").WithArgs(database.MustToDBUUID(docinfo.ID)). + WillReturnRows( + pgxmock.NewRows([]string{"id", "jobId", "hash"}). + AddRow(database.MustToDBUUID(docinfo.ID), database.MustToDBUUID(docinfo.JobID), "example"), + ) + pool.ExpectQuery("name: GetJob :one").WithArgs(database.MustToDBUUID(j.ID)).WillReturnRows( + pgxmock.NewRows([]string{"id", "clientId", "canSync"}). + AddRow(database.MustToDBUUID(j.ID), database.MustToDBUUID(j.ClientID), j.CanSync), + ) + pool.ExpectQuery("name: GetClient :one").WithArgs(database.MustToDBUUID(j.ClientID)). + WillReturnRows( + pgxmock.NewRows([]string{"id", "name", "canSync"}). + AddRow(database.MustToDBUUID(j.ClientID), "client_name", true), + ) + + err = runner.Process(ctx, msg) + assert.NoError(t, err) +} diff --git a/api/docTextRunner/runner_test.go b/api/docTextRunner/runner_test.go index e56c065e..caf49d1b 100644 --- a/api/docTextRunner/runner_test.go +++ b/api/docTextRunner/runner_test.go @@ -9,10 +9,9 @@ import ( "queryorchestration/internal/database/repository" "queryorchestration/internal/document" documenttext "queryorchestration/internal/document/text" + textversion "queryorchestration/internal/document/text/version" "queryorchestration/internal/serviceconfig" - "queryorchestration/internal/serviceconfig/objectstore" "queryorchestration/internal/serviceconfig/queue/querysync" - objectstoremock "queryorchestration/mocks/objectstore" queuemock "queryorchestration/mocks/queue" "testing" @@ -28,7 +27,6 @@ import ( type DocTextConfig struct { serviceconfig.BaseConfig querysync.QuerySyncConfig - objectstore.ObjectStoreConfig } func TestDocCleanRunner(t *testing.T) { @@ -42,15 +40,13 @@ func TestDocCleanRunner(t *testing.T) { cfg := &DocTextConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) - mockStore := objectstoremock.NewMockS3Client(t) - cfg.StoreClient = mockStore mockSQS := queuemock.NewMockSQSClient(t) cfg.QueueClient = mockSQS cfg.QuerySyncURL = "/i/am/here" runner := doctextrunner.New(validator.New(), &doctextrunner.Services{ Text: documenttext.New(cfg, &documenttext.Services{ - Document: document.New(cfg), + Version: textversion.New(cfg), }), }) assert.NotNil(t, runner) diff --git a/api/jobSyncRunner/runner.go b/api/jobSyncRunner/runner.go new file mode 100644 index 00000000..7bd69872 --- /dev/null +++ b/api/jobSyncRunner/runner.go @@ -0,0 +1,54 @@ +package jobsyncrunner + +import ( + "context" + "encoding/json" + jobsync "queryorchestration/internal/job/sync" + + "github.com/go-playground/validator/v10" + "github.com/google/uuid" + + "github.com/aws/aws-sdk-go-v2/service/sqs/types" +) + +const Name = "jobSyncRunner" + +type Services struct { + JobSync *jobsync.Service +} + +type Runner struct { + validator *validator.Validate + svc *Services +} + +func New(validator *validator.Validate, svc *Services) Runner { + return Runner{ + validator: validator, + svc: svc, + } +} + +type Body struct { + ID uuid.UUID `json:"id" validate:"required,uuid"` +} + +func (s *Runner) Process(ctx context.Context, req *types.Message) error { + var body Body + err := json.Unmarshal([]byte(*req.Body), &body) + if err != nil { + return err + } + + err = s.validator.Struct(body) + if err != nil { + return err + } + + err = s.svc.JobSync.Sync(ctx, body.ID) + if err != nil { + return err + } + + return nil +} diff --git a/api/jobSyncRunner/runner_test.go b/api/jobSyncRunner/runner_test.go new file mode 100644 index 00000000..dc31ae0d --- /dev/null +++ b/api/jobSyncRunner/runner_test.go @@ -0,0 +1,82 @@ +package jobsyncrunner_test + +import ( + "context" + "encoding/json" + "fmt" + jobsyncrunner "queryorchestration/api/jobSyncRunner" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + jobsync "queryorchestration/internal/job/sync" + "queryorchestration/internal/server/runner" + "queryorchestration/internal/serviceconfig/queue/documentsync" + queuemock "queryorchestration/mocks/queue" + "testing" + + "github.com/aws/aws-sdk-go-v2/service/sqs" + "github.com/aws/aws-sdk-go-v2/service/sqs/types" + "github.com/go-playground/validator/v10" + "github.com/google/uuid" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" +) + +type JobSyncConfig struct { + runner.BaseConfig + documentsync.DocSyncConfig +} + +func TestQueryRunner(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 := jobsync.New(cfg) + + runner := jobsyncrunner.New(validator.New(), &jobsyncrunner.Services{ + JobSync: svc, + }) + assert.NotNil(t, runner) + + bod := jobsyncrunner.Body{ + ID: uuid.New(), + } + bodyBytes, err := json.Marshal(bod) + assert.NoError(t, err) + body := string(bodyBytes) + msg := &types.Message{ + Body: &body, + } + + docId := uuid.New() + + total := int64(1) + pool.ExpectQuery("name: ListJobDocumentIDsBatch :many").WithArgs(database.MustToDBUUID(bod.ID), int32(100), int32(0)). + WillReturnRows( + pgxmock.NewRows([]string{"id", "totalCount"}). + AddRow(database.MustToDBUUID(docId), &total), + ) + mockSQS.EXPECT(). + SendMessage( + mock.Anything, + mock.MatchedBy(func(in *sqs.SendMessageInput) bool { + return *in.QueueUrl == cfg.DocumentSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", docId.String()) + }), + mock.Anything, + ). + Return(&sqs.SendMessageOutput{}, nil) + + err = runner.Process(ctx, msg) + assert.NoError(t, err) +} diff --git a/api/queryService/controllers.go b/api/queryService/controllers.go index 9af89044..370549c9 100644 --- a/api/queryService/controllers.go +++ b/api/queryService/controllers.go @@ -5,6 +5,7 @@ import ( "queryorchestration/internal/export" "queryorchestration/internal/job" "queryorchestration/internal/job/collector" + collectorupdate "queryorchestration/internal/job/collector/update" "queryorchestration/internal/query" querytest "queryorchestration/internal/query/test" @@ -14,12 +15,13 @@ import ( const Name = "queryService" type Services struct { - Export *export.Service - Collector *collector.Service - Query *query.Service - Client *client.Service - Job *job.Service - QueryTest *querytest.Service + Export *export.Service + Collector *collector.Service + CollectorUpdate *collectorupdate.Service + Query *query.Service + Client *client.Service + Job *job.Service + QueryTest *querytest.Service } type Controllers struct { diff --git a/api/queryService/job_test.go b/api/queryService/job_test.go index 5b320b76..5b8475b5 100644 --- a/api/queryService/job_test.go +++ b/api/queryService/job_test.go @@ -9,7 +9,8 @@ import ( "queryorchestration/internal/client" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" - "queryorchestration/internal/document" + cleanversion "queryorchestration/internal/document/clean/version" + textversion "queryorchestration/internal/document/text/version" "queryorchestration/internal/job" "queryorchestration/internal/job/collector" "queryorchestration/internal/serviceconfig" @@ -37,7 +38,8 @@ func TestCreateJob(t *testing.T) { cons := queryservice.NewControllers(validator.New(), &queryservice.Services{ Job: job.New(cfg, &job.Services{ Collector: collector.New(cfg, &collector.Services{ - Document: document.New(cfg), + CleanVersion: cleanversion.New(cfg), + TextVersion: textversion.New(cfg), }), }), }) diff --git a/api/queryService/jobcollector.go b/api/queryService/jobcollector.go index 13f0431e..b6336c43 100644 --- a/api/queryService/jobcollector.go +++ b/api/queryService/jobcollector.go @@ -3,7 +3,7 @@ package queryservice import ( "fmt" "net/http" - "queryorchestration/internal/job/collector" + collectorupdate "queryorchestration/internal/job/collector/update" "github.com/google/uuid" "github.com/labstack/echo/v4" @@ -51,7 +51,7 @@ func (s *Controllers) UpdateJobCollectorByJobId(ctx echo.Context, jobId types.UU fields = &fs } - err := s.svc.Collector.UpdateByJobId(ctx.Request().Context(), &collector.UpdateParams{ + err := s.svc.CollectorUpdate.UpdateByJobId(ctx.Request().Context(), &collectorupdate.UpdateParams{ JobID: jobId, ActiveVersion: req.ActiveVersion, MinCleanVersion: req.MinimumCleanerVersion, diff --git a/api/queryService/jobcollector_test.go b/api/queryService/jobcollector_test.go index c1841df3..8057d215 100644 --- a/api/queryService/jobcollector_test.go +++ b/api/queryService/jobcollector_test.go @@ -2,17 +2,23 @@ package queryservice_test import ( "encoding/json" + "fmt" "net/http" "net/http/httptest" queryservice "queryorchestration/api/queryService" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" - "queryorchestration/internal/document" + cleanversion "queryorchestration/internal/document/clean/version" + textversion "queryorchestration/internal/document/text/version" "queryorchestration/internal/job/collector" + collectorupdate "queryorchestration/internal/job/collector/update" "queryorchestration/internal/serviceconfig" + "queryorchestration/internal/serviceconfig/queue/jobsync" + queuemock "queryorchestration/mocks/queue" "strings" "testing" + "github.com/aws/aws-sdk-go-v2/service/sqs" "github.com/go-playground/validator/v10" "github.com/google/uuid" "github.com/jackc/pgx/v5" @@ -20,6 +26,7 @@ import ( "github.com/labstack/echo/v4" "github.com/pashagolub/pgxmock/v3" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" ) func TestUpdateJobCollector(t *testing.T) { @@ -27,9 +34,14 @@ func TestUpdateJobCollector(t *testing.T) { if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - cfg := &serviceconfig.BaseConfig{} + cfg := &struct { + serviceconfig.BaseConfig + jobsync.JobSyncConfig + }{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) + mockSQS := queuemock.NewMockSQSClient(t) + cfg.QueueClient = mockSQS current := collector.Collector{ ID: uuid.New(), @@ -39,8 +51,10 @@ func TestUpdateJobCollector(t *testing.T) { } av := int32(2) + cv := int32(1) body := queryservice.JobCollectorUpdate{ - ActiveVersion: &av, + ActiveVersion: &av, + MinimumCleanerVersion: &cv, Fields: &[]queryservice.JobCollectorField{ { Name: "a", @@ -58,8 +72,11 @@ func TestUpdateJobCollector(t *testing.T) { ctx := e.NewContext(req, rec) cons := queryservice.NewControllers(validator.New(), &queryservice.Services{ - Collector: collector.New(cfg, &collector.Services{ - Document: document.New(cfg), + CollectorUpdate: collectorupdate.New(cfg, &collectorupdate.Services{ + Collector: collector.New(cfg, &collector.Services{ + CleanVersion: cleanversion.New(cfg), + TextVersion: textversion.New(cfg), + }), }), }) @@ -76,12 +93,27 @@ func TestUpdateJobCollector(t *testing.T) { AddRow(true), ) pool.ExpectBeginTx(pgx.TxOptions{}) + rv := current.LatestVersion + 1 + pool.ExpectExec("name: RemoveCollectorCodeVersion :exec").WithArgs(&rv, database.MustToDBUUID(current.ID)). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectExec("name: AddCollectorCodeVersion :exec").WithArgs(database.MustToDBUUID(current.ID), int32(5), *body.MinimumCleanerVersion, int32(0)). + WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectExec("name: AddCollectorQuery :exec").WithArgs(database.MustToDBUUID(current.ID), "a", database.MustToDBUUID((*body.Fields)[0].QueryId), int32(5)). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectExec("name: UpdateCollector :exec").WithArgs(int32(5), int32(2), database.MustToDBUUID(current.ID)). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectCommit() + mockSQS.EXPECT(). + SendMessage( + mock.Anything, + mock.MatchedBy(func(in *sqs.SendMessageInput) bool { + return *in.QueueUrl == cfg.JobSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", current.JobID.String()) + }), + mock.Anything, + ). + Return(&sqs.SendMessageOutput{}, nil) + err = cons.UpdateJobCollectorByJobId(ctx, current.JobID) assert.NoError(t, err) assert.Equal(t, http.StatusOK, rec.Code) @@ -103,7 +135,8 @@ func TestGetJobCollectorByJobId(t *testing.T) { ctx := e.NewContext(req, rec) svc := collector.New(cfg, &collector.Services{ - Document: document.New(cfg), + CleanVersion: cleanversion.New(cfg), + TextVersion: textversion.New(cfg), }) cons := queryservice.NewControllers(validator.New(), &queryservice.Services{ Collector: svc, diff --git a/api/queryService/query_test.go b/api/queryService/query_test.go index 3c861e78..3c9ac62b 100644 --- a/api/queryService/query_test.go +++ b/api/queryService/query_test.go @@ -9,6 +9,9 @@ import ( "queryorchestration/internal/database" "queryorchestration/internal/database/repository" "queryorchestration/internal/document" + cleanversion "queryorchestration/internal/document/clean/version" + textversion "queryorchestration/internal/document/text/version" + "queryorchestration/internal/job" "queryorchestration/internal/job/collector" "queryorchestration/internal/query" "queryorchestration/internal/query/result" @@ -208,7 +211,8 @@ func TestTestQuery(t *testing.T) { docsvc := document.New(cfg) col := collector.New(cfg, &collector.Services{ - Document: docsvc, + CleanVersion: cleanversion.New(cfg), + TextVersion: textversion.New(cfg), }) que := query.New(cfg) cons := queryservice.NewControllers(validator.New(), &queryservice.Services{ @@ -223,9 +227,12 @@ func TestTestQuery(t *testing.T) { }), }) + j := job.Job{ + ID: uuid.New(), + } coll := collector.Collector{ ID: uuid.New(), - JobID: uuid.New(), + JobID: j.ID, } doc := document.Document{ ID: uuid.New(), diff --git a/cmd/docCleanRunner/main.go b/cmd/docCleanRunner/main.go index 1fc0da3b..4c80aa7e 100644 --- a/cmd/docCleanRunner/main.go +++ b/cmd/docCleanRunner/main.go @@ -5,10 +5,9 @@ import ( "log/slog" "os" doccleanrunner "queryorchestration/api/docCleanRunner" - "queryorchestration/internal/document" documentclean "queryorchestration/internal/document/clean" + cleanversion "queryorchestration/internal/document/clean/version" "queryorchestration/internal/server/runner" - "queryorchestration/internal/serviceconfig/objectstore" "queryorchestration/internal/serviceconfig/queue/documenttext" _ "github.com/lib/pq" @@ -16,7 +15,6 @@ import ( type DocCleanConfig struct { runner.BaseConfig - objectstore.ObjectStoreConfig documenttext.DocTextConfig } @@ -26,9 +24,8 @@ func main() { cfg := &DocCleanConfig{} cfg.ControllerFunc = func() runner.Controller { - doc := document.New(cfg) clean := documentclean.New(cfg, &documentclean.Services{ - Document: doc, + Version: cleanversion.New(cfg), }) return doccleanrunner.New(cfg.GetValidator(), &doccleanrunner.Services{ diff --git a/cmd/docInitRunner/main.go b/cmd/docInitRunner/main.go index a702de2a..5d501d1c 100644 --- a/cmd/docInitRunner/main.go +++ b/cmd/docInitRunner/main.go @@ -5,20 +5,16 @@ import ( "log/slog" "os" docinitrunner "queryorchestration/api/docInitRunner" - "queryorchestration/internal/client" - "queryorchestration/internal/document" documentinit "queryorchestration/internal/document/init" - "queryorchestration/internal/job" - "queryorchestration/internal/job/collector" "queryorchestration/internal/server/runner" - documentcleanc "queryorchestration/internal/serviceconfig/queue/documentclean" + documentsyncc "queryorchestration/internal/serviceconfig/queue/documentsync" _ "github.com/lib/pq" ) type DocInitConfig struct { runner.BaseConfig - documentcleanc.DocCleanConfig + documentsyncc.DocSyncConfig } func main() { @@ -27,18 +23,7 @@ func main() { cfg := &DocInitConfig{} cfg.ControllerFunc = func() runner.Controller { - cli := client.New(cfg) - doc := document.New(cfg) - col := collector.New(cfg, &collector.Services{ - Document: doc, - }) - j := job.New(cfg, &job.Services{ - Collector: col, - Client: cli, - }) - docinit := documentinit.New(cfg, &documentinit.Services{ - Job: j, - }) + docinit := documentinit.New(cfg) return docinitrunner.New(cfg.GetValidator(), &docinitrunner.Services{ Document: docinit, diff --git a/cmd/docSyncRunner/main.go b/cmd/docSyncRunner/main.go new file mode 100644 index 00000000..63043836 --- /dev/null +++ b/cmd/docSyncRunner/main.go @@ -0,0 +1,61 @@ +package main + +import ( + "context" + "log/slog" + "os" + docsyncrunner "queryorchestration/api/docSyncRunner" + "queryorchestration/internal/client" + "queryorchestration/internal/document" + cleanversion "queryorchestration/internal/document/clean/version" + documentsync "queryorchestration/internal/document/sync" + textversion "queryorchestration/internal/document/text/version" + "queryorchestration/internal/job" + "queryorchestration/internal/job/collector" + "queryorchestration/internal/server/runner" + documentcleanc "queryorchestration/internal/serviceconfig/queue/documentclean" + + _ "github.com/lib/pq" +) + +type DocSyncConfig struct { + runner.BaseConfig + documentcleanc.DocCleanConfig +} + +func main() { + ctx := context.Background() + + cfg := &DocSyncConfig{} + + cfg.ControllerFunc = func() runner.Controller { + tev := textversion.New(cfg) + clv := cleanversion.New(cfg) + col := collector.New(cfg, &collector.Services{ + CleanVersion: clv, + TextVersion: tev, + }) + cli := client.New(cfg) + jbb := job.New(cfg, &job.Services{ + Client: cli, + Collector: col, + }) + doc := document.New(cfg) + docsync := documentsync.New(cfg, &documentsync.Services{ + Document: doc, + Job: jbb, + }) + + return docsyncrunner.New(cfg.GetValidator(), &docsyncrunner.Services{ + Document: docsync, + }) + } + + server, err := runner.New(ctx, cfg) + if err != nil { + slog.Error(err.Error()) + os.Exit(1) + } + + server.Listen(ctx) +} diff --git a/cmd/docTextRunner/main.go b/cmd/docTextRunner/main.go index 66d66642..693f57d3 100644 --- a/cmd/docTextRunner/main.go +++ b/cmd/docTextRunner/main.go @@ -5,10 +5,9 @@ import ( "log/slog" "os" doctextrunner "queryorchestration/api/docTextRunner" - "queryorchestration/internal/document" documenttext "queryorchestration/internal/document/text" + textversion "queryorchestration/internal/document/text/version" "queryorchestration/internal/server/runner" - "queryorchestration/internal/serviceconfig/objectstore" "queryorchestration/internal/serviceconfig/queue/querysync" _ "github.com/lib/pq" @@ -16,7 +15,6 @@ import ( type DocTextConfig struct { runner.BaseConfig - objectstore.ObjectStoreConfig querysync.QuerySyncConfig } @@ -26,9 +24,8 @@ func main() { cfg := &DocTextConfig{} cfg.ControllerFunc = func() runner.Controller { - doc := document.New(cfg) text := documenttext.New(cfg, &documenttext.Services{ - Document: doc, + Version: textversion.New(cfg), }) return doctextrunner.New(cfg.GetValidator(), &doctextrunner.Services{ diff --git a/cmd/jobSyncRunner/main.go b/cmd/jobSyncRunner/main.go new file mode 100644 index 00000000..a3a1a8fb --- /dev/null +++ b/cmd/jobSyncRunner/main.go @@ -0,0 +1,42 @@ +package main + +import ( + "context" + "log/slog" + "os" + jobsyncrunner "queryorchestration/api/jobSyncRunner" + jobsync "queryorchestration/internal/job/sync" + "queryorchestration/internal/server/runner" + "queryorchestration/internal/serviceconfig/queue/documentsync" + + _ "github.com/lib/pq" +) + +type JobSyncConfig struct { + runner.BaseConfig + documentsync.DocSyncConfig +} + +func main() { + ctx := context.Background() + + cfg := &JobSyncConfig{} + + cfg.ControllerFunc = func() runner.Controller { + svc := jobsync.New(cfg) + + c := jobsyncrunner.New(cfg.GetValidator(), &jobsyncrunner.Services{ + JobSync: svc, + }) + + return &c + } + + server, err := runner.New(ctx, cfg) + if err != nil { + slog.Error(err.Error()) + os.Exit(1) + } + + server.Listen(ctx) +} diff --git a/cmd/queryService/main.go b/cmd/queryService/main.go index 734c9efa..4451ccf4 100644 --- a/cmd/queryService/main.go +++ b/cmd/queryService/main.go @@ -8,22 +8,31 @@ import ( queryservice "queryorchestration/api/queryService" "queryorchestration/internal/client" "queryorchestration/internal/document" + cleanversion "queryorchestration/internal/document/clean/version" + textversion "queryorchestration/internal/document/text/version" "queryorchestration/internal/export" "queryorchestration/internal/job" "queryorchestration/internal/job/collector" + collectorupdate "queryorchestration/internal/job/collector/update" "queryorchestration/internal/query" "queryorchestration/internal/query/result" querytest "queryorchestration/internal/query/test" service "queryorchestration/internal/server/service" + "queryorchestration/internal/serviceconfig/queue/jobsync" "github.com/getkin/kin-openapi/openapi3" _ "github.com/lib/pq" ) +type QueryServiceConfig struct { + service.BaseConfig + jobsync.JobSyncConfig +} + func main() { ctx := context.Background() - cfg := &service.BaseConfig{} + cfg := &QueryServiceConfig{} cfg.RegisterHandlersFunc = func() (*openapi3.T, error) { exp := export.New() @@ -31,28 +40,37 @@ func main() { res := result.New(cfg, &result.Services{ Query: que, }) - doc := document.New(cfg) + tev := textversion.New(cfg) + clv := cleanversion.New(cfg) col := collector.New(cfg, &collector.Services{ - Document: doc, + CleanVersion: clv, + TextVersion: tev, }) + colupdate := collectorupdate.New(cfg, &collectorupdate.Services{ + Collector: col, + CleanVersion: clv, + TextVersion: tev, + }) + cli := client.New(cfg) + jbb := job.New(cfg, &job.Services{ + Client: cli, + Collector: col, + }) + doc := document.New(cfg) quetest := querytest.New(cfg, &querytest.Services{ Collector: col, Result: res, Document: doc, }) - cli := client.New(cfg) - jbb := job.New(cfg, &job.Services{ - Collector: col, - Client: cli, - }) services := &queryservice.Services{ - Export: exp, - Collector: col, - Query: que, - Client: cli, - Job: jbb, - QueryTest: quetest, + Export: exp, + Collector: col, + CollectorUpdate: colupdate, + Query: que, + Client: cli, + Job: jbb, + QueryTest: quetest, } cons := queryservice.NewControllers(cfg.GetValidator(), services) diff --git a/database/migrations/00000000000102_job_views.down.sql b/database/migrations/00000000000102_job_views.down.sql new file mode 100644 index 00000000..58d361aa --- /dev/null +++ b/database/migrations/00000000000102_job_views.down.sql @@ -0,0 +1 @@ +DROP VIEW listJobDocumentIDs; diff --git a/database/migrations/00000000000102_job_views.up.sql b/database/migrations/00000000000102_job_views.up.sql new file mode 100644 index 00000000..c027b028 --- /dev/null +++ b/database/migrations/00000000000102_job_views.up.sql @@ -0,0 +1,20 @@ +CREATE OR REPLACE FUNCTION listJobDocumentIDs( + _jobId uuid, + _batchSize INTEGER, + _offset INTEGER +) +RETURNS TABLE (id uuid, totalCount BIGINT) AS $$ +DECLARE + totalCount BIGINT := 0; +BEGIN + SELECT COUNT(*)::BIGINT INTO totalCount FROM documents WHERE jobId = _jobId LIMIT 1; + + RETURN QUERY + SELECT d.id, totalCount + FROM documents as d + WHERE d.jobId = _jobId + ORDER BY d.id + LIMIT _batchSize + OFFSET _offset; +END; +$$ LANGUAGE plpgsql; \ No newline at end of file diff --git a/database/queries/job.sql b/database/queries/job.sql index c92fcc56..ef5da9e0 100644 --- a/database/queries/job.sql +++ b/database/queries/job.sql @@ -14,4 +14,7 @@ SELECT j.id, j.clientId, coalesce(cs.canSync, false) as canSync INSERT INTO jobs (clientId) VALUES ($1) RETURNING id; -- name: AddJobCanSync :exec -INSERT INTO jobCanSync (canSync, jobId) VALUES ($1, $2); \ No newline at end of file +INSERT INTO jobCanSync (canSync, jobId) VALUES ($1, $2); + +-- name: ListJobDocumentIDsBatch :many +SELECT id, totalCount FROM listJobDocumentIDs(@jobId, @batchSize, @pageOffset); \ No newline at end of file diff --git a/deployments/compose.local.yaml b/deployments/compose.local.yaml index 38bb37e9..b2b0a8d1 100644 --- a/deployments/compose.local.yaml +++ b/deployments/compose.local.yaml @@ -9,6 +9,30 @@ services: environment: LOG_LEVEL: DEBUG QUEUE_URL: ${DOCUMENT_INIT_URL} + DOCUMENT_SYNC_URL: ${DOCUMENT_SYNC_URL} + AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID} + AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY} + AWS_SESSION_TOKEN: ${AWS_SESSION_TOKEN} + AWS_REGION: ${AWS_REGION} + DB_USER: ${DB_USER} + DB_PASS: ${DB_PASS} + DB_HOST: db + DB_PORT: 5432 + DB_NAME: ${DB_NAME} + DB_NOSSL: ${DB_NOSSL} + AWS_ENDPOINT_URL: "http://localstack:4566" + AWS_S3_USE_PATH_STYLE: true + networks: + - server-network + doc_sync_runner: + image: queryorchestration:latest + command: ["./docSyncRunner"] + depends_on: + - db + - localstack + environment: + LOG_LEVEL: DEBUG + QUEUE_URL: ${DOCUMENT_SYNC_URL} DOCUMENT_CLEAN_URL: ${DOCUMENT_CLEAN_URL} AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID} AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY} @@ -120,6 +144,30 @@ services: AWS_S3_USE_PATH_STYLE: true networks: - server-network + job_sync_runner: + image: queryorchestration:latest + command: ["./jobSyncRunner"] + depends_on: + - db + - localstack + environment: + LOG_LEVEL: DEBUG + QUEUE_URL: ${JOB_SYNC_URL} + DOCUMENT_SYNC_URL: ${DOCUMENT_SYNC_URL} + AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID} + AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY} + AWS_SESSION_TOKEN: ${AWS_SESSION_TOKEN} + AWS_REGION: ${AWS_REGION} + DB_USER: ${DB_USER} + DB_PASS: ${DB_PASS} + DB_HOST: db + DB_PORT: 5432 + DB_NAME: ${DB_NAME} + DB_NOSSL: ${DB_NOSSL} + AWS_ENDPOINT_URL: "http://localstack:4566" + AWS_S3_USE_PATH_STYLE: true + networks: + - server-network query_service: image: queryorchestration:latest command: ["./queryService"] @@ -132,6 +180,7 @@ services: - 8080 environment: LOG_LEVEL: DEBUG + JOB_SYNC_URL: ${JOB_SYNC_URL} AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID} AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY} AWS_SESSION_TOKEN: ${AWS_SESSION_TOKEN} diff --git a/deployments/compose.test.yaml b/deployments/compose.test.yaml index 69343d10..451c52da 100644 --- a/deployments/compose.test.yaml +++ b/deployments/compose.test.yaml @@ -9,7 +9,7 @@ services: POSTGRES_PASSWORD: ${DB_PASS} POSTGRES_USER: ${DB_USER} expose: - - 5430 + - 5431 volumes: - test-db-data:/var/lib/postgresql/data healthcheck: diff --git a/devbox.json b/devbox.json index 18c3f096..fdbb56d2 100644 --- a/devbox.json +++ b/devbox.json @@ -22,7 +22,7 @@ "shell": { "init_hook": [ "export DB_URI=postgres://${DB_USER}:${DB_PASS}@${DB_HOST}:${DB_PORT}/${DB_NAME}?sslmode=disable", - "export DB_URI_TEST=postgres://${DB_USER}:${DB_PASS}@${DB_HOST}:${DB_PORT_GENERATE}/${DB_NAME}?sslmode=disable", + "export DB_URI_GENERATE=postgres://${DB_USER}:${DB_PASS}@${DB_HOST}:${DB_PORT_GENERATE}/${DB_NAME}?sslmode=disable", "echo 'Welcome to the DoczyAI devbox!'" ] }, @@ -41,15 +41,19 @@ "AWS_REGION": "us-east-1", "AWS_ENDPOINT_URL": "http://localhost:4566", "QNAME_DOCUMENT_INIT": "document_init", + "QNAME_DOCUMENT_SYNC": "document_sync", "QNAME_DOCUMENT_CLEAN": "document_clean", "QNAME_DOCUMENT_TEXT": "document_text", "QNAME_QUERY_SYNC": "query_sync", "QNAME_QUERY_RUNNER": "query_runner", + "QNAME_JOB_SYNC": "job_sync", "DOCUMENT_INIT_URL": "http://localstack:4566/queue/us-east-1/000000000000/document_init", + "DOCUMENT_SYNC_URL": "http://localstack:4566/queue/us-east-1/000000000000/document_sync", "DOCUMENT_CLEAN_URL": "http://localstack:4566/queue/us-east-1/000000000000/document_clean", "DOCUMENT_TEXT_URL": "http://localstack:4566/queue/us-east-1/000000000000/document_text", "QUERY_SYNC_URL": "http://localstack:4566/queue/us-east-1/000000000000/query_sync", "QUERY_URL": "http://localstack:4566/queue/us-east-1/000000000000/query_runner", + "JOB_SYNC_URL": "http://localstack:4566/queue/us-east-1/000000000000/job_sync", "BUCKET_IN": "documentin" }, "env_from": ".env" diff --git a/internal/database/repository/job.sql.go b/internal/database/repository/job.sql.go index 3d987c34..4f6e5017 100644 --- a/internal/database/repository/job.sql.go +++ b/internal/database/repository/job.sql.go @@ -79,3 +79,41 @@ func (q *Queries) GetJob(ctx context.Context, jobid pgtype.UUID) (*GetJobRow, er err := row.Scan(&i.ID, &i.Clientid, &i.Cansync) return &i, err } + +const listJobDocumentIDsBatch = `-- name: ListJobDocumentIDsBatch :many +SELECT id, totalCount FROM listJobDocumentIDs($1, $2, $3) +` + +type ListJobDocumentIDsBatchParams struct { + Jobid pgtype.UUID `db:"jobid"` + Batchsize int32 `db:"batchsize"` + Pageoffset int32 `db:"pageoffset"` +} + +type ListJobDocumentIDsBatchRow struct { + ID pgtype.UUID `db:"id"` + Totalcount *int64 `db:"totalcount"` +} + +// ListJobDocumentIDsBatch +// +// SELECT id, totalCount FROM listJobDocumentIDs($1, $2, $3) +func (q *Queries) ListJobDocumentIDsBatch(ctx context.Context, arg *ListJobDocumentIDsBatchParams) ([]*ListJobDocumentIDsBatchRow, error) { + rows, err := q.db.Query(ctx, listJobDocumentIDsBatch, arg.Jobid, arg.Batchsize, arg.Pageoffset) + if err != nil { + return nil, err + } + defer rows.Close() + items := []*ListJobDocumentIDsBatchRow{} + for rows.Next() { + var i ListJobDocumentIDsBatchRow + if err := rows.Scan(&i.ID, &i.Totalcount); err != nil { + return nil, err + } + items = append(items, &i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} diff --git a/internal/database/repository/job_test.go b/internal/database/repository/job_test.go index fd732b2a..b510bf25 100644 --- a/internal/database/repository/job_test.go +++ b/internal/database/repository/job_test.go @@ -58,3 +58,109 @@ func TestJob(t *testing.T) { Cansync: true, }, job) } +func TestListJobDocumentIDs(t *testing.T) { + if testing.Short() { + t.Skip("Skipping long test in short mode") + } + ctx := context.Background() + + cfg := &serviceconfig.BaseConfig{} + test.SetCfgProvider(t, cfg) + cfg.SetBasePath(path.Join(os.Getenv("PWD"), "../../..")) + _, cleanup := test.CreateDB(t, ctx, &test.CreateDatabaseConfig{ + Cfg: cfg, + RunMigrations: true, + }) + defer cleanup() + + queries := cfg.GetDBQueries() + + clientId, err := queries.CreateClient(ctx, "example_client") + assert.NoError(t, err) + + id, err := queries.CreateJob(ctx, clientId) + assert.NoError(t, err) + assert.NotEmpty(t, id) + + ids, err := queries.ListJobDocumentIDsBatch(ctx, &repository.ListJobDocumentIDsBatchParams{ + Jobid: id, + Batchsize: 1, + Pageoffset: 0, + }) + assert.NoError(t, err) + assert.Len(t, ids, 0) + + docOne, err := queries.CreateDocument(ctx, &repository.CreateDocumentParams{ + Jobid: id, + Hash: "example_hash", + }) + assert.NoError(t, err) + + ids, err = queries.ListJobDocumentIDsBatch(ctx, &repository.ListJobDocumentIDsBatchParams{ + Jobid: id, + Batchsize: 1, + Pageoffset: 0, + }) + assert.NoError(t, err) + assert.Len(t, ids, 1) + total := int64(1) + assert.ElementsMatch(t, []*repository.ListJobDocumentIDsBatchRow{ + { + ID: docOne, + Totalcount: &total, + }, + }, ids) + + docTwo, err := queries.CreateDocument(ctx, &repository.CreateDocumentParams{ + Jobid: id, + Hash: "example_hash_two", + }) + assert.NoError(t, err) + + ids, err = queries.ListJobDocumentIDsBatch(ctx, &repository.ListJobDocumentIDsBatchParams{ + Jobid: id, + Batchsize: 1, + Pageoffset: 0, + }) + assert.NoError(t, err) + assert.Len(t, ids, 1) + total = int64(2) + assert.ElementsMatch(t, []*repository.ListJobDocumentIDsBatchRow{ + { + ID: docOne, + Totalcount: &total, + }, + }, ids) + + ids, err = queries.ListJobDocumentIDsBatch(ctx, &repository.ListJobDocumentIDsBatchParams{ + Jobid: id, + Batchsize: 1, + Pageoffset: 1, + }) + assert.NoError(t, err) + assert.Len(t, ids, 1) + assert.ElementsMatch(t, []*repository.ListJobDocumentIDsBatchRow{ + { + ID: docTwo, + Totalcount: &total, + }, + }, ids) + + ids, err = queries.ListJobDocumentIDsBatch(ctx, &repository.ListJobDocumentIDsBatchParams{ + Jobid: id, + Batchsize: 2, + Pageoffset: 0, + }) + assert.NoError(t, err) + assert.Len(t, ids, 2) + assert.ElementsMatch(t, []*repository.ListJobDocumentIDsBatchRow{ + { + ID: docOne, + Totalcount: &total, + }, + { + ID: docTwo, + Totalcount: &total, + }, + }, ids) +} diff --git a/internal/database/repository/result_test.go b/internal/database/repository/result_test.go index 7707b941..dc192e4f 100644 --- a/internal/database/repository/result_test.go +++ b/internal/database/repository/result_test.go @@ -2,7 +2,6 @@ package repository_test import ( "context" - "log" "os" "path" "queryorchestration/internal/database/repository" @@ -235,9 +234,6 @@ func TestResultValues(t *testing.T) { Documentid: documentID, Version: jsonVersion, }) - for _, r := range qResults { - log.Print(r) - } assert.NoError(t, err) assert.Len(t, qResults, 1) assert.EqualExportedValues(t, []*repository.ListQueryRequirementValuesRow{ diff --git a/internal/document/clean/clean.go b/internal/document/clean/clean.go index eb185d15..938ab73f 100644 --- a/internal/document/clean/clean.go +++ b/internal/document/clean/clean.go @@ -41,7 +41,7 @@ func (s *Service) clean(ctx context.Context, id uuid.UUID) error { return err } - version := s.svc.Document.GetCleanVersion() + version := s.svc.Version.GetVersion() err = s.cfg.GetDBQueries().AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ Documentid: docId, diff --git a/internal/document/clean/clean_test.go b/internal/document/clean/clean_test.go index f77c627c..0c12ddf7 100644 --- a/internal/document/clean/clean_test.go +++ b/internal/document/clean/clean_test.go @@ -5,6 +5,7 @@ import ( "queryorchestration/internal/database" "queryorchestration/internal/database/repository" "queryorchestration/internal/document" + cleanversion "queryorchestration/internal/document/clean/version" "testing" "github.com/google/uuid" @@ -26,7 +27,7 @@ func TestClean(t *testing.T) { svc := Service{ cfg: cfg, svc: &Services{ - Document: document.New(cfg), + Version: cleanversion.New(cfg), }, } diff --git a/internal/document/clean/create_test.go b/internal/document/clean/create_test.go index 2dedde31..040cb371 100644 --- a/internal/document/clean/create_test.go +++ b/internal/document/clean/create_test.go @@ -6,8 +6,8 @@ import ( "queryorchestration/internal/database" "queryorchestration/internal/database/repository" "queryorchestration/internal/document" + cleanversion "queryorchestration/internal/document/clean/version" "queryorchestration/internal/serviceconfig" - "queryorchestration/internal/serviceconfig/objectstore" "queryorchestration/internal/serviceconfig/queue/documenttext" queuemock "queryorchestration/mocks/queue" "testing" @@ -22,7 +22,6 @@ import ( type DocCleanConfig struct { serviceconfig.BaseConfig documenttext.DocTextConfig - objectstore.ObjectStoreConfig } func TestCreate(t *testing.T) { @@ -43,7 +42,7 @@ func TestCreate(t *testing.T) { svc := Service{ cfg: cfg, svc: &Services{ - Document: document.New(cfg), + Version: cleanversion.New(cfg), }, } diff --git a/internal/document/clean/service.go b/internal/document/clean/service.go index 4e843406..784a655b 100644 --- a/internal/document/clean/service.go +++ b/internal/document/clean/service.go @@ -1,20 +1,18 @@ package documentclean import ( - "queryorchestration/internal/document" + cleanversion "queryorchestration/internal/document/clean/version" "queryorchestration/internal/serviceconfig" - "queryorchestration/internal/serviceconfig/objectstore" "queryorchestration/internal/serviceconfig/queue/documenttext" ) type ConfigProvider interface { serviceconfig.ConfigProvider - objectstore.ConfigProvider documenttext.ConfigProvider } type Services struct { - Document *document.Service + Version *cleanversion.Service } type Service struct { diff --git a/internal/document/clean/service_test.go b/internal/document/clean/service_test.go index b0dd5e45..17a965c4 100644 --- a/internal/document/clean/service_test.go +++ b/internal/document/clean/service_test.go @@ -3,7 +3,6 @@ package documentclean_test import ( documentclean "queryorchestration/internal/document/clean" "queryorchestration/internal/serviceconfig" - "queryorchestration/internal/serviceconfig/objectstore" "queryorchestration/internal/serviceconfig/queue/documenttext" "testing" @@ -13,7 +12,6 @@ import ( type DocCleanConfig struct { serviceconfig.BaseConfig documenttext.DocTextConfig - objectstore.ObjectStoreConfig } func TestService(t *testing.T) { diff --git a/internal/document/clean/version/service.go b/internal/document/clean/version/service.go new file mode 100644 index 00000000..0f95171a --- /dev/null +++ b/internal/document/clean/version/service.go @@ -0,0 +1,15 @@ +package cleanversion + +import ( + "queryorchestration/internal/serviceconfig" +) + +type Service struct { + cfg serviceconfig.ConfigProvider +} + +func New(cfg serviceconfig.ConfigProvider) *Service { + return &Service{ + cfg, + } +} diff --git a/internal/document/clean/version/service_test.go b/internal/document/clean/version/service_test.go new file mode 100644 index 00000000..2894665c --- /dev/null +++ b/internal/document/clean/version/service_test.go @@ -0,0 +1,13 @@ +package cleanversion_test + +import ( + cleanversion "queryorchestration/internal/document/clean/version" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestNew(t *testing.T) { + svc := cleanversion.New(nil) + assert.NotNil(t, svc) +} diff --git a/internal/document/clean/version/version.go b/internal/document/clean/version/version.go new file mode 100644 index 00000000..defa953e --- /dev/null +++ b/internal/document/clean/version/version.go @@ -0,0 +1,20 @@ +package cleanversion + +import ( + "fmt" +) + +func (s *Service) GetVersion() int32 { + // TODO - actual version + return 1 +} + +func (s *Service) IsValidVersion(v int32) error { + currentVersion := s.GetVersion() + + if v < 1 || v > currentVersion { + return fmt.Errorf("document clean code version must be in the range 0 < version <= %d", currentVersion) + } + + return nil +} diff --git a/internal/document/clean/version/version_test.go b/internal/document/clean/version/version_test.go new file mode 100644 index 00000000..e29a1545 --- /dev/null +++ b/internal/document/clean/version/version_test.go @@ -0,0 +1,26 @@ +package cleanversion_test + +import ( + cleanversion "queryorchestration/internal/document/clean/version" + "queryorchestration/internal/serviceconfig" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestIsValidVersion(t *testing.T) { + cfg := &serviceconfig.BaseConfig{} + svc := cleanversion.New(cfg) + + assert.Nil(t, svc.IsValidVersion(1)) + assert.Error(t, svc.IsValidVersion(2)) + assert.Error(t, svc.IsValidVersion(0)) + assert.Error(t, svc.IsValidVersion(-1)) +} + +func TestGetVersion(t *testing.T) { + cfg := &serviceconfig.BaseConfig{} + svc := cleanversion.New(cfg) + + assert.Equal(t, int32(1), svc.GetVersion()) +} diff --git a/internal/document/init/create.go b/internal/document/init/create.go index a709ce9d..3396efb5 100644 --- a/internal/document/init/create.go +++ b/internal/document/init/create.go @@ -6,11 +6,10 @@ import ( "errors" "fmt" "log/slog" - doccleanrunner "queryorchestration/api/docCleanRunner" + docsyncrunner "queryorchestration/api/docSyncRunner" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" "queryorchestration/internal/document" - "queryorchestration/internal/job" "queryorchestration/internal/serviceconfig/queue" "regexp" @@ -26,11 +25,6 @@ type Create struct { } func (s *Service) Create(ctx context.Context, doc *Create) (uuid.UUID, error) { - j, err := s.svc.Job.Get(ctx, doc.JobID) - if err != nil { - return uuid.Nil, err - } - params, err := s.getCreateParams(ctx, doc) if err != nil { return uuid.Nil, err @@ -42,7 +36,12 @@ func (s *Service) Create(ctx context.Context, doc *Create) (uuid.UUID, error) { } if params.ID == nil { - err = s.informCreate(ctx, id, j) + err := s.cfg.SendToQueue(ctx, &queue.SendParams{ + QueueURL: s.cfg.GetDocumentSyncURL(), + Body: docsyncrunner.Body{ + ID: id, + }, + }) if err != nil { return uuid.Nil, err } @@ -120,32 +119,19 @@ func (s *Service) submitCreate(ctx context.Context, params *createDocumentParams return id, nil } -func (s *Service) informCreate(ctx context.Context, id uuid.UUID, j *job.Job) error { - if !j.CanSync { - return nil - } - - err := s.cfg.SendToQueue(ctx, &queue.SendParams{ - QueueURL: s.cfg.GetDocumentCleanURL(), - Body: doccleanrunner.Body{ - ID: id, - }, - }) - if err != nil { - return err - } - - return nil -} - func (s *Service) GetJobIDFromKey(key string) (uuid.UUID, error) { - re := regexp.MustCompile(`^.+/(.+)/.+$`) + re := regexp.MustCompile(`^(.+)/(.+)/.+$`) match := re.FindStringSubmatch(key) - if match == nil || len(match) != 2 { + if match == nil || len(match) != 3 { return uuid.Nil, fmt.Errorf("no job id match found in key") } - id, err := uuid.Parse(match[1]) + _, err := uuid.Parse(match[1]) + if err != nil { + return uuid.Nil, err + } + + id, err := uuid.Parse(match[2]) if err != nil { return uuid.Nil, err } diff --git a/internal/document/init/create_test.go b/internal/document/init/create_test.go index 4d0fbd99..0ea39903 100644 --- a/internal/document/init/create_test.go +++ b/internal/document/init/create_test.go @@ -4,15 +4,10 @@ import ( "context" "errors" "fmt" - "queryorchestration/internal/client" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" "queryorchestration/internal/document" "queryorchestration/internal/job" - "queryorchestration/internal/serviceconfig" - "queryorchestration/internal/serviceconfig/objectstore" - "queryorchestration/internal/serviceconfig/queue/documentclean" - objectstoremock "queryorchestration/mocks/objectstore" queuemock "queryorchestration/mocks/queue" "testing" @@ -23,12 +18,6 @@ import ( "github.com/stretchr/testify/mock" ) -type DocInitConfig struct { - serviceconfig.BaseConfig - documentclean.DocCleanConfig - objectstore.ObjectStoreConfig -} - func TestCreate(t *testing.T) { ctx := context.Background() @@ -39,14 +28,11 @@ func TestCreate(t *testing.T) { cfg := &DocInitConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) - mockStore := objectstoremock.NewMockS3Client(t) - cfg.StoreClient = mockStore + mockSQS := queuemock.NewMockSQSClient(t) + cfg.QueueClient = mockSQS + cfg.DocumentSyncURL = "/i/am/here" - svc := New(cfg, &Services{ - Job: job.New(cfg, &job.Services{ - Client: client.New(cfg), - }), - }) + svc := New(cfg) j := job.Job{ ID: uuid.New(), @@ -62,14 +48,6 @@ func TestCreate(t *testing.T) { Key: "/i/am/here", } - pool.ExpectQuery("name: GetJob :one").WithArgs(database.MustToDBUUID(j.ID)).WillReturnRows( - pgxmock.NewRows([]string{"id", "clientId", "canSync"}). - AddRow(database.MustToDBUUID(j.ID), database.MustToDBUUID(j.ClientID), j.CanSync), - ) - pool.ExpectQuery("name: GetClient :one").WithArgs(database.MustToDBUUID(j.ClientID)).WillReturnRows( - pgxmock.NewRows([]string{"id", "name", "canSync"}). - AddRow(database.MustToDBUUID(j.ClientID), "client_name", true), - ) pool.ExpectQuery("name: GetDocumentIDByHash :one").WithArgs(doc.Hash, database.MustToDBUUID(doc.JobID)).WillReturnRows( pgxmock.NewRows([]string{"id"}), ) @@ -82,6 +60,29 @@ func TestCreate(t *testing.T) { pool.ExpectExec("name: AddDocumentEntry :exec").WithArgs(database.MustToDBUUID(doc.ID), location.Bucket, location.Key). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectCommit() + pool.ExpectQuery("name: GetDocument :one").WithArgs(database.MustToDBUUID(doc.ID)). + WillReturnRows( + pgxmock.NewRows([]string{"id", "jobId", "hash"}). + AddRow(database.MustToDBUUID(doc.ID), database.MustToDBUUID(doc.JobID), doc.Hash), + ) + pool.ExpectQuery("name: GetJob :one").WithArgs(database.MustToDBUUID(j.ID)).WillReturnRows( + pgxmock.NewRows([]string{"id", "clientId", "canSync"}). + AddRow(database.MustToDBUUID(j.ID), database.MustToDBUUID(j.ClientID), j.CanSync), + ) + pool.ExpectQuery("name: GetClient :one").WithArgs(database.MustToDBUUID(j.ClientID)).WillReturnRows( + pgxmock.NewRows([]string{"id", "name", "canSync"}). + AddRow(database.MustToDBUUID(j.ClientID), "client_name", true), + ) + + mockSQS.EXPECT(). + SendMessage( + mock.Anything, + mock.MatchedBy(func(in *sqs.SendMessageInput) bool { + return *in.QueueUrl == cfg.DocumentSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", doc.ID.String()) + }), + mock.Anything, + ). + Return(&sqs.SendMessageOutput{}, nil) id, err := svc.Create(ctx, &Create{ JobID: doc.JobID, @@ -92,37 +93,6 @@ func TestCreate(t *testing.T) { assert.Equal(t, doc.ID, id) } -func TestInformCreate(t *testing.T) { - ctx := context.Background() - mockSQS := queuemock.NewMockSQSClient(t) - - cfg := &DocInitConfig{} - cfg.QueueClient = mockSQS - cfg.DocumentCleanURL = "/i/am/here" - svc := Service{ - cfg: cfg, - } - id := uuid.New() - j := &job.Job{} - - err := svc.informCreate(ctx, id, j) - assert.NoError(t, err) - - mockSQS.EXPECT(). - SendMessage( - mock.Anything, - mock.MatchedBy(func(in *sqs.SendMessageInput) bool { - return *in.QueueUrl == cfg.DocumentCleanURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", id) - }), - mock.Anything, - ). - Return(&sqs.SendMessageOutput{}, nil) - - j.CanSync = true - err = svc.informCreate(ctx, id, j) - assert.NoError(t, err) -} - func TestGetCreateParams(t *testing.T) { ctx := context.Background() @@ -133,14 +103,8 @@ func TestGetCreateParams(t *testing.T) { cfg := &DocInitConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) - mockStore := objectstoremock.NewMockS3Client(t) - cfg.StoreClient = mockStore - svc := New(cfg, &Services{ - Job: job.New(cfg, &job.Services{ - Client: client.New(cfg), - }), - }) + svc := New(cfg) doc := document.Document{ JobID: uuid.New(), @@ -178,14 +142,8 @@ func TestGetCreateParamsExisting(t *testing.T) { cfg := &DocInitConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) - mockStore := objectstoremock.NewMockS3Client(t) - cfg.StoreClient = mockStore - svc := New(cfg, &Services{ - Job: job.New(cfg, &job.Services{ - Client: client.New(cfg), - }), - }) + svc := New(cfg) doc := document.Document{ ID: uuid.New(), @@ -227,14 +185,8 @@ func TestGetCreateParamsCurrentDocErr(t *testing.T) { cfg := &DocInitConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) - mockStore := objectstoremock.NewMockS3Client(t) - cfg.StoreClient = mockStore - svc := New(cfg, &Services{ - Job: job.New(cfg, &job.Services{ - Client: client.New(cfg), - }), - }) + svc := New(cfg) doc := document.Document{ JobID: uuid.New(), @@ -266,11 +218,7 @@ func TestSubmitCreate(t *testing.T) { cfg.DBPool = pool cfg.DBQueries = repository.New(pool) - svc := New(cfg, &Services{ - Job: job.New(cfg, &job.Services{ - Client: client.New(cfg), - }), - }) + svc := New(cfg) doc := document.Document{ ID: uuid.New(), @@ -312,11 +260,7 @@ func TestSubmitCreateExists(t *testing.T) { cfg.DBPool = pool cfg.DBQueries = repository.New(pool) - svc := New(cfg, &Services{ - Job: job.New(cfg, &job.Services{ - Client: client.New(cfg), - }), - }) + svc := New(cfg) doc := document.Document{ ID: uuid.New(), @@ -357,7 +301,10 @@ func TestGetJobIDFromKey(t *testing.T) { assert.Error(t, err) id := uuid.New() - aid, err := svc.GetJobIDFromKey(fmt.Sprintf("aaa/%s/bbb", id.String())) + _, err = svc.GetJobIDFromKey(fmt.Sprintf("aaa/%s/bbb", id.String())) + assert.Error(t, err) + + aid, err := svc.GetJobIDFromKey(fmt.Sprintf("%s/%s/bbb", uuid.NewString(), id.String())) assert.NoError(t, err) assert.Equal(t, id, aid) } diff --git a/internal/document/init/service.go b/internal/document/init/service.go index cb62ad90..9a0c5d0d 100644 --- a/internal/document/init/service.go +++ b/internal/document/init/service.go @@ -1,28 +1,21 @@ package documentinit import ( - "queryorchestration/internal/job" "queryorchestration/internal/serviceconfig" - "queryorchestration/internal/serviceconfig/queue/documentclean" + documentsync "queryorchestration/internal/serviceconfig/queue/documentsync" ) -type Services struct { - Job *job.Service -} - type ConfigProvider interface { serviceconfig.ConfigProvider - documentclean.ConfigProvider + documentsync.ConfigProvider } type Service struct { cfg ConfigProvider - svc *Services } -func New(cfg ConfigProvider, svc *Services) *Service { +func New(cfg ConfigProvider) *Service { return &Service{ cfg, - svc, } } diff --git a/internal/document/init/service_test.go b/internal/document/init/service_test.go index 1cb80b7f..b31d67f2 100644 --- a/internal/document/init/service_test.go +++ b/internal/document/init/service_test.go @@ -1,13 +1,20 @@ -package documentinit_test +package documentinit import ( - "queryorchestration/internal/document" + "queryorchestration/internal/serviceconfig" + "queryorchestration/internal/serviceconfig/queue/documentsync" "testing" "github.com/stretchr/testify/assert" ) -func TestNew(t *testing.T) { - svc := document.New(nil) - assert.NotNil(t, svc) +type DocInitConfig struct { + serviceconfig.BaseConfig + documentsync.DocSyncConfig +} + +func TestNew(t *testing.T) { + svc := New(&DocInitConfig{}) + assert.NotNil(t, svc) + } diff --git a/internal/document/sync/service.go b/internal/document/sync/service.go new file mode 100644 index 00000000..c2326615 --- /dev/null +++ b/internal/document/sync/service.go @@ -0,0 +1,30 @@ +package documentsync + +import ( + "queryorchestration/internal/document" + "queryorchestration/internal/job" + "queryorchestration/internal/serviceconfig" + "queryorchestration/internal/serviceconfig/queue/documentclean" +) + +type Services struct { + Document *document.Service + Job *job.Service +} + +type ConfigProvider interface { + serviceconfig.ConfigProvider + documentclean.ConfigProvider +} + +type Service struct { + cfg ConfigProvider + svc *Services +} + +func New(cfg ConfigProvider, svc *Services) *Service { + return &Service{ + cfg, + svc, + } +} diff --git a/internal/document/sync/service_test.go b/internal/document/sync/service_test.go new file mode 100644 index 00000000..9b8115ad --- /dev/null +++ b/internal/document/sync/service_test.go @@ -0,0 +1,20 @@ +package documentsync_test + +import ( + documentsync "queryorchestration/internal/document/sync" + "queryorchestration/internal/serviceconfig" + "queryorchestration/internal/serviceconfig/queue/documentclean" + "testing" + + "github.com/stretchr/testify/assert" +) + +type DocSyncConfig struct { + serviceconfig.BaseConfig + documentclean.DocCleanConfig +} + +func TestNew(t *testing.T) { + svc := documentsync.New(&DocSyncConfig{}, &documentsync.Services{}) + assert.NotNil(t, svc) +} diff --git a/internal/document/sync/sync.go b/internal/document/sync/sync.go new file mode 100644 index 00000000..96aaa572 --- /dev/null +++ b/internal/document/sync/sync.go @@ -0,0 +1,41 @@ +package documentsync + +import ( + "context" + "log/slog" + doccleanrunner "queryorchestration/api/docCleanRunner" + "queryorchestration/internal/serviceconfig/queue" + + "github.com/google/uuid" +) + +func (s *Service) Sync(ctx context.Context, id uuid.UUID) error { + doc, err := s.svc.Document.Get(ctx, id) + if err != nil { + return err + } + + j, err := s.svc.Job.Get(ctx, doc.JobID) + if err != nil { + return err + } + + if !j.CanSync { + slog.Debug("not syncing document", "id", id.String()) + return nil + } + + slog.Debug("syncing document", "id", id.String()) + + err = s.cfg.SendToQueue(ctx, &queue.SendParams{ + QueueURL: s.cfg.GetDocumentCleanURL(), + Body: doccleanrunner.Body{ + ID: id, + }, + }) + if err != nil { + return err + } + + return nil +} diff --git a/internal/document/sync/sync_test.go b/internal/document/sync/sync_test.go new file mode 100644 index 00000000..c68c96c3 --- /dev/null +++ b/internal/document/sync/sync_test.go @@ -0,0 +1,80 @@ +package documentsync_test + +import ( + "context" + "fmt" + "queryorchestration/internal/client" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/document" + documentsync "queryorchestration/internal/document/sync" + "queryorchestration/internal/job" + queuemock "queryorchestration/mocks/queue" + "testing" + + "github.com/aws/aws-sdk-go-v2/service/sqs" + "github.com/google/uuid" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" +) + +func TestSync(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + cfg := &DocSyncConfig{} + cfg.DBPool = pool + cfg.DBQueries = repository.New(pool) + mockSQS := queuemock.NewMockSQSClient(t) + cfg.QueueClient = mockSQS + cfg.DocumentCleanURL = "/i/am/here" + + svc := documentsync.New(cfg, &documentsync.Services{ + Document: document.New(cfg), + Job: job.New(cfg, &job.Services{ + Client: client.New(cfg), + }), + }) + + j := job.Job{ + ID: uuid.New(), + ClientID: uuid.New(), + CanSync: true, + } + doc := document.Document{ + ID: uuid.New(), + JobID: j.ID, + Hash: "example_hash", + } + + pool.ExpectQuery("name: GetDocument :one").WithArgs(database.MustToDBUUID(doc.ID)). + WillReturnRows( + pgxmock.NewRows([]string{"id", "jobId", "hash"}). + AddRow(database.MustToDBUUID(doc.ID), database.MustToDBUUID(doc.JobID), doc.Hash), + ) + pool.ExpectQuery("name: GetJob :one").WithArgs(database.MustToDBUUID(j.ID)).WillReturnRows( + pgxmock.NewRows([]string{"id", "clientId", "canSync"}). + AddRow(database.MustToDBUUID(j.ID), database.MustToDBUUID(j.ClientID), j.CanSync), + ) + pool.ExpectQuery("name: GetClient :one").WithArgs(database.MustToDBUUID(j.ClientID)).WillReturnRows( + pgxmock.NewRows([]string{"id", "name", "canSync"}). + AddRow(database.MustToDBUUID(j.ClientID), "client_name", true), + ) + + mockSQS.EXPECT(). + SendMessage( + mock.Anything, + mock.MatchedBy(func(in *sqs.SendMessageInput) bool { + return *in.QueueUrl == cfg.DocumentCleanURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", doc.ID) + }), + mock.Anything, + ). + Return(&sqs.SendMessageOutput{}, nil) + + err = svc.Sync(ctx, doc.ID) + assert.NoError(t, err) +} diff --git a/internal/document/text/create_test.go b/internal/document/text/create_test.go index da6c0abb..8c3e3669 100644 --- a/internal/document/text/create_test.go +++ b/internal/document/text/create_test.go @@ -6,8 +6,8 @@ import ( "queryorchestration/internal/database" "queryorchestration/internal/database/repository" "queryorchestration/internal/document" + textversion "queryorchestration/internal/document/text/version" "queryorchestration/internal/serviceconfig" - "queryorchestration/internal/serviceconfig/objectstore" "queryorchestration/internal/serviceconfig/queue/querysync" queuemock "queryorchestration/mocks/queue" "testing" @@ -21,7 +21,6 @@ import ( type DocTextConfig struct { serviceconfig.BaseConfig - objectstore.ObjectStoreConfig querysync.QuerySyncConfig } @@ -43,7 +42,7 @@ func TestCreate(t *testing.T) { svc := Service{ cfg: cfg, svc: &Services{ - Document: document.New(cfg), + Version: textversion.New(cfg), }, } diff --git a/internal/document/text/extract.go b/internal/document/text/extract.go index a95ffce8..e43be303 100644 --- a/internal/document/text/extract.go +++ b/internal/document/text/extract.go @@ -38,7 +38,7 @@ func (s *Service) extract(ctx context.Context, id uuid.UUID) error { return err } - version := s.svc.Document.GetTextVersion() + version := s.svc.Version.GetVersion() err = s.cfg.GetDBQueries().AddDocumentTextEntry(ctx, &repository.AddDocumentTextEntryParams{ Documentid: docId, diff --git a/internal/document/text/extract_test.go b/internal/document/text/extract_test.go index 0ddb78f2..b25acb94 100644 --- a/internal/document/text/extract_test.go +++ b/internal/document/text/extract_test.go @@ -5,6 +5,7 @@ import ( "queryorchestration/internal/database" "queryorchestration/internal/database/repository" "queryorchestration/internal/document" + textversion "queryorchestration/internal/document/text/version" "testing" "github.com/google/uuid" @@ -26,7 +27,7 @@ func TestExtract(t *testing.T) { svc := Service{ cfg: cfg, svc: &Services{ - Document: document.New(cfg), + Version: textversion.New(cfg), }, } diff --git a/internal/document/text/service.go b/internal/document/text/service.go index 01689df4..2d8ce42e 100644 --- a/internal/document/text/service.go +++ b/internal/document/text/service.go @@ -1,20 +1,18 @@ package documenttext import ( - "queryorchestration/internal/document" + textversion "queryorchestration/internal/document/text/version" "queryorchestration/internal/serviceconfig" - "queryorchestration/internal/serviceconfig/objectstore" "queryorchestration/internal/serviceconfig/queue/querysync" ) type ConfigProvider interface { serviceconfig.ConfigProvider - objectstore.ConfigProvider querysync.ConfigProvider } type Services struct { - Document *document.Service + Version *textversion.Service } type Service struct { diff --git a/internal/document/text/service_test.go b/internal/document/text/service_test.go index 7c684240..ccdd44a1 100644 --- a/internal/document/text/service_test.go +++ b/internal/document/text/service_test.go @@ -1,10 +1,9 @@ package documenttext_test import ( - "queryorchestration/internal/document" documenttext "queryorchestration/internal/document/text" + textversion "queryorchestration/internal/document/text/version" "queryorchestration/internal/serviceconfig" - "queryorchestration/internal/serviceconfig/objectstore" "queryorchestration/internal/serviceconfig/queue/querysync" "testing" @@ -13,14 +12,13 @@ import ( type DocTextConfig struct { serviceconfig.BaseConfig - objectstore.ObjectStoreConfig querysync.QuerySyncConfig } func TestService(t *testing.T) { cfg := &DocTextConfig{} svc := documenttext.New(cfg, &documenttext.Services{ - Document: document.New(cfg), + Version: textversion.New(cfg), }) assert.NotNil(t, svc) } diff --git a/internal/document/text/version/service.go b/internal/document/text/version/service.go new file mode 100644 index 00000000..3d934196 --- /dev/null +++ b/internal/document/text/version/service.go @@ -0,0 +1,15 @@ +package textversion + +import ( + "queryorchestration/internal/serviceconfig" +) + +type Service struct { + cfg serviceconfig.ConfigProvider +} + +func New(cfg serviceconfig.ConfigProvider) *Service { + return &Service{ + cfg, + } +} diff --git a/internal/document/text/version/service_test.go b/internal/document/text/version/service_test.go new file mode 100644 index 00000000..5f94579b --- /dev/null +++ b/internal/document/text/version/service_test.go @@ -0,0 +1,13 @@ +package textversion_test + +import ( + textversion "queryorchestration/internal/document/clean/version" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestNew(t *testing.T) { + svc := textversion.New(nil) + assert.NotNil(t, svc) +} diff --git a/internal/document/text/version/version.go b/internal/document/text/version/version.go new file mode 100644 index 00000000..7e3ce64a --- /dev/null +++ b/internal/document/text/version/version.go @@ -0,0 +1,20 @@ +package textversion + +import ( + "fmt" +) + +func (s *Service) GetVersion() int32 { + // TODO - actual version + return 1 +} + +func (s *Service) IsValidVersion(v int32) error { + currentVersion := s.GetVersion() + + if v < 1 || v > currentVersion { + return fmt.Errorf("document text code version must be in the range 0 < version <= %d", currentVersion) + } + + return nil +} diff --git a/internal/document/text/version/version_test.go b/internal/document/text/version/version_test.go new file mode 100644 index 00000000..f8068b21 --- /dev/null +++ b/internal/document/text/version/version_test.go @@ -0,0 +1,26 @@ +package textversion_test + +import ( + textversion "queryorchestration/internal/document/clean/version" + "queryorchestration/internal/serviceconfig" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestIsValidVersion(t *testing.T) { + cfg := &serviceconfig.BaseConfig{} + svc := textversion.New(cfg) + + assert.Nil(t, svc.IsValidVersion(1)) + assert.Error(t, svc.IsValidVersion(2)) + assert.Error(t, svc.IsValidVersion(0)) + assert.Error(t, svc.IsValidVersion(-1)) +} + +func TestGetVersion(t *testing.T) { + cfg := &serviceconfig.BaseConfig{} + svc := textversion.New(cfg) + + assert.Equal(t, int32(1), svc.GetVersion()) +} diff --git a/internal/document/version.go b/internal/document/version.go deleted file mode 100644 index 222ffb1e..00000000 --- a/internal/document/version.go +++ /dev/null @@ -1,35 +0,0 @@ -package document - -import ( - "fmt" -) - -func (s *Service) GetCleanVersion() int32 { - // TODO - actual version - return 1 -} - -func (s *Service) GetTextVersion() int32 { - // TODO - actual version - return 1 -} - -func (s *Service) IsValidCleanVersion(v int32) error { - currentVersion := s.GetCleanVersion() - - if v < 1 || v > currentVersion { - return fmt.Errorf("document clean code version must be in the range 0 < version <= %d", currentVersion) - } - - return nil -} - -func (s *Service) IsValidTextVersion(v int32) error { - currentVersion := s.GetTextVersion() - - if v < 1 || v > currentVersion { - return fmt.Errorf("document text code version must be in the range 0 < version <= %d", currentVersion) - } - - return nil -} diff --git a/internal/document/version_test.go b/internal/document/version_test.go deleted file mode 100644 index 96fad2ee..00000000 --- a/internal/document/version_test.go +++ /dev/null @@ -1,43 +0,0 @@ -package document_test - -import ( - "queryorchestration/internal/document" - "queryorchestration/internal/serviceconfig" - "testing" - - "github.com/stretchr/testify/assert" -) - -func TestIsValidCleanVersion(t *testing.T) { - cfg := &serviceconfig.BaseConfig{} - svc := document.New(cfg) - - assert.Nil(t, svc.IsValidCleanVersion(1)) - assert.Error(t, svc.IsValidCleanVersion(2)) - assert.Error(t, svc.IsValidCleanVersion(0)) - assert.Error(t, svc.IsValidCleanVersion(-1)) -} - -func TestIsValidTextVersion(t *testing.T) { - cfg := &serviceconfig.BaseConfig{} - svc := document.New(cfg) - - assert.Nil(t, svc.IsValidTextVersion(1)) - assert.Error(t, svc.IsValidTextVersion(2)) - assert.Error(t, svc.IsValidTextVersion(0)) - assert.Error(t, svc.IsValidTextVersion(-1)) -} - -func TestGetCleanVersion(t *testing.T) { - cfg := &serviceconfig.BaseConfig{} - svc := document.New(cfg) - - assert.Equal(t, int32(1), svc.GetCleanVersion()) -} - -func TestGetTextVersion(t *testing.T) { - cfg := &serviceconfig.BaseConfig{} - svc := document.New(cfg) - - assert.Equal(t, int32(1), svc.GetTextVersion()) -} diff --git a/internal/job/collector/create.go b/internal/job/collector/create.go index 9de50675..8fe4438c 100644 --- a/internal/job/collector/create.go +++ b/internal/job/collector/create.go @@ -42,7 +42,7 @@ type dbCreateParams struct { func (s *Service) getCreateParams(ctx context.Context, params *CreateParams) (*dbCreateParams, error) { minClean := params.MinCleanVersion if minClean != nil { - err := s.svc.Document.IsValidCleanVersion(*minClean) + err := s.svc.CleanVersion.IsValidVersion(*minClean) if err != nil { return nil, err } @@ -50,13 +50,13 @@ func (s *Service) getCreateParams(ctx context.Context, params *CreateParams) (*d minText := params.MinTextVersion if minText != nil { - err := s.svc.Document.IsValidTextVersion(*minText) + err := s.svc.TextVersion.IsValidVersion(*minText) if err != nil { return nil, err } } - fields, err := s.normalizeFieldsToDB(ctx, params.Fields) + fields, err := s.NormalizeFieldsToDB(ctx, params.Fields) if err != nil { return nil, err } @@ -69,7 +69,7 @@ func (s *Service) getCreateParams(ctx context.Context, params *CreateParams) (*d }, nil } -func (s *Service) normalizeFieldsToDB(ctx context.Context, ofields *map[string]uuid.UUID) (*map[string]pgtype.UUID, error) { +func (s *Service) NormalizeFieldsToDB(ctx context.Context, ofields *map[string]uuid.UUID) (*map[string]pgtype.UUID, error) { if ofields == nil || *ofields == nil { return nil, nil } else if len(*ofields) == 0 { diff --git a/internal/job/collector/createprivate_test.go b/internal/job/collector/createprivate_test.go index e3e8d3a4..d3bc6055 100644 --- a/internal/job/collector/createprivate_test.go +++ b/internal/job/collector/createprivate_test.go @@ -132,7 +132,7 @@ func TestNormalizeFieldsToDB(t *testing.T) { AddRow(true), ) - dbparams, err := svc.normalizeFieldsToDB(ctx, &fields) + dbparams, err := svc.NormalizeFieldsToDB(ctx, &fields) assert.NoError(t, err) assert.EqualExportedValues(t, &map[string]pgtype.UUID{ "example_key": database.MustToDBUUID(fields["example_key"]), @@ -145,6 +145,6 @@ func TestNormalizeFieldsToDB(t *testing.T) { AddRow(true), ) - _, err = svc.normalizeFieldsToDB(ctx, &fields) + _, err = svc.NormalizeFieldsToDB(ctx, &fields) assert.Error(t, err) } diff --git a/internal/job/collector/service.go b/internal/job/collector/service.go index cae0a169..e9ddded4 100644 --- a/internal/job/collector/service.go +++ b/internal/job/collector/service.go @@ -1,7 +1,8 @@ package collector import ( - "queryorchestration/internal/document" + cleanversion "queryorchestration/internal/document/clean/version" + textversion "queryorchestration/internal/document/text/version" "queryorchestration/internal/serviceconfig" "github.com/google/uuid" @@ -18,7 +19,8 @@ type Collector struct { } type Services struct { - Document *document.Service + CleanVersion *cleanversion.Service + TextVersion *textversion.Service } type Service struct { diff --git a/internal/job/collector/update/service.go b/internal/job/collector/update/service.go new file mode 100644 index 00000000..36e056e4 --- /dev/null +++ b/internal/job/collector/update/service.go @@ -0,0 +1,32 @@ +package collectorupdate + +import ( + cleanversion "queryorchestration/internal/document/clean/version" + textversion "queryorchestration/internal/document/text/version" + collector "queryorchestration/internal/job/collector" + "queryorchestration/internal/serviceconfig" + "queryorchestration/internal/serviceconfig/queue/jobsync" +) + +type Services struct { + CleanVersion *cleanversion.Service + TextVersion *textversion.Service + Collector *collector.Service +} + +type ConfigProvider interface { + serviceconfig.ConfigProvider + jobsync.ConfigProvider +} + +type Service struct { + cfg ConfigProvider + svc *Services +} + +func New(cfg ConfigProvider, svc *Services) *Service { + return &Service{ + cfg, + svc, + } +} diff --git a/internal/job/collector/update/service_test.go b/internal/job/collector/update/service_test.go new file mode 100644 index 00000000..09f1a4a7 --- /dev/null +++ b/internal/job/collector/update/service_test.go @@ -0,0 +1,24 @@ +package collectorupdate_test + +import ( + "queryorchestration/internal/database/repository" + "queryorchestration/internal/job/collector" + "queryorchestration/internal/serviceconfig" + "testing" + + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +func TestService(t *testing.T) { + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + cfg := &serviceconfig.BaseConfig{} + cfg.DBPool = pool + cfg.DBQueries = repository.New(pool) + + svc := collector.New(cfg, &collector.Services{}) + assert.NotNil(t, svc) +} diff --git a/internal/job/collector/update.go b/internal/job/collector/update/update.go similarity index 82% rename from internal/job/collector/update.go rename to internal/job/collector/update/update.go index 7edff377..6ddb5f35 100644 --- a/internal/job/collector/update.go +++ b/internal/job/collector/update/update.go @@ -1,11 +1,14 @@ -package collector +package collectorupdate import ( "context" "errors" "log/slog" + jobsyncrunner "queryorchestration/api/jobSyncRunner" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" + "queryorchestration/internal/job/collector" + "queryorchestration/internal/serviceconfig/queue" "queryorchestration/internal/validation" "github.com/google/uuid" @@ -21,7 +24,7 @@ type UpdateParams struct { } func (s *Service) UpdateByJobId(ctx context.Context, params *UpdateParams) error { - current, err := s.GetByJobID(ctx, params.JobID) + current, err := s.svc.Collector.GetByJobID(ctx, params.JobID) if err != nil { return err } @@ -36,9 +39,27 @@ func (s *Service) UpdateByJobId(ctx context.Context, params *UpdateParams) error return err } + err = s.informUpdate(ctx, dbparams) + if err != nil { + return err + } + return nil } +func (s *Service) informUpdate(ctx context.Context, update *dbUpdateParams) error { + if update.ActiveVersion == nil { + return nil + } + + return s.cfg.SendToQueue(ctx, &queue.SendParams{ + QueueURL: s.cfg.GetJobSyncURL(), + Body: jobsyncrunner.Body{ + ID: database.MustToUUID(update.JobID), + }, + }) +} + type dbUpdateParams struct { JobID pgtype.UUID ActiveVersion *int32 @@ -47,7 +68,7 @@ type dbUpdateParams struct { Fields *map[string]pgtype.UUID } -func (s *Service) getUpdateParams(ctx context.Context, current *Collector, params *UpdateParams) (*dbUpdateParams, error) { +func (s *Service) getUpdateParams(ctx context.Context, current *collector.Collector, params *UpdateParams) (*dbUpdateParams, error) { err := s.normalizeCodeVersions(current, params) if err != nil { return nil, err @@ -63,22 +84,6 @@ func (s *Service) getUpdateParams(ctx context.Context, current *Collector, param return nil, err } - minClean := params.MinCleanVersion - if minClean != nil { - err := s.svc.Document.IsValidCleanVersion(*minClean) - if err != nil { - return nil, err - } - } - - minText := params.MinTextVersion - if minText != nil { - err := s.svc.Document.IsValidTextVersion(*minText) - if err != nil { - return nil, err - } - } - activeVersionName, err := validation.GetFieldName(params, params.ActiveVersion) if err != nil { return nil, err @@ -97,13 +102,13 @@ func (s *Service) getUpdateParams(ctx context.Context, current *Collector, param return &dbUpdateParams{ JobID: database.MustToDBUUID(params.JobID), ActiveVersion: params.ActiveVersion, - MinCleanVersion: minClean, - MinTextVersion: minText, + MinCleanVersion: params.MinCleanVersion, + MinTextVersion: params.MinTextVersion, Fields: fields, }, nil } -func (s *Service) normalizeCodeVersions(current *Collector, params *UpdateParams) error { +func (s *Service) normalizeCodeVersions(current *collector.Collector, params *UpdateParams) error { if current == nil { return errors.New("current collector required") } @@ -122,7 +127,7 @@ func (s *Service) normalizeCodeVersions(current *Collector, params *UpdateParams if params.MinCleanVersion == nil { params.MinCleanVersion = ¤t.MinCleanVersion } else { - err := s.svc.Document.IsValidCleanVersion(*params.MinCleanVersion) + err := s.svc.CleanVersion.IsValidVersion(*params.MinCleanVersion) if err != nil { return err } @@ -131,7 +136,7 @@ func (s *Service) normalizeCodeVersions(current *Collector, params *UpdateParams if params.MinTextVersion == nil { params.MinTextVersion = ¤t.MinTextVersion } else { - err := s.svc.Document.IsValidTextVersion(*params.MinTextVersion) + err := s.svc.TextVersion.IsValidVersion(*params.MinTextVersion) if err != nil { return err } @@ -140,7 +145,7 @@ func (s *Service) normalizeCodeVersions(current *Collector, params *UpdateParams return nil } -func (s *Service) normalizeActiveVersion(current *Collector, params *UpdateParams) error { +func (s *Service) normalizeActiveVersion(current *collector.Collector, params *UpdateParams) error { if current == nil { return errors.New("current collector required") } @@ -176,10 +181,10 @@ func (s *Service) normalizeUpdateFieldsToDB(ctx context.Context, current map[str return nil, nil } - return s.normalizeFieldsToDB(ctx, ofields) + return s.svc.Collector.NormalizeFieldsToDB(ctx, ofields) } -func (s *Service) submitUpdate(ctx context.Context, current *Collector, params *dbUpdateParams) error { +func (s *Service) submitUpdate(ctx context.Context, current *collector.Collector, params *dbUpdateParams) error { err := s.cfg.ExecuteDBTransaction(ctx, func(ctx context.Context, qtx *repository.Queries) error { latestVersion := current.LatestVersion + 1 id := database.MustToDBUUID(current.ID) diff --git a/internal/job/collector/update_test.go b/internal/job/collector/update/update_test.go similarity index 50% rename from internal/job/collector/update_test.go rename to internal/job/collector/update/update_test.go index 534759b4..068b55a7 100644 --- a/internal/job/collector/update_test.go +++ b/internal/job/collector/update/update_test.go @@ -1,20 +1,30 @@ -package collector_test +package collectorupdate_test import ( "context" + "fmt" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" - "queryorchestration/internal/document" "queryorchestration/internal/job/collector" + collectorupdate "queryorchestration/internal/job/collector/update" "queryorchestration/internal/serviceconfig" + "queryorchestration/internal/serviceconfig/queue/jobsync" + queuemock "queryorchestration/mocks/queue" "testing" + "github.com/aws/aws-sdk-go-v2/service/sqs" "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/pashagolub/pgxmock/v3" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" ) +type CollectorUpdateConfig struct { + serviceconfig.BaseConfig + jobsync.JobSyncConfig +} + func TestUpdate(t *testing.T) { ctx := context.Background() @@ -22,12 +32,15 @@ func TestUpdate(t *testing.T) { if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - cfg := &serviceconfig.BaseConfig{} + cfg := &CollectorUpdateConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) + mockSQS := queuemock.NewMockSQSClient(t) + cfg.QueueClient = mockSQS + cfg.JobSyncURL = "here" - svc := collector.New(cfg, &collector.Services{ - Document: document.New(cfg), + svc := collectorupdate.New(cfg, &collectorupdate.Services{ + Collector: collector.New(cfg, &collector.Services{}), }) current := collector.Collector{ @@ -37,9 +50,11 @@ func TestUpdate(t *testing.T) { LatestVersion: 4, } av := int32(2) - update := collector.UpdateParams{ - JobID: current.JobID, - ActiveVersion: &av, + mv := int32(1) + update := collectorupdate.UpdateParams{ + JobID: current.JobID, + ActiveVersion: &av, + MinCleanVersion: &mv, } pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(database.MustToDBUUID(update.JobID)). @@ -48,10 +63,25 @@ func TestUpdate(t *testing.T) { AddRow(database.MustToDBUUID(current.ID), database.MustToDBUUID(current.JobID), current.MinCleanVersion, current.MinTextVersion, current.ActiveVersion, current.LatestVersion, []byte("")), ) pool.ExpectBeginTx(pgx.TxOptions{}) + rv := current.LatestVersion + 1 + pool.ExpectExec("name: RemoveCollectorCodeVersion :exec").WithArgs(&rv, database.MustToDBUUID(current.ID)). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectExec("name: AddCollectorCodeVersion :exec").WithArgs(database.MustToDBUUID(current.ID), int32(5), *update.MinCleanVersion, int32(0)). + WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectExec("name: UpdateCollector :exec").WithArgs(int32(5), int32(2), database.MustToDBUUID(current.ID)). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectCommit() + mockSQS.EXPECT(). + SendMessage( + mock.Anything, + mock.MatchedBy(func(in *sqs.SendMessageInput) bool { + return *in.QueueUrl == cfg.JobSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", update.JobID.String()) + }), + mock.Anything, + ). + Return(&sqs.SendMessageOutput{}, nil) + err = svc.UpdateByJobId(ctx, &update) assert.NoError(t, err) } diff --git a/internal/job/collector/updateprivate_test.go b/internal/job/collector/update/updateprivate_test.go similarity index 67% rename from internal/job/collector/updateprivate_test.go rename to internal/job/collector/update/updateprivate_test.go index f994bfaa..1e5e0975 100644 --- a/internal/job/collector/updateprivate_test.go +++ b/internal/job/collector/update/updateprivate_test.go @@ -1,20 +1,32 @@ -package collector +package collectorupdate import ( "context" + "fmt" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" - "queryorchestration/internal/document" + cleanversion "queryorchestration/internal/document/clean/version" + textversion "queryorchestration/internal/document/text/version" + "queryorchestration/internal/job/collector" "queryorchestration/internal/serviceconfig" + "queryorchestration/internal/serviceconfig/queue/jobsync" + queuemock "queryorchestration/mocks/queue" "testing" + "github.com/aws/aws-sdk-go-v2/service/sqs" "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" "github.com/pashagolub/pgxmock/v3" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" ) +type CollectorUpdateConfig struct { + serviceconfig.BaseConfig + jobsync.JobSyncConfig +} + func TestGetUpdateParams(t *testing.T) { t.Run("all params", func(t *testing.T) { ctx := context.Background() @@ -23,19 +35,21 @@ func TestGetUpdateParams(t *testing.T) { if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - cfg := &serviceconfig.BaseConfig{} + cfg := &CollectorUpdateConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) svc := Service{ cfg: cfg, - svc: &Services{}, + svc: &Services{ + Collector: collector.New(cfg, &collector.Services{}), + }, } minCleanV := int32(1) minTextV := int32(1) aV := int32(3) - current := Collector{ + current := collector.Collector{ ActiveVersion: 1, LatestVersion: 10, } @@ -79,7 +93,7 @@ func TestGetUpdateParams(t *testing.T) { if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - cfg := &serviceconfig.BaseConfig{} + cfg := &CollectorUpdateConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) @@ -88,7 +102,7 @@ func TestGetUpdateParams(t *testing.T) { svc: &Services{}, } - current := Collector{} + current := collector.Collector{} params := UpdateParams{ JobID: current.JobID, } @@ -103,7 +117,7 @@ func TestGetUpdateParams(t *testing.T) { if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - cfg := &serviceconfig.BaseConfig{} + cfg := &CollectorUpdateConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) @@ -112,7 +126,7 @@ func TestGetUpdateParams(t *testing.T) { svc: &Services{}, } - current := Collector{ + current := collector.Collector{ ActiveVersion: 2, LatestVersion: 4, } @@ -132,7 +146,7 @@ func TestGetUpdateParams(t *testing.T) { if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - cfg := &serviceconfig.BaseConfig{} + cfg := &CollectorUpdateConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) @@ -141,7 +155,7 @@ func TestGetUpdateParams(t *testing.T) { svc: &Services{}, } - current := Collector{ + current := collector.Collector{ Fields: map[string]uuid.UUID{ "example": uuid.New(), }, @@ -164,7 +178,7 @@ func TestSubmitUpdate(t *testing.T) { if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - cfg := &serviceconfig.BaseConfig{} + cfg := &CollectorUpdateConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) @@ -175,7 +189,7 @@ func TestSubmitUpdate(t *testing.T) { minCleanV := int32(2) minTextV := int32(4) - current := Collector{ + current := collector.Collector{ ID: uuid.New(), JobID: uuid.New(), ActiveVersion: 1, @@ -233,7 +247,7 @@ func TestNormalizeActiveVersion(t *testing.T) { }) t.Run("nil update", func(t *testing.T) { - current := Collector{ + current := collector.Collector{ ActiveVersion: 1, LatestVersion: 4, } @@ -243,7 +257,7 @@ func TestNormalizeActiveVersion(t *testing.T) { }) t.Run("no update", func(t *testing.T) { - current := Collector{ + current := collector.Collector{ ActiveVersion: 1, LatestVersion: 4, } @@ -255,7 +269,7 @@ func TestNormalizeActiveVersion(t *testing.T) { t.Run("same version", func(t *testing.T) { version := int32(2) - current := Collector{ + current := collector.Collector{ ActiveVersion: version, LatestVersion: 4, } @@ -268,7 +282,7 @@ func TestNormalizeActiveVersion(t *testing.T) { }) t.Run("latest version plus 1", func(t *testing.T) { - current := Collector{ + current := collector.Collector{ ActiveVersion: 1, LatestVersion: 4, } @@ -282,7 +296,7 @@ func TestNormalizeActiveVersion(t *testing.T) { }) t.Run("latest version plus 2", func(t *testing.T) { - current := Collector{ + current := collector.Collector{ ActiveVersion: 1, LatestVersion: 4, } @@ -296,71 +310,111 @@ func TestNormalizeActiveVersion(t *testing.T) { } func TestNormalizeCodeVersions(t *testing.T) { - cfg := &serviceconfig.BaseConfig{} + cfg := &CollectorUpdateConfig{} svc := Service{ svc: &Services{ - Document: document.New(cfg), + CleanVersion: cleanversion.New(cfg), + TextVersion: textversion.New(cfg), }, } - err := svc.normalizeCodeVersions(nil, nil) - assert.Error(t, err) + t.Run("all nil", func(t *testing.T) { + err := svc.normalizeCodeVersions(nil, nil) + assert.Error(t, err) + }) - current := Collector{} + t.Run("nil update", func(t *testing.T) { + current := collector.Collector{} + err := svc.normalizeCodeVersions(¤t, nil) + assert.NoError(t, err) + }) - err = svc.normalizeCodeVersions(¤t, nil) - assert.NoError(t, err) + t.Run("empty", func(t *testing.T) { + current := collector.Collector{} + update := UpdateParams{} - update := UpdateParams{} + err := svc.normalizeCodeVersions(¤t, &update) + assert.NoError(t, err) + assert.Nil(t, update.MinCleanVersion) + assert.Nil(t, update.MinTextVersion) + }) - err = svc.normalizeCodeVersions(¤t, &update) - assert.NoError(t, err) - assert.Nil(t, update.MinCleanVersion) - assert.Nil(t, update.MinTextVersion) + t.Run("valid clean", func(t *testing.T) { + current := collector.Collector{} + update := UpdateParams{} - cv := int32(1) - update.MinCleanVersion = &cv - update.MinTextVersion = nil - err = svc.normalizeCodeVersions(¤t, &update) - assert.NoError(t, err) - assert.Equal(t, cv, *update.MinCleanVersion) - assert.Equal(t, int32(0), *update.MinTextVersion) + cv := int32(1) + update.MinCleanVersion = &cv + err := svc.normalizeCodeVersions(¤t, &update) + assert.NoError(t, err) + assert.Equal(t, cv, *update.MinCleanVersion) + assert.Equal(t, int32(0), *update.MinTextVersion) + }) - update.MinCleanVersion = nil - tv := int32(1) - update.MinTextVersion = &tv - err = svc.normalizeCodeVersions(¤t, &update) - assert.NoError(t, err) - assert.Equal(t, int32(0), *update.MinCleanVersion) - assert.Equal(t, tv, *update.MinTextVersion) + t.Run("valid text", func(t *testing.T) { + current := collector.Collector{} + update := UpdateParams{} - current.MinCleanVersion = 1 - current.MinTextVersion = 1 - err = svc.normalizeCodeVersions(¤t, &update) - assert.NoError(t, err) - assert.Nil(t, update.MinCleanVersion) - assert.Nil(t, update.MinTextVersion) + tv := int32(1) + update.MinTextVersion = &tv + err := svc.normalizeCodeVersions(¤t, &update) + assert.NoError(t, err) + assert.Equal(t, int32(0), *update.MinCleanVersion) + assert.Equal(t, tv, *update.MinTextVersion) + }) - update.MinCleanVersion = ¤t.MinCleanVersion - update.MinTextVersion = nil - err = svc.normalizeCodeVersions(¤t, &update) - assert.NoError(t, err) - assert.Nil(t, update.MinCleanVersion) - assert.Nil(t, update.MinTextVersion) + t.Run("valid clean and text", func(t *testing.T) { + current := collector.Collector{} + update := UpdateParams{} - update.MinCleanVersion = nil - update.MinTextVersion = ¤t.MinTextVersion - err = svc.normalizeCodeVersions(¤t, &update) - assert.NoError(t, err) - assert.Nil(t, update.MinCleanVersion) - assert.Nil(t, update.MinTextVersion) + current.MinCleanVersion = 1 + current.MinTextVersion = 1 + err := svc.normalizeCodeVersions(¤t, &update) + assert.NoError(t, err) + assert.Nil(t, update.MinCleanVersion) + assert.Nil(t, update.MinTextVersion) + }) - update.MinCleanVersion = ¤t.MinCleanVersion - update.MinTextVersion = ¤t.MinTextVersion - err = svc.normalizeCodeVersions(¤t, &update) - assert.NoError(t, err) - assert.Nil(t, update.MinCleanVersion) - assert.Nil(t, update.MinTextVersion) + t.Run("current clean", func(t *testing.T) { + current := collector.Collector{ + MinCleanVersion: 1, + } + update := UpdateParams{} + + update.MinCleanVersion = ¤t.MinCleanVersion + err := svc.normalizeCodeVersions(¤t, &update) + assert.NoError(t, err) + assert.Nil(t, update.MinCleanVersion) + assert.Nil(t, update.MinTextVersion) + }) + + t.Run("current text", func(t *testing.T) { + current := collector.Collector{ + MinTextVersion: 1, + } + update := UpdateParams{} + + update.MinTextVersion = ¤t.MinTextVersion + err := svc.normalizeCodeVersions(¤t, &update) + assert.NoError(t, err) + assert.Nil(t, update.MinCleanVersion) + assert.Nil(t, update.MinTextVersion) + }) + + t.Run("current clean and text", func(t *testing.T) { + current := collector.Collector{ + MinCleanVersion: 1, + MinTextVersion: 1, + } + update := UpdateParams{} + + update.MinCleanVersion = ¤t.MinCleanVersion + update.MinTextVersion = ¤t.MinTextVersion + err := svc.normalizeCodeVersions(¤t, &update) + assert.NoError(t, err) + assert.Nil(t, update.MinCleanVersion) + assert.Nil(t, update.MinTextVersion) + }) } func TestGetRemoveFields(t *testing.T) { @@ -447,13 +501,15 @@ func TestNormalizeUpdateFieldsToDB(t *testing.T) { if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - cfg := &serviceconfig.BaseConfig{} + cfg := &CollectorUpdateConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) svc := Service{ cfg: cfg, - svc: &Services{}, + svc: &Services{ + Collector: collector.New(cfg, &collector.Services{}), + }, } current := map[string]uuid.UUID{} @@ -480,7 +536,7 @@ func TestNormalizeUpdateFieldsToDB(t *testing.T) { if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - cfg := &serviceconfig.BaseConfig{} + cfg := &CollectorUpdateConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) @@ -505,11 +561,13 @@ func TestNormalizeUpdateFieldsToDB(t *testing.T) { }) t.Run("no changes", func(t *testing.T) { ctx := context.Background() - cfg := &serviceconfig.BaseConfig{} + cfg := &CollectorUpdateConfig{} svc := Service{ cfg: cfg, - svc: &Services{}, + svc: &Services{ + Collector: collector.New(cfg, &collector.Services{}), + }, } current := map[string]uuid.UUID{ @@ -524,3 +582,64 @@ func TestNormalizeUpdateFieldsToDB(t *testing.T) { assert.Nil(t, val) }) } + +func TestInformUpdate(t *testing.T) { + t.Run("with version update", func(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + cfg := &CollectorUpdateConfig{} + cfg.DBPool = pool + cfg.DBQueries = repository.New(pool) + mockSQS := queuemock.NewMockSQSClient(t) + cfg.QueueClient = mockSQS + cfg.JobSyncURL = "here" + + svc := New(cfg, &Services{}) + + av := int32(2) + update := dbUpdateParams{ + JobID: database.MustToDBUUID(uuid.New()), + ActiveVersion: &av, + } + + mockSQS.EXPECT(). + SendMessage( + mock.Anything, + mock.MatchedBy(func(in *sqs.SendMessageInput) bool { + return *in.QueueUrl == cfg.JobSyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", update.JobID.String()) + }), + mock.Anything, + ). + Return(&sqs.SendMessageOutput{}, nil) + + err = svc.informUpdate(ctx, &update) + assert.NoError(t, err) + }) + t.Run("with no update", func(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + cfg := &CollectorUpdateConfig{} + cfg.DBPool = pool + cfg.DBQueries = repository.New(pool) + mockSQS := queuemock.NewMockSQSClient(t) + cfg.QueueClient = mockSQS + cfg.JobSyncURL = "here" + + svc := New(cfg, &Services{}) + + update := dbUpdateParams{ + JobID: database.MustToDBUUID(uuid.New()), + } + + err = svc.informUpdate(ctx, &update) + assert.NoError(t, err) + }) +} diff --git a/internal/job/sync/service.go b/internal/job/sync/service.go new file mode 100644 index 00000000..e1be4b0d --- /dev/null +++ b/internal/job/sync/service.go @@ -0,0 +1,23 @@ +package jobsync + +import ( + "queryorchestration/internal/serviceconfig" + "queryorchestration/internal/serviceconfig/queue/documentsync" +) + +type Service struct { + cfg ConfigProvider + batchSize int32 +} + +type ConfigProvider interface { + serviceconfig.ConfigProvider + documentsync.ConfigProvider +} + +func New(cfg ConfigProvider) *Service { + return &Service{ + cfg: cfg, + batchSize: 100, + } +} diff --git a/internal/job/sync/service_test.go b/internal/job/sync/service_test.go new file mode 100644 index 00000000..b85da688 --- /dev/null +++ b/internal/job/sync/service_test.go @@ -0,0 +1,13 @@ +package jobsync + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestService(t *testing.T) { + svc := New(nil) + assert.NotNil(t, svc) + assert.Equal(t, svc.batchSize, int32(100)) +} diff --git a/internal/job/sync/sync.go b/internal/job/sync/sync.go new file mode 100644 index 00000000..88cc0fda --- /dev/null +++ b/internal/job/sync/sync.go @@ -0,0 +1,51 @@ +package jobsync + +import ( + "context" + "database/sql" + "errors" + docsyncrunner "queryorchestration/api/docSyncRunner" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/serviceconfig/queue" + + "github.com/google/uuid" +) + +func (s *Service) Sync(ctx context.Context, id uuid.UUID) error { + hasMore := true + offset := int32(0) + dbid := database.MustToDBUUID(id) + + for hasMore { + ids, err := s.cfg.GetDBQueries().ListJobDocumentIDsBatch(ctx, &repository.ListJobDocumentIDsBatchParams{ + Batchsize: s.batchSize, + Pageoffset: offset, + Jobid: dbid, + }) + if errors.Is(err, sql.ErrNoRows) { + return nil + } else if err != nil { + return err + } + + for _, id := range ids { + err := s.cfg.SendToQueue(ctx, &queue.SendParams{ + QueueURL: s.cfg.GetDocumentSyncURL(), + Body: docsyncrunner.Body{ + ID: database.MustToUUID(id.ID), + }, + }) + if err != nil { + return err + } + } + + offset += s.batchSize + if int64(offset) >= *ids[0].Totalcount { + hasMore = false + } + } + + return nil +} diff --git a/internal/job/sync/sync_test.go b/internal/job/sync/sync_test.go new file mode 100644 index 00000000..2e5ad64c --- /dev/null +++ b/internal/job/sync/sync_test.go @@ -0,0 +1,88 @@ +package jobsync + +import ( + "context" + "fmt" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/document" + "queryorchestration/internal/serviceconfig" + "queryorchestration/internal/serviceconfig/queue/documentsync" + queuemock "queryorchestration/mocks/queue" + "testing" + + "github.com/aws/aws-sdk-go-v2/service/sqs" + "github.com/google/uuid" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" +) + +type JobSyncConfig struct { + serviceconfig.BaseConfig + documentsync.DocSyncConfig +} + +func TestTrigger(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() + 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) + + err = svc.Sync(ctx, jobId) + assert.NoError(t, err) +} diff --git a/internal/query/test/test_test.go b/internal/query/test/test_test.go index 4b9abcb1..f1588a77 100644 --- a/internal/query/test/test_test.go +++ b/internal/query/test/test_test.go @@ -5,6 +5,8 @@ import ( "queryorchestration/internal/database" "queryorchestration/internal/database/repository" "queryorchestration/internal/document" + cleanversion "queryorchestration/internal/document/clean/version" + textversion "queryorchestration/internal/document/text/version" "queryorchestration/internal/job/collector" "queryorchestration/internal/query" "queryorchestration/internal/query/result" @@ -28,10 +30,11 @@ func TestTest(t *testing.T) { cfg := &serviceconfig.BaseConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) - docsvc := document.New(cfg) col := collector.New(cfg, &collector.Services{ - Document: docsvc, + TextVersion: textversion.New(cfg), + CleanVersion: cleanversion.New(cfg), }) + docsvc := document.New(cfg) svc := querytest.New(cfg, &querytest.Services{ Document: docsvc, Collector: col, @@ -45,7 +48,7 @@ func TestTest(t *testing.T) { JobID: uuid.New(), } doc := document.Document{ - ID: uuid.New(), + ID: coll.JobID, JobID: coll.JobID, Hash: "example_hash", } diff --git a/internal/server/runner/listener.go b/internal/server/runner/listener.go index 05c5f3ee..0d7fff6d 100644 --- a/internal/server/runner/listener.go +++ b/internal/server/runner/listener.go @@ -57,11 +57,6 @@ func New(ctx context.Context, cfg ListenerConfig) (*Server, error) { return nil, err } - err = cfg.SetQueueClient(ctx) - if err != nil { - return nil, err - } - err = cfg.PingQueue(ctx) if err != nil { return nil, err diff --git a/internal/server/server.go b/internal/server/server.go index 6040e1c6..7e4bf1c4 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -52,5 +52,10 @@ func New(ctx context.Context, cfg Config) (func() error, error) { return func() error { return nil }, err } + err = cfg.SetQueueClient(ctx) + if err != nil { + return nil, err + } + return closeTracer, nil } diff --git a/internal/serviceconfig/queue/documentinit/config.go b/internal/serviceconfig/queue/documentinit/config.go new file mode 100644 index 00000000..e81cede9 --- /dev/null +++ b/internal/serviceconfig/queue/documentinit/config.go @@ -0,0 +1,13 @@ +package documentinit + +type DocInitConfig struct { + DocInitURL string `env:"DOCUMENT_INIT_URL,required,notEmpty"` +} + +func (c *DocInitConfig) GetDocInitURL() string { + return c.DocInitURL +} + +type ConfigProvider interface { + GetDocInitURL() string +} diff --git a/internal/serviceconfig/queue/documentinit/config_test.go b/internal/serviceconfig/queue/documentinit/config_test.go new file mode 100644 index 00000000..9cdac18a --- /dev/null +++ b/internal/serviceconfig/queue/documentinit/config_test.go @@ -0,0 +1,20 @@ +package documentinit_test + +import ( + "queryorchestration/internal/serviceconfig/queue/documentinit" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestGetDocInitURL(t *testing.T) { + cfg := documentinit.DocInitConfig{} + + name := cfg.GetDocInitURL() + assert.Equal(t, "", name) + + cfg.DocInitURL = "name" + name = cfg.GetDocInitURL() + assert.Equal(t, "name", name) + assert.Equal(t, cfg.DocInitURL, name) +} diff --git a/internal/serviceconfig/queue/documentsync/config.go b/internal/serviceconfig/queue/documentsync/config.go new file mode 100644 index 00000000..8f804538 --- /dev/null +++ b/internal/serviceconfig/queue/documentsync/config.go @@ -0,0 +1,13 @@ +package documentsync + +type DocSyncConfig struct { + DocumentSyncURL string `env:"DOCUMENT_SYNC_URL,required,notEmpty"` +} + +func (c *DocSyncConfig) GetDocumentSyncURL() string { + return c.DocumentSyncURL +} + +type ConfigProvider interface { + GetDocumentSyncURL() string +} diff --git a/internal/serviceconfig/queue/documentsync/config_test.go b/internal/serviceconfig/queue/documentsync/config_test.go new file mode 100644 index 00000000..fc732f76 --- /dev/null +++ b/internal/serviceconfig/queue/documentsync/config_test.go @@ -0,0 +1,20 @@ +package documentsync_test + +import ( + "queryorchestration/internal/serviceconfig/queue/documentsync" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestGetDocumentSyncURL(t *testing.T) { + cfg := documentsync.DocSyncConfig{} + + name := cfg.GetDocumentSyncURL() + assert.Equal(t, "", name) + + cfg.DocumentSyncURL = "name" + name = cfg.GetDocumentSyncURL() + assert.Equal(t, "name", name) + assert.Equal(t, cfg.DocumentSyncURL, name) +} diff --git a/internal/serviceconfig/queue/jobsync/config.go b/internal/serviceconfig/queue/jobsync/config.go new file mode 100644 index 00000000..c9f2ffa5 --- /dev/null +++ b/internal/serviceconfig/queue/jobsync/config.go @@ -0,0 +1,13 @@ +package jobsync + +type JobSyncConfig struct { + JobSyncURL string `env:"JOB_SYNC_URL,required,notEmpty"` +} + +func (c *JobSyncConfig) GetJobSyncURL() string { + return c.JobSyncURL +} + +type ConfigProvider interface { + GetJobSyncURL() string +} diff --git a/internal/serviceconfig/queue/jobsync/config_test.go b/internal/serviceconfig/queue/jobsync/config_test.go new file mode 100644 index 00000000..3be9f37c --- /dev/null +++ b/internal/serviceconfig/queue/jobsync/config_test.go @@ -0,0 +1,20 @@ +package jobsync_test + +import ( + "queryorchestration/internal/serviceconfig/queue/jobsync" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestGetJobSyncURL(t *testing.T) { + cfg := jobsync.JobSyncConfig{} + + name := cfg.GetJobSyncURL() + assert.Equal(t, "", name) + + cfg.JobSyncURL = "name" + name = cfg.GetJobSyncURL() + assert.Equal(t, "name", name) + assert.Equal(t, cfg.JobSyncURL, name) +} diff --git a/internal/test/ecosystem.go b/internal/test/ecosystem.go index 7b3da90d..c203f698 100644 --- a/internal/test/ecosystem.go +++ b/internal/test/ecosystem.go @@ -125,32 +125,42 @@ func SetCfgProvider(t *testing.T, cfg serviceconfig.ConfigProvider) { } type ServiceNetworkConfig struct { - Name Runner - Env map[string]string + Cfg serviceconfig.ConfigProvider + Network *testcontainers.DockerNetwork + Name Runner + Env map[string]string } func CreateServiceNetwork(t *testing.T, ctx context.Context, scfg *ServiceNetworkConfig) (*Container, func()) { - cfg := &serviceconfig.BaseConfig{} - SetCfgProvider(t, cfg) + if scfg.Cfg == nil { + scfg.Cfg = &serviceconfig.BaseConfig{} + SetCfgProvider(t, scfg.Cfg) + } - network, ncleanup := CreateNetwork(t, ctx) + network := scfg.Network + var ncleanup func() + if network == nil { + network, ncleanup = CreateNetwork(t, ctx) + } _, dbcleanup := CreateDB(t, ctx, &CreateDatabaseConfig{ Network: network, - Cfg: cfg, + Cfg: scfg.Cfg, }) c, ccleanup := CreateService(t, ctx, &ServiceConfig{ Name: scfg.Name, Env: scfg.Env, - Cfg: cfg, + Cfg: scfg.Cfg, Network: network, }) return c, func() { dbcleanup() ccleanup() - ncleanup() + if ncleanup != nil { + ncleanup() + } } } diff --git a/internal/test/ecosystem_test.go b/internal/test/ecosystem_test.go index 03d6786b..e3bd9dfe 100644 --- a/internal/test/ecosystem_test.go +++ b/internal/test/ecosystem_test.go @@ -40,6 +40,9 @@ func TestCreateServiceNetwork(t *testing.T) { conn, cleanup := CreateServiceNetwork(t, ctx, &ServiceNetworkConfig{ Name: QueryService, + Env: map[string]string{ + "JOB_SYNC_URL": "/i/am/here", + }, }) assert.NotNil(t, conn) @@ -68,7 +71,12 @@ func TestCreateRunnersAndServicesNetwork(t *testing.T) { }, }, Services: []*ServiceNetworkConfig{ - {Name: QueryService}, + { + Name: QueryService, + Env: map[string]string{ + "JOB_SYNC_URL": "/i/am/here", + }, + }, }, }) diff --git a/internal/test/runner.go b/internal/test/runner.go index 1682b45b..14baa7c9 100644 --- a/internal/test/runner.go +++ b/internal/test/runner.go @@ -4,7 +4,9 @@ import ( "context" doccleanrunner "queryorchestration/api/docCleanRunner" docinitrunner "queryorchestration/api/docInitRunner" + docsyncrunner "queryorchestration/api/docSyncRunner" doctextrunner "queryorchestration/api/docTextRunner" + jobsyncrunner "queryorchestration/api/jobSyncRunner" queryrunner "queryorchestration/api/queryRunner" "testing" @@ -17,10 +19,12 @@ type Runner = string const ( DocInitRunner = docinitrunner.Name + DocSyncRunner = docsyncrunner.Name DocCleanRunner = doccleanrunner.Name DocTextRunner = doctextrunner.Name QuerySyncRunner = queryrunner.Name QueryRunner = queryrunner.Name + JobSyncRunner = jobsyncrunner.Name ) type RunnerConfig struct { diff --git a/internal/test/service_test.go b/internal/test/service_test.go index 59b9bd30..c8d3583f 100644 --- a/internal/test/service_test.go +++ b/internal/test/service_test.go @@ -33,6 +33,9 @@ func TestCreateService(t *testing.T) { Name: test.QueryService, Cfg: cfg, Network: ncfg, + Env: map[string]string{ + "JOB_SYNC_URL": "/i/am/here", + }, } conn, cleanup := test.CreateService(t, ctx, acfg) diff --git a/scripts/Taskfile.yml b/scripts/Taskfile.yml index 6dc53a8a..6176d456 100644 --- a/scripts/Taskfile.yml +++ b/scripts/Taskfile.yml @@ -39,7 +39,7 @@ tasks: - task: build:nogen - task: lint:nogen - task: test:unit:nogen - - task: test:integration:cmd:nocache + - task: test:endtoend:cmd:nocache build: deps: - generate diff --git a/scripts/database.yml b/scripts/database.yml index 79665660..99e2c4b9 100644 --- a/scripts/database.yml +++ b/scripts/database.yml @@ -12,6 +12,10 @@ tasks: - mig cmds: - sqlc generate --file sqlc.yml + regenerate: + cmds: + - task compose:clean:generate + - task db:generate deps:up: internal: true cmds: @@ -32,4 +36,4 @@ tasks: deps: - deps:up cmds: - - migrate -path {{.MIGRATIONS}} -database {{.DB_URI_TEST}} up + - migrate -path {{.MIGRATIONS}} -database {{.DB_URI_GENERATE}} up diff --git a/scripts/local-deployments.yml b/scripts/local-deployments.yml index 328b7d3b..dd625262 100644 --- a/scripts/local-deployments.yml +++ b/scripts/local-deployments.yml @@ -47,10 +47,12 @@ tasks: cmds: - aws s3 mb s3://$BUCKET_IN - aws sqs create-queue --queue-name $QNAME_DOCUMENT_INIT + - aws sqs create-queue --queue-name $QNAME_DOCUMENT_SYNC - aws sqs create-queue --queue-name $QNAME_DOCUMENT_CLEAN - aws sqs create-queue --queue-name $QNAME_DOCUMENT_TEXT - aws sqs create-queue --queue-name $QNAME_QUERY_SYNC - aws sqs create-queue --queue-name $QNAME_QUERY_RUNNER + - aws sqs create-queue --queue-name $QNAME_JOB_SYNC - | aws s3api put-bucket-notification-configuration \ --bucket $BUCKET_IN \ diff --git a/scripts/tests.yml b/scripts/tests.yml index 54d95552..6759850e 100644 --- a/scripts/tests.yml +++ b/scripts/tests.yml @@ -8,6 +8,7 @@ vars: INTERNAL: "./internal/..." API: "./api/..." PKG: "./pkg/..." + CMD: "./cmd/..." tasks: mocks:generate: @@ -22,7 +23,7 @@ tasks: deps: - mocks:generate cmds: - - go test -short {{.INTERNAL}} {{.API}} {{.PKG}} + - go test -short {{.INTERNAL}} {{.API}} {{.PKG}} {{.CMD}} unit:nogen: cmds: - task: unit:cmd @@ -32,8 +33,8 @@ tasks: cmds: - mkdir -p {{.OUT_DIR}} - | - go test {{.INTERNAL}} {{.API}} {{.PKG}} \ - -coverpkg={{.INTERNAL}},{{.API}},{{.PKG}} \ + go test {{.INTERNAL}} {{.API}} \ + -coverpkg={{.INTERNAL}},{{.API}} \ -coverprofile={{.COVERAGE_FILE}} unit:nocoverage: deps: @@ -59,14 +60,14 @@ tasks: exit 1 fi silent: true - integration: + endtoend: cmds: - task generate - task docker:build - - task: integration:cmd:nocache - integration:cmd:nocache: + - task: endtoend:cmd:nocache + endtoend:cmd:nocache: cmds: - go test -count=1 -v ./test/... - integration:cmd:nobuild: + endtoend:cmd:nobuild: cmds: - go test -v ./test/... diff --git a/sqlc.yml b/sqlc.yml index e1a6712e..580440a5 100644 --- a/sqlc.yml +++ b/sqlc.yml @@ -2,7 +2,7 @@ version: "2" servers: - engine: postgresql - uri: "${DB_URI_TEST}" + uri: "${DB_URI_GENERATE}" sql: - name: "db" engine: "postgresql" diff --git a/test/process_test.go b/test/process_test.go index d483f3ae..456b09dc 100644 --- a/test/process_test.go +++ b/test/process_test.go @@ -1,4 +1,4 @@ -package integration_test +package endtoend_test import ( "context" @@ -64,10 +64,12 @@ func TestProcess(t *testing.T) { }) assert.NoError(t, err) + docsyncurl := test.CreateQueue(t, ctx, cfg, test.DocSyncRunner) doccleanurl := test.CreateQueue(t, ctx, cfg, test.DocCleanRunner) doctexturl := test.CreateQueue(t, ctx, cfg, test.DocTextRunner) querysyncurl := test.CreateQueue(t, ctx, cfg, test.QuerySyncRunner) queryurl := test.CreateQueue(t, ctx, cfg, test.QueryRunner) + jobsyncurl := test.CreateQueue(t, ctx, cfg, test.JobSyncRunner) net, cleanup := test.CreateRunnersAndServicesNetwork(t, ctx, &test.EcosystemNetworkConfig{ Cfg: cfg, @@ -75,6 +77,12 @@ func TestProcess(t *testing.T) { Runners: []*test.RunnerNetworkConfig{ { Name: test.DocInitRunner, + Env: map[string]string{ + "DOCUMENT_SYNC_URL": docsyncurl, + }, + }, + { + Name: test.DocSyncRunner, Env: map[string]string{ "DOCUMENT_CLEAN_URL": doccleanurl, }, @@ -103,10 +111,19 @@ func TestProcess(t *testing.T) { "QUERY_URL": queryurl, }, }, + { + Name: test.JobSyncRunner, + Env: map[string]string{ + "DOCUMENT_SYNC_URL": docsyncurl, + }, + }, }, Services: []*test.ServiceNetworkConfig{ { Name: test.QueryService, + Env: map[string]string{ + "JOB_SYNC_URL": jobsyncurl, + }, }, }, }) diff --git a/test/queryService/client_test.go b/test/queryService/client_test.go index 447705bf..9027994a 100644 --- a/test/queryService/client_test.go +++ b/test/queryService/client_test.go @@ -1,4 +1,4 @@ -package integration_test +package endtoend import ( "context" @@ -14,6 +14,9 @@ func TestClient(t *testing.T) { c, cleanup := test.CreateServiceNetwork(t, ctx, &test.ServiceNetworkConfig{ Name: test.QueryService, + Env: map[string]string{ + "JOB_SYNC_URL": "/i/am/here", + }, }) defer cleanup() diff --git a/test/queryService/exportservice_test.go b/test/queryService/exportservice_test.go index 8948de4a..34ccb15f 100644 --- a/test/queryService/exportservice_test.go +++ b/test/queryService/exportservice_test.go @@ -1,4 +1,4 @@ -package integration_test +package endtoend import ( "context" @@ -14,6 +14,9 @@ func TestExportService(t *testing.T) { c, cleanup := test.CreateServiceNetwork(t, ctx, &test.ServiceNetworkConfig{ Name: test.QueryService, + Env: map[string]string{ + "JOB_SYNC_URL": "/i/am/here", + }, }) defer cleanup() diff --git a/test/queryService/job_test.go b/test/queryService/job_test.go index 79bdf692..8d026b9b 100644 --- a/test/queryService/job_test.go +++ b/test/queryService/job_test.go @@ -1,4 +1,4 @@ -package integration_test +package endtoend import ( "context" @@ -14,6 +14,9 @@ func TestJob(t *testing.T) { c, cleanup := test.CreateServiceNetwork(t, ctx, &test.ServiceNetworkConfig{ Name: test.QueryService, + Env: map[string]string{ + "JOB_SYNC_URL": "/i/am/here", + }, }) defer cleanup() diff --git a/test/queryService/jobcollectorservice_test.go b/test/queryService/jobcollectorservice_test.go index 604154e7..31fc4673 100644 --- a/test/queryService/jobcollectorservice_test.go +++ b/test/queryService/jobcollectorservice_test.go @@ -1,21 +1,53 @@ -package integration_test +package endtoend import ( "context" + "log" + "os" + "path" + "queryorchestration/internal/serviceconfig" + "queryorchestration/internal/serviceconfig/queue" "queryorchestration/internal/test" queryservice "queryorchestration/pkg/queryService" + "regexp" "testing" "github.com/oapi-codegen/runtime/types" "github.com/stretchr/testify/assert" ) +type CollectorConfig struct { + serviceconfig.BaseConfig + queue.QueueConfig +} + func TestJobCollectorService(t *testing.T) { - t.SkipNow() ctx := context.Background() + cfg := &CollectorConfig{} + test.SetCfgProvider(t, cfg) + cfg.SetBasePath(path.Join(os.Getenv("PWD"), "../..")) + + network, ncleanup := test.CreateNetwork(t, ctx) + defer ncleanup() + + _, clean := test.CreateAWSContainer(t, ctx, &test.CreateAWSConfig{ + Cfg: cfg, + Network: network, + }) + defer clean() + + err := cfg.SetQueueClient(ctx) + assert.NoError(t, err) + jobsyncurl := test.CreateQueue(t, ctx, cfg, test.JobSyncRunner) + c, cleanup := test.CreateServiceNetwork(t, ctx, &test.ServiceNetworkConfig{ - Name: test.QueryService, + Cfg: cfg, + Network: network, + Name: test.QueryService, + Env: map[string]string{ + "JOB_SYNC_URL": jobsyncurl, + }, }) defer cleanup() @@ -47,41 +79,39 @@ func TestJobCollectorService(t *testing.T) { assert.NoError(t, err) id := jobRes.JSON201.Id - fields := []queryservice.JobCollectorField{ - { - Name: "json_output", - QueryId: jsonRes.JSON201.Id, - }, - } - collRes, err := client.GetJobCollectorByJobIdWithResponse(ctx, id) assert.NoError(t, err) assert.Equal(t, id, collRes.JSON200.JobId) - assert.Equal(t, 1, collRes.JSON200.ActiveVersion) - assert.Equal(t, 1, collRes.JSON200.LatestVersion) - assert.Equal(t, 1, collRes.JSON200.MinimumCleanerVersion) - assert.Equal(t, 1, collRes.JSON200.MinimumTextVersion) + assert.Equal(t, int32(1), collRes.JSON200.ActiveVersion) + assert.Equal(t, int32(1), collRes.JSON200.LatestVersion) + assert.Equal(t, int32(0), collRes.JSON200.MinimumCleanerVersion) + assert.Equal(t, int32(0), collRes.JSON200.MinimumTextVersion) assert.Len(t, collRes.JSON200.Fields, 0) av := int32(2) - minClean := int32(2) - minText := int32(2) + fields := []queryservice.JobCollectorField{ + { + Name: "json", + QueryId: jsonRes.JSON201.Id, + }, + } uRes, err := client.UpdateJobCollectorByJobIdWithResponse(ctx, id, queryservice.JobCollectorUpdate{ - ActiveVersion: &av, - MinimumCleanerVersion: &minClean, - MinimumTextVersion: &minText, - Fields: &fields, + ActiveVersion: &av, + Fields: &fields, }) assert.NoError(t, err) - assert.Nil(t, uRes) + log.Print(string(uRes.Body)) + assert.Equal(t, 200, uRes.StatusCode()) + + test.AssertMessageBody(t, ctx, cfg, jobsyncurl, regexp.MustCompile(`{"id":".+"}`)) collRes, err = client.GetJobCollectorByJobIdWithResponse(ctx, id) assert.NoError(t, err) assert.Equal(t, id, collRes.JSON200.JobId) - assert.Equal(t, 2, collRes.JSON200.ActiveVersion) - assert.Equal(t, 2, collRes.JSON200.LatestVersion) - assert.Equal(t, 2, collRes.JSON200.MinimumCleanerVersion) - assert.Equal(t, 2, collRes.JSON200.MinimumTextVersion) + assert.Equal(t, int32(2), collRes.JSON200.ActiveVersion) + assert.Equal(t, int32(2), collRes.JSON200.LatestVersion) + assert.Equal(t, int32(0), collRes.JSON200.MinimumCleanerVersion) + assert.Equal(t, int32(0), collRes.JSON200.MinimumTextVersion) assert.Len(t, collRes.JSON200.Fields, 1) assert.ElementsMatch(t, fields, collRes.JSON200.Fields) } diff --git a/test/queryService/openapi_test.go b/test/queryService/openapi_test.go index ae86dc79..02445a20 100644 --- a/test/queryService/openapi_test.go +++ b/test/queryService/openapi_test.go @@ -1,4 +1,4 @@ -package integration_test +package endtoend_test import ( "context" @@ -15,6 +15,9 @@ func TestQueryServiceOpenAPI(t *testing.T) { c, cleanup := test.CreateServiceNetwork(t, ctx, &test.ServiceNetworkConfig{ Name: test.QueryService, + Env: map[string]string{ + "JOB_SYNC_URL": "/i/am/here", + }, }) defer cleanup() diff --git a/test/queryService/queryservice_test.go b/test/queryService/queryservice_test.go index 888d348f..ba3993b3 100644 --- a/test/queryService/queryservice_test.go +++ b/test/queryService/queryservice_test.go @@ -1,4 +1,4 @@ -package integration_test +package endtoend_test import ( "context" @@ -15,6 +15,9 @@ func TestQueryService(t *testing.T) { c, cleanup := test.CreateServiceNetwork(t, ctx, &test.ServiceNetworkConfig{ Name: test.QueryService, + Env: map[string]string{ + "JOB_SYNC_URL": "/i/am/here", + }, }) defer cleanup() diff --git a/test/queryService/testquery_test.go b/test/queryService/testquery_test.go index 4641d7fa..c77de2ee 100644 --- a/test/queryService/testquery_test.go +++ b/test/queryService/testquery_test.go @@ -1,4 +1,4 @@ -package integration_test +package endtoend_test import ( "context" @@ -15,6 +15,9 @@ func TestQueryServiceTest(t *testing.T) { c, cleanup := test.CreateServiceNetwork(t, ctx, &test.ServiceNetworkConfig{ Name: test.QueryService, + Env: map[string]string{ + "JOB_SYNC_URL": "/i/am/here", + }, }) defer cleanup()