import { Context, Elysia, t } from "elysia"; import { NS_misc } from "../lib/misc"; import createError from "http-errors"; import { NS_promise } from "../lib/promise"; import { NS_error } from "../lib/error"; import { Subprocess } from "bun"; import { PassThrough, Readable, Writable } from "node:stream"; import { NS_ffmpeg } from "../lib/ffmpeg"; import * as devalue from "devalue"; import { NS_crypto_client } from "../../../../../packages/lib/helpers/client/crypto"; import { redis } from "../lib/upstash"; const APP_ENV = process.env.APP_ENV; const APP_ACCOUNT_ABC = process.env.APP_ACCOUNT_ABC; const sub_router = new Elysia({ prefix: "/ffmpeg" }) .post( "/probe", async (ctx) => { const { query, request, status, body } = ctx; let playback = body?.playback; let playback_headers = body?.playback_headers; let gen = body?.gen; let playback_type = body?.playback_type; const sbe_id = body?.sbe_id; const picture_quality = body?.picture_quality; const cache_it = body?.cache_it; const REQ0 = request.url; const REQ_ORIGIN = (ctx as any).publicOrigin || new URL(REQ0).origin; const is_valid_playback = (() => { const is_encoded = NS_misc.isUrlEncoded(playback); if (!is_encoded) playback = encodeURI(playback); return NS_misc.isValidUrlFormat(playback); })(); if (!is_valid_playback) { throw createError(400, "Please provide a valid playback URL"); } playback_headers = (() => { if (!playback_headers) { return {}; } if ( typeof playback_headers === "object" && !Array.isArray(playback_headers) ) { return playback_headers; } if (typeof playback_headers === "string") { try { const data = JSON.parse(playback_headers); if (typeof data === "object" && !Array.isArray(data)) { return data; } } catch (error: unknown) {} } return {}; })(); const [E_, R_] = await NS_promise.F_catch_promise({ promise: NS_ffmpeg.F_probe_v2({ playback_url: new URL(playback), context: ctx, playback_headers, playback_type }) }); if (E_ || !R_ || !R_.probe_data) { console.log("\n"); console.error("E_", E_); console.log("\n"); const { status, message } = NS_error.F_collect_error_status_n_message({ ERR: E_ }); throw createError(status, message); } const probe_data = R_.probe_data; if (!gen) { return R_; } let { progress_feedback_url, progress_update_endpoint, live, output_format, output_filename, subtitles } = gen; const is_valid_progress_feedback_url = (() => { if (!progress_feedback_url) return; const is_encoded = NS_misc.isUrlEncoded(progress_feedback_url); if (!is_encoded) progress_feedback_url = encodeURI(progress_feedback_url); return NS_misc.isValidUrlFormat(progress_feedback_url); })(); const tcp_progress = is_valid_progress_feedback_url ? true : false; const subtitles_ary = (() => { if (!subtitles) return; const _ = (() => { if (typeof subtitles === "string") { try { const data = JSON.parse(subtitles); if (typeof data === "object") return data; } catch (error: unknown) { return; } } return subtitles; })(); if (typeof _ === "object") { if (!Array.isArray(_)) { return [_]; } return _; } })(); const out_mime = (() => { switch (output_format) { case "mp4": { return "video/mp4"; } break; case "mkv": { return "video/x-matroska"; } break; case "ts": { return "video/mp2t"; } break; default: { return "video/mp4"; } break; } })(); let output_destination = ( output_filename || `video-${Date.now()}.${output_format}` ).trim(); if (!output_destination.endsWith(`.${output_format}`)) { output_destination = `${output_destination}.${output_format}`; } if ( !playback_type && probe_data && typeof probe_data.format_name === "string" ) { playback_type = probe_data.format_name; } const data = { playback, playback_headers, output_format, output_destination, progress_feedback_url, subtitles: subtitles_ary, out_mime, probe_data, progress_update_endpoint, playback_type, sbe_id, picture_quality, cache_it }; const data_str = devalue.stringify(data); // Encode to Base64 let data_b64 = Buffer.from(data_str).toString("base64"); // Make URL-safe: replace + with -, / with _, and remove = data_b64 = data_b64 .replace(/\+/g, "-") .replace(/\//g, "_") .replace(/=+$/, ""); // store the data in Redis const cache_key = await (async () => { const track_id = NS_crypto_client.F_crypto_randomUUID({ length: 40 }); let key = `llsvkm:${track_id}`; if (APP_ACCOUNT_ABC) key = `${key}:${APP_ACCOUNT_ABC}`; // 30 days = 30 * 24 * 60 * 60 = 2,592,000 seconds const ONE_MONTH_IN_SECONDS = 2592000; await redis.set(key, data_b64, { ex: ONE_MONTH_IN_SECONDS }); return key; })(); const url_o = new URL(`/showtime/${live}/${cache_key}`, REQ_ORIGIN); return { follow: url_o.href, ...data, metadata: R_.metadata, b64: { base: new URL(`/showtime/${live}`, REQ_ORIGIN).toString(), data: data_b64 } }; }, { body: t.Object({ playback: t.String({ minLength: 1, error() { return "Invalid playback url"; } }), playback_headers: t.Optional( t.Union([t.Object({}, { additionalProperties: true }), t.Null()], { error: () => "Invalid playback headers" }) ), playback_type: t.Optional( t.String({ minLength: 1, error() { return "Invalid playback type"; } }) ), sbe_id: t.Optional( t.String({ minLength: 1, error() { return "Invalid sbe id"; } }) ), picture_quality: t.Optional( t.String({ minLength: 1, error() { return "Invalid picture quality"; } }) ), cache_it: t.Optional( t.Boolean({ error() { return "Invalid cache it state"; } }) ), gen: t.Optional( t.Object( { live: t.Union([t.Literal("download"), t.Literal("stream")], { error() { return "Invalid live operation, must be one of: download, stream"; } }), progress_update_endpoint: t.Optional( t.String({ minLength: 1, error() { return "Invalid progress update endpoint"; } }) ), progress_feedback_url: t.Optional( t.String({ minLength: 1, error() { return "Invalid progress feedback url"; } }) ), output_format: t.Optional( t.Union([t.Literal("mp4"), t.Literal("mkv"), t.Literal("ts")], { error() { return "Invalid output format, must be one of: mp4, mkv, ts"; } }) ), output_filename: t.Optional( t.String({ minLength: 1, error() { return "Invalid filename"; } }) ), subtitles: t.Optional( t.Union( [ t.Array( t.Object( { lang: t.String(), url: t.String(), title: t.Optional(t.String()), default: t.Optional(t.Boolean()) }, { additionalProperties: true } ) ), t.Null() ], { error: () => "Invalid subtitle data" } ) ) }, { error: () => "Invalid gen data" } ) ) }) } ) .get( "/:live/:gen?", async (ctx: Context) => { const { params, query, request, status } = ctx; let { live, gen } = params; let { playback_url, output_format, output_filename, progress_feedback_url, subtitle, playback_headers: ph, ops } = query; const REQ0 = request.url; const REQ_ORIGIN = (ctx as any).publicOrigin || new URL(REQ0).origin; const is_valid_playback_url = (() => { const is_encoded = NS_misc.isUrlEncoded(playback_url); if (!is_encoded) playback_url = encodeURI(playback_url); return NS_misc.isValidUrlFormat(playback_url); })(); const is_valid_progress_feedback_url = (() => { if (!progress_feedback_url) return; const is_encoded = NS_misc.isUrlEncoded(progress_feedback_url); if (!is_encoded) progress_feedback_url = encodeURI(progress_feedback_url); return NS_misc.isValidUrlFormat(progress_feedback_url); })(); if (!is_valid_playback_url) { throw createError(400, "Please provide a valid playback URL"); } if (!output_format) { output_format = "mp4"; } const playback_headers = (() => { if (!ph) { return {}; } try { const data = JSON.parse(ph); if (typeof data === "object" && !Array.isArray(data)) { return data; } } catch (error: unknown) {} return {}; })(); const is_network_source = true; const out_mime = (() => { switch (output_format) { case "mp4": { return "video/mp4"; } break; case "mkv": { return "video/x-matroska"; } break; case "ts": { return "video/mp2t"; } break; default: { return "video/mp4"; } break; } })(); let output_destination = ( output_filename || `video-${Date.now()}.${output_format}` ).trim(); if (!output_destination.endsWith(`.${output_format}`)) { output_destination = `${output_destination}.${output_format}`; } const tcp_progress = is_valid_progress_feedback_url ? true : false; const subtitle_obj = (() => { if (!subtitle) return; try { const data = JSON.parse(subtitle); if (typeof data === "object") return data; } catch (error: unknown) { return; } })(); const [E_, R_] = await NS_promise.F_catch_promise({ promise: NS_ffmpeg.F_probe_v2({ playback_url: new URL(playback_url), context: ctx, playback_headers }) }); if (E_ || !R_) { const { status, message } = NS_error.F_collect_error_status_n_message({ ERR: E_ }); throw createError(status, message); } const probe_data = R_.probe_data; if (gen) { const data = { playback: playback_url, playback_headers, output_format, output_destination, progress_feedback_url, subtitle: subtitle_obj, out_mime, probe_data }; const data_str = devalue.stringify(data); // Encode to Base64 let data_b64 = Buffer.from(data_str).toString("base64"); // Make URL-safe: replace + with -, / with _, and remove = data_b64 = data_b64 .replace(/\+/g, "-") .replace(/\//g, "_") .replace(/=+$/, ""); const url_o = new URL(`/showtime/${live}/${data_b64}`, REQ_ORIGIN); return { follow: url_o.href }; } const abort_signal = ctx?.request.signal; const controller = new AbortController(); function kill_controller() { controller.abort(); console.log("[CONTROLLER] Request aborted"); } if (abort_signal) { if (abort_signal.aborted) { kill_controller(); } else { abort_signal.addEventListener( "abort", () => { kill_controller(); }, { once: true } ); } } console.log("\n", "PING FFMPEG NOW", "\n"); // ------------------------------- // Build commands // ------------------------------- const commands = await NS_ffmpeg.F_build_ffmpeg_command_v1({ playback_url: new URL(playback_url), context: ctx, output_format, live, probe_data, subtitles: [subtitle_obj], has_tcp_progress: tcp_progress, is_net_src: is_network_source, playback_headers: playback_headers }); if (tcp_progress) { commands.push("-progress", progress_feedback_url!.trim()); } else { commands.push("-progress", "pipe:2"); // progress goes to stderr, not fd3 use "pipe:3" } commands.push("-y", "pipe:1"); // ------------------------------- // network response headers // ------------------------------- if (is_network_source) { /* ---------- set response headers ---------- */ ctx.set.headers["content-type"] = out_mime; ctx.set.headers["cache-control"] = "no-cache"; if (live === "download") { ctx.set.headers["content-disposition"] = `attachment; filename="${output_destination}"`; } } let t: Timer | undefined; t = setTimeout(kill_controller, NS_ffmpeg.TIMEOUT_MS); function kill_controller_timeout() { if (t) { clearTimeout(t); t = undefined; } } // ------------------------------- // spawn // ------------------------------- console.log("[FFMPEG CMD]", commands, "\n"); const bun_proc = (() => { let proc: Subprocess<"ignore" | "pipe", "pipe", "pipe">; try { // Try with extra pipe (fd:3) proc = Bun.spawn([NS_ffmpeg.ffmpeg_path, ...commands], { stdout: "pipe", // video info, stream mapping stderr: "pipe", // normal logs, warnings, errors stdin: "pipe", stdio: ["ignore", "pipe", "pipe", "pipe"] /* 👈 add fd:3 */, signal: controller.signal }); (proc as any).extraPipe = true; // custom flag so we know later } catch (err) { console.warn( "[Spawn FFmpeg] Failed to create extra pipe, falling back to stderr:", err ); // Fallback: no pipe:3 proc = Bun.spawn([NS_ffmpeg.ffmpeg_path, ...commands], { stdout: "pipe", // video info, stream mapping stderr: "pipe", // normal logs, warnings, errors stdin: "pipe", signal: controller.signal }); (proc as any).extraPipe = false; } return proc; })(); const logged_proc_errors: string[] = []; const has_pipe_3 = (bun_proc as any).extraPipe && bun_proc.stdio[3]; // log progress const decoder = new TextDecoder(); let stderrBuffer = ""; // Buffer for incomplete lines // capture progress from fd:3 if (has_pipe_3) { } let video_duration = probe_data && NS_ffmpeg.F_format_duration(probe_data?.duration); const original_filesize = probe_data && NS_ffmpeg.F_format_size(probe_data?.size); // capture stderr for errors/warnings/misc (async () => { for await (const chunk of bun_proc.stderr) { const log = decoder.decode(chunk).trim(); if (!log) continue; // console.log(log, "\n"); // if ( log.toLowerCase().includes("error") || log.startsWith("[error]") ) { console.error("[ffmpeg log ERROR]", log, "\n"); logged_proc_errors.push(log); continue; } else if (log.toLowerCase().includes("failed")) { console.error("[ffmpeg log FAIL]", log, "\n"); logged_proc_errors.push(log); continue; } let progress_update = NS_ffmpeg.F_stderr_data_read({ output: log }); if (!parseInt(`${progress_update?.current_out_time}`)) { if (progress_update.video_duration && !probe_data.duration) { probe_data.duration = progress_update.video_duration; video_duration = NS_ffmpeg.F_format_duration(probe_data.duration); } (progress_update as any) = undefined; } else { const total_duration = probe_data?.duration || 0; const current_time = progress_update.current_out_time; const current_size_bytes = progress_update.current_size_bytes; progress_update.video_duration = total_duration; (progress_update as any).percentage = Math.min( parseFloat(((current_time / total_duration) * 100).toFixed(2)), 100 ); (progress_update as any).f_video_duration = video_duration; (progress_update as any).f_current_out_time = NS_ffmpeg.F_format_duration(current_time); (progress_update as any).original_filesize = original_filesize; (progress_update as any).current_filesize = NS_ffmpeg.F_format_size(current_size_bytes); } if (APP_ENV === "dev") { if ( !progress_update || !(Object.keys(progress_update).length > 0) ) { // console.log(`[ffmpeg log] ${log}`, "\n"); } } if (progress_update && Object.keys(progress_update).length > 0) { if (APP_ENV === "dev") { console.log("[ffmpeg PROGRESS]:", { progress_update }, "\n"); } if (ctx && ctx.request?.headers) { const ws_client_id = ctx.request.headers.get("x-ws-client-id"); const play_title = ctx.request.headers.get("x-play-title"); // const ws_clients = NS_websockets.ws_clients; // if (ws_client_id && ws_clients) { // const target = ws_clients.get(ws_client_id); // target?.send( // JSON.stringify({ // type: `${ // live === "download" ? "DOWNLOAD" : "STREAM" // } PROGRESS`, // data: { // filename: output_destination, // ...progress_update, // ...(play_title ? { play_title } : {}) // } // }) // ); // } } } } })(); // stream mode const passthrough = new PassThrough(); (async () => { try { for await (const chunk of bun_proc.stdout) { kill_controller_timeout(); // first byte → cancel passthrough.write(chunk); } passthrough.end(); } catch (e) { passthrough.destroy(e as Error); } })(); bun_proc.exited.then((code: number) => { kill_controller_timeout(); if (code !== 0 && !passthrough.destroyed) { passthrough.destroy(new Error(`FFmpeg exited with code ${code}`)); } else if (!passthrough.destroyed) { passthrough.end(); } if (code === 0) { console.log(`FFmpeg finished -> ${output_destination}`, "\n"); } else { console.warn( `FFmpeg exited with code: ${code}`, "\n", "[ffmpeg log ERRORS]", logged_proc_errors, "\n" ); } }); return passthrough; }, { params: t.Object({ live: t.Union([t.Literal("download"), t.Literal("stream")], { error() { return "Invalid live operation, must be one of: download, stream"; } }), gen: t.Optional( t.Union([t.Literal("gen")], { error() { return "Invalid operation, must be one of: gen"; } }) ) }), query: t.Object({ playback_url: t.String({ minLength: 1, error() { return "Invalid playback url"; } }), progress_feedback_url: t.Optional( t.String({ minLength: 1, error() { return "Invalid progress feedback url"; } }) ), output_format: t.Optional( t.Union([t.Literal("mp4"), t.Literal("mkv"), t.Literal("ts")], { error() { return "Invalid output format, must be one of: mp4, mkv, ts"; } }) ), output_filename: t.Optional( t.String({ minLength: 1, error() { return "Invalid filename"; } }) ), subtitle: t.Optional( t.String({ minLength: 1, error() { return "Invalid subtitle data"; } }) ), playback_headers: t.Optional( t.String({ minLength: 1, error() { return "Invalid playback headers"; } }) ), ops: t.Optional( t.Union([t.Literal("cmd"), t.Literal("decode")], { error() { return "Invalid operation, must be one of: cmd decode"; } }) ) }) } ) .get( "/v2/:live", async (ctx) => { const { params, query, request, status } = ctx; let { live } = params; let { playback_url, output_format, output_filename, progress_feedback_url, subtitle } = query; const REQ0 = request.url; const REQ_ORIGIN = (ctx as any).publicOrigin || new URL(REQ0).origin; const is_valid_playback_url = (() => { const is_encoded = NS_misc.isUrlEncoded(playback_url); if (!is_encoded) playback_url = encodeURI(playback_url); return NS_misc.isValidUrlFormat(playback_url); })(); const is_valid_progress_feedback_url = (() => { if (!progress_feedback_url) return; const is_encoded = NS_misc.isUrlEncoded(progress_feedback_url); if (!is_encoded) progress_feedback_url = encodeURI(progress_feedback_url); return NS_misc.isValidUrlFormat(progress_feedback_url); })(); if (!is_valid_playback_url) { throw createError(400, "Please provide a valid playback URL"); } if (!output_format) { output_format = "mp4"; } const playback_headers = {}; const is_network_source = true; const out_mime = (() => { switch (output_format) { case "mp4": { return "video/mp4"; } break; case "mkv": { return "video/x-matroska"; } break; case "ts": { return "video/mp2t"; } break; default: { return "video/mp4"; } break; } })(); let output_destination = ( output_filename || `video-${Date.now()}.${output_format}` ).trim(); if (!output_destination.endsWith(`.${output_format}`)) { output_destination = `${output_destination}.${output_format}`; } const tcp_progress = is_valid_progress_feedback_url ? true : false; const subtitle_obj = (() => { if (!subtitle) return; try { const data = JSON.parse(subtitle); if (typeof data === "object") return data; } catch (error: unknown) { return; } })(); const [E_, R_] = await NS_promise.F_catch_promise({ promise: NS_ffmpeg.F_probe_v2({ playback_url: new URL(playback_url), context: ctx }) }); if (E_ || !R_) { const { status, message } = NS_error.F_collect_error_status_n_message({ ERR: E_ }); throw createError(status, message); } const probe_data = R_.probe_data; // Setup abort handling const abort_signal = ctx?.request.signal; const controller = new AbortController(); let isAborted = false; if (abort_signal) { if (abort_signal.aborted) { controller.abort(); isAborted = true; } else { abort_signal.addEventListener( "abort", () => { console.log("[CLIENT] Request aborted"); isAborted = true; controller.abort(); }, { once: true } ); } } console.log("\n", "PING FFMPEG NOW", "\n"); // ------------------------------- // Build commands // ------------------------------- const commands = await NS_ffmpeg.F_build_ffmpeg_command_v2({ playback_url: new URL(playback_url), context: ctx, output_format, live, probe_data, subtitle: subtitle_obj, has_tcp_progress: tcp_progress, is_net_src: is_network_source, playback_headers: playback_headers }); if (tcp_progress) { commands.addOutputOption("-progress", progress_feedback_url!.trim()); } else { commands.addOutputOption("-progress", "pipe:2"); // progress goes to stderr, not fd3 use "pipe:3" } // ------------------------------- // network response headers // ------------------------------- if (is_network_source) { /* ---------- set response headers ---------- */ ctx.set.headers["content-type"] = out_mime; ctx.set.headers["cache-control"] = "no-cache"; if (live === "download") { ctx.set.headers["content-disposition"] = `attachment; filename="${output_destination}"`; } } // ------------------------------- // spawn // ------------------------------- // Create passthrough stream const passthrough = new PassThrough(); // Track errors const errors: string[] = []; let video_duration = probe_data && NS_ffmpeg.F_format_duration(probe_data?.duration); const original_filesize = probe_data && NS_ffmpeg.F_format_size(probe_data?.size); let killed = false; commands .on("start", (cmd) => { console.log("FFmpeg started:", cmd); }) .on("progress", (p) => console.log("Processing", p.percent)) .on("error", (err) => { if (!killed) { killed = true; console.error("FFmpeg error:", err.message); } }) .on("end", () => { console.log("FFmpeg finished"); }); // Handle abort signa if (controller.signal) { controller.signal.addEventListener( "abort", () => { console.log("[FFMPEG] Aborting process..."); // Kill the process gracefully first, then force if needed try { commands.kill("SIGTERM"); // If process doesn't die in 5 seconds, force kill setTimeout(() => { try { commands.kill("SIGKILL"); } catch (e) { // Already dead, ignore } }, 5000); } catch (e) { console.error("[FFMPEG] Kill error:", e); } }, { once: true } ); } // Pipe to passthrough stream commands.pipe(passthrough, { end: true }); // Clean up on stream close passthrough.on("close", () => { if (!isAborted) { try { commands.kill("SIGTERM"); } catch (e) { // Ignore if already dead } } }); return passthrough; }, { params: t.Object({ live: t.Union([t.Literal("download"), t.Literal("stream")], { error() { return "Invalid live operation, must be one of: download, stream"; } }) }), query: t.Object({ playback_url: t.String({ minLength: 1, error() { return "Invalid playback url"; } }), progress_feedback_url: t.Optional( t.String({ minLength: 1, error() { return "Invalid progress feedback url"; } }) ), output_format: t.Optional( t.Union([t.Literal("mp4"), t.Literal("mkv"), t.Literal("ts")], { error() { return "Invalid output format, must be one of: mp4, mkv, ts"; } }) ), output_filename: t.Optional( t.String({ minLength: 1, error() { return "Invalid filename"; } }) ), subtitle: t.Optional( t.String({ minLength: 1, error() { return "Invalid subtitle data"; } }) ) }) } ); export const FfmpegRouter = sub_router;