distributor/internal/publish/plan.go

package publish

import (
	"context"
	"fmt"
	"time"

	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
	"gitea.maximumdirect.net/eric/distributor/internal/config"
	"gitea.maximumdirect.net/eric/distributor/internal/state"
	"gitea.maximumdirect.net/eric/distributor/internal/storage"
	"gitea.maximumdirect.net/eric/distributor/internal/transform"
)

type Action string

const (
	ActionPublishNew           Action = "publish_new"
	ActionReplaceOlder         Action = "replace_older"
	ActionReplaceConflict      Action = "replace_conflict"
	ActionReplaceNewer         Action = "replace_newer"
	ActionSkipSame             Action = "skip_same"
	ActionSkipDestinationNewer Action = "skip_destination_newer"
	ActionFailConflict         Action = "fail_conflict"
	ActionFailUnmanaged        Action = "fail_unmanaged"
	ActionForceReplace         Action = "force_replace"
	ActionReplaceTakeover      Action = "replace_takeover"
	ActionUpsertAdditive       Action = "upsert_additive"
	ActionReplaceCatalog       Action = "replace_catalog"
)

type Request struct {
	PipelineID            string
	DestinationID         string
	SourceBundle          bundle.Bundle
	SourceBackend         storage.Backend
	DestinationBackend    storage.Backend
	DestinationBundlePath string
	PathMapping           string
	Publish               config.PublishPolicy
	Transform             config.Transform
	Links                 *config.Links
	Workflow              string
	Transformers          TransformerResolver
	DistributorVersion    string
	Force                 bool
	Now                   time.Time
}

type TransformerResolver interface {
	Get(name string) (transform.Transformer, bool)
}

type Output struct {
	SourcePath      string
	DestinationPath string
	Kind            string
	Transform       string
	URL             string
	Data            []byte
	SHA256          string
	Size            int64
}

type Plan struct {
	PipelineID             string
	DestinationID          string
	BundleID               string
	BundlePath             string
	DestinationBundlePath  string
	PathMapping            string
	Action                 Action
	Reason                 string
	Force                  bool
	PrimaryURL             string
	Workflow               string
	OwnerScope             state.OwnerScope
	Outputs                []Output
	ExistingCatalog        *state.CatalogState
	SupersededLegacy       *state.SupersededLegacyState
	CatalogOutputsToWrite  []state.CatalogOutputFile
	CatalogOutputsToRetain []state.CatalogOutputFile
	CatalogOutputsToDelete []state.CatalogOutputFile
	ClearDestinationRoot   bool

	// Retained while the executor is migrated to catalog state.
	StateMode             string
	Reconciliation        config.ReconciliationPolicy
	TakeoverMode          string
	ExistingState         *state.DistributorState
	ExistingSharedRoot    *state.SharedRootState
	OtherOwnerOutputs     []state.SharedRootOutputFile
	TakenOverOwnerOutputs []state.SharedRootOutputFile
	RetainedOwnerOutputs  []state.SharedRootOutputFile
	OwnerOutputsToDelete  []state.SharedRootOutputFile
	OwnerOutputsToWrite   []Output
}

type catalogPlanDetails struct {
	Action                 Action
	Reason                 string
	CatalogOutputsToWrite  []state.CatalogOutputFile
	CatalogOutputsToRetain []state.CatalogOutputFile
	CatalogOutputsToDelete []state.CatalogOutputFile
	ClearDestinationRoot   bool
}

