Write catalog state during publish

internal/publish/execute.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"sort"
 	"time"
 
 	"gitea.maximumdirect.net/eric/distributor/internal/config"
@@ -16,7 +17,9 @@ func Execute(ctx context.Context, req Request, plan Plan) error {
 	switch plan.Action {
 	case ActionSkipSame, ActionSkipDestinationNewer:
 		return nil
-	case ActionPublishNew, ActionReplaceOlder, ActionReplaceConflict, ActionReplaceNewer, ActionReplaceTakeover, ActionForceReplace:
+	case ActionPublishNew, ActionUpsertAdditive, ActionReplaceCatalog:
+		return executeCatalog(ctx, req, plan)
+	case ActionReplaceOlder, ActionReplaceConflict, ActionReplaceNewer, ActionReplaceTakeover, ActionForceReplace:
 		if usesSharedRootState(req, plan) {
 			return executeSharedRoot(ctx, req, plan)
 		}
@@ -138,6 +141,152 @@ func Execute(ctx context.Context, req Request, plan Plan) error {
 	return nil
 }
 
+func executeCatalog(ctx context.Context, req Request, plan Plan) error {
+	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
+}
+
 func executeSharedRoot(ctx context.Context, req Request, plan Plan) error {
 	plan.Reconciliation = normalizeReconciliation(plan.Reconciliation)
 	if plan.Action == ActionForceReplace {