narratio/internal/manifest/store.go

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
	}
}