pipeline/packages/core/src/pipeline.ts

import { randomUUID } from "node:crypto";
import { mkdir } from "node:fs/promises";
import { join } from "node:path";
import type {
  PipelineInput,
  PipelineJob,
  ExportFile,
} from "@pipeline/shared";
import { PLATFORM_PRESETS } from "@pipeline/shared";
import { parseText } from "./stages/parse.js";
import { generateTTS } from "./stages/tts.js";
import { resolveAssets } from "./stages/assets.js";
import { composeProject } from "./stages/compose.js";
import { renderVideo } from "./stages/render.js";
import { exportVideo } from "./stages/export.js";
import { cleanupExpiredOutput } from "./cleanup.js";
import {
  runPublish,
  notifyGenerationFailure,
  type PublishConfig,
} from "./publish/index.js";

export interface PipelineConfig {
  branding: {
    channelName: string;
  };
  llm: {
    baseURL?: string;
    apiKey?: string;
    model: string;
  };
  tts: {
    provider: string;
    voiceId?: string;
    model?: string;
    format?: "mp3" | "wav" | "pcm";
    speed?: number;
  };
  alignment?: {
    provider: "whisper" | "native";
    whisperModel?: string;
    language?: string;
  };
  output: {
    dir: string;
    /** Auto-clean cached outputs older than this many days (0 disables). */
    retentionDays?: number;
  };
  /** OSS upload + Feishu notification, resolved by resolvePublishConfig. */
  publish?: PublishConfig;
  /** Skip the publish stage entirely (e.g. local debugging). */
  skipPublish?: boolean;
  assets: {
    root: string;
    inputDir: string;
  };
  templates: {
    entryPoint: string;
  };
  skipTts?: boolean;
  skipLlm?: boolean;
}

export interface PipelineCallbacks {
  onStageStart?: (stage: string) => void;
  onStageComplete?: (stage: string) => void;
  onError?: (stage: string, error: Error) => void;
}

export async function runPipeline(
  input: PipelineInput,
  config: PipelineConfig,
  callbacks?: PipelineCallbacks
): Promise<PipelineJob> {
  const jobId = randomUUID();
  const workDir = join(config.output.dir, "tmp", jobId);
  await mkdir(workDir, { recursive: true });

  // Opportunistic cache sweep: prune outputs older than retentionDays.
  // Runs on every job (service jobs are low-frequency); never throws.
  if (config.output.retentionDays && config.output.retentionDays > 0) {
    await cleanupExpiredOutput(config.output.dir, config.output.retentionDays);
  }

  const platforms = input.platforms;
  const job: PipelineJob = {
    id: jobId,
    input,
    status: "running",
    createdAt: Date.now(),
    updatedAt: Date.now(),
  };

  try {
    // Stage 1: Parse (shared across all platforms)
    callbacks?.onStageStart?.("parse");
    const parsed = await parseText(input.text, input.template, {
      llm: config.llm,
      skipLlm: config.skipLlm,
      source: input.source,
    });
    job.parsed = parsed;
    callbacks?.onStageComplete?.("parse");

    // Stage 2: TTS (shared across all platforms)
    callbacks?.onStageStart?.("tts");
    const tts = await generateTTS(parsed, workDir, {
      provider: config.tts.provider,
      voiceId: config.tts.voiceId || input.voiceId,
      model: config.tts.model,
      format: config.tts.format,
      speed: config.tts.speed,
      skip: config.skipTts,
      alignment: config.alignment,
    });
    job.tts = tts;
    callbacks?.onStageComplete?.("tts");

    // Stages 3-6: Per-platform loop
    const allExportFiles: ExportFile[] = [];
    job.composed = {};

    for (const platform of platforms) {
      // Stage 3: Assets
      callbacks?.onStageStart?.(`assets:${platform}`);
      const assets = await resolveAssets(
        parsed, workDir, config.assets.root, config.assets.inputDir,
        { template: input.template, aspect: PLATFORM_PRESETS[platform].aspect }
      );
      callbacks?.onStageComplete?.(`assets:${platform}`);

      // Stage 4: Compose
      callbacks?.onStageStart?.(`compose:${platform}`);
      const composed = composeProject(parsed, tts, assets, platform, input.template, config.branding.channelName);
      job.composed[platform] = composed;
      callbacks?.onStageComplete?.(`compose:${platform}`);

      // Stage 5: Render
      callbacks?.onStageStart?.(`render:${platform}`);
      const renderOutput = join(workDir, `render-${platform}.mp4`);
      await renderVideo(composed, renderOutput, config.templates.entryPoint, config.assets.root);
      callbacks?.onStageComplete?.(`render:${platform}`);

      // Stage 6: Export
      callbacks?.onStageStart?.(`export:${platform}`);
      const exported = await exportVideo(
        renderOutput,
        composed,
        config.output.dir,
        jobId
      );
      allExportFiles.push(...exported.files);
      callbacks?.onStageComplete?.(`export:${platform}`);
    }

    job.exported = { jobId, createdAt: Date.now(), files: allExportFiles };
    job.status = "completed";

    // Stage 7: Publish — upload to OSS + notify Feishu. Opt-in via config.
    if (!config.skipPublish && config.publish) {
      callbacks?.onStageStart?.("publish");
      const publishCtx = {
        jobId,
        template: input.template,
        platforms,
      };
      try {
        const outcome = await runPublish(allExportFiles, config.publish, publishCtx);
        const urlByPath = new Map(outcome.uploaded.map((u) => [u.filePath, u.ossUrl]));
        job.exported = {
          ...job.exported,
          files: allExportFiles.map((f) => ({
            ...f,
            ossUrl: urlByPath.get(f.filePath),
          })),
        };
        if (!outcome.ok && outcome.error) {
          job.publishError = outcome.error;
          callbacks?.onError?.("publish", new Error(outcome.error));
        }
        callbacks?.onStageComplete?.("publish");
      } catch (err) {
        // Publish must never mask a successful render.
        const msg = err instanceof Error ? err.message : String(err);
        job.publishError = msg;
        callbacks?.onError?.("publish", err instanceof Error ? err : new Error(msg));
      }
    }
  } catch (err) {
    job.status = "failed";
    job.error = err instanceof Error ? err.message : String(err);
    callbacks?.onError?.("pipeline", err instanceof Error ? err : new Error(String(err)));

    // Render failed before any file existed — push failure info to Feishu.
    if (!config.skipPublish && config.publish) {
      await notifyGenerationFailure(
        config.publish,
        { jobId, template: input.template, platforms },
        job.error ?? "unknown error"
      ).catch(() => {});
    }
  }

  job.updatedAt = Date.now();
  return job;
}