watchparty/server/src/services/TranscodeService.ts

'use strict';

import { spawn, type ChildProcess } from 'child_process';
import fs from 'fs';
import os from 'os';
import path from 'path';
import type { Request, Response } from 'express';
import { EMBY_SERVER_URL } from '../config';
import { Logger } from '../Logger';
import { CatalogService } from './CatalogService';
import type { EmbySession, Party } from '../types';

/** The one shape TranscodeService needs out of Emby's own MediaStreams
 *  array - see absoluteToRelativeIndex below. */
export interface MediaStreamInfo {
  Type: string;
  Index: number;
}

/**
 * Converts Emby's own *absolute* MediaStreams[].Index - a single flat
 * index across video+audio+subtitle streams together (the "stream 6/7/8"
 * this app's own subtitle picker already surfaces, see
 * CatalogService.getPlaybackOptions) - into ffmpeg's *type-relative*
 * stream specifier: `0:s:N`/`0:a:N` count only among streams of that same
 * type. ffmpeg has no notion of Emby's absolute numbering at all, so
 * every burn-in command has to make this conversion first. Pure array
 * math - no ffmpeg or Emby involved, see transcodeService.test.ts.
 */
export function absoluteToRelativeIndex(streams: MediaStreamInfo[], type: string, absoluteIndex: number): number {
  let relative = 0;
  for (const s of streams) {
    if (s.Index === absoluteIndex) return relative;
    if (s.Type === type) relative += 1;
  }
  return 0; // not found - callers only ever pass an index getPlaybackOptions itself produced
}

/**
 * Emby's own "direct/static" file endpoint (untranscoded, original
 * bytes, Range-seekable) - the standard Emby/Jellyfin shape, distinct
 * from `/Videos/{id}/master.m3u8` (Emby's *own* transcode, what
 * CatalogService.proxyStream proxies for every other party). Feeding
 * this to our own ffmpeg is what lets this app burn a PGS subtitle in
 * itself instead of asking Emby's engine to (see docs/architecture.md's
 * "Self-hosted ffmpeg" section for why that kept failing).
 *
 * TODO(verify): unconfirmed against a real Emby server - there isn't one
 * reachable from this dev environment. If the exact shape is wrong, it's
 * isolated to this one function.
 */
function buildSourceUrl(itemId: string, mediaSourceId: string, container: string): string {
  const params = new URLSearchParams({ Static: 'true', MediaSourceId: mediaSourceId });
  return `${EMBY_SERVER_URL}/Videos/${itemId}/stream.${container}?${params}`;
}

const SEGMENT_NAME_RE = /^seg\d{5}\.ts$/;

interface TranscodeSession {
  partyId: string;
  audioIndex: string;
  subtitleIndex: string;
  process: ChildProcess;
  outputDir: string;
  lastAccessUtc: number;
  /** Set once the process exits before ever producing a first segment -
   *  the tail of its stderr, so a route waiting on this session can fail
   *  the request with a real reason instead of hanging until its own
   *  timeout. Left null for a clean exit after having streamed fine (the
   *  party just moved on - not a failure worth surfacing). */
  startupFailure: string | null;
}

/**
 * Runs ffmpeg ourselves to burn a PGS/VobSub/DVB (image, not text)
 * subtitle into the video - the one gap the client-side `<track>`
 * redesign can't close (there's no text to extract). See
 * docs/architecture.md's "Self-hosted ffmpeg" section for the full
 * design, and CatalogService.proxyStream's own comment for why Emby's
 * own transcode was never asked to do this in the first place.
 *
 * One session per PARTY, not per viewer: every member of a video party
 * shares the exact same burn-in stream, the same way everyone already
 * shares one playback position driven by the host (see
 * docs/architecture.md's "host drives, everyone follows") - so this is
 * keyed by partyId, and picking a burn-in subtitle is a host-only,
 * party-wide choice (SocketGateway's "setBurnInSubtitle" handler), not a
 * per-viewer local one the way a text subtitle pick still is.
 */
/** Simplified down to exactly the arguments this service ever passes -
 *  the real `spawn` (more optional params, incl. options) is still a
 *  valid default value for this type. */
type SpawnFn = (command: string, args: string[]) => ChildProcess;

class TranscodeServiceClass {
  private sessions = new Map<string, TranscodeSession>();
  private spawnFn: SpawnFn = spawn;

