diff --git a/api/controllers/parse.go b/api/controllers/parse.go new file mode 100644 index 00000000..70521ab6 --- /dev/null +++ b/api/controllers/parse.go @@ -0,0 +1,45 @@ +package controllers + +import ( + serviceinterfaces "queryorchestration/api/serviceInterfaces" + "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" +) + +func ParseQuery(query *query.Query) *serviceinterfaces.Query { + requiredQueries := make([]string, len(query.RequiredQueryIDs)) + for index, id := range query.RequiredQueryIDs { + requiredQueries[index] = id.String() + } + + return &serviceinterfaces.Query{ + Id: query.ID.String(), + Type: ParseQueryType(query.Type), + ActiveVersion: query.ActiveVersion, + LatestVersion: query.LatestVersion, + RequiredQueries: requiredQueries, + Config: &query.Config, + } +} + +func ParseQueryType(qType queryprocessor.Type) serviceinterfaces.QueryType { + switch qType { + case queryprocessor.TypeJsonExtractor: + return serviceinterfaces.QueryType_QUERY_TYPE_JSON_EXTRACTOR + case queryprocessor.TypeContextFull: + return serviceinterfaces.QueryType_QUERY_TYPE_CONTEXT_FULL + default: + return serviceinterfaces.QueryType_QUERY_TYPE_UNSPECIFIED + } +} + +func ParseSpecQueryType(qType serviceinterfaces.QueryType) queryprocessor.Type { + switch qType { + case serviceinterfaces.QueryType_QUERY_TYPE_JSON_EXTRACTOR: + return queryprocessor.TypeJsonExtractor + case serviceinterfaces.QueryType_QUERY_TYPE_CONTEXT_FULL: + return queryprocessor.TypeContextFull + default: + return queryprocessor.TypeContextFull + } +} diff --git a/api/controllers/query.go b/api/controllers/query.go index d72aad9b..c1ea099a 100644 --- a/api/controllers/query.go +++ b/api/controllers/query.go @@ -2,11 +2,16 @@ package controllers import ( "context" + "fmt" serviceinterfaces "queryorchestration/api/serviceInterfaces" "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" "github.com/go-playground/validator/v10" + "github.com/google/uuid" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "google.golang.org/protobuf/types/known/emptypb" ) @@ -24,25 +29,118 @@ func NewQueryController(querySvc query.Service, validator *validator.Validate) * } func (s *QueryController) List(ctx context.Context, req *serviceinterfaces.QueryFilter) (*serviceinterfaces.Queries, error) { - return &serviceinterfaces.Queries{}, nil + types := make([]queryprocessor.Type, len(req.GetTypes())) + for index, t := range req.GetTypes() { + types[index] = queryprocessor.Type(ParseSpecQueryType(t)) + } + + filters := query.ListFilters{ + Types: types, + } + + queries, err := s.query.List(ctx, filters) + if err != nil { + return nil, status.Error(codes.NotFound, fmt.Sprintf("Unable to list query: %s", err)) + } + + outQueries := make([]*serviceinterfaces.Query, len(queries)) + for index, query := range queries { + outQueries[index] = ParseQuery(query) + } + + return &serviceinterfaces.Queries{ + Queries: outQueries, + }, nil } func (s *QueryController) Get(ctx context.Context, req *serviceinterfaces.IdMessage) (*serviceinterfaces.Query, error) { - return &serviceinterfaces.Query{}, nil + id, err := uuid.Parse(req.GetId()) + if err != nil { + return nil, status.Error(codes.InvalidArgument, "Invalid ID") + } + + query, err := s.query.Get(ctx, id) + if err != nil { + return nil, status.Error(codes.NotFound, fmt.Sprintf("Unable to get query: %s", err)) + } + + return ParseQuery(query), nil } func (s *QueryController) Create(ctx context.Context, req *serviceinterfaces.QueryCreate) (*serviceinterfaces.IdMessage, error) { - return &serviceinterfaces.IdMessage{}, nil + requiredQueryIDs := make([]uuid.UUID, len(req.GetRequiredQueries())) + for index, id := range req.GetRequiredQueries() { + parsedID, err := uuid.Parse(id) + if err != nil { + return nil, status.Error(codes.InvalidArgument, "Invalid ID") + } + + requiredQueryIDs[index] = parsedID + } + + id, err := s.query.Create(ctx, &queryprocessor.Create{ + Type: ParseSpecQueryType(req.GetType()), + RequiredQueryIDs: requiredQueryIDs, + }) + if err != nil { + return nil, status.Error(codes.Internal, fmt.Sprintf("Unable to create query: %s", err)) + } + + return &serviceinterfaces.IdMessage{ + Id: id.String(), + }, nil } func (s *QueryController) Update(ctx context.Context, req *serviceinterfaces.QueryUpdate) (*emptypb.Empty, error) { + id, err := uuid.Parse(req.GetId()) + if err != nil { + return nil, status.Error(codes.InvalidArgument, "Invalid ID") + } + + err = s.query.Update(ctx, &queryprocessor.Update{ + ID: id, + }) + if err != nil { + return nil, status.Error(codes.Internal, fmt.Sprintf("Unable to update query: %s", err)) + } + 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 *serviceinterfaces.IdMessage) (*emptypb.Empty, error) { + id, err := uuid.Parse(req.GetId()) + if err != nil { + return nil, status.Error(codes.InvalidArgument, "Invalid ID") + } + + err = s.query.Deprecate(ctx, id) + if err != nil { + return nil, status.Error(codes.Internal, fmt.Sprintf("Unable to deprecate query: %s", err)) + } + return &emptypb.Empty{}, nil } func (s *QueryController) Test(ctx context.Context, req *serviceinterfaces.QueryTestRequest) (*serviceinterfaces.QueryTestResponse, error) { - return &serviceinterfaces.QueryTestResponse{}, nil + queryId, err := uuid.Parse(req.GetQueryId()) + if err != nil { + return nil, status.Error(codes.InvalidArgument, "Invalid Query ID") + } + docId, err := uuid.Parse(req.GetDocumentId()) + if err != nil { + return nil, status.Error(codes.InvalidArgument, "Invalid Document ID") + } + + value, err := s.query.Test(ctx, query.Test{ + QueryID: queryId, + QueryVersion: req.GetQueryVersion(), + DocumentID: docId, + }) + if err != nil { + return nil, status.Error(codes.Internal, fmt.Sprintf("Unable to test query: %s", err)) + } + + return &serviceinterfaces.QueryTestResponse{ + Value: value, + }, nil } diff --git a/api/serviceInterfaces b/api/serviceInterfaces index c41bce65..6da8d1c2 160000 --- a/api/serviceInterfaces +++ b/api/serviceInterfaces @@ -1 +1 @@ -Subproject commit c41bce6522850bb8d605a5ee33d929708377d10b +Subproject commit 6da8d1c276b875f3e4a1bc2fe3efeb5c764e8e06 diff --git a/cmd/queryRunner/main.go b/cmd/queryRunner/main.go index 4d3e1265..804a646d 100644 --- a/cmd/queryRunner/main.go +++ b/cmd/queryRunner/main.go @@ -34,8 +34,12 @@ func main() { sqsClient := sqs.NewFromConfig(cfg) - conn := database.GetDBConn(ctx) - db := repository.New(conn) + dbPool := database.GetDBPool(ctx) + dbQueries := repository.New(dbPool) + db := &database.Connection{ + Pool: dbPool, + Queries: dbQueries, + } valid := validator.New() controllers := queue.Controllers{ diff --git a/cmd/queryService/main.go b/cmd/queryService/main.go index 8640e540..be2a9b5d 100644 --- a/cmd/queryService/main.go +++ b/cmd/queryService/main.go @@ -37,8 +37,12 @@ func main() { log.Panicf("failed to listen: %v", err) } - pool := database.GetDBPool(ctx) - db := repository.New(pool) + dbPool := database.GetDBPool(ctx) + dbQueries := repository.New(dbPool) + db := &database.Connection{ + Pool: dbPool, + Queries: dbQueries, + } valid := validator.New() grpcServer := grpc.NewServer() 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 aa2aca7b..08488057 100644 --- a/database/queries/query.sql +++ b/database/queries/query.sql @@ -1,2 +1,42 @@ -- name: GetQueryConfig :one -SELECT id, config FROM queryConfigs where queryId = $1 and addedVersion >= $2 and COALESCE(removedVersion, $2 - 1) < $2; \ No newline at end of file +SELECT id, config FROM queryConfigs where queryId = $1 and addedVersion >= $2 and COALESCE(removedVersion, $2 - 1) < $2; + +-- name: DeprecateQuery :exec +INSERT INTO queryDeprecations (queryId) VALUES ($1); + +-- 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, q.latestVersion, c.config, ARRAY_AGG(DISTINCT r.requiredQueryId) as requiredIds + FROM queries AS q + LEFT JOIN queryConfigs AS c ON q.id = c.queryId + LEFT JOIN requiredQueries AS r ON q.id = r.queryId + WHERE q.id = $1 + and c.addedVersion >= q.activeVersion + and COALESCE(c.removedVersion, q.activeVersion - 1) < q.activeVersion + and r.addedVersion >= q.activeVersion + and COALESCE(r.removedVersion, q.activeVersion - 1) < q.activeVersion + GROUP BY q.id, c.config; + +-- name: ListQueries :many +SELECT q.id, q.type, q.activeVersion, q.latestVersion, c.config, ARRAY_AGG(r.requiredQueryId) AS requiredIds + FROM queries AS q + LEFT JOIN queryConfigs AS c ON q.id = c.queryId + LEFT JOIN requiredQueries AS r ON q.id = r.queryId + WHERE c.addedVersion >= q.activeVersion + and COALESCE(c.removedVersion, q.activeVersion - 1) < q.activeVersion + and r.addedVersion >= q.activeVersion + and COALESCE(r.removedVersion, q.activeVersion - 1) < q.activeVersion + GROUP BY q.id, c.config; + +-- name: CreateQuery :one +INSERT INTO queries (type) VALUES ($1) RETURNING id; + +-- name: CreateRequiredQuery :exec +INSERT INTO requiredQueries (queryId, requiredQueryId, addedVersion) VALUES ($1, $2, $3); + +-- name: CreateQueryConfig :exec +INSERT INTO queryConfigs (queryId, config, addedVersion) VALUES ($1, $2, $3); \ No newline at end of file diff --git a/internal/collector/collector.go b/internal/collector/collector.go index e6a6e570..9f178719 100644 --- a/internal/collector/collector.go +++ b/internal/collector/collector.go @@ -3,7 +3,6 @@ package collector import ( "context" "queryorchestration/internal/database" - "queryorchestration/internal/database/repository" "github.com/google/uuid" ) @@ -12,10 +11,10 @@ type Collector struct { ID uuid.UUID MinCleanVersion int32 MinTextVersion int32 - db *repository.Queries + db *database.Connection } -func NewByJobId(ctx context.Context, db *repository.Queries, jobID uuid.UUID) (*Collector, error) { +func NewByJobId(ctx context.Context, db *database.Connection, jobID uuid.UUID) (*Collector, error) { collector := Collector{ db: db, } @@ -31,7 +30,7 @@ func NewByJobId(ctx context.Context, db *repository.Queries, jobID uuid.UUID) (* func (c *Collector) getByJobID(ctx context.Context, jobID uuid.UUID) error { dbJobID := database.MustToDBUUID(jobID) - dbCollector, err := c.db.GetCollectorFromJobID(ctx, dbJobID) + dbCollector, err := c.db.Queries.GetCollectorFromJobID(ctx, dbJobID) if err != nil { return err } diff --git a/internal/collector/service.go b/internal/collector/service.go index 168872d1..814f5227 100644 --- a/internal/collector/service.go +++ b/internal/collector/service.go @@ -1,11 +1,15 @@ package collector -import "queryorchestration/internal/database/repository" +import ( + "queryorchestration/internal/database" +) type Service struct { - db *repository.Queries + db *database.Connection } -func New(db *repository.Queries) *Service { - return &Service{db} +func New(db *database.Connection) *Service { + return &Service{ + db, + } } diff --git a/internal/contextFull/creator.go b/internal/contextFull/creator.go new file mode 100644 index 00000000..67793db0 --- /dev/null +++ b/internal/contextFull/creator.go @@ -0,0 +1,21 @@ +package contextfull + +import ( + "context" + "queryorchestration/internal/database" + queryprocessor "queryorchestration/internal/queryProcessor" +) + +type Creator struct { + db *database.Connection +} + +func NewCreator(db *database.Connection) Creator { + return Creator{db} +} + +func (s Creator) Validate(ctx context.Context, entity *queryprocessor.Create) error { + // TODO + // Type, RequiredQueryIDs, Config + return nil +} diff --git a/internal/contextFull/result.go b/internal/contextFull/result.go index a69841e4..0ad30bf1 100644 --- a/internal/contextFull/result.go +++ b/internal/contextFull/result.go @@ -8,11 +8,11 @@ type Result struct { value string } -func NewResult(value string) *Result { - return &Result{value} +func NewResult(value string) Result { + return Result{value} } -func (r *Result) GetValue(ctx context.Context) (string, error) { +func (r Result) GetValue(ctx context.Context) (string, error) { // TODO - get value from s3 return r.value, nil } diff --git a/internal/contextFull/service.go b/internal/contextFull/service.go index 549ceb3e..39242b71 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.Query, values *[]result.Value) (string, error) { - if len(*values) > 0 { +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..ac24fc4a --- /dev/null +++ b/internal/contextFull/updator.go @@ -0,0 +1,21 @@ +package contextfull + +import ( + "context" + "queryorchestration/internal/database" + queryprocessor "queryorchestration/internal/queryProcessor" +) + +type Updator struct { + db *database.Connection +} + +func NewUpdator(db *database.Connection) Updator { + return Updator{db} +} + +func (s Updator) Validate(ctx context.Context, current *queryprocessor.Query, entity *queryprocessor.Update) error { + // TODO + // Type, RequiredQueryIDs, Config + return nil +} diff --git a/internal/database/parseuuid.go b/internal/database/parseuuid.go new file mode 100644 index 00000000..2a0ec948 --- /dev/null +++ b/internal/database/parseuuid.go @@ -0,0 +1,34 @@ +package database + +import ( + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" +) + +func MustToDBUUIDArray(ids []uuid.UUID) []pgtype.UUID { + dbIDs := make([]pgtype.UUID, len(ids)) + for index, id := range ids { + dbIDs[index] = MustToDBUUID(id) + } + + return dbIDs +} + +func MustToDBUUID(id uuid.UUID) pgtype.UUID { + var dbID pgtype.UUID + dbID.Scan(id.String()) + return dbID +} + +func MustToUUID(id pgtype.UUID) uuid.UUID { + return uuid.Must(uuid.FromBytes(id.Bytes[:])) +} + +func MustToUUIDArray(dbIDs []pgtype.UUID) []uuid.UUID { + ids := make([]uuid.UUID, len(dbIDs)) + for index, id := range dbIDs { + ids[index] = MustToUUID(id) + } + + return ids +} diff --git a/internal/database/pool.go b/internal/database/pool.go new file mode 100644 index 00000000..5324c664 --- /dev/null +++ b/internal/database/pool.go @@ -0,0 +1,17 @@ +package database + +import ( + "context" + "queryorchestration/internal/database/repository" + + "github.com/jackc/pgx/v5" +) + +type Pool interface { + Begin(ctx context.Context) (pgx.Tx, error) +} + +type Connection struct { + Pool Pool + Queries *repository.Queries +} 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 47592cdd..c3ab0ad7 100644 --- a/internal/database/repository/query.sql.go +++ b/internal/database/repository/query.sql.go @@ -11,6 +11,92 @@ import ( "github.com/jackc/pgx/v5/pgtype" ) +const createQuery = `-- name: CreateQuery :one +INSERT INTO queries (type) VALUES ($1) RETURNING id +` + +func (q *Queries) CreateQuery(ctx context.Context, type_ Querytype) (pgtype.UUID, error) { + row := q.db.QueryRow(ctx, createQuery, type_) + var id pgtype.UUID + err := row.Scan(&id) + return id, err +} + +const createQueryConfig = `-- name: CreateQueryConfig :exec +INSERT INTO queryConfigs (queryId, config, addedVersion) VALUES ($1, $2, $3) +` + +type CreateQueryConfigParams struct { + Queryid pgtype.UUID + Config []byte + Addedversion int32 +} + +func (q *Queries) CreateQueryConfig(ctx context.Context, arg CreateQueryConfigParams) error { + _, err := q.db.Exec(ctx, createQueryConfig, arg.Queryid, arg.Config, arg.Addedversion) + return err +} + +const createRequiredQuery = `-- name: CreateRequiredQuery :exec +INSERT INTO requiredQueries (queryId, requiredQueryId, addedVersion) VALUES ($1, $2, $3) +` + +type CreateRequiredQueryParams struct { + Queryid pgtype.UUID + Requiredqueryid pgtype.UUID + Addedversion int32 +} + +func (q *Queries) CreateRequiredQuery(ctx context.Context, arg CreateRequiredQueryParams) error { + _, err := q.db.Exec(ctx, createRequiredQuery, arg.Queryid, arg.Requiredqueryid, arg.Addedversion) + return err +} + +const deprecateQuery = `-- name: DeprecateQuery :exec +INSERT INTO queryDeprecations (queryId) VALUES ($1) +` + +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, q.latestVersion, c.config, ARRAY_AGG(DISTINCT r.requiredQueryId) as requiredIds + FROM queries AS q + LEFT JOIN queryConfigs AS c ON q.id = c.queryId + LEFT JOIN requiredQueries AS r ON q.id = r.queryId + WHERE q.id = $1 + and c.addedVersion >= q.activeVersion + and COALESCE(c.removedVersion, q.activeVersion - 1) < q.activeVersion + and r.addedVersion >= q.activeVersion + and COALESCE(r.removedVersion, q.activeVersion - 1) < q.activeVersion + GROUP BY q.id, c.config +` + +type GetQueryRow struct { + ID pgtype.UUID + Type Querytype + Activeversion int32 + Latestversion int32 + Config []byte + Requiredids []pgtype.UUID +} + +func (q *Queries) GetQuery(ctx context.Context, id pgtype.UUID) (GetQueryRow, error) { + row := q.db.QueryRow(ctx, getQuery, id) + var i GetQueryRow + err := row.Scan( + &i.ID, + &i.Type, + &i.Activeversion, + &i.Latestversion, + &i.Config, + &i.Requiredids, + ) + return i, err +} + const getQueryConfig = `-- name: GetQueryConfig :one SELECT id, config FROM queryConfigs where queryId = $1 and addedVersion >= $2 and COALESCE(removedVersion, $2 - 1) < $2 ` @@ -31,3 +117,64 @@ func (q *Queries) GetQueryConfig(ctx context.Context, arg GetQueryConfigParams) err := row.Scan(&i.ID, &i.Config) return i, 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 +} + +const listQueries = `-- name: ListQueries :many +SELECT q.id, q.type, q.activeVersion, q.latestVersion, c.config, ARRAY_AGG(r.requiredQueryId) AS requiredIds + FROM queries AS q + LEFT JOIN queryConfigs AS c ON q.id = c.queryId + LEFT JOIN requiredQueries AS r ON q.id = r.queryId + WHERE c.addedVersion >= q.activeVersion + and COALESCE(c.removedVersion, q.activeVersion - 1) < q.activeVersion + and r.addedVersion >= q.activeVersion + and COALESCE(r.removedVersion, q.activeVersion - 1) < q.activeVersion + GROUP BY q.id, c.config +` + +type ListQueriesRow struct { + ID pgtype.UUID + Type Querytype + Activeversion int32 + Latestversion int32 + Config []byte + Requiredids []pgtype.UUID +} + +func (q *Queries) ListQueries(ctx context.Context) ([]ListQueriesRow, error) { + rows, err := q.db.Query(ctx, listQueries) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListQueriesRow + for rows.Next() { + var i ListQueriesRow + if err := rows.Scan( + &i.ID, + &i.Type, + &i.Activeversion, + &i.Latestversion, + &i.Config, + &i.Requiredids, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} diff --git a/internal/database/uuid.go b/internal/database/uuid.go deleted file mode 100644 index 66966fc0..00000000 --- a/internal/database/uuid.go +++ /dev/null @@ -1,16 +0,0 @@ -package database - -import ( - "github.com/google/uuid" - "github.com/jackc/pgx/v5/pgtype" -) - -func MustToDBUUID(id uuid.UUID) pgtype.UUID { - var dbID pgtype.UUID - dbID.Scan(id.String()) - return dbID -} - -func MustToUUID(id pgtype.UUID) uuid.UUID { - return uuid.Must(uuid.FromBytes(id.Bytes[:])) -} diff --git a/internal/document/service.go b/internal/document/service.go index 4c8d974e..ba127ce1 100644 --- a/internal/document/service.go +++ b/internal/document/service.go @@ -20,12 +20,12 @@ type Document struct { } type Service struct { - db *repository.Queries + db *database.Connection } -func New(db *repository.Queries) *Service { +func New(db *database.Connection) *Service { return &Service{ - db: db, + db, } } @@ -53,10 +53,10 @@ func (s *Service) Sync(ctx context.Context, doc *Document) error { return nil } -func (s *Service) getResults(ctx context.Context, id uuid.UUID, coll *collector.Collector) (*[]result.Result, error) { +func (s *Service) getResults(ctx context.Context, id uuid.UUID, coll *collector.Collector) ([]*result.Result, error) { docID := database.MustToDBUUID(id) - results, err := s.db.ListResultsByDocumentID(ctx, repository.ListResultsByDocumentIDParams{ + results, err := s.db.Queries.ListResultsByDocumentID(ctx, repository.ListResultsByDocumentIDParams{ Documentid: docID, Textversion: coll.MinTextVersion, Cleanversion: coll.MinCleanVersion, @@ -65,10 +65,10 @@ func (s *Service) getResults(ctx context.Context, id uuid.UUID, coll *collector. return nil, err } - cleanResults := make([]result.Result, len(results)) + cleanResults := make([]*result.Result, len(results)) for index, dbResult := range results { - cleanResults[index] = *result.Parse(&dbResult) + cleanResults[index] = result.Parse(&dbResult) } - return &cleanResults, nil + return cleanResults, nil } diff --git a/internal/export/service.go b/internal/export/service.go index 0a7bf214..d5ae6a25 100644 --- a/internal/export/service.go +++ b/internal/export/service.go @@ -1,11 +1,13 @@ package export -import "queryorchestration/internal/database/repository" +import "queryorchestration/internal/database" type Service struct { - db *repository.Queries + db *database.Connection } -func New(db *repository.Queries) *Service { - return &Service{db} +func New(db *database.Connection) *Service { + return &Service{ + db, + } } diff --git a/internal/jsonExtractor/creator.go b/internal/jsonExtractor/creator.go new file mode 100644 index 00000000..8a328dde --- /dev/null +++ b/internal/jsonExtractor/creator.go @@ -0,0 +1,20 @@ +package jsonextractor + +import ( + "context" + "queryorchestration/internal/database" + queryprocessor "queryorchestration/internal/queryProcessor" +) + +type Creator struct { + db *database.Connection +} + +func NewCreator(db *database.Connection) Creator { + return Creator{db} +} + +func (s Creator) Validate(ctx context.Context, entity *queryprocessor.Create) error { + // TODO + return nil +} diff --git a/internal/jsonExtractor/result.go b/internal/jsonExtractor/result.go index 3ac237ec..7ac82ef0 100644 --- a/internal/jsonExtractor/result.go +++ b/internal/jsonExtractor/result.go @@ -8,10 +8,10 @@ type Result struct { value string } -func NewResult(value string) *Result { - return &Result{value} +func NewResult(value string) Result { + return Result{value} } -func (r *Result) GetValue(ctx context.Context) (string, error) { +func (r Result) GetValue(ctx context.Context) (string, error) { return r.value, nil } diff --git a/internal/jsonExtractor/service.go b/internal/jsonExtractor/service.go index 6b4e4ce2..ad42d707 100644 --- a/internal/jsonExtractor/service.go +++ b/internal/jsonExtractor/service.go @@ -6,35 +6,35 @@ import ( "fmt" "queryorchestration/internal/database" "queryorchestration/internal/database/repository" - "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" "queryorchestration/internal/result" "github.com/tidwall/gjson" ) type Extractor struct { - db *repository.Queries + db *database.Connection } type Config struct { Path string `json:"path"` } -func New(db *repository.Queries) *Extractor { - return &Extractor{db} +func NewExtractor(db *database.Connection) Extractor { + return Extractor{db} } -func (e *Extractor) Process(ctx context.Context, query query.Query, values *[]result.Value) (string, error) { - if len(*values) != 1 { +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") } - value, err := (*values)[0].GetValue(ctx) + value, err := values[0].GetValue(ctx) if err != nil { return "", err } - byteConfig, err := e.db.GetQueryConfig(ctx, repository.GetQueryConfigParams{ + byteConfig, err := e.db.Queries.GetQueryConfig(ctx, repository.GetQueryConfigParams{ Queryid: database.MustToDBUUID(query.ID), Addedversion: query.Version, }) diff --git a/internal/jsonExtractor/updator.go b/internal/jsonExtractor/updator.go new file mode 100644 index 00000000..10697050 --- /dev/null +++ b/internal/jsonExtractor/updator.go @@ -0,0 +1,20 @@ +package jsonextractor + +import ( + "context" + "queryorchestration/internal/database" + queryprocessor "queryorchestration/internal/queryProcessor" +) + +type Updator struct { + db *database.Connection +} + +func NewUpdator(db *database.Connection) Updator { + return Updator{db} +} + +func (s Updator) Validate(ctx context.Context, current *queryprocessor.Query, entity *queryprocessor.Update) error { + // TODO + return nil +} diff --git a/internal/query/create.go b/internal/query/create.go new file mode 100644 index 00000000..b7eb57ac --- /dev/null +++ b/internal/query/create.go @@ -0,0 +1,117 @@ +package query + +import ( + "context" + "fmt" + contextfull "queryorchestration/internal/contextFull" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + jsonextractor "queryorchestration/internal/jsonExtractor" + queryprocessor "queryorchestration/internal/queryProcessor" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" +) + +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 + } + + err = validator.Validate(ctx, entity) + if err != nil { + return uuid.Nil, err + } + + id, err := s.submitCreate(ctx, entity) + if err != nil { + return uuid.Nil, err + } + + return id, err +} + +func (s *Service) submitCreate(ctx context.Context, entity *queryprocessor.Create) (uuid.UUID, error) { + query, err := parseCreateQuery(entity) + if err != nil { + return uuid.Nil, err + } + + tx, err := s.db.Pool.Begin(ctx) + if err != nil { + return uuid.Nil, err + } + // defer tx.Rollback(ctx) + + qtx := s.db.Queries.WithTx(tx) + + dbID, err := qtx.CreateQuery(ctx, query.Type) + if err != nil { + return uuid.Nil, err + } + + for _, reqQuery := range query.RequiredQueryIDs { + err = qtx.CreateRequiredQuery(ctx, repository.CreateRequiredQueryParams{ + Queryid: dbID, + Requiredqueryid: reqQuery, + Addedversion: 1, + }) + if err != nil { + return uuid.Nil, err + } + } + + if query.Config != nil && string(query.Config) != "" { + err = qtx.CreateQueryConfig(ctx, repository.CreateQueryConfigParams{ + Queryid: dbID, + Config: query.Config, + Addedversion: 1, + }) + if err != nil { + return uuid.Nil, err + } + } + + err = tx.Commit(ctx) + if err != nil { + return uuid.Nil, err + } + + id := database.MustToUUID(dbID) + + 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") + } +} + +type createQuery struct { + Type repository.Querytype + RequiredQueryIDs []pgtype.UUID + Config []byte +} + +func parseCreateQuery(q *queryprocessor.Create) (*createQuery, error) { + t, err := queryprocessor.ToDBQueryType(q.Type) + if err != nil { + return nil, err + } + + reqIDs := database.MustToDBUUIDArray(q.RequiredQueryIDs) + + return &createQuery{ + Type: t, + RequiredQueryIDs: reqIDs, + Config: []byte(q.Config), + }, nil +} diff --git a/internal/query/database.go b/internal/query/database.go deleted file mode 100644 index 5c9a0ece..00000000 --- a/internal/query/database.go +++ /dev/null @@ -1,51 +0,0 @@ -package query - -import ( - "fmt" - "queryorchestration/internal/database" - "queryorchestration/internal/database/repository" -) - -func ParseDBQuery(dbQuery *repository.GetCollectorQueriesRow) (*Query, error) { - t, err := ParseDBType(dbQuery.Type) - if err != nil { - return nil, err - } - - return &Query{ - ID: database.MustToUUID(dbQuery.Queryid), - Type: t, - RequiredQueryID: database.MustToUUID(dbQuery.Requiredqueryid), - Version: dbQuery.Queryversion.Int32, - }, nil -} - -func ParseDBType(qType repository.NullQuerytype) (Type, error) { - if !qType.Valid { - return TypeJsonExtractor, fmt.Errorf("invalid database query type") - } - - switch qType.Querytype { - case repository.QuerytypeJsonExtractor: - return TypeJsonExtractor, nil - case repository.QuerytypeContextFull: - return TypeContextFull, nil - default: - return TypeJsonExtractor, fmt.Errorf("invalid database query type") - } -} - -func ToDBQueryType(t Type) (repository.NullQuerytype, error) { - var dbType repository.Querytype - - switch t { - case TypeJsonExtractor: - dbType = repository.QuerytypeJsonExtractor - case TypeContextFull: - dbType = repository.QuerytypeContextFull - default: - return repository.NullQuerytype{}, fmt.Errorf("invalid database query type") - } - - return repository.NullQuerytype{Querytype: dbType, Valid: true}, nil -} diff --git a/internal/query/deprecate.go b/internal/query/deprecate.go new file mode 100644 index 00000000..9fe06525 --- /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.Queries.IsQueryDeprecated(ctx, dbId) + if err != nil { + return err + } else if exists { + return nil + } + + err = s.db.Queries.DeprecateQuery(ctx, dbId) + if err != nil { + return err + } + + return nil +} diff --git a/internal/query/get.go b/internal/query/get.go new file mode 100644 index 00000000..eb38dcdb --- /dev/null +++ b/internal/query/get.go @@ -0,0 +1,46 @@ +package query + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + queryprocessor "queryorchestration/internal/queryProcessor" + + "github.com/google/uuid" +) + +type Query struct { + ID uuid.UUID + Type queryprocessor.Type + ActiveVersion int32 + LatestVersion int32 + RequiredQueryIDs []uuid.UUID + Config string +} + +func (s *Service) Get(ctx context.Context, id uuid.UUID) (*Query, error) { + query, err := s.db.Queries.GetQuery(ctx, database.MustToDBUUID(id)) + if err != nil { + return nil, err + } + + return ParseDBGetQuery(&query) +} + +func ParseDBGetQuery(q *repository.GetQueryRow) (*Query, error) { + reqQueryIDs := database.MustToUUIDArray(q.Requiredids) + + qType, err := queryprocessor.ParseDBType(q.Type) + if err != nil { + return nil, err + } + + return &Query{ + ID: database.MustToUUID(q.ID), + ActiveVersion: q.Activeversion, + LatestVersion: q.Latestversion, + Type: qType, + RequiredQueryIDs: reqQueryIDs, + Config: string(q.Config), + }, nil +} diff --git a/internal/query/list.go b/internal/query/list.go new file mode 100644 index 00000000..7bffbd30 --- /dev/null +++ b/internal/query/list.go @@ -0,0 +1,51 @@ +package query + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + queryprocessor "queryorchestration/internal/queryProcessor" +) + +type ListFilters struct { + Types []queryprocessor.Type +} + +func (s *Service) List(ctx context.Context, filters ListFilters) ([]*Query, error) { + // TODO - use filters + + dbQueries, err := s.db.Queries.ListQueries(ctx) + if err != nil { + return nil, err + } + + queries := make([]*Query, len(dbQueries)) + for index, query := range dbQueries { + q, err := ParseDBListQuery(&query) + if err != nil { + return nil, err + } + + queries[index] = q + } + + return queries, nil +} + +func ParseDBListQuery(q *repository.ListQueriesRow) (*Query, error) { + reqQueryIDs := database.MustToUUIDArray(q.Requiredids) + + qType, err := queryprocessor.ParseDBType(q.Type) + if err != nil { + return nil, err + } + + return &Query{ + ID: database.MustToUUID(q.ID), + ActiveVersion: q.Activeversion, + LatestVersion: q.Latestversion, + Type: qType, + RequiredQueryIDs: reqQueryIDs, + Config: string(q.Config), + }, nil +} diff --git a/internal/query/service.go b/internal/query/service.go index 796ed47a..75dca704 100644 --- a/internal/query/service.go +++ b/internal/query/service.go @@ -1,35 +1,13 @@ package query -import ( - "context" - "queryorchestration/internal/database/repository" - "queryorchestration/internal/result" - - "github.com/google/uuid" -) - -type Type int - -const ( - TypeJsonExtractor = iota - TypeContextFull -) - -type Query struct { - ID uuid.UUID - Type Type - RequiredQueryID uuid.UUID - Version int32 -} - -type Processor interface { - Process(ctx context.Context, query Query, values *[]result.Value) (string, error) -} +import "queryorchestration/internal/database" type Service struct { - db *repository.Queries + db *database.Connection } -func New(db *repository.Queries) *Service { - return &Service{db} +func New(db *database.Connection) *Service { + return &Service{ + db, + } } diff --git a/internal/query/test.go b/internal/query/test.go new file mode 100644 index 00000000..5eed7887 --- /dev/null +++ b/internal/query/test.go @@ -0,0 +1,21 @@ +package query + +import ( + "context" + + "github.com/google/uuid" +) + +type Test struct { + QueryID uuid.UUID + DocumentID uuid.UUID + QueryVersion int32 +} + +func (s *Service) Test(ctx context.Context, filters Test) (string, error) { + // TODO + // Sync doc - documentID + // Run test - queryID, queryVersion + + return "", nil +} diff --git a/internal/query/update.go b/internal/query/update.go new file mode 100644 index 00000000..e8c88634 --- /dev/null +++ b/internal/query/update.go @@ -0,0 +1,60 @@ +package query + +import ( + "context" + "fmt" + contextfull "queryorchestration/internal/contextFull" + jsonextractor "queryorchestration/internal/jsonExtractor" + queryprocessor "queryorchestration/internal/queryProcessor" +) + +func (s *Service) Update(ctx context.Context, entity *queryprocessor.Update) error { + current, err := s.Get(ctx, entity.ID) + if err != nil { + return err + } + + validator, err := s.getUpdator(current.Type) + if err != nil { + return err + } + + err = validator.Validate(ctx, ParseQuery(current), entity) + if err != nil { + return err + } + + err = s.submitUpdate(ctx, entity) + if err != nil { + return err + } + + return nil +} + +func (s *Service) submitUpdate(ctx context.Context, entity *queryprocessor.Update) error { + // TODO - generate new entity + // TODO - submit update - id, type, activeversion, requiredQueryId, Config + 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") + } +} + +func ParseQuery(q *Query) *queryprocessor.Query { + return &queryprocessor.Query{ + ID: q.ID, + Type: q.Type, + Version: q.ActiveVersion, + RequiredQueryIDs: q.RequiredQueryIDs, + Config: q.Config, + } +} diff --git a/internal/queryProcessor/parse.go b/internal/queryProcessor/parse.go new file mode 100644 index 00000000..8304c92b --- /dev/null +++ b/internal/queryProcessor/parse.go @@ -0,0 +1,66 @@ +package queryprocessor + +import ( + "fmt" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" +) + +func ParseDBNullType(qType repository.NullQuerytype) (Type, error) { + if !qType.Valid { + return TypeJsonExtractor, fmt.Errorf("invalid database query type") + } + + return ParseDBType(qType.Querytype) +} + +func ParseDBType(qType repository.Querytype) (Type, error) { + switch qType { + case repository.QuerytypeJsonExtractor: + return TypeJsonExtractor, nil + case repository.QuerytypeContextFull: + return TypeContextFull, nil + default: + return TypeJsonExtractor, fmt.Errorf("invalid database query type") + } +} + +func ToDBQueryType(t Type) (repository.Querytype, error) { + var dbType repository.Querytype + + switch t { + case TypeJsonExtractor: + dbType = repository.QuerytypeJsonExtractor + case TypeContextFull: + dbType = repository.QuerytypeContextFull + default: + return repository.QuerytypeContextFull, fmt.Errorf("invalid database query type") + } + + return dbType, nil +} + +func ToDBNullQueryType(t Type) (repository.NullQuerytype, error) { + dbType, err := ToDBQueryType(t) + if err != nil { + return repository.NullQuerytype{}, err + } + + return repository.NullQuerytype{Querytype: dbType, Valid: true}, nil +} + +func ParseDBCollectorQuery(q *repository.GetCollectorQueriesRow) (*Query, error) { + reqQueryIDs := database.MustToUUIDArray(q.Requiredids) + + 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..7f502b96 --- /dev/null +++ b/internal/queryProcessor/service.go @@ -0,0 +1,47 @@ +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 string +} + +type Update struct { + ID uuid.UUID + RequiredQueryIDs []uuid.UUID + Config string +} + +type Query struct { + ID uuid.UUID + Type Type + Version int32 + RequiredQueryIDs []uuid.UUID + Config string +} + +type Creator interface { + Validate(ctx context.Context, entity *Create) error +} + +type Updator interface { + Validate(ctx context.Context, current *Query, 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 498856f6..4aec4db8 100644 --- a/internal/queryQueue/create.go +++ b/internal/queryQueue/create.go @@ -3,13 +3,13 @@ package queryQueue import ( "context" "queryorchestration/internal/database" - "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" ) func (q *Queue) getUnsyncedQueries() { - for _, query := range *q.collectorQueries { + for _, query := range q.collectorQueries { isSynced := false - for _, result := range *q.results { + for _, result := range q.results { if result.QueryID != query.ID || result.QueryVersion != query.Version { continue } @@ -20,7 +20,7 @@ func (q *Queue) getUnsyncedQueries() { continue } - q.Add(&query) + q.Add(query) } } @@ -31,56 +31,62 @@ func (c *Queue) getCollectorQueries(ctx context.Context) error { id := database.MustToDBUUID(c.collector.ID) - queries, err := c.db.GetCollectorQueries(ctx, id) + queries, err := c.db.Queries.GetCollectorQueries(ctx, id) if err != nil { return err } - cleanQueries := make([]query.Query, len(queries)) + cleanQueries := make([]*queryprocessor.Query, len(queries)) for index, dbQuery := range queries { - cleanQuery, err := query.ParseDBQuery(&dbQuery) + cleanQuery, err := queryprocessor.ParseDBCollectorQuery(&dbQuery) if err != nil { return err } - cleanQueries[index] = *cleanQuery + cleanQueries[index] = cleanQuery } - c.collectorQueries = &cleanQueries + c.collectorQueries = cleanQueries return nil } -func (q *Queue) Add(qu *query.Query) { - dependentQueries := []query.Query{} +func (q *Queue) Add(qu *queryprocessor.Query) { + dependentQueries := []*queryprocessor.Query{} requiredIndex := -1 if q.unsyncedQueue == nil { - q.unsyncedQueue = &[]query.Query{} + q.unsyncedQueue = []*queryprocessor.Query{} } else { - for index, entry := range *q.unsyncedQueue { + 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 _, entry := range q.collectorQueries { + 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.Query{*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.Query{*qu}, *q.unsyncedQueue...) + q.unsyncedQueue = append([]*queryprocessor.Query{qu}, q.unsyncedQueue...) } for _, entry := range dependentQueries { - q.Add(&entry) + q.Add(entry) } } diff --git a/internal/queryQueue/execute.go b/internal/queryQueue/execute.go index 080bc6b1..ddd58bb9 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" ) @@ -19,7 +18,7 @@ func (q *Queue) Execute(ctx context.Context) error { return nil } - for _, query := range *q.unsyncedQueue { + for _, query := range q.unsyncedQueue { err := q.executeQuery(ctx, query) if err != nil { return err @@ -31,25 +30,18 @@ func (q *Queue) Execute(ctx context.Context) error { return nil } -func (q *Queue) executeQuery(ctx context.Context, qu query.Query) 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 { + for _, entry := range q.collectorQueries { if entry.ID == id { queryVersion = entry.Version break } } - for _, entry := range *q.results { + for _, entry := range q.results { if entry.QueryID == id && entry.QueryVersion == queryVersion { resultIDs[index] = database.MustToDBUUID(entry.ID) break @@ -57,7 +49,7 @@ func (q *Queue) executeQuery(ctx context.Context, qu query.Query) error { } } - values, err := q.db.ListResultValuesByID(ctx, resultIDs) + values, err := q.db.Queries.ListResultValuesByID(ctx, resultIDs) if err != nil { return err } @@ -72,7 +64,7 @@ func (q *Queue) executeQuery(ctx context.Context, qu query.Query) error { cleanValues[index] = cleanValue } - err = q.setResult(ctx, qu, &cleanValues) + err = q.setResult(ctx, qu, cleanValues) if err != nil { return err } @@ -81,17 +73,17 @@ func (q *Queue) executeQuery(ctx context.Context, qu query.Query) error { } func (q *Queue) getResultValue(res *repository.ListResultValuesByIDRow) (result.Value, error) { - var queryType query.Type - for _, qu := range *q.collectorQueries { + var queryType queryprocessor.Type + for _, qu := range q.collectorQueries { if qu.ID == database.MustToUUID(res.Queryid) { queryType = qu.Type } } 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 bb776509..633770df 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.Query, 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 @@ -20,7 +20,7 @@ func (q *Queue) setResult(ctx context.Context, qu query.Query, resultValues *[]r return err } - id, err := result.Store(ctx, q.db, &result.ResultStore{ + id, err := result.Store(ctx, q.db.Queries, &result.ResultStore{ QueryID: qu.ID, DocumentID: q.documentId, Value: value, @@ -32,7 +32,7 @@ func (q *Queue) setResult(ctx context.Context, qu query.Query, resultValues *[]r return err } - *q.results = append(*q.results, result.Result{ + q.results = append(q.results, &result.Result{ ID: id, QueryID: qu.ID, QueryVersion: qu.Version, @@ -41,12 +41,12 @@ func (q *Queue) setResult(ctx context.Context, qu query.Query, resultValues *[]r 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 f83fc39d..581b4e53 100644 --- a/internal/queryQueue/service.go +++ b/internal/queryQueue/service.go @@ -3,25 +3,25 @@ package queryQueue import ( "context" "queryorchestration/internal/collector" - "queryorchestration/internal/database/repository" - "queryorchestration/internal/query" + "queryorchestration/internal/database" + queryprocessor "queryorchestration/internal/queryProcessor" "queryorchestration/internal/result" "github.com/google/uuid" ) type Queue struct { - unsyncedQueue *[]query.Query - collectorQueries *[]query.Query - results *[]result.Result + unsyncedQueue []*queryprocessor.Query + collectorQueries []*queryprocessor.Query + results []*result.Result collector *collector.Collector - db *repository.Queries + db *database.Connection cleanVersion int32 textVersion int32 documentId uuid.UUID } -func New(ctx context.Context, db *repository.Queries, coll *collector.Collector, results *[]result.Result, docId uuid.UUID, cleanVersion int32, textVersion int32) (*Queue, error) { +func New(ctx context.Context, db *database.Connection, coll *collector.Collector, results []*result.Result, docId uuid.UUID, cleanVersion int32, textVersion int32) (*Queue, error) { queue := Queue{ db: db, results: results, @@ -41,6 +41,6 @@ func New(ctx context.Context, db *repository.Queries, coll *collector.Collector, return &queue, nil } -func (q *Queue) GetQueue() []query.Query { - return *q.unsyncedQueue +func (q *Queue) GetQueue() []*queryprocessor.Query { + return q.unsyncedQueue } diff --git a/internal/result/service.go b/internal/result/service.go index 75c67fc6..66bde823 100644 --- a/internal/result/service.go +++ b/internal/result/service.go @@ -27,8 +27,8 @@ type ResultStore struct { QueryVersion int32 } -func Store(ctx context.Context, db *repository.Queries, res *ResultStore) (uuid.UUID, error) { - dbId, err := db.SetResult(ctx, repository.SetResultParams{ +func Store(ctx context.Context, dbQueries *repository.Queries, res *ResultStore) (uuid.UUID, error) { + dbId, err := dbQueries.SetResult(ctx, repository.SetResultParams{ Queryid: database.MustToDBUUID(res.QueryID), Documentid: database.MustToDBUUID(res.DocumentID), Value: res.Value, diff --git a/scripts/proto.yml b/scripts/proto.yml index 19086050..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 \ No newline at end of file + - 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/integration/queryrunner_test.go b/test/integration/queryrunner_test.go index d01e511e..2337757f 100644 --- a/test/integration/queryrunner_test.go +++ b/test/integration/queryrunner_test.go @@ -10,7 +10,7 @@ import ( "github.com/stretchr/testify/assert" ) -func TestName(t *testing.T) { +func TestQueryRunner(t *testing.T) { ctx := context.Background() queue, cleanup := createQueueDependencies(t, ctx, "queryrunner") @@ -19,7 +19,7 @@ func TestName(t *testing.T) { document := document.Document{ ID: uuid.New(), JobID: uuid.New(), - Name: "document_name", + Name: "documentname", CleanVersion: int32(1), TextVersion: int32(1), } diff --git a/test/integration/queryservice_test.go b/test/integration/queryservice_test.go index d6f57b7e..4b2f0808 100644 --- a/test/integration/queryservice_test.go +++ b/test/integration/queryservice_test.go @@ -8,7 +8,7 @@ import ( "github.com/stretchr/testify/assert" ) -func TestQuery(t *testing.T) { +func TestQueryService(t *testing.T) { ctx := context.Background() conn, cleanup := createAPIDependencies(t, ctx, "queryservice") @@ -16,11 +16,32 @@ func TestQuery(t *testing.T) { client := serviceinterfaces.NewQueryServiceClient(conn) - id := "sample_id" - - _, err := client.Get(ctx, &serviceinterfaces.IdMessage{ - Id: id, + config := "" + reqQueries := []string{} + idRes, err := client.Create(ctx, &serviceinterfaces.QueryCreate{ + Type: 1, + Config: &config, + RequiredQueries: reqQueries, }) - assert.Nil(t, err) + id := idRes.GetId() + assert.NotEmpty(t, id) + + // queryRes, err := client.Get(ctx, &serviceinterfaces.IdMessage{ + // Id: id, + // }) + // log.Print(err) + // assert.Nil(t, err) + // assert.EqualExportedValues(t, serviceinterfaces.Query{ + // Id: id, + // Type: serviceinterfaces.QueryType_QUERY_TYPE_CONTEXT_FULL, + // ActiveVersion: int32(1), + // LatestVersion: int32(1), + // Config: &config, + // RequiredQueries: reqQueries, + // }, *queryRes) + + //QueryService - List, Update, Test + //JobController - Create, Update, Get + //Export - Trigger } diff --git a/test/unit/internal/collector/service_test.go b/test/unit/internal/collector/service_test.go index e0789dfe..8d307cda 100644 --- a/test/unit/internal/collector/service_test.go +++ b/test/unit/internal/collector/service_test.go @@ -13,29 +13,32 @@ import ( ) func TestService(t *testing.T) { - ctx := context.Background() - - db, err := pgxmock.NewConn() + pool, err := pgxmock.NewPool() if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - defer db.Close(ctx) + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } - queries := repository.New(db) - svc := collector.New(queries) + svc := collector.New(db) assert.NotNil(t, svc) } func TestByJobId(t *testing.T) { ctx := context.Background() - db, err := pgxmock.NewConn() + pool, err := pgxmock.NewPool() if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - defer db.Close(ctx) - - queries := repository.New(db) + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } jobID := uuid.New() fullCollector := collector.Collector{ @@ -44,21 +47,21 @@ func TestByJobId(t *testing.T) { MinTextVersion: int32(1), } - db.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(database.MustToDBUUID(jobID)). + pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(database.MustToDBUUID(jobID)). WillReturnRows( pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}). AddRow(database.MustToDBUUID(fullCollector.ID), database.MustToDBUUID(jobID), fullCollector.MinCleanVersion, fullCollector.MinTextVersion), ) - coll, err := collector.NewByJobId(ctx, queries, jobID) + coll, err := collector.NewByJobId(ctx, db, jobID) assert.Nil(t, err) assert.EqualExportedValues(t, fullCollector, *coll) - db.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(database.MustToDBUUID(jobID)). + pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(database.MustToDBUUID(jobID)). WillReturnRows( pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}), ) - _, err = collector.NewByJobId(ctx, queries, jobID) + _, err = collector.NewByJobId(ctx, db, jobID) assert.EqualError(t, err, "no rows in result set") } diff --git a/test/unit/internal/contextfull/creator_test.go b/test/unit/internal/contextfull/creator_test.go new file mode 100644 index 00000000..e09cf28f --- /dev/null +++ b/test/unit/internal/contextfull/creator_test.go @@ -0,0 +1,40 @@ +package document_test + +import ( + "context" + contextfull "queryorchestration/internal/contextFull" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + queryprocessor "queryorchestration/internal/queryProcessor" + "testing" + + "github.com/google/uuid" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +func TestCreatorValidate(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 := contextfull.NewCreator(db) + assert.NotNil(t, svc) + + entity := &queryprocessor.Create{ + Type: queryprocessor.TypeContextFull, + RequiredQueryIDs: []uuid.UUID{}, + Config: "", + } + + err = svc.Validate(ctx, entity) + assert.Nil(t, err) +} diff --git a/test/unit/internal/contextfull/process_test.go b/test/unit/internal/contextfull/process_test.go index c8c9eab6..0683beff 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,22 +14,22 @@ import ( func TestContextFull(t *testing.T) { ctx := context.Background() - extractor := contextfull.New() + extractor := contextfull.NewExtractor() - query := query.Query{ - 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{} + values := []result.Value{} value, err := extractor.Process(ctx, query, values) assert.Nil(t, err) assert.Equal(t, "", value) - values = &[]result.Value{ + values = []result.Value{ contextfull.NewResult("example_result"), } diff --git a/test/unit/internal/contextfull/updator_test.go b/test/unit/internal/contextfull/updator_test.go new file mode 100644 index 00000000..aab5e241 --- /dev/null +++ b/test/unit/internal/contextfull/updator_test.go @@ -0,0 +1,48 @@ +package document_test + +import ( + "context" + contextfull "queryorchestration/internal/contextFull" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + queryprocessor "queryorchestration/internal/queryProcessor" + "testing" + + "github.com/google/uuid" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +func TestUpdatorValidate(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 := contextfull.NewUpdator(db) + assert.NotNil(t, svc) + + current := &queryprocessor.Query{ + ID: uuid.New(), + Type: queryprocessor.TypeContextFull, + Version: int32(1), + RequiredQueryIDs: []uuid.UUID{}, + Config: "", + } + + entity := &queryprocessor.Update{ + ID: current.ID, + RequiredQueryIDs: []uuid.UUID{}, + Config: "", + } + + err = svc.Validate(ctx, current, entity) + assert.Nil(t, err) +} diff --git a/test/unit/internal/database/parseuuid_test.go b/test/unit/internal/database/parseuuid_test.go new file mode 100644 index 00000000..14b098ef --- /dev/null +++ b/test/unit/internal/database/parseuuid_test.go @@ -0,0 +1,49 @@ +package database_test + +import ( + "queryorchestration/internal/database" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + "github.com/stretchr/testify/assert" +) + +func TestMustToDBUUID(t *testing.T) { + id := uuid.New() + + dbID := database.MustToDBUUID(id) + + assert.Equal(t, true, dbID.Valid) + assert.Equal(t, id.String(), uuid.UUID(dbID.Bytes).String()) +} + +func TestMustToDBUUIDArray(t *testing.T) { + ids := []uuid.UUID{uuid.New(), uuid.New()} + + dbIDs := database.MustToDBUUIDArray(ids) + + assert.Equal(t, len(ids), len(dbIDs)) + for index, id := range dbIDs { + assert.Equal(t, database.MustToDBUUID(ids[index]), id) + } +} + +func TestMustToUUID(t *testing.T) { + dbID := database.MustToDBUUID(uuid.New()) + + id := database.MustToUUID(dbID) + + assert.Equal(t, id.String(), uuid.UUID(dbID.Bytes).String()) +} + +func TestMustToUUIDArray(t *testing.T) { + dbIDs := []pgtype.UUID{database.MustToDBUUID(uuid.New()), database.MustToDBUUID(uuid.New())} + + ids := database.MustToUUIDArray(dbIDs) + + assert.Equal(t, len(ids), len(dbIDs)) + for index, id := range dbIDs { + assert.Equal(t, database.MustToDBUUID(ids[index]), id) + } +} diff --git a/test/unit/internal/database/repository/querytype_test.go b/test/unit/internal/database/repository/querytype_test.go new file mode 100644 index 00000000..81b77b17 --- /dev/null +++ b/test/unit/internal/database/repository/querytype_test.go @@ -0,0 +1,37 @@ +package database_test + +import ( + "fmt" + "queryorchestration/internal/database/repository" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestQueryTypeScan(t *testing.T) { + qType := repository.Querytype(fmt.Sprint(0)) + + stringType := "context_full" + err := qType.Scan(stringType) + assert.Nil(t, err) +} + +func TestNullQueryTypeScan(t *testing.T) { + qType := repository.NullQuerytype{} + + stringType := "context_full" + err := qType.Scan(stringType) + assert.Nil(t, err) +} + +func TestNullQueryTypeValue(t *testing.T) { + qType := repository.NullQuerytype{} + + stringType := "context_full" + err := qType.Scan(stringType) + assert.Nil(t, err) + + val, err := qType.Value() + assert.Nil(t, err) + assert.Equal(t, stringType, val) +} diff --git a/test/unit/internal/database/uuid_test.go b/test/unit/internal/database/uuid_test.go deleted file mode 100644 index 092a2727..00000000 --- a/test/unit/internal/database/uuid_test.go +++ /dev/null @@ -1,18 +0,0 @@ -package database_test - -import ( - "queryorchestration/internal/database" - "testing" - - "github.com/google/uuid" - "github.com/stretchr/testify/assert" -) - -func TestMustToDBUUID(t *testing.T) { - id := uuid.New() - - dbID := database.MustToDBUUID(id) - - assert.Equal(t, true, dbID.Valid) - assert.Equal(t, id.String(), uuid.UUID(dbID.Bytes).String()) -} diff --git a/test/unit/internal/document/sync_test.go b/test/unit/internal/document/sync_test.go index ef776a7e..298f5ea2 100644 --- a/test/unit/internal/document/sync_test.go +++ b/test/unit/internal/document/sync_test.go @@ -17,13 +17,15 @@ import ( func TestSyncIsSynced(t *testing.T) { ctx := context.Background() - db, err := pgxmock.NewConn() + pool, err := pgxmock.NewPool() if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - defer db.Close(ctx) - - queries := repository.New(db) + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } doc := document.Document{ ID: uuid.New(), @@ -36,23 +38,23 @@ func TestSyncIsSynced(t *testing.T) { minCleanVersion := int32(1) minTextVersion := int32(1) - db.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID). + pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID). WillReturnRows( pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}). AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion), ) - db.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), minCleanVersion, minTextVersion). + pool.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), minCleanVersion, minTextVersion). WillReturnRows( pgxmock.NewRows([]string{"id", "queryId", "queryVersion"}). AddRow(pgtype.UUID{}, pgtype.UUID{}, int32(1)), ) - db.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorId). + pool.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) + docSvc := document.New(db) err = docSvc.Sync(ctx, &doc) assert.Nil(t, err) } @@ -60,13 +62,15 @@ func TestSyncIsSynced(t *testing.T) { func TestSyncDBFail(t *testing.T) { ctx := context.Background() - db, err := pgxmock.NewConn() + pool, err := pgxmock.NewPool() if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - defer db.Close(ctx) - - queries := repository.New(db) + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } doc := document.Document{ ID: uuid.New(), @@ -80,64 +84,64 @@ func TestSyncDBFail(t *testing.T) { minCleanVersion := int32(1) minTextVersion := int32(1) - db.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID). + pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID). WillReturnRows( pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}), ) - docSvc := document.New(queries) + docSvc := document.New(db) err = docSvc.Sync(ctx, &doc) assert.EqualError(t, err, "no rows in result set") - db.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID). + pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID). WillReturnRows( pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}). AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion), ) errr := "database failure" - db.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), minCleanVersion, minTextVersion). + pool.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), minCleanVersion, minTextVersion). WillReturnError(errors.New(errr)) - docSvc = document.New(queries) + docSvc = document.New(db) err = docSvc.Sync(ctx, &doc) assert.EqualError(t, err, errr) - db.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID). + pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID). WillReturnRows( pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}). AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion), ) - db.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), minCleanVersion, minTextVersion). + pool.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), minCleanVersion, minTextVersion). WillReturnRows( pgxmock.NewRows([]string{"id", "queryId", "queryVersion"}), ) errr = "database failure" - db.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorId). + pool.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorId). WillReturnError(errors.New(errr)) - docSvc = document.New(queries) + docSvc = document.New(db) err = docSvc.Sync(ctx, &doc) assert.EqualError(t, err, errr) - db.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID). + pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID). WillReturnRows( pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}). AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion), ) - db.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), minCleanVersion, minTextVersion). + pool.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), minCleanVersion, minTextVersion). WillReturnRows( pgxmock.NewRows([]string{"id", "queryId", "queryVersion"}), ) errr = "database failure" - db.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorId). + pool.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{}). + pool.ExpectQuery("name: ListResultValuesByID :many").WithArgs([]pgtype.UUID{}). WillReturnError(errors.New(errr)) - docSvc = document.New(queries) + docSvc = document.New(db) err = docSvc.Sync(ctx, &doc) assert.EqualError(t, err, errr) } @@ -145,13 +149,15 @@ func TestSyncDBFail(t *testing.T) { func TestSync(t *testing.T) { ctx := context.Background() - db, err := pgxmock.NewConn() + pool, err := pgxmock.NewPool() if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - defer db.Close(ctx) - - queries := repository.New(db) + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } doc := document.Document{ ID: uuid.New(), @@ -165,26 +171,26 @@ func TestSync(t *testing.T) { minCleanVersion := int32(1) minTextVersion := int32(1) - db.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID). + pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID). WillReturnRows( pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}). AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion), ) - db.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), minCleanVersion, minTextVersion). + pool.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), minCleanVersion, minTextVersion). WillReturnRows( pgxmock.NewRows([]string{"id", "queryId", "queryVersion"}), ) - db.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorId). + pool.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{}). + pool.ExpectQuery("name: ListResultValuesByID :many").WithArgs([]pgtype.UUID{}). WillReturnRows( pgxmock.NewRows([]string{"id", "queryId", "value"}), ) - docSvc := document.New(queries) + docSvc := document.New(db) err = docSvc.Sync(ctx, &doc) assert.EqualError(t, err, "JSON Extraction requires 1 result") } diff --git a/test/unit/internal/jsonextractor/creator_test.go b/test/unit/internal/jsonextractor/creator_test.go new file mode 100644 index 00000000..59fedf5a --- /dev/null +++ b/test/unit/internal/jsonextractor/creator_test.go @@ -0,0 +1,40 @@ +package document_test + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + jsonextractor "queryorchestration/internal/jsonExtractor" + queryprocessor "queryorchestration/internal/queryProcessor" + "testing" + + "github.com/google/uuid" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +func TestCreatorValidate(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 := jsonextractor.NewCreator(db) + assert.NotNil(t, svc) + + entity := &queryprocessor.Create{ + Type: queryprocessor.TypeJsonExtractor, + RequiredQueryIDs: []uuid.UUID{}, + Config: "", + } + + err = svc.Validate(ctx, entity) + assert.Nil(t, err) +} diff --git a/test/unit/internal/jsonextractor/process_test.go b/test/unit/internal/jsonextractor/process_test.go index 9fea1a61..eb27bc0f 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" @@ -20,32 +20,34 @@ import ( func TestJSONProcess(t *testing.T) { ctx := context.Background() - db, err := pgxmock.NewConn() + pool, err := pgxmock.NewPool() if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - defer db.Close(ctx) + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } - queries := repository.New(db) + extractor := jsonextractor.NewExtractor(db) - extractor := jsonextractor.New(queries) - - query := query.Query{ - 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" jsonString := fmt.Sprintf("{\"key\": \"%s\"}", entryValue) - values := &[]result.Value{ + values := []result.Value{ contextfull.NewResult(jsonString), } config := "{\"path\":\"key\"}" - db.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(query.ID), query.Version). + pool.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(query.ID), query.Version). WillReturnRows( pgxmock.NewRows([]string{"id", "config"}). AddRow(pgtype.UUID{}, []byte(config)), @@ -57,10 +59,10 @@ func TestJSONProcess(t *testing.T) { entryValue = "" jsonString = fmt.Sprintf("{\"key\": \"%s\"}", entryValue) - values = &[]result.Value{ + values = []result.Value{ contextfull.NewResult(jsonString), } - db.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(query.ID), query.Version). + pool.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(query.ID), query.Version). WillReturnRows( pgxmock.NewRows([]string{"id", "config"}). AddRow(pgtype.UUID{}, []byte(config)), @@ -72,10 +74,10 @@ func TestJSONProcess(t *testing.T) { entryValue = "1" jsonString = fmt.Sprintf("{\"key\": %s", entryValue) - values = &[]result.Value{ + values = []result.Value{ contextfull.NewResult(jsonString), } - db.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(query.ID), query.Version). + pool.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(query.ID), query.Version). WillReturnRows( pgxmock.NewRows([]string{"id", "config"}). AddRow(pgtype.UUID{}, []byte(config)), @@ -89,32 +91,34 @@ func TestJSONProcess(t *testing.T) { func TestJSONProcessJSON(t *testing.T) { ctx := context.Background() - db, err := pgxmock.NewConn() + pool, err := pgxmock.NewPool() if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - defer db.Close(ctx) + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } - queries := repository.New(db) + extractor := jsonextractor.NewExtractor(db) - extractor := jsonextractor.New(queries) - - query := query.Query{ - 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" jsonString := fmt.Sprintf("{\"key\": \"%s\"}", entryValue) - values := &[]result.Value{ + values := []result.Value{ contextfull.NewResult(jsonString), } config := "{\"path\":\"invalid_key\"}" - db.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(query.ID), query.Version). + pool.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(query.ID), query.Version). WillReturnRows( pgxmock.NewRows([]string{"id", "config"}). AddRow(pgtype.UUID{}, []byte(config)), @@ -126,7 +130,7 @@ func TestJSONProcessJSON(t *testing.T) { config = "" - db.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(query.ID), query.Version). + pool.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(query.ID), query.Version). WillReturnRows( pgxmock.NewRows([]string{"id", "config"}). AddRow(pgtype.UUID{}, []byte(config)), @@ -138,7 +142,7 @@ func TestJSONProcessJSON(t *testing.T) { config = "{\"path\":\"\"}" - db.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(query.ID), query.Version). + pool.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(query.ID), query.Version). WillReturnRows( pgxmock.NewRows([]string{"id", "config"}). AddRow(pgtype.UUID{}, []byte(config)), @@ -150,7 +154,7 @@ func TestJSONProcessJSON(t *testing.T) { config = "{\"path\":}" - db.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(query.ID), query.Version). + pool.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(query.ID), query.Version). WillReturnRows( pgxmock.NewRows([]string{"id", "config"}). AddRow(pgtype.UUID{}, []byte(config)), @@ -162,7 +166,7 @@ func TestJSONProcessJSON(t *testing.T) { config = "{\"path\":\"key\"}" - db.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(query.ID), query.Version). + pool.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(query.ID), query.Version). WillReturnRows( pgxmock.NewRows([]string{"id", "config"}), ) @@ -175,29 +179,31 @@ func TestJSONProcessJSON(t *testing.T) { func TestJSONProcessResults(t *testing.T) { ctx := context.Background() - db, err := pgxmock.NewConn() + pool, err := pgxmock.NewPool() if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - defer db.Close(ctx) - - queries := repository.New(db) - - extractor := jsonextractor.New(queries) - - query := query.Query{ - ID: uuid.New(), - Type: query.TypeJsonExtractor, - RequiredQueryID: uuid.Nil, - Version: int32(1), + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, } - results := &[]result.Value{} + extractor := jsonextractor.NewExtractor(db) + + query := &queryprocessor.Query{ + ID: uuid.New(), + Type: queryprocessor.TypeJsonExtractor, + RequiredQueryIDs: []uuid.UUID{}, + Version: int32(1), + } + + results := []result.Value{} value, err := extractor.Process(ctx, query, results) assert.EqualError(t, err, "JSON Extraction requires 1 result") assert.Empty(t, value) - results = &[]result.Value{ + results = []result.Value{ contextfull.NewResult(""), contextfull.NewResult(""), } @@ -205,7 +211,7 @@ func TestJSONProcessResults(t *testing.T) { assert.EqualError(t, err, "JSON Extraction requires 1 result") assert.Empty(t, value) - results = &[]result.Value{ + results = []result.Value{ contextfull.NewResult(""), contextfull.NewResult(""), contextfull.NewResult(""), diff --git a/test/unit/internal/jsonextractor/updator_test.go b/test/unit/internal/jsonextractor/updator_test.go new file mode 100644 index 00000000..ec907c09 --- /dev/null +++ b/test/unit/internal/jsonextractor/updator_test.go @@ -0,0 +1,48 @@ +package document_test + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + jsonextractor "queryorchestration/internal/jsonExtractor" + queryprocessor "queryorchestration/internal/queryProcessor" + "testing" + + "github.com/google/uuid" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +func TestUpdatorValidate(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 := jsonextractor.NewUpdator(db) + assert.NotNil(t, svc) + + current := &queryprocessor.Query{ + ID: uuid.New(), + Type: queryprocessor.TypeJsonExtractor, + Version: int32(1), + RequiredQueryIDs: []uuid.UUID{}, + Config: "", + } + + entity := &queryprocessor.Update{ + ID: current.ID, + RequiredQueryIDs: []uuid.UUID{}, + Config: "", + } + + err = svc.Validate(ctx, current, entity) + assert.Nil(t, err) +} diff --git a/test/unit/internal/query/create_test.go b/test/unit/internal/query/create_test.go new file mode 100644 index 00000000..b84c1a02 --- /dev/null +++ b/test/unit/internal/query/create_test.go @@ -0,0 +1,66 @@ +package document_test + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" + "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 := query.New(db) + + config := "{\"path\":\"example_path\"}" + q := query.Query{ + ID: uuid.New(), + Type: queryprocessor.TypeJsonExtractor, + ActiveVersion: int32(1), + LatestVersion: int32(1), + RequiredQueryIDs: []uuid.UUID{ + uuid.New(), + }, + Config: config, + } + create := &queryprocessor.Create{ + Type: q.Type, + RequiredQueryIDs: q.RequiredQueryIDs, + Config: q.Config, + } + + dbType, err := queryprocessor.ToDBQueryType(create.Type) + assert.Nil(t, err) + + pool.ExpectBeginTx(pgx.TxOptions{}) + pool.ExpectQuery("name: CreateQuery :one").WithArgs(dbType).WillReturnRows( + pgxmock.NewRows([]string{"id"}). + AddRow(database.MustToDBUUID(q.ID)), + ) + for _, req := range create.RequiredQueryIDs { + pool.ExpectExec("name: CreateRequiredQuery :exec").WithArgs(database.MustToDBUUID(q.ID), database.MustToDBUUID(req), int32(1)). + WillReturnResult(pgxmock.NewResult("", 1)) + } + pool.ExpectExec("name: CreateQueryConfig :exec").WithArgs(database.MustToDBUUID(q.ID), []byte(create.Config), int32(1)). + WillReturnResult(pgxmock.NewResult("", 1)) + + id, err := svc.Create(ctx, create) + assert.Nil(t, err) + assert.Equal(t, q.ID, id) +} diff --git a/test/unit/internal/query/deprecate_test.go b/test/unit/internal/query/deprecate_test.go new file mode 100644 index 00000000..f17cd82b --- /dev/null +++ b/test/unit/internal/query/deprecate_test.go @@ -0,0 +1,66 @@ +package document_test + +import ( + "context" + "errors" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/query" + "testing" + + "github.com/google/uuid" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +func TestDeprecate(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 := query.New(db) + + id := uuid.New() + + pool.ExpectQuery("name: IsQueryDeprecated :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows( + pgxmock.NewRows([]string{"exists"}). + AddRow(false), + ) + pool.ExpectExec("name: DeprecateQuery :exec").WithArgs(database.MustToDBUUID(id)). + WillReturnResult(pgxmock.NewResult("", 1)) + + err = svc.Deprecate(ctx, id) + assert.Nil(t, err) + + pool.ExpectQuery("name: IsQueryDeprecated :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows( + pgxmock.NewRows([]string{"exists"}). + AddRow(true), + ) + + err = svc.Deprecate(ctx, id) + assert.Nil(t, err) + + pool.ExpectQuery("name: IsQueryDeprecated :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows( + pgxmock.NewRows([]string{"exists"}), + ) + + err = svc.Deprecate(ctx, id) + assert.NotNil(t, err) + + pool.ExpectQuery("name: IsQueryDeprecated :one").WithArgs(database.MustToDBUUID(id)).WillReturnRows( + pgxmock.NewRows([]string{"exists"}). + AddRow(false), + ) + pool.ExpectExec("name: DeprecateQuery :exec").WithArgs(database.MustToDBUUID(id)). + WillReturnError(errors.New("unable to deprecate query")) + + err = svc.Deprecate(ctx, id) + assert.NotNil(t, err) +} diff --git a/test/unit/internal/query/get_test.go b/test/unit/internal/query/get_test.go new file mode 100644 index 00000000..1f038670 --- /dev/null +++ b/test/unit/internal/query/get_test.go @@ -0,0 +1,57 @@ +package document_test + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +func TestGet(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 := query.New(db) + + config := "{\"path\":\"example_path\"}" + query := query.Query{ + ID: uuid.New(), + Type: queryprocessor.TypeJsonExtractor, + ActiveVersion: int32(1), + LatestVersion: int32(1), + RequiredQueryIDs: []uuid.UUID{ + uuid.New(), + }, + Config: config, + } + + dbReqIDs := make([]pgtype.UUID, len(query.RequiredQueryIDs)) + for index, id := range query.RequiredQueryIDs { + dbReqIDs[index] = database.MustToDBUUID(id) + } + + pool.ExpectQuery("name: GetQuery :one").WithArgs(database.MustToDBUUID(query.ID)).WillReturnRows( + pgxmock.NewRows([]string{"id", "type", "activeVersion", "latestVersion", "config", "requiredIds"}). + AddRow(database.MustToDBUUID(query.ID), repository.QuerytypeJsonExtractor, query.ActiveVersion, query.LatestVersion, []byte(config), dbReqIDs), + ) + + returnQuery, err := svc.Get(ctx, query.ID) + assert.Nil(t, err) + + assert.EqualExportedValues(t, query, *returnQuery) +} diff --git a/test/unit/internal/query/list_test.go b/test/unit/internal/query/list_test.go new file mode 100644 index 00000000..66dff32c --- /dev/null +++ b/test/unit/internal/query/list_test.go @@ -0,0 +1,59 @@ +package document_test + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +func TestList(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 := query.New(db) + + config := "{\"path\":\"example_path\"}" + q := &query.Query{ + ID: uuid.New(), + Type: queryprocessor.TypeJsonExtractor, + ActiveVersion: int32(1), + LatestVersion: int32(1), + RequiredQueryIDs: []uuid.UUID{ + uuid.New(), + }, + Config: config, + } + + dbReqIDs := make([]pgtype.UUID, len(q.RequiredQueryIDs)) + for index, id := range q.RequiredQueryIDs { + dbReqIDs[index] = database.MustToDBUUID(id) + } + + filters := query.ListFilters{} + + pool.ExpectQuery("name: ListQueries :many").WithArgs().WillReturnRows( + pgxmock.NewRows([]string{"id", "type", "activeVersion", "latestVersion", "config", "requiredIds"}). + AddRow(database.MustToDBUUID(q.ID), repository.QuerytypeJsonExtractor, q.ActiveVersion, q.LatestVersion, []byte(config), dbReqIDs), + ) + + resList, err := svc.List(ctx, filters) + assert.Nil(t, err) + + assert.EqualExportedValues(t, []*query.Query{q}, resList) +} diff --git a/test/unit/internal/query/parse_test.go b/test/unit/internal/query/parse_test.go index 0727afa4..7c8134ca 100644 --- a/test/unit/internal/query/parse_test.go +++ b/test/unit/internal/query/parse_test.go @@ -1,8 +1,10 @@ package document_test import ( + "queryorchestration/internal/database" "queryorchestration/internal/database/repository" "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" "testing" "github.com/google/uuid" @@ -10,58 +12,76 @@ import ( "github.com/stretchr/testify/assert" ) -func TestParseDBQuery(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{}, +func TestParseQuery(t *testing.T) { + q := &query.Query{ + ID: uuid.New(), + Type: queryprocessor.TypeJsonExtractor, + ActiveVersion: int32(1), + LatestVersion: int32(2), + RequiredQueryIDs: []uuid.UUID{ + uuid.New(), + }, + Config: "example", } - value, err := query.ParseDBQuery(&dbResult) - assert.Nil(t, err) - assert.Equal(t, uuid.Nil, value.ID) - assert.Equal(t, uuid.Nil, value.RequiredQueryID) - assert.Equal(t, int32(0), value.Version) - assert.Equal(t, query.Type(query.TypeJsonExtractor), value.Type) - dbResult.Type = repository.NullQuerytype{} - _, err = query.ParseDBQuery(&dbResult) - assert.EqualError(t, err, "invalid database query type") + out := query.ParseQuery(q) + assert.EqualExportedValues(t, queryprocessor.Query{ + ID: q.ID, + Type: q.Type, + Version: q.ActiveVersion, + RequiredQueryIDs: q.RequiredQueryIDs, + Config: q.Config, + }, *out) } -func TestParseDBType(t *testing.T) { - qType := repository.NullQuerytype{Valid: true, Querytype: repository.QuerytypeJsonExtractor} - value, err := query.ParseDBType(qType) +func TestParseDBListQuery(t *testing.T) { + q := &repository.ListQueriesRow{ + ID: database.MustToDBUUID(uuid.New()), + Type: repository.QuerytypeContextFull, + Activeversion: int32(1), + Latestversion: int32(2), + Requiredids: []pgtype.UUID{ + database.MustToDBUUID(uuid.New()), + }, + Config: []byte("example"), + } + + out, err := query.ParseDBListQuery(q) assert.Nil(t, err) - assert.Equal(t, query.Type(query.TypeJsonExtractor), value) - - qType = repository.NullQuerytype{} - _, err = query.ParseDBType(qType) - assert.EqualError(t, err, "invalid database query type") - - qType = repository.NullQuerytype{Valid: true} - _, err = query.ParseDBType(qType) - assert.EqualError(t, err, "invalid database query type") - - qType = repository.NullQuerytype{Valid: true, Querytype: repository.QuerytypeContextFull} - value, err = query.ParseDBType(qType) - assert.Nil(t, err) - assert.Equal(t, query.Type(query.TypeContextFull), value) + assert.EqualExportedValues(t, query.Query{ + ID: database.MustToUUID(q.ID), + Type: queryprocessor.TypeContextFull, + ActiveVersion: q.Activeversion, + LatestVersion: q.Latestversion, + RequiredQueryIDs: []uuid.UUID{ + database.MustToUUID(q.Requiredids[0]), + }, + Config: string(q.Config), + }, *out) } -func TestToDBQueryType(t *testing.T) { - dbQueryType := query.Type(query.TypeJsonExtractor) - value, err := query.ToDBQueryType(dbQueryType) - assert.Nil(t, err) - assert.Equal(t, repository.NullQuerytype{Querytype: repository.QuerytypeJsonExtractor, Valid: true}, value) +func TestParseDBGetQuery(t *testing.T) { + q := &repository.GetQueryRow{ + ID: database.MustToDBUUID(uuid.New()), + Type: repository.QuerytypeContextFull, + Activeversion: int32(1), + Latestversion: int32(2), + Requiredids: []pgtype.UUID{ + database.MustToDBUUID(uuid.New()), + }, + Config: []byte("example"), + } - dbQueryType = query.Type(-1) - _, err = query.ToDBQueryType(dbQueryType) - assert.EqualError(t, err, "invalid database query type") - - dbQueryType = query.Type(query.TypeContextFull) - value, err = query.ToDBQueryType(dbQueryType) + out, err := query.ParseDBGetQuery(q) assert.Nil(t, err) - assert.Equal(t, repository.NullQuerytype{Querytype: repository.QuerytypeContextFull, Valid: true}, value) + assert.EqualExportedValues(t, query.Query{ + ID: database.MustToUUID(q.ID), + Type: queryprocessor.TypeContextFull, + ActiveVersion: q.Activeversion, + LatestVersion: q.Latestversion, + RequiredQueryIDs: []uuid.UUID{ + database.MustToUUID(q.Requiredids[0]), + }, + Config: string(q.Config), + }, *out) } diff --git a/test/unit/internal/query/service_test.go b/test/unit/internal/query/service_test.go index 23111b58..cd5f43ca 100644 --- a/test/unit/internal/query/service_test.go +++ b/test/unit/internal/query/service_test.go @@ -1,7 +1,7 @@ package document_test import ( - "context" + "queryorchestration/internal/database" "queryorchestration/internal/database/repository" "queryorchestration/internal/query" "testing" @@ -11,15 +11,15 @@ import ( ) func TestService(t *testing.T) { - ctx := context.Background() - - db, err := pgxmock.NewConn() + pool, err := pgxmock.NewPool() if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - defer db.Close(ctx) - - queries := repository.New(db) - svc := query.New(queries) + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } + svc := query.New(db) assert.NotNil(t, svc) } diff --git a/test/unit/internal/query/test_test.go b/test/unit/internal/query/test_test.go new file mode 100644 index 00000000..ed8349e8 --- /dev/null +++ b/test/unit/internal/query/test_test.go @@ -0,0 +1,38 @@ +package document_test + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/query" + "testing" + + "github.com/google/uuid" + "github.com/pashagolub/pgxmock/v3" + "github.com/stretchr/testify/assert" +) + +func TestTest(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 := query.New(db) + + params := &query.Test{ + QueryID: uuid.New(), + DocumentID: uuid.New(), + QueryVersion: int32(1), + } + + result, err := svc.Test(ctx, *params) + assert.Nil(t, err) + assert.Empty(t, result) +} diff --git a/test/unit/internal/query/update_test.go b/test/unit/internal/query/update_test.go new file mode 100644 index 00000000..6bbef7e6 --- /dev/null +++ b/test/unit/internal/query/update_test.go @@ -0,0 +1,58 @@ +package document_test + +import ( + "context" + "queryorchestration/internal/database" + "queryorchestration/internal/database/repository" + "queryorchestration/internal/query" + queryprocessor "queryorchestration/internal/queryProcessor" + "testing" + + "github.com/google/uuid" + "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 := query.New(db) + + config := "{\"path\":\"example_path\"}" + existing := query.Query{ + ID: uuid.New(), + Type: queryprocessor.TypeJsonExtractor, + ActiveVersion: int32(1), + LatestVersion: int32(1), + RequiredQueryIDs: []uuid.UUID{ + uuid.New(), + }, + Config: config, + } + update := &queryprocessor.Update{ + ID: existing.ID, + RequiredQueryIDs: []uuid.UUID{ + uuid.New(), + }, + Config: config, + } + + dbReqIDs := database.MustToDBUUIDArray(existing.RequiredQueryIDs) + + pool.ExpectQuery("name: GetQuery :one").WithArgs(database.MustToDBUUID(update.ID)).WillReturnRows( + pgxmock.NewRows([]string{"id", "type", "activeVersion", "latestVersion", "config", "requiredIds"}). + AddRow(database.MustToDBUUID(existing.ID), repository.QuerytypeJsonExtractor, existing.ActiveVersion, existing.LatestVersion, []byte(config), dbReqIDs), + ) + + err = svc.Update(ctx, update) + assert.Nil(t, err) +} diff --git a/test/unit/internal/queryProcessor/parse_test.go b/test/unit/internal/queryProcessor/parse_test.go new file mode 100644 index 00000000..8f0d37fb --- /dev/null +++ b/test/unit/internal/queryProcessor/parse_test.go @@ -0,0 +1,90 @@ +package document_test + +import ( + "queryorchestration/internal/database/repository" + queryprocessor "queryorchestration/internal/queryProcessor" + "testing" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + "github.com/stretchr/testify/assert" +) + +func TestParseDBCollectorQuery(t *testing.T) { + dbResult := repository.GetCollectorQueriesRow{ + Collectorid: pgtype.UUID{}, + Queryid: pgtype.UUID{}, + Requiredids: []pgtype.UUID{}, + Type: repository.NullQuerytype{Valid: true, Querytype: repository.QuerytypeJsonExtractor}, + Queryversion: pgtype.Int4{}, + } + value, err := queryprocessor.ParseDBCollectorQuery(&dbResult) + assert.Nil(t, err) + assert.Equal(t, uuid.Nil, value.ID) + assert.Equal(t, []uuid.UUID{}, value.RequiredQueryIDs) + assert.Equal(t, int32(0), value.Version) + assert.Equal(t, queryprocessor.Type(queryprocessor.TypeJsonExtractor), value.Type) + + dbResult.Type = repository.NullQuerytype{} + _, err = queryprocessor.ParseDBCollectorQuery(&dbResult) + assert.EqualError(t, err, "invalid database query type") +} + +func TestParseDBNullType(t *testing.T) { + qType := repository.NullQuerytype{Valid: true, Querytype: repository.QuerytypeJsonExtractor} + value, err := queryprocessor.ParseDBNullType(qType) + assert.Nil(t, err) + assert.Equal(t, queryprocessor.Type(queryprocessor.TypeJsonExtractor), value) + + qType = repository.NullQuerytype{} + _, err = queryprocessor.ParseDBNullType(qType) + assert.EqualError(t, err, "invalid database query type") + + qType = repository.NullQuerytype{Valid: true} + _, err = queryprocessor.ParseDBNullType(qType) + assert.EqualError(t, err, "invalid database query type") +} + +func TestParseDBType(t *testing.T) { + qType := repository.QuerytypeJsonExtractor + value, err := queryprocessor.ParseDBType(qType) + assert.Nil(t, err) + assert.Equal(t, queryprocessor.Type(queryprocessor.TypeJsonExtractor), value) + + qType = repository.QuerytypeContextFull + value, err = queryprocessor.ParseDBType(qType) + assert.Nil(t, err) + assert.Equal(t, queryprocessor.Type(queryprocessor.TypeContextFull), value) +} + +func TestToDBQueryType(t *testing.T) { + dbQueryType := queryprocessor.Type(queryprocessor.TypeJsonExtractor) + value, err := queryprocessor.ToDBQueryType(dbQueryType) + assert.Nil(t, err) + assert.Equal(t, repository.Querytype(repository.QuerytypeJsonExtractor), value) + + dbQueryType = queryprocessor.Type(-1) + _, err = queryprocessor.ToDBQueryType(dbQueryType) + assert.EqualError(t, err, "invalid database query type") + + dbQueryType = queryprocessor.Type(queryprocessor.TypeContextFull) + value, err = queryprocessor.ToDBQueryType(dbQueryType) + assert.Nil(t, err) + assert.Equal(t, repository.Querytype(repository.QuerytypeContextFull), value) +} + +func TestToDBNullQueryType(t *testing.T) { + dbQueryType := queryprocessor.Type(queryprocessor.TypeJsonExtractor) + value, err := queryprocessor.ToDBNullQueryType(dbQueryType) + assert.Nil(t, err) + assert.Equal(t, repository.NullQuerytype{Valid: true, Querytype: repository.QuerytypeJsonExtractor}, value) + + dbQueryType = queryprocessor.Type(-1) + _, err = queryprocessor.ToDBNullQueryType(dbQueryType) + assert.EqualError(t, err, "invalid database query type") + + dbQueryType = queryprocessor.Type(queryprocessor.TypeContextFull) + value, err = queryprocessor.ToDBNullQueryType(dbQueryType) + assert.Nil(t, err) + assert.Equal(t, repository.NullQuerytype{Valid: true, Querytype: repository.QuerytypeContextFull}, value) +} diff --git a/test/unit/internal/queryQueue/queue_test.go b/test/unit/internal/queryQueue/queue_test.go index 709a2f8e..e1297c39 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" @@ -21,25 +21,27 @@ import ( func TestQueue(t *testing.T) { ctx := context.Background() - db, err := pgxmock.NewConn() + pool, err := pgxmock.NewPool() if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - defer db.Close(ctx) - - queries := repository.New(db) + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } jobID := uuid.New() dbJobID := database.MustToDBUUID(jobID) collectorID := uuid.New() dbCollectorID := database.MustToDBUUID(collectorID) - db.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID). + pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID). WillReturnRows( pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}). AddRow(dbCollectorID, dbJobID, int32(1), int32(1)), ) - coll, err := collector.NewByJobId(ctx, queries, jobID) + coll, err := collector.NewByJobId(ctx, db, jobID) assert.Nil(t, err) queryOneID := uuid.New() @@ -56,49 +58,52 @@ func TestQueue(t *testing.T) { querySixVersion := int32(6) contextID := uuid.New() contextVersion := int32(1) - collectorQueries := []query.Query{ - {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.ToDBNullQueryType(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) + pool.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorID).WillReturnRows(rows) contextResultID := uuid.New() - results := []result.Result{ + results := []*result.Result{ {ID: contextResultID, QueryID: contextID, QueryVersion: contextVersion}, {ID: uuid.New(), QueryID: queryFourID, QueryVersion: queryFourVersion}, {ID: uuid.New(), QueryID: querySixID, QueryVersion: querySixVersion - 1}, {ID: uuid.New(), QueryID: queryOneID, QueryVersion: queryOneVersion - 1}, } - expectedQueries := []query.Query{ - {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() cleanVersion := int32(1) textVersion := int32(1) - q, err := queryQueue.New(ctx, queries, coll, &results, docID, cleanVersion, textVersion) + q, err := queryQueue.New(ctx, db, coll, results, docID, cleanVersion, textVersion) assert.Nil(t, err) assert.Equal(t, expectedQueries, q.GetQueue()) @@ -108,74 +113,69 @@ func TestQueue(t *testing.T) { valueLayerOne := fmt.Sprintf("{\"%s\":\"%s\"}", keyLayerTwo, valueLayerTwo) valueContext := fmt.Sprintf("{\"%s\":%s}", keyLayerOne, valueLayerOne) - db.ExpectQuery("name: ListResultValuesByID :many").WithArgs([]pgtype.UUID{database.MustToDBUUID(contextResultID)}).WillReturnRows( + pool.ExpectQuery("name: ListResultValuesByID :many").WithArgs([]pgtype.UUID{database.MustToDBUUID(contextResultID)}).WillReturnRows( pgxmock.NewRows([]string{"id", "queryId", "value"}). AddRow(database.MustToDBUUID(contextResultID), database.MustToDBUUID(contextID), valueContext), ) - db.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(querySixID), querySixVersion).WillReturnRows( + pool.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(querySixID), querySixVersion).WillReturnRows( pgxmock.NewRows([]string{"id", "config"}). AddRow(pgtype.UUID{}, []byte(fmt.Sprintf("{\"path\":\"%s\"}", keyLayerOne))), ) - db.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(querySixID), database.MustToDBUUID(docID), valueLayerOne, cleanVersion, textVersion, querySixVersion). + pool.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(querySixID), database.MustToDBUUID(docID), valueLayerOne, cleanVersion, textVersion, querySixVersion). WillReturnRows( - pgxmock.NewRows([]string{"id"}). - AddRow(pgtype.UUID{}), + pgxmock.NewRows([]string{"id"}).AddRow(pgtype.UUID{}), ) - db.ExpectQuery("name: ListResultValuesByID :many").WithArgs(pgxmock.AnyArg()).WillReturnRows( + pool.ExpectQuery("name: ListResultValuesByID :many").WithArgs(pgxmock.AnyArg()).WillReturnRows( pgxmock.NewRows([]string{"id", "queryId", "value"}). AddRow(pgtype.UUID{}, database.MustToDBUUID(querySixID), valueLayerOne), ) - db.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(queryFiveID), queryFiveVersion).WillReturnRows( + pool.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(queryFiveID), queryFiveVersion).WillReturnRows( pgxmock.NewRows([]string{"id", "config"}). AddRow(pgtype.UUID{}, []byte(fmt.Sprintf("{\"path\":\"%s\"}", keyLayerTwo))), ) - db.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(queryFiveID), database.MustToDBUUID(docID), valueLayerTwo, cleanVersion, textVersion, queryFiveVersion). + pool.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(queryFiveID), database.MustToDBUUID(docID), valueLayerTwo, cleanVersion, textVersion, queryFiveVersion). WillReturnRows( - pgxmock.NewRows([]string{"id"}). - AddRow(pgtype.UUID{}), + pgxmock.NewRows([]string{"id"}).AddRow(pgtype.UUID{}), ) - db.ExpectQuery("name: ListResultValuesByID :many").WithArgs(pgxmock.AnyArg()).WillReturnRows( + pool.ExpectQuery("name: ListResultValuesByID :many").WithArgs(pgxmock.AnyArg()).WillReturnRows( pgxmock.NewRows([]string{"id", "queryId", "value"}). AddRow(database.MustToDBUUID(contextResultID), database.MustToDBUUID(contextID), valueContext), ) - db.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(queryOneID), queryOneVersion).WillReturnRows( + pool.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(queryOneID), queryOneVersion).WillReturnRows( pgxmock.NewRows([]string{"id", "config"}). AddRow(pgtype.UUID{}, []byte(fmt.Sprintf("{\"path\":\"%s\"}", keyLayerOne))), ) - db.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(queryOneID), database.MustToDBUUID(docID), valueLayerOne, cleanVersion, textVersion, queryOneVersion). + pool.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(queryOneID), database.MustToDBUUID(docID), valueLayerOne, cleanVersion, textVersion, queryOneVersion). WillReturnRows( - pgxmock.NewRows([]string{"id"}). - AddRow(pgtype.UUID{}), + pgxmock.NewRows([]string{"id"}).AddRow(pgtype.UUID{}), ) - db.ExpectQuery("name: ListResultValuesByID :many").WithArgs(pgxmock.AnyArg()).WillReturnRows( + pool.ExpectQuery("name: ListResultValuesByID :many").WithArgs(pgxmock.AnyArg()).WillReturnRows( pgxmock.NewRows([]string{"id", "queryId", "value"}). AddRow(pgtype.UUID{}, database.MustToDBUUID(queryOneID), valueLayerOne), ) - db.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(queryThreeID), queryThreeVersion).WillReturnRows( + pool.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(queryThreeID), queryThreeVersion).WillReturnRows( pgxmock.NewRows([]string{"id", "config"}). AddRow(pgtype.UUID{}, []byte(fmt.Sprintf("{\"path\":\"%s\"}", keyLayerTwo))), ) - db.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(queryThreeID), database.MustToDBUUID(docID), valueLayerTwo, cleanVersion, textVersion, queryThreeVersion). + pool.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(queryThreeID), database.MustToDBUUID(docID), valueLayerTwo, cleanVersion, textVersion, queryThreeVersion). WillReturnRows( - pgxmock.NewRows([]string{"id"}). - AddRow(pgtype.UUID{}), + pgxmock.NewRows([]string{"id"}).AddRow(pgtype.UUID{}), ) - db.ExpectQuery("name: ListResultValuesByID :many").WithArgs(pgxmock.AnyArg()).WillReturnRows( + pool.ExpectQuery("name: ListResultValuesByID :many").WithArgs(pgxmock.AnyArg()).WillReturnRows( pgxmock.NewRows([]string{"id", "queryId", "value"}). AddRow(pgtype.UUID{}, database.MustToDBUUID(queryOneID), valueLayerOne), ) - db.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(queryTwoID), queryTwoVersion).WillReturnRows( + pool.ExpectQuery("name: GetQueryConfig :one").WithArgs(database.MustToDBUUID(queryTwoID), queryTwoVersion).WillReturnRows( pgxmock.NewRows([]string{"id", "config"}). AddRow(pgtype.UUID{}, []byte(fmt.Sprintf("{\"path\":\"%s\"}", keyLayerTwo))), ) - db.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(queryTwoID), database.MustToDBUUID(docID), valueLayerTwo, cleanVersion, textVersion, queryTwoVersion). + pool.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(queryTwoID), database.MustToDBUUID(docID), valueLayerTwo, cleanVersion, textVersion, queryTwoVersion). WillReturnRows( - pgxmock.NewRows([]string{"id"}). - AddRow(pgtype.UUID{}), + pgxmock.NewRows([]string{"id"}).AddRow(pgtype.UUID{}), ) err = q.Execute(ctx) @@ -185,37 +185,39 @@ func TestQueue(t *testing.T) { func TestQueueFail(t *testing.T) { ctx := context.Background() - db, err := pgxmock.NewConn() + pool, err := pgxmock.NewPool() if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - defer db.Close(ctx) - - queries := repository.New(db) + queries := repository.New(pool) + db := &database.Connection{ + Queries: queries, + Pool: pool, + } jobID := uuid.New() dbJobID := database.MustToDBUUID(jobID) collectorID := uuid.New() dbCollectorID := database.MustToDBUUID(collectorID) - db.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID). + pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID). WillReturnRows( pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}). AddRow(dbCollectorID, dbJobID, int32(1), int32(1)), ) - coll, err := collector.NewByJobId(ctx, queries, jobID) + coll, err := collector.NewByJobId(ctx, db, jobID) assert.Nil(t, err) errr := "database failure" - db.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorID). + pool.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorID). WillReturnError(errors.New(errr)) - results := []result.Result{} + results := []*result.Result{} docID := uuid.New() cleanVersion := int32(1) textVersion := int32(1) - _, err = queryQueue.New(ctx, queries, coll, &results, docID, cleanVersion, textVersion) + _, err = queryQueue.New(ctx, db, coll, results, docID, cleanVersion, textVersion) assert.EqualError(t, err, errr) } diff --git a/test/unit/internal/result/store_test.go b/test/unit/internal/result/store_test.go index 1084fb23..8fc59313 100644 --- a/test/unit/internal/result/store_test.go +++ b/test/unit/internal/result/store_test.go @@ -16,13 +16,11 @@ import ( func TestStore(t *testing.T) { ctx := context.Background() - db, err := pgxmock.NewConn() + pool, err := pgxmock.NewPool() if err != nil { t.Fatalf("failed to open pgxmock database: %v", err) } - defer db.Close(ctx) - - queries := repository.New(db) + queries := repository.New(pool) resultStore := result.ResultStore{ QueryID: uuid.New(), @@ -33,10 +31,9 @@ func TestStore(t *testing.T) { QueryVersion: int32(1), } - db.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(resultStore.QueryID), database.MustToDBUUID(resultStore.DocumentID), resultStore.Value, resultStore.CleanVersion, resultStore.TextVersion, resultStore.QueryVersion). + pool.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(resultStore.QueryID), database.MustToDBUUID(resultStore.DocumentID), resultStore.Value, resultStore.CleanVersion, resultStore.TextVersion, resultStore.QueryVersion). WillReturnRows( - pgxmock.NewRows([]string{"id"}). - AddRow(pgtype.UUID{}), + pgxmock.NewRows([]string{"id"}).AddRow(pgtype.UUID{}), ) id, err := result.Store(ctx, queries, &resultStore) @@ -44,7 +41,7 @@ func TestStore(t *testing.T) { assert.NotNil(t, id) errr := "database failing" - db.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(resultStore.QueryID), database.MustToDBUUID(resultStore.DocumentID), resultStore.Value, resultStore.CleanVersion, resultStore.TextVersion, resultStore.QueryVersion). + pool.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(resultStore.QueryID), database.MustToDBUUID(resultStore.DocumentID), resultStore.Value, resultStore.CleanVersion, resultStore.TextVersion, resultStore.QueryVersion). WillReturnError(fmt.Errorf(errr)) id, err = result.Store(ctx, queries, &resultStore)