abdulive/apps/ingest/src/render.ts

// Render-Warteschlange: aus der Rohdatei wird ein 9:16-Video.
//
// Eine Aufgabe nach der anderen. Waehrend gesendet wird, ist Encodieren
// standardmaessig AUS (`renderWhileLive: false`) - die Kandidatenliste wird
// erst nach dem Stream gebraucht, und der Stream braucht die CPU vorher.
// Siehe docs/07-mvp-plan.md, Risikotabelle.

import fs from 'node:fs/promises';
import path from 'node:path';
import { logger } from '#core/logger.ts';
import type { LayoutProfile } from '#core/config.ts';
import type { Repos } from '#db/repos.ts';
import { FfmpegError, probe, run } from '#media/ffmpeg.ts';
import { thumbnailArgs, verticalArgs } from '#media/vertical.ts';

const log = logger('render');

export type RenderConfig = {
  profile: LayoutProfile;
  tailSeconds: number;
  outputTrack: number;
  encoder: string;
  preset: string;
  crf: number;
  niceness: number;
  renderWhileLive: boolean;
};

export type RenderDeps = {
  cfg: RenderConfig;
  repo: Repos;
  workDir: string;
  isLive: () => boolean;
};

type Job = { clipId: string; rawPath: string; outDir: string };

export function createRenderer(deps: RenderDeps) {
  const queue: Job[] = [];
  let running = false;
  let current: string | null = null;

  function enqueue(job: Job): void {
    if (queue.some((j) => j.clipId === job.clipId) || current === job.clipId) return;
    queue.push(job);
    log.info(`In der Warteschlange: ${job.clipId} (${queue.length} offen)`);
    void pump();
  }

  async function pump(): Promise<void> {
    if (running) return;
    if (!deps.cfg.renderWhileLive && deps.isLive()) {
      log.debug('Stream laeuft - Rendern verschoben');
      return;
    }
    running = true;
    try {
      let job: Job | undefined;
      while ((job = queue.shift())) {
        if (!deps.cfg.renderWhileLive && deps.isLive()) {
          queue.unshift(job);            // zurueck in die Schlange
          break;
        }
        current = job.clipId;
        try {
          await render(job);
        } catch (err) {
          deps.repo.updateClip(job.clipId, { status: 'failed' });
          if (err instanceof FfmpegError) {
            log.error(`Rendern fehlgeschlagen (${job.clipId}): ${err.message}`);
            for (const line of err.tail.split('\n').slice(-6)) log.error(`  ${line}`);
          } else {
            log.error(`Rendern fehlgeschlagen (${job.clipId}): ${(err as Error).message}`);
          }
        } finally {
          current = null;
        }
      }
    } finally {
      running = false;
    }
  }

  async function render(job: Job): Promise<void> {
    const started = Date.now();
    const info = await probe(job.rawPath);
    log.info(`Rendere ${job.clipId}: ${info.width}x${info.height}, ` +
      `${info.durationSeconds.toFixed(1)} s, ${info.audio.length} Tonspur(en)`);

    if (info.width !== 1920 || info.height !== 1080) {
      log.warn(`Quelle ist ${info.width}x${info.height}, das Layout ist fuer 1920x1080 ` +
        'eingemessen - der Ausschnitt sitzt vermutlich falsch.');
    }

    // Faellt die konfigurierte Tonspur weg, nimm die erste vorhandene
    const track = deps.cfg.outputTrack < info.audio.length ? deps.cfg.outputTrack : 0;
    if (track !== deps.cfg.outputTrack) {
      log.warn(`Tonspur ${deps.cfg.outputTrack} gibt es nicht, nehme Spur ${track}`);
    }

    await fs.mkdir(job.outDir, { recursive: true });
    const vertical = path.join(job.outDir, 'vertical.mp4');
    const thumb = path.join(job.outDir, 'thumb.jpg');

    let lastLogged = 0;
    await run('ffmpeg', verticalArgs({
      input: job.rawPath,
      output: vertical,
      profile: deps.cfg.profile,
      tailSeconds: Math.min(deps.cfg.tailSeconds, Math.max(1, Math.floor(info.durationSeconds))),
      audioTrack: track,
      encoder: deps.cfg.encoder,
      preset: deps.cfg.preset,
      crf: deps.cfg.crf,
    }), {
      niceness: deps.cfg.niceness,
      timeoutMs: 20 * 60_000,
      onProgress: (s) => {
        if (s - lastLogged >= 10) { lastLogged = s; log.debug(`  ${s.toFixed(0)} s verarbeitet`); }
      },
    });

    await run('ffmpeg', thumbnailArgs(vertical, thumb), { niceness: deps.cfg.niceness });

    const out = await probe(vertical);
    deps.repo.updateClip(job.clipId, {
      vertical_path: vertical,
      duration_s: out.durationSeconds,
      status: 'rendered',
    });
    deps.repo.event('clip.rendered', {
      clipId: job.clipId, path: vertical, seconds: out.durationSeconds,
    });
    log.info(`Fertig: ${vertical} (${out.durationSeconds.toFixed(1)} s, ` +
      `${((Date.now() - started) / 1000).toFixed(0)} s Rechenzeit)`);
  }

  return {
    enqueue,
    /** Nach Streamende aufrufen: arbeitet ab, was liegen geblieben ist. */
    flush: () => { void pump(); },
    pending: () => queue.length + (current ? 1 : 0),
    currentJob: () => current,
  };
}