  /** Test-only seam (see transcodeService.test.ts) so a test never
   *  actually invokes a real ffmpeg binary - never called in production.
   *  Pass the real `spawn` back (or omit the argument) to restore it. */
  setSpawnForTesting(fn: SpawnFn = spawn): void {
    this.spawnFn = fn;
  }

  /**
   * Returns the party's current burn-in session, starting one if none
   * exists yet, or restarting it if the party's now asking for a
   * different audio/subtitle combination than whatever was already
   * running. Every call refreshes `lastAccessUtc` - this is also how a
   * still-being-watched session avoids the idle reaper.
   */
  async getOrStartSession(party: Party, session: EmbySession, audioIndex: string, subtitleIndex: string): Promise<TranscodeSession> {
    const existing = this.sessions.get(party.id);
    if (existing && existing.audioIndex === audioIndex && existing.subtitleIndex === subtitleIndex) {
      existing.lastAccessUtc = Date.now();
      return existing;
    }
    if (existing) this.endSession(party.id);

    const source = await CatalogService.getMediaSource(session, party.itemId);
    const audioRelative = audioIndex ? absoluteToRelativeIndex(source.streams, 'Audio', Number(audioIndex)) : 0;
    const subtitleRelative = absoluteToRelativeIndex(source.streams, 'Subtitle', Number(subtitleIndex));
    const sourceUrl = buildSourceUrl(party.itemId, source.id, source.container);

    const outputDir = path.join(os.tmpdir(), 'watchparty-transcode', party.id);
    fs.rmSync(outputDir, { recursive: true, force: true });
    fs.mkdirSync(outputDir, { recursive: true });

    const args = [
      '-headers', `X-Emby-Token: ${session.embyToken}\r\n`,
      '-i', sourceUrl,
      '-filter_complex', `[0:v][0:s:${subtitleRelative}]overlay[v]`,
      '-map', '[v]',
      '-map', `0:a:${audioRelative}`,
      '-c:v', 'libx264', '-preset', 'veryfast', '-crf', '20',
      '-c:a', 'aac', '-b:a', '192k',
      // A known filter_complex+HLS gotcha, not something confirmed to
      // happen against this app's own files - included defensively.
      '-max_muxing_queue_size', '1024',
      '-f', 'hls', '-hls_time', '6',
      '-hls_flags', 'independent_segments+append_list',
      '-hls_segment_filename', path.join(outputDir, 'seg%05d.ts'),
      '-y', path.join(outputDir, 'master.m3u8')
    ];

    const proc = this.spawnFn('ffmpeg', args);
    const newSession: TranscodeSession = {
      partyId: party.id,
      audioIndex,
      subtitleIndex,
      process: proc,
      outputDir,
      lastAccessUtc: Date.now(),
      startupFailure: null
    };

    const stderrTail: string[] = [];
    proc.stderr?.on('data', (chunk: Buffer) => {
      stderrTail.push(chunk.toString());
      if (stderrTail.length > 50) stderrTail.shift();
    });
    proc.on('exit', (code) => {
      Logger.transcode(`partyId=${party.id} ffmpeg exited code=${code}`);
      if (code !== 0 && !this.hasProducedSegment(newSession)) {
        newSession.startupFailure = stderrTail.join('').slice(-2000) || `ffmpeg exited with code ${code}`;
      }
    });
    proc.on('error', (err) => {
      // e.g. ffmpeg isn't on PATH at all - spawn itself failing, not the
      // process exiting - see the Dockerfile's `apk add ffmpeg`.
      Logger.transcode(`partyId=${party.id} failed to start ffmpeg: ${err.message}`);
      newSession.startupFailure = err.message;
    });

    Logger.transcode(
      `partyId=${party.id} started itemId=${party.itemId} ` +
      `audioIndex=${audioIndex || '(default)'} subtitleIndex=${subtitleIndex}`
    );
    this.sessions.set(party.id, newSession);
    return newSession;
  }

  /** Kills the process (if any) and removes its temp output directory -
   *  called on a fresh pick replacing this one, the party ending
   *  (PartyService.leave's destroy path / removeIdleParties), or the
   *  idle reaper below. A harmless no-op if this party never had one. */
  endSession(partyId: string): void {
    const s = this.sessions.get(partyId);
    if (!s) return;
    this.sessions.delete(partyId);
    s.process.kill('SIGKILL');
    fs.rm(s.outputDir, { recursive: true, force: true }, () => {});
    Logger.transcode(`partyId=${partyId} ended`);
  }

