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";

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

  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,
    });
    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);
      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";
  } 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)));
  }

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