From 5b7160fe4455566cdd298b284e8d3b0fc9a8fd4d Mon Sep 17 00:00:00 2001 From: Michael McGuinness Date: Thu, 23 Jan 2025 14:56:20 +0000 Subject: [PATCH] Merged in fature/jobs (pull request #34) Job Collector * createstructure * mostupdatevalidation * repocollectorupdate * updateoutline * updatevalidation * scriptupdate * cleanupdockerignore * update * collectorupdateapi --- .dockerignore | 17 + api/queryRunner/queryrunner_test.go | 7 +- api/queryService/jobcollector.go | 37 +- api/queryService/jobcollector_test.go | 61 ++- api/queryService/query_test.go | 7 +- cmd/queryRunner/main.go | 6 +- cmd/queryService/main.go | 10 +- ...sql => 00000000000001_extensions.down.sql} | 0 ...p.sql => 00000000000001_extensions.up.sql} | 0 .../00000000000002_queries.down.sql | 7 + .../migrations/00000000000002_queries.up.sql | 29 ++ ...wn.sql => 00000000000003_clients.down.sql} | 0 ...e.up.sql => 00000000000003_clients.up.sql} | 0 .../migrations/00000000000004_jobs.down.sql | 1 + .../migrations/00000000000004_jobs.up.sql | 6 + .../00000000000005_collectors.down.sql | 3 + .../00000000000005_collectors.up.sql | 30 ++ .../00000000000006_documents.down.sql | 1 + .../00000000000006_documents.up.sql | 5 + ...wn.sql => 00000000000007_results.down.sql} | 0 ...e.up.sql => 00000000000007_results.up.sql} | 1 + .../00000000000100_query_views.down.sql | 1 + .../00000000000100_query_views.up.sql | 29 ++ .../00000000000101_collector_views.down.sql | 3 + .../00000000000101_collector_views.up.sql | 33 ++ ...41202153118_create_querytype_enum.down.sql | 1 - ...0241202153118_create_querytype_enum.up.sql | 1 - ...241202153119_create_queries_table.down.sql | 1 - ...20241202153119_create_queries_table.up.sql | 6 - ...218183030_create_collectors_table.down.sql | 1 - ...41218183030_create_collectors_table.up.sql | 8 - ...41_create_collector_queries_table.down.sql | 1 - ...3541_create_collector_queries_table.up.sql | 11 - ...40508_create_required_query_table.down.sql | 1 - ...9140508_create_required_query_table.up.sql | 10 - ...19161629_create_queryconfig_table.down.sql | 1 - ...1219161629_create_queryconfig_table.up.sql | 9 - ...61630_create_fullactivequery_view.down.sql | 1 - ...9161630_create_fullactivequery_view.up.sql | 10 - ...ivecollectorswithrequiredids_view.down.sql | 1 - ...ctivecollectorswithrequiredids_view.up.sql | 7 - ...636_create_collectorquerydep_view.down.sql | 1 - ...61636_create_collectorquerydep_view.up.sql | 14 - ...0_create_fullactivecollector_view.down.sql | 1 - ...410_create_fullactivecollector_view.up.sql | 8 - database/queries/collector.sql | 14 +- database/queries/document.sql | 5 + database/queries/job.sql | 5 + database/queries/query.sql | 7 +- database/queries/result.sql | 3 +- internal/database/connection.go | 25 ++ internal/database/connection_test.go | 37 ++ internal/database/repository/collector.sql.go | 91 +++- .../database/repository/collector_test.go | 64 ++- internal/database/repository/document.sql.go | 40 ++ internal/database/repository/document_test.go | 40 ++ internal/database/repository/job.sql.go | 40 ++ internal/database/repository/job_test.go | 39 ++ internal/database/repository/models.go | 39 +- internal/database/repository/query.sql.go | 23 ++ internal/database/repository/query_test.go | 68 +++ internal/database/repository/result.sql.go | 6 +- internal/database/repository/result_test.go | 8 +- internal/document_clean/service.go | 18 + internal/document_clean/service_test.go | 13 + internal/job/collector/create.go | 129 ++++++ internal/job/collector/create_test.go | 64 +++ internal/job/collector/createprivate_test.go | 126 ++++++ internal/job/collector/get_test.go | 20 +- internal/job/collector/parse.go | 14 +- internal/job/collector/parse_test.go | 20 +- internal/job/collector/service.go | 17 +- internal/job/collector/service_test.go | 2 +- internal/job/collector/update.go | 275 +++++++++++++ internal/job/collector/update_test.go | 62 +++ internal/job/collector/updateprivate_test.go | 388 ++++++++++++++++++ internal/query/create.go | 68 ++- internal/query/document/sync_test.go | 22 +- internal/query/normalize.go | 20 +- internal/query/normalize_test.go | 26 +- internal/query/update.go | 60 +-- internal/query/update_test.go | 8 +- internal/query/updateprivate_test.go | 48 +-- internal/server/validation/validation.go | 41 ++ internal/server/validation/validation_test.go | 60 +++ internal/text_extraction/service.go | 18 + internal/text_extraction/service_test.go | 13 + scripts/tests.yml | 4 - 88 files changed, 2181 insertions(+), 297 deletions(-) rename database/migrations/{20241202153111_add_pgcrypto.down.sql => 00000000000001_extensions.down.sql} (100%) rename database/migrations/{20241202153111_add_pgcrypto.up.sql => 00000000000001_extensions.up.sql} (100%) create mode 100644 database/migrations/00000000000002_queries.down.sql create mode 100644 database/migrations/00000000000002_queries.up.sql rename database/migrations/{20241218183020_create_client_table.down.sql => 00000000000003_clients.down.sql} (100%) rename database/migrations/{20241218183020_create_client_table.up.sql => 00000000000003_clients.up.sql} (100%) create mode 100644 database/migrations/00000000000004_jobs.down.sql create mode 100644 database/migrations/00000000000004_jobs.up.sql create mode 100644 database/migrations/00000000000005_collectors.down.sql create mode 100644 database/migrations/00000000000005_collectors.up.sql create mode 100644 database/migrations/00000000000006_documents.down.sql create mode 100644 database/migrations/00000000000006_documents.up.sql rename database/migrations/{20241219130217_create_result_table.down.sql => 00000000000007_results.down.sql} (100%) rename database/migrations/{20241219130217_create_result_table.up.sql => 00000000000007_results.up.sql} (86%) create mode 100644 database/migrations/00000000000100_query_views.down.sql create mode 100644 database/migrations/00000000000100_query_views.up.sql create mode 100644 database/migrations/00000000000101_collector_views.down.sql create mode 100644 database/migrations/00000000000101_collector_views.up.sql delete mode 100644 database/migrations/20241202153118_create_querytype_enum.down.sql delete mode 100644 database/migrations/20241202153118_create_querytype_enum.up.sql delete mode 100644 database/migrations/20241202153119_create_queries_table.down.sql delete mode 100644 database/migrations/20241202153119_create_queries_table.up.sql delete mode 100644 database/migrations/20241218183030_create_collectors_table.down.sql delete mode 100644 database/migrations/20241218183030_create_collectors_table.up.sql delete mode 100644 database/migrations/20241218183541_create_collector_queries_table.down.sql delete mode 100644 database/migrations/20241218183541_create_collector_queries_table.up.sql delete mode 100644 database/migrations/20241219140508_create_required_query_table.down.sql delete mode 100644 database/migrations/20241219140508_create_required_query_table.up.sql delete mode 100644 database/migrations/20241219161629_create_queryconfig_table.down.sql delete mode 100644 database/migrations/20241219161629_create_queryconfig_table.up.sql delete mode 100644 database/migrations/20241219161630_create_fullactivequery_view.down.sql delete mode 100644 database/migrations/20241219161630_create_fullactivequery_view.up.sql delete mode 100644 database/migrations/20241219161635_create_activecollectorswithrequiredids_view.down.sql delete mode 100644 database/migrations/20241219161635_create_activecollectorswithrequiredids_view.up.sql delete mode 100644 database/migrations/20241219161636_create_collectorquerydep_view.down.sql delete mode 100644 database/migrations/20241219161636_create_collectorquerydep_view.up.sql delete mode 100644 database/migrations/20250120165410_create_fullactivecollector_view.down.sql delete mode 100644 database/migrations/20250120165410_create_fullactivecollector_view.up.sql create mode 100644 database/queries/document.sql create mode 100644 database/queries/job.sql create mode 100644 internal/database/repository/document.sql.go create mode 100644 internal/database/repository/document_test.go create mode 100644 internal/database/repository/job.sql.go create mode 100644 internal/database/repository/job_test.go create mode 100644 internal/document_clean/service.go create mode 100644 internal/document_clean/service_test.go create mode 100644 internal/job/collector/create.go create mode 100644 internal/job/collector/create_test.go create mode 100644 internal/job/collector/createprivate_test.go create mode 100644 internal/job/collector/update.go create mode 100644 internal/job/collector/update_test.go create mode 100644 internal/job/collector/updateprivate_test.go create mode 100644 internal/server/validation/validation.go create mode 100644 internal/server/validation/validation_test.go create mode 100644 internal/text_extraction/service.go create mode 100644 internal/text_extraction/service_test.go diff --git a/.dockerignore b/.dockerignore index 9e03c484..8a63cadd 100644 --- a/.dockerignore +++ b/.dockerignore @@ -10,6 +10,7 @@ **/.env **/.git **/.gitignore +**/.git* **/.project **/.settings **/.toolstarget @@ -30,3 +31,19 @@ **/values.dev.yaml LICENSE README.md + +.bitbucket/ +.devbox/ +build/ +database/queries/ +deployments/ +mocks/ +out/ +scripts/ +serviceAPIs/ +test/ +**/*_test.go +Taskfile.yml +sqlc.yml +devbox.* +.*.yml \ No newline at end of file diff --git a/api/queryRunner/queryrunner_test.go b/api/queryRunner/queryrunner_test.go index 1d305661..43ba8af7 100644 --- a/api/queryRunner/queryrunner_test.go +++ b/api/queryRunner/queryrunner_test.go @@ -7,6 +7,7 @@ import ( "queryorchestration/internal/database" "queryorchestration/internal/database/repository" "queryorchestration/internal/job/collector" + "queryorchestration/internal/query" "queryorchestration/internal/query/document" "testing" @@ -32,7 +33,9 @@ func TestQueryRunner(t *testing.T) { } svc := document.New(db, &document.Services{ - Collector: collector.New(db), + Collector: collector.New(db, &collector.Services{ + Query: query.New(db), + }), }) runner := controllers.NewQueryRunner(svc, validator.New()) @@ -61,7 +64,7 @@ func TestQueryRunner(t *testing.T) { pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(database.MustToDBUUID(doc.JobID)). WillReturnRows( pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}). - AddRow(collectorID, database.MustToDBUUID(doc.JobID), minCleanVersion, minTextVersion, int32(1), int32(2), []byte("")), + AddRow(collectorID, database.MustToDBUUID(doc.JobID), &minCleanVersion, &minTextVersion, int32(1), int32(2), []byte("")), ) pool.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), minCleanVersion, minTextVersion). WillReturnRows( diff --git a/api/queryService/jobcollector.go b/api/queryService/jobcollector.go index 5578d58a..983a0668 100644 --- a/api/queryService/jobcollector.go +++ b/api/queryService/jobcollector.go @@ -3,6 +3,7 @@ package queryservice import ( "fmt" "net/http" + "queryorchestration/internal/job/collector" "github.com/google/uuid" "github.com/labstack/echo/v4" @@ -40,5 +41,39 @@ func (s *Controllers) GetJobCollectorByJobId(ctx echo.Context, jobId string) err } func (s *Controllers) UpdateJobCollectorByJobId(ctx echo.Context, jobId string) error { - return ctx.JSON(http.StatusOK, map[string]string{"status": "export triggered"}) + req := JobCollectorUpdate{} + if err := ctx.Bind(&req); err != nil { + return echo.NewHTTPError(http.StatusBadRequest, err) + } + + uid, err := uuid.Parse(jobId) + if err != nil { + return echo.NewHTTPError(http.StatusBadRequest, "Invalid ID") + } + + var fields *map[string]uuid.UUID + if req.Fields != nil { + fields := &map[string]uuid.UUID{} + for _, field := range *req.Fields { + queryId, err := uuid.Parse(field.QueryId) + if err != nil { + return echo.NewHTTPError(http.StatusBadRequest, "Invalid ID") + } + + (*fields)[field.Name] = queryId + } + } + + err = s.svc.JobCollector.UpdateByJobId(ctx.Request().Context(), &collector.UpdateParams{ + JobID: uid, + ActiveVersion: req.ActiveVersion, + MinCleanVersion: req.MinimumCleanerVersion, + MinTextVersion: req.MinimumTextVersion, + Fields: fields, + }) + if err != nil { + return echo.NewHTTPError(http.StatusBadRequest, fmt.Sprintf("Unable to update query: %s", err)) + } + + return ctx.NoContent(http.StatusOK) } diff --git a/api/queryService/jobcollector_test.go b/api/queryService/jobcollector_test.go index 921c9db3..abb18d79 100644 --- a/api/queryService/jobcollector_test.go +++ b/api/queryService/jobcollector_test.go @@ -7,31 +7,76 @@ import ( queryservice "queryorchestration/api/queryService" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" + documentclean "queryorchestration/internal/document_clean" "queryorchestration/internal/job/collector" + "queryorchestration/internal/query" + textextraction "queryorchestration/internal/text_extraction" "strings" "testing" "github.com/go-playground/validator/v10" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/labstack/echo/v4" "github.com/pashagolub/pgxmock/v3" "github.com/stretchr/testify/assert" ) func TestUpdateJobCollector(t *testing.T) { + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + + av := int32(2) + body := queryservice.JobCollectorUpdate{ + ActiveVersion: &av, + } + bodyBytes, err := json.Marshal(body) + assert.Nil(t, err) + e := echo.New() - req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader("")) + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(string(bodyBytes))) + req.Header.Set(echo.HeaderContentType, echo.MIMEApplicationJSON) rec := httptest.NewRecorder() ctx := e.NewContext(req, rec) - cons := queryservice.NewControllers(validator.New(), &queryservice.Services{}) + cons := queryservice.NewControllers(validator.New(), &queryservice.Services{ + JobCollector: collector.New(db, &collector.Services{ + Query: query.New(db), + DocumentClean: documentclean.New(), + TextExtraction: textextraction.New(), + }), + }) - id := uuid.New() + current := collector.Collector{ + ID: uuid.New(), + JobID: uuid.New(), + ActiveVersion: 1, + LatestVersion: 4, + } - err := cons.UpdateJobCollectorByJobId(ctx, id.String()) + ctx.Set("id", current.JobID) + + pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(database.MustToDBUUID(current.JobID)). + WillReturnRows( + pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}). + AddRow(database.MustToDBUUID(current.ID), database.MustToDBUUID(current.JobID), ¤t.MinCleanVersion, ¤t.MinTextVersion, current.ActiveVersion, current.LatestVersion, []byte("")), + ) + pool.ExpectBeginTx(pgx.TxOptions{}) + pool.ExpectExec("name: UpdateCollector :exec").WithArgs(int32(5), int32(2), database.MustToDBUUID(current.ID)). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectCommit() + + err = cons.UpdateJobCollectorByJobId(ctx, current.JobID.String()) assert.Nil(t, err) assert.Equal(t, http.StatusOK, rec.Code) - assert.NotEmpty(t, rec.Body.String()) + assert.Empty(t, rec.Body.String()) } func TestGetJobCollectorByJobId(t *testing.T) { @@ -50,7 +95,9 @@ func TestGetJobCollectorByJobId(t *testing.T) { rec := httptest.NewRecorder() ctx := e.NewContext(req, rec) - svc := collector.New(db) + svc := collector.New(db, &collector.Services{ + Query: query.New(db), + }) cons := queryservice.NewControllers(validator.New(), &queryservice.Services{ JobCollector: svc, }) @@ -65,7 +112,7 @@ func TestGetJobCollectorByJobId(t *testing.T) { pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(database.MustToDBUUID(coll.JobID)). WillReturnRows( pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}). - AddRow(database.MustToDBUUID(coll.ID), database.MustToDBUUID(coll.JobID), coll.MinCleanVersion, coll.MinTextVersion, int32(1), int32(2), []byte("")), + AddRow(database.MustToDBUUID(coll.ID), database.MustToDBUUID(coll.JobID), &coll.MinCleanVersion, &coll.MinTextVersion, int32(1), int32(2), []byte("")), ) err = cons.GetJobCollectorByJobId(ctx, coll.JobID.String()) diff --git a/api/queryService/query_test.go b/api/queryService/query_test.go index 9e671a62..efdf4dd6 100644 --- a/api/queryService/query_test.go +++ b/api/queryService/query_test.go @@ -167,7 +167,10 @@ func TestUpdateQuery(t *testing.T) { Query: query.New(db), }) - body := queryservice.QueryUpdate{} + av := int32(2) + body := queryservice.QueryUpdate{ + ActiveVersion: &av, + } bodyBytes, err := json.Marshal(body) assert.Nil(t, err) @@ -187,7 +190,7 @@ func TestUpdateQuery(t *testing.T) { ) pool.ExpectBeginTx(pgx.TxOptions{}) - pool.ExpectExec("name: UpdateQuery :exec").WithArgs(int32(1), int32(2), database.MustToDBUUID(id)). + pool.ExpectExec("name: UpdateQuery :exec").WithArgs(av, int32(3), database.MustToDBUUID(id)). WillReturnResult(pgxmock.NewResult("", 1)) pool.ExpectCommit() diff --git a/cmd/queryRunner/main.go b/cmd/queryRunner/main.go index d26c11a5..14845f8f 100644 --- a/cmd/queryRunner/main.go +++ b/cmd/queryRunner/main.go @@ -4,6 +4,7 @@ import ( "context" controllers "queryorchestration/api/queryRunner" "queryorchestration/internal/job/collector" + "queryorchestration/internal/query" "queryorchestration/internal/query/document" "queryorchestration/internal/server" "queryorchestration/internal/server/queue" @@ -15,7 +16,10 @@ func main() { ctx := context.Background() queryrunner := func(cfg *server.Config) queue.Controller { - coll := collector.New(cfg.Database) + que := query.New(cfg.Database) + coll := collector.New(cfg.Database, &collector.Services{ + Query: que, + }) svc := document.New(cfg.Database, &document.Services{ Collector: coll, diff --git a/cmd/queryService/main.go b/cmd/queryService/main.go index e24ff900..db363a99 100644 --- a/cmd/queryService/main.go +++ b/cmd/queryService/main.go @@ -5,11 +5,13 @@ import ( "log" queryservice "queryorchestration/api/queryService" "queryorchestration/internal/client" + documentclean "queryorchestration/internal/document_clean" "queryorchestration/internal/export" "queryorchestration/internal/job/collector" "queryorchestration/internal/query" "queryorchestration/internal/server" "queryorchestration/internal/server/api" + textextraction "queryorchestration/internal/text_extraction" "github.com/labstack/echo/v4" _ "github.com/lib/pq" @@ -20,8 +22,14 @@ func main() { registerHandlers := func(cfg *server.Config, e *echo.Echo) *api.APIConfig { exp := export.New(cfg.Database) - col := collector.New(cfg.Database) que := query.New(cfg.Database) + extract := textextraction.New() + clean := documentclean.New() + col := collector.New(cfg.Database, &collector.Services{ + Query: que, + TextExtraction: extract, + DocumentClean: clean, + }) cli := client.New(cfg.Database) services := &queryservice.Services{ diff --git a/database/migrations/20241202153111_add_pgcrypto.down.sql b/database/migrations/00000000000001_extensions.down.sql similarity index 100% rename from database/migrations/20241202153111_add_pgcrypto.down.sql rename to database/migrations/00000000000001_extensions.down.sql diff --git a/database/migrations/20241202153111_add_pgcrypto.up.sql b/database/migrations/00000000000001_extensions.up.sql similarity index 100% rename from database/migrations/20241202153111_add_pgcrypto.up.sql rename to database/migrations/00000000000001_extensions.up.sql diff --git a/database/migrations/00000000000002_queries.down.sql b/database/migrations/00000000000002_queries.down.sql new file mode 100644 index 00000000..60000287 --- /dev/null +++ b/database/migrations/00000000000002_queries.down.sql @@ -0,0 +1,7 @@ +DROP TABLE queries; + +DROP TYPE queryType; + +DROP TABLE requiredQueries; + +DROP TABLE queryConfigs; \ No newline at end of file diff --git a/database/migrations/00000000000002_queries.up.sql b/database/migrations/00000000000002_queries.up.sql new file mode 100644 index 00000000..78caec6f --- /dev/null +++ b/database/migrations/00000000000002_queries.up.sql @@ -0,0 +1,29 @@ +CREATE TYPE queryType AS ENUM ('context_full', 'json_extractor'); + +CREATE TABLE queries ( + id uuid primary key DEFAULT gen_random_uuid(), + latestVersion int not null default 1, + activeVersion int not null default 1, + type queryType not null +); + +CREATE TABLE requiredQueries ( + id uuid primary key DEFAULT gen_random_uuid(), + queryId uuid not null, + requiredQueryId uuid not null, + addedVersion int not null, + removedVersion int, + foreign key (queryId) references queries(id), + foreign key (requiredQueryId) references queries(id), + unique (queryId, requiredQueryId, removedVersion) +); + +CREATE TABLE queryConfigs ( + id uuid primary key DEFAULT gen_random_uuid(), + queryId uuid not null, + config jsonb not null, + addedVersion int not null, + removedVersion int, + foreign key (queryId) references queries(id), + unique (queryId, removedVersion) +); \ No newline at end of file diff --git a/database/migrations/20241218183020_create_client_table.down.sql b/database/migrations/00000000000003_clients.down.sql similarity index 100% rename from database/migrations/20241218183020_create_client_table.down.sql rename to database/migrations/00000000000003_clients.down.sql diff --git a/database/migrations/20241218183020_create_client_table.up.sql b/database/migrations/00000000000003_clients.up.sql similarity index 100% rename from database/migrations/20241218183020_create_client_table.up.sql rename to database/migrations/00000000000003_clients.up.sql diff --git a/database/migrations/00000000000004_jobs.down.sql b/database/migrations/00000000000004_jobs.down.sql new file mode 100644 index 00000000..b23f8f7b --- /dev/null +++ b/database/migrations/00000000000004_jobs.down.sql @@ -0,0 +1 @@ +DROP TABLE jobs; \ No newline at end of file diff --git a/database/migrations/00000000000004_jobs.up.sql b/database/migrations/00000000000004_jobs.up.sql new file mode 100644 index 00000000..4f43b3c1 --- /dev/null +++ b/database/migrations/00000000000004_jobs.up.sql @@ -0,0 +1,6 @@ +CREATE TABLE jobs ( + id uuid primary key DEFAULT gen_random_uuid(), + clientId uuid not null, + canSync boolean not null default false, + foreign key (clientId) references clients(id) +); \ No newline at end of file diff --git a/database/migrations/00000000000005_collectors.down.sql b/database/migrations/00000000000005_collectors.down.sql new file mode 100644 index 00000000..02fe466c --- /dev/null +++ b/database/migrations/00000000000005_collectors.down.sql @@ -0,0 +1,3 @@ +DROP TABLE collectorQueries; + +DROP TABLE collectors; diff --git a/database/migrations/00000000000005_collectors.up.sql b/database/migrations/00000000000005_collectors.up.sql new file mode 100644 index 00000000..ec8c5dbd --- /dev/null +++ b/database/migrations/00000000000005_collectors.up.sql @@ -0,0 +1,30 @@ +CREATE TABLE collectors ( + id uuid primary key DEFAULT gen_random_uuid(), + jobId uuid not null, + latestVersion int not null default 1, + activeVersion int not null default 1, + foreign key (jobId) references jobs(id), + unique (jobId) +); + +CREATE TABLE collectorCodeVersions ( + id uuid primary key DEFAULT gen_random_uuid(), + collectorId uuid not null, + minCleanVersion int not null, + minTextVersion int not null, + addedVersion int not null, + removedVersion int, + foreign key (collectorId) references collectors(id) +); + +CREATE TABLE collectorQueries ( + id uuid primary key DEFAULT gen_random_uuid(), + collectorId uuid not null, + name varchar(255) not null, + queryId uuid not null, + addedVersion int not null, + removedVersion int, + foreign key (queryId) references queries(id), + foreign key (collectorId) references collectors(id), + unique (collectorId, name, removedVersion) +); \ No newline at end of file diff --git a/database/migrations/00000000000006_documents.down.sql b/database/migrations/00000000000006_documents.down.sql new file mode 100644 index 00000000..b51b6a91 --- /dev/null +++ b/database/migrations/00000000000006_documents.down.sql @@ -0,0 +1 @@ +DROP TABLE documents; \ No newline at end of file diff --git a/database/migrations/00000000000006_documents.up.sql b/database/migrations/00000000000006_documents.up.sql new file mode 100644 index 00000000..24236f71 --- /dev/null +++ b/database/migrations/00000000000006_documents.up.sql @@ -0,0 +1,5 @@ +CREATE TABLE documents ( + id uuid primary key DEFAULT gen_random_uuid(), + jobId uuid not null, + foreign key (jobId) references jobs(id) +); \ No newline at end of file diff --git a/database/migrations/20241219130217_create_result_table.down.sql b/database/migrations/00000000000007_results.down.sql similarity index 100% rename from database/migrations/20241219130217_create_result_table.down.sql rename to database/migrations/00000000000007_results.down.sql diff --git a/database/migrations/20241219130217_create_result_table.up.sql b/database/migrations/00000000000007_results.up.sql similarity index 86% rename from database/migrations/20241219130217_create_result_table.up.sql rename to database/migrations/00000000000007_results.up.sql index a44152c4..d41879f1 100644 --- a/database/migrations/20241219130217_create_result_table.up.sql +++ b/database/migrations/00000000000007_results.up.sql @@ -7,5 +7,6 @@ CREATE TABLE results ( textVersion int not null, queryVersion int not null, foreign key (queryId) references queries(id), + foreign key (documentId) references documents(id), unique (queryId, documentId, cleanVersion, textVersion, queryVersion) ); \ No newline at end of file diff --git a/database/migrations/00000000000100_query_views.down.sql b/database/migrations/00000000000100_query_views.down.sql new file mode 100644 index 00000000..cf880abe --- /dev/null +++ b/database/migrations/00000000000100_query_views.down.sql @@ -0,0 +1 @@ +DROP VIEW fullActiveQueries; diff --git a/database/migrations/00000000000100_query_views.up.sql b/database/migrations/00000000000100_query_views.up.sql new file mode 100644 index 00000000..a544ae3e --- /dev/null +++ b/database/migrations/00000000000100_query_views.up.sql @@ -0,0 +1,29 @@ +CREATE VIEW fullActiveQueries AS + SELECT DISTINCT q.id, q.type, q.activeVersion, q.latestVersion, coalesce(c.config, null) as config, ARRAY_AGG(DISTINCT r.requiredQueryId)::uuid[] as requiredIds + FROM queries AS q + LEFT JOIN queryConfigs AS c ON q.id = c.queryId + and q.activeVersion >= c.addedVersion + and q.activeVersion < COALESCE(c.removedVersion, q.activeVersion + 1) + LEFT JOIN requiredQueries AS r ON q.id = r.queryId + and q.activeVersion >= r.addedVersion + and q.activeVersion < COALESCE(r.removedVersion, q.activeVersion + 1) + GROUP BY q.id, q.type, q.activeversion, q.latestversion, c.config; + +CREATE VIEW queryActiveDependencies AS + WITH RECURSIVE queryActiveDependencies AS ( + SELECT q.id, r.requiredQueryId + FROM queries AS q + LEFT JOIN requiredQueries AS r ON q.id = r.queryId + and q.activeVersion >= r.addedVersion + and q.activeVersion < COALESCE(r.removedVersion, q.activeVersion + 1) + + UNION ALL + + SELECT qd.id, r.requiredQueryId + FROM queries AS q + LEFT JOIN requiredQueries AS r ON q.id = r.queryId + and q.activeVersion >= r.addedVersion + and q.activeVersion < COALESCE(r.removedVersion, q.activeVersion + 1) + JOIN queryActiveDependencies as qd on q.id = qd.requiredQueryId + ) + SELECT DISTINCT id, requiredQueryId FROM queryActiveDependencies; \ No newline at end of file diff --git a/database/migrations/00000000000101_collector_views.down.sql b/database/migrations/00000000000101_collector_views.down.sql new file mode 100644 index 00000000..8553c48b --- /dev/null +++ b/database/migrations/00000000000101_collector_views.down.sql @@ -0,0 +1,3 @@ +DROP VIEW fullActiveCollectors; + +DROP VIEW collectorQueryDependencyTree; \ No newline at end of file diff --git a/database/migrations/00000000000101_collector_views.up.sql b/database/migrations/00000000000101_collector_views.up.sql new file mode 100644 index 00000000..410b1c0f --- /dev/null +++ b/database/migrations/00000000000101_collector_views.up.sql @@ -0,0 +1,33 @@ +CREATE VIEW fullActiveCollectors AS + SELECT DISTINCT c.id, c.jobId, cv.minCleanVersion, cv.minTextVersion, c.activeVersion, c.latestVersion, + jsonb_object_agg(q.name, q.queryId) FILTER (WHERE q.name is not null) AS fields + FROM collectors AS c + LEFT JOIN collectorCodeVersions AS cv ON c.id = cv.collectorId + AND c.activeVersion >= cv.addedVersion + and c.activeVersion < COALESCE(cv.removedVersion, c.activeVersion + 1) + LEFT JOIN collectorQueries AS q ON c.id = q.collectorId + AND c.activeVersion >= q.addedVersion + and c.activeVersion < COALESCE(q.removedVersion, c.activeVersion + 1) + GROUP BY c.id, c.jobId, cv.minCleanVersion, cv.minTextVersion; + +CREATE VIEW collectorQueryDependencyTree AS + WITH RECURSIVE collectorQueryDependencyTree AS ( + SELECT cq.id as collectorId, aqc.id as queryId, aqc.type, aqc.requiredIds, aqc.activeVersion + FROM fullActiveQueries as aqc + LEFT JOIN ( + SELECT DISTINCT c.id, array_agg(q.queryId) as queryIds + FROM collectors as c + LEFT JOIN collectorQueries as q ON c.id = q.collectorId + AND c.activeVersion >= q.addedVersion + and c.activeVersion < COALESCE(q.removedVersion, c.activeVersion + 1) + GROUP BY c.id + ) as cq ON aqc.id = any(cq.queryIds) + + UNION ALL + + SELECT acq.collectorId, q.id as queryId, q.type, q.requiredIds, q.activeVersion + FROM fullActiveQueries as q + JOIN collectorQueryDependencyTree as acq on q.id = ANY(acq.requiredIds) + ) + SELECT collectorId, queryId, type, activeVersion as queryVersion, requiredIds::uuid[] + FROM collectorQueryDependencyTree; \ No newline at end of file diff --git a/database/migrations/20241202153118_create_querytype_enum.down.sql b/database/migrations/20241202153118_create_querytype_enum.down.sql deleted file mode 100644 index 99e657bc..00000000 --- a/database/migrations/20241202153118_create_querytype_enum.down.sql +++ /dev/null @@ -1 +0,0 @@ -DROP TYPE queryType; \ No newline at end of file diff --git a/database/migrations/20241202153118_create_querytype_enum.up.sql b/database/migrations/20241202153118_create_querytype_enum.up.sql deleted file mode 100644 index c8bfe85c..00000000 --- a/database/migrations/20241202153118_create_querytype_enum.up.sql +++ /dev/null @@ -1 +0,0 @@ -CREATE TYPE queryType AS ENUM ('context_full', 'json_extractor'); \ No newline at end of file diff --git a/database/migrations/20241202153119_create_queries_table.down.sql b/database/migrations/20241202153119_create_queries_table.down.sql deleted file mode 100644 index 9eeaffa8..00000000 --- a/database/migrations/20241202153119_create_queries_table.down.sql +++ /dev/null @@ -1 +0,0 @@ -DROP TABLE queries; \ No newline at end of file diff --git a/database/migrations/20241202153119_create_queries_table.up.sql b/database/migrations/20241202153119_create_queries_table.up.sql deleted file mode 100644 index 18ab6203..00000000 --- a/database/migrations/20241202153119_create_queries_table.up.sql +++ /dev/null @@ -1,6 +0,0 @@ -CREATE TABLE queries ( - id uuid primary key DEFAULT gen_random_uuid(), - latestVersion int not null default 1, - activeVersion int not null default 1, - type queryType not null -); \ No newline at end of file diff --git a/database/migrations/20241218183030_create_collectors_table.down.sql b/database/migrations/20241218183030_create_collectors_table.down.sql deleted file mode 100644 index e656dd3d..00000000 --- a/database/migrations/20241218183030_create_collectors_table.down.sql +++ /dev/null @@ -1 +0,0 @@ -DROP TABLE collectors; \ No newline at end of file diff --git a/database/migrations/20241218183030_create_collectors_table.up.sql b/database/migrations/20241218183030_create_collectors_table.up.sql deleted file mode 100644 index b5b95ae3..00000000 --- a/database/migrations/20241218183030_create_collectors_table.up.sql +++ /dev/null @@ -1,8 +0,0 @@ -CREATE TABLE collectors ( - id uuid primary key DEFAULT gen_random_uuid(), - jobId uuid not null, - minCleanVersion int not null default 1, - minTextVersion int not null default 1, - latestVersion int not null default 1, - activeVersion int not null default 1 -); \ No newline at end of file diff --git a/database/migrations/20241218183541_create_collector_queries_table.down.sql b/database/migrations/20241218183541_create_collector_queries_table.down.sql deleted file mode 100644 index 4609d333..00000000 --- a/database/migrations/20241218183541_create_collector_queries_table.down.sql +++ /dev/null @@ -1 +0,0 @@ -DROP TABLE collectorQueries; \ No newline at end of file diff --git a/database/migrations/20241218183541_create_collector_queries_table.up.sql b/database/migrations/20241218183541_create_collector_queries_table.up.sql deleted file mode 100644 index 48aa61f8..00000000 --- a/database/migrations/20241218183541_create_collector_queries_table.up.sql +++ /dev/null @@ -1,11 +0,0 @@ -CREATE TABLE collectorQueries ( - id uuid primary key DEFAULT gen_random_uuid(), - collectorId uuid not null, - name varchar(255) not null, - queryId uuid not null, - addedVersion int not null, - removedVersion int, - foreign key (queryId) references queries(id), - foreign key (collectorId) references collectors(id), - unique (collectorId, name, removedVersion) -); \ No newline at end of file diff --git a/database/migrations/20241219140508_create_required_query_table.down.sql b/database/migrations/20241219140508_create_required_query_table.down.sql deleted file mode 100644 index 63628833..00000000 --- a/database/migrations/20241219140508_create_required_query_table.down.sql +++ /dev/null @@ -1 +0,0 @@ -DROP TABLE requiredQueries; \ No newline at end of file diff --git a/database/migrations/20241219140508_create_required_query_table.up.sql b/database/migrations/20241219140508_create_required_query_table.up.sql deleted file mode 100644 index 6105256b..00000000 --- a/database/migrations/20241219140508_create_required_query_table.up.sql +++ /dev/null @@ -1,10 +0,0 @@ -CREATE TABLE requiredQueries ( - id uuid primary key DEFAULT gen_random_uuid(), - queryId uuid not null, - requiredQueryId uuid not null, - addedVersion int not null, - removedVersion int, - foreign key (queryId) references queries(id), - foreign key (requiredQueryId) references queries(id), - unique (queryId, requiredQueryId, removedVersion) -); \ No newline at end of file diff --git a/database/migrations/20241219161629_create_queryconfig_table.down.sql b/database/migrations/20241219161629_create_queryconfig_table.down.sql deleted file mode 100644 index 8211282d..00000000 --- a/database/migrations/20241219161629_create_queryconfig_table.down.sql +++ /dev/null @@ -1 +0,0 @@ -DROP TABLE queryConfigs; \ No newline at end of file diff --git a/database/migrations/20241219161629_create_queryconfig_table.up.sql b/database/migrations/20241219161629_create_queryconfig_table.up.sql deleted file mode 100644 index 0e56425e..00000000 --- a/database/migrations/20241219161629_create_queryconfig_table.up.sql +++ /dev/null @@ -1,9 +0,0 @@ -CREATE TABLE queryConfigs ( - id uuid primary key DEFAULT gen_random_uuid(), - queryId uuid not null, - config jsonb not null, - addedVersion int not null, - removedVersion int, - foreign key (queryId) references queries(id), - unique (queryId, removedVersion) -); \ No newline at end of file diff --git a/database/migrations/20241219161630_create_fullactivequery_view.down.sql b/database/migrations/20241219161630_create_fullactivequery_view.down.sql deleted file mode 100644 index 04c4b9a6..00000000 --- a/database/migrations/20241219161630_create_fullactivequery_view.down.sql +++ /dev/null @@ -1 +0,0 @@ -DROP VIEW fullActiveQueries; \ No newline at end of file diff --git a/database/migrations/20241219161630_create_fullactivequery_view.up.sql b/database/migrations/20241219161630_create_fullactivequery_view.up.sql deleted file mode 100644 index 385fc535..00000000 --- a/database/migrations/20241219161630_create_fullactivequery_view.up.sql +++ /dev/null @@ -1,10 +0,0 @@ -CREATE VIEW fullActiveQueries AS - SELECT DISTINCT q.id, q.type, q.activeVersion, q.latestVersion, coalesce(c.config, null) as config, ARRAY_AGG(DISTINCT r.requiredQueryId)::uuid[] as requiredIds - FROM queries AS q - LEFT JOIN queryConfigs AS c ON q.id = c.queryId - and q.activeVersion >= c.addedVersion - and q.activeVersion < COALESCE(c.removedVersion, q.activeVersion + 1) - LEFT JOIN requiredQueries AS r ON q.id = r.queryId - and q.activeVersion >= r.addedVersion - and q.activeVersion < COALESCE(r.removedVersion, q.activeVersion + 1) - GROUP BY q.id, q.type, q.activeversion, q.latestversion, c.config; diff --git a/database/migrations/20241219161635_create_activecollectorswithrequiredids_view.down.sql b/database/migrations/20241219161635_create_activecollectorswithrequiredids_view.down.sql deleted file mode 100644 index 8ac7d42c..00000000 --- a/database/migrations/20241219161635_create_activecollectorswithrequiredids_view.down.sql +++ /dev/null @@ -1 +0,0 @@ -DROP VIEW activeCollectorsWithRequiredIDs; \ No newline at end of file diff --git a/database/migrations/20241219161635_create_activecollectorswithrequiredids_view.up.sql b/database/migrations/20241219161635_create_activecollectorswithrequiredids_view.up.sql deleted file mode 100644 index 15b13c92..00000000 --- a/database/migrations/20241219161635_create_activecollectorswithrequiredids_view.up.sql +++ /dev/null @@ -1,7 +0,0 @@ -CREATE VIEW activeCollectorsWithRequiredIDs AS - SELECT DISTINCT c.id, c.activeVersion, array_agg(q.queryId) as queryIds - FROM collectors as c - LEFT JOIN collectorQueries as q ON c.id = q.collectorId - AND c.activeVersion >= q.addedVersion - and c.activeVersion < COALESCE(q.removedVersion, c.activeVersion + 1) - GROUP BY c.id, c.activeVersion; \ No newline at end of file diff --git a/database/migrations/20241219161636_create_collectorquerydep_view.down.sql b/database/migrations/20241219161636_create_collectorquerydep_view.down.sql deleted file mode 100644 index df524ac4..00000000 --- a/database/migrations/20241219161636_create_collectorquerydep_view.down.sql +++ /dev/null @@ -1 +0,0 @@ -DROP VIEW collectorQueryDependencyTree; \ No newline at end of file diff --git a/database/migrations/20241219161636_create_collectorquerydep_view.up.sql b/database/migrations/20241219161636_create_collectorquerydep_view.up.sql deleted file mode 100644 index 4e7b4ea9..00000000 --- a/database/migrations/20241219161636_create_collectorquerydep_view.up.sql +++ /dev/null @@ -1,14 +0,0 @@ -CREATE VIEW collectorQueryDependencyTree AS -WITH RECURSIVE collectorQueryDependencyTree AS ( - SELECT cq.id, aqc.id as queryId, aqc.type, aqc.requiredIds, aqc.activeVersion - FROM fullActiveQueries as aqc - JOIN activeCollectorsWithRequiredIDs as cq ON aqc.id = ANY(cq.queryIds) - - UNION ALL - - SELECT acq.id, q.id as queryId, q.type, q.requiredIds, q.activeVersion - FROM fullActiveQueries as q - JOIN collectorQueryDependencyTree as acq on q.id = ANY(acq.requiredIds) -) -SELECT id as collectorId, queryId, type, activeVersion as queryVersion, requiredIds::uuid[] - FROM collectorQueryDependencyTree; \ No newline at end of file diff --git a/database/migrations/20250120165410_create_fullactivecollector_view.down.sql b/database/migrations/20250120165410_create_fullactivecollector_view.down.sql deleted file mode 100644 index 2bb7dcc1..00000000 --- a/database/migrations/20250120165410_create_fullactivecollector_view.down.sql +++ /dev/null @@ -1 +0,0 @@ -DROP VIEW fullActiveCollectors; \ No newline at end of file diff --git a/database/migrations/20250120165410_create_fullactivecollector_view.up.sql b/database/migrations/20250120165410_create_fullactivecollector_view.up.sql deleted file mode 100644 index 0e126e7d..00000000 --- a/database/migrations/20250120165410_create_fullactivecollector_view.up.sql +++ /dev/null @@ -1,8 +0,0 @@ -CREATE VIEW fullActiveCollectors AS - SELECT DISTINCT c.id, c.jobId, c.minCleanVersion, c.minTextVersion, c.activeVersion, c.latestVersion, - jsonb_object_agg(q.name, q.queryId) FILTER (WHERE q.name is not null) AS fields - FROM collectors AS c - LEFT JOIN collectorQueries AS q ON c.id = q.collectorId - AND c.activeVersion >= q.addedVersion - and c.activeVersion < COALESCE(q.removedVersion, c.activeVersion + 1) - GROUP BY c.id, c.jobId, c.minCleanVersion, c.minTextVersion; diff --git a/database/queries/collector.sql b/database/queries/collector.sql index 45c97aad..01a8e6cd 100644 --- a/database/queries/collector.sql +++ b/database/queries/collector.sql @@ -8,7 +8,19 @@ SELECT * FROM fullActiveCollectors WHERE jobId = $1 LIMIT 1; SELECT * FROM fullActiveCollectors WHERE id = $1 LIMIT 1; -- name: CreateCollector :one -INSERT INTO collectors (jobId, minCleanVersion, minTextVersion) VALUES ($1, $2, $3) RETURNING id; +INSERT INTO collectors (jobId) VALUES ($1) RETURNING id; + +-- name: UpdateCollector :exec +UPDATE collectors SET latestVersion = $1, activeVersion = $2 WHERE id = $3; + +-- name: AddCollectorCodeVersion :exec +INSERT INTO collectorCodeVersions (collectorId, addedVersion, minCleanVersion, minTextVersion) VALUES ($1, $2, $3, $4); + +-- name: RemoveCollectorCodeVersion :exec +UPDATE collectorCodeVersions SET removedVersion = $1 WHERE collectorId = $2 and removedVersion is null; -- name: AddCollectorQuery :exec INSERT INTO collectorQueries (collectorId, name, queryId, addedVersion) VALUES ($1, $2, $3, $4); + +-- name: RemoveCollectorQuery :exec +UPDATE collectorQueries SET removedVersion = $1 WHERE queryId = $2 and collectorId = $3 and removedVersion is null; diff --git a/database/queries/document.sql b/database/queries/document.sql new file mode 100644 index 00000000..0261c0b4 --- /dev/null +++ b/database/queries/document.sql @@ -0,0 +1,5 @@ +-- name: GetDocument :one +SELECT id, jobId FROM documents WHERE id = $1 LIMIT 1; + +-- name: CreateDocument :one +INSERT INTO documents (jobId) VALUES ($1) RETURNING id; \ No newline at end of file diff --git a/database/queries/job.sql b/database/queries/job.sql new file mode 100644 index 00000000..2db186e5 --- /dev/null +++ b/database/queries/job.sql @@ -0,0 +1,5 @@ +-- name: GetJob :one +SELECT id, clientId, canSync FROM jobs WHERE id = $1 LIMIT 1; + +-- name: CreateJob :one +INSERT INTO jobs (clientId) VALUES ($1) RETURNING id; \ No newline at end of file diff --git a/database/queries/query.sql b/database/queries/query.sql index 6ae17076..4d3dd3bb 100644 --- a/database/queries/query.sql +++ b/database/queries/query.sql @@ -28,4 +28,9 @@ UPDATE queryConfigs SET removedVersion = $1 WHERE queryId = $2 and removedVersio -- name: AllQueriesExist :one SELECT COUNT(*) = COUNT(DISTINCT id) AS all_exist FROM unnest($1::uuid[]) AS input_id - LEFT JOIN queries ON input_id = queries.id; \ No newline at end of file + LEFT JOIN queries ON input_id = queries.id; + +-- name: IsQueryInDependencyTree :one +SELECT EXISTS ( + SELECT 1 FROM queryActiveDependencies WHERE id = any($1) and requiredQueryId = $2 or $2 = any($1) +); \ No newline at end of file diff --git a/database/queries/result.sql b/database/queries/result.sql index 20cdeeb4..75445662 100644 --- a/database/queries/result.sql +++ b/database/queries/result.sql @@ -1,5 +1,6 @@ -- name: ListResultsByDocumentID :many -SELECT id, queryId, queryVersion FROM results where documentId = $1 and cleanVersion >= $2 and textVersion >= $3; +SELECT id, queryId, queryVersion FROM results + where documentId = $1 and cleanVersion >= $2 and textVersion >= $3; -- name: ListResultValuesByID :many SELECT id, queryId, value FROM results where id = ANY($1); diff --git a/internal/database/connection.go b/internal/database/connection.go index 47344e6e..fd3ae4d5 100644 --- a/internal/database/connection.go +++ b/internal/database/connection.go @@ -5,6 +5,7 @@ import ( "fmt" "log" "net/url" + "queryorchestration/internal/database/repository" "queryorchestration/internal/server/env" "github.com/docker/go-connections/nat" @@ -108,3 +109,27 @@ func GetDBPool(ctx context.Context) *pgxpool.Pool { return pool } + +func ExecuteTransaction(ctx context.Context, db *Connection, executeQueries func(context.Context, *repository.Queries) error) error { + tx, err := db.Pool.Begin(ctx) + if err != nil { + return err + } + defer func() { + _ = tx.Rollback(ctx) + }() + + qtx := db.Queries.WithTx(tx) + + err = executeQueries(ctx, qtx) + if err != nil { + return err + } + + err = tx.Commit(ctx) + if err != nil { + return err + } + + return nil +} diff --git a/internal/database/connection_test.go b/internal/database/connection_test.go index a7a20140..fb4eccdb 100644 --- a/internal/database/connection_test.go +++ b/internal/database/connection_test.go @@ -5,9 +5,12 @@ import ( "os" "path" "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" "queryorchestration/internal/test" "testing" + "github.com/google/uuid" + "github.com/pashagolub/pgxmock/v3" "github.com/stretchr/testify/assert" ) @@ -26,3 +29,37 @@ func TestDBConn(t *testing.T) { pool := database.GetDBPool(ctx) assert.NotNil(t, pool) } + +func TestExecuteTransation(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + + clientID := database.MustToDBUUID(uuid.New()) + clientName := "example_client" + + pool.ExpectBegin() + pool.ExpectQuery("name: CreateClient :one").WithArgs(clientName). + WillReturnRows( + pgxmock.NewRows([]string{"id"}). + AddRow(clientID), + ) + pool.ExpectCommit() + + err = database.ExecuteTransaction(ctx, db, func(ctx context.Context, q *repository.Queries) error { + id, err := q.CreateClient(ctx, "example_client") + assert.Nil(t, err) + assert.Equal(t, clientID, id) + + return nil + }) + assert.Nil(t, err) +} diff --git a/internal/database/repository/collector.sql.go b/internal/database/repository/collector.sql.go index baa0149a..37b63584 100644 --- a/internal/database/repository/collector.sql.go +++ b/internal/database/repository/collector.sql.go @@ -11,6 +11,30 @@ import ( "github.com/jackc/pgx/v5/pgtype" ) +const addCollectorCodeVersion = `-- name: AddCollectorCodeVersion :exec +INSERT INTO collectorCodeVersions (collectorId, addedVersion, minCleanVersion, minTextVersion) VALUES ($1, $2, $3, $4) +` + +type AddCollectorCodeVersionParams struct { + Collectorid pgtype.UUID `db:"collectorid"` + Addedversion int32 `db:"addedversion"` + Mincleanversion int32 `db:"mincleanversion"` + Mintextversion int32 `db:"mintextversion"` +} + +// AddCollectorCodeVersion +// +// INSERT INTO collectorCodeVersions (collectorId, addedVersion, minCleanVersion, minTextVersion) VALUES ($1, $2, $3, $4) +func (q *Queries) AddCollectorCodeVersion(ctx context.Context, arg *AddCollectorCodeVersionParams) error { + _, err := q.db.Exec(ctx, addCollectorCodeVersion, + arg.Collectorid, + arg.Addedversion, + arg.Mincleanversion, + arg.Mintextversion, + ) + return err +} + const addCollectorQuery = `-- name: AddCollectorQuery :exec INSERT INTO collectorQueries (collectorId, name, queryId, addedVersion) VALUES ($1, $2, $3, $4) ` @@ -36,20 +60,14 @@ func (q *Queries) AddCollectorQuery(ctx context.Context, arg *AddCollectorQueryP } const createCollector = `-- name: CreateCollector :one -INSERT INTO collectors (jobId, minCleanVersion, minTextVersion) VALUES ($1, $2, $3) RETURNING id +INSERT INTO collectors (jobId) VALUES ($1) RETURNING id ` -type CreateCollectorParams struct { - Jobid pgtype.UUID `db:"jobid"` - Mincleanversion int32 `db:"mincleanversion"` - Mintextversion int32 `db:"mintextversion"` -} - // CreateCollector // -// INSERT INTO collectors (jobId, minCleanVersion, minTextVersion) VALUES ($1, $2, $3) RETURNING id -func (q *Queries) CreateCollector(ctx context.Context, arg *CreateCollectorParams) (pgtype.UUID, error) { - row := q.db.QueryRow(ctx, createCollector, arg.Jobid, arg.Mincleanversion, arg.Mintextversion) +// INSERT INTO collectors (jobId) VALUES ($1) RETURNING id +func (q *Queries) CreateCollector(ctx context.Context, jobid pgtype.UUID) (pgtype.UUID, error) { + row := q.db.QueryRow(ctx, createCollector, jobid) var id pgtype.UUID err := row.Scan(&id) return id, err @@ -131,3 +149,56 @@ func (q *Queries) ListCollectorQueries(ctx context.Context, collectorid pgtype.U } return items, nil } + +const removeCollectorCodeVersion = `-- name: RemoveCollectorCodeVersion :exec +UPDATE collectorCodeVersions SET removedVersion = $1 WHERE collectorId = $2 and removedVersion is null +` + +type RemoveCollectorCodeVersionParams struct { + Removedversion *int32 `db:"removedversion"` + Collectorid pgtype.UUID `db:"collectorid"` +} + +// RemoveCollectorCodeVersion +// +// UPDATE collectorCodeVersions SET removedVersion = $1 WHERE collectorId = $2 and removedVersion is null +func (q *Queries) RemoveCollectorCodeVersion(ctx context.Context, arg *RemoveCollectorCodeVersionParams) error { + _, err := q.db.Exec(ctx, removeCollectorCodeVersion, arg.Removedversion, arg.Collectorid) + return err +} + +const removeCollectorQuery = `-- name: RemoveCollectorQuery :exec +UPDATE collectorQueries SET removedVersion = $1 WHERE queryId = $2 and collectorId = $3 and removedVersion is null +` + +type RemoveCollectorQueryParams struct { + Removedversion *int32 `db:"removedversion"` + Queryid pgtype.UUID `db:"queryid"` + Collectorid pgtype.UUID `db:"collectorid"` +} + +// RemoveCollectorQuery +// +// UPDATE collectorQueries SET removedVersion = $1 WHERE queryId = $2 and collectorId = $3 and removedVersion is null +func (q *Queries) RemoveCollectorQuery(ctx context.Context, arg *RemoveCollectorQueryParams) error { + _, err := q.db.Exec(ctx, removeCollectorQuery, arg.Removedversion, arg.Queryid, arg.Collectorid) + return err +} + +const updateCollector = `-- name: UpdateCollector :exec +UPDATE collectors SET latestVersion = $1, activeVersion = $2 WHERE id = $3 +` + +type UpdateCollectorParams struct { + Latestversion int32 `db:"latestversion"` + Activeversion int32 `db:"activeversion"` + ID pgtype.UUID `db:"id"` +} + +// UpdateCollector +// +// UPDATE collectors SET latestVersion = $1, activeVersion = $2 WHERE id = $3 +func (q *Queries) UpdateCollector(ctx context.Context, arg *UpdateCollectorParams) error { + _, err := q.db.Exec(ctx, updateCollector, arg.Latestversion, arg.Activeversion, arg.ID) + return err +} diff --git a/internal/database/repository/collector_test.go b/internal/database/repository/collector_test.go index 826d2f6b..041084e2 100644 --- a/internal/database/repository/collector_test.go +++ b/internal/database/repository/collector_test.go @@ -36,15 +36,14 @@ func TestCollector(t *testing.T) { }) assert.Nil(t, err) - jobId := database.MustToDBUUID(uuid.New()) + clientId, err := queries.CreateClient(ctx, "example_client") + assert.Nil(t, err) + jobId, err := queries.CreateJob(ctx, clientId) + assert.Nil(t, err) minCleanVersion := int32(2) minTextVersion := int32(4) - collId, err := queries.CreateCollector(ctx, &repository.CreateCollectorParams{ - Jobid: jobId, - Mincleanversion: minCleanVersion, - Mintextversion: minTextVersion, - }) + collId, err := queries.CreateCollector(ctx, jobId) assert.Nil(t, err) coll, err := queries.GetCollector(ctx, collId) @@ -52,8 +51,8 @@ func TestCollector(t *testing.T) { assert.EqualExportedValues(t, &repository.Fullactivecollector{ ID: collId, Jobid: jobId, - Mincleanversion: minCleanVersion, - Mintextversion: minTextVersion, + Mincleanversion: nil, + Mintextversion: nil, Activeversion: 1, Latestversion: 1, }, coll) @@ -63,8 +62,8 @@ func TestCollector(t *testing.T) { assert.EqualExportedValues(t, &repository.Fullactivecollector{ ID: collId, Jobid: jobId, - Mincleanversion: minCleanVersion, - Mintextversion: minTextVersion, + Mincleanversion: nil, + Mintextversion: nil, Activeversion: 1, Latestversion: 1, }, coll) @@ -77,13 +76,21 @@ func TestCollector(t *testing.T) { }) assert.Nil(t, err) + err = queries.AddCollectorCodeVersion(ctx, &repository.AddCollectorCodeVersionParams{ + Collectorid: collId, + Addedversion: 1, + Mincleanversion: minCleanVersion, + Mintextversion: minTextVersion, + }) + assert.Nil(t, err) + coll, err = queries.GetCollector(ctx, collId) assert.Nil(t, err) assert.EqualExportedValues(t, &repository.Fullactivecollector{ ID: collId, Jobid: jobId, - Mincleanversion: minCleanVersion, - Mintextversion: minTextVersion, + Mincleanversion: &minCleanVersion, + Mintextversion: &minTextVersion, Activeversion: 1, Latestversion: 1, Fields: []byte(fmt.Sprintf("{\"example_key\": \"%s\"}", database.MustToUUID(jsonId).String())), @@ -108,4 +115,37 @@ func TestCollector(t *testing.T) { Requiredids: []pgtype.UUID{database.MustToDBUUID(uuid.Nil)}, }, }, qs) + + removeV := int32(2) + err = queries.RemoveCollectorQuery(ctx, &repository.RemoveCollectorQueryParams{ + Collectorid: collId, + Queryid: jsonId, + Removedversion: &removeV, + }) + assert.Nil(t, err) + + err = queries.RemoveCollectorCodeVersion(ctx, &repository.RemoveCollectorCodeVersionParams{ + Collectorid: collId, + Removedversion: &removeV, + }) + assert.Nil(t, err) + + err = queries.UpdateCollector(ctx, &repository.UpdateCollectorParams{ + ID: collId, + Latestversion: 2, + Activeversion: 2, + }) + assert.Nil(t, err) + + coll, err = queries.GetCollector(ctx, collId) + assert.Nil(t, err) + assert.EqualExportedValues(t, &repository.Fullactivecollector{ + ID: collId, + Jobid: jobId, + Mincleanversion: nil, + Mintextversion: nil, + Activeversion: 2, + Latestversion: 2, + Fields: []byte(nil), + }, coll) } diff --git a/internal/database/repository/document.sql.go b/internal/database/repository/document.sql.go new file mode 100644 index 00000000..1647a243 --- /dev/null +++ b/internal/database/repository/document.sql.go @@ -0,0 +1,40 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.27.0 +// source: document.sql + +package repository + +import ( + "context" + + "github.com/jackc/pgx/v5/pgtype" +) + +const createDocument = `-- name: CreateDocument :one +INSERT INTO documents (jobId) VALUES ($1) RETURNING id +` + +// CreateDocument +// +// INSERT INTO documents (jobId) VALUES ($1) RETURNING id +func (q *Queries) CreateDocument(ctx context.Context, jobid pgtype.UUID) (pgtype.UUID, error) { + row := q.db.QueryRow(ctx, createDocument, jobid) + var id pgtype.UUID + err := row.Scan(&id) + return id, err +} + +const getDocument = `-- name: GetDocument :one +SELECT id, jobId FROM documents WHERE id = $1 LIMIT 1 +` + +// GetDocument +// +// SELECT id, jobId FROM documents WHERE id = $1 LIMIT 1 +func (q *Queries) GetDocument(ctx context.Context, id pgtype.UUID) (*Document, error) { + row := q.db.QueryRow(ctx, getDocument, id) + var i Document + err := row.Scan(&i.ID, &i.Jobid) + return &i, err +} diff --git a/internal/database/repository/document_test.go b/internal/database/repository/document_test.go new file mode 100644 index 00000000..4ecb9dc4 --- /dev/null +++ b/internal/database/repository/document_test.go @@ -0,0 +1,40 @@ +package repository_test + +import ( + "context" + "os" + "path" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/test" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestDocument(t *testing.T) { + ctx := context.Background() + db, cleanup := test.CreateDB(t, ctx, &test.CreateDatabaseConfig{ + Migrations: &database.MigrationConfig{ + BasePath: path.Join(os.Getenv("PWD"), "../../.."), + }}) + defer cleanup() + + queries := repository.New(db.Pool) + + clientId, err := queries.CreateClient(ctx, "example_client") + assert.Nil(t, err) + jobId, err := queries.CreateJob(ctx, clientId) + assert.Nil(t, err) + + id, err := queries.CreateDocument(ctx, jobId) + assert.Nil(t, err) + assert.NotEmpty(t, id) + + doc, err := queries.GetDocument(ctx, id) + assert.Nil(t, err) + assert.EqualExportedValues(t, &repository.Document{ + ID: id, + Jobid: jobId, + }, doc) +} diff --git a/internal/database/repository/job.sql.go b/internal/database/repository/job.sql.go new file mode 100644 index 00000000..e9277ca2 --- /dev/null +++ b/internal/database/repository/job.sql.go @@ -0,0 +1,40 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.27.0 +// source: job.sql + +package repository + +import ( + "context" + + "github.com/jackc/pgx/v5/pgtype" +) + +const createJob = `-- name: CreateJob :one +INSERT INTO jobs (clientId) VALUES ($1) RETURNING id +` + +// CreateJob +// +// INSERT INTO jobs (clientId) VALUES ($1) RETURNING id +func (q *Queries) CreateJob(ctx context.Context, clientid pgtype.UUID) (pgtype.UUID, error) { + row := q.db.QueryRow(ctx, createJob, clientid) + var id pgtype.UUID + err := row.Scan(&id) + return id, err +} + +const getJob = `-- name: GetJob :one +SELECT id, clientId, canSync FROM jobs WHERE id = $1 LIMIT 1 +` + +// GetJob +// +// SELECT id, clientId, canSync FROM jobs WHERE id = $1 LIMIT 1 +func (q *Queries) GetJob(ctx context.Context, id pgtype.UUID) (*Job, error) { + row := q.db.QueryRow(ctx, getJob, id) + var i Job + err := row.Scan(&i.ID, &i.Clientid, &i.Cansync) + return &i, err +} diff --git a/internal/database/repository/job_test.go b/internal/database/repository/job_test.go new file mode 100644 index 00000000..3176716d --- /dev/null +++ b/internal/database/repository/job_test.go @@ -0,0 +1,39 @@ +package repository_test + +import ( + "context" + "os" + "path" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/test" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestJob(t *testing.T) { + ctx := context.Background() + db, cleanup := test.CreateDB(t, ctx, &test.CreateDatabaseConfig{ + Migrations: &database.MigrationConfig{ + BasePath: path.Join(os.Getenv("PWD"), "../../.."), + }}) + defer cleanup() + + queries := repository.New(db.Pool) + + clientId, err := queries.CreateClient(ctx, "example_client") + assert.Nil(t, err) + + id, err := queries.CreateJob(ctx, clientId) + assert.Nil(t, err) + assert.NotEmpty(t, id) + + job, err := queries.GetJob(ctx, id) + assert.Nil(t, err) + assert.EqualExportedValues(t, &repository.Job{ + ID: id, + Clientid: clientId, + Cansync: false, + }, job) +} diff --git a/internal/database/repository/models.go b/internal/database/repository/models.go index 8f73bb95..8575a86f 100644 --- a/internal/database/repository/models.go +++ b/internal/database/repository/models.go @@ -62,12 +62,6 @@ func (e Querytype) Valid() bool { return false } -type Activecollectorswithrequiredid struct { - ID pgtype.UUID `db:"id"` - Activeversion int32 `db:"activeversion"` - Queryids interface{} `db:"queryids"` -} - type Client struct { ID pgtype.UUID `db:"id"` Name string `db:"name"` @@ -75,12 +69,19 @@ type Client struct { } type Collector struct { + ID pgtype.UUID `db:"id"` + Jobid pgtype.UUID `db:"jobid"` + Latestversion int32 `db:"latestversion"` + Activeversion int32 `db:"activeversion"` +} + +type Collectorcodeversion struct { ID pgtype.UUID `db:"id"` - Jobid pgtype.UUID `db:"jobid"` + Collectorid pgtype.UUID `db:"collectorid"` Mincleanversion int32 `db:"mincleanversion"` Mintextversion int32 `db:"mintextversion"` - Latestversion int32 `db:"latestversion"` - Activeversion int32 `db:"activeversion"` + Addedversion int32 `db:"addedversion"` + Removedversion *int32 `db:"removedversion"` } type Collectorquery struct { @@ -100,11 +101,16 @@ type Collectorquerydependencytree struct { Requiredids []pgtype.UUID `db:"requiredids"` } +type Document struct { + ID pgtype.UUID `db:"id"` + Jobid pgtype.UUID `db:"jobid"` +} + type Fullactivecollector struct { ID pgtype.UUID `db:"id"` Jobid pgtype.UUID `db:"jobid"` - Mincleanversion int32 `db:"mincleanversion"` - Mintextversion int32 `db:"mintextversion"` + Mincleanversion *int32 `db:"mincleanversion"` + Mintextversion *int32 `db:"mintextversion"` Activeversion int32 `db:"activeversion"` Latestversion int32 `db:"latestversion"` Fields []byte `db:"fields"` @@ -119,6 +125,12 @@ type Fullactivequery struct { Requiredids []pgtype.UUID `db:"requiredids"` } +type Job struct { + ID pgtype.UUID `db:"id"` + Clientid pgtype.UUID `db:"clientid"` + Cansync bool `db:"cansync"` +} + type Query struct { ID pgtype.UUID `db:"id"` Latestversion int32 `db:"latestversion"` @@ -126,6 +138,11 @@ type Query struct { Type Querytype `db:"type"` } +type Queryactivedependency struct { + ID pgtype.UUID `db:"id"` + Requiredqueryid pgtype.UUID `db:"requiredqueryid"` +} + type Queryconfig struct { ID pgtype.UUID `db:"id"` Queryid pgtype.UUID `db:"queryid"` diff --git a/internal/database/repository/query.sql.go b/internal/database/repository/query.sql.go index d4d7a35d..470feb29 100644 --- a/internal/database/repository/query.sql.go +++ b/internal/database/repository/query.sql.go @@ -124,6 +124,29 @@ func (q *Queries) GetQueryConfig(ctx context.Context, arg *GetQueryConfigParams) return &i, err } +const isQueryInDependencyTree = `-- name: IsQueryInDependencyTree :one +SELECT EXISTS ( + SELECT 1 FROM queryActiveDependencies WHERE id = any($1) and requiredQueryId = $2 or $2 = any($1) +) +` + +type IsQueryInDependencyTreeParams struct { + ID []pgtype.UUID `db:"id"` + Requiredqueryid pgtype.UUID `db:"requiredqueryid"` +} + +// IsQueryInDependencyTree +// +// SELECT EXISTS ( +// SELECT 1 FROM queryActiveDependencies WHERE id = any($1) and requiredQueryId = $2 or $2 = any($1) +// ) +func (q *Queries) IsQueryInDependencyTree(ctx context.Context, arg *IsQueryInDependencyTreeParams) (bool, error) { + row := q.db.QueryRow(ctx, isQueryInDependencyTree, arg.ID, arg.Requiredqueryid) + var exists bool + err := row.Scan(&exists) + return exists, err +} + const listQueries = `-- name: ListQueries :many SELECT id, type, activeversion, latestversion, config, requiredids FROM fullActiveQueries ` diff --git a/internal/database/repository/query_test.go b/internal/database/repository/query_test.go index 2226bc70..263b8c39 100644 --- a/internal/database/repository/query_test.go +++ b/internal/database/repository/query_test.go @@ -145,6 +145,74 @@ func TestQueries(t *testing.T) { assert.False(t, all_exist) } +func TestQueryDependencyTree(t *testing.T) { + ctx := context.Background() + + db, cleanup := test.CreateDB(t, ctx, &test.CreateDatabaseConfig{ + Migrations: &database.MigrationConfig{ + BasePath: path.Join(os.Getenv("PWD"), "../../.."), + }}) + defer cleanup() + + queries := repository.New(db.Pool) + + contextQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeContextFull)) + assert.Nil(t, err) + + jsonQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeJsonExtractor)) + assert.Nil(t, err) + err = queries.AddRequiredQuery(ctx, &repository.AddRequiredQueryParams{ + Queryid: jsonQueryID, + Requiredqueryid: contextQueryID, + Addedversion: 1, + }) + assert.Nil(t, err) + + secondJsonQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeJsonExtractor)) + assert.Nil(t, err) + err = queries.AddRequiredQuery(ctx, &repository.AddRequiredQueryParams{ + Queryid: secondJsonQueryID, + Requiredqueryid: jsonQueryID, + Addedversion: 1, + }) + assert.Nil(t, err) + + isdependent, err := queries.IsQueryInDependencyTree(ctx, &repository.IsQueryInDependencyTreeParams{ + Requiredqueryid: jsonQueryID, + ID: []pgtype.UUID{contextQueryID}, + }) + assert.Nil(t, err) + assert.False(t, isdependent) + + isdependent, err = queries.IsQueryInDependencyTree(ctx, &repository.IsQueryInDependencyTreeParams{ + Requiredqueryid: jsonQueryID, + ID: []pgtype.UUID{secondJsonQueryID}, + }) + assert.Nil(t, err) + assert.True(t, isdependent) + + isdependent, err = queries.IsQueryInDependencyTree(ctx, &repository.IsQueryInDependencyTreeParams{ + Requiredqueryid: jsonQueryID, + ID: []pgtype.UUID{jsonQueryID}, + }) + assert.Nil(t, err) + assert.True(t, isdependent) + + isdependent, err = queries.IsQueryInDependencyTree(ctx, &repository.IsQueryInDependencyTreeParams{ + Requiredqueryid: secondJsonQueryID, + ID: []pgtype.UUID{jsonQueryID, contextQueryID}, + }) + assert.Nil(t, err) + assert.False(t, isdependent) + + isdependent, err = queries.IsQueryInDependencyTree(ctx, &repository.IsQueryInDependencyTreeParams{ + Requiredqueryid: contextQueryID, + ID: []pgtype.UUID{jsonQueryID, secondJsonQueryID}, + }) + assert.Nil(t, err) + assert.True(t, isdependent) +} + func TestQueriesList(t *testing.T) { ctx := context.Background() diff --git a/internal/database/repository/result.sql.go b/internal/database/repository/result.sql.go index 4472c327..b38b5fc6 100644 --- a/internal/database/repository/result.sql.go +++ b/internal/database/repository/result.sql.go @@ -45,7 +45,8 @@ func (q *Queries) ListResultValuesByID(ctx context.Context, id []pgtype.UUID) ([ } const listResultsByDocumentID = `-- name: ListResultsByDocumentID :many -SELECT id, queryId, queryVersion FROM results where documentId = $1 and cleanVersion >= $2 and textVersion >= $3 +SELECT id, queryId, queryVersion FROM results + where documentId = $1 and cleanVersion >= $2 and textVersion >= $3 ` type ListResultsByDocumentIDParams struct { @@ -62,7 +63,8 @@ type ListResultsByDocumentIDRow struct { // ListResultsByDocumentID // -// SELECT id, queryId, queryVersion FROM results where documentId = $1 and cleanVersion >= $2 and textVersion >= $3 +// SELECT id, queryId, queryVersion FROM results +// where documentId = $1 and cleanVersion >= $2 and textVersion >= $3 func (q *Queries) ListResultsByDocumentID(ctx context.Context, arg *ListResultsByDocumentIDParams) ([]*ListResultsByDocumentIDRow, error) { rows, err := q.db.Query(ctx, listResultsByDocumentID, arg.Documentid, arg.Cleanversion, arg.Textversion) if err != nil { diff --git a/internal/database/repository/result_test.go b/internal/database/repository/result_test.go index 9bec883c..521ea817 100644 --- a/internal/database/repository/result_test.go +++ b/internal/database/repository/result_test.go @@ -9,7 +9,6 @@ import ( "queryorchestration/internal/test" "testing" - "github.com/google/uuid" "github.com/jackc/pgx/v5/pgtype" "github.com/stretchr/testify/assert" ) @@ -28,7 +27,12 @@ func TestResults(t *testing.T) { jsonQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeJsonExtractor)) assert.Nil(t, err) - documentID := database.MustToDBUUID(uuid.New()) + clientId, err := queries.CreateClient(ctx, "example_client") + assert.Nil(t, err) + jobId, err := queries.CreateJob(ctx, clientId) + assert.Nil(t, err) + documentID, err := queries.CreateDocument(ctx, jobId) + assert.Nil(t, err) jsonQuery, err := queries.GetQuery(ctx, jsonQueryID) assert.Nil(t, err) diff --git a/internal/document_clean/service.go b/internal/document_clean/service.go new file mode 100644 index 00000000..3de769fd --- /dev/null +++ b/internal/document_clean/service.go @@ -0,0 +1,18 @@ +package documentclean + +import "errors" + +type Service struct { +} + +func New() *Service { + return &Service{} +} + +func (s *Service) IsValidVersion(v int32) error { + if v <= 0 { + return errors.New("document clean code version must be > 0") + } + + return nil +} diff --git a/internal/document_clean/service_test.go b/internal/document_clean/service_test.go new file mode 100644 index 00000000..adaf4266 --- /dev/null +++ b/internal/document_clean/service_test.go @@ -0,0 +1,13 @@ +package documentclean_test + +import ( + documentclean "queryorchestration/internal/document_clean" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestService(t *testing.T) { + svc := documentclean.New() + assert.NotNil(t, svc) +} diff --git a/internal/job/collector/create.go b/internal/job/collector/create.go new file mode 100644 index 00000000..3e9f988c --- /dev/null +++ b/internal/job/collector/create.go @@ -0,0 +1,129 @@ +package collector + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/server/validation" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" +) + +type CreateParams struct { + JobID uuid.UUID + MinCleanVersion *int32 + MinTextVersion *int32 + Fields map[string]uuid.UUID +} + +func (s *Service) Create(ctx context.Context, params *CreateParams) (uuid.UUID, error) { + dbparams, err := s.getCreateParams(ctx, params) + if err != nil { + return uuid.Nil, err + } + + id, err := s.submitCreate(ctx, dbparams) + if err != nil { + return uuid.Nil, err + } + + return id, nil +} + +type dbCreateParams struct { + JobID pgtype.UUID + MinCleanVersion *int32 + MinTextVersion *int32 + Fields map[string]pgtype.UUID +} + +type fields struct { + values *[]uuid.UUID +} + +func (f *fields) GetRequiredQueryIDs() *[]uuid.UUID { + return f.values +} + +func (f *fields) SetRequiredQueryIDs(v *[]uuid.UUID) { + f.values = v +} + +func (s *Service) getCreateParams(ctx context.Context, params *CreateParams) (*dbCreateParams, error) { + minClean := params.MinCleanVersion + if minClean != nil { + err := s.svc.DocumentClean.IsValidVersion(*minClean) + if err != nil { + return nil, err + } + } + + minText := params.MinTextVersion + if minText != nil { + err := s.svc.TextExtraction.IsValidVersion(*minText) + if err != nil { + return nil, err + } + } + + fields, err := s.normalizeFieldsToDB(ctx, params.Fields) + if err != nil { + return nil, err + } + + return &dbCreateParams{ + JobID: database.MustToDBUUID(params.JobID), + MinCleanVersion: minClean, + MinTextVersion: minText, + Fields: fields, + }, nil +} + +func (s *Service) submitCreate(ctx context.Context, params *dbCreateParams) (uuid.UUID, error) { + var dbID pgtype.UUID + err := database.ExecuteTransaction(ctx, s.db, func(ctx context.Context, qtx *repository.Queries) error { + dID, err := qtx.CreateCollector(ctx, params.JobID) + if err != nil { + return err + } + dbID = dID + latestVersion := int32(1) + + if params.MinCleanVersion != nil || params.MinTextVersion != nil { + err = qtx.AddCollectorCodeVersion(ctx, &repository.AddCollectorCodeVersionParams{ + Collectorid: dbID, + Mincleanversion: validation.GetUpdatedValue(1, params.MinCleanVersion), + Mintextversion: validation.GetUpdatedValue(1, params.MinTextVersion), + Addedversion: latestVersion, + }) + if err != nil { + return err + } + } + + for key, field := range params.Fields { + err = qtx.AddCollectorQuery(ctx, &repository.AddCollectorQueryParams{ + Collectorid: dbID, + Name: key, + Queryid: field, + Addedversion: latestVersion, + }) + if err != nil { + return err + } + } + + return nil + }) + if err != nil { + return uuid.Nil, err + } + + id, err := database.ToUUID(dbID) + if err != nil { + return uuid.Nil, err + } + + return id, nil +} diff --git a/internal/job/collector/create_test.go b/internal/job/collector/create_test.go new file mode 100644 index 00000000..d9df07dc --- /dev/null +++ b/internal/job/collector/create_test.go @@ -0,0 +1,64 @@ +package collector_test + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/job/collector" + "queryorchestration/internal/query" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +func TestCreate(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + + svc := collector.New(db, &collector.Services{ + Query: query.New(db), + }) + + id := uuid.New() + minCleanV := int32(2) + minTextV := int32(4) + create := collector.CreateParams{ + JobID: uuid.New(), + MinCleanVersion: &minCleanV, + MinTextVersion: &minTextV, + Fields: map[string]uuid.UUID{ + "example_key": uuid.New(), + }, + } + + pool.ExpectQuery("name: AllQueriesExist :one").WithArgs(database.MustToDBUUIDArray([]uuid.UUID{create.Fields["example_key"]})).WillReturnRows( + pgxmock.NewRows([]string{"all_exist"}). + AddRow(true), + ) + pool.ExpectBeginTx(pgx.TxOptions{}) + pool.ExpectQuery("name: CreateCollector :one").WithArgs(database.MustToDBUUID(create.JobID)).WillReturnRows( + pgxmock.NewRows([]string{"id"}). + AddRow(database.MustToDBUUID(id)), + ) + pool.ExpectExec("name: AddCollectorCodeVersion :exec").WithArgs(database.MustToDBUUID(id), int32(1), *create.MinCleanVersion, *create.MinTextVersion). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectExec("name: AddCollectorQuery :exec").WithArgs(database.MustToDBUUID(id), "example_key", database.MustToDBUUID(create.Fields["example_key"]), int32(1)). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectCommit() + + aid, err := svc.Create(ctx, &create) + assert.Nil(t, err) + assert.Equal(t, id, aid) +} diff --git a/internal/job/collector/createprivate_test.go b/internal/job/collector/createprivate_test.go new file mode 100644 index 00000000..82b04af0 --- /dev/null +++ b/internal/job/collector/createprivate_test.go @@ -0,0 +1,126 @@ +package collector + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/query" + "testing" + + "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" +) + +func TestGetCreateParams(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + + svc := Service{ + db: db, + svc: &Services{ + Query: query.New(db), + }, + } + + minCleanV := int32(2) + minTextV := int32(4) + params := CreateParams{ + JobID: uuid.New(), + MinCleanVersion: &minCleanV, + MinTextVersion: &minTextV, + Fields: map[string]uuid.UUID{ + "example_key": uuid.New(), + }, + } + + pool.ExpectQuery("name: AllQueriesExist :one").WithArgs(database.MustToDBUUIDArray([]uuid.UUID{params.Fields["example_key"]})).WillReturnRows( + pgxmock.NewRows([]string{"all_exist"}). + AddRow(true), + ) + + dbparams, err := svc.getCreateParams(ctx, ¶ms) + assert.Nil(t, err) + assert.EqualExportedValues(t, &dbCreateParams{ + JobID: database.MustToDBUUID(params.JobID), + MinCleanVersion: &minCleanV, + MinTextVersion: &minTextV, + Fields: map[string]pgtype.UUID{ + "example_key": database.MustToDBUUID(params.Fields["example_key"]), + }, + }, dbparams) + + params.Fields["second_key"] = params.Fields["example_key"] + assert.Len(t, params.Fields, 2) + _, err = svc.getCreateParams(ctx, ¶ms) + assert.Error(t, err) +} + +func TestSubmitCreate(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + + svc := Service{ + db: db, + svc: &Services{ + Query: query.New(db), + }, + } + + minCleanV := int32(2) + minTextV := int32(4) + id := uuid.New() + params := dbCreateParams{ + JobID: database.MustToDBUUID(uuid.New()), + MinCleanVersion: &minCleanV, + MinTextVersion: &minTextV, + Fields: map[string]pgtype.UUID{ + "example_key": database.MustToDBUUID(uuid.New()), + }, + } + + pool.ExpectBeginTx(pgx.TxOptions{}) + pool.ExpectQuery("name: CreateCollector :one").WithArgs(params.JobID).WillReturnRows( + pgxmock.NewRows([]string{"id"}). + AddRow(database.MustToDBUUID(id)), + ) + pool.ExpectExec("name: AddCollectorCodeVersion :exec").WithArgs(database.MustToDBUUID(id), int32(1), *params.MinCleanVersion, *params.MinTextVersion). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectExec("name: AddCollectorQuery :exec").WithArgs(database.MustToDBUUID(id), "example_key", params.Fields["example_key"], int32(1)). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectCommit() + + aid, err := svc.submitCreate(ctx, ¶ms) + assert.Nil(t, err) + assert.Equal(t, id, aid) +} + +func TestFields(t *testing.T) { + f := fields{} + assert.Nil(t, f.GetRequiredQueryIDs()) + + values := []uuid.UUID{uuid.New()} + f.SetRequiredQueryIDs(&values) + assert.NotNil(t, f.values) + assert.ElementsMatch(t, *f.values, *f.GetRequiredQueryIDs()) +} diff --git a/internal/job/collector/get_test.go b/internal/job/collector/get_test.go index 385d6609..c5ed884d 100644 --- a/internal/job/collector/get_test.go +++ b/internal/job/collector/get_test.go @@ -26,13 +26,15 @@ func TestGet(t *testing.T) { Pool: pool, } - svc := collector.New(db) + svc := collector.New(db, &collector.Services{}) + minCleanV := int32(2) + minTextV := int32(4) ogc := collector.Collector{ ID: uuid.New(), JobID: uuid.New(), - MinCleanVersion: 2, - MinTextVersion: 4, + MinCleanVersion: minCleanV, + MinTextVersion: minTextV, Fields: map[string]uuid.UUID{ "example_key": uuid.New(), }, @@ -41,7 +43,7 @@ func TestGet(t *testing.T) { pool.ExpectQuery("name: GetCollector :one").WithArgs(database.MustToDBUUID(ogc.ID)). WillReturnRows( pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}). - AddRow(database.MustToDBUUID(ogc.ID), database.MustToDBUUID(ogc.JobID), ogc.MinCleanVersion, ogc.MinTextVersion, ogc.ActiveVersion, ogc.LatestVersion, []byte(fmt.Sprintf("{\"example_key\":\"%s\"}", ogc.Fields["example_key"].String()))), + AddRow(database.MustToDBUUID(ogc.ID), database.MustToDBUUID(ogc.JobID), &ogc.MinCleanVersion, &ogc.MinTextVersion, ogc.ActiveVersion, ogc.LatestVersion, []byte(fmt.Sprintf("{\"example_key\":\"%s\"}", ogc.Fields["example_key"].String()))), ) coll, err := svc.Get(ctx, ogc.ID) @@ -62,13 +64,15 @@ func TestGetByJobID(t *testing.T) { Pool: pool, } - svc := collector.New(db) + svc := collector.New(db, &collector.Services{}) + minCleanV := int32(2) + minTextV := int32(4) ogc := collector.Collector{ ID: uuid.New(), JobID: uuid.New(), - MinCleanVersion: 2, - MinTextVersion: 4, + MinCleanVersion: minCleanV, + MinTextVersion: minTextV, Fields: map[string]uuid.UUID{ "example_key": uuid.New(), }, @@ -77,7 +81,7 @@ func TestGetByJobID(t *testing.T) { pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(database.MustToDBUUID(ogc.JobID)). WillReturnRows( pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}). - AddRow(database.MustToDBUUID(ogc.ID), database.MustToDBUUID(ogc.JobID), ogc.MinCleanVersion, ogc.MinTextVersion, ogc.ActiveVersion, ogc.LatestVersion, []byte(fmt.Sprintf("{\"example_key\":\"%s\"}", ogc.Fields["example_key"].String()))), + AddRow(database.MustToDBUUID(ogc.ID), database.MustToDBUUID(ogc.JobID), &ogc.MinCleanVersion, &ogc.MinTextVersion, ogc.ActiveVersion, ogc.LatestVersion, []byte(fmt.Sprintf("{\"example_key\":\"%s\"}", ogc.Fields["example_key"].String()))), ) coll, err := svc.GetByJobID(ctx, ogc.JobID) diff --git a/internal/job/collector/parse.go b/internal/job/collector/parse.go index dc5445dd..f1a45416 100644 --- a/internal/job/collector/parse.go +++ b/internal/job/collector/parse.go @@ -25,11 +25,21 @@ func parseDBCollector(c *repository.Fullactivecollector) (*Collector, error) { fields = map[string]uuid.UUID{} } + minCleanVersion := int32(0) + if c.Mincleanversion != nil { + minCleanVersion = *c.Mincleanversion + } + + minTextVersion := int32(0) + if c.Mintextversion != nil { + minTextVersion = *c.Mintextversion + } + return &Collector{ ID: database.MustToUUID(c.ID), JobID: database.MustToUUID(c.Jobid), - MinCleanVersion: c.Mincleanversion, - MinTextVersion: c.Mintextversion, + MinCleanVersion: minCleanVersion, + MinTextVersion: minTextVersion, ActiveVersion: c.Activeversion, LatestVersion: c.Latestversion, Fields: fields, diff --git a/internal/job/collector/parse_test.go b/internal/job/collector/parse_test.go index 091589b4..2d54d9d3 100644 --- a/internal/job/collector/parse_test.go +++ b/internal/job/collector/parse_test.go @@ -15,11 +15,13 @@ func TestParseDBCollector(t *testing.T) { assert.Nil(t, err) assert.Nil(t, c) + minCleanV := int32(1) + minTextV := int32(2) ogc := Collector{ ID: uuid.New(), JobID: uuid.New(), - MinCleanVersion: 1, - MinTextVersion: 2, + MinCleanVersion: minCleanV, + MinTextVersion: minTextV, Fields: map[string]uuid.UUID{ "example_key": uuid.New(), }, @@ -27,10 +29,20 @@ func TestParseDBCollector(t *testing.T) { c, err = parseDBCollector(&repository.Fullactivecollector{ ID: database.MustToDBUUID(ogc.ID), Jobid: database.MustToDBUUID(ogc.JobID), - Mincleanversion: ogc.MinCleanVersion, - Mintextversion: ogc.MinTextVersion, + Mincleanversion: &ogc.MinCleanVersion, + Mintextversion: &ogc.MinTextVersion, Fields: []byte(fmt.Sprintf("{\"example_key\":\"%s\"}", ogc.Fields["example_key"])), }) assert.Nil(t, err) assert.EqualExportedValues(t, ogc, *c) + + ogc.MinCleanVersion = 0 + ogc.MinTextVersion = 0 + c, err = parseDBCollector(&repository.Fullactivecollector{ + ID: database.MustToDBUUID(ogc.ID), + Jobid: database.MustToDBUUID(ogc.JobID), + Fields: []byte(fmt.Sprintf("{\"example_key\":\"%s\"}", ogc.Fields["example_key"])), + }) + assert.Nil(t, err) + assert.EqualExportedValues(t, ogc, *c) } diff --git a/internal/job/collector/service.go b/internal/job/collector/service.go index 57cdbcd3..f911e518 100644 --- a/internal/job/collector/service.go +++ b/internal/job/collector/service.go @@ -2,6 +2,9 @@ package collector import ( "queryorchestration/internal/database" + documentclean "queryorchestration/internal/document_clean" + "queryorchestration/internal/query" + textextraction "queryorchestration/internal/text_extraction" "github.com/google/uuid" ) @@ -16,12 +19,20 @@ type Collector struct { Fields map[string]uuid.UUID } -type Service struct { - db *database.Connection +type Services struct { + Query *query.Service + DocumentClean *documentclean.Service + TextExtraction *textextraction.Service } -func New(db *database.Connection) *Service { +type Service struct { + db *database.Connection + svc *Services +} + +func New(db *database.Connection, svc *Services) *Service { return &Service{ db, + svc, } } diff --git a/internal/job/collector/service_test.go b/internal/job/collector/service_test.go index 01920db6..dea647dd 100644 --- a/internal/job/collector/service_test.go +++ b/internal/job/collector/service_test.go @@ -21,6 +21,6 @@ func TestService(t *testing.T) { Pool: pool, } - svc := collector.New(db) + svc := collector.New(db, &collector.Services{}) assert.NotNil(t, svc) } diff --git a/internal/job/collector/update.go b/internal/job/collector/update.go new file mode 100644 index 00000000..3645627b --- /dev/null +++ b/internal/job/collector/update.go @@ -0,0 +1,275 @@ +package collector + +import ( + "context" + "errors" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/server/validation" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" +) + +type UpdateParams struct { + JobID uuid.UUID + ActiveVersion *int32 + MinCleanVersion *int32 + MinTextVersion *int32 + Fields *map[string]uuid.UUID +} + +func (s *Service) UpdateByJobId(ctx context.Context, params *UpdateParams) error { + current, err := s.GetByJobID(ctx, params.JobID) + if err != nil { + return err + } + + dbparams, err := s.getUpdateParams(ctx, current, params) + if err != nil { + return err + } + + err = s.submitUpdate(ctx, current, dbparams) + if err != nil { + return err + } + + return nil +} + +type dbUpdateParams struct { + JobID pgtype.UUID + ActiveVersion *int32 + MinCleanVersion *int32 + MinTextVersion *int32 + Fields *map[string]pgtype.UUID +} + +func (s *Service) getUpdateParams(ctx context.Context, current *Collector, params *UpdateParams) (*dbUpdateParams, error) { + err := s.normalizeCodeVersions(current, params) + if err != nil { + return nil, err + } + + var ufields *map[string]pgtype.UUID + if params.Fields != nil { + ufs, err := s.normalizeFieldsToDB(ctx, *params.Fields) + if err != nil { + return nil, err + } + ufields = &ufs + } + + err = s.normalizeActiveVersion(current, params) + if err != nil { + return nil, err + } + + if params.ActiveVersion == nil && + params.MinCleanVersion == nil && + params.MinTextVersion == nil && + ufields == nil { + return nil, errors.New("no changes") + } + + return &dbUpdateParams{ + JobID: database.MustToDBUUID(params.JobID), + ActiveVersion: params.ActiveVersion, + MinCleanVersion: params.MinCleanVersion, + MinTextVersion: params.MinTextVersion, + Fields: ufields, + }, nil +} + +func (s *Service) normalizeCodeVersions(current *Collector, params *UpdateParams) error { + if current == nil { + return errors.New("current collector required") + } + + if params == nil || params.MinCleanVersion == nil && params.MinTextVersion == nil { + return nil + } + + if (params.MinCleanVersion == nil || *params.MinCleanVersion == current.MinCleanVersion) && + (params.MinTextVersion == nil || *params.MinTextVersion == current.MinTextVersion) { + params.MinCleanVersion = nil + params.MinTextVersion = nil + return nil + } + + if params.MinCleanVersion == nil { + params.MinCleanVersion = ¤t.MinCleanVersion + } else { + err := s.svc.DocumentClean.IsValidVersion(*params.MinCleanVersion) + if err != nil { + return err + } + } + + if params.MinTextVersion == nil { + params.MinTextVersion = ¤t.MinTextVersion + } else { + err := s.svc.TextExtraction.IsValidVersion(*params.MinTextVersion) + if err != nil { + return err + } + } + + return nil +} + +func (s *Service) normalizeActiveVersion(current *Collector, params *UpdateParams) error { + if current == nil { + return errors.New("current collector required") + } + + if params == nil || params.ActiveVersion == nil { + return nil + } + + err := validation.NormalizeInClosedInterval(¶ms.ActiveVersion, current.ActiveVersion, 1, current.LatestVersion+1) + if err != nil { + return err + } + + return nil +} + +func (s *Service) normalizeFieldsToDB(ctx context.Context, ofields map[string]uuid.UUID) (map[string]pgtype.UUID, error) { + fids := []uuid.UUID{} + for _, id := range ofields { + fids = append(fids, id) + } + vfields := fields{ + values: &fids, + } + + err := s.svc.Query.NormalizeQueryIDs(ctx, &vfields) + if err != nil { + return nil, err + } + if len(ofields) != len(*vfields.values) { + return nil, errors.New("duplicate output fields") + } + + ufields := map[string]pgtype.UUID{} + for key, value := range ofields { + ufields[key] = database.MustToDBUUID(value) + } + + return ufields, nil +} + +func (s *Service) submitUpdate(ctx context.Context, current *Collector, params *dbUpdateParams) error { + err := database.ExecuteTransaction(ctx, s.db, func(ctx context.Context, qtx *repository.Queries) error { + latestVersion := current.LatestVersion + 1 + id := database.MustToDBUUID(current.ID) + + if params.MinCleanVersion != nil || params.MinTextVersion != nil { + err := qtx.RemoveCollectorCodeVersion(ctx, &repository.RemoveCollectorCodeVersionParams{ + Collectorid: id, + Removedversion: &latestVersion, + }) + if err != nil { + return err + } + + err = qtx.AddCollectorCodeVersion(ctx, &repository.AddCollectorCodeVersionParams{ + Collectorid: id, + Mincleanversion: *params.MinCleanVersion, + Mintextversion: *params.MinTextVersion, + Addedversion: latestVersion, + }) + if err != nil { + return err + } + } + + removeIDs := getRemoveFields(current.Fields, params.Fields) + for _, field := range removeIDs { + err := qtx.RemoveCollectorQuery(ctx, &repository.RemoveCollectorQueryParams{ + Collectorid: id, + Queryid: field, + Removedversion: &latestVersion, + }) + if err != nil { + return err + } + } + + addIDs := getAddFields(current.Fields, params.Fields) + for key, field := range addIDs { + err := qtx.AddCollectorQuery(ctx, &repository.AddCollectorQueryParams{ + Collectorid: id, + Name: key, + Queryid: field, + Addedversion: latestVersion, + }) + if err != nil { + return err + } + } + + activeVersion := params.ActiveVersion + if activeVersion == nil { + activeVersion = ¤t.ActiveVersion + } + + err := qtx.UpdateCollector(ctx, &repository.UpdateCollectorParams{ + ID: id, + Latestversion: latestVersion, + Activeversion: *activeVersion, + }) + if err != nil { + return err + } + + return nil + }) + if err != nil { + return err + } + + return nil +} + +func getRemoveFields(current map[string]uuid.UUID, update *map[string]pgtype.UUID) []pgtype.UUID { + diff := []pgtype.UUID{} + if update == nil { + return diff + } + + for ckey, cid := range current { + found := false + for ukey, uid := range *update { + if cid == database.MustToUUID(uid) && + ckey == ukey { + found = true + } + } + if !found { + diff = append(diff, database.MustToDBUUID(cid)) + } + } + + return diff +} + +func getAddFields(current map[string]uuid.UUID, update *map[string]pgtype.UUID) map[string]pgtype.UUID { + diff := map[string]pgtype.UUID{} + if update == nil { + return diff + } + + for ckey, cid := range current { + for ukey, uid := range *update { + if ckey == ukey && cid != database.MustToUUID(uid) || + current[ukey] == uuid.Nil { + diff[ukey] = uid + } + } + } + + return diff +} diff --git a/internal/job/collector/update_test.go b/internal/job/collector/update_test.go new file mode 100644 index 00000000..bb9fe340 --- /dev/null +++ b/internal/job/collector/update_test.go @@ -0,0 +1,62 @@ +package collector_test + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + documentclean "queryorchestration/internal/document_clean" + "queryorchestration/internal/job/collector" + "queryorchestration/internal/query" + textextraction "queryorchestration/internal/text_extraction" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +func TestUpdate(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + + svc := collector.New(db, &collector.Services{ + Query: query.New(db), + DocumentClean: documentclean.New(), + TextExtraction: textextraction.New(), + }) + + current := collector.Collector{ + ID: uuid.New(), + JobID: uuid.New(), + ActiveVersion: 1, + LatestVersion: 4, + } + av := int32(2) + update := collector.UpdateParams{ + JobID: current.JobID, + ActiveVersion: &av, + } + + pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(database.MustToDBUUID(update.JobID)). + WillReturnRows( + pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}). + AddRow(database.MustToDBUUID(current.ID), database.MustToDBUUID(current.JobID), ¤t.MinCleanVersion, ¤t.MinTextVersion, current.ActiveVersion, current.LatestVersion, []byte("")), + ) + pool.ExpectBeginTx(pgx.TxOptions{}) + pool.ExpectExec("name: UpdateCollector :exec").WithArgs(int32(5), int32(2), database.MustToDBUUID(current.ID)). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectCommit() + + err = svc.UpdateByJobId(ctx, &update) + assert.Nil(t, err) +} diff --git a/internal/job/collector/updateprivate_test.go b/internal/job/collector/updateprivate_test.go new file mode 100644 index 00000000..3e92d0a7 --- /dev/null +++ b/internal/job/collector/updateprivate_test.go @@ -0,0 +1,388 @@ +package collector + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + documentclean "queryorchestration/internal/document_clean" + "queryorchestration/internal/query" + textextraction "queryorchestration/internal/text_extraction" + "testing" + + "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" +) + +func TestGetUpdateParams(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + + svc := Service{ + db: db, + svc: &Services{ + Query: query.New(db), + }, + } + + minCleanV := int32(2) + minTextV := int32(4) + aV := int32(3) + current := Collector{ + ActiveVersion: 1, + LatestVersion: 10, + } + params := UpdateParams{ + JobID: uuid.New(), + ActiveVersion: &aV, + MinCleanVersion: &minCleanV, + MinTextVersion: &minTextV, + Fields: &map[string]uuid.UUID{ + "example_key": uuid.New(), + }, + } + + pool.ExpectQuery("name: AllQueriesExist :one").WithArgs(database.MustToDBUUIDArray([]uuid.UUID{(*params.Fields)["example_key"]})).WillReturnRows( + pgxmock.NewRows([]string{"all_exist"}). + AddRow(true), + ) + + dbparams, err := svc.getUpdateParams(ctx, ¤t, ¶ms) + assert.Nil(t, err) + assert.EqualExportedValues(t, &dbUpdateParams{ + JobID: database.MustToDBUUID(params.JobID), + ActiveVersion: &aV, + MinCleanVersion: &minCleanV, + MinTextVersion: &minTextV, + Fields: &map[string]pgtype.UUID{ + "example_key": database.MustToDBUUID((*params.Fields)["example_key"]), + }, + }, dbparams) + + (*params.Fields)["second_key"] = (*params.Fields)["example_key"] + assert.Len(t, *params.Fields, 2) + _, err = svc.getUpdateParams(ctx, ¤t, ¶ms) + assert.Error(t, err) + + params = UpdateParams{ + JobID: current.JobID, + } + _, err = svc.getUpdateParams(ctx, ¤t, ¶ms) + assert.Error(t, err) +} + +func TestSubmitUpdate(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + + svc := Service{ + db: db, + svc: &Services{ + Query: query.New(db), + }, + } + + minCleanV := int32(2) + minTextV := int32(4) + current := Collector{ + ID: uuid.New(), + JobID: uuid.New(), + ActiveVersion: 1, + LatestVersion: 1, + MinCleanVersion: 1, + MinTextVersion: 2, + Fields: map[string]uuid.UUID{ + "original_key": uuid.New(), + "og_key": uuid.New(), + }, + } + aV := int32(2) + params := dbUpdateParams{ + JobID: database.MustToDBUUID(current.JobID), + ActiveVersion: &aV, + MinCleanVersion: &minCleanV, + MinTextVersion: &minTextV, + Fields: &map[string]pgtype.UUID{ + "example_key": database.MustToDBUUID(uuid.New()), + "second_key": database.MustToDBUUID(uuid.New()), + "changed_key": database.MustToDBUUID(current.Fields["original_key"]), + "og_key": database.MustToDBUUID(current.Fields["og_key"]), + }, + } + + pool.ExpectBeginTx(pgx.TxOptions{}) + rv := int32(2) + 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(2), *params.MinCleanVersion, *params.MinTextVersion). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectExec("name: RemoveCollectorQuery :exec").WithArgs(&rv, database.MustToDBUUID(current.Fields["original_key"]), database.MustToDBUUID(current.ID)). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectExec("name: AddCollectorQuery :exec").WithArgs(database.MustToDBUUID(current.ID), "example_key", (*params.Fields)["example_key"], int32(2)). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectExec("name: AddCollectorQuery :exec").WithArgs(database.MustToDBUUID(current.ID), "second_key", (*params.Fields)["second_key"], int32(2)). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectExec("name: AddCollectorQuery :exec").WithArgs(database.MustToDBUUID(current.ID), "changed_key", (*params.Fields)["changed_key"], int32(2)). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectExec("name: UpdateCollector :exec").WithArgs(int32(2), int32(2), database.MustToDBUUID(current.ID)). + WillReturnResult(pgxmock.NewResult("", 1)) + pool.ExpectCommit() + + err = svc.submitUpdate(ctx, ¤t, ¶ms) + assert.Nil(t, err) +} + +func TestNormalizeFieldsToDB(t *testing.T) { + ctx := context.Background() + + pool, err := pgxmock.NewPool() + if err != nil { + t.Fatalf("failed to open pgxmock database: %v", err) + } + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + + svc := Service{ + db: db, + svc: &Services{ + Query: query.New(db), + }, + } + + fields := map[string]uuid.UUID{ + "example_key": uuid.New(), + } + + pool.ExpectQuery("name: AllQueriesExist :one").WithArgs(database.MustToDBUUIDArray([]uuid.UUID{fields["example_key"]})).WillReturnRows( + pgxmock.NewRows([]string{"all_exist"}). + AddRow(true), + ) + + dbparams, err := svc.normalizeFieldsToDB(ctx, fields) + assert.Nil(t, err) + assert.EqualExportedValues(t, map[string]pgtype.UUID{ + "example_key": database.MustToDBUUID(fields["example_key"]), + }, dbparams) + + fields["second_key"] = fields["example_key"] + + pool.ExpectQuery("name: AllQueriesExist :one").WithArgs(database.MustToDBUUIDArray([]uuid.UUID{fields["example_key"]})).WillReturnRows( + pgxmock.NewRows([]string{"all_exist"}). + AddRow(true), + ) + + _, err = svc.normalizeFieldsToDB(ctx, fields) + assert.Error(t, err) +} + +func TestNormalizeActiveVersion(t *testing.T) { + svc := Service{} + + err := svc.normalizeActiveVersion(nil, nil) + assert.Error(t, err) + + current := Collector{ + ActiveVersion: 1, + LatestVersion: 4, + } + + err = svc.normalizeActiveVersion(¤t, nil) + assert.Nil(t, err) + + update := UpdateParams{} + + err = svc.normalizeActiveVersion(¤t, &update) + assert.Nil(t, err) + assert.Nil(t, update.ActiveVersion) + + update.ActiveVersion = ¤t.ActiveVersion + err = svc.normalizeActiveVersion(¤t, &update) + assert.Nil(t, err) + assert.Nil(t, update.ActiveVersion) + + av := current.LatestVersion + 2 + update.ActiveVersion = &av + err = svc.normalizeActiveVersion(¤t, &update) + assert.Error(t, err) + + av = current.ActiveVersion + 1 + update.ActiveVersion = &av + err = svc.normalizeActiveVersion(¤t, &update) + assert.Nil(t, err) + assert.Equal(t, av, *update.ActiveVersion) +} + +func TestNormalizeCodeVersions(t *testing.T) { + svc := Service{ + svc: &Services{ + DocumentClean: documentclean.New(), + TextExtraction: textextraction.New(), + }, + } + + err := svc.normalizeCodeVersions(nil, nil) + assert.Error(t, err) + + current := Collector{} + + err = svc.normalizeCodeVersions(¤t, nil) + assert.Nil(t, err) + + update := UpdateParams{} + + err = svc.normalizeCodeVersions(¤t, &update) + assert.Nil(t, err) + assert.Nil(t, update.MinCleanVersion) + assert.Nil(t, update.MinTextVersion) + + cv := int32(2) + update.MinCleanVersion = &cv + update.MinTextVersion = nil + err = svc.normalizeCodeVersions(¤t, &update) + assert.Nil(t, err) + assert.Equal(t, cv, *update.MinCleanVersion) + assert.Equal(t, int32(0), *update.MinTextVersion) + + update.MinCleanVersion = nil + tv := int32(2) + update.MinTextVersion = &tv + err = svc.normalizeCodeVersions(¤t, &update) + assert.Nil(t, err) + assert.Equal(t, int32(0), *update.MinCleanVersion) + assert.Equal(t, tv, *update.MinTextVersion) + + current.MinCleanVersion = 2 + current.MinTextVersion = 2 + err = svc.normalizeCodeVersions(¤t, &update) + assert.Nil(t, err) + assert.Nil(t, update.MinCleanVersion) + assert.Nil(t, update.MinTextVersion) + + update.MinCleanVersion = ¤t.MinCleanVersion + update.MinTextVersion = nil + err = svc.normalizeCodeVersions(¤t, &update) + assert.Nil(t, err) + assert.Nil(t, update.MinCleanVersion) + assert.Nil(t, update.MinTextVersion) + + update.MinCleanVersion = nil + update.MinTextVersion = ¤t.MinTextVersion + err = svc.normalizeCodeVersions(¤t, &update) + assert.Nil(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.Nil(t, err) + assert.Nil(t, update.MinCleanVersion) + assert.Nil(t, update.MinTextVersion) + + cv = current.MinCleanVersion + 1 + update.MinCleanVersion = &cv + update.MinTextVersion = ¤t.MinTextVersion + err = svc.normalizeCodeVersions(¤t, &update) + assert.Nil(t, err) + assert.Equal(t, cv, *update.MinCleanVersion) + assert.Equal(t, current.MinTextVersion, *update.MinTextVersion) + + cv = current.MinCleanVersion + 1 + update.MinCleanVersion = &cv + tv = current.MinTextVersion + 1 + update.MinTextVersion = &tv + err = svc.normalizeCodeVersions(¤t, &update) + assert.Nil(t, err) + assert.Equal(t, cv, *update.MinCleanVersion) + assert.Equal(t, tv, *update.MinTextVersion) +} + +func TestGetRemoveFields(t *testing.T) { + current := map[string]uuid.UUID{} + var update *map[string]pgtype.UUID + + remove := getRemoveFields(current, update) + assert.Len(t, remove, 0) + + current["a"] = uuid.New() + remove = getRemoveFields(current, update) + assert.Len(t, remove, 0) + + update = &map[string]pgtype.UUID{ + "b": database.MustToDBUUID(uuid.New()), + } + + (*update)["a"] = database.MustToDBUUID(current["a"]) + remove = getRemoveFields(current, update) + assert.Len(t, remove, 0) + + (*update)["a"] = database.MustToDBUUID(uuid.New()) + remove = getRemoveFields(current, update) + assert.Len(t, remove, 1) + assert.ElementsMatch(t, []pgtype.UUID{database.MustToDBUUID(current["a"])}, remove) + + (*update)["a"] = database.MustToDBUUID(uuid.Nil) + (*update)["c"] = database.MustToDBUUID(current["a"]) + remove = getRemoveFields(current, update) + assert.Len(t, remove, 1) + assert.ElementsMatch(t, []pgtype.UUID{database.MustToDBUUID(current["a"])}, remove) +} + +func TestGetAddFields(t *testing.T) { + current := map[string]uuid.UUID{} + var update *map[string]pgtype.UUID + + add := getAddFields(current, update) + assert.Len(t, add, 0) + + current["a"] = uuid.New() + add = getAddFields(current, update) + assert.Len(t, add, 0) + + update = &map[string]pgtype.UUID{} + + add = getAddFields(current, update) + assert.Len(t, add, 0) + + (*update)["a"] = database.MustToDBUUID(current["a"]) + add = getAddFields(current, update) + assert.Len(t, add, 0) + + (*update)["a"] = database.MustToDBUUID(uuid.New()) + add = getAddFields(current, update) + assert.Len(t, add, 1) + assert.Equal(t, map[string]pgtype.UUID{ + "a": (*update)["a"], + }, add) + + (*update)["a"] = database.MustToDBUUID(current["a"]) + (*update)["b"] = database.MustToDBUUID(uuid.New()) + add = getAddFields(current, update) + assert.Len(t, add, 1) + assert.Equal(t, map[string]pgtype.UUID{ + "b": (*update)["b"], + }, add) +} diff --git a/internal/query/create.go b/internal/query/create.go index a353bb37..3fe61551 100644 --- a/internal/query/create.go +++ b/internal/query/create.go @@ -28,7 +28,7 @@ func (s *Service) Create(ctx context.Context, entity *queryprocessor.Create) (uu } func (s *Service) normalizeCreate(ctx context.Context, entity *queryprocessor.Create) error { - err := s.normalizeQueryIDs(ctx, entity) + err := s.NormalizeQueryIDs(ctx, entity) if err != nil { return err } @@ -57,51 +57,47 @@ func (s *Service) submitCreate(ctx context.Context, entity *queryprocessor.Creat return uuid.Nil, err } - tx, err := s.db.Pool.Begin(ctx) - if err != nil { - return uuid.Nil, err - } - defer func() { - _ = tx.Rollback(ctx) - }() + var dbID pgtype.UUID + err = database.ExecuteTransaction(ctx, s.db, func(ctx context.Context, qtx *repository.Queries) error { + dbID, err = qtx.CreateQuery(ctx, query.Type) + if err != nil { + return err + } - qtx := s.db.Queries.WithTx(tx) - - dbID, err := qtx.CreateQuery(ctx, query.Type) - if err != nil { - return uuid.Nil, err - } - - if query.RequiredQueryIDs != nil { - for _, reqQuery := range *query.RequiredQueryIDs { - err = qtx.AddRequiredQuery(ctx, &repository.AddRequiredQueryParams{ - Queryid: dbID, - Requiredqueryid: reqQuery, - Addedversion: 1, - }) - if err != nil { - return uuid.Nil, err + if query.RequiredQueryIDs != nil { + for _, reqQuery := range *query.RequiredQueryIDs { + err = qtx.AddRequiredQuery(ctx, &repository.AddRequiredQueryParams{ + Queryid: dbID, + Requiredqueryid: reqQuery, + Addedversion: 1, + }) + if err != nil { + return err + } } } - } - if query.Config != nil && string(*query.Config) != "" { - err = qtx.AddQueryConfig(ctx, &repository.AddQueryConfigParams{ - Queryid: dbID, - Config: *query.Config, - Addedversion: 1, - }) - if err != nil { - return uuid.Nil, err + if query.Config != nil && string(*query.Config) != "" { + err = qtx.AddQueryConfig(ctx, &repository.AddQueryConfigParams{ + Queryid: dbID, + Config: *query.Config, + Addedversion: 1, + }) + if err != nil { + return err + } } - } - err = tx.Commit(ctx) + return nil + }) if err != nil { return uuid.Nil, err } - id := database.MustToUUID(dbID) + id, err := database.ToUUID(dbID) + if err != nil { + return uuid.Nil, err + } return id, nil diff --git a/internal/query/document/sync_test.go b/internal/query/document/sync_test.go index c5bd1f75..2f200945 100644 --- a/internal/query/document/sync_test.go +++ b/internal/query/document/sync_test.go @@ -45,7 +45,7 @@ func TestSyncIsSynced(t *testing.T) { pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(database.MustToDBUUID(doc.JobID)). WillReturnRows( pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}). - AddRow(database.MustToDBUUID(coll.ID), database.MustToDBUUID(coll.JobID), coll.MinCleanVersion, coll.MinTextVersion, int32(1), int32(2), []byte("")), + AddRow(database.MustToDBUUID(coll.ID), database.MustToDBUUID(coll.JobID), &coll.MinCleanVersion, &coll.MinTextVersion, int32(1), int32(2), []byte("")), ) pool.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), coll.MinCleanVersion, coll.MinTextVersion). WillReturnRows( @@ -59,7 +59,7 @@ func TestSyncIsSynced(t *testing.T) { ) docSvc := document.New(db, &document.Services{ - Collector: collector.New(db), + Collector: collector.New(db, &collector.Services{}), }) err = docSvc.Sync(ctx, &doc) assert.Nil(t, err) @@ -96,7 +96,7 @@ func TestSyncDBFail(t *testing.T) { ) docSvc := document.New(db, &document.Services{ - Collector: collector.New(db), + Collector: collector.New(db, &collector.Services{}), }) err = docSvc.Sync(ctx, &doc) assert.EqualError(t, err, "no rows in result set") @@ -104,14 +104,14 @@ func TestSyncDBFail(t *testing.T) { pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(dbJobID). WillReturnRows( pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}). - AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion, int32(1), int32(2), []byte("")), + AddRow(dbCollectorId, dbJobID, &minCleanVersion, &minTextVersion, int32(1), int32(2), []byte("")), ) dbErr := "database failure" pool.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), minCleanVersion, minTextVersion). WillReturnError(errors.New(dbErr)) docSvc = document.New(db, &document.Services{ - Collector: collector.New(db), + Collector: collector.New(db, &collector.Services{}), }) err = docSvc.Sync(ctx, &doc) assert.EqualError(t, err, dbErr) @@ -119,7 +119,7 @@ func TestSyncDBFail(t *testing.T) { pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(dbJobID). WillReturnRows( pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}). - AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion, int32(1), int32(2), []byte("")), + AddRow(dbCollectorId, dbJobID, &minCleanVersion, &minTextVersion, int32(1), int32(2), []byte("")), ) pool.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), minCleanVersion, minTextVersion). WillReturnRows( @@ -130,7 +130,7 @@ func TestSyncDBFail(t *testing.T) { WillReturnError(errors.New(dbErr)) docSvc = document.New(db, &document.Services{ - Collector: collector.New(db), + Collector: collector.New(db, &collector.Services{}), }) err = docSvc.Sync(ctx, &doc) assert.EqualError(t, err, dbErr) @@ -138,7 +138,7 @@ func TestSyncDBFail(t *testing.T) { pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(dbJobID). WillReturnRows( pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}). - AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion, int32(1), int32(2), []byte("")), + AddRow(dbCollectorId, dbJobID, &minCleanVersion, &minTextVersion, int32(1), int32(2), []byte("")), ) qV := int32(1) reqID := database.MustToDBUUID(uuid.New()) @@ -159,7 +159,7 @@ func TestSyncDBFail(t *testing.T) { WillReturnError(errors.New(dbErr)) docSvc = document.New(db, &document.Services{ - Collector: collector.New(db), + Collector: collector.New(db, &collector.Services{}), }) err = docSvc.Sync(ctx, &doc) assert.EqualError(t, err, dbErr) @@ -193,7 +193,7 @@ func TestSync(t *testing.T) { pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(dbJobID). WillReturnRows( pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}). - AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion, int32(1), int32(2), []byte("")), + AddRow(dbCollectorId, dbJobID, &minCleanVersion, &minTextVersion, int32(1), int32(2), []byte("")), ) pool.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), minCleanVersion, minTextVersion). WillReturnRows( @@ -211,7 +211,7 @@ func TestSync(t *testing.T) { ) docSvc := document.New(db, &document.Services{ - Collector: collector.New(db), + Collector: collector.New(db, &collector.Services{}), }) err = docSvc.Sync(ctx, &doc) assert.EqualError(t, err, "JSON Extraction requires 1 result") diff --git a/internal/query/normalize.go b/internal/query/normalize.go index 29cbf42c..837c72b5 100644 --- a/internal/query/normalize.go +++ b/internal/query/normalize.go @@ -7,6 +7,7 @@ import ( "fmt" "queryorchestration/internal/database" queryprocessor "queryorchestration/internal/query/processor" + "queryorchestration/internal/server/validation" "strings" "github.com/google/uuid" @@ -55,7 +56,7 @@ type RequiredQueryIDs interface { SetRequiredQueryIDs(*[]uuid.UUID) } -func (s *Service) normalizeQueryIDs(ctx context.Context, ids RequiredQueryIDs) error { +func (s *Service) NormalizeQueryIDs(ctx context.Context, ids RequiredQueryIDs) error { if ids == nil || ids.GetRequiredQueryIDs() == nil { return nil } @@ -67,7 +68,10 @@ func (s *Service) normalizeQueryIDs(ctx context.Context, ids RequiredQueryIDs) e return nil } - dbids := database.MustToDBUUIDArray(*ide) + dedup := validation.DeduplicateArray(*ide) + ids.SetRequiredQueryIDs(&dedup) + + dbids := database.MustToDBUUIDArray(dedup) exist, err := s.db.Queries.AllQueriesExist(ctx, dbids) if err != nil { @@ -84,17 +88,13 @@ func (s *Service) normalizeActiveVersion(current *Query, entity *queryprocessor. return errors.New("current query required") } - if entity == nil || entity.ActiveVersion == nil { + if entity == nil { return nil } - if entity.ActiveVersion == ¤t.ActiveVersion { - entity.ActiveVersion = nil - return nil - } - - if *entity.ActiveVersion < 1 || *entity.ActiveVersion > current.LatestVersion+1 { - return fmt.Errorf("active version must be in the range: 1 <= activeVersion <= %d", current.LatestVersion+1) + err := validation.NormalizeInClosedInterval(&entity.ActiveVersion, current.ActiveVersion, 1, current.LatestVersion+1) + if err != nil { + return err } return nil diff --git a/internal/query/normalize_test.go b/internal/query/normalize_test.go index 259c1752..d4f0d168 100644 --- a/internal/query/normalize_test.go +++ b/internal/query/normalize_test.go @@ -81,18 +81,18 @@ func TestNormalizeQueryIDs(t *testing.T) { } s := Service{db: db} - err = s.normalizeQueryIDs(ctx, nil) + err = s.NormalizeQueryIDs(ctx, nil) assert.Nil(t, err) entity := queryprocessor.Create{} entity.RequiredQueryIDs = nil - err = s.normalizeQueryIDs(ctx, &entity) + err = s.NormalizeQueryIDs(ctx, &entity) assert.Nil(t, err) assert.Nil(t, entity.RequiredQueryIDs) entity.RequiredQueryIDs = &[]uuid.UUID{} - err = s.normalizeQueryIDs(ctx, &entity) + err = s.NormalizeQueryIDs(ctx, &entity) assert.Nil(t, err) assert.Nil(t, entity.RequiredQueryIDs) @@ -105,7 +105,7 @@ func TestNormalizeQueryIDs(t *testing.T) { AddRow(true), ) - err = s.normalizeQueryIDs(ctx, &entity) + err = s.NormalizeQueryIDs(ctx, &entity) assert.Nil(t, err) assert.ElementsMatch(t, ids, *entity.RequiredQueryIDs) @@ -114,16 +114,30 @@ func TestNormalizeQueryIDs(t *testing.T) { AddRow(false), ) - err = s.normalizeQueryIDs(ctx, &entity) + err = s.NormalizeQueryIDs(ctx, &entity) assert.Error(t, err) assert.ElementsMatch(t, ids, *entity.RequiredQueryIDs) pool.ExpectQuery("name: AllQueriesExist :one").WithArgs(dbids). WillReturnError(errors.New("database failure")) - err = s.normalizeQueryIDs(ctx, &entity) + err = s.NormalizeQueryIDs(ctx, &entity) assert.Error(t, err) assert.ElementsMatch(t, ids, *entity.RequiredQueryIDs) + + singleid := uuid.New() + entity.RequiredQueryIDs = &[]uuid.UUID{singleid, singleid} + outids := []uuid.UUID{singleid} + dbids = database.MustToDBUUIDArray(outids) + + pool.ExpectQuery("name: AllQueriesExist :one").WithArgs(dbids).WillReturnRows( + pgxmock.NewRows([]string{"all_exist"}). + AddRow(true), + ) + + err = s.NormalizeQueryIDs(ctx, &entity) + assert.Nil(t, err) + assert.ElementsMatch(t, outids, *entity.RequiredQueryIDs) } func TestNormalizeActiveVersion(t *testing.T) { diff --git a/internal/query/update.go b/internal/query/update.go index 404b0f98..bf0801c4 100644 --- a/internal/query/update.go +++ b/internal/query/update.go @@ -2,6 +2,7 @@ package query import ( "context" + "errors" "fmt" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" @@ -37,16 +38,34 @@ func (s *Service) normalizeUpdate(ctx context.Context, current *Query, entity *q return err } - err = s.normalizeQueryIDs(ctx, entity) + err = s.NormalizeQueryIDs(ctx, entity) if err != nil { return err } + if entity.RequiredQueryIDs != nil { + createsloop, err := s.db.Queries.IsQueryInDependencyTree(ctx, &repository.IsQueryInDependencyTreeParams{ + Requiredqueryid: database.MustToDBUUID(current.ID), + ID: database.MustToDBUUIDArray(*entity.RequiredQueryIDs), + }) + if err != nil { + return err + } else if createsloop { + return errors.New("required ids create a loop") + } + } + err = s.normalizeConfig(entity) if err != nil { return err } + if entity.ActiveVersion == nil && + entity.RequiredQueryIDs == nil && + entity.Config == nil { + return errors.New("no changes") + } + validator, err := s.getUpdator(current.Type) if err != nil { return err @@ -74,8 +93,6 @@ func (s *Service) submitUpdate(ctx context.Context, current *Query, entity *quer latestVersion := current.LatestVersion + 1 id := database.MustToDBUUID(entity.ID) - hasChanges := false - addIDs := getSetDifference(entity.RequiredQueryIDs, current.RequiredQueryIDs) for _, qID := range addIDs { err = qtx.AddRequiredQuery(ctx, &repository.AddRequiredQueryParams{ @@ -86,7 +103,6 @@ func (s *Service) submitUpdate(ctx context.Context, current *Query, entity *quer if err != nil { return err } - hasChanges = true } removeIDs := getSetDifference(current.RequiredQueryIDs, entity.RequiredQueryIDs) @@ -99,8 +115,6 @@ func (s *Service) submitUpdate(ctx context.Context, current *Query, entity *quer if err != nil { return err } - - hasChanges = true } if entity.Config != nil && *entity.Config != "" { @@ -120,33 +134,25 @@ func (s *Service) submitUpdate(ctx context.Context, current *Query, entity *quer if err != nil { return err } - - hasChanges = true } - activeVersion := current.ActiveVersion - if entity.ActiveVersion != nil { - activeVersion = *entity.ActiveVersion + activeVersion := entity.ActiveVersion + if activeVersion == nil { + activeVersion = ¤t.ActiveVersion } - if activeVersion != current.ActiveVersion { - hasChanges = true + err = qtx.UpdateQuery(ctx, &repository.UpdateQueryParams{ + Latestversion: latestVersion, + Activeversion: *activeVersion, + ID: id, + }) + if err != nil { + return err } - if hasChanges { - err = qtx.UpdateQuery(ctx, &repository.UpdateQueryParams{ - Latestversion: latestVersion, - Activeversion: activeVersion, - ID: id, - }) - if err != nil { - return err - } - - err = tx.Commit(ctx) - if err != nil { - return err - } + err = tx.Commit(ctx) + if err != nil { + return err } return nil diff --git a/internal/query/update_test.go b/internal/query/update_test.go index 0a598a08..c5c02833 100644 --- a/internal/query/update_test.go +++ b/internal/query/update_test.go @@ -9,7 +9,6 @@ import ( "testing" "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" @@ -44,11 +43,6 @@ func TestUpdate(t *testing.T) { AddRow(database.MustToDBUUID(existing.ID), repository.QuerytypeJsonExtractor, existing.ActiveVersion, existing.LatestVersion, []byte(config), []pgtype.UUID{}), ) - pool.ExpectBeginTx(pgx.TxOptions{}) - pool.ExpectExec("name: UpdateQuery :exec").WithArgs(int32(1), int32(2), database.MustToDBUUID(update.ID)). - WillReturnResult(pgxmock.NewResult("", 1)) - pool.ExpectCommit() - err = svc.Update(ctx, update) - assert.Nil(t, err) + assert.Error(t, err) } diff --git a/internal/query/updateprivate_test.go b/internal/query/updateprivate_test.go index 397e3b12..42eeb6a2 100644 --- a/internal/query/updateprivate_test.go +++ b/internal/query/updateprivate_test.go @@ -213,36 +213,6 @@ func TestSubmitUpdateActiveVersion(t *testing.T) { assert.Nil(t, err) } -func TestSubmitUpdateNoChange(t *testing.T) { - ctx := context.Background() - - pool, err := pgxmock.NewPool() - if err != nil { - t.Fatalf("failed to open pgxmock database: %v", err) - } - queries := repository.New(pool) - db := &database.Connection{ - Queries: queries, - Pool: pool, - } - svc := New(db) - - q := Query{ - ID: uuid.New(), - ActiveVersion: int32(1), - LatestVersion: int32(2), - } - update := &queryprocessor.Update{ - ID: q.ID, - } - - pool.ExpectBeginTx(pgx.TxOptions{}) - pool.ExpectRollback() - - err = svc.submitUpdate(ctx, &q, update) - assert.Nil(t, err) -} - func TestGetSetDifference(t *testing.T) { commonUUID := uuid.New() listA := []uuid.UUID{uuid.New(), commonUUID} @@ -300,4 +270,22 @@ func TestNormalizeUpdate(t *testing.T) { Config: &cfg, RequiredQueryIDs: nil, }, *update) + + update.RequiredQueryIDs = &[]uuid.UUID{uuid.New()} + + dbids = database.MustToDBUUIDArray(*update.RequiredQueryIDs) + + pool.ExpectQuery("name: IsQueryInDependencyTree :one").WithArgs(dbids).WillReturnRows( + pgxmock.NewRows([]string{"all_exist"}). + AddRow(true), + ) + + err = svc.normalizeUpdate(ctx, current, update) + assert.Error(t, err) + + update = &queryprocessor.Update{ + ID: current.ID, + } + err = svc.normalizeUpdate(ctx, current, update) + assert.Error(t, err) } diff --git a/internal/server/validation/validation.go b/internal/server/validation/validation.go new file mode 100644 index 00000000..8e2dffdc --- /dev/null +++ b/internal/server/validation/validation.go @@ -0,0 +1,41 @@ +package validation + +import "fmt" + +func GetUpdatedValue[T any](def T, new *T) T { + if new != nil { + return *new + } + return def +} + +func DeduplicateArray[T comparable](arr []T) []T { + occurred := make(map[T]bool) + result := []T{} + + for _, value := range arr { + if !occurred[value] { + occurred[value] = true + result = append(result, value) + } + } + + return result +} + +func NormalizeInClosedInterval(updated **int32, current int32, min int32, max int32) error { + if updated == nil || *updated == nil { + return nil + } + + if **updated == current { + *updated = nil + return nil + } + + if **updated < min || **updated > max { + return fmt.Errorf("update must be in range: %d <= update <= %d", min, max) + } + + return nil +} diff --git a/internal/server/validation/validation_test.go b/internal/server/validation/validation_test.go new file mode 100644 index 00000000..f36799a7 --- /dev/null +++ b/internal/server/validation/validation_test.go @@ -0,0 +1,60 @@ +package validation_test + +import ( + "queryorchestration/internal/server/validation" + "testing" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" +) + +func TestGetUpdatedValue(t *testing.T) { + assert.Equal(t, 1, validation.GetUpdatedValue(1, nil)) + val := 3 + assert.Equal(t, 3, validation.GetUpdatedValue(1, &val)) + + assert.Equal(t, "hello", validation.GetUpdatedValue("hello", nil)) + valstr := "byebye" + assert.Equal(t, "byebye", validation.GetUpdatedValue("hello", &valstr)) +} + +func TestDeduplicateArray(t *testing.T) { + assert.ElementsMatch(t, []int{}, validation.DeduplicateArray([]int{})) + assert.ElementsMatch(t, []int{1, 2, 3}, validation.DeduplicateArray([]int{1, 2, 3})) + assert.ElementsMatch(t, []int{1, 2, 3}, validation.DeduplicateArray([]int{1, 2, 1, 3})) + assert.ElementsMatch(t, []int{1}, validation.DeduplicateArray([]int{1, 1, 1, 1})) + id := uuid.New() + assert.ElementsMatch(t, []uuid.UUID{id}, validation.DeduplicateArray([]uuid.UUID{id, id, id})) +} + +func TestNormalizeInClosedInterval(t *testing.T) { + err := validation.NormalizeInClosedInterval(nil, 1, 1, 1) + assert.Nil(t, err) + + var param *int32 + err = validation.NormalizeInClosedInterval(¶m, 1, 1, 1) + assert.Nil(t, err) + assert.Nil(t, param) + + updated := int32(1) + param = &updated + err = validation.NormalizeInClosedInterval(¶m, 1, 1, 1) + assert.Nil(t, err) + assert.Equal(t, int32(1), updated) + + updated = -1 + param = &updated + err = validation.NormalizeInClosedInterval(¶m, 1, 1, 1) + assert.Error(t, err) + + updated = 2 + param = &updated + err = validation.NormalizeInClosedInterval(¶m, 1, 1, 1) + assert.Error(t, err) + + updated = 2 + param = &updated + err = validation.NormalizeInClosedInterval(¶m, 1, 1, 3) + assert.Nil(t, err) + assert.Equal(t, int32(2), updated) +} diff --git a/internal/text_extraction/service.go b/internal/text_extraction/service.go new file mode 100644 index 00000000..35dc5794 --- /dev/null +++ b/internal/text_extraction/service.go @@ -0,0 +1,18 @@ +package textextraction + +import "errors" + +type Service struct { +} + +func New() *Service { + return &Service{} +} + +func (s *Service) IsValidVersion(v int32) error { + if v <= 0 { + return errors.New("document clean code version must be > 0") + } + + return nil +} diff --git a/internal/text_extraction/service_test.go b/internal/text_extraction/service_test.go new file mode 100644 index 00000000..1fa82b24 --- /dev/null +++ b/internal/text_extraction/service_test.go @@ -0,0 +1,13 @@ +package textextraction_test + +import ( + documentclean "queryorchestration/internal/document_clean" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestService(t *testing.T) { + svc := documentclean.New() + assert.NotNil(t, svc) +} diff --git a/scripts/tests.yml b/scripts/tests.yml index ce5f34c2..a329ff7c 100644 --- a/scripts/tests.yml +++ b/scripts/tests.yml @@ -51,10 +51,6 @@ tasks: integration: cmds: - task generate - - task docker:build - - task test:integration:nobuild - integration:nocache: - cmds: - task docker:build - go test -count=1 -v ./test/... integration:nobuild: