Extract shared read-only session inspection checks
This commit is contained in:
274
internal/app/operator_inspection.go
Normal file
274
internal/app/operator_inspection.go
Normal file
@@ -0,0 +1,274 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"path"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
||||
)
|
||||
|
||||
type stableInputCheck struct {
|
||||
Name string
|
||||
Path string
|
||||
Err error
|
||||
}
|
||||
|
||||
type localAudioCheck struct {
|
||||
Checked bool
|
||||
Paths []string
|
||||
Err error
|
||||
}
|
||||
|
||||
type remoteAudioCheck struct {
|
||||
Checked bool
|
||||
Prefix string
|
||||
Keys []string
|
||||
Err error
|
||||
}
|
||||
|
||||
type previousArtifactReadiness struct {
|
||||
Requirements []artifacts.PreviousArtifactRequirement
|
||||
MissingID bool
|
||||
Err error
|
||||
}
|
||||
|
||||
type remoteCurrentStateCheck struct {
|
||||
State *RemoteCurrentState
|
||||
Err error
|
||||
}
|
||||
|
||||
type effectiveLocksCheck struct {
|
||||
Locks *effectiveLocks
|
||||
Err error
|
||||
}
|
||||
|
||||
func inspectStableInputs(cfg *config.Config) []stableInputCheck {
|
||||
items := []struct {
|
||||
name string
|
||||
in config.ResolvedInputFile
|
||||
}{
|
||||
{name: "speakers", in: cfg.StableInputs.SpeakersFile},
|
||||
{name: "autocorrect", in: cfg.StableInputs.AutocorrectFile},
|
||||
{name: "glossary", in: cfg.StableInputs.GlossaryFile},
|
||||
}
|
||||
out := make([]stableInputCheck, 0, len(items))
|
||||
for _, item := range items {
|
||||
path, err := resolveHelperConfigRelativePath(item.in)
|
||||
if err != nil {
|
||||
out = append(out, stableInputCheck{Name: item.name, Err: err})
|
||||
continue
|
||||
}
|
||||
if _, err := os.Stat(path); err != nil {
|
||||
out = append(out, stableInputCheck{Name: item.name, Path: path, Err: err})
|
||||
continue
|
||||
}
|
||||
out = append(out, stableInputCheck{Name: item.name, Path: path})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func inspectLocalAudioPresence(cfg *config.Config) localAudioCheck {
|
||||
if cfg.Session.Inputs.AudioS3 != nil {
|
||||
return localAudioCheck{}
|
||||
}
|
||||
|
||||
sessionDir := filepath.Dir(cfg.SessionPath)
|
||||
resolved, err := resolveLocalInspectionAudioPaths(sessionDir, cfg.Session.Inputs)
|
||||
if err != nil {
|
||||
return localAudioCheck{Checked: true, Err: err}
|
||||
}
|
||||
return localAudioCheck{
|
||||
Checked: true,
|
||||
Paths: resolved,
|
||||
}
|
||||
}
|
||||
|
||||
func inspectRemoteAudioPresence(ctx context.Context, cfg *config.Config, store storage.ObjectStore) remoteAudioCheck {
|
||||
if cfg.Session.Inputs.AudioS3 == nil {
|
||||
return remoteAudioCheck{}
|
||||
}
|
||||
if store == nil {
|
||||
return remoteAudioCheck{Checked: true, Err: fmt.Errorf("storage backend is required for remote audio checks")}
|
||||
}
|
||||
|
||||
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 remoteAudioCheck{Checked: true, Prefix: audioPrefix, Err: err}
|
||||
}
|
||||
|
||||
keys := make([]string, 0, len(objects))
|
||||
seenBase := map[string]string{}
|
||||
for _, obj := range objects {
|
||||
key := strings.TrimSpace(obj.Key)
|
||||
if key == "" || strings.HasSuffix(key, "/") || !isInspectionFlacPath(key) {
|
||||
continue
|
||||
}
|
||||
base := path.Base(key)
|
||||
if prev, exists := seenBase[base]; exists && prev != key {
|
||||
return remoteAudioCheck{
|
||||
Checked: true,
|
||||
Prefix: audioPrefix,
|
||||
Err: fmt.Errorf("duplicate s3 audio basename %q from %q and %q", base, prev, key),
|
||||
}
|
||||
}
|
||||
seenBase[base] = key
|
||||
keys = append(keys, key)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
if len(keys) == 0 {
|
||||
return remoteAudioCheck{
|
||||
Checked: true,
|
||||
Prefix: audioPrefix,
|
||||
Err: fmt.Errorf("no .flac files found under s3 audio prefix %q", audioPrefix),
|
||||
}
|
||||
}
|
||||
return remoteAudioCheck{
|
||||
Checked: true,
|
||||
Prefix: audioPrefix,
|
||||
Keys: keys,
|
||||
}
|
||||
}
|
||||
|
||||
func inspectPreviousArtifactReadiness(
|
||||
ctx context.Context,
|
||||
cfg *config.Config,
|
||||
store storage.ObjectStore,
|
||||
requirements []artifacts.PreviousArtifactRequirement,
|
||||
) previousArtifactReadiness {
|
||||
out := previousArtifactReadiness{
|
||||
Requirements: append([]artifacts.PreviousArtifactRequirement(nil), requirements...),
|
||||
}
|
||||
if len(requirements) == 0 {
|
||||
return out
|
||||
}
|
||||
if strings.TrimSpace(cfg.Session.PreviousSessionID) == "" {
|
||||
out.MissingID = true
|
||||
return out
|
||||
}
|
||||
if store == nil {
|
||||
out.Err = fmt.Errorf("previous-session artifacts cannot be checked because storage is unavailable")
|
||||
return out
|
||||
}
|
||||
|
||||
prefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, cfg.Session.Campaign, cfg.Session.PreviousSessionID)
|
||||
if _, err := artifacts.LoadCurrentState(ctx, store, prefix, artifacts.CurrentStateValidation{
|
||||
ExpectedSessionID: strings.TrimSpace(cfg.Session.PreviousSessionID),
|
||||
ExpectedCampaign: strings.TrimSpace(cfg.Session.Campaign),
|
||||
ValidateRunID: true,
|
||||
}); err != nil {
|
||||
out.Err = fmt.Errorf("remote %v", err)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func inspectRemoteCurrentState(ctx context.Context, cfg *config.Config, store storage.ObjectStore) remoteCurrentStateCheck {
|
||||
if store == nil {
|
||||
return remoteCurrentStateCheck{}
|
||||
}
|
||||
current, err := discoverRemoteCurrentStateFn(ctx, cfg, store)
|
||||
if err != nil {
|
||||
return remoteCurrentStateCheck{Err: err}
|
||||
}
|
||||
return remoteCurrentStateCheck{State: current}
|
||||
}
|
||||
|
||||
func inspectEffectiveLocks(ctx context.Context, cfg *config.Config, store storage.ObjectStore) effectiveLocksCheck {
|
||||
locks, err := loadEffectiveLocks(ctx, cfg, store)
|
||||
if err != nil {
|
||||
return effectiveLocksCheck{Err: err}
|
||||
}
|
||||
return effectiveLocksCheck{Locks: locks}
|
||||
}
|
||||
|
||||
func resolveLocalInspectionAudioPaths(sessionDir string, inputs config.SessionInputsConfig) ([]string, error) {
|
||||
if len(inputs.AudioFiles) > 0 {
|
||||
out := make([]string, 0, len(inputs.AudioFiles))
|
||||
seenBase := map[string]string{}
|
||||
for _, item := range inputs.AudioFiles {
|
||||
resolved, err := resolveInspectionPath(sessionDir, item)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !isInspectionFlacPath(resolved) {
|
||||
return nil, fmt.Errorf("audio file %q must have .flac extension", resolved)
|
||||
}
|
||||
if err := requireInspectionFile(resolved, "audio file"); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
base := filepath.Base(resolved)
|
||||
if prev, exists := seenBase[base]; exists && prev != resolved {
|
||||
return nil, fmt.Errorf("duplicate audio basename %q from %q and %q", base, prev, resolved)
|
||||
}
|
||||
seenBase[base] = resolved
|
||||
out = append(out, resolved)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
audioDir, err := resolveInspectionPath(sessionDir, inputs.AudioDir)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
entries, err := os.ReadDir(audioDir)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("read audio directory %q: %w", audioDir, err)
|
||||
}
|
||||
out := make([]string, 0, len(entries))
|
||||
for _, entry := range entries {
|
||||
if entry.IsDir() {
|
||||
continue
|
||||
}
|
||||
full := filepath.Join(audioDir, entry.Name())
|
||||
if !isInspectionFlacPath(full) {
|
||||
continue
|
||||
}
|
||||
if err := requireInspectionFile(full, "audio file"); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, full)
|
||||
}
|
||||
if len(out) == 0 {
|
||||
return nil, fmt.Errorf("no .flac files found in audio directory %q", audioDir)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func resolveInspectionPath(baseDir, inputPath string) (string, error) {
|
||||
pathValue := strings.TrimSpace(inputPath)
|
||||
if pathValue == "" {
|
||||
return "", fmt.Errorf("path is required")
|
||||
}
|
||||
if filepath.IsAbs(pathValue) {
|
||||
return filepath.Clean(pathValue), nil
|
||||
}
|
||||
return filepath.Clean(filepath.Join(baseDir, pathValue)), nil
|
||||
}
|
||||
|
||||
func requireInspectionFile(path, label string) error {
|
||||
info, err := os.Stat(path)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return fmt.Errorf("%s %q does not exist", label, path)
|
||||
}
|
||||
return fmt.Errorf("stat %s %q: %w", label, path, err)
|
||||
}
|
||||
if info.IsDir() {
|
||||
return fmt.Errorf("%s %q is a directory", label, path)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func isInspectionFlacPath(path string) bool {
|
||||
return strings.EqualFold(filepath.Ext(strings.TrimSpace(path)), ".flac")
|
||||
}
|
||||
Reference in New Issue
Block a user