query-orchestration/internal/query/update.go

package query

import (
	"context"
	"errors"
	"fmt"
	"queryorchestration/internal/database"
	"queryorchestration/internal/database/repository"
	queryprocessor "queryorchestration/internal/query/processor"
	contextfull "queryorchestration/internal/query/types/contextFull"
	jsonextractor "queryorchestration/internal/query/types/jsonExtractor"

	"github.com/google/uuid"
)

func (s *Service) Update(ctx context.Context, entity *queryprocessor.Update) error {
	current, err := s.Get(ctx, entity.ID)
	if err != nil {
		return err
	}

	err = s.normalizeUpdate(ctx, current, entity)
	if err != nil {
		return err
	}

	err = s.submitUpdate(ctx, current, entity)
	if err != nil {
		return err
	}

	return nil
}

func (s *Service) normalizeUpdate(ctx context.Context, current *Query, entity *queryprocessor.Update) error {
	err := s.normalizeActiveVersion(current, entity)
	if err != nil {
		return err
	}

	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
	}

	err = validator.Validate(ctx, ParseQuery(current), entity)
	if err != nil {
		return err
	}

	return nil
}

func (s *Service) submitUpdate(ctx context.Context, current *Query, entity *queryprocessor.Update) error {
	tx, err := s.db.Pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer func() {
		_ = tx.Rollback(ctx)
	}()

	qtx := s.db.Queries.WithTx(tx)

	latestVersion := current.LatestVersion + 1
	id := database.MustToDBUUID(entity.ID)

	addIDs := getSetDifference(entity.RequiredQueryIDs, current.RequiredQueryIDs)
	for _, qID := range addIDs {
		err = qtx.AddRequiredQuery(ctx, &repository.AddRequiredQueryParams{
			Queryid:         id,
			Requiredqueryid: database.MustToDBUUID(qID),
			Addedversion:    latestVersion,
		})
		if err != nil {
			return err
		}
	}

	removeIDs := getSetDifference(current.RequiredQueryIDs, entity.RequiredQueryIDs)
	for _, qID := range removeIDs {
		err = qtx.RemoveRequiredQuery(ctx, &repository.RemoveRequiredQueryParams{
			Queryid:         id,
			Requiredqueryid: database.MustToDBUUID(qID),
			Removedversion:  &latestVersion,
		})
		if err != nil {
			return err
		}
	}

	if entity.Config != nil && *entity.Config != "" {
		err = qtx.RemoveQueryConfig(ctx, &repository.RemoveQueryConfigParams{
			Queryid:        id,
			Removedversion: &latestVersion,
		})
		if err != nil {
			return err
		}

		err = qtx.AddQueryConfig(ctx, &repository.AddQueryConfigParams{
			Queryid:      id,
			Config:       []byte(*entity.Config),
			Addedversion: latestVersion,
		})
		if err != nil {
			return err
		}
	}

	activeVersion := entity.ActiveVersion
	if activeVersion == nil {
		activeVersion = &current.ActiveVersion
	}

	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
	}

	return nil
}

func getSetDifference(setA *[]uuid.UUID, setB *[]uuid.UUID) []uuid.UUID {
	if setA == nil {
		return []uuid.UUID{}
	} else if setB == nil {
		return *setA
	}

	diff := []uuid.UUID{}
	for _, q := range *setA {
		isFound := false
		for _, eq := range *setB {
			if q == eq {
				isFound = true
				break
			}
		}
		if !isFound {
			diff = append(diff, q)
		}
	}

	return diff
}

func (s *Service) getUpdator(qType queryprocessor.Type) (queryprocessor.Updator, error) {
	switch qType {
	case queryprocessor.TypeJsonExtractor:
		return jsonextractor.NewUpdator(s.db), nil
	case queryprocessor.TypeContextFull:
		return contextfull.NewUpdator(s.db), nil
	default:
		return nil, fmt.Errorf("attempting to process invalid query type")
	}
}

func ParseQuery(q *Query) *queryprocessor.Query {
	return &queryprocessor.Query{
		ID:               q.ID,
		Type:             q.Type,
		Version:          q.ActiveVersion,
		RequiredQueryIDs: q.RequiredQueryIDs,
		Config:           q.Config,
	}
}