src/lib/ffmpeg-v2.ts

import ffmpeg from "ffmpeg-static";
import ffprobe from "ffprobe-static";
import fluent_ffmpeg, { type FfprobeData } from "fluent-ffmpeg";
import createError from "http-errors";
import { Context } from "elysia";
// import { T_Probe_v2 } from "./types";
import { NS_misc } from "./misc";
import { NS_promise } from "./promise";
import { NS_error } from "./error";
import UserAgent from "user-agents";
import { exec, execSync, spawn } from "node:child_process";
import { T_live, T_playback } from "../sub-router/ffmpeg/validator-v2";
import { NS_runtime } from "../../../../../packages/lib/env.runtime";

const APP_ENV = () => NS_runtime.ENV_VAR_DB?.APP_ENV || process.env.APP_ENV;

const FFMPEG_V = process.env.FFMPEG_V;
const FFPROBE_V = process.env.FFPROBE_V;
const CPU_0V1 = process.env.CPU_0V1;

const is_prod =
  process.env.NODE_ENV === "production" ||
  APP_ENV() === "prod" ||
  APP_ENV() === "pprod" ||
  APP_ENV() === "pre_prod";

const ffmpeg_path = is_prod ? "ffmpeg" : ffmpeg || "ffmpeg";
const ffprobe_path = is_prod ? "ffprobe" : ffprobe.path || "ffprobe";

fluent_ffmpeg.setFfmpegPath(ffmpeg_path);
fluent_ffmpeg.setFfprobePath(ffprobe_path);

let hw_encoder: any = await get_hardware_encoder();
hw_encoder =
  typeof hw_encoder !== "string" || !hw_encoder.trim()
    ? "libx264"
    : hw_encoder.trim();
// hw_encoder = "libx264";

console.log(`🚀 Using hardware encoder: ${hw_encoder}`);

const TIMEOUT_MS = 30_000; // probe + transcode timeout
const PROBE_TIMEOUT_MS = 30_000; // probe + transcode timeout

function get_hardware_encoder() {
  check_if_fFmpeg_active();

  return new Promise((resolve) => {
    // We wrap ffmpeg path in quotes in case there are spaces in the path
    const checkCmd = `"${ffmpeg_path}" -encoders | grep -E "nvenc|qsv|vaapi"`;

    exec(checkCmd, (error, stdout) => {
      if (error || !stdout) {
        return resolve(null);
      }

      const output = stdout.toLowerCase();
      if (output.includes("nvenc")) return resolve("h264_nvenc");
      if (output.includes("qsv")) return resolve("h264_qsv");
      if (output.includes("vaapi")) return resolve("h264_vaapi");

      resolve(null);
    });
  });
}

function check_if_fFmpeg_active() {
  try {
    const version = execSync(`${ffmpeg_path} -version`, { encoding: "utf8" });
    console.log("\n");
    console.log("⚙️ ", version.split("\n")[0], "\n");
    return true;
  } catch {
    console.error("❗FFmpeg not found in PATH", "\n");
    return false;
  }
}

export const NS_ffmpeg_v2 = {
  F_probe_t1,
  F_scan_raw_probe_metadata,
  F_build_ffmpeg_flags_t1,
  proc_error_pattern,
  F_get_qLevel
};

