From be57e675e042714b5ad3ecdccf2e3252b7fcc1b6 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 23 May 2026 14:03:31 +0000 Subject: [PATCH] Split operator helper implementations by command responsibility --- docs/roadmap/cleanup.md | 4 +- internal/app/operator_artifact_rendering.go | 136 +++ internal/app/operator_artifacts_list.go | 39 + internal/app/operator_findings.go | 203 +++++ internal/app/operator_helpers.go | 951 -------------------- internal/app/operator_locks.go | 222 +++++ internal/app/operator_session_init.go | 268 ++++++ internal/app/operator_session_validate.go | 81 ++ internal/app/operator_status.go | 88 ++ 9 files changed, 1039 insertions(+), 953 deletions(-) create mode 100644 internal/app/operator_artifact_rendering.go create mode 100644 internal/app/operator_artifacts_list.go create mode 100644 internal/app/operator_findings.go create mode 100644 internal/app/operator_locks.go create mode 100644 internal/app/operator_session_init.go create mode 100644 internal/app/operator_session_validate.go create mode 100644 internal/app/operator_status.go diff --git a/docs/roadmap/cleanup.md b/docs/roadmap/cleanup.md index ab1e0c7..136ae0d 100644 --- a/docs/roadmap/cleanup.md +++ b/docs/roadmap/cleanup.md @@ -1,8 +1,8 @@ # Roadmap: Pre-1.0 Code Cleanup -Status: Planned +Status: Implemented (Stages 1-6 complete) -This roadmap turns the findings in `docs/roadmap/audit.md` into staged cleanup work for the 1.0 release. It is planning-only. Do not implement these refactors until a stage is explicitly selected for implementation. +This roadmap turns the findings in `docs/roadmap/audit.md` into staged cleanup work for the 1.0 release and records completion status for each selected stage. The cleanup work must follow the policy documents under `docs/policy/`, especially these invariants: diff --git a/internal/app/operator_artifact_rendering.go b/internal/app/operator_artifact_rendering.go new file mode 100644 index 0000000..62b7afb --- /dev/null +++ b/internal/app/operator_artifact_rendering.go @@ -0,0 +1,136 @@ +package app + +import ( + "context" + "fmt" + "io" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" + "gitea.maximumdirect.net/eric/narratio/internal/artifactpolicy" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/config" +) + +func buildHelperArtifactCatalog(cfg *config.Config) (*artifacts.ArtifactCatalog, error) { + catalog := artifacts.NewArtifactCatalog() + if err := catalog.RegisterBuiltIns(); err != nil { + return nil, err + } + configured := map[string]artifacts.ConfiguredArtifactDefinition{} + if cfg.Pipeline.Scriptorium != nil { + for key, item := range cfg.Pipeline.Scriptorium.Artifacts { + configured[key] = artifacts.ConfiguredArtifactDefinition{Enabled: item.Enabled, OutputPath: item.OutputPath} + } + } + if err := catalog.RegisterConfiguredArtifacts(configured, nil); err != nil { + return nil, err + } + return catalog, nil +} + +func writeArtifactList(out io.Writer, cfg *config.Config, catalog *artifacts.ArtifactCatalog, locks *effectiveLocks, publishedRemoteState map[string]string) { + lockSet := lockSourceSet(locks.All) + fmt.Fprintln(out, "Built-in:") + for _, id := range []string{ + artifacts.ArtifactTranscriptBase, + artifacts.ArtifactTranscriptPolished, + artifacts.ArtifactTranscriptFinal, + artifacts.ArtifactTranscriptFinalTrimmed, + artifacts.ArtifactBoundsSession, + } { + writeArtifactLine(out, id, lockSet) + } + fmt.Fprintln(out, "Configured:") + for _, entry := range catalog.ListConfigured() { + writeArtifactLine(out, entry.SourceID, lockSet) + } + fmt.Fprintln(out, "Previous-session:") + for _, req := range artifacts.CollectPreviousArtifactRequirements(configuredScriptoriumArtifacts(cfg)) { + fmt.Fprintf(out, "- %s required=%t\n", artifactpolicy.PreviousSessionSourceID(req.Name), req.Required) + } + fmt.Fprintln(out, "Published:") + for _, rule := range cfg.Pipeline.Publish.Outputs { + writePublishedOutputLine(out, rule, catalog, lockSet, publishedRemoteState) + } +} + +func writeArtifactLine(out io.Writer, source string, lockSet map[string]config.PublishLockRule) { + parts := []string{source} + if _, ok := lockSet[source]; ok { + parts = append(parts, "locked") + } + fmt.Fprintf(out, "- %s\n", strings.Join(parts, " ")) +} + +func writePublishedOutputLine(out io.Writer, rule config.PublishOutputRule, catalog *artifacts.ArtifactCatalog, lockSet map[string]config.PublishLockRule, remoteState map[string]string) { + source := strings.TrimSpace(rule.Source) + parts := []string{source} + if _, ok := lockSet[source]; ok { + parts = append(parts, "locked") + } + dest, showDest, err := helperPublishedOutputDest(rule, catalog) + if err != nil { + parts = append(parts, "remote=error") + fmt.Fprintf(out, "- %s\n", strings.Join(parts, " ")) + return + } + if showDest { + parts = append(parts, "dest="+dest) + } + if state := remoteState[publishedOutputRemoteStateKey(source, dest)]; state != "" { + parts = append(parts, state) + } + fmt.Fprintf(out, "- %s\n", strings.Join(parts, " ")) +} + +func remotePublishedOutputAvailability(ctx context.Context, cfg *config.Config, store storage.ObjectStore, catalog *artifacts.ArtifactCatalog) map[string]string { + out := map[string]string{} + sessionPrefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, cfg.Session.Campaign, cfg.Session.SessionID) + for _, rule := range cfg.Pipeline.Publish.Outputs { + source := strings.TrimSpace(rule.Source) + dest, _, err := helperPublishedOutputDest(rule, catalog) + if err != nil { + out[publishedOutputRemoteStateKey(source, "")] = "remote=error" + continue + } + key := artifacts.S3PublishedOutputKey(sessionPrefix, dest) + if exists, err := store.Exists(ctx, key); err == nil && exists { + out[publishedOutputRemoteStateKey(source, dest)] = "remote=published" + } else if err != nil { + out[publishedOutputRemoteStateKey(source, dest)] = "remote=error" + } else { + out[publishedOutputRemoteStateKey(source, dest)] = "remote=missing" + } + } + return out +} + +func helperPublishedOutputDest(rule config.PublishOutputRule, catalog *artifacts.ArtifactCatalog) (string, bool, error) { + source := strings.TrimSpace(rule.Source) + normalized, err := artifactpolicy.ResolvePublishedDestination(source, rule.Dest, helperConfiguredOutputPathMap(catalog)) + if err != nil { + return "", false, err + } + entry, ok := catalog.Lookup(source) + showDest := !ok || strings.TrimSpace(entry.CanonicalRelPath) != normalized + return normalized, showDest, nil +} + +func helperConfiguredOutputPathMap(catalog *artifacts.ArtifactCatalog) map[string]string { + out := map[string]string{} + if catalog == nil { + return out + } + for _, entry := range catalog.ListConfigured() { + if strings.TrimSpace(entry.ConfiguredKey) == "" { + continue + } + out[entry.ConfiguredKey] = strings.TrimSpace(entry.CanonicalRelPath) + } + return out +} + +func publishedOutputRemoteStateKey(source, dest string) string { + return strings.TrimSpace(source) + "\x00" + strings.TrimSpace(dest) +} diff --git a/internal/app/operator_artifacts_list.go b/internal/app/operator_artifacts_list.go new file mode 100644 index 0000000..5960425 --- /dev/null +++ b/internal/app/operator_artifacts_list.go @@ -0,0 +1,39 @@ +package app + +import ( + "context" + "flag" + "fmt" + "io" + "strings" +) + +// ArtifactsList lists effective artifact sources. +func ArtifactsList(ctx context.Context, args []string, out io.Writer) error { + fs := flag.NewFlagSet("artifacts list", flag.ContinueOnError) + fs.SetOutput(io.Discard) + var flags commonConfigFlags + var remote bool + addCommonConfigFlags(fs, &flags) + fs.BoolVar(&remote, "remote", false, "inspect remote publish availability") + if err := parseSessionAwareFlags("artifacts list", fs, args, &flags.sessionID); err != nil { + return err + } + if strings.TrimSpace(flags.sessionID) == "" { + return fmt.Errorf("artifacts list: session_id is required") + } + cfg, store, locks, _, err := loadHelperContext(ctx, flags, remote) + if err != nil { + return fmt.Errorf("artifacts list: %w", err) + } + catalog, err := buildHelperArtifactCatalog(cfg) + if err != nil { + return fmt.Errorf("artifacts list: %w", err) + } + publishedRemoteState := map[string]string{} + if remote && store != nil { + publishedRemoteState = remotePublishedOutputAvailability(ctx, cfg, store, catalog) + } + writeArtifactList(out, cfg, catalog, locks, publishedRemoteState) + return nil +} diff --git a/internal/app/operator_findings.go b/internal/app/operator_findings.go new file mode 100644 index 0000000..cf89a0b --- /dev/null +++ b/internal/app/operator_findings.go @@ -0,0 +1,203 @@ +package app + +import ( + "context" + "fmt" + "io" + "os" + "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" +) + +type finding struct { + Severity string + Category string + Message string +} + +type findingError struct { + count int +} + +func (e findingError) Error() string { + return fmt.Sprintf("%d validation error(s)", e.count) +} + +func renderFindings(out io.Writer, campaign, sessionID string, findings []finding) error { + if campaign != "" || sessionID != "" { + fmt.Fprintf(out, "Campaign: %s\n", campaign) + fmt.Fprintf(out, "Session: %s\n\n", sessionID) + } + errorsCount := 0 + for _, f := range findings { + if f.Severity == "ERROR" { + errorsCount++ + } + fmt.Fprintf(out, "%-5s %-10s %s\n", f.Severity, f.Category, f.Message) + } + if errorsCount > 0 { + return findingError{count: errorsCount} + } + return nil +} + +func okFinding(category, msg string) finding { return finding{"OK", category, msg} } +func infoFinding(category, msg string) finding { return finding{"INFO", category, msg} } +func warnFinding(category, msg string) finding { return finding{"WARN", category, msg} } +func errorFinding(category, msg string) finding { return finding{"ERROR", category, msg} } + +func sessionSourceSummary(cfg *config.Config) string { + source := cfg.SessionSource.Source + if source == "" { + source = "session_config" + } + if cfg.SessionSource.S3Key != "" { + return source + " " + cfg.SessionSource.S3Key + } + return source + " " + cfg.SessionPath +} + +func validateStableInputFindings(cfg *config.Config) []finding { + items := []struct { + name string + in config.ResolvedInputFile + }{ + {"speakers", cfg.StableInputs.SpeakersFile}, + {"autocorrect", cfg.StableInputs.AutocorrectFile}, + {"glossary", cfg.StableInputs.GlossaryFile}, + } + out := make([]finding, 0, len(items)) + for _, item := range items { + path, err := resolveHelperConfigRelativePath(item.in) + if err != nil { + out = append(out, errorFinding("inputs", item.name+": "+err.Error())) + continue + } + if _, err := os.Stat(path); err != nil { + out = append(out, errorFinding("inputs", fmt.Sprintf("%s missing: %v", item.name, err))) + } else { + out = append(out, okFinding("inputs", item.name+": "+path)) + } + } + return out +} + +func resolveHelperConfigRelativePath(input config.ResolvedInputFile) (string, error) { + if strings.TrimSpace(input.ConfigPath) == "" { + return "", fmt.Errorf("source config path is required") + } + path := strings.TrimSpace(input.Path) + if path == "" { + return "", fmt.Errorf("path is required") + } + if filepath.IsAbs(path) { + return filepath.Clean(path), nil + } + return filepath.Clean(filepath.Join(filepath.Dir(input.ConfigPath), path)), nil +} + +func validateLocalAudioFindings(cfg *config.Config) []finding { + if cfg.Session.Inputs.AudioS3 != nil { + return nil + } + audioDir := strings.TrimSpace(cfg.Session.Inputs.AudioDir) + if audioDir == "" && len(cfg.Session.Inputs.AudioFiles) == 0 { + return []finding{errorFinding("audio", "audio_dir, audio_files, or audio_s3 is required")} + } + base := filepath.Dir(cfg.SessionPath) + paths := []string{} + if audioDir != "" { + dir := audioDir + if !filepath.IsAbs(dir) { + dir = filepath.Join(base, dir) + } + matches, err := filepath.Glob(filepath.Join(dir, "*.flac")) + if err != nil || len(matches) == 0 { + return []finding{errorFinding("audio", "no .flac files found in "+dir)} + } + paths = append(paths, matches...) + } + for _, file := range cfg.Session.Inputs.AudioFiles { + p := file + if !filepath.IsAbs(p) { + p = filepath.Join(base, p) + } + paths = append(paths, p) + } + for _, p := range paths { + if _, err := os.Stat(p); err != nil { + return []finding{errorFinding("audio", fmt.Sprintf("audio file missing: %v", err))} + } + } + return []finding{okFinding("audio", fmt.Sprintf("%d local audio file(s)", len(paths)))} +} + +func validateRemoteAudioFinding(ctx context.Context, cfg *config.Config, store storage.ObjectStore) finding { + sessionPrefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, cfg.Session.Campaign, cfg.Session.SessionID) + audioPrefix := artifacts.S3AudioPrefix(sessionPrefix, cfg.Session.Inputs.AudioS3.Prefix) + objects, err := store.List(ctx, audioPrefix) + if err != nil { + return errorFinding("audio", err.Error()) + } + count := 0 + for _, obj := range objects { + if strings.HasSuffix(strings.ToLower(obj.Key), ".flac") { + count++ + } + } + if count == 0 { + return errorFinding("audio", "no remote .flac objects found under "+audioPrefix) + } + return okFinding("audio", fmt.Sprintf("%d remote .flac object(s)", count)) +} + +func validatePreviousArtifactFindings(ctx context.Context, cfg *config.Config, store storage.ObjectStore, requirements []artifacts.PreviousArtifactRequirement) []finding { + out := []finding{} + prefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, cfg.Session.Campaign, cfg.Session.PreviousSessionID) + _, err := artifacts.LoadCurrentState(ctx, store, prefix, artifacts.CurrentStateValidation{ + ExpectedSessionID: strings.TrimSpace(cfg.Session.PreviousSessionID), + ExpectedCampaign: strings.TrimSpace(cfg.Session.Campaign), + ValidateRunID: true, + }) + if err != nil { + out = append(out, errorFinding("previous", fmt.Sprintf("remote %v", err))) + return out + } + for _, req := range requirements { + out = append(out, okFinding("previous", fmt.Sprintf("%s required=%t", req.Name, req.Required))) + } + return out +} + +func loadLocalManifest(ctx context.Context, path string) (*manifest.Manifest, error) { + if _, err := os.Stat(path); err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, err + } + store := &manifest.LocalStore{} + return store.Load(ctx, path) +} + +func writeStageStatuses(out io.Writer, m *manifest.Manifest) { + if m == nil || len(m.Stages) == 0 { + fmt.Fprintln(out, "stages: no stages recorded") + return + } + fmt.Fprintln(out, "stages:") + names := make([]string, 0, len(m.Stages)) + for name := range m.Stages { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + fmt.Fprintf(out, "- %s: %s\n", name, m.Stages[name].Status) + } +} diff --git a/internal/app/operator_helpers.go b/internal/app/operator_helpers.go index 37efb08..f6725a6 100644 --- a/internal/app/operator_helpers.go +++ b/internal/app/operator_helpers.go @@ -5,18 +5,12 @@ import ( "flag" "fmt" "io" - "os" - "path/filepath" - "regexp" - "sort" "strings" "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" - "gitea.maximumdirect.net/eric/narratio/internal/artifactpolicy" "gitea.maximumdirect.net/eric/narratio/internal/artifacts" "gitea.maximumdirect.net/eric/narratio/internal/config" "gitea.maximumdirect.net/eric/narratio/internal/manifest" - "gopkg.in/yaml.v3" ) type commonConfigFlags struct { @@ -28,20 +22,6 @@ type commonConfigFlags struct { previousSessionID string } -type finding struct { - Severity string - Category string - Message string -} - -type findingError struct { - count int -} - -func (e findingError) Error() string { - return fmt.Sprintf("%d validation error(s)", e.count) -} - func addCommonConfigFlags(fs *flag.FlagSet, flags *commonConfigFlags) { fs.StringVar(&flags.pipelinePath, "config", "", "path to pipeline.yml (optional; defaults searched)") fs.StringVar(&flags.campaignPath, "campaign", "", "campaign ID") @@ -110,456 +90,6 @@ func Artifacts(ctx context.Context, args []string, out io.Writer) error { } } -// SessionValidate performs a read-only session preflight. -func SessionValidate(ctx context.Context, args []string, out io.Writer) error { - fs := flag.NewFlagSet("session validate", flag.ContinueOnError) - fs.SetOutput(io.Discard) - var flags commonConfigFlags - addCommonConfigFlags(fs, &flags) - if err := parseSessionAwareFlags("session validate", fs, args, &flags.sessionID); err != nil { - return err - } - if strings.TrimSpace(flags.sessionID) == "" { - return fmt.Errorf("session validate: session_id is required") - } - - findings := []finding{} - cfg, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) - if err != nil { - findings = append(findings, errorFinding("config", err.Error())) - return renderFindings(out, "", "", findings) - } - if err := config.Validate(cfg); err != nil { - findings = append(findings, errorFinding("config", err.Error())) - } else { - findings = append(findings, okFinding("config", "resolved pipeline, campaign, and session config")) - } - findings = append(findings, okFinding("session", fmt.Sprintf("session source: %s", sessionSourceSummary(cfg)))) - - paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID) - findings = append(findings, validateStableInputFindings(cfg)...) - findings = append(findings, validateLocalAudioFindings(cfg)...) - - store, storeErr := objectStoreIfConfigured(ctx, cfg) - if storeErr != nil { - findings = append(findings, errorFinding("storage", storeErr.Error())) - } - if cfg.Session.Inputs.AudioS3 != nil { - if storeErr != nil { - findings = append(findings, errorFinding("audio", "remote audio cannot be checked because storage is unavailable")) - } else { - findings = append(findings, validateRemoteAudioFinding(ctx, cfg, store)) - } - } - - requirements := artifacts.CollectPreviousArtifactRequirements(configuredScriptoriumArtifacts(cfg)) - if len(requirements) == 0 { - findings = append(findings, okFinding("previous", "no previous-session artifacts required")) - } else if strings.TrimSpace(cfg.Session.PreviousSessionID) == "" { - findings = append(findings, errorFinding("previous", "previous_session_id is required by configured previous-session artifacts")) - } else if storeErr != nil { - findings = append(findings, errorFinding("previous", "previous-session artifacts cannot be checked because storage is unavailable")) - } else { - findings = append(findings, validatePreviousArtifactFindings(ctx, cfg, store, requirements)...) - } - - locks, lockErr := loadEffectiveLocks(ctx, cfg, store) - if lockErr != nil { - findings = append(findings, errorFinding("locks", lockErr.Error())) - } else if len(locks.All) == 0 { - findings = append(findings, okFinding("locks", "no effective publish locks")) - } else { - for _, lock := range locks.All { - findings = append(findings, warnFinding("locks", fmt.Sprintf("%s locked: %s", lock.Source, strings.TrimSpace(lock.Reason)))) - } - } - if paths.ManifestPath != "" { - findings = append(findings, infoFinding("workspace", "manifest path: "+paths.ManifestPath)) - } - return renderFindings(out, cfg.Session.Campaign, cfg.Session.SessionID, findings) -} - -// Status reports effective local/remote session state. -func Status(ctx context.Context, args []string, out io.Writer) error { - fs := flag.NewFlagSet("status", flag.ContinueOnError) - fs.SetOutput(io.Discard) - var flags commonConfigFlags - addCommonConfigFlags(fs, &flags) - if err := parseSessionAwareFlags("status", fs, args, &flags.sessionID); err != nil { - return err - } - if strings.TrimSpace(flags.sessionID) == "" { - return fmt.Errorf("status: session_id is required") - } - cfg, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) - if err != nil { - return fmt.Errorf("status: %w", err) - } - if err := config.Validate(cfg); err != nil { - return fmt.Errorf("status: %w", err) - } - paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID) - fmt.Fprintf(out, "Session: %s\n", cfg.Session.SessionID) - fmt.Fprintf(out, "Campaign: %s\n", cfg.Session.Campaign) - fmt.Fprintf(out, "Workspace: %s\n", paths.Root) - fmt.Fprintf(out, "Session config: %s\n", sessionSourceSummary(cfg)) - - if m, err := loadLocalManifest(ctx, paths.ManifestPath); err != nil { - fmt.Fprintf(out, "Local manifest: error: %v\n", err) - } else if m == nil { - fmt.Fprintln(out, "Local manifest: missing") - } else { - fmt.Fprintf(out, "Local manifest: %s\n", paths.ManifestPath) - writeStageStatuses(out, m) - } - - store, storeErr := objectStoreIfConfigured(ctx, cfg) - if storeErr != nil { - fmt.Fprintf(out, "Remote publish: unavailable: %v\n", storeErr) - } else if store != nil { - current, err := discoverRemoteCurrentStateFn(ctx, cfg, store) - if err != nil { - fmt.Fprintf(out, "Remote publish: missing or unavailable: %v\n", err) - } else { - fmt.Fprintf(out, "Remote publish: current run %s\n", current.RunID) - fmt.Fprintf(out, "Remote manifest: %s\n", current.CurrentManifestKey) - } - } - - locks, err := loadEffectiveLocks(ctx, cfg, store) - if catalog, catalogErr := buildHelperArtifactCatalog(cfg); catalogErr != nil { - fmt.Fprintf(out, "Remote outputs: error: %v\n", catalogErr) - } else if storeErr == nil { - catalogLocks := locks - if err != nil { - catalogLocks = &effectiveLocks{ - Static: staticPublishLocks(cfg), - All: staticPublishLocks(cfg), - } - } - publishedRemoteState := map[string]string{} - if store != nil { - publishedRemoteState = remotePublishedOutputAvailability(ctx, cfg, store, catalog) - } - fmt.Fprintln(out, "Remote outputs:") - writeArtifactList(out, cfg, catalog, catalogLocks, publishedRemoteState) - } - if err != nil { - fmt.Fprintf(out, "Publish locks: error: %v\n", err) - } else { - writeLocks(out, cfg, locks) - } - fmt.Fprintln(out, "Next actions:") - fmt.Fprintf(out, "- narratio session validate %s\n", cfg.Session.SessionID) - fmt.Fprintf(out, "- narratio session restore %s --dry-run\n", cfg.Session.SessionID) - return nil -} - -// SessionInit creates a local or remote session.yml skeleton. -func SessionInit(ctx context.Context, args []string, out io.Writer) error { - fs := flag.NewFlagSet("session init", flag.ContinueOnError) - fs.SetOutput(io.Discard) - var pipelinePath, campaignPath, campaignFilePath, sessionID, previousSessionID, date, title, output, audioS3Prefix, audioDir string - var remote, force bool - fs.StringVar(&pipelinePath, "config", "", "path to pipeline.yml (optional; defaults searched)") - fs.StringVar(&campaignPath, "campaign", "", "campaign ID") - fs.StringVar(&campaignFilePath, "campaign-file", "", "path to campaign.yml") - fs.StringVar(&sessionID, "session-id", "", "session identifier") - fs.StringVar(&previousSessionID, "previous-session-id", "", "previous session identifier") - fs.StringVar(&date, "date", "", "session date") - fs.StringVar(&title, "title", "", "session title") - fs.StringVar(&output, "output", "", "local output session.yml path") - fs.StringVar(&audioS3Prefix, "audio-s3-prefix", "", "session audio S3 prefix") - fs.StringVar(&audioDir, "audio-dir", "", "local audio directory") - fs.BoolVar(&remote, "remote", false, "write session.yml to S3 session prefix") - fs.BoolVar(&force, "force", false, "overwrite existing target") - if err := parseSessionAwareFlags("session init", fs, args, &sessionID); err != nil { - return err - } - if strings.TrimSpace(sessionID) == "" { - return fmt.Errorf("session init: session_id is required") - } - if (strings.TrimSpace(output) == "") == !remote { - return fmt.Errorf("session init: specify exactly one target: --output or --remote") - } - if strings.TrimSpace(audioDir) != "" && strings.TrimSpace(audioS3Prefix) != "" { - return fmt.Errorf("session init: --audio-dir and --audio-s3-prefix are mutually exclusive") - } - - base, err := loadPipelineCampaignConfig(pipelinePath, campaignPath, campaignFilePath) - if err != nil { - return fmt.Errorf("session init: %w", err) - } - - input := sessionInitInput{ - Campaign: config.CampaignID(base.Campaign), - CampaignPath: base.CampaignPath, - TemplateFile: base.Campaign.SessionTemplateFile, - SessionID: sessionID, - PreviousSessionID: previousSessionID, - Date: date, - Title: title, - AudioS3Prefix: audioS3Prefix, - AudioDir: audioDir, - } - data, err := buildSessionInitYAML(input) - if err != nil { - return fmt.Errorf("session init: %w", err) - } - label := strings.TrimSpace(output) - if label == "" { - label = "remote session.yml" - } - sessionCfg, err := config.LoadSessionBytesWithOptions(label, data, config.SessionLoadOptions{ - SessionID: sessionID, - PreviousSessionID: previousSessionID, - }) - if err != nil { - return fmt.Errorf("session init: %w", err) - } - cfg, err := config.Resolve(base.PipelinePath, base.Pipeline, base.CampaignPath, base.Campaign, label, sessionCfg, config.SessionSource{Source: "session_config", LocalPath: label}) - if err != nil { - return fmt.Errorf("session init: %w", err) - } - if err := config.Validate(cfg); err != nil { - return fmt.Errorf("session init: %w", err) - } - - if !remote { - if err := writeLocalFile(output, data, force); err != nil { - return fmt.Errorf("session init: %w", err) - } - _, err := fmt.Fprintf(out, "narratio session init: wrote %s\n", filepath.Clean(output)) - return err - } - - store, err := newCommandObjectStore(ctx, cfg, nil) - if err != nil { - return fmt.Errorf("session init: %w", err) - } - sessionPrefix := artifacts.S3SessionPrefix(base.Pipeline.Storage.S3.RootPrefix, config.CampaignID(base.Campaign), sessionID) - key := artifacts.S3SessionConfigKey(sessionPrefix) - exists, err := store.Exists(ctx, key) - if err != nil { - return fmt.Errorf("session init: check remote session %q: %w", key, err) - } - if exists && !force { - return fmt.Errorf("session init: remote session %q already exists; pass --force to overwrite", key) - } - tmp, err := os.CreateTemp("", "narratio-session-init-*.yml") - if err != nil { - return fmt.Errorf("session init: create temp file: %w", err) - } - tmpPath := tmp.Name() - defer func() { _ = os.Remove(tmpPath) }() - if _, err := tmp.Write(data); err != nil { - _ = tmp.Close() - return fmt.Errorf("session init: write temp file: %w", err) - } - if err := tmp.Close(); err != nil { - return fmt.Errorf("session init: close temp file: %w", err) - } - if _, err := store.Upload(ctx, tmpPath, key, storage.UploadOptions{ContentType: "application/x-yaml; charset=utf-8"}); err != nil { - return fmt.Errorf("session init: upload remote session %q: %w", key, err) - } - _, err = fmt.Fprintf(out, "narratio session init: wrote s3://%s/%s\n", s3BucketName(base.Pipeline), key) - return err -} - -// ArtifactsList lists effective artifact sources. -func ArtifactsList(ctx context.Context, args []string, out io.Writer) error { - fs := flag.NewFlagSet("artifacts list", flag.ContinueOnError) - fs.SetOutput(io.Discard) - var flags commonConfigFlags - var remote bool - addCommonConfigFlags(fs, &flags) - fs.BoolVar(&remote, "remote", false, "inspect remote publish availability") - if err := parseSessionAwareFlags("artifacts list", fs, args, &flags.sessionID); err != nil { - return err - } - if strings.TrimSpace(flags.sessionID) == "" { - return fmt.Errorf("artifacts list: session_id is required") - } - cfg, store, locks, _, err := loadHelperContext(ctx, flags, remote) - if err != nil { - return fmt.Errorf("artifacts list: %w", err) - } - catalog, err := buildHelperArtifactCatalog(cfg) - if err != nil { - return fmt.Errorf("artifacts list: %w", err) - } - publishedRemoteState := map[string]string{} - if remote && store != nil { - publishedRemoteState = remotePublishedOutputAvailability(ctx, cfg, store, catalog) - } - writeArtifactList(out, cfg, catalog, locks, publishedRemoteState) - return nil -} - -// Locks dispatches publish lock list and mutation helpers. -func Locks(ctx context.Context, args []string, out io.Writer) error { - if len(args) > 0 && !strings.HasPrefix(args[0], "-") { - switch args[0] { - case "add": - return LocksAdd(ctx, args[1:], out) - case "remove": - return LocksRemove(ctx, args[1:], out) - default: - return fmt.Errorf("locks: unknown subcommand %q", args[0]) - } - } - return LocksList(ctx, args, out) -} - -// LocksList lists effective publish locks. -func LocksList(ctx context.Context, args []string, out io.Writer) error { - fs := flag.NewFlagSet("locks", flag.ContinueOnError) - fs.SetOutput(io.Discard) - var flags commonConfigFlags - addCommonConfigFlags(fs, &flags) - if err := parseSessionAwareFlags("locks", fs, args, &flags.sessionID); err != nil { - return err - } - if strings.TrimSpace(flags.sessionID) == "" { - return fmt.Errorf("locks: session_id is required") - } - cfg, _, locks, _, err := loadHelperContext(ctx, flags, true) - if err != nil { - return fmt.Errorf("locks: %w", err) - } - writeLocks(out, cfg, locks) - return nil -} - -// LocksAdd adds or updates one remote lock. -func LocksAdd(ctx context.Context, args []string, out io.Writer) error { - var positionalSessionID string - var source string - if len(args) >= 2 && !isCLIFlagToken(args[0]) && !isCLIFlagToken(args[1]) { - positionalSessionID = strings.TrimSpace(args[0]) - source = strings.TrimSpace(args[1]) - args = append([]string(nil), args[2:]...) - } - fs := flag.NewFlagSet("locks add", flag.ContinueOnError) - fs.SetOutput(io.Discard) - var flags commonConfigFlags - var reason string - var force bool - addCommonConfigFlags(fs, &flags) - fs.StringVar(&reason, "reason", "", "lock reason") - fs.BoolVar(&force, "force", false, "update existing remote lock") - if err := fs.Parse(args); err != nil { - return fmt.Errorf("locks add: invalid flags: %w", err) - } - if source == "" { - switch fs.NArg() { - case 2: - positionalSessionID = strings.TrimSpace(fs.Arg(0)) - source = strings.TrimSpace(fs.Arg(1)) - case 1: - if strings.TrimSpace(flags.sessionID) == "" { - return fmt.Errorf("locks add: expected session_id and source id") - } - source = strings.TrimSpace(fs.Arg(0)) - default: - return fmt.Errorf("locks add: expected session_id and source id") - } - } else if fs.NArg() != 0 { - return fmt.Errorf("locks add: unexpected positional arguments") - } - if err := applyPositionalSessionID("locks add", positionalSessionID, &flags.sessionID); err != nil { - return err - } - if strings.TrimSpace(flags.sessionID) == "" { - return fmt.Errorf("locks add: session_id is required") - } - cfg, store, locks, _, err := loadHelperContext(ctx, flags, true) - if err != nil { - return fmt.Errorf("locks add: %w", err) - } - if _, err := config.ValidatePublishLockRules([]config.PublishLockRule{{Source: source}}, cfg.Pipeline.Scriptorium, "locks add"); err != nil { - return fmt.Errorf("locks add: %w", err) - } - if _, ok := lockSourceSet(locks.Static)[source]; ok { - return fmt.Errorf("locks add: source %q is locked by pipeline config and cannot be modified remotely", source) - } - remoteSet := lockSourceSet(locks.Remote) - if _, exists := remoteSet[source]; exists && !force { - return fmt.Errorf("locks add: remote lock for %q already exists; pass --force to update", source) - } - remoteSet[source] = config.PublishLockRule{Source: source, Reason: strings.TrimSpace(reason)} - remoteLocks := lockMapValues(remoteSet) - if _, err := config.ValidatePublishLockRules(remoteLocks, cfg.Pipeline.Scriptorium, "locks"); err != nil { - return fmt.Errorf("locks add: %w", err) - } - if err := uploadRemoteLockStore(ctx, store, locks.Key, &config.PublishLockStore{Locks: remoteLocks}); err != nil { - return fmt.Errorf("locks add: %w", err) - } - _, err = fmt.Fprintf(out, "narratio session locks add: locked %s\n", source) - return err -} - -// LocksRemove removes one remote lock. -func LocksRemove(ctx context.Context, args []string, out io.Writer) error { - var positionalSessionID string - var source string - if len(args) >= 2 && !isCLIFlagToken(args[0]) && !isCLIFlagToken(args[1]) { - positionalSessionID = strings.TrimSpace(args[0]) - source = strings.TrimSpace(args[1]) - args = append([]string(nil), args[2:]...) - } - fs := flag.NewFlagSet("locks remove", flag.ContinueOnError) - fs.SetOutput(io.Discard) - var flags commonConfigFlags - addCommonConfigFlags(fs, &flags) - if err := fs.Parse(args); err != nil { - return fmt.Errorf("locks remove: invalid flags: %w", err) - } - if source == "" { - switch fs.NArg() { - case 2: - positionalSessionID = strings.TrimSpace(fs.Arg(0)) - source = strings.TrimSpace(fs.Arg(1)) - case 1: - if strings.TrimSpace(flags.sessionID) == "" { - return fmt.Errorf("locks remove: expected session_id and source id") - } - source = strings.TrimSpace(fs.Arg(0)) - default: - return fmt.Errorf("locks remove: expected session_id and source id") - } - } else if fs.NArg() != 0 { - return fmt.Errorf("locks remove: unexpected positional arguments") - } - if err := applyPositionalSessionID("locks remove", positionalSessionID, &flags.sessionID); err != nil { - return err - } - if strings.TrimSpace(flags.sessionID) == "" { - return fmt.Errorf("locks remove: session_id is required") - } - cfg, store, locks, _, err := loadHelperContext(ctx, flags, true) - if err != nil { - return fmt.Errorf("locks remove: %w", err) - } - if _, err := config.ValidatePublishLockRules([]config.PublishLockRule{{Source: source}}, cfg.Pipeline.Scriptorium, "locks remove"); err != nil { - return fmt.Errorf("locks remove: %w", err) - } - remoteSet := lockSourceSet(locks.Remote) - if _, ok := remoteSet[source]; !ok { - if _, static := lockSourceSet(locks.Static)[source]; static { - return fmt.Errorf("locks remove: source %q is locked by pipeline config and cannot be unlocked remotely", source) - } - return fmt.Errorf("locks remove: remote lock for %q does not exist", source) - } - delete(remoteSet, source) - remoteLocks := lockMapValues(remoteSet) - if err := uploadRemoteLockStore(ctx, store, locks.Key, &config.PublishLockStore{Locks: remoteLocks}); err != nil { - return fmt.Errorf("locks remove: %w", err) - } - _, err = fmt.Fprintf(out, "narratio session locks remove: unlocked %s\n", source) - return err -} - func loadHelperContext(ctx context.Context, flags commonConfigFlags, needStore bool) (*config.Config, storage.ObjectStore, *effectiveLocks, *manifest.Manifest, error) { cfg, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) if err != nil { @@ -599,484 +129,3 @@ func objectStoreIfConfigured(ctx context.Context, cfg *config.Config) (storage.O } return store, nil } - -func buildSessionYAML(campaign, sessionID, previousSessionID, date, title, audioS3Prefix, audioDir string) ([]byte, error) { - if strings.TrimSpace(date) == "" && regexp.MustCompile(`^\d{4}-\d{2}-\d{2}$`).MatchString(strings.TrimSpace(sessionID)) { - date = strings.TrimSpace(sessionID) - } - type audioS3 struct { - Prefix string `yaml:"prefix"` - } - type inputs struct { - AudioDir string `yaml:"audio_dir,omitempty"` - AudioS3 *audioS3 `yaml:"audio_s3,omitempty"` - } - type sessionYAML struct { - Campaign string `yaml:"campaign"` - SessionID string `yaml:"session_id"` - PreviousSessionID string `yaml:"previous_session_id,omitempty"` - Date string `yaml:"date,omitempty"` - Title string `yaml:"title,omitempty"` - Inputs inputs `yaml:"inputs"` - } - in := inputs{AudioDir: strings.TrimSpace(audioDir)} - if in.AudioDir == "" { - prefix := strings.TrimSpace(audioS3Prefix) - if prefix == "" { - prefix = "audio/" - } - in.AudioS3 = &audioS3{Prefix: prefix} - } - data, err := yaml.Marshal(sessionYAML{ - Campaign: strings.TrimSpace(campaign), - SessionID: strings.TrimSpace(sessionID), - PreviousSessionID: strings.TrimSpace(previousSessionID), - Date: strings.TrimSpace(date), - Title: strings.TrimSpace(title), - Inputs: in, - }) - if err != nil { - return nil, err - } - return data, nil -} - -type sessionInitInput struct { - Campaign string - CampaignPath string - TemplateFile string - SessionID string - PreviousSessionID string - Date string - Title string - AudioS3Prefix string - AudioDir string -} - -func buildSessionInitYAML(in sessionInitInput) ([]byte, error) { - if strings.TrimSpace(in.TemplateFile) == "" { - return buildSessionYAML(in.Campaign, in.SessionID, in.PreviousSessionID, in.Date, in.Title, in.AudioS3Prefix, in.AudioDir) - } - templatePath := resolveSessionInitTemplatePath(in.CampaignPath, in.TemplateFile) - templateBytes, err := os.ReadFile(templatePath) - if err != nil { - return nil, fmt.Errorf("read session template %q: %w", templatePath, err) - } - rendered, err := renderSessionInitTemplate(string(templateBytes), in) - if err != nil { - return nil, fmt.Errorf("render session template %q: %w", templatePath, err) - } - return []byte(rendered), nil -} - -func resolveSessionInitTemplatePath(campaignPath, templateFile string) string { - templateFile = strings.TrimSpace(templateFile) - if filepath.IsAbs(templateFile) { - return filepath.Clean(templateFile) - } - return filepath.Clean(filepath.Join(filepath.Dir(campaignPath), templateFile)) -} - -var sessionInitTemplatePattern = regexp.MustCompile(`\{\{\s*([a-zA-Z_][a-zA-Z0-9_]*)\s*\}\}`) - -func renderSessionInitTemplate(content string, in sessionInitInput) (string, error) { - values := map[string]string{ - "session_id": strings.TrimSpace(in.SessionID), - "previous_session_id": strings.TrimSpace(in.PreviousSessionID), - "date": strings.TrimSpace(in.Date), - "title": strings.TrimSpace(in.Title), - "audio_s3_prefix": strings.TrimSpace(in.AudioS3Prefix), - "audio_dir": strings.TrimSpace(in.AudioDir), - } - used := map[string]struct{}{} - unknown := map[string]struct{}{} - missing := map[string]struct{}{} - rendered := sessionInitTemplatePattern.ReplaceAllStringFunc(content, func(match string) string { - parts := sessionInitTemplatePattern.FindStringSubmatch(match) - if len(parts) < 2 { - return match - } - name := parts[1] - value, ok := values[name] - if !ok { - unknown[name] = struct{}{} - return match - } - used[name] = struct{}{} - if value == "" { - missing[name] = struct{}{} - return match - } - return value - }) - if len(unknown) > 0 { - return "", fmt.Errorf("unsupported template variable(s): %s", sortedStringSet(unknown)) - } - if len(missing) > 0 { - return "", fmt.Errorf("missing required template variable value(s): %s", sortedStringSet(missing)) - } - unused := map[string]struct{}{} - for _, name := range []string{"previous_session_id", "date", "title", "audio_s3_prefix", "audio_dir"} { - if values[name] == "" { - continue - } - if _, ok := used[name]; !ok { - unused[name] = struct{}{} - } - } - if len(unused) > 0 { - return "", fmt.Errorf("unused template variable value(s): %s", sortedStringSet(unused)) - } - return rendered, nil -} - -func sortedStringSet(set map[string]struct{}) string { - items := make([]string, 0, len(set)) - for item := range set { - items = append(items, item) - } - sort.Strings(items) - return strings.Join(items, ", ") -} - -func renderFindings(out io.Writer, campaign, sessionID string, findings []finding) error { - if campaign != "" || sessionID != "" { - fmt.Fprintf(out, "Campaign: %s\n", campaign) - fmt.Fprintf(out, "Session: %s\n\n", sessionID) - } - errorsCount := 0 - for _, f := range findings { - if f.Severity == "ERROR" { - errorsCount++ - } - fmt.Fprintf(out, "%-5s %-10s %s\n", f.Severity, f.Category, f.Message) - } - if errorsCount > 0 { - return findingError{count: errorsCount} - } - return nil -} - -func okFinding(category, msg string) finding { return finding{"OK", category, msg} } -func infoFinding(category, msg string) finding { return finding{"INFO", category, msg} } -func warnFinding(category, msg string) finding { return finding{"WARN", category, msg} } -func errorFinding(category, msg string) finding { return finding{"ERROR", category, msg} } - -func sessionSourceSummary(cfg *config.Config) string { - source := cfg.SessionSource.Source - if source == "" { - source = "session_config" - } - if cfg.SessionSource.S3Key != "" { - return source + " " + cfg.SessionSource.S3Key - } - return source + " " + cfg.SessionPath -} - -func validateStableInputFindings(cfg *config.Config) []finding { - items := []struct { - name string - in config.ResolvedInputFile - }{ - {"speakers", cfg.StableInputs.SpeakersFile}, - {"autocorrect", cfg.StableInputs.AutocorrectFile}, - {"glossary", cfg.StableInputs.GlossaryFile}, - } - out := make([]finding, 0, len(items)) - for _, item := range items { - path, err := resolveHelperConfigRelativePath(item.in) - if err != nil { - out = append(out, errorFinding("inputs", item.name+": "+err.Error())) - continue - } - if _, err := os.Stat(path); err != nil { - out = append(out, errorFinding("inputs", fmt.Sprintf("%s missing: %v", item.name, err))) - } else { - out = append(out, okFinding("inputs", item.name+": "+path)) - } - } - return out -} - -func resolveHelperConfigRelativePath(input config.ResolvedInputFile) (string, error) { - if strings.TrimSpace(input.ConfigPath) == "" { - return "", fmt.Errorf("source config path is required") - } - path := strings.TrimSpace(input.Path) - if path == "" { - return "", fmt.Errorf("path is required") - } - if filepath.IsAbs(path) { - return filepath.Clean(path), nil - } - return filepath.Clean(filepath.Join(filepath.Dir(input.ConfigPath), path)), nil -} - -func validateLocalAudioFindings(cfg *config.Config) []finding { - if cfg.Session.Inputs.AudioS3 != nil { - return nil - } - audioDir := strings.TrimSpace(cfg.Session.Inputs.AudioDir) - if audioDir == "" && len(cfg.Session.Inputs.AudioFiles) == 0 { - return []finding{errorFinding("audio", "audio_dir, audio_files, or audio_s3 is required")} - } - base := filepath.Dir(cfg.SessionPath) - paths := []string{} - if audioDir != "" { - dir := audioDir - if !filepath.IsAbs(dir) { - dir = filepath.Join(base, dir) - } - matches, err := filepath.Glob(filepath.Join(dir, "*.flac")) - if err != nil || len(matches) == 0 { - return []finding{errorFinding("audio", "no .flac files found in "+dir)} - } - paths = append(paths, matches...) - } - for _, file := range cfg.Session.Inputs.AudioFiles { - p := file - if !filepath.IsAbs(p) { - p = filepath.Join(base, p) - } - paths = append(paths, p) - } - for _, p := range paths { - if _, err := os.Stat(p); err != nil { - return []finding{errorFinding("audio", fmt.Sprintf("audio file missing: %v", err))} - } - } - return []finding{okFinding("audio", fmt.Sprintf("%d local audio file(s)", len(paths)))} -} - -func validateRemoteAudioFinding(ctx context.Context, cfg *config.Config, store storage.ObjectStore) finding { - sessionPrefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, cfg.Session.Campaign, cfg.Session.SessionID) - audioPrefix := artifacts.S3AudioPrefix(sessionPrefix, cfg.Session.Inputs.AudioS3.Prefix) - objects, err := store.List(ctx, audioPrefix) - if err != nil { - return errorFinding("audio", err.Error()) - } - count := 0 - for _, obj := range objects { - if strings.HasSuffix(strings.ToLower(obj.Key), ".flac") { - count++ - } - } - if count == 0 { - return errorFinding("audio", "no remote .flac objects found under "+audioPrefix) - } - return okFinding("audio", fmt.Sprintf("%d remote .flac object(s)", count)) -} - -func validatePreviousArtifactFindings(ctx context.Context, cfg *config.Config, store storage.ObjectStore, requirements []artifacts.PreviousArtifactRequirement) []finding { - out := []finding{} - prefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, cfg.Session.Campaign, cfg.Session.PreviousSessionID) - _, err := artifacts.LoadCurrentState(ctx, store, prefix, artifacts.CurrentStateValidation{ - ExpectedSessionID: strings.TrimSpace(cfg.Session.PreviousSessionID), - ExpectedCampaign: strings.TrimSpace(cfg.Session.Campaign), - ValidateRunID: true, - }) - if err != nil { - out = append(out, errorFinding("previous", fmt.Sprintf("remote %v", err))) - return out - } - for _, req := range requirements { - out = append(out, okFinding("previous", fmt.Sprintf("%s required=%t", req.Name, req.Required))) - } - return out -} - -func loadLocalManifest(ctx context.Context, path string) (*manifest.Manifest, error) { - if _, err := os.Stat(path); err != nil { - if os.IsNotExist(err) { - return nil, nil - } - return nil, err - } - store := &manifest.LocalStore{} - return store.Load(ctx, path) -} - -func writeStageStatuses(out io.Writer, m *manifest.Manifest) { - if m == nil || len(m.Stages) == 0 { - fmt.Fprintln(out, "stages: no stages recorded") - return - } - fmt.Fprintln(out, "stages:") - names := make([]string, 0, len(m.Stages)) - for name := range m.Stages { - names = append(names, name) - } - sort.Strings(names) - for _, name := range names { - fmt.Fprintf(out, "- %s: %s\n", name, m.Stages[name].Status) - } -} - -func buildHelperArtifactCatalog(cfg *config.Config) (*artifacts.ArtifactCatalog, error) { - catalog := artifacts.NewArtifactCatalog() - if err := catalog.RegisterBuiltIns(); err != nil { - return nil, err - } - configured := map[string]artifacts.ConfiguredArtifactDefinition{} - if cfg.Pipeline.Scriptorium != nil { - for key, item := range cfg.Pipeline.Scriptorium.Artifacts { - configured[key] = artifacts.ConfiguredArtifactDefinition{Enabled: item.Enabled, OutputPath: item.OutputPath} - } - } - if err := catalog.RegisterConfiguredArtifacts(configured, nil); err != nil { - return nil, err - } - return catalog, nil -} - -func writeArtifactList(out io.Writer, cfg *config.Config, catalog *artifacts.ArtifactCatalog, locks *effectiveLocks, publishedRemoteState map[string]string) { - lockSet := lockSourceSet(locks.All) - fmt.Fprintln(out, "Built-in:") - for _, id := range []string{ - artifacts.ArtifactTranscriptBase, - artifacts.ArtifactTranscriptPolished, - artifacts.ArtifactTranscriptFinal, - artifacts.ArtifactTranscriptFinalTrimmed, - artifacts.ArtifactBoundsSession, - } { - writeArtifactLine(out, id, lockSet) - } - fmt.Fprintln(out, "Configured:") - for _, entry := range catalog.ListConfigured() { - writeArtifactLine(out, entry.SourceID, lockSet) - } - fmt.Fprintln(out, "Previous-session:") - for _, req := range artifacts.CollectPreviousArtifactRequirements(configuredScriptoriumArtifacts(cfg)) { - fmt.Fprintf(out, "- %s required=%t\n", artifactpolicy.PreviousSessionSourceID(req.Name), req.Required) - } - fmt.Fprintln(out, "Published:") - for _, rule := range cfg.Pipeline.Publish.Outputs { - writePublishedOutputLine(out, rule, catalog, lockSet, publishedRemoteState) - } -} - -func writeArtifactLine(out io.Writer, source string, lockSet map[string]config.PublishLockRule) { - parts := []string{source} - if _, ok := lockSet[source]; ok { - parts = append(parts, "locked") - } - fmt.Fprintf(out, "- %s\n", strings.Join(parts, " ")) -} - -func writePublishedOutputLine(out io.Writer, rule config.PublishOutputRule, catalog *artifacts.ArtifactCatalog, lockSet map[string]config.PublishLockRule, remoteState map[string]string) { - source := strings.TrimSpace(rule.Source) - parts := []string{source} - if _, ok := lockSet[source]; ok { - parts = append(parts, "locked") - } - dest, showDest, err := helperPublishedOutputDest(rule, catalog) - if err != nil { - parts = append(parts, "remote=error") - fmt.Fprintf(out, "- %s\n", strings.Join(parts, " ")) - return - } - if showDest { - parts = append(parts, "dest="+dest) - } - if state := remoteState[publishedOutputRemoteStateKey(source, dest)]; state != "" { - parts = append(parts, state) - } - fmt.Fprintf(out, "- %s\n", strings.Join(parts, " ")) -} - -func remotePublishedOutputAvailability(ctx context.Context, cfg *config.Config, store storage.ObjectStore, catalog *artifacts.ArtifactCatalog) map[string]string { - out := map[string]string{} - sessionPrefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, cfg.Session.Campaign, cfg.Session.SessionID) - for _, rule := range cfg.Pipeline.Publish.Outputs { - source := strings.TrimSpace(rule.Source) - dest, _, err := helperPublishedOutputDest(rule, catalog) - if err != nil { - out[publishedOutputRemoteStateKey(source, "")] = "remote=error" - continue - } - key := artifacts.S3PublishedOutputKey(sessionPrefix, dest) - if exists, err := store.Exists(ctx, key); err == nil && exists { - out[publishedOutputRemoteStateKey(source, dest)] = "remote=published" - } else if err != nil { - out[publishedOutputRemoteStateKey(source, dest)] = "remote=error" - } else { - out[publishedOutputRemoteStateKey(source, dest)] = "remote=missing" - } - } - return out -} - -func helperPublishedOutputDest(rule config.PublishOutputRule, catalog *artifacts.ArtifactCatalog) (string, bool, error) { - source := strings.TrimSpace(rule.Source) - normalized, err := artifactpolicy.ResolvePublishedDestination(source, rule.Dest, helperConfiguredOutputPathMap(catalog)) - if err != nil { - return "", false, err - } - entry, ok := catalog.Lookup(source) - showDest := !ok || strings.TrimSpace(entry.CanonicalRelPath) != normalized - return normalized, showDest, nil -} - -func helperConfiguredOutputPathMap(catalog *artifacts.ArtifactCatalog) map[string]string { - out := map[string]string{} - if catalog == nil { - return out - } - for _, entry := range catalog.ListConfigured() { - if strings.TrimSpace(entry.ConfiguredKey) == "" { - continue - } - out[entry.ConfiguredKey] = strings.TrimSpace(entry.CanonicalRelPath) - } - return out -} - -func publishedOutputRemoteStateKey(source, dest string) string { - return strings.TrimSpace(source) + "\x00" + strings.TrimSpace(dest) -} - -func writeLocks(out io.Writer, cfg *config.Config, locks *effectiveLocks) { - if locks == nil || len(locks.All) == 0 { - fmt.Fprintln(out, "Publish locks: none") - return - } - fmt.Fprintln(out, "Publish locks:") - published := map[string]config.PublishOutputRule{} - if cfg != nil && cfg.Pipeline != nil && cfg.Pipeline.Publish != nil { - for _, rule := range cfg.Pipeline.Publish.Outputs { - published[strings.TrimSpace(rule.Source)] = rule - } - } - staticSet := lockSourceSet(locks.Static) - for _, lock := range locks.All { - origin := "remote" - if _, ok := staticSet[lock.Source]; ok { - origin = "pipeline" - } - promo := "not-published" - if _, ok := published[lock.Source]; ok { - promo = "published" - } - reason := strings.TrimSpace(lock.Reason) - if reason == "" { - reason = "(no reason)" - } - fmt.Fprintf(out, "- %s origin=%s %s reason=%s\n", lock.Source, origin, promo, reason) - } -} - -func lockMapValues(in map[string]config.PublishLockRule) []config.PublishLockRule { - keys := make([]string, 0, len(in)) - for key := range in { - keys = append(keys, key) - } - sort.Strings(keys) - out := make([]config.PublishLockRule, 0, len(keys)) - for _, key := range keys { - item := in[key] - item.Source = key - item.Reason = strings.TrimSpace(item.Reason) - out = append(out, item) - } - return out -} diff --git a/internal/app/operator_locks.go b/internal/app/operator_locks.go new file mode 100644 index 0000000..cf9cb78 --- /dev/null +++ b/internal/app/operator_locks.go @@ -0,0 +1,222 @@ +package app + +import ( + "context" + "flag" + "fmt" + "io" + "sort" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/config" +) + +// Locks dispatches publish lock list and mutation helpers. +func Locks(ctx context.Context, args []string, out io.Writer) error { + if len(args) > 0 && !strings.HasPrefix(args[0], "-") { + switch args[0] { + case "add": + return LocksAdd(ctx, args[1:], out) + case "remove": + return LocksRemove(ctx, args[1:], out) + default: + return fmt.Errorf("locks: unknown subcommand %q", args[0]) + } + } + return LocksList(ctx, args, out) +} + +// LocksList lists effective publish locks. +func LocksList(ctx context.Context, args []string, out io.Writer) error { + fs := flag.NewFlagSet("locks", flag.ContinueOnError) + fs.SetOutput(io.Discard) + var flags commonConfigFlags + addCommonConfigFlags(fs, &flags) + if err := parseSessionAwareFlags("locks", fs, args, &flags.sessionID); err != nil { + return err + } + if strings.TrimSpace(flags.sessionID) == "" { + return fmt.Errorf("locks: session_id is required") + } + cfg, _, locks, _, err := loadHelperContext(ctx, flags, true) + if err != nil { + return fmt.Errorf("locks: %w", err) + } + writeLocks(out, cfg, locks) + return nil +} + +// LocksAdd adds or updates one remote lock. +func LocksAdd(ctx context.Context, args []string, out io.Writer) error { + var positionalSessionID string + var source string + if len(args) >= 2 && !isCLIFlagToken(args[0]) && !isCLIFlagToken(args[1]) { + positionalSessionID = strings.TrimSpace(args[0]) + source = strings.TrimSpace(args[1]) + args = append([]string(nil), args[2:]...) + } + fs := flag.NewFlagSet("locks add", flag.ContinueOnError) + fs.SetOutput(io.Discard) + var flags commonConfigFlags + var reason string + var force bool + addCommonConfigFlags(fs, &flags) + fs.StringVar(&reason, "reason", "", "lock reason") + fs.BoolVar(&force, "force", false, "update existing remote lock") + if err := fs.Parse(args); err != nil { + return fmt.Errorf("locks add: invalid flags: %w", err) + } + if source == "" { + switch fs.NArg() { + case 2: + positionalSessionID = strings.TrimSpace(fs.Arg(0)) + source = strings.TrimSpace(fs.Arg(1)) + case 1: + if strings.TrimSpace(flags.sessionID) == "" { + return fmt.Errorf("locks add: expected session_id and source id") + } + source = strings.TrimSpace(fs.Arg(0)) + default: + return fmt.Errorf("locks add: expected session_id and source id") + } + } else if fs.NArg() != 0 { + return fmt.Errorf("locks add: unexpected positional arguments") + } + if err := applyPositionalSessionID("locks add", positionalSessionID, &flags.sessionID); err != nil { + return err + } + if strings.TrimSpace(flags.sessionID) == "" { + return fmt.Errorf("locks add: session_id is required") + } + cfg, store, locks, _, err := loadHelperContext(ctx, flags, true) + if err != nil { + return fmt.Errorf("locks add: %w", err) + } + if _, err := config.ValidatePublishLockRules([]config.PublishLockRule{{Source: source}}, cfg.Pipeline.Scriptorium, "locks add"); err != nil { + return fmt.Errorf("locks add: %w", err) + } + if _, ok := lockSourceSet(locks.Static)[source]; ok { + return fmt.Errorf("locks add: source %q is locked by pipeline config and cannot be modified remotely", source) + } + remoteSet := lockSourceSet(locks.Remote) + if _, exists := remoteSet[source]; exists && !force { + return fmt.Errorf("locks add: remote lock for %q already exists; pass --force to update", source) + } + remoteSet[source] = config.PublishLockRule{Source: source, Reason: strings.TrimSpace(reason)} + remoteLocks := lockMapValues(remoteSet) + if _, err := config.ValidatePublishLockRules(remoteLocks, cfg.Pipeline.Scriptorium, "locks"); err != nil { + return fmt.Errorf("locks add: %w", err) + } + if err := uploadRemoteLockStore(ctx, store, locks.Key, &config.PublishLockStore{Locks: remoteLocks}); err != nil { + return fmt.Errorf("locks add: %w", err) + } + _, err = fmt.Fprintf(out, "narratio session locks add: locked %s\n", source) + return err +} + +// LocksRemove removes one remote lock. +func LocksRemove(ctx context.Context, args []string, out io.Writer) error { + var positionalSessionID string + var source string + if len(args) >= 2 && !isCLIFlagToken(args[0]) && !isCLIFlagToken(args[1]) { + positionalSessionID = strings.TrimSpace(args[0]) + source = strings.TrimSpace(args[1]) + args = append([]string(nil), args[2:]...) + } + fs := flag.NewFlagSet("locks remove", flag.ContinueOnError) + fs.SetOutput(io.Discard) + var flags commonConfigFlags + addCommonConfigFlags(fs, &flags) + if err := fs.Parse(args); err != nil { + return fmt.Errorf("locks remove: invalid flags: %w", err) + } + if source == "" { + switch fs.NArg() { + case 2: + positionalSessionID = strings.TrimSpace(fs.Arg(0)) + source = strings.TrimSpace(fs.Arg(1)) + case 1: + if strings.TrimSpace(flags.sessionID) == "" { + return fmt.Errorf("locks remove: expected session_id and source id") + } + source = strings.TrimSpace(fs.Arg(0)) + default: + return fmt.Errorf("locks remove: expected session_id and source id") + } + } else if fs.NArg() != 0 { + return fmt.Errorf("locks remove: unexpected positional arguments") + } + if err := applyPositionalSessionID("locks remove", positionalSessionID, &flags.sessionID); err != nil { + return err + } + if strings.TrimSpace(flags.sessionID) == "" { + return fmt.Errorf("locks remove: session_id is required") + } + cfg, store, locks, _, err := loadHelperContext(ctx, flags, true) + if err != nil { + return fmt.Errorf("locks remove: %w", err) + } + if _, err := config.ValidatePublishLockRules([]config.PublishLockRule{{Source: source}}, cfg.Pipeline.Scriptorium, "locks remove"); err != nil { + return fmt.Errorf("locks remove: %w", err) + } + remoteSet := lockSourceSet(locks.Remote) + if _, ok := remoteSet[source]; !ok { + if _, static := lockSourceSet(locks.Static)[source]; static { + return fmt.Errorf("locks remove: source %q is locked by pipeline config and cannot be unlocked remotely", source) + } + return fmt.Errorf("locks remove: remote lock for %q does not exist", source) + } + delete(remoteSet, source) + remoteLocks := lockMapValues(remoteSet) + if err := uploadRemoteLockStore(ctx, store, locks.Key, &config.PublishLockStore{Locks: remoteLocks}); err != nil { + return fmt.Errorf("locks remove: %w", err) + } + _, err = fmt.Fprintf(out, "narratio session locks remove: unlocked %s\n", source) + return err +} + +func writeLocks(out io.Writer, cfg *config.Config, locks *effectiveLocks) { + if locks == nil || len(locks.All) == 0 { + fmt.Fprintln(out, "Publish locks: none") + return + } + fmt.Fprintln(out, "Publish locks:") + published := map[string]config.PublishOutputRule{} + if cfg != nil && cfg.Pipeline != nil && cfg.Pipeline.Publish != nil { + for _, rule := range cfg.Pipeline.Publish.Outputs { + published[strings.TrimSpace(rule.Source)] = rule + } + } + staticSet := lockSourceSet(locks.Static) + for _, lock := range locks.All { + origin := "remote" + if _, ok := staticSet[lock.Source]; ok { + origin = "pipeline" + } + promo := "not-published" + if _, ok := published[lock.Source]; ok { + promo = "published" + } + reason := strings.TrimSpace(lock.Reason) + if reason == "" { + reason = "(no reason)" + } + fmt.Fprintf(out, "- %s origin=%s %s reason=%s\n", lock.Source, origin, promo, reason) + } +} + +func lockMapValues(in map[string]config.PublishLockRule) []config.PublishLockRule { + keys := make([]string, 0, len(in)) + for key := range in { + keys = append(keys, key) + } + sort.Strings(keys) + out := make([]config.PublishLockRule, 0, len(keys)) + for _, key := range keys { + item := in[key] + item.Source = key + item.Reason = strings.TrimSpace(item.Reason) + out = append(out, item) + } + return out +} diff --git a/internal/app/operator_session_init.go b/internal/app/operator_session_init.go new file mode 100644 index 0000000..a69b9cf --- /dev/null +++ b/internal/app/operator_session_init.go @@ -0,0 +1,268 @@ +package app + +import ( + "context" + "flag" + "fmt" + "io" + "os" + "path/filepath" + "regexp" + "sort" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/config" + "gopkg.in/yaml.v3" +) + +// SessionInit creates a local or remote session.yml skeleton. +func SessionInit(ctx context.Context, args []string, out io.Writer) error { + fs := flag.NewFlagSet("session init", flag.ContinueOnError) + fs.SetOutput(io.Discard) + var pipelinePath, campaignPath, campaignFilePath, sessionID, previousSessionID, date, title, output, audioS3Prefix, audioDir string + var remote, force bool + fs.StringVar(&pipelinePath, "config", "", "path to pipeline.yml (optional; defaults searched)") + fs.StringVar(&campaignPath, "campaign", "", "campaign ID") + fs.StringVar(&campaignFilePath, "campaign-file", "", "path to campaign.yml") + fs.StringVar(&sessionID, "session-id", "", "session identifier") + fs.StringVar(&previousSessionID, "previous-session-id", "", "previous session identifier") + fs.StringVar(&date, "date", "", "session date") + fs.StringVar(&title, "title", "", "session title") + fs.StringVar(&output, "output", "", "local output session.yml path") + fs.StringVar(&audioS3Prefix, "audio-s3-prefix", "", "session audio S3 prefix") + fs.StringVar(&audioDir, "audio-dir", "", "local audio directory") + fs.BoolVar(&remote, "remote", false, "write session.yml to S3 session prefix") + fs.BoolVar(&force, "force", false, "overwrite existing target") + if err := parseSessionAwareFlags("session init", fs, args, &sessionID); err != nil { + return err + } + if strings.TrimSpace(sessionID) == "" { + return fmt.Errorf("session init: session_id is required") + } + if (strings.TrimSpace(output) == "") == !remote { + return fmt.Errorf("session init: specify exactly one target: --output or --remote") + } + if strings.TrimSpace(audioDir) != "" && strings.TrimSpace(audioS3Prefix) != "" { + return fmt.Errorf("session init: --audio-dir and --audio-s3-prefix are mutually exclusive") + } + + base, err := loadPipelineCampaignConfig(pipelinePath, campaignPath, campaignFilePath) + if err != nil { + return fmt.Errorf("session init: %w", err) + } + + input := sessionInitInput{ + Campaign: config.CampaignID(base.Campaign), + CampaignPath: base.CampaignPath, + TemplateFile: base.Campaign.SessionTemplateFile, + SessionID: sessionID, + PreviousSessionID: previousSessionID, + Date: date, + Title: title, + AudioS3Prefix: audioS3Prefix, + AudioDir: audioDir, + } + data, err := buildSessionInitYAML(input) + if err != nil { + return fmt.Errorf("session init: %w", err) + } + label := strings.TrimSpace(output) + if label == "" { + label = "remote session.yml" + } + sessionCfg, err := config.LoadSessionBytesWithOptions(label, data, config.SessionLoadOptions{ + SessionID: sessionID, + PreviousSessionID: previousSessionID, + }) + if err != nil { + return fmt.Errorf("session init: %w", err) + } + cfg, err := config.Resolve(base.PipelinePath, base.Pipeline, base.CampaignPath, base.Campaign, label, sessionCfg, config.SessionSource{Source: "session_config", LocalPath: label}) + if err != nil { + return fmt.Errorf("session init: %w", err) + } + if err := config.Validate(cfg); err != nil { + return fmt.Errorf("session init: %w", err) + } + + if !remote { + if err := writeLocalFile(output, data, force); err != nil { + return fmt.Errorf("session init: %w", err) + } + _, err := fmt.Fprintf(out, "narratio session init: wrote %s\n", filepath.Clean(output)) + return err + } + + store, err := newCommandObjectStore(ctx, cfg, nil) + if err != nil { + return fmt.Errorf("session init: %w", err) + } + sessionPrefix := artifacts.S3SessionPrefix(base.Pipeline.Storage.S3.RootPrefix, config.CampaignID(base.Campaign), sessionID) + key := artifacts.S3SessionConfigKey(sessionPrefix) + exists, err := store.Exists(ctx, key) + if err != nil { + return fmt.Errorf("session init: check remote session %q: %w", key, err) + } + if exists && !force { + return fmt.Errorf("session init: remote session %q already exists; pass --force to overwrite", key) + } + tmp, err := os.CreateTemp("", "narratio-session-init-*.yml") + if err != nil { + return fmt.Errorf("session init: create temp file: %w", err) + } + tmpPath := tmp.Name() + defer func() { _ = os.Remove(tmpPath) }() + if _, err := tmp.Write(data); err != nil { + _ = tmp.Close() + return fmt.Errorf("session init: write temp file: %w", err) + } + if err := tmp.Close(); err != nil { + return fmt.Errorf("session init: close temp file: %w", err) + } + if _, err := store.Upload(ctx, tmpPath, key, storage.UploadOptions{ContentType: "application/x-yaml; charset=utf-8"}); err != nil { + return fmt.Errorf("session init: upload remote session %q: %w", key, err) + } + _, err = fmt.Fprintf(out, "narratio session init: wrote s3://%s/%s\n", s3BucketName(base.Pipeline), key) + return err +} + +func buildSessionYAML(campaign, sessionID, previousSessionID, date, title, audioS3Prefix, audioDir string) ([]byte, error) { + if strings.TrimSpace(date) == "" && regexp.MustCompile(`^\d{4}-\d{2}-\d{2}$`).MatchString(strings.TrimSpace(sessionID)) { + date = strings.TrimSpace(sessionID) + } + type audioS3 struct { + Prefix string `yaml:"prefix"` + } + type inputs struct { + AudioDir string `yaml:"audio_dir,omitempty"` + AudioS3 *audioS3 `yaml:"audio_s3,omitempty"` + } + type sessionYAML struct { + Campaign string `yaml:"campaign"` + SessionID string `yaml:"session_id"` + PreviousSessionID string `yaml:"previous_session_id,omitempty"` + Date string `yaml:"date,omitempty"` + Title string `yaml:"title,omitempty"` + Inputs inputs `yaml:"inputs"` + } + in := inputs{AudioDir: strings.TrimSpace(audioDir)} + if in.AudioDir == "" { + prefix := strings.TrimSpace(audioS3Prefix) + if prefix == "" { + prefix = "audio/" + } + in.AudioS3 = &audioS3{Prefix: prefix} + } + data, err := yaml.Marshal(sessionYAML{ + Campaign: strings.TrimSpace(campaign), + SessionID: strings.TrimSpace(sessionID), + PreviousSessionID: strings.TrimSpace(previousSessionID), + Date: strings.TrimSpace(date), + Title: strings.TrimSpace(title), + Inputs: in, + }) + if err != nil { + return nil, err + } + return data, nil +} + +type sessionInitInput struct { + Campaign string + CampaignPath string + TemplateFile string + SessionID string + PreviousSessionID string + Date string + Title string + AudioS3Prefix string + AudioDir string +} + +func buildSessionInitYAML(in sessionInitInput) ([]byte, error) { + if strings.TrimSpace(in.TemplateFile) == "" { + return buildSessionYAML(in.Campaign, in.SessionID, in.PreviousSessionID, in.Date, in.Title, in.AudioS3Prefix, in.AudioDir) + } + templatePath := resolveSessionInitTemplatePath(in.CampaignPath, in.TemplateFile) + templateBytes, err := os.ReadFile(templatePath) + if err != nil { + return nil, fmt.Errorf("read session template %q: %w", templatePath, err) + } + rendered, err := renderSessionInitTemplate(string(templateBytes), in) + if err != nil { + return nil, fmt.Errorf("render session template %q: %w", templatePath, err) + } + return []byte(rendered), nil +} + +func resolveSessionInitTemplatePath(campaignPath, templateFile string) string { + templateFile = strings.TrimSpace(templateFile) + if filepath.IsAbs(templateFile) { + return filepath.Clean(templateFile) + } + return filepath.Clean(filepath.Join(filepath.Dir(campaignPath), templateFile)) +} + +var sessionInitTemplatePattern = regexp.MustCompile(`\{\{\s*([a-zA-Z_][a-zA-Z0-9_]*)\s*\}\}`) + +func renderSessionInitTemplate(content string, in sessionInitInput) (string, error) { + values := map[string]string{ + "session_id": strings.TrimSpace(in.SessionID), + "previous_session_id": strings.TrimSpace(in.PreviousSessionID), + "date": strings.TrimSpace(in.Date), + "title": strings.TrimSpace(in.Title), + "audio_s3_prefix": strings.TrimSpace(in.AudioS3Prefix), + "audio_dir": strings.TrimSpace(in.AudioDir), + } + used := map[string]struct{}{} + unknown := map[string]struct{}{} + missing := map[string]struct{}{} + rendered := sessionInitTemplatePattern.ReplaceAllStringFunc(content, func(match string) string { + parts := sessionInitTemplatePattern.FindStringSubmatch(match) + if len(parts) < 2 { + return match + } + name := parts[1] + value, ok := values[name] + if !ok { + unknown[name] = struct{}{} + return match + } + used[name] = struct{}{} + if value == "" { + missing[name] = struct{}{} + return match + } + return value + }) + if len(unknown) > 0 { + return "", fmt.Errorf("unsupported template variable(s): %s", sortedStringSet(unknown)) + } + if len(missing) > 0 { + return "", fmt.Errorf("missing required template variable value(s): %s", sortedStringSet(missing)) + } + unused := map[string]struct{}{} + for _, name := range []string{"previous_session_id", "date", "title", "audio_s3_prefix", "audio_dir"} { + if values[name] == "" { + continue + } + if _, ok := used[name]; !ok { + unused[name] = struct{}{} + } + } + if len(unused) > 0 { + return "", fmt.Errorf("unused template variable value(s): %s", sortedStringSet(unused)) + } + return rendered, nil +} + +func sortedStringSet(set map[string]struct{}) string { + items := make([]string, 0, len(set)) + for item := range set { + items = append(items, item) + } + sort.Strings(items) + return strings.Join(items, ", ") +} diff --git a/internal/app/operator_session_validate.go b/internal/app/operator_session_validate.go new file mode 100644 index 0000000..cb9b7ec --- /dev/null +++ b/internal/app/operator_session_validate.go @@ -0,0 +1,81 @@ +package app + +import ( + "context" + "flag" + "fmt" + "io" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/config" +) + +// SessionValidate performs a read-only session preflight. +func SessionValidate(ctx context.Context, args []string, out io.Writer) error { + fs := flag.NewFlagSet("session validate", flag.ContinueOnError) + fs.SetOutput(io.Discard) + var flags commonConfigFlags + addCommonConfigFlags(fs, &flags) + if err := parseSessionAwareFlags("session validate", fs, args, &flags.sessionID); err != nil { + return err + } + if strings.TrimSpace(flags.sessionID) == "" { + return fmt.Errorf("session validate: session_id is required") + } + + findings := []finding{} + cfg, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) + if err != nil { + findings = append(findings, errorFinding("config", err.Error())) + return renderFindings(out, "", "", findings) + } + if err := config.Validate(cfg); err != nil { + findings = append(findings, errorFinding("config", err.Error())) + } else { + findings = append(findings, okFinding("config", "resolved pipeline, campaign, and session config")) + } + findings = append(findings, okFinding("session", fmt.Sprintf("session source: %s", sessionSourceSummary(cfg)))) + + paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID) + findings = append(findings, validateStableInputFindings(cfg)...) + findings = append(findings, validateLocalAudioFindings(cfg)...) + + store, storeErr := objectStoreIfConfigured(ctx, cfg) + if storeErr != nil { + findings = append(findings, errorFinding("storage", storeErr.Error())) + } + if cfg.Session.Inputs.AudioS3 != nil { + if storeErr != nil { + findings = append(findings, errorFinding("audio", "remote audio cannot be checked because storage is unavailable")) + } else { + findings = append(findings, validateRemoteAudioFinding(ctx, cfg, store)) + } + } + + requirements := artifacts.CollectPreviousArtifactRequirements(configuredScriptoriumArtifacts(cfg)) + if len(requirements) == 0 { + findings = append(findings, okFinding("previous", "no previous-session artifacts required")) + } else if strings.TrimSpace(cfg.Session.PreviousSessionID) == "" { + findings = append(findings, errorFinding("previous", "previous_session_id is required by configured previous-session artifacts")) + } else if storeErr != nil { + findings = append(findings, errorFinding("previous", "previous-session artifacts cannot be checked because storage is unavailable")) + } else { + findings = append(findings, validatePreviousArtifactFindings(ctx, cfg, store, requirements)...) + } + + locks, lockErr := loadEffectiveLocks(ctx, cfg, store) + if lockErr != nil { + findings = append(findings, errorFinding("locks", lockErr.Error())) + } else if len(locks.All) == 0 { + findings = append(findings, okFinding("locks", "no effective publish locks")) + } else { + for _, lock := range locks.All { + findings = append(findings, warnFinding("locks", fmt.Sprintf("%s locked: %s", lock.Source, strings.TrimSpace(lock.Reason)))) + } + } + if paths.ManifestPath != "" { + findings = append(findings, infoFinding("workspace", "manifest path: "+paths.ManifestPath)) + } + return renderFindings(out, cfg.Session.Campaign, cfg.Session.SessionID, findings) +} diff --git a/internal/app/operator_status.go b/internal/app/operator_status.go new file mode 100644 index 0000000..2c7c611 --- /dev/null +++ b/internal/app/operator_status.go @@ -0,0 +1,88 @@ +package app + +import ( + "context" + "flag" + "fmt" + "io" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/config" +) + +// Status reports effective local/remote session state. +func Status(ctx context.Context, args []string, out io.Writer) error { + fs := flag.NewFlagSet("status", flag.ContinueOnError) + fs.SetOutput(io.Discard) + var flags commonConfigFlags + addCommonConfigFlags(fs, &flags) + if err := parseSessionAwareFlags("status", fs, args, &flags.sessionID); err != nil { + return err + } + if strings.TrimSpace(flags.sessionID) == "" { + return fmt.Errorf("status: session_id is required") + } + cfg, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) + if err != nil { + return fmt.Errorf("status: %w", err) + } + if err := config.Validate(cfg); err != nil { + return fmt.Errorf("status: %w", err) + } + paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID) + fmt.Fprintf(out, "Session: %s\n", cfg.Session.SessionID) + fmt.Fprintf(out, "Campaign: %s\n", cfg.Session.Campaign) + fmt.Fprintf(out, "Workspace: %s\n", paths.Root) + fmt.Fprintf(out, "Session config: %s\n", sessionSourceSummary(cfg)) + + if m, err := loadLocalManifest(ctx, paths.ManifestPath); err != nil { + fmt.Fprintf(out, "Local manifest: error: %v\n", err) + } else if m == nil { + fmt.Fprintln(out, "Local manifest: missing") + } else { + fmt.Fprintf(out, "Local manifest: %s\n", paths.ManifestPath) + writeStageStatuses(out, m) + } + + store, storeErr := objectStoreIfConfigured(ctx, cfg) + if storeErr != nil { + fmt.Fprintf(out, "Remote publish: unavailable: %v\n", storeErr) + } else if store != nil { + current, err := discoverRemoteCurrentStateFn(ctx, cfg, store) + if err != nil { + fmt.Fprintf(out, "Remote publish: missing or unavailable: %v\n", err) + } else { + fmt.Fprintf(out, "Remote publish: current run %s\n", current.RunID) + fmt.Fprintf(out, "Remote manifest: %s\n", current.CurrentManifestKey) + } + } + + locks, err := loadEffectiveLocks(ctx, cfg, store) + if catalog, catalogErr := buildHelperArtifactCatalog(cfg); catalogErr != nil { + fmt.Fprintf(out, "Remote outputs: error: %v\n", catalogErr) + } else if storeErr == nil { + catalogLocks := locks + if err != nil { + catalogLocks = &effectiveLocks{ + Static: staticPublishLocks(cfg), + All: staticPublishLocks(cfg), + } + } + publishedRemoteState := map[string]string{} + if store != nil { + publishedRemoteState = remotePublishedOutputAvailability(ctx, cfg, store, catalog) + } + fmt.Fprintln(out, "Remote outputs:") + writeArtifactList(out, cfg, catalog, catalogLocks, publishedRemoteState) + } + if err != nil { + fmt.Fprintf(out, "Publish locks: error: %v\n", err) + } else { + writeLocks(out, cfg, locks) + } + fmt.Fprintln(out, "Next actions:") + fmt.Fprintf(out, "- narratio session validate %s\n", cfg.Session.SessionID) + fmt.Fprintf(out, "- narratio session restore %s --dry-run\n", cfg.Session.SessionID) + return nil +}