From 788b21594c03373b27d21f3871eea7f075f061cb Mon Sep 17 00:00:00 2001 From: Michael McGuinness Date: Fri, 3 Jan 2025 13:41:07 +0000 Subject: [PATCH] creatorupdatordeprecate --- Taskfile.yml | 11 ---- api/controllers/parse.go | 15 ++--- api/controllers/query.go | 14 +++-- api/serviceInterfaces | 2 +- ...103132852_query_deprecation_table.down.sql | 1 + ...50103132852_query_deprecation_table.up.sql | 8 +++ database/queries/collector.sql | 5 +- database/queries/query.sql | 11 ++++ internal/contextFull/creator.go | 32 ++++++++++ internal/contextFull/service.go | 9 +-- internal/contextFull/updator.go | 30 ++++++++++ internal/database/repository/collector.sql.go | 17 +++--- internal/database/repository/models.go | 7 +++ internal/database/repository/query.sql.go | 33 +++++++++++ internal/jsonExtractor/creator.go | 32 ++++++++++ internal/jsonExtractor/service.go | 8 +-- internal/jsonExtractor/updator.go | 30 ++++++++++ internal/query/create.go | 36 ++++++++++++ internal/query/deprecate.go | 26 +++++++++ internal/query/list.go | 10 ++++ internal/query/service.go | 58 ++----------------- internal/query/test.go | 13 +++++ internal/query/update.go | 45 ++++++++++++++ .../database.go => queryProcessor/parse.go} | 37 +++++++----- internal/queryProcessor/service.go | 49 ++++++++++++++++ internal/queryQueue/create.go | 30 ++++++---- internal/queryQueue/execute.go | 22 +++---- internal/queryQueue/result.go | 14 ++--- internal/queryQueue/service.go | 8 +-- scripts/proto.yml | 2 - .../unit/internal/contextfull/process_test.go | 14 ++--- test/unit/internal/document/sync_test.go | 12 ++-- .../internal/jsonextractor/process_test.go | 38 ++++++------ test/unit/internal/query/get_test.go | 3 +- test/unit/internal/query/parse_test.go | 48 +++++++-------- test/unit/internal/queryQueue/queue_test.go | 41 +++++++------ 36 files changed, 545 insertions(+), 226 deletions(-) delete mode 100644 Taskfile.yml create mode 100644 database/migrations/20250103132852_query_deprecation_table.down.sql create mode 100644 database/migrations/20250103132852_query_deprecation_table.up.sql create mode 100644 internal/contextFull/creator.go create mode 100644 internal/contextFull/updator.go create mode 100644 internal/jsonExtractor/creator.go create mode 100644 internal/jsonExtractor/updator.go create mode 100644 internal/query/create.go create mode 100644 internal/query/deprecate.go create mode 100644 internal/query/list.go create mode 100644 internal/query/test.go create mode 100644 internal/query/update.go rename internal/{query/database.go => queryProcessor/parse.go} (69%) create mode 100644 internal/queryProcessor/service.go diff --git a/Taskfile.yml b/Taskfile.yml deleted file mode 100644 index 8e74433f..00000000 --- a/Taskfile.yml +++ /dev/null @@ -1,11 +0,0 @@ -# https://taskfile.dev - -version: '3' - -includes: - lib: - taskfile: scripts/Taskfile.yml - flatten: true - vars: - IMAGE_NAME: queryorchestration - COVERAGE_THRESHOLD: 80 diff --git a/api/controllers/parse.go b/api/controllers/parse.go index 45f16d57..35c020a9 100644 --- a/api/controllers/parse.go +++ b/api/controllers/parse.go @@ -3,6 +3,7 @@ package controllers import ( serviceinterfaces "queryorchestration/api/serviceInterfaces" "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" ) func ParseQuery(query *query.Query) *serviceinterfaces.Query { @@ -19,24 +20,24 @@ func ParseQuery(query *query.Query) *serviceinterfaces.Query { } } -func ParseQueryType(qType query.Type) serviceinterfaces.QueryType { +func ParseQueryType(qType queryprocessor.Type) serviceinterfaces.QueryType { switch qType { - case query.TypeJsonExtractor: + case queryprocessor.TypeJsonExtractor: return serviceinterfaces.QueryType_QUERY_TYPE_JSON_EXTRACTOR - case query.TypeContextFull: + case queryprocessor.TypeContextFull: return serviceinterfaces.QueryType_QUERY_TYPE_CONTEXT_FULL default: return serviceinterfaces.QueryType_QUERY_TYPE_UNSPECIFIED } } -func ParseSpecQueryType(qType serviceinterfaces.QueryType) query.Type { +func ParseSpecQueryType(qType serviceinterfaces.QueryType) queryprocessor.Type { switch qType { case serviceinterfaces.QueryType_QUERY_TYPE_JSON_EXTRACTOR: - return query.TypeJsonExtractor + return queryprocessor.TypeJsonExtractor case serviceinterfaces.QueryType_QUERY_TYPE_CONTEXT_FULL: - return query.TypeContextFull + return queryprocessor.TypeContextFull default: - return query.TypeContextFull + return queryprocessor.TypeContextFull } } diff --git a/api/controllers/query.go b/api/controllers/query.go index a0e3db03..e56328b1 100644 --- a/api/controllers/query.go +++ b/api/controllers/query.go @@ -2,8 +2,10 @@ package controllers import ( "context" + "queryorchestration/api/grpc/spec" serviceinterfaces "queryorchestration/api/serviceInterfaces" "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" "github.com/go-playground/validator/v10" "github.com/google/uuid" @@ -25,9 +27,9 @@ func NewQueryController(querySvc query.Service, validator *validator.Validate) * } func (s *QueryController) List(ctx context.Context, req *serviceinterfaces.QueryFilter) (*serviceinterfaces.Queries, error) { - types := make([]query.Type, len(req.GetTypes())) + types := make([]queryprocessor.Type, len(req.GetTypes())) for index, t := range req.GetTypes() { - types[index] = query.Type(ParseSpecQueryType(t)) + types[index] = queryprocessor.Type(ParseSpecQueryType(t)) } filters := query.ListFilters{ @@ -74,7 +76,7 @@ func (s *QueryController) Create(ctx context.Context, req *serviceinterfaces.Que requiredQueryIDs[index] = parsedID } - id, err := s.query.Create(ctx, query.Create{ + id, err := s.query.Create(ctx, &queryprocessor.Create{ Type: ParseSpecQueryType(req.GetType()), RequiredQueryIDs: requiredQueryIDs, }) @@ -93,7 +95,7 @@ func (s *QueryController) Update(ctx context.Context, req *serviceinterfaces.Que return nil, err } - err = s.query.Update(ctx, query.Update{ + err = s.query.Update(ctx, &queryprocessor.Update{ ID: id, }) if err != nil { @@ -103,13 +105,13 @@ func (s *QueryController) Update(ctx context.Context, req *serviceinterfaces.Que return &emptypb.Empty{}, nil } -func (s *QueryController) Remove(ctx context.Context, req *serviceinterfaces.IdMessage) (*emptypb.Empty, error) { +func (s *QueryController) Deprecate(ctx context.Context, req *spec.IdMessage) (*emptypb.Empty, error) { id, err := uuid.Parse(req.GetId()) if err != nil { return nil, err } - err = s.query.Remove(ctx, id) + err = s.query.Deprecate(ctx, id) if err != nil { return nil, err } diff --git a/api/serviceInterfaces b/api/serviceInterfaces index c41bce65..e3d7e63a 160000 --- a/api/serviceInterfaces +++ b/api/serviceInterfaces @@ -1 +1 @@ -Subproject commit c41bce6522850bb8d605a5ee33d929708377d10b +Subproject commit e3d7e63a2745b19bf2f4a29506a5f89d2aadfd13 diff --git a/database/migrations/20250103132852_query_deprecation_table.down.sql b/database/migrations/20250103132852_query_deprecation_table.down.sql new file mode 100644 index 00000000..c0590869 --- /dev/null +++ b/database/migrations/20250103132852_query_deprecation_table.down.sql @@ -0,0 +1 @@ +DROP TABLE queryDeprecations; \ No newline at end of file diff --git a/database/migrations/20250103132852_query_deprecation_table.up.sql b/database/migrations/20250103132852_query_deprecation_table.up.sql new file mode 100644 index 00000000..efe4d7d4 --- /dev/null +++ b/database/migrations/20250103132852_query_deprecation_table.up.sql @@ -0,0 +1,8 @@ +CREATE TABLE queryDeprecations ( + id uuid primary key DEFAULT gen_random_uuid(), + queryId uuid not null, + time timestamp not null DEFAULT NOW(), + removedAt timestamp, + foreign key (queryId) references queries(id), + unique (queryId, removedAt) +); \ No newline at end of file diff --git a/database/queries/collector.sql b/database/queries/collector.sql index 6da5e6f6..31320091 100644 --- a/database/queries/collector.sql +++ b/database/queries/collector.sql @@ -1,5 +1,8 @@ -- name: GetCollectorQueries :many -SELECT collectorId, queryId, type, requiredQueryId, queryVersion FROM collectorQueryDependencyTree WHERE collectorId = $1; +SELECT collectorId, queryId, type, queryVersion, ARRAY_AGG(requiredQueryId) AS requiredIds + FROM collectorQueryDependencyTree + WHERE collectorId = $1 + GROUP BY queryId, collectorId, type, queryVersion; -- name: GetCollectorFromJobID :one SELECT id, jobId, minCleanVersion, minTextVersion FROM collectors WHERE jobId = $1 LIMIT 1; \ No newline at end of file diff --git a/database/queries/query.sql b/database/queries/query.sql index 541f958d..15bb71fc 100644 --- a/database/queries/query.sql +++ b/database/queries/query.sql @@ -1,6 +1,17 @@ -- name: GetQueryConfig :one SELECT id, config FROM queryConfigs where queryId = $1 and addedVersion >= $2 and COALESCE(removedVersion, $2 - 1) < $2; +-- name: GetQueryType :one +SELECT type FROM queries where id = $1; + +-- name: DeprecateQuery :exec +INSERT INTO queryDeprecations (queryId) VALUES ($1) RETURNING id; + +-- name: IsQueryDeprecated :one +SELECT EXISTS ( + SELECT 1 FROM queryDeprecations where queryId = $1 and removedAt is not null +); + -- name: GetQuery :one SELECT q.id, q.type, q.activeVersion, c.config, ARRAY_AGG(r.requiredQueryId) AS requiredIds FROM queries AS q diff --git a/internal/contextFull/creator.go b/internal/contextFull/creator.go new file mode 100644 index 00000000..3fcc90e6 --- /dev/null +++ b/internal/contextFull/creator.go @@ -0,0 +1,32 @@ +package contextfull + +import ( + "context" + "queryorchestration/internal/database/repository" + queryprocessor "queryorchestration/internal/queryProcessor" + + "github.com/google/uuid" +) + +type Creator struct { + db *repository.Queries +} + +func NewCreator(db *repository.Queries) Creator { + return Creator{db} +} + +func (s Creator) Validate(ctx context.Context, entity *queryprocessor.Create) error { + // TODO + return nil +} + +func (s Creator) Create(ctx context.Context, entity *queryprocessor.Create) (uuid.UUID, error) { + err := s.Validate(ctx, entity) + if err != nil { + return uuid.Nil, err + } + + // TODO + return uuid.Nil, nil +} diff --git a/internal/contextFull/service.go b/internal/contextFull/service.go index 59e7c159..67957832 100644 --- a/internal/contextFull/service.go +++ b/internal/contextFull/service.go @@ -3,21 +3,22 @@ package contextfull import ( "context" "fmt" - "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" "queryorchestration/internal/result" ) type Extractor struct { } -func New() *Extractor { - return &Extractor{} +func NewExtractor() Extractor { + return Extractor{} } -func (e *Extractor) Process(ctx context.Context, query query.QueryRow, values *[]result.Value) (string, error) { +func (e Extractor) Process(ctx context.Context, query queryprocessor.Query, values *[]result.Value) (string, error) { if len(*values) > 0 { return "", fmt.Errorf("no requirements expected") } + // TODO return "", nil } diff --git a/internal/contextFull/updator.go b/internal/contextFull/updator.go new file mode 100644 index 00000000..5dfde987 --- /dev/null +++ b/internal/contextFull/updator.go @@ -0,0 +1,30 @@ +package contextfull + +import ( + "context" + "queryorchestration/internal/database/repository" + queryprocessor "queryorchestration/internal/queryProcessor" +) + +type Updator struct { + db *repository.Queries +} + +func NewUpdator(db *repository.Queries) Updator { + return Updator{db} +} + +func (s Updator) Validate(ctx context.Context, entity *queryprocessor.Update) error { + // TODO + return nil +} + +func (s Updator) Update(ctx context.Context, entity *queryprocessor.Update) error { + err := s.Validate(ctx, entity) + if err != nil { + return err + } + + // TODO + return nil +} diff --git a/internal/database/repository/collector.sql.go b/internal/database/repository/collector.sql.go index ea5eb8ae..f85da2ac 100644 --- a/internal/database/repository/collector.sql.go +++ b/internal/database/repository/collector.sql.go @@ -35,15 +35,18 @@ func (q *Queries) GetCollectorFromJobID(ctx context.Context, jobid pgtype.UUID) } const getCollectorQueries = `-- name: GetCollectorQueries :many -SELECT collectorId, queryId, type, requiredQueryId, queryVersion FROM collectorQueryDependencyTree WHERE collectorId = $1 +SELECT collectorId, queryId, type, queryVersion, ARRAY_AGG(requiredQueryId) AS requiredIds + FROM collectorQueryDependencyTree + WHERE collectorId = $1 + GROUP BY queryId, collectorId, type, queryVersion ` type GetCollectorQueriesRow struct { - Collectorid pgtype.UUID - Queryid pgtype.UUID - Type NullQuerytype - Requiredqueryid pgtype.UUID - Queryversion pgtype.Int4 + Collectorid pgtype.UUID + Queryid pgtype.UUID + Type NullQuerytype + Queryversion pgtype.Int4 + Requiredids []pgtype.UUID } func (q *Queries) GetCollectorQueries(ctx context.Context, collectorid pgtype.UUID) ([]GetCollectorQueriesRow, error) { @@ -59,8 +62,8 @@ func (q *Queries) GetCollectorQueries(ctx context.Context, collectorid pgtype.UU &i.Collectorid, &i.Queryid, &i.Type, - &i.Requiredqueryid, &i.Queryversion, + &i.Requiredids, ); err != nil { return nil, err } diff --git a/internal/database/repository/models.go b/internal/database/repository/models.go index db1fccdf..fda9edcf 100644 --- a/internal/database/repository/models.go +++ b/internal/database/repository/models.go @@ -107,6 +107,13 @@ type Queryconfig struct { Removedversion pgtype.Int4 } +type Querydeprecation struct { + ID pgtype.UUID + Queryid pgtype.UUID + Time pgtype.Timestamp + Removedat pgtype.Timestamp +} + type Requiredquery struct { ID pgtype.UUID Queryid pgtype.UUID diff --git a/internal/database/repository/query.sql.go b/internal/database/repository/query.sql.go index edd998b1..0403c2fd 100644 --- a/internal/database/repository/query.sql.go +++ b/internal/database/repository/query.sql.go @@ -11,6 +11,15 @@ import ( "github.com/jackc/pgx/v5/pgtype" ) +const deprecateQuery = `-- name: DeprecateQuery :exec +INSERT INTO queryDeprecations (queryId) VALUES ($1) RETURNING id +` + +func (q *Queries) DeprecateQuery(ctx context.Context, queryid pgtype.UUID) error { + _, err := q.db.Exec(ctx, deprecateQuery, queryid) + return err +} + const getQuery = `-- name: GetQuery :one SELECT q.id, q.type, q.activeVersion, c.config, ARRAY_AGG(r.requiredQueryId) AS requiredIds FROM queries AS q @@ -65,3 +74,27 @@ func (q *Queries) GetQueryConfig(ctx context.Context, arg GetQueryConfigParams) err := row.Scan(&i.ID, &i.Config) return i, err } + +const getQueryType = `-- name: GetQueryType :one +SELECT type FROM queries where id = $1 +` + +func (q *Queries) GetQueryType(ctx context.Context, id pgtype.UUID) (Querytype, error) { + row := q.db.QueryRow(ctx, getQueryType, id) + var type_ Querytype + err := row.Scan(&type_) + return type_, err +} + +const isQueryDeprecated = `-- name: IsQueryDeprecated :one +SELECT EXISTS ( + SELECT 1 FROM queryDeprecations where queryId = $1 and removedAt is not null +) +` + +func (q *Queries) IsQueryDeprecated(ctx context.Context, queryid pgtype.UUID) (bool, error) { + row := q.db.QueryRow(ctx, isQueryDeprecated, queryid) + var exists bool + err := row.Scan(&exists) + return exists, err +} diff --git a/internal/jsonExtractor/creator.go b/internal/jsonExtractor/creator.go new file mode 100644 index 00000000..0298162a --- /dev/null +++ b/internal/jsonExtractor/creator.go @@ -0,0 +1,32 @@ +package jsonextractor + +import ( + "context" + "queryorchestration/internal/database/repository" + queryprocessor "queryorchestration/internal/queryProcessor" + + "github.com/google/uuid" +) + +type Creator struct { + db *repository.Queries +} + +func NewCreator(db *repository.Queries) Creator { + return Creator{db} +} + +func (s Creator) Validate(ctx context.Context, entity *queryprocessor.Create) error { + // TODO + return nil +} + +func (s Creator) Create(ctx context.Context, entity *queryprocessor.Create) (uuid.UUID, error) { + err := s.Validate(ctx, entity) + if err != nil { + return uuid.Nil, err + } + + // TODO + return uuid.Nil, nil +} diff --git a/internal/jsonExtractor/service.go b/internal/jsonExtractor/service.go index ed9609d7..2762a4a1 100644 --- a/internal/jsonExtractor/service.go +++ b/internal/jsonExtractor/service.go @@ -6,7 +6,7 @@ import ( "fmt" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" - "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" "queryorchestration/internal/result" "github.com/tidwall/gjson" @@ -20,11 +20,11 @@ type Config struct { Path string `json:"path"` } -func New(db *repository.Queries) *Extractor { - return &Extractor{db} +func NewExtractor(db *repository.Queries) Extractor { + return Extractor{db} } -func (e *Extractor) Process(ctx context.Context, query query.QueryRow, values *[]result.Value) (string, error) { +func (e Extractor) Process(ctx context.Context, query queryprocessor.Query, values *[]result.Value) (string, error) { if len(*values) != 1 { return "", fmt.Errorf("JSON Extraction requires 1 result") } diff --git a/internal/jsonExtractor/updator.go b/internal/jsonExtractor/updator.go new file mode 100644 index 00000000..a9853824 --- /dev/null +++ b/internal/jsonExtractor/updator.go @@ -0,0 +1,30 @@ +package jsonextractor + +import ( + "context" + "queryorchestration/internal/database/repository" + queryprocessor "queryorchestration/internal/queryProcessor" +) + +type Updator struct { + db *repository.Queries +} + +func NewUpdator(db *repository.Queries) Updator { + return Updator{db} +} + +func (s Updator) Validate(ctx context.Context, entity *queryprocessor.Update) error { + // TODO + return nil +} + +func (s Updator) Update(ctx context.Context, entity *queryprocessor.Update) error { + err := s.Validate(ctx, entity) + if err != nil { + return err + } + + // TODO + return nil +} diff --git a/internal/query/create.go b/internal/query/create.go new file mode 100644 index 00000000..109ca74f --- /dev/null +++ b/internal/query/create.go @@ -0,0 +1,36 @@ +package query + +import ( + "context" + "fmt" + contextfull "queryorchestration/internal/contextFull" + jsonextractor "queryorchestration/internal/jsonExtractor" + queryprocessor "queryorchestration/internal/queryProcessor" + + "github.com/google/uuid" +) + +func (s *Service) Create(ctx context.Context, entity *queryprocessor.Create) (uuid.UUID, error) { + validator, err := s.getCreator(entity.Type) + if err != nil { + return uuid.Nil, err + } + + id, err := validator.Create(ctx, entity) + if err != nil { + return uuid.Nil, err + } + + return id, nil +} + +func (s *Service) getCreator(qType queryprocessor.Type) (queryprocessor.Creator, error) { + switch qType { + case queryprocessor.TypeJsonExtractor: + return jsonextractor.NewCreator(s.db), nil + case queryprocessor.TypeContextFull: + return contextfull.NewCreator(s.db), nil + default: + return nil, fmt.Errorf("attempting to process invalid query type") + } +} diff --git a/internal/query/deprecate.go b/internal/query/deprecate.go new file mode 100644 index 00000000..094f4337 --- /dev/null +++ b/internal/query/deprecate.go @@ -0,0 +1,26 @@ +package query + +import ( + "context" + "queryorchestration/internal/database" + + "github.com/google/uuid" +) + +func (s *Service) Deprecate(ctx context.Context, id uuid.UUID) error { + dbId := database.MustToDBUUID(id) + + exists, err := s.db.IsQueryDeprecated(ctx, dbId) + if err != nil { + return err + } else if exists { + return nil + } + + err = s.db.DeprecateQuery(ctx, dbId) + if err != nil { + return err + } + + return nil +} diff --git a/internal/query/list.go b/internal/query/list.go new file mode 100644 index 00000000..81932355 --- /dev/null +++ b/internal/query/list.go @@ -0,0 +1,10 @@ +package query + +import ( + "context" +) + +func (s *Service) List(ctx context.Context, filters ListFilters) (*[]Query, error) { + // TODO + return nil, nil +} diff --git a/internal/query/service.go b/internal/query/service.go index 2d076832..a6be2760 100644 --- a/internal/query/service.go +++ b/internal/query/service.go @@ -4,54 +4,27 @@ import ( "context" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" - "queryorchestration/internal/result" + queryprocessor "queryorchestration/internal/queryProcessor" "github.com/google/uuid" ) -type Type int - -const ( - TypeJsonExtractor = iota - TypeContextFull -) - -type QueryRow struct { - ID uuid.UUID - Type Type - RequiredQueryID uuid.UUID - Version int32 -} - type Query struct { ID uuid.UUID - Type Type + Type queryprocessor.Type RequiredQueryIDs []uuid.UUID Version int32 Config interface{} } type ListFilters struct { - Types []Type -} - -type Create struct { - Type Type - RequiredQueryIDs []uuid.UUID -} - -type Update struct { - ID uuid.UUID + Types []queryprocessor.Type } type Test struct { ID uuid.UUID } -type Processor interface { - Process(ctx context.Context, query QueryRow, values *[]result.Value) (string, error) -} - type Service struct { db *repository.Queries } @@ -66,7 +39,7 @@ func (s *Service) Get(ctx context.Context, id uuid.UUID) (*Query, error) { return nil, err } - queryType, err := ParseDBType(query.Type) + queryType, err := queryprocessor.ParseDBType(query.Type) if err != nil { return nil, err } @@ -84,26 +57,3 @@ func (s *Service) Get(ctx context.Context, id uuid.UUID) (*Query, error) { Config: string(query.Config), }, nil } - -func (s *Service) List(ctx context.Context, filters ListFilters) (*[]Query, error) { - return nil, nil -} - -func (s *Service) Create(ctx context.Context, entity Create) (uuid.UUID, error) { - return uuid.Nil, nil -} - -func (s *Service) Update(ctx context.Context, entity Update) error { - return nil -} - -func (s *Service) Remove(ctx context.Context, id uuid.UUID) error { - return nil -} - -func (s *Service) Test(ctx context.Context, filters Test) (string, error) { - // Sync doc - // Run test - - return "", nil -} diff --git a/internal/query/test.go b/internal/query/test.go new file mode 100644 index 00000000..42fe2673 --- /dev/null +++ b/internal/query/test.go @@ -0,0 +1,13 @@ +package query + +import ( + "context" +) + +func (s *Service) Test(ctx context.Context, filters Test) (string, error) { + // TODO + // Sync doc + // Run test + + return "", nil +} diff --git a/internal/query/update.go b/internal/query/update.go new file mode 100644 index 00000000..a23630a2 --- /dev/null +++ b/internal/query/update.go @@ -0,0 +1,45 @@ +package query + +import ( + "context" + "fmt" + contextfull "queryorchestration/internal/contextFull" + "queryorchestration/internal/database" + jsonextractor "queryorchestration/internal/jsonExtractor" + queryprocessor "queryorchestration/internal/queryProcessor" +) + +func (s *Service) Update(ctx context.Context, entity *queryprocessor.Update) error { + dbType, err := s.db.GetQueryType(ctx, database.MustToDBUUID(entity.ID)) + if err != nil { + return err + } + + qType, err := queryprocessor.ParseDBType(dbType) + if err != nil { + return err + } + + validator, err := s.getUpdator(qType) + if err != nil { + return err + } + + err = validator.Update(ctx, entity) + if err != nil { + return err + } + + return nil +} + +func (s *Service) getUpdator(qType queryprocessor.Type) (queryprocessor.Updator, error) { + switch qType { + case queryprocessor.TypeJsonExtractor: + return jsonextractor.NewUpdator(s.db), nil + case queryprocessor.TypeContextFull: + return contextfull.NewUpdator(s.db), nil + default: + return nil, fmt.Errorf("attempting to process invalid query type") + } +} diff --git a/internal/query/database.go b/internal/queryProcessor/parse.go similarity index 69% rename from internal/query/database.go rename to internal/queryProcessor/parse.go index 28afeb29..70cd71e5 100644 --- a/internal/query/database.go +++ b/internal/queryProcessor/parse.go @@ -1,25 +1,13 @@ -package query +package queryprocessor import ( "fmt" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" + + "github.com/google/uuid" ) -func ParseDBQueryRow(dbQuery *repository.GetCollectorQueriesRow) (*QueryRow, error) { - t, err := ParseDBNullType(dbQuery.Type) - if err != nil { - return nil, err - } - - return &QueryRow{ - ID: database.MustToUUID(dbQuery.Queryid), - Type: t, - RequiredQueryID: database.MustToUUID(dbQuery.Requiredqueryid), - Version: dbQuery.Queryversion.Int32, - }, nil -} - func ParseDBNullType(qType repository.NullQuerytype) (Type, error) { if !qType.Valid { return TypeJsonExtractor, fmt.Errorf("invalid database query type") @@ -53,3 +41,22 @@ func ToDBQueryType(t Type) (repository.NullQuerytype, error) { return repository.NullQuerytype{Querytype: dbType, Valid: true}, nil } + +func ParseDBQuery(q *repository.GetCollectorQueriesRow) (*Query, error) { + reqQueryIDs := make([]uuid.UUID, len(q.Requiredids)) + for index, id := range q.Requiredids { + reqQueryIDs[index] = database.MustToUUID(id) + } + + qType, err := ParseDBNullType(q.Type) + if err != nil { + return nil, err + } + + return &Query{ + ID: database.MustToUUID(q.Queryid), + Version: q.Queryversion.Int32, + Type: qType, + RequiredQueryIDs: reqQueryIDs, + }, nil +} diff --git a/internal/queryProcessor/service.go b/internal/queryProcessor/service.go new file mode 100644 index 00000000..7603a924 --- /dev/null +++ b/internal/queryProcessor/service.go @@ -0,0 +1,49 @@ +package queryprocessor + +import ( + "context" + "queryorchestration/internal/result" + + "github.com/google/uuid" +) + +type Type int + +const ( + TypeJsonExtractor = iota + TypeContextFull +) + +type Create struct { + Type Type + RequiredQueryIDs []uuid.UUID + Config interface{} +} + +type Update struct { + ID uuid.UUID + RequiredQueryIDs []uuid.UUID + Config interface{} +} + +type Query struct { + ID uuid.UUID + Type Type + Version int32 + RequiredQueryIDs []uuid.UUID + Config interface{} +} + +type Creator interface { + Validate(ctx context.Context, entity *Create) error + Create(ctx context.Context, entity *Create) (uuid.UUID, error) +} + +type Updator interface { + Validate(ctx context.Context, entity *Update) error + Update(ctx context.Context, entity *Update) error +} + +type Processor interface { + Process(ctx context.Context, query Query, values *[]result.Value) (string, error) +} diff --git a/internal/queryQueue/create.go b/internal/queryQueue/create.go index 37cbf181..bfeb0618 100644 --- a/internal/queryQueue/create.go +++ b/internal/queryQueue/create.go @@ -3,7 +3,7 @@ package queryQueue import ( "context" "queryorchestration/internal/database" - "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" ) func (q *Queue) getUnsyncedQueries() { @@ -36,9 +36,9 @@ func (c *Queue) getCollectorQueries(ctx context.Context) error { return err } - cleanQueries := make([]query.QueryRow, len(queries)) + cleanQueries := make([]queryprocessor.Query, len(queries)) for index, dbQuery := range queries { - cleanQuery, err := query.ParseDBQueryRow(&dbQuery) + cleanQuery, err := queryprocessor.ParseDBQuery(&dbQuery) if err != nil { return err } @@ -51,33 +51,39 @@ func (c *Queue) getCollectorQueries(ctx context.Context) error { return nil } -func (q *Queue) Add(qu *query.QueryRow) { - dependentQueries := []query.QueryRow{} +func (q *Queue) Add(qu *queryprocessor.Query) { + dependentQueries := []queryprocessor.Query{} requiredIndex := -1 if q.unsyncedQueue == nil { - q.unsyncedQueue = &[]query.QueryRow{} + q.unsyncedQueue = &[]queryprocessor.Query{} } else { for index, entry := range *q.unsyncedQueue { if entry.ID == qu.ID { return } - if entry.ID == qu.RequiredQueryID { - requiredIndex = index + for _, id := range qu.RequiredQueryIDs { + if entry.ID == id { + requiredIndex = index + break + } } } } for _, entry := range *q.collectorQueries { - if entry.RequiredQueryID == qu.ID { - dependentQueries = append(dependentQueries, entry) + for _, id := range entry.RequiredQueryIDs { + if qu.ID == id { + dependentQueries = append(dependentQueries, entry) + break + } } } if requiredIndex != -1 { - *q.unsyncedQueue = append((*q.unsyncedQueue)[:requiredIndex+1], append([]query.QueryRow{*qu}, (*q.unsyncedQueue)[requiredIndex+1:]...)...) + *q.unsyncedQueue = append((*q.unsyncedQueue)[:requiredIndex+1], append([]queryprocessor.Query{*qu}, (*q.unsyncedQueue)[requiredIndex+1:]...)...) } else { - *q.unsyncedQueue = append([]query.QueryRow{*qu}, *q.unsyncedQueue...) + *q.unsyncedQueue = append([]queryprocessor.Query{*qu}, *q.unsyncedQueue...) } for _, entry := range dependentQueries { diff --git a/internal/queryQueue/execute.go b/internal/queryQueue/execute.go index 3829b733..f82faf42 100644 --- a/internal/queryQueue/execute.go +++ b/internal/queryQueue/execute.go @@ -7,10 +7,9 @@ import ( "queryorchestration/internal/database" "queryorchestration/internal/database/repository" jsonextractor "queryorchestration/internal/jsonExtractor" - "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" "queryorchestration/internal/result" - "github.com/google/uuid" "github.com/jackc/pgx/v5/pgtype" ) @@ -31,16 +30,9 @@ func (q *Queue) Execute(ctx context.Context) error { return nil } -func (q *Queue) executeQuery(ctx context.Context, qu query.QueryRow) error { - requiredQueryIDs := []uuid.UUID{} - for _, entry := range *q.collectorQueries { - if entry.ID == qu.ID && entry.RequiredQueryID != uuid.Nil { - requiredQueryIDs = append(requiredQueryIDs, entry.RequiredQueryID) - } - } - - resultIDs := make([]pgtype.UUID, len(requiredQueryIDs)) - for index, id := range requiredQueryIDs { +func (q *Queue) executeQuery(ctx context.Context, qu queryprocessor.Query) error { + resultIDs := make([]pgtype.UUID, len(qu.RequiredQueryIDs)) + for index, id := range qu.RequiredQueryIDs { var queryVersion int32 for _, entry := range *q.collectorQueries { if entry.ID == id { @@ -81,7 +73,7 @@ func (q *Queue) executeQuery(ctx context.Context, qu query.QueryRow) error { } func (q *Queue) getResultValue(res *repository.ListResultValuesByIDRow) (result.Value, error) { - var queryType query.Type + var queryType queryprocessor.Type for _, qu := range *q.collectorQueries { if qu.ID == database.MustToUUID(res.Queryid) { queryType = qu.Type @@ -89,9 +81,9 @@ func (q *Queue) getResultValue(res *repository.ListResultValuesByIDRow) (result. } switch queryType { - case query.TypeJsonExtractor: + case queryprocessor.TypeJsonExtractor: return jsonextractor.NewResult(res.Value), nil - case query.TypeContextFull: + case queryprocessor.TypeContextFull: return contextfull.NewResult(res.Value), nil default: return nil, fmt.Errorf("attempting to process invalid query type") diff --git a/internal/queryQueue/result.go b/internal/queryQueue/result.go index 6e350953..c5cddf07 100644 --- a/internal/queryQueue/result.go +++ b/internal/queryQueue/result.go @@ -5,11 +5,11 @@ import ( "fmt" contextfull "queryorchestration/internal/contextFull" jsonextractor "queryorchestration/internal/jsonExtractor" - "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" "queryorchestration/internal/result" ) -func (q *Queue) setResult(ctx context.Context, qu query.QueryRow, resultValues *[]result.Value) error { +func (q *Queue) setResult(ctx context.Context, qu queryprocessor.Query, resultValues *[]result.Value) error { processor, err := q.getProcessor(qu.Type) if err != nil { return err @@ -41,12 +41,12 @@ func (q *Queue) setResult(ctx context.Context, qu query.QueryRow, resultValues * return nil } -func (q *Queue) getProcessor(queryType query.Type) (query.Processor, error) { +func (q *Queue) getProcessor(queryType queryprocessor.Type) (queryprocessor.Processor, error) { switch queryType { - case query.TypeJsonExtractor: - return jsonextractor.New(q.db), nil - case query.TypeContextFull: - return contextfull.New(), nil + case queryprocessor.TypeJsonExtractor: + return jsonextractor.NewExtractor(q.db), nil + case queryprocessor.TypeContextFull: + return contextfull.NewExtractor(), nil default: return nil, fmt.Errorf("attempting to process invalid query type") } diff --git a/internal/queryQueue/service.go b/internal/queryQueue/service.go index 88070874..152df861 100644 --- a/internal/queryQueue/service.go +++ b/internal/queryQueue/service.go @@ -4,15 +4,15 @@ import ( "context" "queryorchestration/internal/collector" "queryorchestration/internal/database/repository" - "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" "queryorchestration/internal/result" "github.com/google/uuid" ) type Queue struct { - unsyncedQueue *[]query.QueryRow - collectorQueries *[]query.QueryRow + unsyncedQueue *[]queryprocessor.Query + collectorQueries *[]queryprocessor.Query results *[]result.Result collector *collector.Collector db *repository.Queries @@ -41,6 +41,6 @@ func New(ctx context.Context, db *repository.Queries, coll *collector.Collector, return &queue, nil } -func (q *Queue) GetQueue() []query.QueryRow { +func (q *Queue) GetQueue() []queryprocessor.Query { return *q.unsyncedQueue } diff --git a/scripts/proto.yml b/scripts/proto.yml index 02dea2b9..9644a489 100644 --- a/scripts/proto.yml +++ b/scripts/proto.yml @@ -15,6 +15,4 @@ tasks: - protolint lint -fix {{.PROTO_DIR}} generate: cmds: - - pwd - - mkdir -p {{.API_DIR}}/serviceInterfaces - protoc --proto_path={{.PROTO_DIR}} --go_out={{.API_DIR}}/serviceInterfaces --go-grpc_out={{.API_DIR}} --go_opt=paths=source_relative --experimental_allow_proto3_optional main.proto diff --git a/test/unit/internal/contextfull/process_test.go b/test/unit/internal/contextfull/process_test.go index 027d2257..73ec923a 100644 --- a/test/unit/internal/contextfull/process_test.go +++ b/test/unit/internal/contextfull/process_test.go @@ -3,7 +3,7 @@ package document_test import ( "context" contextfull "queryorchestration/internal/contextFull" - "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" "queryorchestration/internal/result" "testing" @@ -14,13 +14,13 @@ import ( func TestContextFull(t *testing.T) { ctx := context.Background() - extractor := contextfull.New() + extractor := contextfull.NewExtractor() - query := query.QueryRow{ - ID: uuid.New(), - Type: query.TypeJsonExtractor, - RequiredQueryID: uuid.Nil, - Version: int32(1), + query := queryprocessor.Query{ + ID: uuid.New(), + Type: queryprocessor.TypeJsonExtractor, + RequiredQueryIDs: []uuid.UUID{}, + Version: int32(1), } values := &[]result.Value{} diff --git a/test/unit/internal/document/sync_test.go b/test/unit/internal/document/sync_test.go index ef776a7e..dafb210d 100644 --- a/test/unit/internal/document/sync_test.go +++ b/test/unit/internal/document/sync_test.go @@ -48,8 +48,8 @@ func TestSyncIsSynced(t *testing.T) { ) db.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorId). WillReturnRows( - pgxmock.NewRows([]string{"collectorId", "queryId", "type", "requiredQueryId", "queryVersion"}). - AddRow(dbCollectorId, pgtype.UUID{}, repository.NullQuerytype{Querytype: repository.QuerytypeJsonExtractor, Valid: true}, pgtype.UUID{}, pgtype.Int4{Int32: int32(1), Valid: true}), + pgxmock.NewRows([]string{"collectorId", "queryId", "type", "queryVersion", "requiredIds"}). + AddRow(dbCollectorId, pgtype.UUID{}, repository.NullQuerytype{Querytype: repository.QuerytypeJsonExtractor, Valid: true}, pgtype.Int4{Int32: int32(1), Valid: true}, []pgtype.UUID{}), ) docSvc := document.New(queries) @@ -131,8 +131,8 @@ func TestSyncDBFail(t *testing.T) { errr = "database failure" db.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorId). WillReturnRows( - pgxmock.NewRows([]string{"collectorId", "queryId", "type", "requiredQueryId", "queryVersion"}). - AddRow(dbCollectorId, dbQueryID, repository.NullQuerytype{Querytype: repository.QuerytypeJsonExtractor, Valid: true}, pgtype.UUID{}, pgtype.Int4{Int32: int32(1), Valid: true}), + pgxmock.NewRows([]string{"collectorId", "queryId", "type", "queryVersion", "requiredIds"}). + AddRow(dbCollectorId, dbQueryID, repository.NullQuerytype{Querytype: repository.QuerytypeJsonExtractor, Valid: true}, pgtype.Int4{Int32: int32(1), Valid: true}, []pgtype.UUID{}), ) db.ExpectQuery("name: ListResultValuesByID :many").WithArgs([]pgtype.UUID{}). WillReturnError(errors.New(errr)) @@ -176,8 +176,8 @@ func TestSync(t *testing.T) { ) db.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorId). WillReturnRows( - pgxmock.NewRows([]string{"collectorId", "queryId", "type", "requiredQueryId", "queryVersion"}). - AddRow(dbCollectorId, dbQueryID, repository.NullQuerytype{Querytype: repository.QuerytypeJsonExtractor, Valid: true}, pgtype.UUID{}, pgtype.Int4{Int32: int32(1), Valid: true}), + pgxmock.NewRows([]string{"collectorId", "queryId", "type", "queryVersion", "requiredIds"}). + AddRow(dbCollectorId, dbQueryID, repository.NullQuerytype{Querytype: repository.QuerytypeJsonExtractor, Valid: true}, pgtype.Int4{Int32: int32(1), Valid: true}, []pgtype.UUID{}), ) db.ExpectQuery("name: ListResultValuesByID :many").WithArgs([]pgtype.UUID{}). WillReturnRows( diff --git a/test/unit/internal/jsonextractor/process_test.go b/test/unit/internal/jsonextractor/process_test.go index 7033b6de..0cfe4221 100644 --- a/test/unit/internal/jsonextractor/process_test.go +++ b/test/unit/internal/jsonextractor/process_test.go @@ -7,7 +7,7 @@ import ( "queryorchestration/internal/database" "queryorchestration/internal/database/repository" jsonextractor "queryorchestration/internal/jsonExtractor" - "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" "queryorchestration/internal/result" "testing" @@ -28,13 +28,13 @@ func TestJSONProcess(t *testing.T) { queries := repository.New(db) - extractor := jsonextractor.New(queries) + extractor := jsonextractor.NewExtractor(queries) - query := query.QueryRow{ - ID: uuid.New(), - Type: query.TypeJsonExtractor, - RequiredQueryID: uuid.Nil, - Version: int32(1), + query := queryprocessor.Query{ + ID: uuid.New(), + Type: queryprocessor.TypeJsonExtractor, + RequiredQueryIDs: []uuid.UUID{}, + Version: int32(1), } entryValue := "value" @@ -97,13 +97,13 @@ func TestJSONProcessJSON(t *testing.T) { queries := repository.New(db) - extractor := jsonextractor.New(queries) + extractor := jsonextractor.NewExtractor(queries) - query := query.QueryRow{ - ID: uuid.New(), - Type: query.TypeJsonExtractor, - RequiredQueryID: uuid.Nil, - Version: int32(1), + query := queryprocessor.Query{ + ID: uuid.New(), + Type: queryprocessor.TypeJsonExtractor, + RequiredQueryIDs: []uuid.UUID{}, + Version: int32(1), } entryValue := "value" @@ -183,13 +183,13 @@ func TestJSONProcessResults(t *testing.T) { queries := repository.New(db) - extractor := jsonextractor.New(queries) + extractor := jsonextractor.NewExtractor(queries) - query := query.QueryRow{ - ID: uuid.New(), - Type: query.TypeJsonExtractor, - RequiredQueryID: uuid.Nil, - Version: int32(1), + query := queryprocessor.Query{ + ID: uuid.New(), + Type: queryprocessor.TypeJsonExtractor, + RequiredQueryIDs: []uuid.UUID{}, + Version: int32(1), } results := &[]result.Value{} diff --git a/test/unit/internal/query/get_test.go b/test/unit/internal/query/get_test.go index 4760e76d..9d47d08c 100644 --- a/test/unit/internal/query/get_test.go +++ b/test/unit/internal/query/get_test.go @@ -5,6 +5,7 @@ import ( "queryorchestration/internal/database" "queryorchestration/internal/database/repository" "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" "testing" "github.com/google/uuid" @@ -28,7 +29,7 @@ func TestGet(t *testing.T) { config := "{\"path\":\"example_path\"}" query := query.Query{ ID: uuid.New(), - Type: query.TypeJsonExtractor, + Type: queryprocessor.TypeJsonExtractor, Version: int32(1), RequiredQueryIDs: []uuid.UUID{ uuid.New(), diff --git a/test/unit/internal/query/parse_test.go b/test/unit/internal/query/parse_test.go index b3423fb5..5c423173 100644 --- a/test/unit/internal/query/parse_test.go +++ b/test/unit/internal/query/parse_test.go @@ -2,7 +2,7 @@ package document_test import ( "queryorchestration/internal/database/repository" - "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" "testing" "github.com/google/uuid" @@ -12,63 +12,63 @@ import ( func TestParseDBQueryRow(t *testing.T) { dbResult := repository.GetCollectorQueriesRow{ - Collectorid: pgtype.UUID{}, - Queryid: pgtype.UUID{}, - Requiredqueryid: pgtype.UUID{}, - Type: repository.NullQuerytype{Valid: true, Querytype: repository.QuerytypeJsonExtractor}, - Queryversion: pgtype.Int4{}, + Collectorid: pgtype.UUID{}, + Queryid: pgtype.UUID{}, + Requiredids: []pgtype.UUID{}, + Type: repository.NullQuerytype{Valid: true, Querytype: repository.QuerytypeJsonExtractor}, + Queryversion: pgtype.Int4{}, } - value, err := query.ParseDBQueryRow(&dbResult) + value, err := queryprocessor.ParseDBQuery(&dbResult) assert.Nil(t, err) assert.Equal(t, uuid.Nil, value.ID) - assert.Equal(t, uuid.Nil, value.RequiredQueryID) + assert.Equal(t, []uuid.UUID{}, value.RequiredQueryIDs) assert.Equal(t, int32(0), value.Version) - assert.Equal(t, query.Type(query.TypeJsonExtractor), value.Type) + assert.Equal(t, queryprocessor.Type(queryprocessor.TypeJsonExtractor), value.Type) dbResult.Type = repository.NullQuerytype{} - _, err = query.ParseDBQueryRow(&dbResult) + _, err = queryprocessor.ParseDBQuery(&dbResult) assert.EqualError(t, err, "invalid database query type") } func TestParseDBNullType(t *testing.T) { qType := repository.NullQuerytype{Valid: true, Querytype: repository.QuerytypeJsonExtractor} - value, err := query.ParseDBNullType(qType) + value, err := queryprocessor.ParseDBNullType(qType) assert.Nil(t, err) - assert.Equal(t, query.Type(query.TypeJsonExtractor), value) + assert.Equal(t, queryprocessor.Type(queryprocessor.TypeJsonExtractor), value) qType = repository.NullQuerytype{} - _, err = query.ParseDBNullType(qType) + _, err = queryprocessor.ParseDBNullType(qType) assert.EqualError(t, err, "invalid database query type") qType = repository.NullQuerytype{Valid: true} - _, err = query.ParseDBNullType(qType) + _, err = queryprocessor.ParseDBNullType(qType) assert.EqualError(t, err, "invalid database query type") } func TestParseDBType(t *testing.T) { qType := repository.QuerytypeJsonExtractor - value, err := query.ParseDBType(qType) + value, err := queryprocessor.ParseDBType(qType) assert.Nil(t, err) - assert.Equal(t, query.Type(query.TypeJsonExtractor), value) + assert.Equal(t, queryprocessor.Type(queryprocessor.TypeJsonExtractor), value) qType = repository.QuerytypeContextFull - value, err = query.ParseDBType(qType) + value, err = queryprocessor.ParseDBType(qType) assert.Nil(t, err) - assert.Equal(t, query.Type(query.TypeContextFull), value) + assert.Equal(t, queryprocessor.Type(queryprocessor.TypeContextFull), value) } func TestToDBQueryType(t *testing.T) { - dbQueryType := query.Type(query.TypeJsonExtractor) - value, err := query.ToDBQueryType(dbQueryType) + dbQueryType := queryprocessor.Type(queryprocessor.TypeJsonExtractor) + value, err := queryprocessor.ToDBQueryType(dbQueryType) assert.Nil(t, err) assert.Equal(t, repository.NullQuerytype{Querytype: repository.QuerytypeJsonExtractor, Valid: true}, value) - dbQueryType = query.Type(-1) - _, err = query.ToDBQueryType(dbQueryType) + dbQueryType = queryprocessor.Type(-1) + _, err = queryprocessor.ToDBQueryType(dbQueryType) assert.EqualError(t, err, "invalid database query type") - dbQueryType = query.Type(query.TypeContextFull) - value, err = query.ToDBQueryType(dbQueryType) + dbQueryType = queryprocessor.Type(queryprocessor.TypeContextFull) + value, err = queryprocessor.ToDBQueryType(dbQueryType) assert.Nil(t, err) assert.Equal(t, repository.NullQuerytype{Querytype: repository.QuerytypeContextFull, Valid: true}, value) } diff --git a/test/unit/internal/queryQueue/queue_test.go b/test/unit/internal/queryQueue/queue_test.go index ba334555..e2284933 100644 --- a/test/unit/internal/queryQueue/queue_test.go +++ b/test/unit/internal/queryQueue/queue_test.go @@ -7,7 +7,7 @@ import ( "queryorchestration/internal/collector" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" - "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" "queryorchestration/internal/queryQueue" "queryorchestration/internal/result" "testing" @@ -56,24 +56,27 @@ func TestQueue(t *testing.T) { querySixVersion := int32(6) contextID := uuid.New() contextVersion := int32(1) - collectorQueries := []query.QueryRow{ - {ID: contextID, Type: query.TypeContextFull, RequiredQueryID: uuid.Nil, Version: contextVersion}, - {ID: queryOneID, Type: query.TypeJsonExtractor, RequiredQueryID: contextID, Version: queryOneVersion}, - {ID: queryTwoID, Type: query.TypeJsonExtractor, RequiredQueryID: queryOneID, Version: queryTwoVersion}, - {ID: queryThreeID, Type: query.TypeJsonExtractor, RequiredQueryID: queryOneID, Version: queryThreeVersion}, - {ID: queryFourID, Type: query.TypeJsonExtractor, RequiredQueryID: contextID, Version: queryFourVersion}, - {ID: queryFiveID, Type: query.TypeJsonExtractor, RequiredQueryID: querySixID, Version: queryFiveVersion}, - {ID: querySixID, Type: query.TypeJsonExtractor, RequiredQueryID: contextID, Version: querySixVersion}, + collectorQueries := []queryprocessor.Query{ + {ID: contextID, Type: queryprocessor.TypeContextFull, RequiredQueryIDs: []uuid.UUID{}, Version: contextVersion}, + {ID: queryOneID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{contextID}, Version: queryOneVersion}, + {ID: queryTwoID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{queryOneID}, Version: queryTwoVersion}, + {ID: queryThreeID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{queryOneID}, Version: queryThreeVersion}, + {ID: queryFourID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{contextID}, Version: queryFourVersion}, + {ID: queryFiveID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{querySixID}, Version: queryFiveVersion}, + {ID: querySixID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{contextID}, Version: querySixVersion}, } - rows := pgxmock.NewRows([]string{"collectorId", "queryId", "type", "requiredQueryId", "queryVersion"}) + rows := pgxmock.NewRows([]string{"collectorId", "queryId", "type", "queryVersion", "requiredIds"}) for _, q := range collectorQueries { dbID := database.MustToDBUUID(q.ID) - dbReqID := database.MustToDBUUID(q.RequiredQueryID) - ty, err := query.ToDBQueryType(q.Type) + dbReqIDs := make([]pgtype.UUID, len(q.RequiredQueryIDs)) + for index, id := range q.RequiredQueryIDs { + dbReqIDs[index] = database.MustToDBUUID(id) + } + ty, err := queryprocessor.ToDBQueryType(q.Type) assert.Nil(t, err) rows = rows. - AddRow(dbCollectorID, dbID, ty, dbReqID, pgtype.Int4{Int32: int32(q.Version), Valid: true}) + AddRow(dbCollectorID, dbID, ty, pgtype.Int4{Int32: int32(q.Version), Valid: true}, dbReqIDs) } db.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorID).WillReturnRows(rows) @@ -86,12 +89,12 @@ func TestQueue(t *testing.T) { {ID: uuid.New(), QueryID: queryOneID, QueryVersion: queryOneVersion - 1}, } - expectedQueries := []query.QueryRow{ - {ID: querySixID, Type: query.TypeJsonExtractor, RequiredQueryID: contextID, Version: querySixVersion}, - {ID: queryFiveID, Type: query.TypeJsonExtractor, RequiredQueryID: querySixID, Version: queryFiveVersion}, - {ID: queryOneID, Type: query.TypeJsonExtractor, RequiredQueryID: contextID, Version: queryOneVersion}, - {ID: queryThreeID, Type: query.TypeJsonExtractor, RequiredQueryID: queryOneID, Version: queryThreeVersion}, - {ID: queryTwoID, Type: query.TypeJsonExtractor, RequiredQueryID: queryOneID, Version: queryTwoVersion}, + expectedQueries := []queryprocessor.Query{ + {ID: querySixID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{contextID}, Version: querySixVersion}, + {ID: queryFiveID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{querySixID}, Version: queryFiveVersion}, + {ID: queryOneID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{contextID}, Version: queryOneVersion}, + {ID: queryThreeID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{queryOneID}, Version: queryThreeVersion}, + {ID: queryTwoID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{queryOneID}, Version: queryTwoVersion}, } docID := uuid.New()