async function F_probe_t1(arg0: {
  playback: Omit<T_playback, "picture_quality">;
  context?: Context;
  just_cmd?: boolean;
  timeout?: number;
}) {
  const { context, playback, just_cmd, timeout } = arg0;

  if (!NS_misc.isValidUrlFormat(playback.url)) {
    if (playback.protocol !== "local") {
      throw createError(400, "invalid playback URL");
    }
  }

  const controller = new AbortController();

  function kill_controller() {
    controller.abort();
    console.log("[CONTROLLER] Request aborted");
  }

  const _PROBE_TIMEOUT_MS =
    typeof timeout === "number" ? timeout : PROBE_TIMEOUT_MS;

  const t =
    _PROBE_TIMEOUT_MS <= 0
      ? undefined
      : setTimeout(() => kill_controller(), _PROBE_TIMEOUT_MS);

  const abort_signal = context?.request.signal;

  if (abort_signal) {
    if (abort_signal.aborted) {
      kill_controller();
    } else {
      abort_signal.addEventListener(
        "abort",
        () => {
          kill_controller();
        },
        {
          once: true
        }
      );
    }
  }

  const flags = [];

  flags.push("-v", "debug");

  flags.push("-protocol_whitelist", "file,http,https,tcp,tls,crypto");
  // if (playback.type === "hls") flags.push("-allowed_extensions", "ALL");

  flags.push("-probesize", `${1_000_000 * 10}`); // Limit probe size to 10MB (default is 5MB)
  flags.push("-analyzeduration", `${1_000_000 * 10}`); // Analyze only 10 seconds (default is 5s)

  flags.push("-show_format");
  flags.push("-show_streams");

  flags.push("-print_format", "json");

  (() => {
    const playback_headers = playback.headers;

    if (
      typeof playback_headers !== "object" ||
      Array.isArray(playback_headers)
    ) {
      return;
    }

    const headers_str = Object.entries(playback_headers)
      .map(([k, v]) => `${k}: ${v}`)
      .join("\r\n");

    flags.push("-headers", headers_str + "\r\n");
  })();

  flags.push("-i", playback.url);

  if (APP_ENV() === "dev") {
    console.log("[PROBE FLAGS]", flags, "\n");
  }

  if (just_cmd) {
    clearTimeout(t);

    return { cmd: flags.join(" ") };
  }

  const timeoutPromise = () =>
    new Promise<never>((_, reject) => {
      setTimeout(() => {
        clearTimeout(t);
        reject(new Error("PROBE is taking way to long"));
      }, _PROBE_TIMEOUT_MS);
    });

  const [E_, R_] = await NS_promise.F_catch_promise({
    promise: Promise.race([
      probe_runner({
        ffprobe_path,
        args: flags,
        controller
      }),
      timeoutPromise()
    ])
  });

  clearTimeout(t);

  if (E_) {
    // timeout / hard failure -> this IS your "slow" bucket per your own rule
    throw E_;
  }

  const { data, bytesRead, durationMs } = R_ as any;
  const throughputMbps = (bytesRead * 8) / (durationMs / 1000) / 1_000_000;

  console.log({ throughputMbps });

  (() => {
    // SPEED TEST
    const bit_rate = data?.format?.bit_rate;
    const streams = data?.streams;

    let safetyFactor = 1.5;

    let sourceBitrateMbps = bit_rate ? Number(bit_rate) / 1_000_000 : undefined;

    // HLS variant playlists often don't set format.bit_rate — fall back to summing streams
    if (!sourceBitrateMbps) {
      const streamSum = (streams ?? []).reduce(
        (s: number, st: { bit_rate: any }) =>
          s + (st.bit_rate ? Number(st.bit_rate) : 0),
        0
      );
      sourceBitrateMbps = streamSum > 0 ? streamSum / 1_000_000 : undefined;
    }

    const _ = { tier: "unknown", throughputMbps };

    if(sourceBitrateMbps) {
      _.tier = "unknown";
      return;
    }

    const requiredMbps = sourceBitrateMbps * safetyFactor;
    const tier =
      throughputMbps >= requiredMbps * 1.5 ? "fast" :
      throughputMbps >= requiredMbps ? "medium" : "slow";

    _.tier = tier

    console.log({ _ });
  })();

  return data;
}