func Build(ctx context.Context, req Request) (Plan, error) {
	if err := validateRequest(req); err != nil {
		return Plan{}, err
	}
	outputs, err := PlanOutputs(ctx, req)
	if err != nil {
		return Plan{}, err
	}
	outputs, primaryURL, err := PlanLinks(req, outputs)
	if err != nil {
		return Plan{}, err
	}
	status, err := inspectDestination(ctx, req.DestinationBackend, req.DestinationBundlePath)
	if err != nil {
		return Plan{}, err
	}
	workflow := normalizeWorkflow(req.Workflow)
	scope := state.CurrentOwnerScope(req.PipelineID, req.DestinationID)
	now := requestTime(req)
	plan := Plan{
		PipelineID:            req.PipelineID,
		DestinationID:         req.DestinationID,
		BundleID:              req.SourceBundle.Manifest.ID,
		BundlePath:            req.SourceBundle.RootRelativePath,
		DestinationBundlePath: req.DestinationBundlePath,
		PathMapping:           req.PathMapping,
		PrimaryURL:            primaryURL,
		Workflow:              workflow,
		OwnerScope:            scope,
		Outputs:               outputs,
		ExistingCatalog:       status.Catalog,
		SupersededLegacy:      status.SupersededLegacy,
	}
	if status.StateErr != nil {
		plan.Action = ActionFailConflict
		plan.Reason = status.StateErr.Error()
		return plan, fmt.Errorf("%s: %s", plan.Action, plan.Reason)
	}

	var details catalogPlanDetails
	switch {
	case status.Catalog != nil:
		details, err = planExistingCatalog(ctx, req, *status.Catalog, outputs, workflow, scope, now)
	case status.SupersededLegacy != nil:
		details = planSupersededLegacy(req, outputs, workflow, scope, now)
	default:
		details, err = planWithoutCatalog(ctx, req, outputs, workflow, scope, now)
	}
	plan.Action = details.Action
	plan.Reason = details.Reason
	plan.CatalogOutputsToWrite = details.CatalogOutputsToWrite
	plan.CatalogOutputsToRetain = details.CatalogOutputsToRetain
	plan.CatalogOutputsToDelete = details.CatalogOutputsToDelete
	plan.ClearDestinationRoot = details.ClearDestinationRoot
	if err != nil {
		return plan, err
	}
	return plan, nil
}

func validateRequest(req Request) error {
	if req.PipelineID == "" {
		return fmt.Errorf("pipeline id is required")
	}
	if req.DestinationID == "" {
		return fmt.Errorf("destination id is required")
	}
	if req.SourceBackend == nil {
		return fmt.Errorf("source backend is required")
	}
	if req.DestinationBackend == nil {
		return fmt.Errorf("destination backend is required")
	}
	if err := config.ValidatePublishTransformPolicy(req.Publish, req.Transform); err != nil {
		return fmt.Errorf("publish/transform policy: %w", err)
	}
	switch normalizeWorkflow(req.Workflow) {
	case config.WorkflowAdditive, config.WorkflowReplacement:
	default:
		return fmt.Errorf("destination.workflow must be %s or %s", config.WorkflowAdditive, config.WorkflowReplacement)
	}
	return nil
}

func normalizeWorkflow(workflow string) string {
	if workflow == "" {
		return config.WorkflowAdditive
	}
	return workflow
}

func requestTime(req Request) time.Time {
	if req.Now.IsZero() {
		return time.Now().UTC()
	}
	return req.Now.UTC()
}

func planExistingCatalog(ctx context.Context, req Request, catalog state.CatalogState, outputs []Output, workflow string, scope state.OwnerScope, now time.Time) (catalogPlanDetails, error) {
	if err := rejectCatalogUnmanagedCollisions(ctx, req.DestinationBackend, req.DestinationBundlePath, catalog.Outputs, outputs); err != nil {
		return catalogPlanDetails{
			Action: ActionFailUnmanaged,
			Reason: err.Error(),
		}, fmt.Errorf("%s: %s", ActionFailUnmanaged, err)
	}
	planned := outputPathSet(outputs)
	details := catalogPlanDetails{
		Action:                actionForWorkflow(workflow),
		CatalogOutputsToWrite: catalogOutputsForPlan(req, outputs, catalog.Outputs, scope, now),
	}
	for _, output := range catalog.Outputs {
		if _, exists := planned[output.Path]; exists {
			continue
		}
		if workflow == config.WorkflowReplacement && output.PipelineID == scope.PipelineID && output.DestinationID == scope.DestinationID {
			details.CatalogOutputsToDelete = append(details.CatalogOutputsToDelete, output)
			continue
		}
		details.CatalogOutputsToRetain = append(details.CatalogOutputsToRetain, output)
	}
	return details, nil
}

func planSupersededLegacy(req Request, outputs []Output, workflow string, scope state.OwnerScope, now time.Time) catalogPlanDetails {
	details := catalogPlanDetails{
		Action:                actionForWorkflow(workflow),
		CatalogOutputsToWrite: catalogOutputsForPlan(req, outputs, nil, scope, now),
	}
	if workflow == config.WorkflowReplacement {
		details.ClearDestinationRoot = true
	}
	return details
}

