nextav/src/lib/transcode/orchestrator.ts

import { ChildProcessWithoutNullStreams, spawn } from 'child_process';
import crypto from 'crypto';
import fs from 'fs';
import path from 'path';
import { getDatabase } from '@/db';
import { getTranscodeConfig } from './config';
import { ConcurrencyGate, ConcurrencyLimitError } from './concurrency';
import { decideTranscode, TranscodeDecision } from './decision';
import { buildRemuxArgs, buildTranscodeArgs } from './ffmpeg-args';
import { parseStoredProbe, probeMedia, ProbeInfo } from './ffprobe';
import { HeartbeatManager } from './heartbeat';
import { SegmentJanitor } from './janitor';
import { listSegmentFiles } from './playlist';
import { getDefaultProfileName, TranscodeProfileName } from './profiles';
import { TranscodeThrottler } from './throttler';

export type TranscodeJobStatus = 'pending' | 'encoding' | 'ready' | 'paused' | 'failed' | 'killed';

export interface TranscodeJob {
  id: string;
  mediaId: number;
  mediaPath: string;
  profile: TranscodeProfileName;
  outDir: string;
  startTime: number;
  status: TranscodeJobStatus;
  process?: ChildProcessWithoutNullStreams;
  createdAt: number;
  lastPingAt: number;
  lastClientPosition: number;
  knownDuration: number;
  error?: string;
  stderrRing: string[];
  decision: TranscodeDecision;
  throttler?: TranscodeThrottler;
}

interface MediaRow {
  id: number;
  path: string;
  codec_info?: string;
}

export class TranscodeDisabledError extends Error {
  constructor() {
    super('Live transcoding is disabled');
    this.name = 'TranscodeDisabledError';
  }
}

class TranscodeOrchestrator {
  private readonly jobs = new Map<string, TranscodeJob>();
  private readonly gate = new ConcurrencyGate(getTranscodeConfig().maxConcurrent);
  private readonly heartbeat = new HeartbeatManager();
  private readonly janitor = new SegmentJanitor();
  private recovered = false;

  constructor() {
    const config = getTranscodeConfig();
    fs.mkdirSync(config.tmpDir, { recursive: true });
  }

  recover(): void {
    if (this.recovered) return;
    this.recovered = true;
    const db = getDatabase();
    db.prepare("UPDATE transcode_jobs SET status = 'killed', error = COALESCE(error, 'server restarted') WHERE status IN ('pending', 'encoding', 'paused')").run();
  }

  async start(mediaId: number, requestedProfile: string | undefined, startTime: number): Promise<TranscodeJob> {
    const config = getTranscodeConfig();
    if (!config.enabled) {
      throw new TranscodeDisabledError();
    }

    this.recover();
    this.janitor.enforceBudget();

    const profile = getDefaultProfileName(requestedProfile || config.defaultProfile);
    const existing = this.findReusableJob(mediaId, profile, startTime);
    if (existing) return existing;

    const media = this.getMedia(mediaId);
    const probe = await this.getProbe(media);
    const decision = decideTranscode(media.path, probe, profile);

    if (decision.kind === 'direct') {
      throw new Error('Media is directly playable and does not need live transcoding');
    }

    this.gate.acquire();

    const jobId = crypto.randomUUID();
    const outDir = path.join(config.tmpDir, String(mediaId), `${decision.profile}-${jobId}`);
    fs.mkdirSync(outDir, { recursive: true });

    const job: TranscodeJob = {
      id: jobId,
      mediaId,
      mediaPath: media.path,
      profile: decision.profile,
      outDir,
      startTime,
      status: 'pending',
      createdAt: Date.now(),
      lastPingAt: Date.now(),
      lastClientPosition: startTime,
      knownDuration: probe.duration,
      stderrRing: [],
      decision,
    };

    this.jobs.set(jobId, job);
    this.persistJob(job);

    try {
      await this.spawnJob(job);
      return job;
    } catch (error) {
      await this.kill(jobId, error instanceof Error ? error.message : 'startup failed', 0);
      throw error;
    }
  }

  get(jobId: string): TranscodeJob | undefined {
    return this.jobs.get(jobId);
  }

