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 { createLogger } from "@pipeline/shared/node";
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 log = createLogger("pipeline");
  const t0 = Date.now();
  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) {
    const cleaned = await cleanupExpiredOutput(config.output.dir, config.output.retentionDays);
    log.debug(
      `cleanup scanned=${cleaned.scanned} removed=${cleaned.removed} freed=${Math.round(cleaned.bytesFreed / 1024 / 1024)}MB`
    );
  }

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

  log.info(
    `job ${jobId} start template=${input.template} platforms=${platforms.join(",")} ` +
      `chars=${input.text.length} skipLlm=${!!config.skipLlm} skipTts=${!!config.skipTts}`
  );

  try {
    // Stage 1: Parse (shared across all platforms)
    callbacks?.onStageStart?.("parse");
    const tParse = Date.now();
    const parsed = await parseText(input.text, input.template, {
      llm: config.llm,
      skipLlm: config.skipLlm,
      source: input.source,
    });
    job.parsed = parsed;
    log.info(`parse ok scenes=${parsed.scenes?.length ?? 0} (${Date.now() - tParse}ms)`);
    callbacks?.onStageComplete?.("parse");

    // Stage 2: TTS (shared across all platforms)
    callbacks?.onStageStart?.("tts");
    const tTts = Date.now();
    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;
    log.info(
      `tts ok provider=${config.tts.provider} scenes=${tts.scenes?.length ?? 0} ` +
        `duration=${(tts.totalDurationSeconds ?? 0).toFixed(1)}s (${Date.now() - tTts}ms)`
    );
    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`);
      const tRender = Date.now();
      await renderVideo(composed, renderOutput, config.templates.entryPoint, config.assets.root);
      log.info(`render ${platform} ok -> ${renderOutput} (${Date.now() - tRender}ms)`);
      callbacks?.onStageComplete?.(`render:${platform}`);

      // Stage 6: Export
      callbacks?.onStageStart?.(`export:${platform}`);
      const exported = await exportVideo(
        renderOutput,
        composed,
        config.output.dir,
        jobId
      );
      const expFile = exported.files[0];
      log.info(
        `export ${platform} -> ${expFile?.filePath} ` +
          `(${((expFile?.fileSizeBytes ?? 0) / 1024 / 1024).toFixed(1)}MB)`
      );
      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) {
      log.info(
        `publish oss=${config.publish.oss ? "on" : "off"} feishu=${config.publish.feishu ? "on" : "off"}`
      );
      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;
          log.error(`publish failed: ${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;
        log.error(`publish error: ${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);
    log.error(`job ${jobId} failed: ${job.error}`);
    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(() => {});
    }
  }

  const elapsed = ((Date.now() - t0) / 1000).toFixed(1);
  log.info(
    `job ${jobId} ${job.status} in ${elapsed}s` +
      (job.publishError ? ` publishError=${job.publishError.slice(0, 160)}` : "")
  );
  job.updatedAt = Date.now();
  return job;
}