src/sub-router/ffmpeg.ts

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";
import { NS_runtime } from "../../../../../packages/lib/env.runtime";

const APP_ENV = () => NS_runtime.ENV_VAR_DB?.APP_ENV || process.env.APP_ENV;
const APP_ACCOUNT_ABC = () =>
  NS_runtime.ENV_VAR_DB?.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;