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

Merged in feature/jobcollector (pull request #30)

Browse Source 
Initial Job Collector (changes pending)

* movearroundtocleancollector

* internalgetfunctions

* completecollectorquery

* simplify

* fixtests

* addvendor

* noplaceholder
This commit is contained in:
Michael McGuinness
2025-01-21 12:28:46 +00:00
parent b888e3450f
commit 4ccb980593
46 changed files with 951 additions and 655 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/query/document/service.go
+19 -6
View File
@@ -19,28 +19,41 @@ type Document struct {
TextVersion int32 `json:"textVersion" validate:"required,gt=0"`
}
type Service struct {
db *database.Connection
type Services struct {
Collector *collector.Service
}
func New(db *database.Connection) *Service {
type Service struct {
db *database.Connection
svc *Services
}
func New(db *database.Connection, svc *Services) *Service {
return &Service{
db,
svc,
}
}
func (s *Service) Sync(ctx context.Context, doc *Document) error {
collector, err := collector.NewByJobId(ctx, s.db, doc.JobID)
coll, err := s.svc.Collector.GetByJobID(ctx, doc.JobID)
if err != nil {
return err
}
results, err := s.getResults(ctx, doc.ID, collector)
results, err := s.getResults(ctx, doc.ID, coll)
if err != nil {
return err
}
queue, err := queryqueue.New(ctx, s.db, collector, results, doc.ID, doc.CleanVersion, doc.TextVersion)
queue, err := queryqueue.New(ctx, &queryqueue.NewConfig{
DB: s.db,
Collector: coll,
Results: results,
DocumentID: doc.ID,
CleanVersion: doc.CleanVersion,
TextVersion: doc.TextVersion,
})
if err != nil {
return err
}
internal/query/document/sync_test.go
+53 -38
View File
@@ -5,6 +5,7 @@ import (
"errors"
"queryorchestration/internal/database"
"queryorchestration/internal/database/repository"
"queryorchestration/internal/job/collector"
"queryorchestration/internal/query/document"
"testing"
@@ -33,29 +34,33 @@ func TestSyncIsSynced(t *testing.T) {
Name: "document_name",
}
dbCollectorId := database.MustToDBUUID(uuid.New())
dbJobID := database.MustToDBUUID(doc.JobID)
minCleanVersion := int32(1)
minTextVersion := int32(1)
qV := int32(1)
queryVersion := int32(1)
coll := collector.Collector{
ID: uuid.New(),
JobID: doc.JobID,
MinCleanVersion: int32(1),
MinTextVersion: int32(2),
}
pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID).
pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(database.MustToDBUUID(doc.JobID)).
WillReturnRows(
pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}).
AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion),
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("")),
)
pool.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), minCleanVersion, minTextVersion).
pool.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), coll.MinCleanVersion, coll.MinTextVersion).
WillReturnRows(
pgxmock.NewRows([]string{"id", "queryId", "queryVersion"}).
AddRow(pgtype.UUID{}, pgtype.UUID{}, qV),
AddRow(pgtype.UUID{}, pgtype.UUID{}, queryVersion),
)
pool.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorId).
pool.ExpectQuery("name: ListCollectorQueries :many").WithArgs(database.MustToDBUUID(coll.ID)).
WillReturnRows(
pgxmock.NewRows([]string{"collectorId", "queryId", "type", "queryVersion", "requiredIds"}).
AddRow(dbCollectorId, pgtype.UUID{}, repository.NullQuerytype{Querytype: repository.QuerytypeJsonExtractor, Valid: true}, &qV, []pgtype.UUID{}),
AddRow(database.MustToDBUUID(coll.ID), pgtype.UUID{}, repository.QuerytypeJsonExtractor, queryVersion, []pgtype.UUID{}),
)
docSvc := document.New(db)
docSvc := document.New(db, &document.Services{
Collector: collector.New(db),
})
err = docSvc.Sync(ctx, &doc)
assert.Nil(t, err)
}
@@ -85,49 +90,55 @@ func TestSyncDBFail(t *testing.T) {
minCleanVersion := int32(1)
minTextVersion := int32(1)
pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID).
pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(dbJobID).
WillReturnRows(
pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}),
pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}),
)
docSvc := document.New(db)
docSvc := document.New(db, &document.Services{
Collector: collector.New(db),
})
err = docSvc.Sync(ctx, &doc)
assert.EqualError(t, err, "no rows in result set")
pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID).
pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(dbJobID).
WillReturnRows(
pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}).
AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion),
pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}).
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)
docSvc = document.New(db, &document.Services{
Collector: collector.New(db),
})
err = docSvc.Sync(ctx, &doc)
assert.EqualError(t, err, dbErr)
pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID).
pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(dbJobID).
WillReturnRows(
pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}).
AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion),
pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}).
AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion, int32(1), int32(2), []byte("")),
)
pool.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), minCleanVersion, minTextVersion).
WillReturnRows(
pgxmock.NewRows([]string{"id", "queryId", "queryVersion"}),
)
dbErr = "database failure"
pool.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorId).
pool.ExpectQuery("name: ListCollectorQueries :many").WithArgs(dbCollectorId).
WillReturnError(errors.New(dbErr))
docSvc = document.New(db)
docSvc = document.New(db, &document.Services{
Collector: collector.New(db),
})
err = docSvc.Sync(ctx, &doc)
assert.EqualError(t, err, dbErr)
pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID).
pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(dbJobID).
WillReturnRows(
pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}).
AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion),
pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}).
AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion, int32(1), int32(2), []byte("")),
)
qV := int32(1)
reqID := database.MustToDBUUID(uuid.New())
@@ -138,16 +149,18 @@ func TestSyncDBFail(t *testing.T) {
AddRow(resID, reqID, qV),
)
dbErr = "database failure"
pool.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorId).
pool.ExpectQuery("name: ListCollectorQueries :many").WithArgs(dbCollectorId).
WillReturnRows(
pgxmock.NewRows([]string{"collectorId", "queryId", "type", "queryVersion", "requiredIds"}).
AddRow(dbCollectorId, dbQueryID, repository.NullQuerytype{Querytype: repository.QuerytypeJsonExtractor, Valid: true}, &qV, []pgtype.UUID{reqID}).
AddRow(dbCollectorId, reqID, repository.NullQuerytype{Querytype: repository.QuerytypeContextFull, Valid: true}, &qV, []pgtype.UUID{}),
AddRow(dbCollectorId, dbQueryID, repository.QuerytypeJsonExtractor, qV, []pgtype.UUID{reqID}).
AddRow(dbCollectorId, reqID, repository.QuerytypeContextFull, qV, []pgtype.UUID{}),
)
pool.ExpectQuery("name: ListResultValuesByID :many").WithArgs([]pgtype.UUID{resID}).
WillReturnError(errors.New(dbErr))
docSvc = document.New(db)
docSvc = document.New(db, &document.Services{
Collector: collector.New(db),
})
err = docSvc.Sync(ctx, &doc)
assert.EqualError(t, err, dbErr)
}
@@ -177,27 +190,29 @@ func TestSync(t *testing.T) {
minCleanVersion := int32(1)
minTextVersion := int32(1)
pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID).
pool.ExpectQuery("name: GetCollectorByJobID :one").WithArgs(dbJobID).
WillReturnRows(
pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}).
AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion),
pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion", "activeVersion", "latestVersion", "fields"}).
AddRow(dbCollectorId, dbJobID, minCleanVersion, minTextVersion, int32(1), int32(2), []byte("")),
)
pool.ExpectQuery("name: ListResultsByDocumentID :many").WithArgs(database.MustToDBUUID(doc.ID), minCleanVersion, minTextVersion).
WillReturnRows(
pgxmock.NewRows([]string{"id", "queryId", "queryVersion"}),
)
qV := int32(1)
pool.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorId).
pool.ExpectQuery("name: ListCollectorQueries :many").WithArgs(dbCollectorId).
WillReturnRows(
pgxmock.NewRows([]string{"collectorId", "queryId", "type", "queryVersion", "requiredIds"}).
AddRow(dbCollectorId, dbQueryID, repository.NullQuerytype{Querytype: repository.QuerytypeJsonExtractor, Valid: true}, &qV, []pgtype.UUID{}),
AddRow(dbCollectorId, dbQueryID, repository.QuerytypeJsonExtractor, qV, []pgtype.UUID{}),
)
pool.ExpectQuery("name: ListResultValuesByID :many").WithArgs([]pgtype.UUID{}).
WillReturnRows(
pgxmock.NewRows([]string{"id", "queryId", "value"}),
)
docSvc := document.New(db)
docSvc := document.New(db, &document.Services{
Collector: collector.New(db),
})
err = docSvc.Sync(ctx, &doc)
assert.EqualError(t, err, "JSON Extraction requires 1 result")
}
internal/query/processor/parse.go
+3 -3
View File
@@ -65,21 +65,21 @@ func ToDBNullQueryType(t Type) (repository.NullQuerytype, error) {
return repository.NullQuerytype{Querytype: dbType, Valid: true}, nil
}
func ParseDBCollectorQuery(q *repository.GetCollectorQueriesRow) (*Query, error) {
func ParseDBCollectorQuery(q *repository.Collectorquerydependencytree) (*Query, error) {
var reqQueryIDs *[]uuid.UUID
if len(q.Requiredids) > 0 {
ids := database.MustToUUIDArray(q.Requiredids)
reqQueryIDs = &ids
}
qType, err := ParseDBNullType(q.Type)
qType, err := ParseDBType(q.Type)
if err != nil {
return nil, err
}
return &Query{
ID: database.MustToUUID(q.Queryid),
Version: *q.Queryversion,
Version: q.Queryversion,
Type: qType,
RequiredQueryIDs: reqQueryIDs,
}, nil
internal/query/processor/parse_test.go
+5 -6
View File
@@ -11,22 +11,21 @@ import (
)
func TestParseDBCollectorQuery(t *testing.T) {
qV := int32(0)
dbResult := repository.GetCollectorQueriesRow{
dbResult := repository.Collectorquerydependencytree{
Collectorid: pgtype.UUID{},
Queryid: pgtype.UUID{},
Requiredids: []pgtype.UUID{},
Type: repository.NullQuerytype{Valid: true, Querytype: repository.QuerytypeJsonExtractor},
Queryversion: &qV,
Type: repository.QuerytypeJsonExtractor,
Queryversion: 1,
}
value, err := queryprocessor.ParseDBCollectorQuery(&dbResult)
assert.Nil(t, err)
assert.Equal(t, uuid.Nil, value.ID)
assert.Nil(t, value.RequiredQueryIDs)
assert.Equal(t, int32(0), value.Version)
assert.Equal(t, int32(1), value.Version)
assert.Equal(t, queryprocessor.Type(queryprocessor.TypeJsonExtractor), value.Type)
dbResult.Type = repository.NullQuerytype{}
dbResult.Type = repository.Querytype("")
_, err = queryprocessor.ParseDBCollectorQuery(&dbResult)
assert.EqualError(t, err, "invalid database query type")
}
internal/query/queue/create.go
+1 -1
View File
@@ -33,7 +33,7 @@ func (c *Queue) getCollectorQueries(ctx context.Context) error {
id := database.MustToDBUUID(c.collector.ID)
queries, err := c.db.Queries.GetCollectorQueries(ctx, id)
queries, err := c.db.Queries.ListCollectorQueries(ctx, id)
if err != nil {
return err
}
internal/query/queue/create_test.go
+3 -3
View File
@@ -71,13 +71,13 @@ func TestGetCollectorQueries(t *testing.T) {
if q.RequiredQueryIDs != nil {
dbReqIDs = database.MustToDBUUIDArray(*q.RequiredQueryIDs)
}
ty, err := queryprocessor.ToDBNullQueryType(q.Type)
ty, err := queryprocessor.ToDBQueryType(q.Type)
assert.Nil(t, err)
rows = rows.
AddRow(dbCollectorID, dbID, ty, &q.Version, dbReqIDs)
AddRow(dbCollectorID, dbID, ty, q.Version, dbReqIDs)
}
pool.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorID).WillReturnRows(rows)
pool.ExpectQuery("name: ListCollectorQueries :many").WithArgs(dbCollectorID).WillReturnRows(rows)
err = svc.getCollectorQueries(ctx)
assert.Nil(t, err)
internal/query/queue/execute_test.go
+10 -17
View File
@@ -29,19 +29,10 @@ func TestExecute(t *testing.T) {
Queries: queries,
Pool: pool,
}
jobID := uuid.New()
dbJobID := database.MustToDBUUID(jobID)
collectorID := uuid.New()
dbCollectorID := database.MustToDBUUID(collectorID)
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, db, jobID)
assert.Nil(t, err)
coll := collector.Collector{
ID: uuid.New(),
}
dbCollectorID := database.MustToDBUUID(coll.ID)
queryOneID := uuid.New()
queryOneVersion := int32(1)
@@ -74,13 +65,13 @@ func TestExecute(t *testing.T) {
if q.RequiredQueryIDs != nil {
dbReqIDs = database.MustToDBUUIDArray(*q.RequiredQueryIDs)
}
ty, err := queryprocessor.ToDBNullQueryType(q.Type)
ty, err := queryprocessor.ToDBQueryType(q.Type)
assert.Nil(t, err)
rows = rows.
AddRow(dbCollectorID, dbID, ty, &q.Version, dbReqIDs)
AddRow(dbCollectorID, dbID, ty, q.Version, dbReqIDs)
}
pool.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorID).WillReturnRows(rows)
pool.ExpectQuery("name: ListCollectorQueries :many").WithArgs(dbCollectorID).WillReturnRows(rows)
contextResultID := uuid.New()
results := []*result.Result{
@@ -102,7 +93,9 @@ func TestExecute(t *testing.T) {
cleanVersion := int32(1)
textVersion := int32(1)
q, err := queryqueue.New(ctx, db, coll, results, docID, cleanVersion, textVersion)
q, err := queryqueue.New(ctx, &queryqueue.NewConfig{
db, &coll, results, docID, cleanVersion, textVersion,
})
assert.Nil(t, err)
assert.Equal(t, expectedQueries, q.GetQueue())
internal/query/queue/service.go
+17 -8
View File
@@ -16,19 +16,28 @@ type Queue struct {
results []*result.Result
collector *collector.Collector
db *database.Connection
documentId uuid.UUID
cleanVersion int32
textVersion int32
documentId uuid.UUID
}
func New(ctx context.Context, db *database.Connection, coll *collector.Collector, results []*result.Result, docId uuid.UUID, cleanVersion int32, textVersion int32) (*Queue, error) {
type NewConfig struct {
DB *database.Connection
Collector *collector.Collector
Results []*result.Result
DocumentID uuid.UUID
CleanVersion int32
TextVersion int32
}
func New(ctx context.Context, cfg *NewConfig) (*Queue, error) {
queue := Queue{
db: db,
results: results,
collector: coll,
documentId: docId,
cleanVersion: cleanVersion,
textVersion: textVersion,
db: cfg.DB,
results: cfg.Results,
collector: cfg.Collector,
documentId: cfg.DocumentID,
cleanVersion: cfg.CleanVersion,
textVersion: cfg.TextVersion,
}
err := queue.getCollectorQueries(ctx)
internal/query/queue/service_test.go
+18 -31
View File
@@ -29,19 +29,11 @@ func TestService(t *testing.T) {
Queries: queries,
Pool: pool,
}
jobID := uuid.New()
dbJobID := database.MustToDBUUID(jobID)
collectorID := uuid.New()
dbCollectorID := database.MustToDBUUID(collectorID)
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, db, jobID)
assert.Nil(t, err)
coll := collector.Collector{
ID: uuid.New(),
}
dbCollectorID := database.MustToDBUUID(coll.ID)
queryOneID := uuid.New()
queryOneVersion := int32(1)
@@ -74,13 +66,13 @@ func TestService(t *testing.T) {
if q.RequiredQueryIDs != nil {
dbReqIDs = database.MustToDBUUIDArray(*q.RequiredQueryIDs)
}
ty, err := queryprocessor.ToDBNullQueryType(q.Type)
ty, err := queryprocessor.ToDBQueryType(q.Type)
assert.Nil(t, err)
rows = rows.
AddRow(dbCollectorID, dbID, ty, &q.Version, dbReqIDs)
AddRow(dbCollectorID, dbID, ty, q.Version, dbReqIDs)
}
pool.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorID).WillReturnRows(rows)
pool.ExpectQuery("name: ListCollectorQueries :many").WithArgs(dbCollectorID).WillReturnRows(rows)
contextResultID := uuid.New()
results := []*result.Result{
@@ -102,7 +94,9 @@ func TestService(t *testing.T) {
cleanVersion := int32(1)
textVersion := int32(1)
q, err := queryqueue.New(ctx, db, coll, results, docID, cleanVersion, textVersion)
q, err := queryqueue.New(ctx, &queryqueue.NewConfig{
db, &coll, results, docID, cleanVersion, textVersion,
})
assert.Nil(t, err)
assert.Equal(t, expectedQueries, q.GetQueue())
}
@@ -119,22 +113,13 @@ func TestQueueFail(t *testing.T) {
Queries: queries,
Pool: pool,
}
jobID := uuid.New()
dbJobID := database.MustToDBUUID(jobID)
collectorID := uuid.New()
dbCollectorID := database.MustToDBUUID(collectorID)
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, db, jobID)
assert.Nil(t, err)
coll := collector.Collector{
ID: uuid.New(),
}
dbCollectorID := database.MustToDBUUID(coll.ID)
dbErr := "database failure"
pool.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorID).
pool.ExpectQuery("name: ListCollectorQueries :many").WithArgs(dbCollectorID).
WillReturnError(errors.New(dbErr))
results := []*result.Result{}
@@ -143,6 +128,8 @@ func TestQueueFail(t *testing.T) {
cleanVersion := int32(1)
textVersion := int32(1)
_, err = queryqueue.New(ctx, db, coll, results, docID, cleanVersion, textVersion)
_, err = queryqueue.New(ctx, &queryqueue.NewConfig{
db, &coll, results, docID, cleanVersion, textVersion,
})
assert.EqualError(t, err, dbErr)
}