func planWithoutCatalog(ctx context.Context, req Request, outputs []Output, workflow string, scope state.OwnerScope, now time.Time) (catalogPlanDetails, error) {
	if err := rejectCatalogUnmanagedCollisions(ctx, req.DestinationBackend, req.DestinationBundlePath, nil, outputs); err != nil {
		return catalogPlanDetails{
			Action: ActionFailUnmanaged,
			Reason: err.Error(),
		}, fmt.Errorf("%s: %s", ActionFailUnmanaged, err)
	}
	return catalogPlanDetails{
		Action:                ActionPublishNew,
		CatalogOutputsToWrite: catalogOutputsForPlan(req, outputs, nil, scope, now),
	}, nil
}

func actionForWorkflow(workflow string) Action {
	if workflow == config.WorkflowReplacement {
		return ActionReplaceCatalog
	}
	return ActionUpsertAdditive
}

func catalogOutputsForPlan(req Request, outputs []Output, existing []state.CatalogOutputFile, scope state.OwnerScope, now time.Time) []state.CatalogOutputFile {
	files := make([]state.CatalogOutputFile, 0, len(outputs))
	source := state.CatalogSourceIdentity{
		ID:      req.SourceBundle.Manifest.ID,
		Digest:  req.SourceBundle.Manifest.Digest,
		Created: req.SourceBundle.Manifest.Created,
	}
	for _, output := range outputs {
		createdAt := now
		if existingOutput, ok := state.FindCatalogOutputByPath(existing, output.DestinationPath); ok {
			createdAt = existingOutput.CreatedAt
		}
		file := state.CatalogOutputFile{
			Path:          output.DestinationPath,
			PipelineID:    scope.PipelineID,
			DestinationID: scope.DestinationID,
			Source:        source,
			Kind:          output.Kind,
			URL:           output.URL,
			SHA256:        output.SHA256,
			Size:          output.Size,
			CreatedAt:     createdAt,
			UpdatedAt:     now,
		}
		if output.Kind == state.OutputKindGenerated {
			file.SourcePath = output.SourcePath
			file.Transform = output.Transform
		}
		files = append(files, file)
	}
	return files
}

func rejectCatalogUnmanagedCollisions(ctx context.Context, backend storage.Backend, bundlePath string, existing []state.CatalogOutputFile, outputs []Output) error {
	managed := catalogOutputPathSet(existing)
	for _, output := range outputs {
		if _, exists := managed[output.DestinationPath]; exists {
			continue
		}
		destinationPath, err := storage.Join(bundlePath, output.DestinationPath)
		if err != nil {
			return err
		}
		if _, err := backend.Stat(ctx, destinationPath); err == nil {
			return fmt.Errorf("destination output path %s exists but is not managed by catalog state", storage.DisplayPath(output.DestinationPath))
		} else if !storage.IsNotFound(err) {
			return err
		}
	}
	return nil
}

func outputPaths(outputs []Output) []string {
	paths := make([]string, 0, len(outputs))
	for _, output := range outputs {
		paths = append(paths, output.DestinationPath)
	}
	return paths
}

func outputPathSet(outputs []Output) map[string]struct{} {
	paths := make(map[string]struct{}, len(outputs))
	for _, output := range outputs {
		paths[output.DestinationPath] = struct{}{}
	}
	return paths
}

func catalogOutputPathSet(outputs []state.CatalogOutputFile) map[string]struct{} {
	paths := make(map[string]struct{}, len(outputs))
	for _, output := range outputs {
		paths[output.Path] = struct{}{}
	}
	return paths
}

func normalizeReconciliation(policy config.ReconciliationPolicy) config.ReconciliationPolicy {
	if policy.Mode == "" {
		policy.Mode = config.ReconciliationModeReplace
	}
	return policy
}

func isReconciliationReplacementAction(action Action) bool {
	return action == ActionReplaceOlder || action == ActionReplaceNewer
}

func sharedRootOutputPathSet(outputs []state.SharedRootOutputFile) map[string]struct{} {
	paths := make(map[string]struct{}, len(outputs))
	for _, output := range outputs {
		paths[output.Path] = struct{}{}
	}
	return paths
}