function probe_runner(arg0: {
  ffprobe_path: string;
  args: string[];
  controller?: AbortController;
}) {
  const { ffprobe_path, args, controller } = arg0;

  return new Promise((resolve, reject) => {
    const start = performance.now();

    const probe = spawn(ffprobe_path, args, { signal: controller?.signal });

    let dataOutput = "";
    let errorOutput = "";

    probe.stdout.on("data", (data) => {
      dataOutput += data;
    });

    probe.stderr.on("data", (data) => {
      errorOutput += data;
    });

    probe.on("close", (code) => {
      if (code !== 0) {
        reject(
          new Error(
            JSON.stringify({
              code,
              message: `ffprobe exited with code ${code}: ${errorOutput}`
            })
          )
        );
        return;
      }

      const durationMs = performance.now() - start;

      const bytesRead = extract_bytes_read(errorOutput);

      try {
        resolve({ data: JSON.parse(dataOutput), bytesRead, durationMs });
      } catch (err: any) {
        reject(
          new Error(`PROBE Failed to parse ffprobe output: ${err.message}`)
        );
      }
    });

    probe.on("error", (err) => {
      // This will be called with err being an AbortError if the controller aborts
      const { message } = NS_error.F_collect_error_message({ ERR: err });

      reject(new Error(`PROBE FAILED: ${message}`));
    });
  });

  function extract_bytes_read(stderr: string): number {
    // robust to ffmpeg version differences in exact line formatting
    const matches = [...stderr.matchAll(/(\d+)\s*bytes read/gi)];
    return matches.reduce((sum, m) => sum + Number(m[1]), 0);
  }
}

function F_scan_raw_probe_metadata(arg0: { metadata: Record<string, any> }) {
  const { metadata } = arg0;

  const streams = metadata.streams;

  if (!streams) return;

  const audioStreams: Record<string, any>[] = [];
  const videoStreams: Record<string, any>[] = [];
  const subStreams: Record<string, any>[] = [];

  let hasSubs = false;

  // Single pass through streams
  for (const stream of streams) {
    if (stream.codec_type === "video") {
      // if (!videoStream) videoStream = stream;
      videoStreams.push(stream);
    } else if (stream.codec_type === "audio") {
      // if (!audioStream) audioStream = stream;
      audioStreams.push(stream);
    } else if (stream.codec_type === "subtitle") {
      hasSubs = true;
      subStreams.push(stream);
    }
  }

  const format = (() => {
    if (!metadata.format) return;

    const xx = metadata.format;

    return {
      name: xx.format_name,
      duration: xx.duration,
      bit_rate: xx.bit_rate,
      size: xx.size,
      probe_score: xx.probe_score
    };
  })();

  return { audioStreams, videoStreams, subStreams, format };
}

