package previouscache import ( "context" "fmt" "os" "path" "path/filepath" "sort" "strings" "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" "gitea.maximumdirect.net/eric/narratio/internal/artifacts" "gitea.maximumdirect.net/eric/narratio/internal/config" "gitea.maximumdirect.net/eric/narratio/internal/manifest" ) const ( InputKindManifest = "previous_manifest" InputKindArtifact = "previous_artifact" InputSource = "previous_session_publish.current" ) type Plan struct { Records []Record SkippedMissing []string PreviousRunID string } type Record struct { Kind string RequirementName string Required bool LocalRelativePath string LocalPath string RemoteKey string S3Bucket string } func BuildPlan( ctx context.Context, cfg *config.Config, paths artifacts.SessionPaths, requirements []artifacts.PreviousArtifactRequirement, store storage.ObjectStore, ) (*Plan, error) { if len(requirements) == 0 { return &Plan{}, nil } if cfg == nil || cfg.Session == nil || cfg.Pipeline == nil { return nil, fmt.Errorf("resolved config with session/pipeline is required") } orderedRequirements := append([]artifacts.PreviousArtifactRequirement(nil), requirements...) sort.Slice(orderedRequirements, func(i, j int) bool { return orderedRequirements[i].Name < orderedRequirements[j].Name }) requiredNames := requiredPreviousArtifactNames(orderedRequirements) optionalNames := optionalPreviousArtifactNames(orderedRequirements) previousSessionID := strings.TrimSpace(cfg.Session.PreviousSessionID) if previousSessionID == "" { if len(requiredNames) > 0 { return nil, fmt.Errorf( "previous_session_id is required for required previous-session artifacts: %s", strings.Join(requiredNames, ", "), ) } return &Plan{SkippedMissing: optionalNames}, nil } if store == nil { return nil, fmt.Errorf("previous-session artifact hydration requires object store backend") } if cfg.Pipeline.Storage.S3 == nil { return nil, fmt.Errorf("pipeline.storage.s3 configuration is required for previous-session artifact hydration") } campaign := strings.TrimSpace(cfg.Session.Campaign) if campaign == "" { return nil, fmt.Errorf("session campaign is required for previous-session artifact hydration") } bucket := strings.TrimSpace(cfg.Pipeline.Storage.S3.Bucket) if bucket == "" { return nil, fmt.Errorf("pipeline.storage.s3.bucket is required for previous-session artifact hydration") } previousSessionPrefix := artifacts.S3SessionPrefix( cfg.Pipeline.Storage.S3.RootPrefix, campaign, previousSessionID, ) currentManifestKey, currentRunIDKey := artifacts.ResolveArchiveCurrentStateKeys(previousSessionPrefix) result := &Plan{} runPointerExists, err := store.Exists(ctx, currentRunIDKey) if err != nil { return nil, fmt.Errorf("check previous-session current run pointer %q: %w", currentRunIDKey, err) } if !runPointerExists { if len(requiredNames) > 0 { return nil, fmt.Errorf("required previous-session artifacts unavailable: remote current run pointer missing: %q", currentRunIDKey) } result.SkippedMissing = optionalNames return result, nil } runIDTemp, err := downloadObjectToTemp(ctx, store, currentRunIDKey, "narratio-previous-run-id-*.txt") if err != nil { return nil, fmt.Errorf("download previous-session current run pointer %q: %w", currentRunIDKey, err) } defer func() { _ = os.Remove(runIDTemp) }() runIDBytes, err := os.ReadFile(runIDTemp) if err != nil { return nil, fmt.Errorf("read previous-session current run pointer %q: %w", currentRunIDKey, err) } previousRunID := strings.TrimSpace(string(runIDBytes)) if previousRunID == "" { return nil, fmt.Errorf("previous-session current run pointer %q is empty", currentRunIDKey) } result.PreviousRunID = previousRunID manifestExists, err := store.Exists(ctx, currentManifestKey) if err != nil { return nil, fmt.Errorf("check previous-session current manifest %q: %w", currentManifestKey, err) } if !manifestExists { if len(requiredNames) > 0 { return nil, fmt.Errorf("required previous-session artifacts unavailable: remote current manifest missing: %q", currentManifestKey) } result.SkippedMissing = optionalNames return result, nil } manifestTemp, err := downloadObjectToTemp(ctx, store, currentManifestKey, "narratio-previous-manifest-*.json") if err != nil { return nil, fmt.Errorf("download previous-session current manifest %q: %w", currentManifestKey, err) } defer func() { _ = os.Remove(manifestTemp) }() manifestStore := &manifest.LocalStore{} previousManifest, err := manifestStore.Load(ctx, manifestTemp) if err != nil { return nil, fmt.Errorf("decode downloaded previous-session manifest %q: %w", currentManifestKey, err) } if strings.TrimSpace(previousManifest.SessionID) != previousSessionID { return nil, fmt.Errorf( "previous-session manifest session_id %q does not match configured previous_session_id %q", strings.TrimSpace(previousManifest.SessionID), previousSessionID, ) } if strings.TrimSpace(previousManifest.Campaign) != campaign { return nil, fmt.Errorf( "previous-session manifest campaign %q does not match current campaign %q", strings.TrimSpace(previousManifest.Campaign), campaign, ) } if strings.TrimSpace(previousManifest.RunID) == "" { return nil, fmt.Errorf("previous-session manifest run_id is required") } if strings.TrimSpace(previousManifest.RunID) != previousRunID { return nil, fmt.Errorf( "previous-session current run pointer %q references run %q but current manifest run_id is %q", currentRunIDKey, previousRunID, strings.TrimSpace(previousManifest.RunID), ) } manifestRel, err := relativeToSession(paths, paths.PreviousManifestPath) if err != nil { return nil, err } result.Records = append(result.Records, Record{ Kind: InputKindManifest, LocalRelativePath: manifestRel, LocalPath: paths.PreviousManifestPath, RemoteKey: currentManifestKey, S3Bucket: bucket, }) for _, requirement := range orderedRequirements { candidates := artifactRelativePathCandidates(requirement.Name, previousManifest, cfg) if len(candidates) == 0 { if requirement.Required { return nil, fmt.Errorf( "required previous-session artifact %q is unavailable in previous-session manifest/archive", requirement.Name, ) } result.SkippedMissing = append(result.SkippedMissing, requirement.Name) continue } selectedRel := "" selectedKey := "" for _, candidate := range candidates { remoteKey := artifacts.S3PublishedOutputKey(previousSessionPrefix, candidate) exists, err := store.Exists(ctx, remoteKey) if err != nil { return nil, fmt.Errorf("check previous-session artifact object %q: %w", remoteKey, err) } if !exists { continue } selectedRel = candidate selectedKey = remoteKey break } if selectedRel == "" { if requirement.Required { return nil, fmt.Errorf( "required previous-session artifact %q object missing from archive candidate keys", requirement.Name, ) } result.SkippedMissing = append(result.SkippedMissing, requirement.Name) continue } localPath := artifacts.SessionPreviousArtifactPath(paths, selectedRel) localRel, err := relativeToSession(paths, localPath) if err != nil { return nil, err } result.Records = append(result.Records, Record{ Kind: InputKindArtifact, RequirementName: requirement.Name, Required: requirement.Required, LocalRelativePath: localRel, LocalPath: localPath, RemoteKey: selectedKey, S3Bucket: bucket, }) } sort.Strings(result.SkippedMissing) sort.Slice(result.Records, func(i, j int) bool { if result.Records[i].LocalRelativePath != result.Records[j].LocalRelativePath { return result.Records[i].LocalRelativePath < result.Records[j].LocalRelativePath } return result.Records[i].RemoteKey < result.Records[j].RemoteKey }) return result, nil } func requiredPreviousArtifactNames(requirements []artifacts.PreviousArtifactRequirement) []string { names := make([]string, 0, len(requirements)) for _, requirement := range requirements { if requirement.Required { names = append(names, strings.TrimSpace(requirement.Name)) } } sort.Strings(names) return names } func optionalPreviousArtifactNames(requirements []artifacts.PreviousArtifactRequirement) []string { names := make([]string, 0, len(requirements)) for _, requirement := range requirements { if requirement.Required { continue } names = append(names, strings.TrimSpace(requirement.Name)) } sort.Strings(names) return names } func artifactRelativePathCandidates( artifactName string, previousManifest *manifest.Manifest, cfg *config.Config, ) []string { candidates := []string{} appendCandidate := func(v string) { normalized, err := normalizeArchiveRelativePath(v) if err != nil { return } candidates = append(candidates, normalized) } sourceID := artifacts.ConfiguredArtifactSourceID(artifactName) if rel, ok := manifestArtifactRelativePathBySourceID(previousManifest, sourceID); ok { appendCandidate(rel) base := path.Base(rel) for _, published := range manifestPublishedPaths(previousManifest) { if path.Base(published) == base { appendCandidate(published) } } } if cfg != nil && cfg.Pipeline != nil && cfg.Pipeline.Scriptorium != nil { if artifactCfg, ok := cfg.Pipeline.Scriptorium.Artifacts[artifactName]; ok { appendCandidate(artifactCfg.OutputPath) } } return dedupeOrderedStrings(candidates) } func manifestArtifactRelativePathBySourceID(previousManifest *manifest.Manifest, sourceID string) (string, bool) { if previousManifest == nil || len(previousManifest.Stages) == 0 { return "", false } sourceID = strings.TrimSpace(sourceID) if sourceID == "" { return "", false } stageNames := make([]string, 0, len(previousManifest.Stages)) if _, ok := previousManifest.Stages["analyze"]; ok { stageNames = append(stageNames, "analyze") } for stageName := range previousManifest.Stages { if stageName == "analyze" { continue } stageNames = append(stageNames, stageName) } start := 0 if len(stageNames) > 0 && stageNames[0] == "analyze" { start = 1 } sort.Strings(stageNames[start:]) for _, stageName := range stageNames { sr := previousManifest.Stages[stageName] if sr == nil { continue } for _, out := range sr.Outputs { if strings.TrimSpace(out.SourceID) != sourceID { continue } rel, ok := deriveManifestRelativePath(previousManifest, out.LocalPath) if ok { return rel, true } } } return "", false } func deriveManifestRelativePath(previousManifest *manifest.Manifest, localPath string) (string, bool) { trimmed := strings.TrimSpace(localPath) if trimmed == "" { return "", false } if !filepath.IsAbs(trimmed) { normalized, err := normalizeArchiveRelativePath(filepath.ToSlash(trimmed)) if err != nil { return "", false } return normalized, true } sessionRoot, ok := manifestSessionRoot(previousManifest) if !ok { return "", false } rel, err := filepath.Rel(sessionRoot, trimmed) if err != nil { return "", false } normalized, err := normalizeArchiveRelativePath(filepath.ToSlash(rel)) if err != nil { return "", false } return normalized, true } func manifestSessionRoot(previousManifest *manifest.Manifest) (string, bool) { if previousManifest == nil { return "", false } runRoot := filepath.Clean(strings.TrimSpace(previousManifest.LocalWorkDir)) runID := strings.TrimSpace(previousManifest.RunID) if runRoot == "" || runID == "" { return "", false } if filepath.Base(runRoot) != runID { return "", false } runsDir := filepath.Dir(runRoot) if filepath.Base(runsDir) != config.PathRunsDirSegment { return "", false } return filepath.Dir(runsDir), true } func manifestPublishedPaths(previousManifest *manifest.Manifest) []string { if previousManifest == nil || len(previousManifest.Stages) == 0 { return nil } sr := previousManifest.Stages["publish"] if sr == nil || sr.Metadata == nil { return nil } raw, ok := sr.Metadata["published_paths"] if !ok { return nil } values, ok := raw.([]any) if !ok { return nil } out := make([]string, 0, len(values)) for _, value := range values { asString, ok := value.(string) if !ok { continue } normalized, err := normalizeArchiveRelativePath(asString) if err != nil { continue } out = append(out, normalized) } return dedupeOrderedStrings(out) } func normalizeArchiveRelativePath(rel string) (string, error) { trimmed := strings.TrimSpace(rel) if trimmed == "" { return "", fmt.Errorf("relative path is required") } cleaned := filepath.ToSlash(filepath.Clean(filepath.FromSlash(trimmed))) if cleaned == "." || cleaned == "" { return "", fmt.Errorf("relative path is required") } if filepath.IsAbs(trimmed) || strings.HasPrefix(cleaned, "/") || cleaned == ".." || strings.HasPrefix(cleaned, "../") { return "", fmt.Errorf("path must be a clean relative path") } return cleaned, nil } func relativeToSession(paths artifacts.SessionPaths, localPath string) (string, error) { root := filepath.Clean(paths.Root) if strings.TrimSpace(root) == "" { return "", fmt.Errorf("session root is required") } rel, err := filepath.Rel(root, filepath.Clean(localPath)) if err != nil { return "", fmt.Errorf("resolve previous-cache relative path: %w", err) } normalized, err := normalizeArchiveRelativePath(filepath.ToSlash(rel)) if err != nil { return "", fmt.Errorf("resolve previous-cache relative path: %w", err) } return normalized, nil } func dedupeOrderedStrings(values []string) []string { if len(values) == 0 { return nil } seen := map[string]struct{}{} out := make([]string, 0, len(values)) for _, value := range values { trimmed := strings.TrimSpace(value) if trimmed == "" { continue } if _, ok := seen[trimmed]; ok { continue } seen[trimmed] = struct{}{} out = append(out, trimmed) } return out } func downloadObjectToTemp(ctx context.Context, store storage.ObjectStore, key, pattern string) (string, error) { tmp, err := os.CreateTemp("", pattern) if err != nil { return "", fmt.Errorf("create temp file: %w", err) } path := tmp.Name() if err := tmp.Close(); err != nil { _ = os.Remove(path) return "", fmt.Errorf("close temp file: %w", err) } if err := store.Download(ctx, key, path); err != nil { _ = os.Remove(path) return "", err } return path, nil }