package manifest import ( "bytes" "context" "encoding/json" "fmt" "io" "os" "path/filepath" "strings" "time" "gitea.maximumdirect.net/eric/narratio/internal/fileops" "gitea.maximumdirect.net/eric/narratio/internal/pathsafe" ) // Store persists manifests to and from durable storage. type Store interface { Create(ctx context.Context, sessionID string) (*Manifest, error) Load(ctx context.Context, path string) (*Manifest, error) Save(ctx context.Context, path string, m *Manifest) error } // LocalStore stores manifests as JSON on the local filesystem. type LocalStore struct{} // Create returns a new in-memory manifest for a session. func (s *LocalStore) Create(ctx context.Context, sessionID string) (*Manifest, error) { if err := checkContext(ctx); err != nil { return nil, err } if strings.TrimSpace(sessionID) == "" { return nil, fmt.Errorf("create manifest: sessionID is required") } now := time.Now().UTC() return New(sessionID, now), nil } // Load reads and validates a local JSON manifest from path. func (s *LocalStore) Load(ctx context.Context, path string) (*Manifest, error) { if err := checkContext(ctx); err != nil { return nil, err } if strings.TrimSpace(path) == "" { return nil, fmt.Errorf("load manifest: path is required") } file, err := os.Open(path) if err != nil { return nil, fmt.Errorf("load manifest %q: %w", path, err) } defer file.Close() return s.LoadReader(ctx, file) } // LoadReader reads and validates a manifest from a caller-owned reader. func (s *LocalStore) LoadReader(ctx context.Context, source io.Reader) (*Manifest, error) { if err := checkContext(ctx); err != nil { return nil, err } if source == nil { return nil, fmt.Errorf("load manifest: source is required") } data, err := io.ReadAll(source) if err != nil { return nil, fmt.Errorf("read manifest: %w", err) } var m Manifest decoder := json.NewDecoder(bytes.NewReader(data)) decoder.DisallowUnknownFields() if err := decoder.Decode(&m); err != nil { return nil, fmt.Errorf("decode manifest: %w", err) } var trailing any if err := decoder.Decode(&trailing); err != io.EOF { if err == nil { return nil, fmt.Errorf("decode manifest: multiple JSON values are not allowed") } return nil, fmt.Errorf("decode manifest: %w", err) } if err := validateLoadedManifest(&m); err != nil { return nil, fmt.Errorf("manifest invalid: %w", err) } normalizeManifest(&m) return &m, nil } // Save writes the manifest to path through the durable file replacement primitive. func (s *LocalStore) Save(ctx context.Context, path string, m *Manifest) error { if err := checkContext(ctx); err != nil { return err } if strings.TrimSpace(path) == "" { return fmt.Errorf("save manifest: path is required") } if m == nil { return fmt.Errorf("save manifest: manifest is nil") } if err := validateSessionManifestSchemaVersion(m.SchemaVersion); err != nil { return fmt.Errorf("save manifest: %w", err) } if strings.TrimSpace(m.SessionID) == "" { return fmt.Errorf("save manifest: session_id is required") } if m.CreatedAt.IsZero() { return fmt.Errorf("save manifest: created_at is required") } if err := validateManifestIdentities(m.SessionID, m.Campaign, m.RunID); err != nil { return fmt.Errorf("save manifest: %w", err) } if err := validateSessionStages(m.Stages); err != nil { return fmt.Errorf("save manifest: %w", err) } for name, stage := range m.Stages { normalizeAnalyzeArtifactCollection(stage.AnalyzeArtifacts) if err := validateTranscribeStageState(name, stage.TranscribeStateVersion, stage.TranscribeSpeakers); err != nil { return fmt.Errorf("save manifest: stages.%s: %w", name, err) } if err := validateAnalyzeStageState(name, stage.AnalyzeStateVersion, stage.AnalyzeArtifacts); err != nil { return fmt.Errorf("save manifest: stages.%s: %w", name, err) } } m.UpdatedAt = time.Now().UTC() data, err := json.MarshalIndent(m, "", " ") if err != nil { return fmt.Errorf("save manifest: marshal: %w", err) } data = append(data, '\n') if err := writeJSONAtomically(ctx, path, data); err != nil { return fmt.Errorf("save manifest: %w", err) } return nil } func validateLoadedManifest(m *Manifest) error { if m == nil { return fmt.Errorf("manifest is nil") } if err := validateSessionManifestSchemaVersion(m.SchemaVersion); err != nil { return err } if strings.TrimSpace(m.SessionID) == "" { return fmt.Errorf("session_id is required") } if m.CreatedAt.IsZero() { return fmt.Errorf("created_at is required") } if m.UpdatedAt.IsZero() { return fmt.Errorf("updated_at is required") } if err := validateManifestIdentities(m.SessionID, m.Campaign, m.RunID); err != nil { return err } if err := validateSessionStages(m.Stages); err != nil { return err } for name, stage := range m.Stages { if err := validateTranscribeStageState(name, stage.TranscribeStateVersion, stage.TranscribeSpeakers); err != nil { return fmt.Errorf("stages.%s: %w", name, err) } if err := validateAnalyzeStageState(name, stage.AnalyzeStateVersion, stage.AnalyzeArtifacts); err != nil { return fmt.Errorf("stages.%s: %w", name, err) } if stage.SemanticConfig != nil { if err := stage.SemanticConfig.Validate(); err != nil { return fmt.Errorf("stages.%s.semantic_config: %w", name, err) } } } return nil } func normalizeManifest(m *Manifest) { for _, stage := range m.Stages { normalizeAnalyzeArtifactCollection(stage.AnalyzeArtifacts) } } func validateManifestIdentities(sessionID, campaign, runID string) error { identities := []struct { field string value string required bool }{ {field: "session_id", value: sessionID, required: true}, {field: "campaign", value: campaign}, {field: "run_id", value: runID}, } for _, identity := range identities { value := strings.TrimSpace(identity.value) if value == "" && !identity.required { continue } if err := pathsafe.ValidateOpaqueSegment(identity.value); err != nil { return fmt.Errorf( "manifest %s %q is not a portable opaque identifier; migrate the legacy manifest before use: %w", identity.field, identity.value, err, ) } } return nil } func validateSessionManifestSchemaVersion(version string) error { return validateManifestSchemaVersion("session manifest", version, SessionManifestSchemaVersion) } func validateManifestSchemaVersion(kind, version, supported string) error { if version == supported { return nil } if version == "" { return fmt.Errorf("%s schema_version is required; replace or regenerate this unsupported state (supported: %s)", kind, supported) } return fmt.Errorf("unsupported %s schema_version %q; replace or regenerate this unsupported state (supported: %s)", kind, version, supported) } func validateSessionStages(stages map[string]*StageRecord) error { if stages == nil { return fmt.Errorf("stages is required") } for name, stage := range stages { if stage == nil { return fmt.Errorf("stages.%s is nil", name) } if stage.Name == "" { return fmt.Errorf("stages.%s.name is required", name) } if stage.Name != name { return fmt.Errorf("stages.%s.name %q does not match map key", name, stage.Name) } } return nil } func writeJSONAtomically(ctx context.Context, path string, data []byte) error { dir := filepath.Dir(path) if err := fileops.EnsureWorkspaceDirectory(dir); err != nil { return fmt.Errorf("create directory %q: %w", dir, err) } return fileops.ReplaceFileAtomic(path, data, fileops.ReplaceFileOptions{ Mode: fileops.WorkspaceFileMode, BeforeRename: func() error { return checkContext(ctx) }, }) } func checkContext(ctx context.Context) error { if ctx == nil { return nil } select { case <-ctx.Done(): return ctx.Err() default: return nil } }