Add shared-root destination state model

internal/state/outputs.go

@@ -3,6 +3,8 @@ package state
 import (
 	"fmt"
 	"time"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
 )
 
 type OutputProjection struct {
@@ -80,3 +82,185 @@ func ProjectOutputs(outputs []OutputProjection, existing []OutputFile, now time.
 	}
 	return files
 }
+
+func CurrentOwnerScope(pipelineID, destinationID string) OwnerScope {
+	return OwnerScope{PipelineID: pipelineID, DestinationID: destinationID}
+}
+
+func (s SharedRootState) Owner(scope OwnerScope) (OwnerRecord, bool) {
+	for _, owner := range s.Owners {
+		if owner.Scope == scope {
+			return owner, true
+		}
+	}
+	return OwnerRecord{}, false
+}
+
+func (s SharedRootState) SourceManifest(scope OwnerScope) (bundle.Manifest, bool) {
+	owner, ok := s.Owner(scope)
+	if !ok {
+		return bundle.Manifest{}, false
+	}
+	return owner.Source.Manifest, true
+}
+
+func (s SharedRootState) ManagedOutputPaths(scope OwnerScope) []string {
+	paths := make([]string, 0, len(s.Outputs))
+	for _, output := range s.Outputs {
+		if output.Owner == scope {
+			paths = append(paths, output.Path)
+		}
+	}
+	return paths
+}
+
+func (s SharedRootState) AllManagedOutputPaths() []string {
+	paths := make([]string, 0, len(s.Outputs))
+	for _, output := range s.Outputs {
+		paths = append(paths, output.Path)
+	}
+	return paths
+}
+
+func (s SharedRootState) OutputOwner(path string) (OwnerScope, bool) {
+	for _, output := range s.Outputs {
+		if output.Path == path {
+			return output.Owner, true
+		}
+	}
+	return OwnerScope{}, false
+}
+
+func (s SharedRootState) PathOwnershipConflict(scope OwnerScope, paths []string) (PathOwnershipConflict, bool) {
+	for _, path := range paths {
+		owner, exists := s.OutputOwner(path)
+		if exists && owner != scope {
+			return PathOwnershipConflict{Path: path, Owner: owner}, true
+		}
+	}
+	return PathOwnershipConflict{}, false
+}
+
+func ProjectSharedRootOutputs(outputs []OutputProjection, existing []SharedRootOutputFile, scope OwnerScope, source bundle.Manifest, now time.Time) []SharedRootOutputFile {
+	now = now.UTC()
+	files := make([]SharedRootOutputFile, 0, len(outputs))
+	for _, output := range outputs {
+		createdAt := now
+		if existingOutput, ok := findSharedRootOutput(existing, output.Path); ok && existingOutput.Owner == scope {
+			createdAt = existingOutput.CreatedAt
+		}
+		files = append(files, SharedRootOutputFile{
+			Path:          output.Path,
+			Kind:          output.Kind,
+			SourcePath:    output.SourcePath,
+			Transform:     output.Transform,
+			URL:           output.URL,
+			SHA256:        output.SHA256,
+			Size:          output.Size,
+			Owner:         scope,
+			SourceID:      source.ID,
+			SourceDigest:  source.Digest,
+			SourceCreated: source.Created,
+			CreatedAt:     createdAt,
+			UpdatedAt:     now,
+		})
+	}
+	return files
+}
+
+func ReplaceOwnerOutputs(s SharedRootState, scope OwnerScope, owner OwnerRecord, planned []SharedRootOutputFile) (SharedRootState, error) {
+	if conflict, ok := s.PathOwnershipConflict(scope, sharedRootOutputPaths(planned)); ok {
+		return SharedRootState{}, fmt.Errorf("state output path %q is owned by %s/%s", conflict.Path, conflict.Owner.PipelineID, conflict.Owner.DestinationID)
+	}
+	if err := validatePlannedSharedRootOutputs(scope, planned); err != nil {
+		return SharedRootState{}, err
+	}
+	next := s
+	next.Owners = upsertOwner(s.Owners, owner)
+	next.Outputs = make([]SharedRootOutputFile, 0, len(s.Outputs)+len(planned))
+	for _, output := range s.Outputs {
+		if output.Owner != scope {
+			next.Outputs = append(next.Outputs, output)
+		}
+	}
+	next.Outputs = append(next.Outputs, planned...)
+	return next, nil
+}
+
+func MergeOwnerOutputs(s SharedRootState, scope OwnerScope, owner OwnerRecord, planned []SharedRootOutputFile) (SharedRootState, error) {
+	if conflict, ok := s.PathOwnershipConflict(scope, sharedRootOutputPaths(planned)); ok {
+		return SharedRootState{}, fmt.Errorf("state output path %q is owned by %s/%s", conflict.Path, conflict.Owner.PipelineID, conflict.Owner.DestinationID)
+	}
+	if err := validatePlannedSharedRootOutputs(scope, planned); err != nil {
+		return SharedRootState{}, err
+	}
+	next := s
+	next.Owners = upsertOwner(s.Owners, owner)
+	outputs := make([]SharedRootOutputFile, 0, len(s.Outputs)+len(planned))
+	indexByPath := make(map[string]int, len(s.Outputs)+len(planned))
+	for _, output := range s.Outputs {
+		indexByPath[output.Path] = len(outputs)
+		outputs = append(outputs, output)
+	}
+	for _, output := range planned {
+		if index, exists := indexByPath[output.Path]; exists {
+			outputs[index] = output
+			continue
+		}
+		indexByPath[output.Path] = len(outputs)
+		outputs = append(outputs, output)
+	}
+	next.Outputs = outputs
+	return next, nil
+}
+
+func findSharedRootOutput(outputs []SharedRootOutputFile, path string) (SharedRootOutputFile, bool) {
+	for _, output := range outputs {
+		if output.Path == path {
+			return output, true
+		}
+	}
+	return SharedRootOutputFile{}, false
+}
+
+func sharedRootOutputPaths(outputs []SharedRootOutputFile) []string {
+	paths := make([]string, 0, len(outputs))
+	for _, output := range outputs {
+		paths = append(paths, output.Path)
+	}
+	return paths
+}
+
+func rejectDuplicateSharedRootOutputs(outputs []SharedRootOutputFile) error {
+	seen := make(map[string]struct{}, len(outputs))
+	for _, output := range outputs {
+		if _, exists := seen[output.Path]; exists {
+			return fmt.Errorf("state output path %q is duplicated", output.Path)
+		}
+		seen[output.Path] = struct{}{}
+	}
+	return nil
+}
+
+func validatePlannedSharedRootOutputs(scope OwnerScope, outputs []SharedRootOutputFile) error {
+	if err := rejectDuplicateSharedRootOutputs(outputs); err != nil {
+		return err
+	}
+	for _, output := range outputs {
+		if output.Owner != scope {
+			return fmt.Errorf("state output path %q is owned by %s/%s, not %s/%s", output.Path, output.Owner.PipelineID, output.Owner.DestinationID, scope.PipelineID, scope.DestinationID)
+		}
+	}
+	return nil
+}
+
+func upsertOwner(owners []OwnerRecord, owner OwnerRecord) []OwnerRecord {
+	next := append([]OwnerRecord(nil), owners...)
+	for index, existing := range next {
+		if existing.Scope == owner.Scope {
+			next[index] = owner
+			return next
+		}
+	}
+	return append(next, owner)
+}