526 lines
16 KiB
Go
526 lines
16 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{}
|
|
|
|
var archivePrerequisiteStages = []string{
|
|
"prepare",
|
|
"transcribe",
|
|
"merge",
|
|
"polish",
|
|
"normalize",
|
|
"trim",
|
|
"analyze",
|
|
}
|
|
|
|
var archiveRunUploadDirs = []string{
|
|
"inputs",
|
|
"transcripts",
|
|
"artifacts",
|
|
"reports",
|
|
"config",
|
|
"logs",
|
|
}
|
|
|
|
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")
|
|
}
|
|
|
|
workDir, err := archiveWorkDir(env, m)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: resolve local workdir: %w", err)
|
|
}
|
|
workDirInfo, err := os.Stat(workDir)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: local workdir %q: %w", workDir, err)
|
|
}
|
|
if !workDirInfo.IsDir() {
|
|
return nil, fmt.Errorf("archive: local workdir %q is not a directory", workDir)
|
|
}
|
|
|
|
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")
|
|
}
|
|
|
|
runFiles, err := collectArchiveRunFiles(workDir)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: collect run files: %w", err)
|
|
}
|
|
promotions, err := resolveArchivePromotions(workDir, env.Config.Pipeline.Archive.PromoteArtifacts)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("archive: resolve promotion rules: %w", err)
|
|
}
|
|
currentManifestSource := filepath.Join(workDir, "manifest.json")
|
|
if info, err := os.Stat(currentManifestSource); err != nil {
|
|
return nil, fmt.Errorf("archive: current manifest source %q: %w", currentManifestSource, err)
|
|
} else if info.IsDir() {
|
|
return nil, fmt.Errorf("archive: current manifest source %q is a directory", currentManifestSource)
|
|
}
|
|
|
|
runUploaded := make([]string, 0, len(runFiles))
|
|
for _, rel := range runFiles {
|
|
localPath := filepath.Join(workDir, filepath.FromSlash(rel))
|
|
key := artifacts.S3RunRelativeDestinationKey(runPrefix, rel)
|
|
if _, err := env.ObjectStore.Upload(ctx, localPath, key, storage.UploadOptions{}); err != nil {
|
|
return nil, fmt.Errorf("archive: upload run file %q to %q: %w", rel, key, err)
|
|
}
|
|
runUploaded = append(runUploaded, rel)
|
|
}
|
|
|
|
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 archiveWorkDir(env *Env, m *manifest.Manifest) (string, error) {
|
|
workDir := strings.TrimSpace(m.LocalWorkDir)
|
|
if workDir != "" {
|
|
cleaned := filepath.Clean(workDir)
|
|
if info, err := os.Stat(cleaned); err == nil && info.IsDir() {
|
|
return cleaned, 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)
|
|
}
|
|
runID := strings.TrimSpace(m.RunID)
|
|
if runID == "" {
|
|
return "", fmt.Errorf("run id is required")
|
|
}
|
|
if campaign == "" || sessionID == "" {
|
|
return "", fmt.Errorf("campaign and session id are required")
|
|
}
|
|
runScoped := artifacts.SessionRunWorkDir(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID)
|
|
if info, err := os.Stat(runScoped); err == nil && info.IsDir() {
|
|
return runScoped, nil
|
|
}
|
|
legacy := artifacts.SessionWorkDir(env.Config.Pipeline.Workspace.Root, sessionID)
|
|
return legacy, 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(workDir string, rules []config.ArchivePromotionRule) ([]archivePromotion, error) {
|
|
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
|
|
|
|
localPath, err := resolveWorkDirRelativePath(workDir, from)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("promotion from %q: %w", from, err)
|
|
}
|
|
info, err := os.Stat(localPath)
|
|
exists := err == nil && !info.IsDir()
|
|
if err != nil && !os.IsNotExist(err) {
|
|
return nil, fmt.Errorf("promotion source %q: %w", from, err)
|
|
}
|
|
|
|
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(workDir string) ([]string, error) {
|
|
files := make([]string, 0, 64)
|
|
|
|
for _, dirName := range archiveRunUploadDirs {
|
|
fullDir := filepath.Join(workDir, dirName)
|
|
info, err := os.Stat(fullDir)
|
|
if err != nil {
|
|
if os.IsNotExist(err) {
|
|
continue
|
|
}
|
|
return nil, fmt.Errorf("stat %q: %w", fullDir, err)
|
|
}
|
|
if !info.IsDir() {
|
|
continue
|
|
}
|
|
|
|
if err := filepath.WalkDir(fullDir, func(path string, d fs.DirEntry, walkErr error) error {
|
|
if walkErr != nil {
|
|
return walkErr
|
|
}
|
|
if d.IsDir() {
|
|
return nil
|
|
}
|
|
rel, err := filepath.Rel(workDir, path)
|
|
if err != nil {
|
|
return fmt.Errorf("relative path from %q to %q: %w", workDir, path, err)
|
|
}
|
|
rel = filepath.ToSlash(rel)
|
|
files = append(files, rel)
|
|
return nil
|
|
}); err != nil {
|
|
return nil, fmt.Errorf("walk %q: %w", fullDir, err)
|
|
}
|
|
}
|
|
|
|
manifestPath := filepath.Join(workDir, "manifest.json")
|
|
manifestInfo, err := os.Stat(manifestPath)
|
|
if err != nil {
|
|
if os.IsNotExist(err) {
|
|
return nil, fmt.Errorf("manifest.json not found in workdir %q", workDir)
|
|
}
|
|
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, "manifest.json")
|
|
|
|
sort.Strings(files)
|
|
return files, nil
|
|
}
|
|
|
|
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,
|
|
}
|
|
}
|