path: root/vmedia/server.js

// Video help for the feed, served at v3.coinos.io/vmedia/ (nginx → this
// container, which carries ffmpeg):
//
//   /vmedia/thumb.jpg?u=<clip>  a still from the clip's first half second,
//     480px wide — the poster a post's imeta didn't give. A 4K or 134 MB
//     clip waits for a tap, so without one the feed showed a black box.
//   /vmedia/h264.mp4?u=<clip>   the clip as H.264/AAC at 720p, for a browser
//     that can't decode the original's picture (an iPhone's HEVC on Chrome
//     for Linux plays its sound over black). The first ask streams the
//     transcode as it's made (fragmented MP4); once done it's a plain
//     faststart file served in ranges and cached for everyone.
//
// Only https sources on public addresses; ffmpeg may speak nothing else. A
// few jobs at a time, a per-address hourly allowance for new transcodes,
// and both caches trimmed to a size.
import { mkdirSync, existsSync, statSync, readdirSync, unlinkSync, renameSync, utimesSync, openSync, readSync, closeSync } from 'node:fs';
import { lookup } from 'node:dns/promises';
import { isIP } from 'node:net';
import { createHash } from 'node:crypto';

const DATA = process.env.DATA || '/data';
const THUMBS = DATA + '/thumbs', CLIPS = DATA + '/h264', WORK = DATA + '/work';
for (const d of [THUMBS, CLIPS, WORK]) mkdirSync(d, { recursive: true });
const THUMB_CAP = 2e9, CLIP_CAP = 25e9;
const THUMB_JOBS = 4, CLIP_JOBS = 2, CLIP_QUEUE = 6;
const CLIP_PER_HOUR = 12; // new transcodes per address
const MAX_SECONDS = 900;
const FAIL_TTL = 3_600_000;

// ---- sources --------------------------------------------------------------
const privateV4 = (ip) => {
  const [a, b] = ip.split('.').map(Number);
  return a === 0 || a === 10 || a === 127 || (a === 100 && b >= 64 && b < 128) || (a === 169 && b === 254)
    || (a === 172 && b >= 16 && b < 32) || (a === 192 && b === 168) || a >= 224;
};
const privateV6 = (ip) => {
  const x = ip.toLowerCase();
  if (x === '::' || x === '::1') return true;
  const m = /^::ffff:(\d+\.\d+\.\d+\.\d+)$/.exec(x);
  if (m) return privateV4(m[1]);
  return /^(fc|fd|fe8|fe9|fea|feb|ff)/.test(x);
};
export const privateIp = (ip) => (isIP(ip) === 6 ? privateV6(ip) : privateV4(ip));
// a clip worth fetching: https, an ordinary port, a public name
export async function sourceOk(raw, resolve = (h) => lookup(h, { all: true })) {
  let u;
  try { u = new URL(raw); } catch { return null; }
  if (u.protocol !== 'https:' || u.username || u.password || (u.port && u.port !== '443') || raw.length > 2000) return null;
  const host = u.hostname.replace(/^\[|\]$/g, '');
  if (/^(localhost|.*\.local|.*\.internal)$/i.test(host)) return null;
  try {
    const addrs = isIP(host) ? [{ address: host }] : await resolve(host);
    if (!addrs.length || addrs.some((a) => privateIp(a.address))) return null;
  } catch { return null; }
  return u.href;
}

// ---- ffmpeg ---------------------------------------------------------------
const NET = ['-protocol_whitelist', 'https,tls,tcp,crypto', '-rw_timeout', '15000000', '-user_agent', 'coinos-vmedia'];
// longest side at most 1280, even numbers for the encoder
export const SCALE_720 = "scale='trunc(min(1,1280/max(iw,ih))*iw/2)*2':'trunc(min(1,1280/max(iw,ih))*ih/2)*2'";
export const thumbArgs = (src, at) => ['-hide_banner', '-loglevel', 'error', ...NET, ...(at ? ['-ss', String(at)] : []), '-i', src,
  '-frames:v', '1', '-vf', "scale='min(480,iw)':-2", '-q:v', '5', '-f', 'image2', '-c:v', 'mjpeg', 'pipe:1'];
export const clipArgs = (src, out) => ['-hide_banner', '-loglevel', 'error', '-y', ...NET, '-i', src,
  '-map', '0:v:0', '-map', '0:a:0?', '-t', String(MAX_SECONDS), '-vf', SCALE_720,
  '-c:v', 'libx264', '-preset', 'veryfast', '-crf', '26', '-profile:v', 'high', '-pix_fmt', 'yuv420p', '-threads', '4',
  '-c:a', 'aac', '-b:a', '128k', '-ac', '2',
  '-movflags', 'frag_keyframe+empty_moov+default_base_moof', '-f', 'mp4', out];