  /** Backstop for parties that don't end cleanly (a crash, every
   *  member's tab just closing) - mirrors PartyService.removeIdleParties,
   *  same wiring in index.ts's own setInterval. */
  reapIdleSessions(idleTimeoutMs: number): void {
    const cutoff = Date.now() - idleTimeoutMs;
    for (const [partyId, s] of this.sessions) {
      if (s.lastAccessUtc < cutoff) {
        Logger.transcode(`partyId=${partyId} reaped (idle)`);
        this.endSession(partyId);
      }
    }
  }

  /**
   * GET /api/catalog/burn/:partyId/master.m3u8 - the caller (index.ts)
   * has already checked the requester is a member of `party` and that
   * `party.burnInSubtitleIndex` is actually set. Waits briefly for the
   * first segment to exist (ffmpeg writes the playlist file itself
   * almost immediately, but with zero segments listed until the first
   * one finishes - serving it before then would just hand hls.js an
   * empty manifest) before serving the current playlist contents.
   */
  async serveMasterPlaylist(req: Request, res: Response, party: Party, embySession: EmbySession): Promise<void> {
    let s: TranscodeSession;
    try {
      s = await this.getOrStartSession(party, embySession, party.burnInAudioIndex || '', party.burnInSubtitleIndex!);
    } catch (err) {
      Logger.transcode(`partyId=${party.id} failed to negotiate source: ${(err as Error).message}`);
      res.status(502).json({ error: "Impossible de préparer le flux incrusté." });
      return;
    }

    const ok = await this.waitUntil(() => this.hasProducedSegment(s) || !!s.startupFailure, 20000);
    if (s.startupFailure) {
      res.status(502).json({ error: 'ffmpeg a échoué au démarrage du flux incrusté.' });
      return;
    }
    if (!ok) {
      res.status(504).json({ error: 'Le flux incrusté met trop de temps à démarrer.' });
      return;
    }

    const playlistPath = path.join(s.outputDir, 'master.m3u8');
    res.setHeader('Content-Type', 'application/vnd.apple.mpegurl');
    fs.createReadStream(playlistPath).pipe(res);
  }

  /** GET /api/catalog/burn/:partyId/:segment - `segment` is validated
   *  against the exact filename shape ffmpeg itself produces before ever
   *  touching the filesystem with it (never trust a URL param as a path
   *  component - see proxyRawStreamAsset's own equivalent guard). */
  async serveSegment(req: Request, res: Response, party: Party, segment: string): Promise<void> {
    if (!SEGMENT_NAME_RE.test(segment)) {
      res.status(400).end();
      return;
    }
    const s = this.sessions.get(party.id);
    if (!s) {
      res.status(404).end();
      return;
    }
    s.lastAccessUtc = Date.now();

    const segmentPath = path.join(s.outputDir, segment);
    const ok = await this.waitUntil(() => fs.existsSync(segmentPath) || !!s.startupFailure, 15000);
    if (!ok || !fs.existsSync(segmentPath)) {
      res.status(404).end();
      return;
    }
    res.setHeader('Content-Type', 'video/mp2t');
    fs.createReadStream(segmentPath).pipe(res);
  }

  /** Whether ffmpeg has written at least one real segment yet - checked
   *  by reading the output directory rather than tracked via an event,
   *  since ffmpeg's own HLS muxer gives no signal for "first segment
   *  flushed" beyond the file simply existing on disk. */
  private hasProducedSegment(s: TranscodeSession): boolean {
    try {
      return fs.readdirSync(s.outputDir).some((f) => SEGMENT_NAME_RE.test(f));
    } catch {
      return false;
    }
  }

  private waitUntil(predicate: () => boolean, timeoutMs: number): Promise<boolean> {
    if (predicate()) return Promise.resolve(true);
    return new Promise((resolve) => {
      const deadline = Date.now() + timeoutMs;
      const interval = setInterval(() => {
        if (predicate()) {
          clearInterval(interval);
          resolve(true);
        } else if (Date.now() >= deadline) {
          clearInterval(interval);
          resolve(false);
        }
      }, 250);
    });
  }
}

export const TranscodeService = new TranscodeServiceClass();