Implemented operations helper commands for validation, locking, and status
This commit is contained in:
870
internal/app/operator_helpers.go
Normal file
870
internal/app/operator_helpers.go
Normal file
@@ -0,0 +1,870 @@
|
||||
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"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
||||
"gopkg.in/yaml.v3"
|
||||
)
|
||||
|
||||
type commonConfigFlags struct {
|
||||
pipelinePath string
|
||||
campaignPath string
|
||||
sessionPath string
|
||||
sessionID string
|
||||
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", "", "path to campaign.yml (optional; defaults searched)")
|
||||
fs.StringVar(&flags.sessionPath, "session", "", "path to session.yml")
|
||||
fs.StringVar(&flags.sessionID, "session-id", "", "session identifier for session.yml templates")
|
||||
fs.StringVar(&flags.previousSessionID, "previous-session-id", "", "previous session identifier for session.yml templates")
|
||||
}
|
||||
|
||||
func (f commonConfigFlags) sessionOptions() config.SessionLoadOptions {
|
||||
return config.SessionLoadOptions{
|
||||
SessionID: f.sessionID,
|
||||
PreviousSessionID: f.previousSessionID,
|
||||
}
|
||||
}
|
||||
|
||||
// Session dispatches session helper subcommands.
|
||||
func Session(ctx context.Context, args []string, out io.Writer) error {
|
||||
if len(args) == 0 {
|
||||
return fmt.Errorf("session: expected subcommand: validate|init")
|
||||
}
|
||||
switch args[0] {
|
||||
case "validate":
|
||||
return SessionValidate(ctx, args[1:], out)
|
||||
case "init":
|
||||
return SessionInit(ctx, args[1:], out)
|
||||
default:
|
||||
return fmt.Errorf("session: unknown subcommand %q", args[0])
|
||||
}
|
||||
}
|
||||
|
||||
// Artifacts dispatches artifact helper subcommands.
|
||||
func Artifacts(ctx context.Context, args []string, out io.Writer) error {
|
||||
if len(args) == 0 {
|
||||
return fmt.Errorf("artifacts: expected subcommand: list")
|
||||
}
|
||||
switch args[0] {
|
||||
case "list":
|
||||
return ArtifactsList(ctx, args[1:], out)
|
||||
default:
|
||||
return fmt.Errorf("artifacts: unknown subcommand %q", args[0])
|
||||
}
|
||||
}
|
||||
|
||||
// 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 := fs.Parse(args); err != nil {
|
||||
return fmt.Errorf("session validate: invalid flags: %w", err)
|
||||
}
|
||||
if fs.NArg() != 0 {
|
||||
return fmt.Errorf("session validate: unexpected positional arguments")
|
||||
}
|
||||
|
||||
findings := []finding{}
|
||||
cfg, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, 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 archive 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 either a requested manifest or 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 manifestPath string
|
||||
var flags commonConfigFlags
|
||||
fs.StringVar(&manifestPath, "manifest", "", "path to manifest.json")
|
||||
addCommonConfigFlags(fs, &flags)
|
||||
if err := fs.Parse(args); err != nil {
|
||||
return fmt.Errorf("status: invalid flags: %w", err)
|
||||
}
|
||||
if fs.NArg() != 0 {
|
||||
return fmt.Errorf("status: unexpected positional arguments")
|
||||
}
|
||||
if strings.TrimSpace(manifestPath) != "" {
|
||||
return statusManifest(ctx, manifestPath, out)
|
||||
}
|
||||
if flags.pipelinePath == "" && flags.campaignPath == "" && flags.sessionPath == "" && flags.sessionID == "" && flags.previousSessionID == "" {
|
||||
return fmt.Errorf("status: --manifest is required")
|
||||
}
|
||||
cfg, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, 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 archive: unavailable: %v\n", storeErr)
|
||||
} else if store != nil {
|
||||
current, err := discoverRemoteCurrentStateFn(ctx, cfg, store)
|
||||
if err != nil {
|
||||
fmt.Fprintf(out, "Remote archive: missing or unavailable: %v\n", err)
|
||||
} else {
|
||||
fmt.Fprintf(out, "Remote archive: current run %s\n", current.RunID)
|
||||
fmt.Fprintf(out, "Remote manifest: %s\n", current.CurrentManifestKey)
|
||||
}
|
||||
}
|
||||
|
||||
locks, err := loadEffectiveLocks(ctx, cfg, store)
|
||||
if err != nil {
|
||||
fmt.Fprintf(out, "Archive locks: error: %v\n", err)
|
||||
} else {
|
||||
writeLocks(out, cfg, locks)
|
||||
}
|
||||
fmt.Fprintln(out, "Next actions:")
|
||||
fmt.Fprintf(out, "- narratio session validate --session-id %s\n", cfg.Session.SessionID)
|
||||
fmt.Fprintf(out, "- narratio restore --session-id %s --dry-run\n", cfg.Session.SessionID)
|
||||
return nil
|
||||
}
|
||||
|
||||
func statusManifest(ctx context.Context, manifestPath string, out io.Writer) error {
|
||||
store := &manifest.LocalStore{}
|
||||
m, err := store.Load(ctx, manifestPath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("status: %w", err)
|
||||
}
|
||||
if _, err := fmt.Fprintf(out, "session_id: %s\n", m.SessionID); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := fmt.Fprintf(out, "updated_at: %s\n", m.UpdatedAt.UTC().Format("2006-01-02T15:04:05Z07:00")); err != nil {
|
||||
return err
|
||||
}
|
||||
writeStageStatuses(out, m)
|
||||
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, sessionID, previousSessionID, date, title, output, audioS3Prefix, audioDir string
|
||||
var remote, force bool
|
||||
fs.StringVar(&pipelinePath, "config", "", "path to pipeline.yml")
|
||||
fs.StringVar(&campaignPath, "campaign", "", "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 := fs.Parse(args); err != nil {
|
||||
return fmt.Errorf("session init: invalid flags: %w", err)
|
||||
}
|
||||
if fs.NArg() != 0 {
|
||||
return fmt.Errorf("session init: unexpected positional arguments")
|
||||
}
|
||||
if strings.TrimSpace(pipelinePath) == "" || strings.TrimSpace(campaignPath) == "" || strings.TrimSpace(sessionID) == "" {
|
||||
return fmt.Errorf("session init: --config, --campaign, and --session-id are required")
|
||||
}
|
||||
if (strings.TrimSpace(output) == "") == !remote {
|
||||
return fmt.Errorf("session init: specify exactly one target: --output <path> or --remote")
|
||||
}
|
||||
if strings.TrimSpace(audioDir) != "" && strings.TrimSpace(audioS3Prefix) != "" {
|
||||
return fmt.Errorf("session init: --audio-dir and --audio-s3-prefix are mutually exclusive")
|
||||
}
|
||||
|
||||
resolvedPipeline, err := resolvePipelineConfigPath(pipelinePath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("session init: %w", err)
|
||||
}
|
||||
resolvedCampaign, err := resolveCampaignConfigPath(campaignPath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("session init: %w", err)
|
||||
}
|
||||
pipelineCfg, err := config.LoadPipeline(resolvedPipeline)
|
||||
if err != nil {
|
||||
return fmt.Errorf("session init: %w", err)
|
||||
}
|
||||
campaignCfg, err := config.LoadCampaign(resolvedCampaign)
|
||||
if err != nil {
|
||||
return fmt.Errorf("session init: %w", err)
|
||||
}
|
||||
|
||||
data, err := buildSessionYAML(campaignCfg.Campaign, sessionID, previousSessionID, date, title, audioS3Prefix, audioDir)
|
||||
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(resolvedPipeline, pipelineCfg, resolvedCampaign, campaignCfg, 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 := newObjectStoreFromConfigFn(ctx, cfg)
|
||||
if err != nil {
|
||||
return fmt.Errorf("session init: %w", err)
|
||||
}
|
||||
sessionPrefix := artifacts.S3SessionPrefix(pipelineCfg.Storage.S3.RootPrefix, campaignCfg.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(pipelineCfg), 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 archive availability")
|
||||
if err := fs.Parse(args); err != nil {
|
||||
return fmt.Errorf("artifacts list: invalid flags: %w", err)
|
||||
}
|
||||
if fs.NArg() != 0 {
|
||||
return fmt.Errorf("artifacts list: unexpected positional arguments")
|
||||
}
|
||||
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)
|
||||
}
|
||||
remoteState := map[string]string{}
|
||||
if remote && store != nil {
|
||||
remoteState = remoteArtifactAvailability(ctx, cfg, store, catalog)
|
||||
}
|
||||
writeArtifactList(out, cfg, catalog, locks, remoteState)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Locks lists effective archive locks.
|
||||
func Locks(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 := fs.Parse(args); err != nil {
|
||||
return fmt.Errorf("locks: invalid flags: %w", err)
|
||||
}
|
||||
if fs.NArg() != 0 {
|
||||
return fmt.Errorf("locks: unexpected positional arguments")
|
||||
}
|
||||
cfg, _, locks, _, err := loadHelperContext(ctx, flags, true)
|
||||
if err != nil {
|
||||
return fmt.Errorf("locks: %w", err)
|
||||
}
|
||||
writeLocks(out, cfg, locks)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Lock adds or updates one remote lock.
|
||||
func Lock(ctx context.Context, args []string, out io.Writer) error {
|
||||
fs := flag.NewFlagSet("lock", 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("lock: invalid flags: %w", err)
|
||||
}
|
||||
if fs.NArg() != 1 {
|
||||
return fmt.Errorf("lock: expected exactly one source id")
|
||||
}
|
||||
source := strings.TrimSpace(fs.Arg(0))
|
||||
cfg, store, locks, _, err := loadHelperContext(ctx, flags, true)
|
||||
if err != nil {
|
||||
return fmt.Errorf("lock: %w", err)
|
||||
}
|
||||
if _, err := config.ValidateArchiveLockRules([]config.ArchiveLockRule{{Source: source}}, cfg.Pipeline.Scriptorium, "lock"); err != nil {
|
||||
return fmt.Errorf("lock: %w", err)
|
||||
}
|
||||
if _, ok := lockSourceSet(locks.Static)[source]; ok {
|
||||
return fmt.Errorf("lock: 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("lock: remote lock for %q already exists; pass --force to update", source)
|
||||
}
|
||||
remoteSet[source] = config.ArchiveLockRule{Source: source, Reason: strings.TrimSpace(reason)}
|
||||
remoteLocks := lockMapValues(remoteSet)
|
||||
if _, err := config.ValidateArchiveLockRules(remoteLocks, cfg.Pipeline.Scriptorium, "locks"); err != nil {
|
||||
return fmt.Errorf("lock: %w", err)
|
||||
}
|
||||
if err := uploadRemoteLockStore(ctx, store, locks.Key, &config.ArchiveLockStore{Locks: remoteLocks}); err != nil {
|
||||
return fmt.Errorf("lock: %w", err)
|
||||
}
|
||||
_, err = fmt.Fprintf(out, "narratio lock: locked %s\n", source)
|
||||
return err
|
||||
}
|
||||
|
||||
// Unlock removes one remote lock.
|
||||
func Unlock(ctx context.Context, args []string, out io.Writer) error {
|
||||
fs := flag.NewFlagSet("unlock", flag.ContinueOnError)
|
||||
fs.SetOutput(io.Discard)
|
||||
var flags commonConfigFlags
|
||||
addCommonConfigFlags(fs, &flags)
|
||||
if err := fs.Parse(args); err != nil {
|
||||
return fmt.Errorf("unlock: invalid flags: %w", err)
|
||||
}
|
||||
if fs.NArg() != 1 {
|
||||
return fmt.Errorf("unlock: expected exactly one source id")
|
||||
}
|
||||
source := strings.TrimSpace(fs.Arg(0))
|
||||
cfg, store, locks, _, err := loadHelperContext(ctx, flags, true)
|
||||
if err != nil {
|
||||
return fmt.Errorf("unlock: %w", err)
|
||||
}
|
||||
if _, err := config.ValidateArchiveLockRules([]config.ArchiveLockRule{{Source: source}}, cfg.Pipeline.Scriptorium, "unlock"); err != nil {
|
||||
return fmt.Errorf("unlock: %w", err)
|
||||
}
|
||||
remoteSet := lockSourceSet(locks.Remote)
|
||||
if _, ok := remoteSet[source]; !ok {
|
||||
if _, static := lockSourceSet(locks.Static)[source]; static {
|
||||
return fmt.Errorf("unlock: source %q is locked by pipeline config and cannot be unlocked remotely", source)
|
||||
}
|
||||
return fmt.Errorf("unlock: remote lock for %q does not exist", source)
|
||||
}
|
||||
delete(remoteSet, source)
|
||||
remoteLocks := lockMapValues(remoteSet)
|
||||
if err := uploadRemoteLockStore(ctx, store, locks.Key, &config.ArchiveLockStore{Locks: remoteLocks}); err != nil {
|
||||
return fmt.Errorf("unlock: %w", err)
|
||||
}
|
||||
_, err = fmt.Fprintf(out, "narratio unlock: 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.sessionPath, flags.sessionOptions())
|
||||
if err != nil {
|
||||
return nil, nil, nil, nil, err
|
||||
}
|
||||
if err := config.Validate(cfg); err != nil {
|
||||
return nil, nil, nil, nil, err
|
||||
}
|
||||
var store storage.ObjectStore
|
||||
if needStore {
|
||||
store, err = newObjectStoreFromConfigFn(ctx, cfg)
|
||||
if err != nil {
|
||||
return nil, nil, nil, nil, err
|
||||
}
|
||||
} else {
|
||||
store, _ = objectStoreIfConfigured(ctx, cfg)
|
||||
}
|
||||
locks, err := loadEffectiveLocks(ctx, cfg, store)
|
||||
if err != nil {
|
||||
return nil, nil, nil, nil, err
|
||||
}
|
||||
paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID)
|
||||
m, err := loadLocalManifest(ctx, paths.ManifestPath)
|
||||
if err != nil {
|
||||
return nil, nil, nil, nil, err
|
||||
}
|
||||
return cfg, store, locks, m, nil
|
||||
}
|
||||
|
||||
func objectStoreIfConfigured(ctx context.Context, cfg *config.Config) (storage.ObjectStore, error) {
|
||||
if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Storage.S3 == nil || strings.TrimSpace(cfg.Pipeline.Storage.S3.Bucket) == "" {
|
||||
return nil, nil
|
||||
}
|
||||
store, err := newObjectStoreFromConfigFn(ctx, cfg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
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
|
||||
}
|
||||
|
||||
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)
|
||||
manifestKey, runIDKey := artifacts.ResolveArchiveCurrentStateKeys(prefix)
|
||||
for _, key := range []string{runIDKey, manifestKey} {
|
||||
exists, err := store.Exists(ctx, key)
|
||||
if err != nil {
|
||||
out = append(out, errorFinding("previous", fmt.Sprintf("check %s: %v", key, err)))
|
||||
return out
|
||||
}
|
||||
if !exists {
|
||||
out = append(out, errorFinding("previous", "missing "+key))
|
||||
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, remoteState map[string]string) {
|
||||
lockSet := lockSourceSet(locks.All)
|
||||
fmt.Fprintln(out, "Built-in:")
|
||||
for _, id := range []string{
|
||||
artifacts.ArtifactTranscriptMerged,
|
||||
artifacts.ArtifactTranscriptPolished,
|
||||
artifacts.ArtifactTranscriptFull,
|
||||
artifacts.ArtifactTranscriptTrimmed,
|
||||
artifacts.ArtifactBoundsSession,
|
||||
} {
|
||||
writeArtifactLine(out, id, lockSet, remoteState)
|
||||
}
|
||||
fmt.Fprintln(out, "Configured:")
|
||||
for _, entry := range catalog.ListConfigured() {
|
||||
writeArtifactLine(out, entry.SourceID, lockSet, remoteState)
|
||||
}
|
||||
fmt.Fprintln(out, "Previous-session:")
|
||||
for _, req := range artifacts.CollectPreviousArtifactRequirements(configuredScriptoriumArtifacts(cfg)) {
|
||||
fmt.Fprintf(out, "- narratio.previous_session.artifact.%s required=%t\n", req.Name, req.Required)
|
||||
}
|
||||
fmt.Fprintln(out, "Promoted:")
|
||||
for _, rule := range cfg.Pipeline.Archive.PromoteArtifacts {
|
||||
writeArtifactLine(out, rule.Source, lockSet, remoteState)
|
||||
}
|
||||
}
|
||||
|
||||
func writeArtifactLine(out io.Writer, source string, lockSet map[string]config.ArchiveLockRule, remoteState map[string]string) {
|
||||
parts := []string{source}
|
||||
if _, ok := lockSet[source]; ok {
|
||||
parts = append(parts, "locked")
|
||||
}
|
||||
if state := remoteState[source]; state != "" {
|
||||
parts = append(parts, state)
|
||||
}
|
||||
fmt.Fprintf(out, "- %s\n", strings.Join(parts, " "))
|
||||
}
|
||||
|
||||
func remoteArtifactAvailability(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 _, source := range allCatalogSources(catalog) {
|
||||
entry, ok := catalog.Lookup(source)
|
||||
if !ok || strings.TrimSpace(entry.CanonicalRelPath) == "" {
|
||||
continue
|
||||
}
|
||||
key := artifacts.S3PromotedArtifactKey(sessionPrefix, entry.CanonicalRelPath)
|
||||
if exists, err := store.Exists(ctx, key); err == nil && exists {
|
||||
out[source] = "remote=promoted"
|
||||
} else if err != nil {
|
||||
out[source] = "remote=error"
|
||||
} else {
|
||||
out[source] = "remote=missing"
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func allCatalogSources(catalog *artifacts.ArtifactCatalog) []string {
|
||||
out := []string{
|
||||
artifacts.ArtifactTranscriptMerged,
|
||||
artifacts.ArtifactTranscriptPolished,
|
||||
artifacts.ArtifactTranscriptFull,
|
||||
artifacts.ArtifactTranscriptTrimmed,
|
||||
artifacts.ArtifactBoundsSession,
|
||||
}
|
||||
for _, entry := range catalog.ListConfigured() {
|
||||
out = append(out, entry.SourceID)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func writeLocks(out io.Writer, cfg *config.Config, locks *effectiveLocks) {
|
||||
if locks == nil || len(locks.All) == 0 {
|
||||
fmt.Fprintln(out, "Archive locks: none")
|
||||
return
|
||||
}
|
||||
fmt.Fprintln(out, "Archive locks:")
|
||||
promoted := map[string]config.ArchivePromotionRule{}
|
||||
if cfg != nil && cfg.Pipeline != nil && cfg.Pipeline.Archive != nil {
|
||||
for _, rule := range cfg.Pipeline.Archive.PromoteArtifacts {
|
||||
promoted[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-promoted"
|
||||
if _, ok := promoted[lock.Source]; ok {
|
||||
promo = "promoted"
|
||||
}
|
||||
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.ArchiveLockRule) []config.ArchiveLockRule {
|
||||
keys := make([]string, 0, len(in))
|
||||
for key := range in {
|
||||
keys = append(keys, key)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
out := make([]config.ArchiveLockRule, 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
|
||||
}
|
||||
Reference in New Issue
Block a user