From 24a038ec3d9927d75eebd74ccab65177a59a133f Mon Sep 17 00:00:00 2001 From: Michael McGuinness Date: Fri, 7 Feb 2025 14:15:06 +0000 Subject: [PATCH] Merged in feature/doctextstructure (pull request #56) DocTextRunner Structure * firstround * shorttests --- api/docCleanRunner/runner_test.go | 4 +- api/docTextRunner/runner.go | 54 +++++++++ api/docTextRunner/runner_test.go | 96 ++++++++++++++++ api/queryRunner/runner_test.go | 4 +- api/querySyncRunner/runner.go | 11 ++ cmd/docTextRunner/main.go | 46 ++++++++ cmd/queryRunner/main.go | 3 - cmd/queryService/main.go | 3 - .../00000000000006_documents.up.sql | 9 ++ database/queries/clean.sql | 11 +- database/queries/text.sql | 21 ++++ deployments/compose.local.yaml | 23 ++++ devbox.json | 2 + internal/database/repository/clean.sql.go | 32 ++++++ internal/database/repository/clean_test.go | 15 +++ internal/database/repository/client_test.go | 3 + .../database/repository/collector_test.go | 3 + internal/database/repository/document_test.go | 3 + internal/database/repository/job_test.go | 3 + internal/database/repository/models.go | 8 ++ internal/database/repository/query_test.go | 9 ++ internal/database/repository/result_test.go | 9 ++ internal/database/repository/text.sql.go | 96 ++++++++++++++++ internal/database/repository/text_test.go | 92 +++++++++++++++ internal/document/clean/create.go | 4 +- internal/document/text/create.go | 45 ++++++++ internal/document/text/create_test.go | 105 ++++++++++++++++++ internal/document/text/extract.go | 54 +++++++++ internal/document/text/extract_test.go | 62 +++++++++++ internal/document/text/service.go | 39 ++++--- internal/document/text/service_test.go | 26 +++-- internal/document/version.go | 13 +++ internal/document/version_test.go | 12 ++ internal/query/service.go | 2 - internal/query/sync.go | 4 +- internal/query/test_test.go | 3 - .../serviceconfig/queue/querysync/config.go | 13 +++ .../queue/querysync/config_test.go | 20 ++++ internal/test/runner.go | 10 +- scripts/compose.yml | 1 + test/process_test.go | 11 +- 41 files changed, 929 insertions(+), 55 deletions(-) create mode 100644 api/docTextRunner/runner.go create mode 100644 api/docTextRunner/runner_test.go create mode 100644 api/querySyncRunner/runner.go create mode 100644 cmd/docTextRunner/main.go create mode 100644 database/queries/text.sql create mode 100644 internal/database/repository/text.sql.go create mode 100644 internal/database/repository/text_test.go create mode 100644 internal/document/text/create.go create mode 100644 internal/document/text/create_test.go create mode 100644 internal/document/text/extract.go create mode 100644 internal/document/text/extract_test.go create mode 100644 internal/serviceconfig/queue/querysync/config.go create mode 100644 internal/serviceconfig/queue/querysync/config_test.go diff --git a/api/docCleanRunner/runner_test.go b/api/docCleanRunner/runner_test.go index b559aed1..75f154e5 100644 --- a/api/docCleanRunner/runner_test.go +++ b/api/docCleanRunner/runner_test.go @@ -25,7 +25,7 @@ import ( "github.com/stretchr/testify/mock" ) -type DocTextConfig struct { +type DocCleanConfig struct { serviceconfig.BaseConfig documenttext.DocTextConfig objectstore.ObjectStoreConfig @@ -39,7 +39,7 @@ func TestDocCleanRunner(t *testing.T) { t.Fatalf("failed to open pgxmock database: %v", err) } - cfg := &DocTextConfig{} + cfg := &DocCleanConfig{} cfg.DBPool = pool cfg.DBQueries = repository.New(pool) mockStore := objectstoremock.NewMockS3Client(t) diff --git a/api/docTextRunner/runner.go b/api/docTextRunner/runner.go new file mode 100644 index 00000000..f9216429 --- /dev/null +++ b/api/docTextRunner/runner.go @@ -0,0 +1,54 @@ +package doctextrunner + +import ( + "context" + "encoding/json" + documenttext "queryorchestration/internal/document/text" + + "github.com/go-playground/validator/v10" + "github.com/google/uuid" + + "github.com/aws/aws-sdk-go-v2/service/sqs/types" +) + +const Name = "docTextRunner" + +type Services struct { + Text *documenttext.Service +} + +type Runner struct { + validator *validator.Validate + svc *Services +} + +func New(validator *validator.Validate, svc *Services) Runner { + return Runner{ + validator: validator, + svc: svc, + } +} + +type Create struct { + ID uuid.UUID `json:"id" validate:"required,uuid"` +} + +func (s Runner) Process(ctx context.Context, req *types.Message) error { + var body Create + 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.Text.Create(ctx, body.ID) + if err != nil { + return err + } + + return nil +} diff --git a/api/docTextRunner/runner_test.go b/api/docTextRunner/runner_test.go new file mode 100644 index 00000000..a67f15a5 --- /dev/null +++ b/api/docTextRunner/runner_test.go @@ -0,0 +1,96 @@ +package doctextrunner_test + +import ( + "context" + "encoding/json" + "fmt" + doctextrunner "queryorchestration/api/docTextRunner" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/document" + documenttext "queryorchestration/internal/document/text" + "queryorchestration/internal/serviceconfig" + "queryorchestration/internal/serviceconfig/objectstore" + "queryorchestration/internal/serviceconfig/queue/querysync" + objectstoremock "queryorchestration/mocks/objectstore" + 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 DocTextConfig struct { + serviceconfig.BaseConfig + querysync.QuerySyncConfig + objectstore.ObjectStoreConfig +} + +func TestDocCleanRunner(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + + 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), + }), + }) + assert.NotNil(t, runner) + + doc := doctextrunner.Create{ + ID: uuid.New(), + } + bodyBytes, err := json.Marshal(doc) + assert.NoError(t, err) + body := string(bodyBytes) + msg := &types.Message{ + Body: &body, + } + + inloc := document.Location{ + Bucket: "bucket_name", + Key: "/i/am/here", + } + + pool.ExpectQuery("name: IsDocumentTextExtracted :one").WithArgs(database.MustToDBUUID(doc.ID)).WillReturnRows( + pgxmock.NewRows([]string{"isextracted"}). + AddRow(false), + ) + pool.ExpectQuery("name: GetDocumentCleanEntry :one").WithArgs(database.MustToDBUUID(doc.ID)).WillReturnRows( + pgxmock.NewRows([]string{"documentId", "bucket", "key"}). + AddRow(database.MustToDBUUID(doc.ID), inloc.Bucket, inloc.Key), + ) + pool.ExpectExec("name: AddDocumentTextEntry :exec").WithArgs(database.MustToDBUUID(doc.ID), int32(1), inloc.Bucket, inloc.Key). + WillReturnResult(pgxmock.NewResult("", 1)) + + mockSQS.EXPECT(). + SendMessage( + mock.Anything, + mock.MatchedBy(func(in *sqs.SendMessageInput) bool { + return *in.QueueUrl == cfg.QuerySyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", doc.ID.String()) + }), + mock.Anything, + ). + Return(&sqs.SendMessageOutput{}, nil) + + err = runner.Process(ctx, msg) + assert.NoError(t, err) +} diff --git a/api/queryRunner/runner_test.go b/api/queryRunner/runner_test.go index 39667377..142f7b15 100644 --- a/api/queryRunner/runner_test.go +++ b/api/queryRunner/runner_test.go @@ -6,7 +6,7 @@ import ( controllers "queryorchestration/api/queryRunner" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" - documenttext "queryorchestration/internal/document/text" + "queryorchestration/internal/document" "queryorchestration/internal/job/collector" "queryorchestration/internal/query" "queryorchestration/internal/query/result" @@ -34,8 +34,8 @@ func TestQueryRunner(t *testing.T) { svc := query.New(cfg, &query.Services{ Result: result.New(cfg), - Text: documenttext.New(), Collector: collector.New(cfg, &collector.Services{}), + Document: document.New(cfg), }) runner := controllers.New(validator.New(), &controllers.Services{ diff --git a/api/querySyncRunner/runner.go b/api/querySyncRunner/runner.go new file mode 100644 index 00000000..a9f12f0a --- /dev/null +++ b/api/querySyncRunner/runner.go @@ -0,0 +1,11 @@ +package querysyncrunner + +import ( + "github.com/google/uuid" +) + +const Name = "querySyncRunner" + +type Create struct { + ID uuid.UUID `json:"id" validate:"required,uuid"` +} diff --git a/cmd/docTextRunner/main.go b/cmd/docTextRunner/main.go new file mode 100644 index 00000000..66d66642 --- /dev/null +++ b/cmd/docTextRunner/main.go @@ -0,0 +1,46 @@ +package main + +import ( + "context" + "log/slog" + "os" + doctextrunner "queryorchestration/api/docTextRunner" + "queryorchestration/internal/document" + documenttext "queryorchestration/internal/document/text" + "queryorchestration/internal/server/runner" + "queryorchestration/internal/serviceconfig/objectstore" + "queryorchestration/internal/serviceconfig/queue/querysync" + + _ "github.com/lib/pq" +) + +type DocTextConfig struct { + runner.BaseConfig + objectstore.ObjectStoreConfig + querysync.QuerySyncConfig +} + +func main() { + ctx := context.Background() + + cfg := &DocTextConfig{} + + cfg.ControllerFunc = func() runner.Controller { + doc := document.New(cfg) + text := documenttext.New(cfg, &documenttext.Services{ + Document: doc, + }) + + return doctextrunner.New(cfg.GetValidator(), &doctextrunner.Services{ + Text: text, + }) + } + + server, err := runner.New(ctx, cfg) + if err != nil { + slog.Error(err.Error()) + os.Exit(1) + } + + server.Listen(ctx) +} diff --git a/cmd/queryRunner/main.go b/cmd/queryRunner/main.go index 763e295a..c6f81ec2 100644 --- a/cmd/queryRunner/main.go +++ b/cmd/queryRunner/main.go @@ -6,7 +6,6 @@ import ( "os" controllers "queryorchestration/api/queryRunner" "queryorchestration/internal/document" - documenttext "queryorchestration/internal/document/text" "queryorchestration/internal/job/collector" "queryorchestration/internal/query" "queryorchestration/internal/query/result" @@ -21,7 +20,6 @@ func main() { cfg := &runner.BaseConfig{} cfg.ControllerFunc = func() runner.Controller { - text := documenttext.New() doc := document.New(cfg) col := collector.New(cfg, &collector.Services{ Document: doc, @@ -29,7 +27,6 @@ func main() { res := result.New(cfg) svc := query.New(cfg, &query.Services{ Result: res, - Text: text, Collector: col, Document: doc, }) diff --git a/cmd/queryService/main.go b/cmd/queryService/main.go index 9bb79ea5..ff9e79ff 100644 --- a/cmd/queryService/main.go +++ b/cmd/queryService/main.go @@ -8,7 +8,6 @@ import ( queryservice "queryorchestration/api/queryService" "queryorchestration/internal/client" "queryorchestration/internal/document" - documenttext "queryorchestration/internal/document/text" "queryorchestration/internal/export" "queryorchestration/internal/job" "queryorchestration/internal/job/collector" @@ -27,14 +26,12 @@ func main() { cfg.RegisterHandlersFunc = func() (*openapi3.T, error) { exp := export.New() - extract := documenttext.New() res := result.New(cfg) doc := document.New(cfg) col := collector.New(cfg, &collector.Services{ Document: doc, }) que := query.New(cfg, &query.Services{ - Text: extract, Result: res, Collector: col, Document: doc, diff --git a/database/migrations/00000000000006_documents.up.sql b/database/migrations/00000000000006_documents.up.sql index 68ca6d3e..1f04281a 100644 --- a/database/migrations/00000000000006_documents.up.sql +++ b/database/migrations/00000000000006_documents.up.sql @@ -22,3 +22,12 @@ CREATE TABLE documentCleans ( key text not null, foreign key (documentId) references documents(id) ); + +CREATE TABLE documentTextExtractions ( + id uuid primary key DEFAULT uuid_generate_v7(), + documentId uuid not null, + version int not null, + bucket text not null, + key text not null, + foreign key (documentId) references documents(id) +); diff --git a/database/queries/clean.sql b/database/queries/clean.sql index 690826c0..aa915c11 100644 --- a/database/queries/clean.sql +++ b/database/queries/clean.sql @@ -9,4 +9,13 @@ SELECT EXISTS( ); -- name: AddDocumentCleanEntry :exec -INSERT INTO documentCleans (documentId, version, bucket, key) VALUES ($1, $2, $3, $4); \ No newline at end of file +INSERT INTO documentCleans (documentId, version, bucket, key) VALUES ($1, $2, $3, $4); + +-- name: GetDocumentCleanEntry :one +SELECT dc.documentId, dc.bucket, dc.key + FROM documentCleans AS dc + JOIN documents AS d ON d.id = dc.documentId + JOIN collectors as c ON d.jobId = c.jobId + JOIN collectorCodeVersions as cv ON c.id = cv.collectorId + WHERE dc.documentId = $1 and dc.version >= cv.minCleanVersion + ORDER BY d.id DESC LIMIT 1; \ No newline at end of file diff --git a/database/queries/text.sql b/database/queries/text.sql new file mode 100644 index 00000000..5b53294b --- /dev/null +++ b/database/queries/text.sql @@ -0,0 +1,21 @@ +-- name: IsDocumentTextExtracted :one +SELECT EXISTS( + SELECT 1 + FROM documentTextExtractions AS dc + JOIN documents AS d ON d.id = dc.documentId + JOIN collectors as c ON d.jobId = c.jobId + JOIN collectorCodeVersions as cv ON c.id = cv.collectorId + WHERE dc.documentId = $1 and dc.version >= cv.minTextVersion +); + +-- name: AddDocumentTextEntry :exec +INSERT INTO documentTextExtractions (documentId, version, bucket, key) VALUES ($1, $2, $3, $4); + +-- name: GetDocumentTextEntry :one +SELECT dc.documentId, dc.bucket, dc.key + FROM documentTextExtractions AS dc + JOIN documents AS d ON d.id = dc.documentId + JOIN collectors as c ON d.jobId = c.jobId + JOIN collectorCodeVersions as cv ON c.id = cv.collectorId + WHERE dc.documentId = $1 and dc.version >= cv.minTextVersion + ORDER BY d.id DESC LIMIT 1; \ No newline at end of file diff --git a/deployments/compose.local.yaml b/deployments/compose.local.yaml index 589575e6..b8ca4092 100644 --- a/deployments/compose.local.yaml +++ b/deployments/compose.local.yaml @@ -46,6 +46,29 @@ services: AWS_S3_USE_PATH_STYLE: true networks: - server-network + doc_text_runner: + image: queryorchestration:latest + command: ["./docTextRunner"] + depends_on: + - db + - localstack + environment: + QUEUE_URL: ${DOCUMENT_TEXT_URL} + QUERY_SYNC_URL: ${QUERY_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_runner: image: queryorchestration:latest command: ["./queryRunner"] diff --git a/devbox.json b/devbox.json index a360a08f..5e981ffe 100644 --- a/devbox.json +++ b/devbox.json @@ -43,10 +43,12 @@ "QNAME_DOCUMENT_INIT": "document_init", "QNAME_DOCUMENT_CLEAN": "document_clean", "QNAME_DOCUMENT_TEXT": "document_text", + "QNAME_QUERY_SYNC": "query_sync", "QNAME_QUERY_RUNNER": "query_runner", "DOCUMENT_INIT_URL": "http://localstack:4566/queue/us-east-1/000000000000/document_init", "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_RUNNER_URL": "http://localstack:4566/queue/us-east-1/000000000000/query_runner", "BUCKET_IN": "documentin" }, diff --git a/internal/database/repository/clean.sql.go b/internal/database/repository/clean.sql.go index a92f92a8..2c53430f 100644 --- a/internal/database/repository/clean.sql.go +++ b/internal/database/repository/clean.sql.go @@ -35,6 +35,38 @@ func (q *Queries) AddDocumentCleanEntry(ctx context.Context, arg *AddDocumentCle return err } +const getDocumentCleanEntry = `-- name: GetDocumentCleanEntry :one +SELECT dc.documentId, dc.bucket, dc.key + FROM documentCleans AS dc + JOIN documents AS d ON d.id = dc.documentId + JOIN collectors as c ON d.jobId = c.jobId + JOIN collectorCodeVersions as cv ON c.id = cv.collectorId + WHERE dc.documentId = $1 and dc.version >= cv.minCleanVersion + ORDER BY d.id DESC LIMIT 1 +` + +type GetDocumentCleanEntryRow struct { + Documentid pgtype.UUID `db:"documentid"` + Bucket string `db:"bucket"` + Key string `db:"key"` +} + +// GetDocumentCleanEntry +// +// SELECT dc.documentId, dc.bucket, dc.key +// FROM documentCleans AS dc +// JOIN documents AS d ON d.id = dc.documentId +// JOIN collectors as c ON d.jobId = c.jobId +// JOIN collectorCodeVersions as cv ON c.id = cv.collectorId +// WHERE dc.documentId = $1 and dc.version >= cv.minCleanVersion +// ORDER BY d.id DESC LIMIT 1 +func (q *Queries) GetDocumentCleanEntry(ctx context.Context, documentid pgtype.UUID) (*GetDocumentCleanEntryRow, error) { + row := q.db.QueryRow(ctx, getDocumentCleanEntry, documentid) + var i GetDocumentCleanEntryRow + err := row.Scan(&i.Documentid, &i.Bucket, &i.Key) + return &i, err +} + const isDocumentClean = `-- name: IsDocumentClean :one SELECT EXISTS( SELECT 1 diff --git a/internal/database/repository/clean_test.go b/internal/database/repository/clean_test.go index 9b929c43..4bc56281 100644 --- a/internal/database/repository/clean_test.go +++ b/internal/database/repository/clean_test.go @@ -13,6 +13,9 @@ import ( ) func TestClean(t *testing.T) { + if testing.Short() { + t.Skip("Skipping long test in short mode") + } ctx := context.Background() cfg := &serviceconfig.BaseConfig{} @@ -53,9 +56,13 @@ func TestClean(t *testing.T) { assert.NoError(t, err) assert.False(t, isclean) + bucket := "example_bucket" + key := "example_key" err = queries.AddDocumentCleanEntry(ctx, &repository.AddDocumentCleanEntryParams{ Documentid: id, Version: 1, + Bucket: bucket, + Key: key, }) assert.NoError(t, err) @@ -74,4 +81,12 @@ func TestClean(t *testing.T) { isclean, err = queries.IsDocumentClean(ctx, id) assert.NoError(t, err) assert.True(t, isclean) + + clean, err := queries.GetDocumentCleanEntry(ctx, id) + assert.NoError(t, err) + assert.EqualExportedValues(t, &repository.GetDocumentCleanEntryRow{ + Documentid: id, + Bucket: bucket, + Key: key, + }, clean) } diff --git a/internal/database/repository/client_test.go b/internal/database/repository/client_test.go index 924e858f..1fd25fac 100644 --- a/internal/database/repository/client_test.go +++ b/internal/database/repository/client_test.go @@ -13,6 +13,9 @@ import ( ) func TestClient(t *testing.T) { + if testing.Short() { + t.Skip("Skipping long test in short mode") + } ctx := context.Background() cfg := &serviceconfig.BaseConfig{} diff --git a/internal/database/repository/collector_test.go b/internal/database/repository/collector_test.go index 2a1251df..76be778b 100644 --- a/internal/database/repository/collector_test.go +++ b/internal/database/repository/collector_test.go @@ -16,6 +16,9 @@ import ( ) func TestCollector(t *testing.T) { + if testing.Short() { + t.Skip("Skipping long test in short mode") + } ctx := context.Background() cfg := &serviceconfig.BaseConfig{} diff --git a/internal/database/repository/document_test.go b/internal/database/repository/document_test.go index 20feb7f7..9e65c4c7 100644 --- a/internal/database/repository/document_test.go +++ b/internal/database/repository/document_test.go @@ -13,6 +13,9 @@ import ( ) func TestDocument(t *testing.T) { + if testing.Short() { + t.Skip("Skipping long test in short mode") + } ctx := context.Background() cfg := &serviceconfig.BaseConfig{} diff --git a/internal/database/repository/job_test.go b/internal/database/repository/job_test.go index bfb2de82..fd732b2a 100644 --- a/internal/database/repository/job_test.go +++ b/internal/database/repository/job_test.go @@ -13,6 +13,9 @@ import ( ) func TestJob(t *testing.T) { + if testing.Short() { + t.Skip("Skipping long test in short mode") + } ctx := context.Background() cfg := &serviceconfig.BaseConfig{} diff --git a/internal/database/repository/models.go b/internal/database/repository/models.go index e235c748..3542b779 100644 --- a/internal/database/repository/models.go +++ b/internal/database/repository/models.go @@ -128,6 +128,14 @@ type Documententry struct { Key string `db:"key"` } +type Documenttextextraction struct { + ID pgtype.UUID `db:"id"` + Documentid pgtype.UUID `db:"documentid"` + Version int32 `db:"version"` + Bucket string `db:"bucket"` + Key string `db:"key"` +} + type Fullactivecollector struct { ID pgtype.UUID `db:"id"` Jobid pgtype.UUID `db:"jobid"` diff --git a/internal/database/repository/query_test.go b/internal/database/repository/query_test.go index b8c6eda1..fbdbabe5 100644 --- a/internal/database/repository/query_test.go +++ b/internal/database/repository/query_test.go @@ -16,6 +16,9 @@ import ( ) func TestQueries(t *testing.T) { + if testing.Short() { + t.Skip("Skipping long test in short mode") + } ctx := context.Background() cfg := &serviceconfig.BaseConfig{} @@ -164,6 +167,9 @@ func TestQueries(t *testing.T) { } func TestQueryDependencyTree(t *testing.T) { + if testing.Short() { + t.Skip("Skipping long test in short mode") + } ctx := context.Background() cfg := &serviceconfig.BaseConfig{} @@ -235,6 +241,9 @@ func TestQueryDependencyTree(t *testing.T) { } func TestQueriesList(t *testing.T) { + if testing.Short() { + t.Skip("Skipping long test in short mode") + } ctx := context.Background() cfg := &serviceconfig.BaseConfig{} diff --git a/internal/database/repository/result_test.go b/internal/database/repository/result_test.go index 9c63f2f4..786ad003 100644 --- a/internal/database/repository/result_test.go +++ b/internal/database/repository/result_test.go @@ -14,6 +14,9 @@ import ( ) func TestResults(t *testing.T) { + if testing.Short() { + t.Skip("Skipping long test in short mode") + } ctx := context.Background() cfg := &serviceconfig.BaseConfig{} @@ -73,6 +76,9 @@ func TestResults(t *testing.T) { } func TestResultValues(t *testing.T) { + if testing.Short() { + t.Skip("Skipping long test in short mode") + } ctx := context.Background() cfg := &serviceconfig.BaseConfig{} @@ -171,6 +177,9 @@ func TestResultValues(t *testing.T) { } func TestUnsyncedQueries(t *testing.T) { + if testing.Short() { + t.Skip("Skipping long test in short mode") + } ctx := context.Background() cfg := &serviceconfig.BaseConfig{} diff --git a/internal/database/repository/text.sql.go b/internal/database/repository/text.sql.go new file mode 100644 index 00000000..6efdb616 --- /dev/null +++ b/internal/database/repository/text.sql.go @@ -0,0 +1,96 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.27.0 +// source: text.sql + +package repository + +import ( + "context" + + "github.com/jackc/pgx/v5/pgtype" +) + +const addDocumentTextEntry = `-- name: AddDocumentTextEntry :exec +INSERT INTO documentTextExtractions (documentId, version, bucket, key) VALUES ($1, $2, $3, $4) +` + +type AddDocumentTextEntryParams struct { + Documentid pgtype.UUID `db:"documentid"` + Version int32 `db:"version"` + Bucket string `db:"bucket"` + Key string `db:"key"` +} + +// AddDocumentTextEntry +// +// INSERT INTO documentTextExtractions (documentId, version, bucket, key) VALUES ($1, $2, $3, $4) +func (q *Queries) AddDocumentTextEntry(ctx context.Context, arg *AddDocumentTextEntryParams) error { + _, err := q.db.Exec(ctx, addDocumentTextEntry, + arg.Documentid, + arg.Version, + arg.Bucket, + arg.Key, + ) + return err +} + +const getDocumentTextEntry = `-- name: GetDocumentTextEntry :one +SELECT dc.documentId, dc.bucket, dc.key + FROM documentTextExtractions AS dc + JOIN documents AS d ON d.id = dc.documentId + JOIN collectors as c ON d.jobId = c.jobId + JOIN collectorCodeVersions as cv ON c.id = cv.collectorId + WHERE dc.documentId = $1 and dc.version >= cv.minTextVersion + ORDER BY d.id DESC LIMIT 1 +` + +type GetDocumentTextEntryRow struct { + Documentid pgtype.UUID `db:"documentid"` + Bucket string `db:"bucket"` + Key string `db:"key"` +} + +// GetDocumentTextEntry +// +// SELECT dc.documentId, dc.bucket, dc.key +// FROM documentTextExtractions AS dc +// JOIN documents AS d ON d.id = dc.documentId +// JOIN collectors as c ON d.jobId = c.jobId +// JOIN collectorCodeVersions as cv ON c.id = cv.collectorId +// WHERE dc.documentId = $1 and dc.version >= cv.minTextVersion +// ORDER BY d.id DESC LIMIT 1 +func (q *Queries) GetDocumentTextEntry(ctx context.Context, documentid pgtype.UUID) (*GetDocumentTextEntryRow, error) { + row := q.db.QueryRow(ctx, getDocumentTextEntry, documentid) + var i GetDocumentTextEntryRow + err := row.Scan(&i.Documentid, &i.Bucket, &i.Key) + return &i, err +} + +const isDocumentTextExtracted = `-- name: IsDocumentTextExtracted :one +SELECT EXISTS( + SELECT 1 + FROM documentTextExtractions AS dc + JOIN documents AS d ON d.id = dc.documentId + JOIN collectors as c ON d.jobId = c.jobId + JOIN collectorCodeVersions as cv ON c.id = cv.collectorId + WHERE dc.documentId = $1 and dc.version >= cv.minTextVersion +) +` + +// IsDocumentTextExtracted +// +// SELECT EXISTS( +// SELECT 1 +// FROM documentTextExtractions AS dc +// JOIN documents AS d ON d.id = dc.documentId +// JOIN collectors as c ON d.jobId = c.jobId +// JOIN collectorCodeVersions as cv ON c.id = cv.collectorId +// WHERE dc.documentId = $1 and dc.version >= cv.minTextVersion +// ) +func (q *Queries) IsDocumentTextExtracted(ctx context.Context, documentid pgtype.UUID) (bool, error) { + row := q.db.QueryRow(ctx, isDocumentTextExtracted, documentid) + var exists bool + err := row.Scan(&exists) + return exists, err +} diff --git a/internal/database/repository/text_test.go b/internal/database/repository/text_test.go new file mode 100644 index 00000000..003e45ab --- /dev/null +++ b/internal/database/repository/text_test.go @@ -0,0 +1,92 @@ +package repository_test + +import ( + "context" + "os" + "path" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/serviceconfig" + "queryorchestration/internal/test" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestTextExtraction(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) + jobId, err := queries.CreateJob(ctx, clientId) + assert.NoError(t, err) + + hash := "example_hash" + id, err := queries.CreateDocument(ctx, &repository.CreateDocumentParams{ + Jobid: jobId, + Hash: hash, + }) + assert.NoError(t, err) + assert.NotEmpty(t, id) + + collId, err := queries.CreateCollector(ctx, jobId) + assert.NoError(t, err) + err = queries.AddCollectorCodeVersion(ctx, &repository.AddCollectorCodeVersionParams{ + Collectorid: collId, + Addedversion: 1, + Mincleanversion: 1, + Mintextversion: 2, + }) + assert.NoError(t, err) + + isextract, err := queries.IsDocumentTextExtracted(ctx, id) + assert.NoError(t, err) + assert.False(t, isextract) + + bucket := "example_bucket" + key := "example_key" + err = queries.AddDocumentTextEntry(ctx, &repository.AddDocumentTextEntryParams{ + Documentid: id, + Version: 1, + Bucket: bucket, + Key: key, + }) + assert.NoError(t, err) + + isextract, err = queries.IsDocumentTextExtracted(ctx, id) + assert.NoError(t, err) + assert.False(t, isextract) + + err = queries.AddCollectorCodeVersion(ctx, &repository.AddCollectorCodeVersionParams{ + Collectorid: collId, + Addedversion: 1, + Mincleanversion: 1, + Mintextversion: 1, + }) + assert.NoError(t, err) + + isextract, err = queries.IsDocumentTextExtracted(ctx, id) + assert.NoError(t, err) + assert.True(t, isextract) + + clean, err := queries.GetDocumentTextEntry(ctx, id) + assert.NoError(t, err) + assert.EqualExportedValues(t, &repository.GetDocumentTextEntryRow{ + Documentid: id, + Bucket: bucket, + Key: key, + }, clean) +} diff --git a/internal/document/clean/create.go b/internal/document/clean/create.go index 6e2fa8d1..953bb437 100644 --- a/internal/document/clean/create.go +++ b/internal/document/clean/create.go @@ -2,8 +2,8 @@ package documentclean import ( "context" + doctextrunner "queryorchestration/api/docTextRunner" "queryorchestration/internal/database" - documenttext "queryorchestration/internal/document/text" "queryorchestration/internal/serviceconfig/queue" "github.com/google/uuid" @@ -33,7 +33,7 @@ func (s *Service) Create(ctx context.Context, id uuid.UUID) error { func (s *Service) informClean(ctx context.Context, id uuid.UUID) error { err := s.cfg.SendToQueue(ctx, &queue.SendParams{ QueueURL: s.cfg.GetDocumentTextURL(), - Body: documenttext.Create{ + Body: doctextrunner.Create{ ID: id, }, }) diff --git a/internal/document/text/create.go b/internal/document/text/create.go new file mode 100644 index 00000000..93e5bef4 --- /dev/null +++ b/internal/document/text/create.go @@ -0,0 +1,45 @@ +package documenttext + +import ( + "context" + querysyncrunner "queryorchestration/api/querySyncRunner" + "queryorchestration/internal/database" + "queryorchestration/internal/serviceconfig/queue" + + "github.com/google/uuid" +) + +func (s *Service) Create(ctx context.Context, id uuid.UUID) error { + isextracted, err := s.cfg.GetDBQueries().IsDocumentTextExtracted(ctx, database.MustToDBUUID(id)) + if err != nil { + return err + } + + if !isextracted { + err = s.extract(ctx, id) + if err != nil { + return err + } + } + + err = s.informExtraction(ctx, id) + if err != nil { + return err + } + + return nil +} + +func (s *Service) informExtraction(ctx context.Context, id uuid.UUID) error { + err := s.cfg.SendToQueue(ctx, &queue.SendParams{ + QueueURL: s.cfg.GetQuerySyncURL(), + Body: querysyncrunner.Create{ + ID: id, + }, + }) + if err != nil { + return err + } + + return nil +} diff --git a/internal/document/text/create_test.go b/internal/document/text/create_test.go new file mode 100644 index 00000000..05619d17 --- /dev/null +++ b/internal/document/text/create_test.go @@ -0,0 +1,105 @@ +package documenttext + +import ( + "context" + "fmt" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/document" + "queryorchestration/internal/serviceconfig" + "queryorchestration/internal/serviceconfig/objectstore" + "queryorchestration/internal/serviceconfig/queue/querysync" + 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 DocTextConfig struct { + serviceconfig.BaseConfig + objectstore.ObjectStoreConfig + querysync.QuerySyncConfig +} + +func TestCreate(t *testing.T) { + ctx := context.Background() + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + + mockSQS := queuemock.NewMockSQSClient(t) + + cfg := &DocTextConfig{} + cfg.QueueClient = mockSQS + cfg.QuerySyncURL = "/i/am/here" + cfg.DBPool = pool + cfg.DBQueries = repository.New(pool) + + svc := Service{ + cfg: cfg, + svc: &Services{ + Document: document.New(cfg), + }, + } + + id := uuid.New() + inloc := document.Location{ + Bucket: "bucket_name", + Key: "/i/am/here", + } + + pool.ExpectQuery("name: IsDocumentTextExtracted :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows( + pgxmock.NewRows([]string{"isextracted"}). + AddRow(false), + ) + pool.ExpectQuery("name: GetDocumentCleanEntry :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows( + pgxmock.NewRows([]string{"documentId", "bucket", "key"}). + AddRow(database.MustToDBUUID(id), inloc.Bucket, inloc.Key), + ) + pool.ExpectExec("name: AddDocumentTextEntry :exec").WithArgs(database.MustToDBUUID(id), int32(1), inloc.Bucket, inloc.Key). + WillReturnResult(pgxmock.NewResult("", 1)) + + mockSQS.EXPECT(). + SendMessage( + mock.Anything, + mock.MatchedBy(func(in *sqs.SendMessageInput) bool { + return *in.QueueUrl == cfg.QuerySyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", id) + }), + mock.Anything, + ). + Return(&sqs.SendMessageOutput{}, nil) + + err = svc.Create(ctx, id) + assert.NoError(t, err) +} + +func TestInformClean(t *testing.T) { + ctx := context.Background() + mockSQS := queuemock.NewMockSQSClient(t) + + cfg := &DocTextConfig{} + cfg.QueueClient = mockSQS + cfg.QuerySyncURL = "/i/am/here" + svc := Service{ + cfg: cfg, + } + id := uuid.New() + + mockSQS.EXPECT(). + SendMessage( + mock.Anything, + mock.MatchedBy(func(in *sqs.SendMessageInput) bool { + return *in.QueueUrl == cfg.QuerySyncURL && *in.MessageBody == fmt.Sprintf("{\"id\":\"%s\"}", id) + }), + mock.Anything, + ). + Return(&sqs.SendMessageOutput{}, nil) + + err := svc.informExtraction(ctx, id) + assert.NoError(t, err) +} diff --git a/internal/document/text/extract.go b/internal/document/text/extract.go new file mode 100644 index 00000000..a95ffce8 --- /dev/null +++ b/internal/document/text/extract.go @@ -0,0 +1,54 @@ +package documenttext + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/document" + + "github.com/google/uuid" +) + +type ExtractionParams struct { + ID uuid.UUID + Location document.Location +} + +func (s *Service) executeExtraction(params *ExtractionParams) (*document.Location, error) { + // TODO - various extraction tasks + return ¶ms.Location, nil +} + +func (s *Service) extract(ctx context.Context, id uuid.UUID) error { + docId := database.MustToDBUUID(id) + + entry, err := s.cfg.GetDBQueries().GetDocumentCleanEntry(ctx, docId) + if err != nil { + return err + } + + outLocation, err := s.executeExtraction(&ExtractionParams{ + ID: id, + Location: document.Location{ + Bucket: entry.Bucket, + Key: entry.Key, + }, + }) + if err != nil { + return err + } + + version := s.svc.Document.GetTextVersion() + + err = s.cfg.GetDBQueries().AddDocumentTextEntry(ctx, &repository.AddDocumentTextEntryParams{ + Documentid: docId, + Version: version, + Bucket: outLocation.Bucket, + Key: outLocation.Key, + }) + if err != nil { + return err + } + + return nil +} diff --git a/internal/document/text/extract_test.go b/internal/document/text/extract_test.go new file mode 100644 index 00000000..ff555f2b --- /dev/null +++ b/internal/document/text/extract_test.go @@ -0,0 +1,62 @@ +package documenttext + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/document" + "testing" + + "github.com/google/uuid" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +func TestExtract(t *testing.T) { + ctx := context.Background() + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + + cfg := &DocTextConfig{} + cfg.DBPool = pool + cfg.DBQueries = repository.New(pool) + + svc := Service{ + cfg: cfg, + svc: &Services{ + Document: document.New(cfg), + }, + } + + id := uuid.New() + inloc := document.Location{ + Bucket: "bucket_name", + Key: "/i/am/here", + } + + pool.ExpectQuery("name: GetDocumentCleanEntry :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows( + pgxmock.NewRows([]string{"documentId", "bucket", "key"}). + AddRow(database.MustToDBUUID(id), inloc.Bucket, inloc.Key), + ) + pool.ExpectExec("name: AddDocumentTextEntry :exec").WithArgs(database.MustToDBUUID(id), int32(1), inloc.Bucket, inloc.Key). + WillReturnResult(pgxmock.NewResult("", 1)) + + err = svc.extract(ctx, id) + assert.NoError(t, err) +} + +func TestExecuteExtraction(t *testing.T) { + svc := Service{} + + id := uuid.New() + location := document.Location{} + + outloc, err := svc.executeExtraction(&ExtractionParams{ + ID: id, + Location: location, + }) + assert.NoError(t, err) + assert.EqualExportedValues(t, location, *outloc) +} diff --git a/internal/document/text/service.go b/internal/document/text/service.go index fed7e67d..01689df4 100644 --- a/internal/document/text/service.go +++ b/internal/document/text/service.go @@ -1,27 +1,30 @@ package documenttext import ( - "github.com/google/uuid" + "queryorchestration/internal/document" + "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 +} + type Service struct { + cfg ConfigProvider + svc *Services } -func New() *Service { - return &Service{} -} - -// TODO - move to api/ -type Create struct { - ID uuid.UUID `json:"id" validate:"required,uuid"` -} - -type IsExtractedParams struct { - DocumentID uuid.UUID - MinCleanVersion int32 - MinTextVersion int32 -} - -func (s *Service) IsExtracted(params *IsExtractedParams) error { - return nil +func New(cfg ConfigProvider, svc *Services) *Service { + return &Service{ + cfg: cfg, + svc: svc, + } } diff --git a/internal/document/text/service_test.go b/internal/document/text/service_test.go index 965ee2a8..7c684240 100644 --- a/internal/document/text/service_test.go +++ b/internal/document/text/service_test.go @@ -1,24 +1,26 @@ package documenttext_test import ( + "queryorchestration/internal/document" documenttext "queryorchestration/internal/document/text" + "queryorchestration/internal/serviceconfig" + "queryorchestration/internal/serviceconfig/objectstore" + "queryorchestration/internal/serviceconfig/queue/querysync" "testing" - "github.com/google/uuid" "github.com/stretchr/testify/assert" ) +type DocTextConfig struct { + serviceconfig.BaseConfig + objectstore.ObjectStoreConfig + querysync.QuerySyncConfig +} + func TestService(t *testing.T) { - svc := documenttext.New() + cfg := &DocTextConfig{} + svc := documenttext.New(cfg, &documenttext.Services{ + Document: document.New(cfg), + }) assert.NotNil(t, svc) } - -func TestIsExtracted(t *testing.T) { - svc := documenttext.New() - - assert.Nil(t, svc.IsExtracted(&documenttext.IsExtractedParams{ - DocumentID: uuid.New(), - MinCleanVersion: int32(1), - MinTextVersion: int32(1), - })) -} diff --git a/internal/document/version.go b/internal/document/version.go index 222ffb1e..8e3279f1 100644 --- a/internal/document/version.go +++ b/internal/document/version.go @@ -2,8 +2,21 @@ package document import ( "fmt" + + "github.com/google/uuid" ) +type IsTextExtractedParams struct { + DocumentID uuid.UUID + MinCleanVersion int32 + MinTextVersion int32 +} + +func (s *Service) IsTextExtracted(params *IsTextExtractedParams) error { + // TODO + return nil +} + func (s *Service) GetCleanVersion() int32 { // TODO - actual version return 1 diff --git a/internal/document/version_test.go b/internal/document/version_test.go index 96fad2ee..e2399ab9 100644 --- a/internal/document/version_test.go +++ b/internal/document/version_test.go @@ -5,6 +5,7 @@ import ( "queryorchestration/internal/serviceconfig" "testing" + "github.com/google/uuid" "github.com/stretchr/testify/assert" ) @@ -41,3 +42,14 @@ func TestGetTextVersion(t *testing.T) { assert.Equal(t, int32(1), svc.GetTextVersion()) } + +func TestIsTextExtracted(t *testing.T) { + cfg := &serviceconfig.BaseConfig{} + svc := document.New(cfg) + + assert.Nil(t, svc.IsTextExtracted(&document.IsTextExtractedParams{ + DocumentID: uuid.New(), + MinCleanVersion: int32(1), + MinTextVersion: int32(1), + })) +} diff --git a/internal/query/service.go b/internal/query/service.go index 1622c811..de621b7b 100644 --- a/internal/query/service.go +++ b/internal/query/service.go @@ -2,7 +2,6 @@ package query import ( "queryorchestration/internal/document" - documenttext "queryorchestration/internal/document/text" "queryorchestration/internal/job/collector" "queryorchestration/internal/query/result" "queryorchestration/internal/serviceconfig" @@ -11,7 +10,6 @@ import ( ) type Services struct { - Text *documenttext.Service Result *result.Service Collector *collector.Service Document *document.Service diff --git a/internal/query/sync.go b/internal/query/sync.go index 23ad7250..86942685 100644 --- a/internal/query/sync.go +++ b/internal/query/sync.go @@ -3,7 +3,7 @@ package query import ( "context" "errors" - documenttext "queryorchestration/internal/document/text" + "queryorchestration/internal/document" "queryorchestration/internal/query/result" resultprocessor "queryorchestration/internal/query/result/processor" "sync" @@ -12,7 +12,7 @@ import ( ) func (s *Service) Sync(ctx context.Context, doc *Document) error { - err := s.svc.Text.IsExtracted(&documenttext.IsExtractedParams{ + err := s.svc.Document.IsTextExtracted(&document.IsTextExtractedParams{ DocumentID: doc.ID, MinCleanVersion: doc.CleanVersion, MinTextVersion: doc.TextVersion, diff --git a/internal/query/test_test.go b/internal/query/test_test.go index aae8d3bd..1059d989 100644 --- a/internal/query/test_test.go +++ b/internal/query/test_test.go @@ -5,7 +5,6 @@ import ( "queryorchestration/internal/database" "queryorchestration/internal/database/repository" "queryorchestration/internal/document" - documenttext "queryorchestration/internal/document/text" "queryorchestration/internal/job/collector" "queryorchestration/internal/query" "queryorchestration/internal/query/result" @@ -32,9 +31,7 @@ func TestTest(t *testing.T) { col := collector.New(cfg, &collector.Services{ Document: docsvc, }) - text := documenttext.New() svc := query.New(cfg, &query.Services{ - Text: text, Document: docsvc, Collector: col, Result: result.New(cfg), diff --git a/internal/serviceconfig/queue/querysync/config.go b/internal/serviceconfig/queue/querysync/config.go new file mode 100644 index 00000000..fa0cc323 --- /dev/null +++ b/internal/serviceconfig/queue/querysync/config.go @@ -0,0 +1,13 @@ +package querysync + +type QuerySyncConfig struct { + QuerySyncURL string `env:"QUERY_SYNC_URL,required,notEmpty"` +} + +func (c *QuerySyncConfig) GetQuerySyncURL() string { + return c.QuerySyncURL +} + +type ConfigProvider interface { + GetQuerySyncURL() string +} diff --git a/internal/serviceconfig/queue/querysync/config_test.go b/internal/serviceconfig/queue/querysync/config_test.go new file mode 100644 index 00000000..b920b60f --- /dev/null +++ b/internal/serviceconfig/queue/querysync/config_test.go @@ -0,0 +1,20 @@ +package querysync_test + +import ( + "queryorchestration/internal/serviceconfig/queue/querysync" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestGetQuerySyncURL(t *testing.T) { + cfg := querysync.QuerySyncConfig{} + + name := cfg.GetQuerySyncURL() + assert.Equal(t, "", name) + + cfg.QuerySyncURL = "name" + name = cfg.GetQuerySyncURL() + assert.Equal(t, "name", name) + assert.Equal(t, cfg.QuerySyncURL, name) +} diff --git a/internal/test/runner.go b/internal/test/runner.go index b8f47b10..1682b45b 100644 --- a/internal/test/runner.go +++ b/internal/test/runner.go @@ -2,7 +2,9 @@ package test import ( "context" + doccleanrunner "queryorchestration/api/docCleanRunner" docinitrunner "queryorchestration/api/docInitRunner" + doctextrunner "queryorchestration/api/docTextRunner" queryrunner "queryorchestration/api/queryRunner" "testing" @@ -14,9 +16,11 @@ import ( type Runner = string const ( - DocInitRunner = docinitrunner.Name - DocCleanRunner = docinitrunner.Name - QueryRunner = queryrunner.Name + DocInitRunner = docinitrunner.Name + DocCleanRunner = doccleanrunner.Name + DocTextRunner = doctextrunner.Name + QuerySyncRunner = queryrunner.Name + QueryRunner = queryrunner.Name ) type RunnerConfig struct { diff --git a/scripts/compose.yml b/scripts/compose.yml index 123d4f58..72ec5fe7 100644 --- a/scripts/compose.yml +++ b/scripts/compose.yml @@ -36,6 +36,7 @@ tasks: - aws sqs create-queue --queue-name $QNAME_DOCUMENT_INIT - 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 s3api put-bucket-notification-configuration \ diff --git a/test/process_test.go b/test/process_test.go index 64110e9f..036fea50 100644 --- a/test/process_test.go +++ b/test/process_test.go @@ -65,7 +65,8 @@ func TestProcess(t *testing.T) { assert.NoError(t, err) doccleanurl := test.CreateQueue(t, ctx, cfg, test.DocCleanRunner) - doctexturl := test.CreateQueue(t, ctx, cfg, "doctextrunner") + doctexturl := test.CreateQueue(t, ctx, cfg, test.DocTextRunner) + querysyncurl := test.CreateQueue(t, ctx, cfg, test.QuerySyncRunner) net, cleanup := test.CreateRunnersAndServicesNetwork(t, ctx, &test.EcosystemNetworkConfig{ Cfg: cfg, @@ -83,6 +84,12 @@ func TestProcess(t *testing.T) { "DOCUMENT_TEXT_URL": doctexturl, }, }, + { + Name: test.DocTextRunner, + Env: map[string]string{ + "QUERY_SYNC_URL": querysyncurl, + }, + }, }, Services: []*test.ServiceNetworkConfig{ { @@ -124,5 +131,5 @@ func TestProcess(t *testing.T) { assert.NoError(t, err) resRegex := regexp.MustCompile(`{"id": "[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"`) - test.AssertMessageBody(t, ctx, cfg, doctexturl, resRegex) + test.AssertMessageBody(t, ctx, cfg, querysyncurl, resRegex) }