distributor/internal/state/outputs.go

package state

import (
	"fmt"
	"time"

	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
)

type OutputProjection struct {
	Path       string
	Kind       string
	SourcePath string
	Transform  string
	URL        string
	SHA256     string
	Size       int64
}

func FindOutputByPath(outputs []OutputFile, path string) (OutputFile, bool) {
	for _, output := range outputs {
		if output.Path == path {
			return output, true
		}
	}
	return OutputFile{}, false
}

func MergeOutputFiles(retained, planned []OutputFile) ([]OutputFile, error) {
	outputs := make([]OutputFile, 0, len(retained)+len(planned))
	indexByPath := make(map[string]int, len(retained)+len(planned))
	for _, output := range retained {
		if _, exists := indexByPath[output.Path]; exists {
			return nil, fmt.Errorf("state output path %q is duplicated", output.Path)
		}
		indexByPath[output.Path] = len(outputs)
		outputs = append(outputs, output)
	}
	seenPlanned := make(map[string]struct{}, len(planned))
	for _, output := range planned {
		if _, exists := seenPlanned[output.Path]; exists {
			return nil, fmt.Errorf("state output path %q is duplicated", output.Path)
		}
		seenPlanned[output.Path] = struct{}{}
		if index, exists := indexByPath[output.Path]; exists {
			outputs[index] = output
			continue
		}
		indexByPath[output.Path] = len(outputs)
		outputs = append(outputs, output)
	}
	return outputs, nil
}

func ManagedOutputPaths(s DistributorState) []string {
	paths := make([]string, 0, len(s.Outputs))
	for _, output := range s.Outputs {
		paths = append(paths, output.Path)
	}
	return paths
}

func RemoveMissingOutputs(s DistributorState, missingPaths []string) (DistributorState, bool) {
	if len(missingPaths) == 0 {
		return s, false
	}
	missing := pathSet(missingPaths)
	next := s
	next.Outputs = make([]OutputFile, 0, len(s.Outputs))
	changed := false
	for _, output := range s.Outputs {
		if _, remove := missing[output.Path]; remove {
			changed = true
			continue
		}
		next.Outputs = append(next.Outputs, output)
	}
	return next, changed
}

func ProjectOutputs(outputs []OutputProjection, existing []OutputFile, now time.Time) []OutputFile {
	now = now.UTC()
	files := make([]OutputFile, 0, len(outputs))
	for _, output := range outputs {
		createdAt := now
		if existingOutput, ok := FindOutputByPath(existing, output.Path); ok {
			createdAt = existingOutput.CreatedAt
		}
		files = append(files, OutputFile{
			Path:       output.Path,
			Kind:       output.Kind,
			SourcePath: output.SourcePath,
			Transform:  output.Transform,
			URL:        output.URL,
			SHA256:     output.SHA256,
			Size:       output.Size,
			CreatedAt:  createdAt,
			UpdatedAt:  now,
		})
	}
	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 RemoveMissingSharedRootOwnerOutputs(s SharedRootState, scope OwnerScope, missingPaths []string) (SharedRootState, bool) {
	if len(missingPaths) == 0 {
		return s, false
	}
	missing := pathSet(missingPaths)
	next := s
	next.Outputs = make([]SharedRootOutputFile, 0, len(s.Outputs))
	changed := false
	for _, output := range s.Outputs {
		if output.Owner == scope {
			if _, remove := missing[output.Path]; remove {
				changed = true
				continue
			}
		}
		next.Outputs = append(next.Outputs, output)
	}
	return next, changed
}

func RemoveMissingSharedRootOutputs(s SharedRootState, missingPaths []string) (SharedRootState, bool) {
	if len(missingPaths) == 0 {
		return s, false
	}
	missing := pathSet(missingPaths)
	next := s
	next.Outputs = make([]SharedRootOutputFile, 0, len(s.Outputs))
	changed := false
	for _, output := range s.Outputs {
		if _, remove := missing[output.Path]; remove {
			changed = true
			continue
		}
		next.Outputs = append(next.Outputs, output)
	}
	return next, changed
}

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)
}

func pathSet(paths []string) map[string]struct{} {
	set := make(map[string]struct{}, len(paths))
	for _, path := range paths {
		set[path] = struct{}{}
	}
	return set
}