Add prune retention planning

internal/app/prune.go

@@ -0,0 +1,106 @@
+package app
+
+import (
+	"fmt"
+	"time"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/config"
+	"gitea.maximumdirect.net/eric/distributor/internal/state"
+)
+
+type PrunePlanOptions struct {
+	PipelineID    string
+	DestinationID string
+	Now           time.Time
+}
+
+type PrunePlanReport struct {
+	PipelineID       string              `json:"pipeline_id"`
+	DestinationID    string              `json:"destination_id"`
+	OwnerScope       PruneOwnerScope     `json:"owner_scope"`
+	Enabled          bool                `json:"enabled"`
+	CheckedCount     int                 `json:"checked_count"`
+	PrunedOutputs    []PruneOutputRecord `json:"pruned_outputs"`
+	PreservedOutputs []PruneOutputRecord `json:"preserved_outputs"`
+}
+
+type PruneOwnerScope struct {
+	PipelineID    string `json:"pipeline_id"`
+	DestinationID string `json:"destination_id"`
+}
+
+type PruneOutputRecord struct {
+	Path      string           `json:"path"`
+	UpdatedAt string           `json:"updated_at"`
+	Owner     *PruneOwnerScope `json:"owner,omitempty"`
+}
+
+func PlanPrune(document state.StateDocument, policy config.PrunePolicy, options PrunePlanOptions) (PrunePlanReport, error) {
+	scope := state.CurrentOwnerScope(options.PipelineID, options.DestinationID)
+	report := PrunePlanReport{
+		PipelineID:       options.PipelineID,
+		DestinationID:    options.DestinationID,
+		OwnerScope:       PruneOwnerScope{PipelineID: scope.PipelineID, DestinationID: scope.DestinationID},
+		Enabled:          policy.Enabled,
+		PrunedOutputs:    []PruneOutputRecord{},
+		PreservedOutputs: []PruneOutputRecord{},
+	}
+	if !policy.Enabled {
+		return report, nil
+	}
+
+	candidates, err := pruneCandidatesForDocument(document, scope)
+	if err != nil {
+		return PrunePlanReport{}, err
+	}
+	report.CheckedCount = len(candidates)
+	plan := state.PlanPrune(candidates, state.PrunePlanOptions{
+		Now:        options.Now,
+		OlderThan:  pruneOlderThan(policy),
+		KeepLatest: policy.KeepLatest,
+	})
+	report.PrunedOutputs = pruneOutputRecords(plan.Pruned)
+	report.PreservedOutputs = pruneOutputRecords(plan.Preserved)
+	return report, nil
+}
+
+func pruneCandidatesForDocument(document state.StateDocument, scope state.OwnerScope) ([]state.PruneCandidate, error) {
+	if document.SingleOwner != nil {
+		singleOwner := *document.SingleOwner
+		if singleOwner.PipelineID != scope.PipelineID || singleOwner.DestinationID != scope.DestinationID {
+			return nil, fmt.Errorf("state owner is %s/%s, not %s/%s", singleOwner.PipelineID, singleOwner.DestinationID, scope.PipelineID, scope.DestinationID)
+		}
+		return state.SingleOwnerPruneCandidates(singleOwner), nil
+	}
+	if document.SharedRoot != nil {
+		return state.SharedRootPruneCandidates(*document.SharedRoot, scope), nil
+	}
+	return nil, fmt.Errorf("destination state document is empty")
+}
+
+func pruneOlderThan(policy config.PrunePolicy) *time.Duration {
+	if policy.OlderThan == nil {
+		return nil
+	}
+	duration := policy.OlderThan.AsDuration()
+	return &duration
+}
+
+func pruneOutputRecords(candidates []state.PruneCandidate) []PruneOutputRecord {
+	records := make([]PruneOutputRecord, 0, len(candidates))
+	for _, candidate := range candidates {
+		var owner *PruneOwnerScope
+		if candidate.Owner != nil {
+			owner = &PruneOwnerScope{
+				PipelineID:    candidate.Owner.PipelineID,
+				DestinationID: candidate.Owner.DestinationID,
+			}
+		}
+		records = append(records, PruneOutputRecord{
+			Path:      candidate.Path,
+			UpdatedAt: candidate.UpdatedAt.UTC().Format(time.RFC3339),
+			Owner:     owner,
+		})
+	}
+	return records
+}