feat: interaktif panel kontrolleri (Faz 3)

Yayın durdurma, kanal duraklat/devam (tekli+toplu), şimdi kontrol et,
manuel URL ile zorla kayıt başlatma, canlı player+log önizleme, oturum
süre sayacı, başarısız/takılı render için tekrar dene ve iptal, poll
aralığı + ham segment TTL'yi panelden canlı değiştirme, cookie bayatlık
uyarısı, Telegram bildirim aç/kapa, ve kullanım/maliyet özeti (/stats).

api-daemon'ın yeni state-değiştiren endpoint'leri (stop/force-start/
render/cancel) INTERNAL_API_TOKEN ile korunuyor — bu daemon'ın portu
Coolify'de public'e açık olduğu için korumasız bırakmak güvenlik açığı
olurdu.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

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

@@ -1,4 +1,4 @@
-import { spawn } from "node:child_process";
+import { spawn, type ChildProcess } from "node:child_process";
 import { mkdir, readFile } from "node:fs/promises";
 import path from "node:path";
 import { Worker, type Job } from "bullmq";
@@ -17,6 +17,24 @@ const SEGMENT_LIST_POLL_MS = 5_000;
  */
 export const lastCaptureDebug = new Map<string, string[]>();
 
+/**
+ * Live yt-dlp processes, keyed by sessionId — lets a panel "stop" action
+ * reach a specific in-progress capture. Only yt-dlp is tracked/killed:
+ * ffmpeg is downstream of it in the pipe, so ending yt-dlp closes its stdin
+ * with EOF and ffmpeg finishes the same way it does on a natural stream end
+ * (flushes the last partial segment, exits, existing finalize code runs
+ * unchanged).
+ */
+const activeCaptures = new Map<string, ChildProcess>();
+
+/** Returns false if the session has no tracked live process (already ended, or this daemon didn't start it — e.g. after a restart). */
+export function stopCapture(sessionId: string): boolean {
+  const ytdlp = activeCaptures.get(sessionId);
+  if (!ytdlp) return false;
+  ytdlp.kill("SIGTERM");
+  return true;
+}
+
 function appendCaptureDebug(sessionId: string, line: string) {
   const lines = lastCaptureDebug.get(sessionId) ?? [];
   lines.push(line);
@@ -95,6 +113,7 @@ async function runCapture(job: Job<StreamIngestJob>): Promise<void> {
     ],
     { stdio: ["ignore", "pipe", "pipe"] },
   );
+  activeCaptures.set(sessionId, ytdlp);
 
   const ffmpeg = spawn(
     "ffmpeg",
@@ -153,6 +172,7 @@ async function runCapture(job: Job<StreamIngestJob>): Promise<void> {
   });
 
   clearInterval(pollTimer);
+  activeCaptures.delete(sessionId);
 
   // final sweep in case segments closed between the last poll and process exit
   const raw = await readFile(segmentListPath, "utf8").catch(() => "");

apps/api-daemon/src/cleanup.ts

@@ -3,6 +3,13 @@ import { prisma } from "@streamclipper/db";
 import { env } from "./env";
 
 const CHECK_INTERVAL_MS = 60 * 60 * 1000; // hourly
