distributor/internal/publish/execute.go

package publish

import (
	"context"
	"encoding/json"
	"fmt"
	"sort"

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

func Execute(ctx context.Context, req Request, plan Plan) error {
	switch plan.Action {
	case ActionSkipSame:
		return nil
	case ActionPublishNew, ActionUpsertAdditive, ActionReplaceCatalog, ActionForceReplace:
		return executeCatalog(ctx, req, plan)
	default:
		return fmt.Errorf("cannot execute action %s: %s", plan.Action, plan.Reason)
	}
}

func executeCatalog(ctx context.Context, req Request, plan Plan) error {
	if plan.Action == ActionForceReplace {
		if err := req.DestinationBackend.DeletePrefix(ctx, req.DestinationBundlePath, storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil {
			return err
		}
		if err := ensureDestinationEmpty(ctx, req.DestinationBackend, req.DestinationBundlePath); err != nil {
			return err
		}
	} else if plan.Action == ActionReplaceCatalog {
		if plan.ClearDestinationRoot {
			if err := req.DestinationBackend.DeletePrefix(ctx, req.DestinationBundlePath, storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil {
				return err
			}
			if err := ensureDestinationEmpty(ctx, req.DestinationBackend, req.DestinationBundlePath); err != nil {
				return err
			}
		} else if len(plan.CatalogOutputsToDelete) > 0 {
			if err := req.DestinationBackend.DeleteManagedOutputs(ctx, req.DestinationBundlePath, catalogOutputPaths(plan.CatalogOutputsToDelete), storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true}); err != nil {
				return err
			}
		}
	}

	writtenOutputs := make([]Output, 0, len(plan.Outputs))
	newOutputs := make([]Output, 0, len(plan.Outputs))
	cleanup := func() {
		outputs := writtenOutputs
		if plan.Action == ActionUpsertAdditive {
			outputs = newOutputs
		}
		_ = req.DestinationBackend.DeleteManagedOutputs(ctx, req.DestinationBundlePath, ManagedOutputPaths(outputs), storage.DeleteOptions{IgnoreMissing: true, PruneEmptyDirs: true})
	}
	for _, output := range plan.Outputs {
		destinationPath, err := storage.Join(req.DestinationBundlePath, output.DestinationPath)
		if err != nil {
			cleanup()
			return err
		}
		created, err := catalogWriteCreatesOutput(ctx, req.DestinationBackend, destinationPath)
		if err != nil {
			cleanup()
			return err
		}
		data := output.Data
		if output.Kind == state.OutputKindSource {
			sourcePath, err := storage.Join(req.SourceBundle.RootRelativePath, output.SourcePath)
			if err != nil {
				cleanup()
				return err
			}
			data, err = req.SourceBackend.ReadFile(ctx, sourcePath)
			if err != nil {
				cleanup()
				return err
			}
		}
		if _, err := req.DestinationBackend.WriteFile(ctx, destinationPath, data, storage.WriteOptions{Overwrite: catalogOutputOverwriteAllowed(plan, output), PreferAtomic: true}); err != nil {
			cleanup()
			return err
		}
		writtenOutputs = append(writtenOutputs, output)
		if created {
			newOutputs = append(newOutputs, output)
		}
	}

	catalogState := catalogStateForPlan(req, plan)
	if err := state.ValidateCatalog(catalogState); err != nil {
		cleanup()
		return err
	}
	data, err := json.MarshalIndent(catalogState, "", "  ")
	if err != nil {
		cleanup()
		return err
	}
	data = append(data, '\n')
	statePath, err := storage.StatePath(req.DestinationBundlePath)
	if err != nil {
		cleanup()
		return err
	}
	if _, err := req.DestinationBackend.WriteFile(ctx, statePath, data, storage.WriteOptions{Overwrite: catalogStateWriteOverwrites(plan), PreferAtomic: true}); err != nil {
		cleanup()
		return err
	}
	return nil
}

func catalogWriteCreatesOutput(ctx context.Context, backend storage.Backend, destinationPath string) (bool, error) {
	if _, err := backend.Stat(ctx, destinationPath); err == nil {
		return false, nil
	} else if storage.IsNotFound(err) {
		return true, nil
	} else {
		return false, err
	}
}

func catalogOutputOverwriteAllowed(plan Plan, output Output) bool {
	if plan.ClearDestinationRoot {
		return false
	}
	if plan.SupersededLegacy != nil {
		return true
	}
	if plan.ExistingCatalog == nil {
		return false
	}
	_, ok := state.FindCatalogOutputByPath(plan.ExistingCatalog.Outputs, output.DestinationPath)
	return ok
}

func catalogStateForPlan(req Request, plan Plan) state.CatalogState {
	now := requestTime(req)
	createdAt := now
	if plan.ExistingCatalog != nil {
		createdAt = plan.ExistingCatalog.CreatedAt
	}
	outputs := make([]state.CatalogOutputFile, 0, len(plan.CatalogOutputsToRetain)+len(plan.CatalogOutputsToWrite))
	outputs = append(outputs, plan.CatalogOutputsToRetain...)
	outputs = append(outputs, plan.CatalogOutputsToWrite...)
	sort.SliceStable(outputs, func(i, j int) bool {
		if outputs[i].Path != outputs[j].Path {
			return outputs[i].Path < outputs[j].Path
		}
		if outputs[i].PipelineID != outputs[j].PipelineID {
			return outputs[i].PipelineID < outputs[j].PipelineID
		}
		return outputs[i].DestinationID < outputs[j].DestinationID
	})
	return state.CatalogState{
		SchemaVersion:      state.CatalogSchemaVersion,
		DistributorVersion: req.DistributorVersion,
		CreatedAt:          createdAt,
		UpdatedAt:          now,
		State:              state.StatePolicy{Mode: state.StateModeCatalog},
		Outputs:            outputs,
	}
}

func catalogStateWriteOverwrites(plan Plan) bool {
	return plan.ExistingCatalog != nil || plan.SupersededLegacy != nil
}

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