const run = (args, ms) => {
  const p = Bun.spawn(['ffmpeg', ...args], { stdout: 'pipe', stderr: 'pipe' });
  const timer = setTimeout(() => p.kill(9), ms);
  return { p, done: p.exited.finally(() => clearTimeout(timer)) };
};

const keyOf = (src) => createHash('sha256').update(src).digest('hex');
const failed = new Map(); // key -> at
const failedLately = (k) => { const at = failed.get(k); return at && Date.now() - at < FAIL_TTL; };
const touch = (p) => { try { const n = new Date(); utimesSync(p, n, n); } catch {} };

// a few at a time, the rest wait their turn
function limiter(n) {
  let busy = 0;
  const q = [];
  const next = () => { while (busy < n && q.length) { busy++; const { fn, res, rej } = q.shift(); fn().then(res, rej).finally(() => { busy--; next(); }); } };
  const go = (fn) => new Promise((res, rej) => { q.push({ fn, res, rej }); next(); });
  go.waiting = () => q.length;
  return go;
}

// ---- thumbnails -----------------------------------------------------------
const thumbSlots = limiter(THUMB_JOBS);
const thumbJobs = new Map();
async function makeThumb(src, key) {
  const out = `${THUMBS}/${key}.jpg`;
  for (const at of [0.5, 0]) { // a clip shorter than half a second has only its start
    const { p, done } = run(thumbArgs(src, at), 25_000);
    const [buf] = await Promise.all([new Response(p.stdout).arrayBuffer(), done]);
    if (p.exitCode === 0 && buf.byteLength > 500) { await Bun.write(out, buf); return out; }
  }
  failed.set(key, Date.now());
  return null;
}
async function thumb(src) {
  const key = keyOf(src), out = `${THUMBS}/${key}.jpg`;
  if (existsSync(out)) { touch(out); return out; }
  if (failedLately(key)) return null;
  if (!thumbJobs.has(key)) thumbJobs.set(key, thumbSlots(() => makeThumb(src, key)).finally(() => thumbJobs.delete(key)));
  return thumbJobs.get(key);
}

// ---- transcodes -----------------------------------------------------------
const clipSlots = limiter(CLIP_JOBS);
const clipJobs = new Map(); // key -> { work, started, done, ok }
const allowance = new Map(); // ip -> [times]
function allowed(ip) {
  const now = Date.now(), list = (allowance.get(ip) || []).filter((t) => now - t < 3_600_000);
  if (list.length >= CLIP_PER_HOUR) { allowance.set(ip, list); return false; }
  list.push(now); allowance.set(ip, list);
  if (allowance.size > 5000) allowance.delete(allowance.keys().next().value);
  return true;
}
function startClip(src, key) {
  const work = `${WORK}/${key}.mp4`, final = `${CLIPS}/${key}.mp4`;
  const job = { work, ok: false, finished: false };
  let begin;
  job.started = new Promise((r) => (begin = r));
  job.done = clipSlots(async () => {
    begin();
    const { p, done } = run(clipArgs(src, work), 20 * 60_000);
    await done;
    if (p.exitCode !== 0 || !existsSync(work) || statSync(work).size < 1000) {
      failed.set(key, Date.now());
      console.log('transcode failed', src, (await new Response(p.stderr).text()).slice(-300));
      return;
    }
    // the finished file, its index first, for range requests and the cache
    const fs = run(['-hide_banner', '-loglevel', 'error', '-y', '-i', work, '-c', 'copy', '-movflags', '+faststart', '-f', 'mp4', final + '.part'], 120_000);
    await fs.done;
    if (fs.p.exitCode === 0) { renameSync(final + '.part', final); job.ok = true; }
    else failed.set(key, Date.now());
  }).catch(() => failed.set(key, Date.now())).finally(() => {
    job.finished = true;
    // readers still tailing the work file finish from it; then it goes
    setTimeout(() => { clipJobs.delete(key); try { unlinkSync(work); } catch {} }, 10 * 60_000);
  });
  clipJobs.set(key, job);
  return job;
}
const sizeOf = (p) => { try { return statSync(p).size; } catch { return 0; } };
// the work file as it grows, to the end of the job
function tail(job) {
  let at = 0, fd = null;
  return new ReadableStream({
    async pull(ctl) {
      for (;;) {
        const size = sizeOf(job.work);
        if (size > at) {
          if (fd == null) fd = openSync(job.work, 'r');
          const buf = new Uint8Array(Math.min(size - at, 1 << 20));
          const n = readSync(fd, buf, 0, buf.length, at);
          at += n;
          ctl.enqueue(buf.subarray(0, n));
          return;
        }
        if (job.finished) { if (fd != null) closeSync(fd); ctl.close(); return; }
        await Bun.sleep(200);
      }
    },
    cancel() { if (fd != null) try { closeSync(fd); } catch {} },
  });
}

