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";

const SEGMENT_LIST_POLL_MS = 5_000;

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",
    ["--extractor-args", "youtube:player_client=android", "-f", "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.stderr.on("data", (chunk) => console.error(`[yt-dlp:${sessionId}]`, chunk.toString().trim()));
  ffmpeg.stderr.on("data", () => {
    /* ffmpeg logs are extremely verbose; only surface on error below */
  });

  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,
    concurrency: 3,
    // Captures run for the lifetime of the stream (can be hours) — extend the
    // default lock well beyond BullMQ's stalled-job assumptions for an MVP.
    lockDuration: 24 * 60 * 60 * 1000,
  });
}