  status() {
    return {
      enabled: getTranscodeConfig().enabled,
      concurrency: this.gate.snapshot(),
      jobs: Array.from(this.jobs.values()).map(job => ({
        id: job.id,
        mediaId: job.mediaId,
        profile: job.profile,
        status: job.status,
        pid: job.process?.pid,
        startTime: job.startTime,
        lastPingAt: job.lastPingAt,
        lastClientPosition: job.lastClientPosition,
        segmentCount: listSegmentFiles(job.outDir).length,
        decision: job.decision.kind,
        error: job.error,
      })),
    };
  }

  ping(jobId: string, position: number, isPaused: boolean): boolean {
    const job = this.jobs.get(jobId);
    if (!job) return false;
    const config = getTranscodeConfig();
    job.lastPingAt = Date.now();
    job.lastClientPosition = position;
    job.throttler?.updateClient(position, isPaused);
    this.heartbeat.arm(jobId, config.heartbeatTimeoutMs, () => {
      this.kill(jobId, 'heartbeat timeout').catch(error => console.error('[Transcode] heartbeat kill failed', error));
    });
    getDatabase().prepare('UPDATE transcode_jobs SET last_ping_at = CURRENT_TIMESTAMP WHERE job_id = ?').run(jobId);
    return true;
  }

  async seek(jobId: string, position: number): Promise<{ reused: true } | { reused: false; job: TranscodeJob }> {
    const job = this.jobs.get(jobId);
    if (!job) throw new Error('Transcode job not found');

    const encodedSeconds = listSegmentFiles(job.outDir).length * getTranscodeConfig().segmentDuration;
    if (position >= job.startTime && position <= job.startTime + Math.max(0, encodedSeconds - 8)) {
      return { reused: true };
    }

    await this.kill(jobId, 'seek restart', getTranscodeConfig().cacheGraceMs);
    const newJob = await this.start(job.mediaId, job.profile, position);
    return { reused: false, job: newJob };
  }

  async kill(jobId: string, reason = 'client closed', cleanupDelay = getTranscodeConfig().cacheGraceMs): Promise<void> {
    const job = this.jobs.get(jobId);
    if (!job) return;

    job.status = reason === 'completed' ? 'ready' : 'killed';
    job.error = reason === 'completed' ? undefined : reason;
    this.heartbeat.clear(jobId);
    job.throttler?.stop();

    const proc = job.process;
    if (proc && proc.exitCode === null) {
      try {
        proc.stdin.write('q\n');
      } catch {
        // Process may already be gone.
      }
      const exited = await waitForExit(proc, 3000);
      if (!exited && proc.exitCode === null) {
        proc.kill('SIGTERM');
        await waitForExit(proc, 2000);
      }
      if (proc.exitCode === null) {
        proc.kill('SIGKILL');
      }
    }

    getDatabase().prepare('UPDATE transcode_jobs SET status = ?, error = ? WHERE job_id = ?').run(job.status, job.error || null, jobId);
    this.gate.release();
    this.jobs.delete(jobId);
    this.janitor.scheduleCleanup(job.outDir, cleanupDelay);
  }

  private findReusableJob(mediaId: number, profile: TranscodeProfileName, startTime: number): TranscodeJob | undefined {
    const now = Date.now();
    return Array.from(this.jobs.values()).find(job =>
      job.mediaId === mediaId &&
      job.profile === profile &&
      Math.abs(job.startTime - startTime) < 1 &&
      (job.status === 'encoding' || job.status === 'ready') &&
      now - job.lastPingAt < 30_000
    );
  }

  private getMedia(mediaId: number): MediaRow {
    const media = getDatabase().prepare("SELECT id, path, codec_info FROM media WHERE id = ? AND type = 'video'").get(mediaId) as MediaRow | undefined;
    if (!media) throw new Error('Video not found');
    if (!fs.existsSync(media.path)) throw new Error('Video file not found on disk');
    return media;
  }

  private async getProbe(media: MediaRow): Promise<ProbeInfo> {
    const stored = parseStoredProbe(media.codec_info);
    if (stored) return stored;
    const probed = await probeMedia(media.path);
    getDatabase().prepare('UPDATE media SET codec_info = ? WHERE id = ?').run(JSON.stringify(probed), media.id);
    return probed;
  }

