query-orchestration/internal/job/collector/create.go

package collector

import (
	"context"
	"errors"

	"queryorchestration/internal/database"
	"queryorchestration/internal/database/repository"
	"queryorchestration/internal/validation"

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

type CreateParams struct {
	JobID           uuid.UUID
	MinCleanVersion *int32
	MinTextVersion  *int32
	Fields          *map[string]uuid.UUID
}

func (s *Service) Create(ctx context.Context, params *CreateParams) error {
	dbparams, err := s.getCreateParams(ctx, params)
	if err != nil {
		return err
	}

	return s.submitCreate(ctx, dbparams)
}

type dbCreateParams struct {
	JobID           pgtype.UUID
	MinCleanVersion *int32
	MinTextVersion  *int32
	Fields          *map[string]pgtype.UUID
}

func (s *Service) getCreateParams(ctx context.Context, params *CreateParams) (*dbCreateParams, error) {
	minClean := params.MinCleanVersion
	if minClean != nil {
		err := s.svc.CleanVersion.IsValidVersion(*minClean)
		if err != nil {
			return nil, err
		}
	}

	minText := params.MinTextVersion
	if minText != nil {
		err := s.svc.TextVersion.IsValidVersion(*minText)
		if err != nil {
			return nil, err
		}
	}

	fields, err := s.NormalizeFieldsToDB(ctx, params.Fields)
	if err != nil {
		return nil, err
	}

	return &dbCreateParams{
		JobID:           database.MustToDBUUID(params.JobID),
		MinCleanVersion: minClean,
		MinTextVersion:  minText,
		Fields:          fields,
	}, nil
}

func (s *Service) NormalizeFieldsToDB(ctx context.Context, ofields *map[string]uuid.UUID) (*map[string]pgtype.UUID, error) {
	if ofields == nil || *ofields == nil {
		return nil, nil
	} else if len(*ofields) == 0 {
		return nil, nil
	}

	dbm := map[string]pgtype.UUID{}
	for name, id := range *ofields {
		dbm[name] = database.MustToDBUUID(id)
	}

	dbids := []pgtype.UUID{}
	for _, id := range dbm {
		dbids = append(dbids, id)
	}

	dedup := validation.DeduplicateArray(dbids)

	if len(dedup) != len(dbids) {
		return nil, errors.New("duplicate output fields")
	}

	exist, err := s.cfg.GetDBQueries().AllQueriesExist(ctx, dbids)
	if err != nil {
		return nil, err
	} else if !exist {
		return nil, errors.New("not all required ids are present")
	}

	return &dbm, nil
}

func (s *Service) submitCreate(ctx context.Context, params *dbCreateParams) error {
	return s.cfg.ExecuteDBTransaction(ctx, func(ctx context.Context, qtx *repository.Queries) error {
		version, err := qtx.AddLatestCollectorVersion(ctx, params.JobID)
		if err != nil {
			return err
		}

		err = qtx.SetActiveCollectorVersion(ctx, &repository.SetActiveCollectorVersionParams{
			Jobid:     params.JobID,
			Versionid: version,
		})
		if err != nil {
			return err
		}

		if params.MinCleanVersion != nil {
			err = qtx.SetCollectorCleanVersion(ctx, &repository.SetCollectorCleanVersionParams{
				Jobid:        params.JobID,
				Versionid:    *params.MinCleanVersion,
				Addedversion: version,
			})
			if err != nil {
				return err
			}
		}

		if params.MinTextVersion != nil {
			err = qtx.SetCollectorTextVersion(ctx, &repository.SetCollectorTextVersionParams{
				Jobid:        params.JobID,
				Versionid:    *params.MinTextVersion,
				Addedversion: version,
			})
			if err != nil {
				return err
			}
		}

		if params.Fields != nil {
			for key, field := range *params.Fields {
				err = qtx.AddCollectorQuery(ctx, &repository.AddCollectorQueryParams{
					Jobid:        params.JobID,
					Name:         key,
					Queryid:      field,
					Addedversion: version,
				})
				if err != nil {
					return err
				}
			}
		}

		return nil
	})
}