async function F_build_ffmpeg_flags_t1(arg0: {
  playback: Omit<T_playback, "picture_quality">;
  subtitles?: T_live["subtitles"];
  output_destination: NonNullable<T_live["delivery_name"]>;
  delivery_mode: T_live["delivery_mode"];
  delivery_format: T_live["delivery_format"];
  probe_metadata: Record<string, any>;
  flash_ping?: boolean;
}) {
  const {
    playback,
    subtitles,
    output_destination,
    delivery_format,
    delivery_mode,
    probe_metadata
  } = arg0;

  let { flash_ping } = arg0;

  const flags: string[] = ["-hide_banner", "-loglevel", "info", "-stats"];

  flags.push("-reconnect", "1");
  flags.push("-reconnect_streamed", "1");
  if (playback.type === "hls" && delivery_mode === "stream") {
    flags.push("-reconnect_at_eof", "1");
  }
  flags.push("-reconnect_delay_max", "30");
  flags.push("-rw_timeout", `${1_000_000 * 30}`);
  flags.push("-reconnect_on_network_error", "1");
  // flags.push("-reconnect_on_http_error", "429,500,502,503,504");
  flags.push("-reconnect_on_http_error", "500,502,503,504");

  // flags.push("-fflags", "+discardcorrupt+genpts+nobuffer");
  // flags.push("-live_start_index", "-2"); // For live streams: start near the end
  // flags.push("-fflags", "+discardcorrupt"); // Don't hang on truncated segments

  if (playback.type === "hls") {
    if (flash_ping === undefined || flash_ping === true) {
      flags.push("-http_persistent", "1");
      flags.push("-http_multiple", "1");
      flags.push("-http_seekable", "1");
      flash_ping = true;
    } else {
      flags.push("-http_persistent", "0");
      flags.push("-http_multiple", "0");
      flags.push("-http_seekable", "0");
    }
  }

  flags.push("-protocol_whitelist", "file,http,https,tcp,tls,crypto");
  if (playback.type === "hls") flags.push("-allowed_extensions", "ALL");

  (() => {
    const playback_headers = playback.headers;

    if (
      typeof playback_headers !== "object" ||
      Array.isArray(playback_headers)
    ) {
      return;
    }

    const headers_str = Object.entries(playback_headers)
      .map(([k, v]) => `${k}: ${v}`)
      .join("\r\n");

    flags.push("-headers", headers_str + "\r\n");
  })();

  flags.push("-i", playback.url);

  const clean_metadata = NS_ffmpeg_v2.F_scan_raw_probe_metadata({
    metadata: probe_metadata
  });

  let subs: any[] | undefined = undefined;

  (() => {
    (() => {
      if (delivery_format === "ts") {
        subs = undefined;
      }

      if (clean_metadata?.subStreams && clean_metadata.subStreams.length) {
        subs = clean_metadata.subStreams;

        return;
      }

      if (!subtitles || !subtitles.length) return;

      subs = subtitles;

      for (let i = 0; i < subtitles.length; i++) {
        const sub = subtitles[i];

        flags.push("-i", sub.url);
      }
    })();

    if (subs?.length) {
      const limit = 5;
      subs = subs.slice(0, limit);
    }
  })();

  const videoCodec = clean_metadata?.videoStreams[0].codec_name;
  const audioCodec = clean_metadata?.audioStreams[0].codec_name;

  const supported_codecs = (() => {
    switch (delivery_format) {
      case "mp4":
        {
          return {
            video: ["h264", "hevc", "vp9", "av1", "mpeg2video"],
            audio: ["aac", "mp3", "opus", "ac3"],
            subtitle: ["mov_text"]
          };
        }

        break;

      case "ts":
        {
          return {
            video: ["h264", "hevc", "mpeg2video"],
            audio: ["aac", "mp3", "ac3"],
            subtitle: []
          };
        }

        break;

      case "mkv":
        {
          return {
            video: [
              "h264",
              "hevc",
              "vp9",
              "av1",
              "mpeg2video",
              "vp8",
              "theora"
            ],
            audio: [
              "aac",
              "mp3",
              "opus",
              "ac3",
              "flac",
              "vorbis",
              "dts",
              "truehd"
            ],
            subtitle: ["srt", "ass", "ssa", "vobsub", "pgs", "webvtt"]
          };
        }

        break;
    }
  })();

  const must_re_encode =
    !supported_codecs.video.includes(videoCodec) ||
    !supported_codecs.audio.includes(audioCodec);

  flags.push("-map", "0:v");
  flags.push("-map", "0:a");

  if (subs) {
    if (clean_metadata?.subStreams.length) {
      flags.push("-map", "0:s");
    } else {
      subs.forEach((_, i) => flags.push("-map", String(i + 1)));
    }
  }

  if (!must_re_encode) {
    flags.push("-c:v", "copy");
    flags.push("-c:a", "copy");
  } else {
    flags.push("-c:v", hw_encoder || "libx264");
    flags.push("-c:a", "aac", "-b:a", "256k");
  }

  if (playback.type === "hls") flags.push("-bsf:a", "aac_adtstoasc");

  if (subs) {
    flags.push("-c:s", "mov_text");

    subs.forEach((s, i) => {
      let lang = s.lang || s.tags?.language;

      if (!lang) {
        lang = "und";
      }

      lang = lang.slice(0, 3).toLowerCase();

      flags.push(`-metadata:s:s:${i}`, `language=${lang}`);
    });
  }

  if (delivery_mode === "stream") {
    flags.push("-tune", "zerolatency");
  }

  if (must_re_encode || delivery_mode === "stream") {
    const use_ultrafast = delivery_mode === "stream";

    if (hw_encoder.includes("h264_qsv")) {
      flags.push("-preset", use_ultrafast ? "7" : "7"); // QSV uses numbers: 7 is fastest speed (ultrafast), 4 is balanced (medium)
    } else {
      flags.push("-crf", "23");
      flags.push("-preset", use_ultrafast ? "ultrafast" : "ultrafast"); // ultrafast veryfast faster medium
    }
  }

  if (must_re_encode) flags.push("-pix_fmt", "yuv420p");

  flags.push("-max_interleave_delta", "0");

  if (delivery_mode === "stream") {
    //   flags.push("-f", "ismv");
    // } else {
    flags.push("-f", delivery_format === "ts" ? "mpegts" : delivery_format);
  }

  flags.push("-avoid_negative_ts", "make_zero");

  if (delivery_mode === "download") {
    flags.push("-max_muxing_queue_size", "9999");

    flags.push("-movflags", "+faststart");

    flags.push("-y", output_destination);
  } else {
    flags.push("-fps_mode", "passthrough");

    flags.push("-movflags", "frag_keyframe+dash+delay_moov");

    flags.push("-y", "pipe:1");
  }

  return {
    flash_ping,
    flags
  };
}

