narratio/internal/stage/publish.go

package stage

import (
	"context"
	"fmt"
	"os"
	"path/filepath"
	"strings"

	"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
)

type publishStage struct{}

var publishPrerequisiteStages = []string{
	"prepare",
	"transcribe",
	"merge",
	"polish",
	"normalize",
	"trim",
	"render",
	"analyze",
}

func (publishStage) Name() string { return "publish" }

func (publishStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
	if env == nil || env.Config == nil || env.Config.Pipeline == nil || env.Config.Session == nil {
		return nil, fmt.Errorf("publish: resolved config must include pipeline and session")
	}

	if publishDisabled(env) {
		return &StageResult{
			Metadata: map[string]any{
				"stage":                "publish",
				"skipped":              true,
				"publish_enabled":      false,
				"audio_upload_skipped": true,
			},
		}, nil
	}
	if publishRunUploadDisabled(env) {
		return &StageResult{
			Metadata: map[string]any{
				"stage":                "publish",
				"skipped":              true,
				"upload_run_enabled":   false,
				"audio_upload_skipped": true,
			},
		}, nil
	}

	if err := validatePublishPrerequisites(m); err != nil {
		return nil, fmt.Errorf("publish: %w", err)
	}
	if env.ObjectStore == nil {
		return nil, fmt.Errorf("publish: remote object store backend is required when publish run upload is enabled")
	}

	runRoot, err := resolvePublishRunRoot(env, m)
	if err != nil {
		return nil, fmt.Errorf("publish: resolve run root: %w", err)
	}

	runPrefix, err := artifacts.ResolvePublishRunPrefix(env.Config, m)
	if err != nil {
		return nil, fmt.Errorf("publish: resolve s3 run prefix: %w", err)
	}
	sessionPrefix, err := artifacts.ResolvePublishSessionPrefix(env.Config, m)
	if err != nil {
		return nil, fmt.Errorf("publish: resolve s3 session prefix: %w", err)
	}
	bucket := artifacts.ResolvePublishBucket(env.Config, m)
	if bucket == "" {
		return nil, fmt.Errorf("publish: resolve s3 bucket: bucket is required")
	}
	runID := strings.TrimSpace(m.RunID)
	if runID == "" {
		return nil, fmt.Errorf("publish: run id is required")
	}

	runFiles, err := collectPublishRunFiles(runRoot, m)
	if err != nil {
		return nil, fmt.Errorf("publish: collect run files: %w", err)
	}
	runSources, err := openPublishRunFiles(runRoot, runFiles)
	if err != nil {
		return nil, fmt.Errorf("publish: verify run files: %w", err)
	}
	defer closePublishSources(runSources)
	uploader, ok := env.ObjectStore.(storage.ReaderUploader)
	if !ok {
		return nil, fmt.Errorf("publish: object store does not support verified source uploads")
	}
	sessionPaths := publishSessionPaths(env, m)
	previousFiles, err := collectPublishPreviousFiles(ctx, sessionPaths.PreviousDir)
	if err != nil {
		return nil, fmt.Errorf("publish: collect previous files: %w", err)
	}
	previousSources, err := openPublishRunFiles(sessionPaths.PreviousDir, previousFiles)
	if err != nil {
		return nil, fmt.Errorf("publish: verify previous files: %w", err)
	}
	defer closePublishSources(previousSources)
	runtimeCatalog, err := buildPublishRuntimeArtifactCatalog(
		sessionPaths,
		m,
		env.Config.Pipeline.Scriptorium,
		env.Config.Pipeline.Notarius,
	)
	if err != nil {
		return nil, fmt.Errorf("publish: build runtime artifact catalog: %w", err)
	}
	publishRules, err := resolvePublishRules(
		runtimeCatalog,
		env.Config.Pipeline.Publish.Outputs,
		env.SelectedArtifactKeys,
	)
	if err != nil {
		return nil, fmt.Errorf("publish: resolve publish output rules: %w", err)
	}
	initialLocks, err := readPublishOutputLocks(ctx, env.ObjectStore, sessionPrefix, publishRules)
	if err != nil {
		return nil, fmt.Errorf("publish: read publish output locks: %w", err)
	}
	var priorCommit *artifacts.RemoteCommitManifest
	if publishRulesNeedPriorCommit(publishRules, initialLocks) {
		priorCommit, err = loadPriorPublishCommit(ctx, env.ObjectStore, sessionPrefix, m.Campaign, m.SessionID)
		if err != nil {
			return nil, fmt.Errorf("publish: load prior current commit: %w", err)
		}
	}
	discoveredOutputs, err := discoverPublishOutputs(sessionPaths, m, runtimeCatalog, publishRules, initialLocks)
	if err != nil {
		return nil, fmt.Errorf("publish: discover publish outputs: %w", err)
	}
	outputPlan, err := buildPublishOutputPlan(sessionPrefix, runPrefix, publishRules, discoveredOutputs, initialLocks, priorCommit)
	if err != nil {
		return nil, fmt.Errorf("publish: plan publish outputs: %w", err)
	}

	runUploaded := publishUploadPaths(runFiles)
	publishedUploaded := publishOutputPaths(outputPlan)
	previousUploaded := publishUploadPaths(previousFiles)
	commitKey := artifacts.S3RunCommitKey(sessionPrefix, runID)
	pointerKey := artifacts.S3CurrentCommitPointerKey(sessionPrefix)

	carriedDeclarations, err := verifyCarriedPublishOutputs(ctx, env.ObjectStore, outputPlan)
	if err != nil {
		return nil, fmt.Errorf("publish: %w", err)
	}
	transfers, err := buildPublishTransferPlan(runPrefix, runFiles, runSources, previousFiles, previousSources, outputPlan)
	if err != nil {
		return nil, fmt.Errorf("publish: build immutable upload mapping: %w", err)
	}
	defer func() { closePublishTransfers(transfers) }()

	manifestTempPath, err := writeCommittedSessionManifest(m, publishMetadata(
		bucket,
		runPrefix,
		runID,
		runUploaded,
		publishedUploaded,
		previousUploaded,
		outputPlan,
		commitKey,
		pointerKey,
	))
	if err != nil {
		return nil, fmt.Errorf("publish: build committed session manifest: %w", err)
	}
	if err := appendCommittedSessionManifest(&transfers, manifestTempPath, artifacts.S3RunSessionManifestKey(sessionPrefix, runID)); err != nil {
		return nil, fmt.Errorf("publish: add committed session manifest: %w", err)
	}
	if err := validatePublishTransferPlan(transfers, commitKey); err != nil {
		return nil, fmt.Errorf("publish: validate immutable upload mapping: %w", err)
	}

	// Durability boundary: transfer every declared immutable object, write the
	// immutable commit, perform pre-pointer checks, then write the sole mutable
	// current pointer exactly once.
	uploadedDeclarations, err := uploadPublishTransfers(ctx, env.ObjectStore, uploader, transfers)
	if err != nil {
		return nil, fmt.Errorf("publish: %w", err)
	}
	declarations := combinePublishDeclarations(uploadedDeclarations, carriedDeclarations)
	committed, err := writeImmutablePublishCommit(
		ctx,
		env.ObjectStore,
		uploader,
		sessionPrefix,
		m.Campaign,
		m.SessionID,
		runID,
		declarations,
	)
	if err != nil {
		return nil, fmt.Errorf("publish: %w", err)
	}
	if err := revalidatePublishOutputLocks(ctx, env.ObjectStore, sessionPrefix, outputPlan); err != nil {
		return nil, fmt.Errorf("publish: revalidate locks before current commit selection: %w", err)
	}
	if err := writePublishCurrentPointer(ctx, uploader, committed); err != nil {
		return nil, fmt.Errorf("publish: %w", err)
	}

	return &StageResult{Metadata: publishMetadata(
		bucket,
		runPrefix,
		runID,
		runUploaded,
		publishedUploaded,
		previousUploaded,
		outputPlan,
		commitKey,
		pointerKey,
	)}, nil
}

