From c282a052ef092b0064a2de0e3f3d1901e3f924d1 Mon Sep 17 00:00:00 2001 From: Michael McGuinness Date: Mon, 6 Jan 2025 12:26:28 +0000 Subject: [PATCH] dbconn --- cmd/queryRunner/main.go | 8 ++- cmd/queryService/main.go | 8 ++- database/queries/query.sql | 11 +++- internal/collector/collector.go | 7 +- internal/collector/service.go | 12 ++-- internal/contextFull/creator.go | 6 +- internal/contextFull/updator.go | 6 +- internal/database/pool.go | 12 ++++ internal/database/repository/query.sql.go | 41 ++++++++++++ internal/document/service.go | 8 +-- internal/export/service.go | 10 +-- internal/jsonExtractor/creator.go | 6 +- internal/jsonExtractor/service.go | 6 +- internal/jsonExtractor/updator.go | 6 +- internal/query/create.go | 80 ++++++++++++++++++++++- internal/query/deprecate.go | 4 +- internal/query/get.go | 2 +- internal/query/list.go | 2 +- internal/query/service.go | 12 ++-- internal/queryProcessor/parse.go | 13 +++- internal/queryQueue/create.go | 2 +- internal/queryQueue/execute.go | 2 +- internal/queryQueue/result.go | 2 +- internal/queryQueue/service.go | 6 +- internal/result/service.go | 4 +- 25 files changed, 218 insertions(+), 58 deletions(-) create mode 100644 internal/database/pool.go 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/queries/query.sql b/database/queries/query.sql index f9eba802..4355ed2a 100644 --- a/database/queries/query.sql +++ b/database/queries/query.sql @@ -30,4 +30,13 @@ SELECT q.id, q.type, q.activeVersion, q.latestVersion, c.config, ARRAY_AGG(r.req 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; \ No newline at end of file + 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 index d74cf825..67793db0 100644 --- a/internal/contextFull/creator.go +++ b/internal/contextFull/creator.go @@ -2,15 +2,15 @@ package contextfull import ( "context" - "queryorchestration/internal/database/repository" + "queryorchestration/internal/database" queryprocessor "queryorchestration/internal/queryProcessor" ) type Creator struct { - db *repository.Queries + db *database.Connection } -func NewCreator(db *repository.Queries) Creator { +func NewCreator(db *database.Connection) Creator { return Creator{db} } diff --git a/internal/contextFull/updator.go b/internal/contextFull/updator.go index 5dd299e6..ac24fc4a 100644 --- a/internal/contextFull/updator.go +++ b/internal/contextFull/updator.go @@ -2,15 +2,15 @@ package contextfull import ( "context" - "queryorchestration/internal/database/repository" + "queryorchestration/internal/database" queryprocessor "queryorchestration/internal/queryProcessor" ) type Updator struct { - db *repository.Queries + db *database.Connection } -func NewUpdator(db *repository.Queries) Updator { +func NewUpdator(db *database.Connection) Updator { return Updator{db} } diff --git a/internal/database/pool.go b/internal/database/pool.go new file mode 100644 index 00000000..576d6d75 --- /dev/null +++ b/internal/database/pool.go @@ -0,0 +1,12 @@ +package database + +import ( + "queryorchestration/internal/database/repository" + + "github.com/jackc/pgx/v5/pgxpool" +) + +type Connection struct { + Pool *pgxpool.Pool + Queries *repository.Queries +} diff --git a/internal/database/repository/query.sql.go b/internal/database/repository/query.sql.go index 36e81644..431cc359 100644 --- a/internal/database/repository/query.sql.go +++ b/internal/database/repository/query.sql.go @@ -11,6 +11,47 @@ 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) ` diff --git a/internal/document/service.go b/internal/document/service.go index 4c8d974e..3f09bc53 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, } } @@ -56,7 +56,7 @@ func (s *Service) Sync(ctx context.Context, doc *Document) 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, 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 index e69f990e..8a328dde 100644 --- a/internal/jsonExtractor/creator.go +++ b/internal/jsonExtractor/creator.go @@ -2,15 +2,15 @@ package jsonextractor import ( "context" - "queryorchestration/internal/database/repository" + "queryorchestration/internal/database" queryprocessor "queryorchestration/internal/queryProcessor" ) type Creator struct { - db *repository.Queries + db *database.Connection } -func NewCreator(db *repository.Queries) Creator { +func NewCreator(db *database.Connection) Creator { return Creator{db} } diff --git a/internal/jsonExtractor/service.go b/internal/jsonExtractor/service.go index 2762a4a1..36098949 100644 --- a/internal/jsonExtractor/service.go +++ b/internal/jsonExtractor/service.go @@ -13,14 +13,14 @@ import ( ) type Extractor struct { - db *repository.Queries + db *database.Connection } type Config struct { Path string `json:"path"` } -func NewExtractor(db *repository.Queries) Extractor { +func NewExtractor(db *database.Connection) Extractor { return Extractor{db} } @@ -34,7 +34,7 @@ func (e Extractor) Process(ctx context.Context, query queryprocessor.Query, valu 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 index 2a689d03..10697050 100644 --- a/internal/jsonExtractor/updator.go +++ b/internal/jsonExtractor/updator.go @@ -2,15 +2,15 @@ package jsonextractor import ( "context" - "queryorchestration/internal/database/repository" + "queryorchestration/internal/database" queryprocessor "queryorchestration/internal/queryProcessor" ) type Updator struct { - db *repository.Queries + db *database.Connection } -func NewUpdator(db *repository.Queries) Updator { +func NewUpdator(db *database.Connection) Updator { return Updator{db} } diff --git a/internal/query/create.go b/internal/query/create.go index 3a50b256..5c4acb22 100644 --- a/internal/query/create.go +++ b/internal/query/create.go @@ -4,10 +4,13 @@ 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) { @@ -21,10 +24,59 @@ func (s *Service) Create(ctx context.Context, entity *queryprocessor.Create) (uu return uuid.Nil, err } - // TODO - submit create - type, requiredids, config - id := uuid.New() + 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 { + err = qtx.CreateQueryConfig(ctx, repository.CreateQueryConfigParams{ + Queryid: dbID, + Config: query.Config, + Addedversion: 1, + }) + if err != nil { + return uuid.Nil, err + } + } + + id := database.MustToUUID(dbID) return id, nil + } func (s *Service) getCreator(qType queryprocessor.Type) (queryprocessor.Creator, error) { @@ -37,3 +89,27 @@ func (s *Service) getCreator(qType queryprocessor.Type) (queryprocessor.Creator, 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 := make([]pgtype.UUID, len(q.RequiredQueryIDs)) + for index, id := range q.RequiredQueryIDs { + reqIDs[index] = database.MustToDBUUID(id) + } + + return &createQuery{ + Type: t, + RequiredQueryIDs: reqIDs, + Config: []byte(q.Config), + }, nil +} diff --git a/internal/query/deprecate.go b/internal/query/deprecate.go index 094f4337..9fe06525 100644 --- a/internal/query/deprecate.go +++ b/internal/query/deprecate.go @@ -10,14 +10,14 @@ import ( func (s *Service) Deprecate(ctx context.Context, id uuid.UUID) error { dbId := database.MustToDBUUID(id) - exists, err := s.db.IsQueryDeprecated(ctx, dbId) + exists, err := s.db.Queries.IsQueryDeprecated(ctx, dbId) if err != nil { return err } else if exists { return nil } - err = s.db.DeprecateQuery(ctx, dbId) + err = s.db.Queries.DeprecateQuery(ctx, dbId) if err != nil { return err } diff --git a/internal/query/get.go b/internal/query/get.go index cfeb30d2..d5b366b3 100644 --- a/internal/query/get.go +++ b/internal/query/get.go @@ -19,7 +19,7 @@ type Query struct { } func (s *Service) Get(ctx context.Context, id uuid.UUID) (*Query, error) { - query, err := s.db.GetQuery(ctx, database.MustToDBUUID(id)) + query, err := s.db.Queries.GetQuery(ctx, database.MustToDBUUID(id)) if err != nil { return nil, err } diff --git a/internal/query/list.go b/internal/query/list.go index f3fb9c97..c07339cd 100644 --- a/internal/query/list.go +++ b/internal/query/list.go @@ -14,7 +14,7 @@ type ListFilters struct { } func (s *Service) List(ctx context.Context, filters ListFilters) (*[]Query, error) { - dbQueries, err := s.db.ListQueries(ctx) + dbQueries, err := s.db.Queries.ListQueries(ctx) if err != nil { return nil, err } diff --git a/internal/query/service.go b/internal/query/service.go index 8f773c59..75dca704 100644 --- a/internal/query/service.go +++ b/internal/query/service.go @@ -1,13 +1,13 @@ package query -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/queryProcessor/parse.go b/internal/queryProcessor/parse.go index 0d5e5717..73464fbe 100644 --- a/internal/queryProcessor/parse.go +++ b/internal/queryProcessor/parse.go @@ -27,7 +27,7 @@ func ParseDBType(qType repository.Querytype) (Type, error) { } } -func ToDBQueryType(t Type) (repository.NullQuerytype, error) { +func ToDBQueryType(t Type) (repository.Querytype, error) { var dbType repository.Querytype switch t { @@ -36,7 +36,16 @@ func ToDBQueryType(t Type) (repository.NullQuerytype, error) { case TypeContextFull: dbType = repository.QuerytypeContextFull default: - return repository.NullQuerytype{}, fmt.Errorf("invalid database query type") + 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 diff --git a/internal/queryQueue/create.go b/internal/queryQueue/create.go index e9a1aa30..2d6731e5 100644 --- a/internal/queryQueue/create.go +++ b/internal/queryQueue/create.go @@ -31,7 +31,7 @@ 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 } diff --git a/internal/queryQueue/execute.go b/internal/queryQueue/execute.go index f82faf42..da5f5632 100644 --- a/internal/queryQueue/execute.go +++ b/internal/queryQueue/execute.go @@ -49,7 +49,7 @@ func (q *Queue) executeQuery(ctx context.Context, qu queryprocessor.Query) error } } - values, err := q.db.ListResultValuesByID(ctx, resultIDs) + values, err := q.db.Queries.ListResultValuesByID(ctx, resultIDs) if err != nil { return err } diff --git a/internal/queryQueue/result.go b/internal/queryQueue/result.go index c5cddf07..5b2a1431 100644 --- a/internal/queryQueue/result.go +++ b/internal/queryQueue/result.go @@ -20,7 +20,7 @@ func (q *Queue) setResult(ctx context.Context, qu queryprocessor.Query, resultVa 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, diff --git a/internal/queryQueue/service.go b/internal/queryQueue/service.go index 152df861..d6307336 100644 --- a/internal/queryQueue/service.go +++ b/internal/queryQueue/service.go @@ -3,7 +3,7 @@ package queryQueue import ( "context" "queryorchestration/internal/collector" - "queryorchestration/internal/database/repository" + "queryorchestration/internal/database" queryprocessor "queryorchestration/internal/queryProcessor" "queryorchestration/internal/result" @@ -15,13 +15,13 @@ type Queue struct { 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, 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,