screenclipper/apps/api-daemon/src/capture/streamIngest.ts

import { spawn } from "node:child_process";
import { mkdir, readFile } from "node:fs/promises";
import path from "node:path";
import { Worker, type Job } from "bullmq";
import { prisma } from "@streamclipper/db";
import { env } from "../env";
import { QUEUE_NAMES, signalDetectionQueue, type StreamIngestJob } from "../queues";
import { redisConnection } from "../redis";
import { ytdlpAntiBotArgs } from "../ytdlpCookies";

const SEGMENT_LIST_POLL_MS = 5_000;

/**
 * Coolify's /logs endpoint can't reliably isolate one docker-compose
 * service's stdout, so the last capture attempt's yt-dlp/ffmpeg stderr tail
 * is kept here and surfaced via GET /status for debugging.
 */
export const lastCaptureDebug = new Map<string, string[]>();

function appendCaptureDebug(sessionId: string, line: string) {
  const lines = lastCaptureDebug.get(sessionId) ?? [];
  lines.push(line);
  if (lines.length > 40) lines.shift();
  lastCaptureDebug.set(sessionId, lines);
}

interface SegmentListEntry {
  fileName: string;
  startSec: number;
  endSec: number;
}

function parseSegmentListLines(raw: string): SegmentListEntry[] {
  return raw
    .split("\n")
    .map((line) => line.trim())
    .filter(Boolean)
    .map((line) => {
      const [fileName, startStr, endStr] = line.split(",");
      return { fileName, startSec: Number(startStr), endSec: Number(endStr) };
    });
}

async function processCompletedSegments(
  entries: SegmentListEntry[],
  fromIndex: number,
  outDir: string,
  sessionId: string,
): Promise<number> {
  let processed = fromIndex;

  for (let i = fromIndex; i < entries.length; i++) {
    const entry = entries[i];
    const filePath = path.join(outDir, entry.fileName);
    const duration = Math.round(entry.endSec - entry.startSec);

    const segment = await prisma.rawSegment.create({
      data: {
        sessionId,
        filePath,
        duration,
        startedAt: new Date(Date.now() - duration * 1000),
        status: "PENDING",
      },
    });

    await signalDetectionQueue.add("analyze-segment", {
      rawSegmentId: segment.id,
      filePath,
      sessionId,
    });

    console.log(`[stream-ingest] segment ready: ${entry.fileName} -> raw_segment ${segment.id}`);
    processed = i + 1;
  }

  return processed;
}

async function runCapture(job: Job<StreamIngestJob>): Promise<void> {
  const { sessionId, youtubeUrl } = job.data;
  const outDir = path.join(env.sharedMediaRoot, "raw", sessionId);
  await mkdir(outDir, { recursive: true });

  const segmentListPath = path.join(outDir, "segments.csv");

  console.log(`[stream-ingest] starting capture for session ${sessionId}: ${youtubeUrl}`);

  const ytdlp = spawn(
    "yt-dlp",
    [
      "--js-runtimes", "node",
      ...ytdlpAntiBotArgs(),
      "-f", "bestvideo+bestaudio/best", "-o", "-", youtubeUrl,
    ],
    { stdio: ["ignore", "pipe", "pipe"] },
  );

  const ffmpeg = spawn(
    "ffmpeg",
    [
      "-i", "pipe:0",
      "-c", "copy",
      "-f", "segment",
      "-segment_time", String(env.segmentTimeSec),
      "-reset_timestamps", "1",
      "-segment_list", segmentListPath,
      "-segment_list_type", "csv",
      path.join(outDir, "segment_%03d.mp4"),
    ],
    { stdio: ["pipe", "pipe", "pipe"] },
  );

  ytdlp.stdout.pipe(ffmpeg.stdin);
  ytdlp.on("error", (err) => appendCaptureDebug(sessionId, `[yt-dlp spawn error] ${err.message}`));
  ytdlp.stderr.on("data", (chunk) => {
    const text = chunk.toString().trim();
    console.error(`[yt-dlp:${sessionId}]`, text);
    appendCaptureDebug(sessionId, `[yt-dlp] ${text}`);
  });
  ffmpeg.on("error", (err) => appendCaptureDebug(sessionId, `[ffmpeg spawn error] ${err.message}`));
  ffmpeg.stderr.on("data", (chunk) => {
    /* ffmpeg logs are extremely verbose; keep only the tail for debugging */
    appendCaptureDebug(sessionId, `[ffmpeg] ${chunk.toString().trim().split("\n").pop()}`);
  });

  let lastProcessedIndex = 0;
  const pollTimer = setInterval(async () => {
    try {
      const raw = await readFile(segmentListPath, "utf8").catch(() => "");
      if (!raw) return;
      const entries = parseSegmentListLines(raw);
      if (entries.length > lastProcessedIndex) {
        lastProcessedIndex = await processCompletedSegments(entries, lastProcessedIndex, outDir, sessionId);
        await prisma.streamSession.update({
          where: { id: sessionId },
          data: { totalSegments: lastProcessedIndex },
        });
      }
    } catch (err) {
      console.error(`[stream-ingest] segment list poll error for ${sessionId}:`, err);
    }
  }, SEGMENT_LIST_POLL_MS);

  await new Promise<void>((resolve) => {
    ffmpeg.on("exit", (code) => {
      console.log(`[stream-ingest] ffmpeg exited (${code}) for session ${sessionId}`);
      resolve();
    });
    ytdlp.on("exit", (code) => {
      if (code !== 0) console.error(`[stream-ingest] yt-dlp exited with code ${code} for session ${sessionId}`);
    });
  });

  clearInterval(pollTimer);

  // final sweep in case segments closed between the last poll and process exit
  const raw = await readFile(segmentListPath, "utf8").catch(() => "");
  const entries = parseSegmentListLines(raw);
  if (entries.length > lastProcessedIndex) {
    lastProcessedIndex = await processCompletedSegments(entries, lastProcessedIndex, outDir, sessionId);
  }

  await prisma.streamSession.update({
    where: { id: sessionId },
    data: { endedAt: new Date(), totalSegments: lastProcessedIndex },
  });

  console.log(`[stream-ingest] capture finished for session ${sessionId}, ${lastProcessedIndex} segments`);
}

export function startStreamIngestWorker(): Worker<StreamIngestJob> {
  return new Worker<StreamIngestJob>(QUEUE_NAMES.STREAM_INGEST, runCapture, {
    connection: redisConnection,
    // Each capture job runs for the entire duration of a live stream
    // (potentially hours) before its BullMQ job resolves. If concurrency is
    // lower than the number of channels going live at once, the extra
    // channels queue behind already-running captures and may not start
    // recording until one of those finishes — i.e. effectively never, for
    // streams still live. This must cover the realistic simultaneous-channel
    // count; the real ceiling is the server's CPU/bandwidth for running that
    // many yt-dlp+ffmpeg pairs at once, not this number.
    concurrency: env.maxConcurrentCaptures,
    // BullMQ renews this lock automatically while a worker is alive, so a
    // low value doesn't cap how long a capture can run — it only controls
    // how fast a *dead* worker's job is detected as stalled and recovered.
    // (An earlier 24h value meant a killed process's job wouldn't be
    // recovered for up to 24h; closeOrphanedSessions() in index.ts is what
    // actually unblocks that channel's StreamSession in the meantime.)
    lockDuration: 10 * 60 * 1000,
  });
}