query-orchestration/internal/query/service.go

package query

import (
	"context"
	"gotemplate/internal/collector"
	"gotemplate/internal/database"
	"gotemplate/internal/database/repository"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgtype"
)

type QueryType int

const (
	QueryTypeJsonExtractor = iota
)

type Query struct {
	ID              uuid.UUID
	Type            QueryType
	RequiredQueryID uuid.UUID
	Version         int32
}

type Result struct {
	ID           uuid.UUID
	QueryID      uuid.UUID
	QueryVersion int32
}

type ResultValue struct {
	ID      uuid.UUID
	QueryID uuid.UUID
	Value   string
}

type Queue struct {
	unsyncedQueue    *[]Query
	collectorQueries *[]Query
	results          *[]Result
	collector        *collector.Collector
	db               *repository.Queries
	cleanVersion     int32
	textVersion      int32
	documentId       uuid.UUID
}

func NewQueue(ctx context.Context, db *repository.Queries, coll *collector.Collector, results *[]Result, docId uuid.UUID, cleanVersion int32, textVersion int32) (*Queue, error) {
	queue := Queue{
		db:           db,
		results:      results,
		collector:    coll,
		documentId:   docId,
		cleanVersion: cleanVersion,
		textVersion:  textVersion,
	}

	err := queue.getCollectorQueries(ctx)
	if err != nil {
		return nil, err
	}

	queue.getUnsyncedQueries()

	return &queue, nil
}

func (q *Queue) GetQueue() []Query {
	return *q.unsyncedQueue
}

func (q *Queue) getUnsyncedQueries() {
	for _, query := range *q.collectorQueries {
		isSynced := false
		for _, result := range *q.results {
			if result.QueryID != query.ID || result.QueryVersion != query.Version {
				continue
			}
			isSynced = true
			break
		}
		if isSynced {
			continue
		}

		q.Add(&query)
	}
}

func (c *Queue) getCollectorQueries(ctx context.Context) error {
	if c.collectorQueries != nil {
		return nil
	}

	id, err := database.ToDBUUID(c.collector.ID)
	if err != nil {
		return err
	}

	queries, err := c.db.GetCollectorQueries(ctx, id)
	if err != nil {
		return err
	}

	cleanQueries := make([]Query, len(queries))
	for index, dbQuery := range queries {
		cleanQuery, err := ParseDBQuery(&dbQuery)
		if err != nil {
			return err
		}

		cleanQueries[index] = *cleanQuery
	}

	c.collectorQueries = &cleanQueries

	return nil
}

func (q *Queue) Add(query *Query) {
	dependentQueries := []Query{}
	requiredIndex := -1

	if q.unsyncedQueue == nil {
		q.unsyncedQueue = &[]Query{}
	} else {
		for index, entry := range *q.unsyncedQueue {
			if entry.ID == query.ID {
				return
			}
			if entry.ID == query.RequiredQueryID {
				requiredIndex = index
			}
		}
	}

	for _, entry := range *q.collectorQueries {
		if entry.RequiredQueryID == query.ID {
			dependentQueries = append(dependentQueries, entry)
		}
	}

	if requiredIndex != -1 {
		*q.unsyncedQueue = append((*q.unsyncedQueue)[:requiredIndex], append([]Query{*query}, (*q.unsyncedQueue)[requiredIndex:]...)...)
	} else {
		*q.unsyncedQueue = append([]Query{*query}, *q.unsyncedQueue...)
	}

	for _, entry := range dependentQueries {
		q.Add(&entry)
	}
}

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, query Query) error {
	requiredQueryIDs := []uuid.UUID{}
	for _, entry := range *q.collectorQueries {
		if entry.ID == query.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([]ResultValue, len(values))
	for _, result := range values {
		cleanValue := q.parseResultValue(result)
		cleanValues = append(cleanValues, cleanValue)
	}

	err = q.getResult(ctx, query, &cleanValues)
	if err != nil {
		return err
	}

	return nil
}

func (q *Queue) parseResultValue(result repository.ListResultValuesByIDRow) ResultValue {
	return ResultValue{
		ID:      database.MustToUUID(result.ID),
		QueryID: database.MustToUUID(result.Queryid),
		Value:   result.Value,
	}
}

func (q *Queue) getResult(ctx context.Context, query Query, resultValues *[]ResultValue) error {
	// TODO - execute query
	value := "the value"

	id := uuid.New()

	err := q.db.SetResult(ctx, repository.SetResultParams{
		ID:           database.MustToDBUUID(id),
		Queryid:      database.MustToDBUUID(query.ID),
		Documentid:   database.MustToDBUUID(q.documentId),
		Value:        value,
		Cleanversion: q.cleanVersion,
		Textversion:  q.textVersion,
		Queryversion: query.Version,
	})
	if err != nil {
		return err
	}

	*q.results = append(*q.results, Result{
		ID:           id,
		QueryID:      query.ID,
		QueryVersion: query.Version,
	})

	return nil
}