diff --git a/.env.example b/.env.example index cf958ec..3ab47a9 100644 --- a/.env.example +++ b/.env.example @@ -55,3 +55,9 @@ RAW_SEGMENT_TTL_HOURS=24 # işinden çok daha ağır — düşük tutulur ki aynı anda birden fazla render # worker container'ını boğmasın. VIDEO_RENDER_CONCURRENCY=2 + +# api-daemon'ın durdur/zorla-başlat/render-tetikle gibi durum-değiştiren +# endpoint'lerini korur (bu daemon'ın portu Coolify'de public'e açık). +# frontend ve api-daemon aynı değeri kullanmalı. Generate: openssl rand -hex 24 +# Boş bırakılırsa (sadece local dev için) bu endpoint'ler korumasız kalır. +INTERNAL_API_TOKEN= diff --git a/apps/api-daemon/src/capture/streamIngest.ts b/apps/api-daemon/src/capture/streamIngest.ts index 55bc1de..405de12 100644 --- a/apps/api-daemon/src/capture/streamIngest.ts +++ b/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(); +/** + * 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(); + +/** 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): Promise { ], { stdio: ["ignore", "pipe", "pipe"] }, ); + activeCaptures.set(sessionId, ytdlp); const ffmpeg = spawn( "ffmpeg", @@ -153,6 +172,7 @@ async function runCapture(job: Job): Promise { }); 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(() => ""); diff --git a/apps/api-daemon/src/cleanup.ts b/apps/api-daemon/src/cleanup.ts index 6976ab4..3ca0454 100644 --- a/apps/api-daemon/src/cleanup.ts +++ b/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 { + 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 { - 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 { } 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)); diff --git a/apps/api-daemon/src/env.ts b/apps/api-daemon/src/env.ts index 074a3d7..f0eaf82 100644 --- a/apps/api-daemon/src/env.ts +++ b/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 ?? "", }; diff --git a/apps/api-daemon/src/queues.ts b/apps/api-daemon/src/queues.ts index 4693928..d0280a2 100644 --- a/apps/api-daemon/src/queues.ts +++ b/apps/api-daemon/src/queues.ts @@ -34,3 +34,13 @@ export const streamIngestQueue = new Queue(QUEUE_NAMES.STREAM_I export const signalDetectionQueue = new Queue(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(QUEUE_NAMES.VIDEO_RENDER, { + connection: redisConnection, +}); diff --git a/apps/api-daemon/src/server.ts b/apps/api-daemon/src/server.ts index 451a6b8..5d94a8a 100644 --- a/apps/api-daemon/src/server.ts +++ b/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}`); }); diff --git a/apps/api-daemon/src/youtubePolling.ts b/apps/api-daemon/src/youtubePolling.ts index 8b44bc8..e05188e 100644 --- a/apps/api-daemon/src/youtubePolling.ts +++ b/apps/api-daemon/src/youtubePolling.ts @@ -69,11 +69,22 @@ async function findLiveVideoId(channelId: string): Promise { } } -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 { @@ -94,25 +127,39 @@ export async function pollOnce(): Promise { 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 { + 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(); } diff --git a/apps/frontend/app/actions.ts b/apps/frontend/app/actions.ts index 11f8ce5..b7580a4 100644 --- a/apps/frontend/app/actions.ts +++ b/apps/frontend/app/actions.ts @@ -4,6 +4,7 @@ import { unlink } from "node:fs/promises"; import { prisma } from "@streamclipper/db"; import { revalidatePath } from "next/cache"; import { redirect } from "next/navigation"; +import { postToApiDaemon } from "../lib/apiDaemon"; export async function addChannel(formData: FormData) { const name = String(formData.get("name") ?? "").trim(); @@ -59,6 +60,85 @@ export async function deleteChannel(channelId: string) { redirect("/"); } +export async function toggleChannelActive(channelId: string, nextActive: boolean) { + await prisma.channel.update({ where: { id: channelId }, data: { isActive: nextActive } }); + revalidatePath("/", "layout"); +} + +export async function toggleAllChannels(nextActive: boolean) { + await prisma.channel.updateMany({ data: { isActive: nextActive } }); + revalidatePath("/", "layout"); +} + +export async function stopSession(sessionId: string) { + await postToApiDaemon(`/sessions/${sessionId}/stop`); + revalidatePath("/", "layout"); +} + +export async function checkChannelNow(channelId: string) { + await postToApiDaemon(`/channels/${channelId}/check-now`); + revalidatePath("/", "layout"); +} + +export async function forceStartCapture(channelId: string, formData: FormData) { + const url = String(formData.get("url") ?? "").trim(); + if (!url) throw new Error("YouTube URL zorunlu."); + await postToApiDaemon(`/channels/${channelId}/force-start`, { url }); + revalidatePath("/", "layout"); +} + +export async function retryRender(candidateId: string) { + await postToApiDaemon(`/shorts/${candidateId}/render`); + revalidatePath("/segments"); + revalidatePath("/", "layout"); +} + +export async function cancelRender(candidateId: string) { + await postToApiDaemon(`/shorts/${candidateId}/cancel`); + revalidatePath("/segments"); + revalidatePath("/", "layout"); +} + +export async function updateSystemSettings(formData: FormData) { + const pollSeconds = Number(formData.get("pollIntervalSec")); + const ttlHours = Number(formData.get("ttlHours")); + + if (Number.isFinite(pollSeconds) && pollSeconds >= 5) { + await prisma.appSetting.upsert({ + where: { key: "poll_interval_ms" }, + create: { key: "poll_interval_ms", value: String(pollSeconds * 1000) }, + update: { value: String(pollSeconds * 1000) }, + }); + } + if (Number.isFinite(ttlHours) && ttlHours > 0) { + await prisma.appSetting.upsert({ + where: { key: "raw_segment_ttl_hours" }, + create: { key: "raw_segment_ttl_hours", value: String(ttlHours) }, + update: { value: String(ttlHours) }, + }); + } + + revalidatePath("/settings"); +} + +export async function updateNotificationPrefs(formData: FormData) { + const streamStart = formData.get("notifyStreamStart") === "on"; + const renderDone = formData.get("notifyRenderDone") === "on"; + + await prisma.appSetting.upsert({ + where: { key: "notify_stream_start" }, + create: { key: "notify_stream_start", value: String(streamStart) }, + update: { value: String(streamStart) }, + }); + await prisma.appSetting.upsert({ + where: { key: "notify_render_done" }, + create: { key: "notify_render_done", value: String(renderDone) }, + update: { value: String(renderDone) }, + }); + + revalidatePath("/settings"); +} + export async function updateYtdlpCookies(formData: FormData) { const value = String(formData.get("cookies") ?? "").trim(); diff --git a/apps/frontend/app/api/debug/capture/[sessionId]/route.ts b/apps/frontend/app/api/debug/capture/[sessionId]/route.ts new file mode 100644 index 0000000..3910ec5 --- /dev/null +++ b/apps/frontend/app/api/debug/capture/[sessionId]/route.ts @@ -0,0 +1,11 @@ +import { NextResponse } from "next/server"; +import { API_DAEMON_URL } from "../../../../../lib/apiDaemon"; + +/** Client components can't reach api-daemon's internal Docker hostname directly — this proxies the (auth-free, read-only) debug endpoint through the frontend's own origin. */ +export async function GET(_req: Request, { params }: { params: Promise<{ sessionId: string }> }) { + const { sessionId } = await params; + const res = await fetch(`${API_DAEMON_URL}/debug/capture/${sessionId}`, { cache: "no-store" }); + if (!res.ok) return NextResponse.json({ lines: [] }, { status: res.status }); + const data = await res.json(); + return NextResponse.json(data); +} diff --git a/apps/frontend/app/channels/[id]/page.tsx b/apps/frontend/app/channels/[id]/page.tsx index 3b14a35..8fb90ee 100644 --- a/apps/frontend/app/channels/[id]/page.tsx +++ b/apps/frontend/app/channels/[id]/page.tsx @@ -2,8 +2,19 @@ import { notFound } from "next/navigation"; import Link from "next/link"; import { prisma } from "@streamclipper/db"; import { fetchChannelStatuses } from "../../../lib/apiDaemon"; -import { deleteRawSegment, deleteCandidateSegment, deleteChannel } from "../../actions"; +import { + deleteRawSegment, + deleteCandidateSegment, + deleteChannel, + stopSession, + forceStartCapture, + retryRender, + cancelRender, +} from "../../actions"; import { DeleteButton } from "../../components/DeleteButton"; +import { ConfirmButton } from "../../components/ConfirmButton"; +import { LiveLogViewer } from "../../components/LiveLogViewer"; +import { SessionTimer } from "../../components/SessionTimer"; export const dynamic = "force-dynamic"; @@ -48,6 +59,7 @@ export default async function ChannelDetailPage({ params }: { params: Promise<{ if (!channel) notFound(); const status = statuses.get(channel.id); + const activeSession = channel.sessions.find((s) => s.endedAt === null); return ( <> @@ -64,8 +76,21 @@ export default async function ChannelDetailPage({ params }: { params: Promise<{
Canlı Durum - - {status?.recording ? "🔴 Kayıtta" : "Kayıtta değil"} + + + {status?.recording ? "🔴 Kayıtta" : "Kayıtta değil"} + {status?.recording && activeSession && ( + <> + {" · "} + + + )} + + {status?.recording && activeSession && ( +
+ + + )}
@@ -85,6 +110,27 @@ export default async function ChannelDetailPage({ params }: { params: Promise<{
)} + {status?.recording && activeSession?.liveVideoId && activeSession.liveVideoId !== "forced" && ( +