jacob-mathison/query-orchestration
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Merged in fature/jobs (pull request #34)

Browse Source 
Job Collector

* createstructure

* mostupdatevalidation

* repocollectorupdate

* updateoutline

* updatevalidation

* scriptupdate

* cleanupdockerignore

* update

* collectorupdateapi
This commit is contained in:
Michael McGuinness
2025-01-23 14:56:20 +00:00
parent 36967fc946
commit 5b7160fe44
88 changed files with 2181 additions and 297 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/query/create.go
+32 -36
View File
@@ -28,7 +28,7 @@ func (s *Service) Create(ctx context.Context, entity *queryprocessor.Create) (uu
}
func (s *Service) normalizeCreate(ctx context.Context, entity *queryprocessor.Create) error {
err := s.normalizeQueryIDs(ctx, entity)
err := s.NormalizeQueryIDs(ctx, entity)
if err != nil {
return err
}
@@ -57,51 +57,47 @@ func (s *Service) submitCreate(ctx context.Context, entity *queryprocessor.Creat
return uuid.Nil, err
}
tx, err := s.db.Pool.Begin(ctx)
if err != nil {
return uuid.Nil, err
}
defer func() {
_ = tx.Rollback(ctx)
}()
var dbID pgtype.UUID
err = database.ExecuteTransaction(ctx, s.db, func(ctx context.Context, qtx *repository.Queries) error {
dbID, err = qtx.CreateQuery(ctx, query.Type)
if err != nil {
return err
}
qtx := s.db.Queries.WithTx(tx)
dbID, err := qtx.CreateQuery(ctx, query.Type)
if err != nil {
return uuid.Nil, err
}
if query.RequiredQueryIDs != nil {
for _, reqQuery := range *query.RequiredQueryIDs {
err = qtx.AddRequiredQuery(ctx, &repository.AddRequiredQueryParams{
Queryid: dbID,
Requiredqueryid: reqQuery,
Addedversion: 1,
})
if err != nil {
return uuid.Nil, err
if query.RequiredQueryIDs != nil {
for _, reqQuery := range *query.RequiredQueryIDs {
err = qtx.AddRequiredQuery(ctx, &repository.AddRequiredQueryParams{
Queryid: dbID,
Requiredqueryid: reqQuery,
Addedversion: 1,
})
if err != nil {
return err
}
}
}
}
if query.Config != nil && string(*query.Config) != "" {
err = qtx.AddQueryConfig(ctx, &repository.AddQueryConfigParams{
Queryid: dbID,
Config: *query.Config,
Addedversion: 1,
})
if err != nil {
return uuid.Nil, err
if query.Config != nil && string(*query.Config) != "" {
err = qtx.AddQueryConfig(ctx, &repository.AddQueryConfigParams{
Queryid: dbID,
Config: *query.Config,
Addedversion: 1,
})
if err != nil {
return err
}
}
}
err = tx.Commit(ctx)
return nil
})
if err != nil {
return uuid.Nil, err
}
id := database.MustToUUID(dbID)
id, err := database.ToUUID(dbID)
if err != nil {
return uuid.Nil, err
}
return id, nil
internal/query/document/sync_test.go
+11 -11
View File
@@ -45,7 +45,7 @@ func TestSyncIsSynced(t *testing.T) {
pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(database.MustToDBUUID(doc.JobID)).
WillReturnRows(
pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}).
AddRow(database.MustToDBUUID(coll.ID), database.MustToDBUUID(coll.JobID), coll.MinCleanVersion, coll.MinTextVersion, int32(1), int32(2), []byte("")),
AddRow(database.MustToDBUUID(coll.ID), database.MustToDBUUID(coll.JobID), &coll.MinCleanVersion, &coll.MinTextVersion, int32(1), int32(2), []byte("")),
)
pool.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), coll.MinCleanVersion, coll.MinTextVersion).
WillReturnRows(
@@ -59,7 +59,7 @@ func TestSyncIsSynced(t *testing.T) {
)
docSvc := document.New(db, &document.Services{
Collector: collector.New(db),
Collector: collector.New(db, &collector.Services{}),
})
err = docSvc.Sync(ctx, &doc)
assert.Nil(t, err)
@@ -96,7 +96,7 @@ func TestSyncDBFail(t *testing.T) {
)
docSvc := document.New(db, &document.Services{
Collector: collector.New(db),
Collector: collector.New(db, &collector.Services{}),
})
err = docSvc.Sync(ctx, &doc)
assert.EqualError(t, err, "no rows in result set")
@@ -104,14 +104,14 @@ func TestSyncDBFail(t *testing.T) {
pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(dbJobID).
WillReturnRows(
pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}).
AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion, int32(1), int32(2), []byte("")),
AddRow(dbCollectorId, dbJobID, &minCleanVersion, &minTextVersion, int32(1), int32(2), []byte("")),
)
dbErr := "database failure"
pool.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), minCleanVersion, minTextVersion).
WillReturnError(errors.New(dbErr))
docSvc = document.New(db, &document.Services{
Collector: collector.New(db),
Collector: collector.New(db, &collector.Services{}),
})
err = docSvc.Sync(ctx, &doc)
assert.EqualError(t, err, dbErr)
@@ -119,7 +119,7 @@ func TestSyncDBFail(t *testing.T) {
pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(dbJobID).
WillReturnRows(
pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}).
AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion, int32(1), int32(2), []byte("")),
AddRow(dbCollectorId, dbJobID, &minCleanVersion, &minTextVersion, int32(1), int32(2), []byte("")),
)
pool.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), minCleanVersion, minTextVersion).
WillReturnRows(
@@ -130,7 +130,7 @@ func TestSyncDBFail(t *testing.T) {
WillReturnError(errors.New(dbErr))
docSvc = document.New(db, &document.Services{
Collector: collector.New(db),
Collector: collector.New(db, &collector.Services{}),
})
err = docSvc.Sync(ctx, &doc)
assert.EqualError(t, err, dbErr)
@@ -138,7 +138,7 @@ func TestSyncDBFail(t *testing.T) {
pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(dbJobID).
WillReturnRows(
pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}).
AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion, int32(1), int32(2), []byte("")),
AddRow(dbCollectorId, dbJobID, &minCleanVersion, &minTextVersion, int32(1), int32(2), []byte("")),
)
qV := int32(1)
reqID := database.MustToDBUUID(uuid.New())
@@ -159,7 +159,7 @@ func TestSyncDBFail(t *testing.T) {
WillReturnError(errors.New(dbErr))
docSvc = document.New(db, &document.Services{
Collector: collector.New(db),
Collector: collector.New(db, &collector.Services{}),
})
err = docSvc.Sync(ctx, &doc)
assert.EqualError(t, err, dbErr)
@@ -193,7 +193,7 @@ func TestSync(t *testing.T) {
pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(dbJobID).
WillReturnRows(
pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}).
AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion, int32(1), int32(2), []byte("")),
AddRow(dbCollectorId, dbJobID, &minCleanVersion, &minTextVersion, int32(1), int32(2), []byte("")),
)
pool.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), minCleanVersion, minTextVersion).
WillReturnRows(
@@ -211,7 +211,7 @@ func TestSync(t *testing.T) {
)
docSvc := document.New(db, &document.Services{
Collector: collector.New(db),
Collector: collector.New(db, &collector.Services{}),
})
err = docSvc.Sync(ctx, &doc)
assert.EqualError(t, err, "JSON Extraction requires 1 result")
internal/query/normalize.go
+10 -10
View File
@@ -7,6 +7,7 @@ import (
"fmt"
"queryorchestration/internal/database"
queryprocessor "queryorchestration/internal/query/processor"
"queryorchestration/internal/server/validation"
"strings"
"github.com/google/uuid"
@@ -55,7 +56,7 @@ type RequiredQueryIDs interface {
SetRequiredQueryIDs(*[]uuid.UUID)
}
func (s *Service) normalizeQueryIDs(ctx context.Context, ids RequiredQueryIDs) error {
func (s *Service) NormalizeQueryIDs(ctx context.Context, ids RequiredQueryIDs) error {
if ids == nil || ids.GetRequiredQueryIDs() == nil {
return nil
}
@@ -67,7 +68,10 @@ func (s *Service) normalizeQueryIDs(ctx context.Context, ids RequiredQueryIDs) e
return nil
}
dbids := database.MustToDBUUIDArray(*ide)
dedup := validation.DeduplicateArray(*ide)
ids.SetRequiredQueryIDs(&dedup)
dbids := database.MustToDBUUIDArray(dedup)
exist, err := s.db.Queries.AllQueriesExist(ctx, dbids)
if err != nil {
@@ -84,17 +88,13 @@ func (s *Service) normalizeActiveVersion(current *Query, entity *queryprocessor.
return errors.New("current query required")
}
if entity == nil || entity.ActiveVersion == nil {
if entity == nil {
return nil
}
if entity.ActiveVersion == &current.ActiveVersion {
entity.ActiveVersion = nil
return nil
}
if *entity.ActiveVersion < 1 || *entity.ActiveVersion > current.LatestVersion+1 {
return fmt.Errorf("active version must be in the range: 1 <= activeVersion <= %d", current.LatestVersion+1)
err := validation.NormalizeInClosedInterval(&entity.ActiveVersion, current.ActiveVersion, 1, current.LatestVersion+1)
if err != nil {
return err
}
return nil
internal/query/normalize_test.go
+20 -6
View File
@@ -81,18 +81,18 @@ func TestNormalizeQueryIDs(t *testing.T) {
}
s := Service{db: db}
err = s.normalizeQueryIDs(ctx, nil)
err = s.NormalizeQueryIDs(ctx, nil)
assert.Nil(t, err)
entity := queryprocessor.Create{}
entity.RequiredQueryIDs = nil
err = s.normalizeQueryIDs(ctx, &entity)
err = s.NormalizeQueryIDs(ctx, &entity)
assert.Nil(t, err)
assert.Nil(t, entity.RequiredQueryIDs)
entity.RequiredQueryIDs = &[]uuid.UUID{}
err = s.normalizeQueryIDs(ctx, &entity)
err = s.NormalizeQueryIDs(ctx, &entity)
assert.Nil(t, err)
assert.Nil(t, entity.RequiredQueryIDs)
@@ -105,7 +105,7 @@ func TestNormalizeQueryIDs(t *testing.T) {
AddRow(true),
)
err = s.normalizeQueryIDs(ctx, &entity)
err = s.NormalizeQueryIDs(ctx, &entity)
assert.Nil(t, err)
assert.ElementsMatch(t, ids, *entity.RequiredQueryIDs)
@@ -114,16 +114,30 @@ func TestNormalizeQueryIDs(t *testing.T) {
AddRow(false),
)
err = s.normalizeQueryIDs(ctx, &entity)
err = s.NormalizeQueryIDs(ctx, &entity)
assert.Error(t, err)
assert.ElementsMatch(t, ids, *entity.RequiredQueryIDs)
pool.ExpectQuery("name: AllQueriesExist :one").WithArgs(dbids).
WillReturnError(errors.New("database failure"))
err = s.normalizeQueryIDs(ctx, &entity)
err = s.NormalizeQueryIDs(ctx, &entity)
assert.Error(t, err)
assert.ElementsMatch(t, ids, *entity.RequiredQueryIDs)
singleid := uuid.New()
entity.RequiredQueryIDs = &[]uuid.UUID{singleid, singleid}
outids := []uuid.UUID{singleid}
dbids = database.MustToDBUUIDArray(outids)
pool.ExpectQuery("name: AllQueriesExist :one").WithArgs(dbids).WillReturnRows(
pgxmock.NewRows([]string{"all_exist"}).
AddRow(true),
)
err = s.NormalizeQueryIDs(ctx, &entity)
assert.Nil(t, err)
assert.ElementsMatch(t, outids, *entity.RequiredQueryIDs)
}
func TestNormalizeActiveVersion(t *testing.T) {
internal/query/update.go
+33 -27
View File
@@ -2,6 +2,7 @@ package query
import (
"context"
"errors"
"fmt"
"queryorchestration/internal/database"
"queryorchestration/internal/database/repository"
@@ -37,16 +38,34 @@ func (s *Service) normalizeUpdate(ctx context.Context, current *Query, entity *q
return err
}
err = s.normalizeQueryIDs(ctx, entity)
err = s.NormalizeQueryIDs(ctx, entity)
if err != nil {
return err
}
if entity.RequiredQueryIDs != nil {
createsloop, err := s.db.Queries.IsQueryInDependencyTree(ctx, &repository.IsQueryInDependencyTreeParams{
Requiredqueryid: database.MustToDBUUID(current.ID),
ID: database.MustToDBUUIDArray(*entity.RequiredQueryIDs),
})
if err != nil {
return err
} else if createsloop {
return errors.New("required ids create a loop")
}
}
err = s.normalizeConfig(entity)
if err != nil {
return err
}
if entity.ActiveVersion == nil &&
entity.RequiredQueryIDs == nil &&
entity.Config == nil {
return errors.New("no changes")
}
validator, err := s.getUpdator(current.Type)
if err != nil {
return err
@@ -74,8 +93,6 @@ func (s *Service) submitUpdate(ctx context.Context, current *Query, entity *quer
latestVersion := current.LatestVersion + 1
id := database.MustToDBUUID(entity.ID)
hasChanges := false
addIDs := getSetDifference(entity.RequiredQueryIDs, current.RequiredQueryIDs)
for _, qID := range addIDs {
err = qtx.AddRequiredQuery(ctx, &repository.AddRequiredQueryParams{
@@ -86,7 +103,6 @@ func (s *Service) submitUpdate(ctx context.Context, current *Query, entity *quer
if err != nil {
return err
}
hasChanges = true
}
removeIDs := getSetDifference(current.RequiredQueryIDs, entity.RequiredQueryIDs)
@@ -99,8 +115,6 @@ func (s *Service) submitUpdate(ctx context.Context, current *Query, entity *quer
if err != nil {
return err
}
hasChanges = true
}
if entity.Config != nil && *entity.Config != "" {
@@ -120,33 +134,25 @@ func (s *Service) submitUpdate(ctx context.Context, current *Query, entity *quer
if err != nil {
return err
}
hasChanges = true
}
activeVersion := current.ActiveVersion
if entity.ActiveVersion != nil {
activeVersion = *entity.ActiveVersion
activeVersion := entity.ActiveVersion
if activeVersion == nil {
activeVersion = &current.ActiveVersion
}
if activeVersion != current.ActiveVersion {
hasChanges = true
err = qtx.UpdateQuery(ctx, &repository.UpdateQueryParams{
Latestversion: latestVersion,
Activeversion: *activeVersion,
ID: id,
})
if err != nil {
return err
}
if hasChanges {
err = qtx.UpdateQuery(ctx, &repository.UpdateQueryParams{
Latestversion: latestVersion,
Activeversion: activeVersion,
ID: id,
})
if err != nil {
return err
}
err = tx.Commit(ctx)
if err != nil {
return err
}
err = tx.Commit(ctx)
if err != nil {
return err
}
return nil
internal/query/update_test.go
+1 -7
View File
@@ -9,7 +9,6 @@ import (
"testing"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgtype"
"github.com/pashagolub/pgxmock/v3"
"github.com/stretchr/testify/assert"
@@ -44,11 +43,6 @@ func TestUpdate(t *testing.T) {
AddRow(database.MustToDBUUID(existing.ID), repository.QuerytypeJsonExtractor, existing.ActiveVersion, existing.LatestVersion, []byte(config), []pgtype.UUID{}),
)
pool.ExpectBeginTx(pgx.TxOptions{})
pool.ExpectExec("name: UpdateQuery :exec").WithArgs(int32(1), int32(2), database.MustToDBUUID(update.ID)).
WillReturnResult(pgxmock.NewResult("", 1))
pool.ExpectCommit()
err = svc.Update(ctx, update)
assert.Nil(t, err)
assert.Error(t, err)
}
internal/query/updateprivate_test.go
+18 -30
View File
@@ -213,36 +213,6 @@ func TestSubmitUpdateActiveVersion(t *testing.T) {
assert.Nil(t, err)
}
func TestSubmitUpdateNoChange(t *testing.T) {
ctx := context.Background()
pool, err := pgxmock.NewPool()
if err != nil {
t.Fatalf("failed to open pgxmock database: %v", err)
}
queries := repository.New(pool)
db := &database.Connection{
Queries: queries,
Pool: pool,
}
svc := New(db)
q := Query{
ID: uuid.New(),
ActiveVersion: int32(1),
LatestVersion: int32(2),
}
update := &queryprocessor.Update{
ID: q.ID,
}
pool.ExpectBeginTx(pgx.TxOptions{})
pool.ExpectRollback()
err = svc.submitUpdate(ctx, &q, update)
assert.Nil(t, err)
}
func TestGetSetDifference(t *testing.T) {
commonUUID := uuid.New()
listA := []uuid.UUID{uuid.New(), commonUUID}
@@ -300,4 +270,22 @@ func TestNormalizeUpdate(t *testing.T) {
Config: &cfg,
RequiredQueryIDs: nil,
}, *update)
update.RequiredQueryIDs = &[]uuid.UUID{uuid.New()}
dbids = database.MustToDBUUIDArray(*update.RequiredQueryIDs)
pool.ExpectQuery("name: IsQueryInDependencyTree :one").WithArgs(dbids).WillReturnRows(
pgxmock.NewRows([]string{"all_exist"}).
AddRow(true),
)
err = svc.normalizeUpdate(ctx, current, update)
assert.Error(t, err)
update = &queryprocessor.Update{
ID: current.ID,
}
err = svc.normalizeUpdate(ctx, current, update)
assert.Error(t, err)
}