Michael McGuinness
100 lines
2.2 KiB
Go
100 lines
2.2 KiB
Go
package queryQueue
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
contextfull "queryorchestration/internal/contextFull"
|
|
"queryorchestration/internal/database"
|
|
"queryorchestration/internal/database/repository"
|
|
jsonextractor "queryorchestration/internal/jsonExtractor"
|
|
"queryorchestration/internal/query"
|
|
"queryorchestration/internal/result"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|
)
|
|
|
|
func (q *Queue) Execute(ctx context.Context) error {
|
|
if q.unsyncedQueue == nil {
|
|
return nil
|
|
}
|
|
|
|
for _, query := range *q.unsyncedQueue {
|
|
err := q.executeQuery(ctx, query)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
q.unsyncedQueue = nil
|
|
|
|
return nil
|
|
}
|
|
|
|
func (q *Queue) executeQuery(ctx context.Context, qu query.QueryRow) 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 {
|
|
var queryVersion int32
|
|
for _, entry := range *q.collectorQueries {
|
|
if entry.ID == id {
|
|
queryVersion = entry.Version
|
|
break
|
|
}
|
|
}
|
|
|
|
for _, entry := range *q.results {
|
|
if entry.QueryID == id && entry.QueryVersion == queryVersion {
|
|
resultIDs[index] = database.MustToDBUUID(entry.ID)
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
values, err := q.db.ListResultValuesByID(ctx, resultIDs)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
cleanValues := make([]result.Value, len(values))
|
|
for index, r := range values {
|
|
cleanValue, err := q.getResultValue(&r)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
cleanValues[index] = cleanValue
|
|
}
|
|
|
|
err = q.setResult(ctx, qu, &cleanValues)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (q *Queue) getResultValue(res *repository.ListResultValuesByIDRow) (result.Value, error) {
|
|
var queryType query.Type
|
|
for _, qu := range *q.collectorQueries {
|
|
if qu.ID == database.MustToUUID(res.Queryid) {
|
|
queryType = qu.Type
|
|
}
|
|
}
|
|
|
|
switch queryType {
|
|
case query.TypeJsonExtractor:
|
|
return jsonextractor.NewResult(res.Value), nil
|
|
case query.TypeContextFull:
|
|
return contextfull.NewResult(res.Value), nil
|
|
default:
|
|
return nil, fmt.Errorf("attempting to process invalid query type")
|
|
}
|
|
}
|