// ---- responses --------------------------------------------------------------
const WEEK = 'public, max-age=604800, immutable';
function fileResponse(req, path, type) {
  const file = Bun.file(path), size = file.size;
  const base = { 'Content-Type': type, 'Accept-Ranges': 'bytes', 'Cache-Control': WEEK, 'Access-Control-Allow-Origin': '*' };
  const m = /^bytes=(\d*)-(\d*)$/.exec(req.headers.get('range') || '');
  if (m && (m[1] || m[2])) {
    let start = m[1] ? +m[1] : Math.max(0, size - +m[2]);
    let end = m[1] && m[2] ? Math.min(+m[2], size - 1) : size - 1;
    if (start >= size || start > end) return new Response(null, { status: 416, headers: { ...base, 'Content-Range': `bytes */${size}` } });
    return new Response(req.method === 'HEAD' ? null : file.slice(start, end + 1),
      { status: 206, headers: { ...base, 'Content-Range': `bytes ${start}-${end}/${size}`, 'Content-Length': String(end - start + 1) } });
  }
  return new Response(req.method === 'HEAD' ? null : file, { headers: { ...base, 'Content-Length': String(size) } });
}
const nope = (status, text) => new Response(text, { status, headers: { 'Cache-Control': status === 404 ? 'public, max-age=3600' : 'no-store', 'Access-Control-Allow-Origin': '*' } });

async function clipResponse(req, src, ip) {
  const key = keyOf(src), final = `${CLIPS}/${key}.mp4`;
  if (existsSync(final)) { touch(final); return fileResponse(req, final, 'video/mp4'); }
  if (failedLately(key)) return nope(404, 'Not playable');
  let job = clipJobs.get(key);
  if (job && job.finished && job.ok) return fileResponse(req, final, 'video/mp4');
  if (job && job.finished) return nope(404, 'Not playable');
  if (!job) {
    if (clipSlots.waiting() >= CLIP_QUEUE) return nope(503, 'Busy');
    if (!allowed(ip)) return nope(429, 'Too many');
    job = startClip(src, key);
  }
  // answer once there's something to play (or nothing ever will be)
  await job.started;
  const t0 = Date.now();
  while (!job.finished && sizeOf(job.work) < 64_000 && Date.now() - t0 < 60_000) await Bun.sleep(150);
  if (job.finished && job.ok && existsSync(final)) return fileResponse(req, final, 'video/mp4');
  if (sizeOf(job.work) === 0) return nope(job.finished ? 404 : 503, 'Not playable');
  return new Response(req.method === 'HEAD' ? null : tail(job),
    { headers: { 'Content-Type': 'video/mp4', 'Cache-Control': 'no-store', 'Access-Control-Allow-Origin': '*' } });
}

// ---- the caches stay a size ------------------------------------------------
function trim(dir, cap) {
  try {
    const files = readdirSync(dir).filter((f) => !f.endsWith('.part')).map((f) => {
      const p = dir + '/' + f, s = statSync(p);
      return { p, size: s.size, at: s.mtimeMs };
    }).sort((a, b) => a.at - b.at);
    let total = files.reduce((n, f) => n + f.size, 0);
    for (const f of files) { if (total <= cap) break; try { unlinkSync(f.p); total -= f.size; } catch {} }
  } catch {}
}
export function startTrim() {
  const go = () => { trim(THUMBS, THUMB_CAP); trim(CLIPS, CLIP_CAP); };
  go();
  setInterval(go, 10 * 60_000);
  // work left over from a restart is no good to anyone
  for (const f of readdirSync(WORK)) try { unlinkSync(WORK + '/' + f); } catch {}
}

export async function handle(req) {
  const u = new URL(req.url);
  if (!['GET', 'HEAD'].includes(req.method)) return nope(405, 'GET only');
  const route = u.pathname.replace(/^\/vmedia/, '');
  if (route !== '/thumb.jpg' && route !== '/h264.mp4') return nope(404, 'Not found');
  const src = await sourceOk(u.searchParams.get('u') || '');
  if (!src) return nope(400, 'Not a public https URL');
  const ip = req.headers.get('cf-connecting-ip') || req.headers.get('x-real-ip') || 'unknown';
  if (route === '/thumb.jpg') {
    const p = await thumb(src);
    return p ? fileResponse(req, p, 'image/jpeg') : nope(404, 'No picture');
  }
  return clipResponse(req, src, ip);
}

if (import.meta.main) {
  startTrim();
  const server = Bun.serve({ hostname: process.env.HOST || '0.0.0.0', port: Number(process.env.PORT || 8801), idleTimeout: 0, fetch: handle });
  console.log('vmedia on ' + server.url);
}