func publishDisabled(env *Env) bool {
	cfg := env.Config.Pipeline.Publish
	if cfg == nil {
		return true
	}
	return cfg.Enabled != nil && !*cfg.Enabled
}

func publishRunUploadDisabled(env *Env) bool {
	cfg := env.Config.Pipeline.Publish
	if cfg == nil {
		return true
	}
	return cfg.UploadRun != nil && !*cfg.UploadRun
}

func validatePublishPrerequisites(m *manifest.Manifest) error {
	if m == nil {
		return fmt.Errorf("manifest is required")
	}
	for _, stageName := range publishPrerequisiteStages {
		sr := m.Stages[stageName]
		if sr == nil {
			return fmt.Errorf("prerequisite stage %q has not succeeded", stageName)
		}
		if sr.Status != manifest.StatusSucceeded {
			return fmt.Errorf("prerequisite stage %q status is %q (want %q)", stageName, sr.Status, manifest.StatusSucceeded)
		}
	}
	return nil
}

func resolvePublishRunRoot(env *Env, m *manifest.Manifest) (string, error) {
	sessionID := strings.TrimSpace(env.Config.Session.SessionID)
	if sessionID == "" && m != nil {
		sessionID = strings.TrimSpace(m.SessionID)
	}
	campaign := strings.TrimSpace(env.Config.Session.Campaign)
	if campaign == "" && m != nil {
		campaign = strings.TrimSpace(m.Campaign)
	}
	runID := ""
	if m != nil {
		runID = strings.TrimSpace(m.RunID)
	}
	if sessionID == "" || campaign == "" {
		return "", fmt.Errorf("campaign and session id are required")
	}
	if runID == "" {
		return "", fmt.Errorf("run id is required")
	}

	canonical := filepath.Clean(artifacts.SessionRunRootForCampaign(env.Config.Pipeline.Workspace.Root, campaign, sessionID, runID))
	info, err := os.Lstat(canonical)
	if err != nil {
		return "", fmt.Errorf("inspect canonical run root %q: %w", canonical, err)
	}
	if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
		return "", fmt.Errorf("canonical run root %q is not a directory", canonical)
	}
	return canonical, nil
}