+const TTL_SETTING_KEY = "raw_segment_ttl_hours";
+
+async function getTtlHours(): Promise<number> {
+  const setting = await prisma.appSetting.findUnique({ where: { key: TTL_SETTING_KEY } });
+  const parsed = setting ? Number(setting.value) : NaN;
+  return Number.isFinite(parsed) && parsed > 0 ? parsed : env.rawSegmentTtlHours;
+}
 
 /**
  * PRD's auto-purge lifecycle: processed raw recordings get deleted 24h after
@@ -11,7 +18,8 @@ const CHECK_INTERVAL_MS = 60 * 60 * 1000; // hourly
  * file mid-analysis or mid-transcription would break the pipeline for it.
  */
 async function cleanupOnce(): Promise<void> {
-  const cutoff = new Date(Date.now() - env.rawSegmentTtlHours * 60 * 60 * 1000);
+  const ttlHours = await getTtlHours();
+  const cutoff = new Date(Date.now() - ttlHours * 60 * 60 * 1000);
 
   const staleSegments = await prisma.rawSegment.findMany({
     where: {
@@ -28,12 +36,12 @@ async function cleanupOnce(): Promise<void> {
   }
 
   if (staleSegments.length > 0) {
-    console.log(`[cleanup] removed ${staleSegments.length} raw segment(s) older than ${env.rawSegmentTtlHours}h`);
+    console.log(`[cleanup] removed ${staleSegments.length} raw segment(s) older than ${ttlHours}h`);
   }
 }
 
 export function startCleanupLoop(): NodeJS.Timeout {
-  console.log(`[cleanup] checking every ${CHECK_INTERVAL_MS}ms, TTL ${env.rawSegmentTtlHours}h`);
+  console.log(`[cleanup] checking every ${CHECK_INTERVAL_MS}ms, default TTL ${env.rawSegmentTtlHours}h`);
   cleanupOnce().catch((err) => console.error("[cleanup] initial run failed:", err));
   return setInterval(() => {
     cleanupOnce().catch((err) => console.error("[cleanup] run failed:", err));

apps/api-daemon/src/env.ts

@@ -11,4 +11,5 @@ export const env = {
   proxyUrl: process.env.PROXY_URL ?? "",
   maxConcurrentCaptures: Number(process.env.MAX_CONCURRENT_CAPTURES ?? 10),
   rawSegmentTtlHours: Number(process.env.RAW_SEGMENT_TTL_HOURS ?? 24),
+  internalApiToken: process.env.INTERNAL_API_TOKEN ?? "",
 };

apps/api-daemon/src/queues.ts

@@ -34,3 +34,13 @@ export const streamIngestQueue = new Queue<StreamIngestJob>(QUEUE_NAMES.STREAM_I
 export const signalDetectionQueue = new Queue<SignalDetectionJob>(QUEUE_NAMES.SIGNAL_DETECTION, {
   connection: redisConnection,
 });
+
+export interface VideoRenderJob {
+  candidateSegmentId: string;
+}
+
+// Consumed by the Python worker (apps/worker/worker/video_render.py) — job
+// name/payload shape must match what stt_scoring.py already enqueues.
+export const videoRenderQueue = new Queue<VideoRenderJob>(QUEUE_NAMES.VIDEO_RENDER, {
+  connection: redisConnection,
+});

apps/api-daemon/src/server.ts

@@ -1,11 +1,13 @@
 import express from "express";
 import { prisma } from "@streamclipper/db";
 import { env } from "./env";
-import { lastPollErrors } from "./youtubePolling";
-import { lastCaptureDebug } from "./capture/streamIngest";
+import { lastPollErrors, checkChannel, startCaptureSession } from "./youtubePolling";
+import { lastCaptureDebug, stopCapture } from "./capture/streamIngest";
+import { videoRenderQueue } from "./queues";
 
 export function startServer() {
   const app = express();
+  app.use(express.json());
 
   app.get("/health", (_req, res) => res.json({ ok: true }));
 
@@ -37,6 +39,70 @@ export function startServer() {
     res.json({ lines: lastCaptureDebug.get(req.params.sessionId) ?? [] });
   });
 
+  /**
+   * These endpoints change state (stop a capture, start one against an
+   * arbitrary URL, re-run a render) — unlike /health and /status they must
+   * not be reachable by anyone who finds this daemon's public sslip.io URL.
+   * If INTERNAL_API_TOKEN isn't set (local dev), they stay open — matches
+   * SESSION_SECRET's existing "insecure default, fine for local" pattern.
+   */
+  function requireInternalToken(req: express.Request, res: express.Response, next: express.NextFunction) {
+    if (!env.internalApiToken) return next();
+    const header = req.header("authorization") ?? "";
+    if (header === `Bearer ${env.internalApiToken}`) return next();
+    res.status(401).json({ error: "unauthorized" });
+  }
+
+  app.post("/sessions/:sessionId/stop", requireInternalToken, (req, res) => {
+    const stopped = stopCapture(req.params.sessionId);
+    res.json({ stopped });
+  });
+
+  app.post("/channels/:channelId/check-now", requireInternalToken, async (req, res) => {
+    const channel = await prisma.channel.findUnique({ where: { id: req.params.channelId } });
+    if (!channel) return res.status(404).json({ error: "channel not found" });
+    const result = await checkChannel(channel);
+    res.json(result);
+  });
+
+  app.post("/channels/:channelId/force-start", requireInternalToken, async (req, res) => {
+    const channel = await prisma.channel.findUnique({ where: { id: req.params.channelId } });
+    if (!channel) return res.status(404).json({ error: "channel not found" });
+    const youtubeUrl = String(req.body?.url ?? "").trim();
+    if (!youtubeUrl) return res.status(400).json({ error: "url is required" });
+    const result = await startCaptureSession(channel, youtubeUrl);
+    res.json(result);
+  });
+
+  app.post("/shorts/:candidateId/render", requireInternalToken, async (req, res) => {
+    const candidateId = req.params.candidateId;
+    const candidate = await prisma.candidateSegment.findUnique({ where: { id: candidateId } });
+    if (!candidate) return res.status(404).json({ error: "candidate not found" });
+
+    await prisma.shortVideo.upsert({
+      where: { candidateId },
+      create: { candidateId, status: "PENDING" },
+      update: { status: "PENDING", errorMessage: null },
+    });
+    await videoRenderQueue.add("render-short", { candidateSegmentId: candidateId });
+    res.json({ queued: true });
+  });
+
+  app.post("/shorts/:candidateId/cancel", requireInternalToken, async (req, res) => {
+    const candidateId = req.params.candidateId;
+    const short = await prisma.shortVideo.findUnique({ where: { candidateId } });
+    if (!short) return res.status(404).json({ error: "short not found" });
+
+    // Doesn't kill the in-flight Python render process — just clears the
+    // stuck DB state so the panel can offer a retry instead of showing
+    // "RENDERING" forever if a worker died mid-job.
+    await prisma.shortVideo.update({
+      where: { candidateId },
+      data: { status: "FAILED", errorMessage: "Manuel olarak iptal edildi." },
+    });
+    res.json({ cancelled: true });
+  });
+
   app.listen(env.port, () => {
     console.log(`[server] api-daemon listening on :${env.port}`);
   });

apps/api-daemon/src/youtubePolling.ts

@@ -69,11 +69,22 @@ async function findLiveVideoId(channelId: string): Promise<string | null> {
   }
 }
 
-async function startSession(channel: { id: string; name: string }, liveVideoId: string) {
+/**
+ * Shared by the automatic poll loop and the panel's manual "force-start"
+ * action — both just need a channel + a URL to begin capturing. A no-op if
+ * that channel already has an open session (covers both callers: the poll
+ * loop re-checking a channel it's already recording, and someone hitting
+ * force-start on a channel that's already live).
+ */
+export async function startCaptureSession(
+  channel: { id: string; name: string },
+  youtubeUrl: string,
+  liveVideoId: string | null = null,
+): Promise<{ started: boolean }> {
   const existing = await prisma.streamSession.findFirst({
     where: { channelId: channel.id, endedAt: null },
   });
-  if (existing) return;
+  if (existing) return { started: false };
 
   const session = await prisma.streamSession.create({
     data: { channelId: channel.id, liveVideoId },
@@ -82,11 +93,33 @@ async function startSession(channel: { id: string; name: string }, liveVideoId:
   await streamIngestQueue.add("start-capture", {
     sessionId: session.id,
     channelDbId: channel.id,
-    youtubeUrl: `https://www.youtube.com/watch?v=${liveVideoId}`,
+    youtubeUrl,
   });
 
-  await sendTelegramMessage(`🔴 *${channel.name}* canlıya geçti, kayıt başlatılıyor.`);
+  const notifySetting = await prisma.appSetting.findUnique({ where: { key: "notify_stream_start" } });
+  if (notifySetting?.value !== "false") {
+    await sendTelegramMessage(`🔴 *${channel.name}* canlıya geçti, kayıt başlatılıyor.`);
+  }
   console.log(`[youtube-polling] started session ${session.id} for channel ${channel.name}`);
+  return { started: true };
+}
+
+/** Runs the live-check for a single channel and applies its result — the unit both `pollOnce` and the panel's manual "şimdi kontrol et" action reuse. */
+export async function checkChannel(channel: { id: string; name: string; channelId: string }): Promise<{
+  liveVideoId: string | null;
+  error: string | null;
+}> {
+  const liveVideoId = await findLiveVideoId(channel.channelId);
+  await prisma.channel.update({
+    where: { id: channel.id },
+    data: { lastCheckedAt: new Date() },
+  });
+
+  if (liveVideoId) {
+    await startCaptureSession(channel, `https://www.youtube.com/watch?v=${liveVideoId}`, liveVideoId);
+  }
+
+  return { liveVideoId, error: lastPollErrors.get(channel.channelId) ?? null };
 }
 
 export async function pollOnce(): Promise<void> {
@@ -94,25 +127,39 @@ export async function pollOnce(): Promise<void> {
 
   for (const channel of channels) {
     try {
-      const liveVideoId = await findLiveVideoId(channel.channelId);
-      await prisma.channel.update({
-        where: { id: channel.id },
-        data: { lastCheckedAt: new Date() },
-      });
-
-      if (liveVideoId) {
-        await startSession(channel, liveVideoId);
-      }
+      await checkChannel(channel);
     } catch (err) {
       console.error(`[youtube-polling] error polling channel ${channel.name}:`, err);
     }
   }
 }
 
-export function startPollingLoop(): NodeJS.Timeout {
-  console.log(`[youtube-polling] polling every ${env.pollIntervalMs}ms`);
-  pollOnce().catch((err) => console.error("[youtube-polling] initial poll failed:", err));
-  return setInterval(() => {
-    pollOnce().catch((err) => console.error("[youtube-polling] poll failed:", err));
-  }, env.pollIntervalMs);
+const POLL_INTERVAL_SETTING_KEY = "poll_interval_ms";
+
+async function getPollIntervalMs(): Promise<number> {
+  const setting = await prisma.appSetting.findUnique({ where: { key: POLL_INTERVAL_SETTING_KEY } });
+  const parsed = setting ? Number(setting.value) : NaN;
+  return Number.isFinite(parsed) && parsed >= 5_000 ? parsed : env.pollIntervalMs;
+}
+
+/**
+ * Self-rescheduling instead of setInterval so a panel-side change to the
+ * poll_interval_ms AppSetting (read fresh on every iteration, same pattern
+ * as ytdlpAntiBotArgs()'s cookies) takes effect on the very next cycle
+ * without a redeploy.
+ */
+export function startPollingLoop(): void {
+  let stopped = false;
+
+  async function tick() {
+    if (stopped) return;
+    await pollOnce().catch((err) => console.error("[youtube-polling] poll failed:", err));
+    const intervalMs = await getPollIntervalMs().catch(() => env.pollIntervalMs);
+    if (!stopped) setTimeout(tick, intervalMs);
+  }
+
+  getPollIntervalMs()
+    .then((ms) => console.log(`[youtube-polling] polling every ${ms}ms`))
+    .catch(() => {});
+  tick();
 }