  private persistJob(job: TranscodeJob): void {
    getDatabase().prepare(`
      INSERT INTO transcode_jobs (job_id, media_id, profile, out_dir, start_ts, status, pid, last_ping_at, error)
      VALUES (?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP, ?)
    `).run(job.id, job.mediaId, job.profile, job.outDir, job.startTime, job.status, null, null);
  }

  private async spawnJob(job: TranscodeJob): Promise<void> {
    const config = getTranscodeConfig();
    const args = job.decision.kind === 'hls-remux'
      ? buildRemuxArgs(job.mediaPath, job.outDir, job.startTime)
      : buildTranscodeArgs(job.mediaPath, job.outDir, job.profile as Exclude<TranscodeProfileName, 'remux-copy'>, job.startTime);

    const proc = spawn(config.ffmpegPath, args, { stdio: ['pipe', 'pipe', 'pipe'] });
    job.process = proc;
    job.status = 'encoding';
    job.throttler = new TranscodeThrottler(proc);
    job.throttler.start();

    getDatabase().prepare("UPDATE transcode_jobs SET status = 'encoding', pid = ? WHERE job_id = ?").run(proc.pid || null, job.id);
    this.heartbeat.arm(job.id, config.heartbeatTimeoutMs, () => {
      this.kill(job.id, 'heartbeat timeout').catch(error => console.error('[Transcode] heartbeat kill failed', error));
    });

    proc.stderr.setEncoding('utf8');
    proc.stderr.on('data', (chunk: string) => {
      const lines = chunk.split(/\r?\n/).filter(Boolean);
      for (const line of lines) {
        job.stderrRing.push(line);
        if (job.stderrRing.length > 50) job.stderrRing.shift();
        job.throttler?.updateProgress(line);
      }
    });

    proc.on('exit', code => {
      job.throttler?.stop();
      if (code === 0 && job.status !== 'killed') {
        job.status = 'ready';
        getDatabase().prepare("UPDATE transcode_jobs SET status = 'ready' WHERE job_id = ?").run(job.id);
      } else if (job.status !== 'killed') {
        job.status = 'failed';
        job.error = job.stderrRing.slice(-10).join('\n') || `ffmpeg exited with code ${code}`;
        getDatabase().prepare("UPDATE transcode_jobs SET status = 'failed', error = ? WHERE job_id = ?").run(job.error, job.id);
        this.gate.release();
      }
    });

    await waitForFirstSegment(job.outDir, config.startupTimeoutMs);
  }
}

function waitForExit(proc: ChildProcessWithoutNullStreams, timeoutMs: number): Promise<boolean> {
  return new Promise(resolve => {
    if (proc.exitCode !== null) {
      resolve(true);
      return;
    }

    const timeout = setTimeout(() => {
      cleanup();
      resolve(false);
    }, timeoutMs);
    timeout.unref();

    const onExit = () => {
      cleanup();
      resolve(true);
    };

    const cleanup = () => {
      clearTimeout(timeout);
      proc.off('exit', onExit);
    };

    proc.once('exit', onExit);
  });
}

function waitForFirstSegment(outDir: string, timeoutMs: number): Promise<void> {
  return new Promise((resolve, reject) => {
    if (listSegmentFiles(outDir).length > 0) {
      resolve();
      return;
    }

    const timeout = setTimeout(() => {
      watcher.close();
      reject(new Error('TRANSCODE_STARTUP_TIMEOUT'));
    }, timeoutMs);
    timeout.unref();

    const watcher = fs.watch(outDir, () => {
      if (listSegmentFiles(outDir).length > 0) {
        clearTimeout(timeout);
        watcher.close();
        resolve();
      }
    });
  });
}

const globalForTranscode = globalThis as typeof globalThis & {
  nextavTranscodeOrchestrator?: TranscodeOrchestrator;
};

export const transcodeOrchestrator =
  globalForTranscode.nextavTranscodeOrchestrator ?? new TranscodeOrchestrator();

globalForTranscode.nextavTranscodeOrchestrator = transcodeOrchestrator;

export { ConcurrencyLimitError };