function proc_error_pattern() {
  const errorPatterns = [
    /^(\[error\]|\[fatal\]|\[panic\])/i,
    /error/i,
    /failed/i,
    /failure/i,
    /invalid/i,
    /unknown/i,
    /unsupported/i,
    /not found/i,
    /cannot/i,
    /could not/i,
    /unable to/i,
    /permission denied/i,
    /no such file/i,
    /broken pipe/i,
    /protocol not found/i,
    /codec not found/i,
    /encoder not found/i,
    /decoder not found/i,
    /filter not found/i,
    /option not found/i,
    /conversion failed/i,
    /av_interleaved_write_frame/i,
    /out of memory/i,
    /segmentation fault/i,
    /assertion failed/i,
    /corrupt/i,
    /truncated/i,
    /malformed/i,
    /incompatible/i,
    /mismatch/i,
    /overflow/i,
    /underflow/i,
    /timeout/i,
    /connection refused/i,
    /connection reset/i,
    /network is unreachable/i,
    /no route to host/i,
    /dns/i, // DNS resolution failures
    /unrecognized/i
  ];

  return errorPatterns;
}

function F_get_qLevel(arg0: { metadata: Record<string, any> }) {
  const { metadata } = arg0;

  if (!NS_misc.is_plain_object(metadata)) {
    return;
  }

  const clean_metadata = F_scan_raw_probe_metadata({
    metadata: metadata
  });

  const videoStreams = clean_metadata?.videoStreams;
  if (!videoStreams) return;

  const seen = new Set();
  const uniqueByDimensions = videoStreams.filter((xx) => {
    const key = `${xx.width}x${xx.height}`;
    if (seen.has(key)) {
      return false;
    } else {
      seen.add(key);
      return true;
    }
  });

  if (
    !uniqueByDimensions ||
    !Array.isArray(uniqueByDimensions) ||
    !uniqueByDimensions.length ||
    uniqueByDimensions.length > 1
  ) {
    return;
  }

  const _ = uniqueByDimensions[0];

  if (_.width <= 0) {
    return;
  }

  switch (_.width) {
    case 3840:
      {
        return "4K";
      }

      break;

    case 2048:
      {
        return "2K";
      }

      break;

    case 1920:
      {
        return "1080P";
      }

      break;

    case 1280:
      {
        return "720P";
      }

      break;

    case 854:
      {
        return "480P";
      }

      break;

    case 640:
      {
        return "360P";
      }

      break;

    case 480:
      {
        return "320P";
      }

      break;

    case 426:
      {
        return "240P";
      }

      break;

    default:
      {
        return `${_.height}P`;
      }
      break;
  }
}