dbconn

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{

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()

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;
+    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);

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
 	}

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,
+	}
 }

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}
 }
 

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}
 }
 

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
+}

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)
 `

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,

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,
+	}
 }

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}
 }
 

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,
 	})

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}
 }
 

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
+}

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
 	}

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
 	}

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
 	}

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,
+	}
 }

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

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
 	}

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
 	}

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,

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,

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,