query-orchestration/internal/queryQueue/execute.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")
	}
}