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) (uuid.UUID, error) {
	dbparams, err := s.getCreateParams(ctx, params)
	if err != nil {
		return uuid.Nil, err
	}

	id, err := s.submitCreate(ctx, dbparams)
	if err != nil {
		return uuid.Nil, err
	}

	return id, nil
}

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.Document.IsValidCleanVersion(*minClean)
		if err != nil {
			return nil, err
		}
	}

	minText := params.MinTextVersion
	if minText != nil {
		err := s.svc.Document.IsValidTextVersion(*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) (uuid.UUID, error) {
	var dbID pgtype.UUID
	err := s.cfg.ExecuteDBTransaction(ctx, func(ctx context.Context, qtx *repository.Queries) error {
		dID, err := qtx.CreateCollector(ctx, params.JobID)
		if err != nil {
			return err
		}
		dbID = dID
		latestVersion := int32(1)

		if params.MinCleanVersion != nil || params.MinTextVersion != nil {
			err = qtx.AddCollectorCodeVersion(ctx, &repository.AddCollectorCodeVersionParams{
				Collectorid:     dbID,
				Mincleanversion: validation.GetUpdatedValue(1, params.MinCleanVersion),
				Mintextversion:  validation.GetUpdatedValue(1, params.MinTextVersion),
				Addedversion:    latestVersion,
			})
			if err != nil {
				return err
			}
		}

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

		return nil
	})
	if err != nil {
		return uuid.Nil, err
	}

	id, err := database.ToUUID(dbID)
	if err != nil {
		return uuid.Nil, err
	}

	return id, nil
}