func publishSessionPaths(env *Env, m *manifest.Manifest) artifacts.SessionPaths {
	sessionID := strings.TrimSpace(env.Config.Session.SessionID)
	if sessionID == "" && m != nil {
		sessionID = strings.TrimSpace(m.SessionID)
	}
	campaign := strings.TrimSpace(env.Config.Session.Campaign)
	if campaign == "" && m != nil {
		campaign = strings.TrimSpace(m.Campaign)
	}
	store := artifacts.NewLocalStore(env.Config.Pipeline.Workspace.Root)
	return store.SessionPathsFor(campaign, sessionID)
}

func publishMetadata(
	bucket, runPrefix, runID string,
	runUploaded []string,
	publishedUploaded []string,
	previousUploaded []string,
	outputPlan *publishOutputPlan,
	commitKey string,
	pointerKey string,
) map[string]any {
	return map[string]any{
		"stage":                      "publish",
		"uploaded":                   true,
		"s3_bucket":                  bucket,
		"s3_run_prefix":              runPrefix,
		"published_run_id":           runID,
		"run_files_uploaded":         len(runUploaded),
		"run_uploaded_paths":         append([]string(nil), runUploaded...),
		"published_files_uploaded":   len(publishedUploaded),
		"published_paths":            append([]string(nil), publishedUploaded...),
		"previous_files_uploaded":    len(previousUploaded),
		"previous_uploaded_paths":    append([]string(nil), previousUploaded...),
		"publish_output_outcomes":    publishOutputOutcomeMetadata(outputPlan),
		"remote_commit_key":          commitKey,
		"current_commit_pointer_key": pointerKey,
		"audio_upload_skipped":       true,
	}
}