package stage import ( "bytes" "context" "crypto/sha256" "encoding/hex" "encoding/json" "fmt" "io" "os" "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/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) } publishOutputs, skippedOptionalOutputs, skippedUnselectedOutputs, lockedOutputs, err := resolvePublishOutputs( sessionPaths, m, runtimeCatalog, env.Config.Pipeline.Publish.Outputs, env.Config.Pipeline.Publish.Locks, env.SelectedArtifactKeys, runPrefix, ) if err != nil { return nil, fmt.Errorf("publish: resolve publish output rules: %w", err) } runUploaded := publishUploadPaths(runFiles) publishedUploaded := publishOutputPaths(publishOutputs) previousUploaded := publishUploadPaths(previousFiles) commitKey := artifacts.S3RunCommitKey(sessionPrefix, runID) pointerKey := artifacts.S3CurrentCommitPointerKey(sessionPrefix) transfers, err := buildPublishTransferPlan(runPrefix, runFiles, runSources, previousFiles, previousSources, publishOutputs) 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, skippedOptionalOutputs, skippedUnselectedOutputs, lockedOutputs, 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) } commit := artifacts.RemoteCommitManifest{ FormatVersion: artifacts.RemoteCommitFormatVersion, Campaign: strings.TrimSpace(m.Campaign), SessionID: strings.TrimSpace(m.SessionID), RunID: runID, Artifacts: make([]artifacts.RemoteArtifact, 0, len(transfers)), } for index := range transfers { artifact, err := uploadVerifiedPublishTransfer(ctx, env.ObjectStore, uploader, &transfers[index]) if err != nil { return nil, fmt.Errorf("publish: upload immutable object %q: %w", transfers[index].Artifact.DestinationKey, err) } commit.Artifacts = append(commit.Artifacts, artifact) } commitData, err := artifacts.EncodeRemoteCommitManifest(commit) if err != nil { return nil, fmt.Errorf("publish: encode immutable commit: %w", err) } commitInfo, err := uploadVerifiedPublishBytes(ctx, env.ObjectStore, uploader, commitKey, commitData, "application/json") if err != nil { return nil, fmt.Errorf("publish: upload immutable commit %q: %w", commitKey, err) } if err := revalidatePublishCommitLocks(ctx, env, publishOutputs); err != nil { return nil, fmt.Errorf("publish: revalidate locks before current commit selection: %w", err) } pointerData, err := artifacts.EncodeCurrentCommitPointer(artifacts.CurrentCommitPointer{ FormatVersion: artifacts.RemoteCommitFormatVersion, Campaign: commit.Campaign, SessionID: commit.SessionID, RunID: commit.RunID, CommitKey: commitKey, CommitSHA256: publishSHA256(commitData), CommitSize: int64(len(commitData)), CommitGeneration: commitInfo.ETag, }) if err != nil { return nil, fmt.Errorf("publish: encode current commit pointer: %w", err) } if _, err := uploader.UploadReader(ctx, bytes.NewReader(pointerData), pointerKey, storage.UploadOptions{ContentType: "application/json"}); err != nil { return nil, fmt.Errorf("publish: update current commit pointer %q: %w", pointerKey, err) } return &StageResult{ Metadata: map[string]any{ "stage": "publish", "uploaded": true, "s3_bucket": bucket, "s3_run_prefix": runPrefix, "run_files_uploaded": len(runUploaded), "run_uploaded_paths": runUploaded, "published_files_uploaded": len(publishedUploaded), "published_paths": publishedUploaded, "previous_files_uploaded": len(previousUploaded), "previous_uploaded_paths": previousUploaded, "skipped_optional_outputs": skippedOptionalOutputs, "skipped_unselected_outputs": skippedUnselectedOutputMetadata(skippedUnselectedOutputs), "locked_output_count": len(lockedOutputs), "locked_outputs": lockedOutputMetadata(lockedOutputs), "remote_commit_key": commitKey, "current_commit_pointer_key": pointerKey, "audio_upload_skipped": true, }, }, nil } func revalidatePublishCommitLocks(ctx context.Context, env *Env, outputs []publishOutput) error { if env == nil || env.RevalidatePublishLocks == nil || len(outputs) == 0 { return nil } locks, err := env.RevalidatePublishLocks(ctx) if err != nil { return err } lockSet := publishLockSet(locks) for _, output := range outputs { if lock, locked := lockSet[output.Source]; locked { return fmt.Errorf("source %q is locked: %s", output.Source, strings.TrimSpace(lock.Reason)) } } return nil } type publishTransfer struct { Artifact artifacts.RemoteArtifact Source *os.File TemporaryPath string } func buildPublishTransferPlan( runPrefix string, runFiles []publishUploadFile, runSources []*os.File, previousFiles []publishUploadFile, previousSources []*os.File, publishOutputs []publishOutput, ) ([]publishTransfer, error) { if len(runFiles) != len(runSources) || len(previousFiles) != len(previousSources) { return nil, fmt.Errorf("verified publish sources do not match their declared files") } transfers := make([]publishTransfer, 0, len(runFiles)+len(previousFiles)+len(publishOutputs)) add := func(artifact artifacts.RemoteArtifact, source *os.File) error { if source == nil { return fmt.Errorf("source for %q is required", artifact.DestinationKey) } transfers = append(transfers, publishTransfer{Artifact: artifact, Source: source}) return nil } for index, file := range runFiles { if err := add(artifacts.RemoteArtifact{ Type: artifacts.RemoteArtifactTypeRunFile, Source: "run:" + file.RelativePath, DestinationKey: artifacts.S3RunRelativeDestinationKey(runPrefix, file.RelativePath), }, runSources[index]); err != nil { closePublishTransfers(transfers) return nil, err } } for index, file := range previousFiles { if err := add(artifacts.RemoteArtifact{ Type: artifacts.RemoteArtifactTypePreviousArtifact, Source: "previous:" + file.SourceRelativePath, DestinationKey: artifacts.S3RunRelativeDestinationKey(runPrefix, file.RelativePath), }, previousSources[index]); err != nil { closePublishTransfers(transfers) return nil, err } } for _, output := range publishOutputs { source, err := openPublishOutputSource(output.LocalPath) if err != nil { closePublishTransfers(transfers) return nil, fmt.Errorf("open output source %q: %w", output.Source, err) } if err := add(artifacts.RemoteArtifact{ Type: artifacts.RemoteArtifactTypePublishedOutput, Source: output.Source, DestinationKey: publishOutputRemoteKey(runPrefix, output.Dest), }, source); err != nil { _ = source.Close() closePublishTransfers(transfers) return nil, err } } sort.Slice(transfers, func(i, j int) bool { return transfers[i].Artifact.DestinationKey < transfers[j].Artifact.DestinationKey }) return transfers, nil } func appendCommittedSessionManifest(transfers *[]publishTransfer, localPath, destinationKey string) error { if transfers == nil { return fmt.Errorf("transfer plan is required") } source, err := os.Open(localPath) if err != nil { return err } *transfers = append(*transfers, publishTransfer{ Artifact: artifacts.RemoteArtifact{ Type: artifacts.RemoteArtifactTypeSessionManifest, Source: "session.manifest", DestinationKey: destinationKey, }, Source: source, TemporaryPath: localPath, }) sort.Slice(*transfers, func(i, j int) bool { return (*transfers)[i].Artifact.DestinationKey < (*transfers)[j].Artifact.DestinationKey }) return nil } func validatePublishTransferPlan(transfers []publishTransfer, commitKey string) error { artifactsToUpload := make([]artifacts.RemoteArtifact, 0, len(transfers)) for _, transfer := range transfers { if transfer.Artifact.DestinationKey == commitKey { return fmt.Errorf("artifact destination %q is reserved for the immutable commit", commitKey) } artifactsToUpload = append(artifactsToUpload, transfer.Artifact) } return artifacts.ValidateRemoteArtifactMapping(artifactsToUpload) } func uploadVerifiedPublishTransfer( ctx context.Context, store storage.ObjectStore, uploader storage.ReaderUploader, transfer *publishTransfer, ) (artifacts.RemoteArtifact, error) { if transfer == nil || transfer.Source == nil { return artifacts.RemoteArtifact{}, fmt.Errorf("transfer source is required") } if err := preparePublishTransfer(transfer); err != nil { return artifacts.RemoteArtifact{}, err } if existing, found, err := verifyExistingImmutableObject(ctx, store, transfer.Artifact.DestinationKey, transfer.Artifact.SHA256, transfer.Artifact.Size); err != nil { return artifacts.RemoteArtifact{}, err } else if found { transfer.Artifact.Generation = existing.ETag return transfer.Artifact, nil } info, err := uploader.UploadReader(ctx, transfer.Source, transfer.Artifact.DestinationKey, storage.UploadOptions{}) if err != nil { return artifacts.RemoteArtifact{}, err } verified, err := verifyPublishedObject(ctx, store, transfer.Artifact.DestinationKey, transfer.Artifact.Size, info.ETag) if err != nil { return artifacts.RemoteArtifact{}, err } transfer.Artifact.Generation = verified.ETag return transfer.Artifact, nil } func uploadVerifiedPublishBytes( ctx context.Context, store storage.ObjectStore, uploader storage.ReaderUploader, key string, data []byte, contentType string, ) (storage.ObjectInfo, error) { checksum := publishSHA256(data) if existing, found, err := verifyExistingImmutableObject(ctx, store, key, checksum, int64(len(data))); err != nil { return storage.ObjectInfo{}, err } else if found { return existing, nil } info, err := uploader.UploadReader(ctx, bytes.NewReader(data), key, storage.UploadOptions{ContentType: contentType}) if err != nil { return storage.ObjectInfo{}, err } return verifyPublishedObject(ctx, store, key, int64(len(data)), info.ETag) } func verifyExistingImmutableObject(ctx context.Context, store storage.ObjectStore, key, wantSHA256 string, wantSize int64) (storage.ObjectInfo, bool, error) { exists, err := store.Exists(ctx, key) if err != nil { return storage.ObjectInfo{}, false, fmt.Errorf("check immutable object %q: %w", key, err) } if !exists { return storage.ObjectInfo{}, false, nil } path, err := storage.DownloadObjectToTemp(ctx, store, key, "narratio-verify-immutable-*.bin") if err != nil { return storage.ObjectInfo{}, false, fmt.Errorf("download existing immutable object %q: %w", key, err) } defer func() { _ = os.Remove(path) }() file, err := os.Open(path) if err != nil { return storage.ObjectInfo{}, false, fmt.Errorf("open existing immutable object %q: %w", key, err) } hash := sha256.New() size, copyErr := io.Copy(hash, file) closeErr := file.Close() if copyErr != nil { return storage.ObjectInfo{}, false, fmt.Errorf("checksum existing immutable object %q: %w", key, copyErr) } if closeErr != nil { return storage.ObjectInfo{}, false, fmt.Errorf("close existing immutable object %q: %w", key, closeErr) } if size != wantSize || hex.EncodeToString(hash.Sum(nil)) != wantSHA256 { return storage.ObjectInfo{}, false, fmt.Errorf("immutable object %q already exists with different content", key) } info, err := verifyPublishedObject(ctx, store, key, wantSize, "") if err != nil { return storage.ObjectInfo{}, false, err } return info, true, nil } func preparePublishTransfer(transfer *publishTransfer) error { if _, err := transfer.Source.Seek(0, io.SeekStart); err != nil { return fmt.Errorf("rewind source %q: %w", transfer.Artifact.Source, err) } info, err := transfer.Source.Stat() if err != nil { return fmt.Errorf("stat source %q: %w", transfer.Artifact.Source, err) } if !info.Mode().IsRegular() { return fmt.Errorf("source %q is not a regular file", transfer.Artifact.Source) } hash := sha256.New() if _, err := io.Copy(hash, transfer.Source); err != nil { return fmt.Errorf("checksum source %q: %w", transfer.Artifact.Source, err) } if _, err := transfer.Source.Seek(0, io.SeekStart); err != nil { return fmt.Errorf("rewind source %q after checksum: %w", transfer.Artifact.Source, err) } transfer.Artifact.Size = info.Size() transfer.Artifact.SHA256 = hex.EncodeToString(hash.Sum(nil)) return nil } func verifyPublishedObject(ctx context.Context, store storage.ObjectStore, key string, wantSize int64, uploadedGeneration string) (storage.ObjectInfo, error) { objects, err := store.List(ctx, key) if err != nil { return storage.ObjectInfo{}, fmt.Errorf("list uploaded object %q: %w", key, err) } var found *storage.ObjectInfo for _, object := range objects { if object.Key != key { continue } if found != nil { return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q is ambiguous", key) } copy := object found = © } if found == nil { return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q is missing", key) } if found.Size != wantSize { return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q size is %d, want %d", key, found.Size, wantSize) } if strings.TrimSpace(found.ETag) == "" { return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q has no generation", key) } if strings.TrimSpace(uploadedGeneration) != "" && found.ETag != uploadedGeneration { return storage.ObjectInfo{}, fmt.Errorf("uploaded object %q generation changed during verification", key) } return *found, nil } func closePublishTransfers(transfers []publishTransfer) { for _, transfer := range transfers { if transfer.Source != nil { _ = transfer.Source.Close() } if transfer.TemporaryPath != "" { _ = os.Remove(transfer.TemporaryPath) } } } func publishSHA256(data []byte) string { sum := sha256.Sum256(data) return hex.EncodeToString(sum[:]) } 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 writeCommittedSessionManifest(m *manifest.Manifest, publishMetadata map[string]any) (string, error) { if m == nil { return "", fmt.Errorf("manifest is required") } clone := *m clone.Stages = make(map[string]*manifest.StageRecord, len(m.Stages)) for name, sr := range m.Stages { if sr == nil { continue } stageCopy := *sr if sr.Outputs != nil { stageCopy.Outputs = append([]manifest.ArtifactRecord(nil), sr.Outputs...) } if sr.Logs != nil { stageCopy.Logs = append([]string(nil), sr.Logs...) } if sr.GeneratedConfigs != nil { stageCopy.GeneratedConfigs = append([]string(nil), sr.GeneratedConfigs...) } if sr.Metadata != nil { metaCopy := make(map[string]any, len(sr.Metadata)) for k, v := range sr.Metadata { metaCopy[k] = v } stageCopy.Metadata = metaCopy } clone.Stages[name] = &stageCopy } clone.MarkStageSucceeded("publish", clone.UpdatedAt, nil) if sr := clone.Stages["publish"]; sr != nil { sr.Metadata = publishMetadata } data, err := json.MarshalIndent(&clone, "", " ") if err != nil { return "", fmt.Errorf("marshal manifest: %w", err) } data = append(data, '\n') tmp, err := os.CreateTemp("", "narratio-committed-session-manifest-*.json") if err != nil { return "", fmt.Errorf("create temp file: %w", err) } path := tmp.Name() if _, err := tmp.Write(data); err != nil { _ = tmp.Close() return "", fmt.Errorf("write temp manifest: %w", err) } if err := tmp.Close(); err != nil { return "", fmt.Errorf("close temp manifest: %w", err) } return path, nil } func publishMetadata( bucket, runPrefix, runID string, runUploaded []string, publishedUploaded []string, previousUploaded []string, skippedOptionalOutputs []string, skippedUnselectedOutputs []publishSkippedUnselectedOutput, lockedOutputs []publishLockedOutput, 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...), "skipped_optional_outputs": append([]string(nil), skippedOptionalOutputs...), "skipped_unselected_outputs": skippedUnselectedOutputMetadata(skippedUnselectedOutputs), "locked_output_count": len(lockedOutputs), "locked_outputs": lockedOutputMetadata(lockedOutputs), "remote_commit_key": commitKey, "current_commit_pointer_key": pointerKey, "audio_upload_skipped": true, } } func skippedUnselectedOutputMetadata(skipped []publishSkippedUnselectedOutput) []map[string]any { out := make([]map[string]any, 0, len(skipped)) for _, item := range skipped { out = append(out, map[string]any{ "source": item.Source, "dest": item.Dest, "required": item.Required, }) } return out } func lockedOutputMetadata(locked []publishLockedOutput) []map[string]any { out := make([]map[string]any, 0, len(locked)) for _, item := range locked { out = append(out, map[string]any{ "source": item.Source, "dest": item.Dest, "remote_key": item.RemoteKey, "reason": item.Reason, "required": item.Required, "local_path": item.LocalPath, "provenance": item.Provenance, }) } return out }