583 lines
17 KiB
Go
583 lines
17 KiB
Go
package stage
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io/fs"
|
|
"os"
|
|
"path/filepath"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"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"
|
|
)
|
|
|
|
type archiveStage struct{}
|
|
type archiveUploadFile struct {
|
|
RelativePath string
|
|
LocalPath string
|
|
}
|
|
|
|
var archivePrerequisiteStages = []string{
|
|
"prepare",
|
|
"transcribe",
|
|
"merge",
|
|
"polish",
|
|
"normalize",
|
|
"trim",
|
|
"analyze",
|
|
}
|
|
|
|
func (archiveStage) Name() string { return "archive" }
|
|
|
|
func (archiveStage) Declares() IODecl {
|
|
return IODecl{
|
|
Inputs: []artifacts.Ref{
|
|
{Kind: "manifest", Category: "input", RelativePath: "manifest.json"},
|
|
},
|
|
}
|
|
}
|
|
|
|
func (archiveStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
|
|
if env == nil || env.Config == nil || env.Config.Pipeline == nil || env.Config.Session == nil {
|
|
return nil, fmt.Errorf("archive: resolved config must include pipeline and session")
|
|
}
|
|
|
|
if archiveDisabled(env) {
|
|
return &StageResult{
|
|
Metadata: map[string]any{
|
|
"stage": "archive",
|
|
"skipped": true,
|
|
"archive_enabled": false,
|
|
"audio_upload_skipped": true,
|
|
"current_pointer_written": false,
|
|
},
|
|
}, nil
|
|
}
|
|
if archiveRunUploadDisabled(env) {
|
|
return &StageResult{
|
|
Metadata: map[string]any{
|
|
"stage": "archive",
|
|
"skipped": true,
|
|
"upload_run_enabled": false,
|
|
"audio_upload_skipped": true,
|
|
"current_pointer_written": false,
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
if err := validateArchivePrerequisites(m); err != nil {
|
|
return nil, fmt.Errorf("archive: %w", err)
|
|
}
|
|
if env.ObjectStore == nil {
|
|
return nil, fmt.Errorf("archive: remote object store backend is required when archive run upload is enabled")
|
|
}
|
|
|
|
runRoot, err := resolveArchiveRunRoot(env, m)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: resolve run root: %w", err)
|
|
}
|
|
runRootInfo, err := os.Stat(runRoot)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: run root %q: %w", runRoot, err)
|
|
}
|
|
if !runRootInfo.IsDir() {
|
|
return nil, fmt.Errorf("archive: run root %q is not a directory", runRoot)
|
|
}
|
|
|
|
runPrefix, err := archiveRunPrefix(env, m)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: resolve s3 run prefix: %w", err)
|
|
}
|
|
sessionPrefix, err := archiveSessionPrefix(env, m)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: resolve s3 session prefix: %w", err)
|
|
}
|
|
bucket := archiveBucket(env, m)
|
|
if bucket == "" {
|
|
return nil, fmt.Errorf("archive: resolve s3 bucket: bucket is required")
|
|
}
|
|
runID := strings.TrimSpace(m.RunID)
|
|
if runID == "" {
|
|
return nil, fmt.Errorf("archive: run id is required")
|
|
}
|
|
|
|
manifestSource, err := resolveArchiveRunManifestSource(runRoot)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: resolve run manifest source: %w", err)
|
|
}
|
|
|
|
runFiles, err := collectArchiveRunFiles(runRoot, manifestSource)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: collect run files: %w", err)
|
|
}
|
|
sessionRoot, err := resolveArchiveSessionRoot(env, m)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: resolve session root for promotions: %w", err)
|
|
}
|
|
promotions, err := resolveArchivePromotions(sessionRoot, env.Config.Pipeline.Archive.PromoteArtifacts)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: resolve promotion rules: %w", err)
|
|
}
|
|
runUploaded := make([]string, 0, len(runFiles))
|
|
for _, file := range runFiles {
|
|
key := artifacts.S3RunRelativeDestinationKey(runPrefix, file.RelativePath)
|
|
if _, err := env.ObjectStore.Upload(ctx, file.LocalPath, key, storage.UploadOptions{}); err != nil {
|
|
return nil, fmt.Errorf("archive: upload run file %q to %q: %w", file.RelativePath, key, err)
|
|
}
|
|
runUploaded = append(runUploaded, file.RelativePath)
|
|
}
|
|
|
|
promotedUploaded := make([]string, 0, len(promotions))
|
|
skippedOptional := make([]string, 0)
|
|
for _, promotion := range promotions {
|
|
if !promotion.Exists {
|
|
if promotion.Required {
|
|
return nil, fmt.Errorf("archive: required promotion source missing: %q", promotion.From)
|
|
}
|
|
skippedOptional = append(skippedOptional, promotion.To)
|
|
continue
|
|
}
|
|
key := artifacts.S3PromotedArtifactKey(sessionPrefix, promotion.To)
|
|
if _, err := env.ObjectStore.Upload(ctx, promotion.LocalPath, key, storage.UploadOptions{}); err != nil {
|
|
return nil, fmt.Errorf("archive: upload promoted output %q to %q: %w", promotion.From, key, err)
|
|
}
|
|
promotedUploaded = append(promotedUploaded, promotion.To)
|
|
}
|
|
|
|
currentManifestKey := artifacts.S3CurrentManifestKey(sessionPrefix)
|
|
manifestTempPath, err := writeCurrentManifestSnapshot(m, archiveMetadataPreview(
|
|
bucket,
|
|
runPrefix,
|
|
sessionPrefix,
|
|
runUploaded,
|
|
promotedUploaded,
|
|
skippedOptional,
|
|
currentManifestKey,
|
|
))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: build current manifest snapshot: %w", err)
|
|
}
|
|
defer func() { _ = os.Remove(manifestTempPath) }()
|
|
|
|
if _, err := env.ObjectStore.Upload(ctx, manifestTempPath, currentManifestKey, storage.UploadOptions{
|
|
ContentType: "application/json",
|
|
}); err != nil {
|
|
return nil, fmt.Errorf("archive: upload current manifest to %q: %w", currentManifestKey, err)
|
|
}
|
|
|
|
currentRunPointerKey := artifacts.S3CurrentRunPointerKey(sessionPrefix)
|
|
runIDTempPath, err := writeCurrentRunIDPointer(runID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: build current run id pointer: %w", err)
|
|
}
|
|
defer func() { _ = os.Remove(runIDTempPath) }()
|
|
|
|
if _, err := env.ObjectStore.Upload(ctx, runIDTempPath, currentRunPointerKey, storage.UploadOptions{
|
|
ContentType: "text/plain; charset=utf-8",
|
|
}); err != nil {
|
|
return nil, fmt.Errorf("archive: upload current run pointer to %q: %w", currentRunPointerKey, err)
|
|
}
|
|
|
|
return &StageResult{
|
|
Metadata: map[string]any{
|
|
"stage": "archive",
|
|
"uploaded": true,
|
|
"s3_bucket": bucket,
|
|
"s3_run_prefix": runPrefix,
|
|
"run_files_uploaded": len(runUploaded),
|
|
"run_uploaded_paths": runUploaded,
|
|
"promoted_files_uploaded": len(promotedUploaded),
|
|
"promoted_paths": promotedUploaded,
|
|
"skipped_optional_promotions": skippedOptional,
|
|
"current_manifest_key": currentManifestKey,
|
|
"current_run_id_key": currentRunPointerKey,
|
|
"current_pointer_written": true,
|
|
"audio_upload_skipped": true,
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
type archivePromotion struct {
|
|
From string
|
|
To string
|
|
Required bool
|
|
LocalPath string
|
|
Exists bool
|
|
}
|
|
|
|
func archiveDisabled(env *Env) bool {
|
|
cfg := env.Config.Pipeline.Archive
|
|
if cfg == nil {
|
|
return true
|
|
}
|
|
return cfg.Enabled != nil && !*cfg.Enabled
|
|
}
|
|
|
|
func archiveRunUploadDisabled(env *Env) bool {
|
|
cfg := env.Config.Pipeline.Archive
|
|
if cfg == nil {
|
|
return true
|
|
}
|
|
return cfg.UploadRun != nil && !*cfg.UploadRun
|
|
}
|
|
|
|
func validateArchivePrerequisites(m *manifest.Manifest) error {
|
|
if m == nil {
|
|
return fmt.Errorf("manifest is required")
|
|
}
|
|
for _, stageName := range archivePrerequisiteStages {
|
|
sr := m.Stages[stageName]
|
|
if sr == nil {
|
|
return fmt.Errorf("prerequisite stage %q has not succeeded", stageName)
|
|
}
|
|
if sr.Status != manifest.StatusSucceeded {
|
|
return fmt.Errorf("prerequisite stage %q status is %q (want %q)", stageName, sr.Status, manifest.StatusSucceeded)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func resolveArchiveRunRoot(env *Env, m *manifest.Manifest) (string, error) {
|
|
sessionID := strings.TrimSpace(env.Config.Session.SessionID)
|
|
if sessionID == "" && m != nil {
|
|
sessionID = strings.TrimSpace(m.SessionID)
|
|
}
|
|
campaign := strings.TrimSpace(env.Config.Session.Campaign)
|
|
if campaign == "" && m != nil {
|
|
campaign = strings.TrimSpace(m.Campaign)
|
|
}
|
|
runID := ""
|
|
if m != nil {
|
|
runID = strings.TrimSpace(m.RunID)
|
|
}
|
|
if sessionID == "" || campaign == "" {
|
|
return "", fmt.Errorf("campaign and session id are required")
|
|
}
|
|
if runID == "" {
|
|
return "", fmt.Errorf("run id is required")
|
|
}
|
|
|
|
canonical := filepath.Clean(artifacts.SessionRunRootForCampaign(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID))
|
|
canonicalExists, err := directoryExists(canonical)
|
|
if err != nil {
|
|
return "", fmt.Errorf("check canonical run root %q: %w", canonical, err)
|
|
}
|
|
if !canonicalExists {
|
|
return "", fmt.Errorf("run root not found for campaign %q session %q run %q at canonical path %q", campaign, sessionID, runID, canonical)
|
|
}
|
|
return canonical, nil
|
|
}
|
|
|
|
func resolveArchiveSessionRoot(env *Env, m *manifest.Manifest) (string, error) {
|
|
sessionID := strings.TrimSpace(env.Config.Session.SessionID)
|
|
if sessionID == "" && m != nil {
|
|
sessionID = strings.TrimSpace(m.SessionID)
|
|
}
|
|
campaign := strings.TrimSpace(env.Config.Session.Campaign)
|
|
if campaign == "" && m != nil {
|
|
campaign = strings.TrimSpace(m.Campaign)
|
|
}
|
|
if sessionID == "" {
|
|
return "", fmt.Errorf("session id is required")
|
|
}
|
|
if campaign == "" {
|
|
return "", fmt.Errorf("campaign is required")
|
|
}
|
|
return filepath.Clean(artifacts.SessionWorkDirForCampaign(env.Config.Pipeline.Workspace.Root, campaign, sessionID)), nil
|
|
}
|
|
|
|
func archiveRunPrefix(env *Env, m *manifest.Manifest) (string, error) {
|
|
runPrefix := strings.TrimSpace(m.S3RunPrefix)
|
|
if runPrefix != "" {
|
|
return runPrefix, nil
|
|
}
|
|
|
|
sessionPrefix, err := archiveSessionPrefix(env, m)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
runID := strings.TrimSpace(m.RunID)
|
|
if runID == "" {
|
|
return "", fmt.Errorf("run id is required")
|
|
}
|
|
return artifacts.S3RunPrefix(sessionPrefix, runID), nil
|
|
}
|
|
|
|
func archiveSessionPrefix(env *Env, m *manifest.Manifest) (string, error) {
|
|
if m != nil && strings.TrimSpace(m.S3SessionPrefix) != "" {
|
|
return strings.TrimSpace(m.S3SessionPrefix), nil
|
|
}
|
|
|
|
sessionID := strings.TrimSpace(env.Config.Session.SessionID)
|
|
if sessionID == "" {
|
|
sessionID = strings.TrimSpace(m.SessionID)
|
|
}
|
|
campaign := strings.TrimSpace(env.Config.Session.Campaign)
|
|
if campaign == "" {
|
|
campaign = strings.TrimSpace(m.Campaign)
|
|
}
|
|
if env.Config.Pipeline.Storage.S3 == nil {
|
|
return "", fmt.Errorf("pipeline.storage.s3 configuration is required")
|
|
}
|
|
sessionPrefix := artifacts.S3SessionPrefix(env.Config.Pipeline.Storage.S3.RootPrefix, campaign, sessionID)
|
|
if strings.TrimSpace(sessionPrefix) == "" {
|
|
return "", fmt.Errorf("session prefix is required")
|
|
}
|
|
return sessionPrefix, nil
|
|
}
|
|
|
|
func archiveBucket(env *Env, m *manifest.Manifest) string {
|
|
if m != nil && strings.TrimSpace(m.S3Bucket) != "" {
|
|
return strings.TrimSpace(m.S3Bucket)
|
|
}
|
|
if env == nil || env.Config == nil || env.Config.Pipeline == nil || env.Config.Pipeline.Storage.S3 == nil {
|
|
return ""
|
|
}
|
|
return strings.TrimSpace(env.Config.Pipeline.Storage.S3.Bucket)
|
|
}
|
|
|
|
func resolveArchivePromotions(sessionRoot string, rules []config.ArchivePromotionRule) ([]archivePromotion, error) {
|
|
sessionRoot = filepath.Clean(strings.TrimSpace(sessionRoot))
|
|
if sessionRoot == "" {
|
|
return nil, fmt.Errorf("session root is required")
|
|
}
|
|
out := make([]archivePromotion, 0, len(rules))
|
|
for _, rule := range rules {
|
|
from := strings.TrimSpace(rule.From)
|
|
to := strings.TrimSpace(rule.To)
|
|
required := rule.Required == nil || *rule.Required
|
|
|
|
resolvedPath, err := resolveWorkDirRelativePath(sessionRoot, from)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("promotion from %q: %w", from, err)
|
|
}
|
|
info, err := os.Stat(resolvedPath)
|
|
exists := err == nil && !info.IsDir()
|
|
if err != nil && !os.IsNotExist(err) {
|
|
return nil, fmt.Errorf("promotion source %q: %w", from, err)
|
|
}
|
|
localPath := resolvedPath
|
|
|
|
out = append(out, archivePromotion{
|
|
From: from,
|
|
To: to,
|
|
Required: required,
|
|
LocalPath: localPath,
|
|
Exists: exists,
|
|
})
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func resolveWorkDirRelativePath(workDir, rel string) (string, error) {
|
|
rel = filepath.Clean(filepath.FromSlash(strings.TrimSpace(rel)))
|
|
if rel == "." || rel == "" {
|
|
return "", fmt.Errorf("relative path is required")
|
|
}
|
|
full := filepath.Join(workDir, rel)
|
|
cleanedWork := filepath.Clean(workDir)
|
|
cleanedFull := filepath.Clean(full)
|
|
relative, err := filepath.Rel(cleanedWork, cleanedFull)
|
|
if err != nil {
|
|
return "", fmt.Errorf("compute relative path: %w", err)
|
|
}
|
|
if relative == ".." || strings.HasPrefix(relative, ".."+string(filepath.Separator)) {
|
|
return "", fmt.Errorf("path escapes workdir")
|
|
}
|
|
return cleanedFull, nil
|
|
}
|
|
|
|
func collectArchiveRunFiles(runRoot, manifestPath string) ([]archiveUploadFile, error) {
|
|
files := make([]archiveUploadFile, 0, 64)
|
|
err := filepath.WalkDir(runRoot, func(path string, d fs.DirEntry, walkErr error) error {
|
|
if walkErr != nil {
|
|
return walkErr
|
|
}
|
|
if d.IsDir() {
|
|
if path == runRoot {
|
|
return nil
|
|
}
|
|
relDir, err := filepath.Rel(runRoot, path)
|
|
if err != nil {
|
|
return fmt.Errorf("relative dir from %q to %q: %w", runRoot, path, err)
|
|
}
|
|
relDir = filepath.ToSlash(relDir)
|
|
// Preserve existing behavior: audio is not uploaded in archive run record.
|
|
if relDir == "audio" || strings.HasPrefix(relDir, "audio/") {
|
|
return filepath.SkipDir
|
|
}
|
|
return nil
|
|
}
|
|
rel, err := filepath.Rel(runRoot, path)
|
|
if err != nil {
|
|
return fmt.Errorf("relative path from %q to %q: %w", runRoot, path, err)
|
|
}
|
|
rel = filepath.ToSlash(rel)
|
|
files = append(files, archiveUploadFile{
|
|
RelativePath: rel,
|
|
LocalPath: path,
|
|
})
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("walk %q: %w", runRoot, err)
|
|
}
|
|
|
|
manifestInfo, err := os.Stat(manifestPath)
|
|
if err != nil {
|
|
if os.IsNotExist(err) {
|
|
return nil, fmt.Errorf("manifest.json not found (checked path %q)", manifestPath)
|
|
}
|
|
return nil, fmt.Errorf("stat %q: %w", manifestPath, err)
|
|
}
|
|
if manifestInfo.IsDir() {
|
|
return nil, fmt.Errorf("manifest path %q is a directory", manifestPath)
|
|
}
|
|
files = append(files, archiveUploadFile{
|
|
RelativePath: "manifest.json",
|
|
LocalPath: manifestPath,
|
|
})
|
|
seen := map[string]archiveUploadFile{}
|
|
for _, file := range files {
|
|
seen[file.RelativePath] = file
|
|
}
|
|
files = files[:0]
|
|
for _, file := range seen {
|
|
files = append(files, file)
|
|
}
|
|
|
|
sort.Slice(files, func(i, j int) bool {
|
|
return files[i].RelativePath < files[j].RelativePath
|
|
})
|
|
return files, nil
|
|
}
|
|
|
|
func resolveArchiveRunManifestSource(runRoot string) (string, error) {
|
|
path := filepath.Join(filepath.Clean(runRoot), "manifest.json")
|
|
info, err := os.Stat(path)
|
|
if err != nil {
|
|
if os.IsNotExist(err) {
|
|
return "", fmt.Errorf("manifest.json not found in run root %q", runRoot)
|
|
}
|
|
return "", fmt.Errorf("stat %q: %w", path, err)
|
|
}
|
|
if info.IsDir() {
|
|
return "", fmt.Errorf("manifest path %q is a directory", path)
|
|
}
|
|
return path, nil
|
|
}
|
|
|
|
func directoryExists(path string) (bool, error) {
|
|
info, err := os.Stat(path)
|
|
if err == nil {
|
|
return info.IsDir(), nil
|
|
}
|
|
if os.IsNotExist(err) {
|
|
return false, nil
|
|
}
|
|
return false, err
|
|
}
|
|
|
|
func writeCurrentManifestSnapshot(m *manifest.Manifest, archiveMetadata map[string]any) (string, error) {
|
|
if m == nil {
|
|
return "", fmt.Errorf("manifest is required")
|
|
}
|
|
clone := *m
|
|
clone.Stages = make(map[string]*manifest.StageRecord, len(m.Stages))
|
|
for name, sr := range m.Stages {
|
|
if sr == nil {
|
|
continue
|
|
}
|
|
stageCopy := *sr
|
|
if sr.Outputs != nil {
|
|
stageCopy.Outputs = append([]manifest.ArtifactRecord(nil), sr.Outputs...)
|
|
}
|
|
if sr.Logs != nil {
|
|
stageCopy.Logs = append([]string(nil), sr.Logs...)
|
|
}
|
|
if sr.GeneratedConfigs != nil {
|
|
stageCopy.GeneratedConfigs = append([]string(nil), sr.GeneratedConfigs...)
|
|
}
|
|
if sr.Metadata != nil {
|
|
metaCopy := make(map[string]any, len(sr.Metadata))
|
|
for k, v := range sr.Metadata {
|
|
metaCopy[k] = v
|
|
}
|
|
stageCopy.Metadata = metaCopy
|
|
}
|
|
clone.Stages[name] = &stageCopy
|
|
}
|
|
|
|
now := time.Now().UTC()
|
|
clone.MarkStageSucceeded("archive", now, nil)
|
|
if sr := clone.Stages["archive"]; sr != nil {
|
|
sr.Metadata = archiveMetadata
|
|
}
|
|
|
|
data, err := json.MarshalIndent(&clone, "", " ")
|
|
if err != nil {
|
|
return "", fmt.Errorf("marshal manifest: %w", err)
|
|
}
|
|
data = append(data, '\n')
|
|
|
|
tmp, err := os.CreateTemp("", "narratio-current-manifest-*.json")
|
|
if err != nil {
|
|
return "", fmt.Errorf("create temp file: %w", err)
|
|
}
|
|
path := tmp.Name()
|
|
if _, err := tmp.Write(data); err != nil {
|
|
_ = tmp.Close()
|
|
return "", fmt.Errorf("write temp manifest: %w", err)
|
|
}
|
|
if err := tmp.Close(); err != nil {
|
|
return "", fmt.Errorf("close temp manifest: %w", err)
|
|
}
|
|
return path, nil
|
|
}
|
|
|
|
func writeCurrentRunIDPointer(runID string) (string, error) {
|
|
tmp, err := os.CreateTemp("", "narratio-current-run-id-*.txt")
|
|
if err != nil {
|
|
return "", fmt.Errorf("create temp file: %w", err)
|
|
}
|
|
path := tmp.Name()
|
|
if _, err := tmp.WriteString(runID + "\n"); err != nil {
|
|
_ = tmp.Close()
|
|
return "", fmt.Errorf("write temp run id pointer: %w", err)
|
|
}
|
|
if err := tmp.Close(); err != nil {
|
|
return "", fmt.Errorf("close temp run id pointer: %w", err)
|
|
}
|
|
return path, nil
|
|
}
|
|
|
|
func archiveMetadataPreview(
|
|
bucket, runPrefix, sessionPrefix string,
|
|
runUploaded []string,
|
|
promotedUploaded []string,
|
|
skippedOptional []string,
|
|
currentManifestKey string,
|
|
) map[string]any {
|
|
return map[string]any{
|
|
"stage": "archive",
|
|
"uploaded": true,
|
|
"s3_bucket": bucket,
|
|
"s3_run_prefix": runPrefix,
|
|
"run_files_uploaded": len(runUploaded),
|
|
"run_uploaded_paths": append([]string(nil), runUploaded...),
|
|
"promoted_files_uploaded": len(promotedUploaded),
|
|
"promoted_paths": append([]string(nil), promotedUploaded...),
|
|
"skipped_optional_promotions": append([]string(nil), skippedOptional...),
|
|
"current_manifest_key": currentManifestKey,
|
|
"current_run_id_key": artifacts.S3CurrentRunPointerKey(sessionPrefix),
|
|
"current_pointer_written": false,
|
|
"audio_upload_skipped": true,
|
|
}
|
|
}
|