openvibers/openvibe.media Commits
Thumbnails: fix frozen live frames on still-growing recordings
Changes
@@@ -224,9 +224,17 @@ router.get('/live/:sel/frame.jpg', async (req, res) => {};try {res.set('Access-Control-Allow-Origin', '*');++ // Each cache miss spawns ffprobe+ffmpeg, so the endpoint is guarded by a++ // per-IP token bucket before any work is done.++ if (!frames.rateLimitOk(req.ip)) {++ res.set('Retry-After', '2');++ if (wantJson) return res.status(429).json({ error: 'Too many requests' });++ return res.status(429).end();++ }const appId = String(req.query.app || 'live');-- let w = parseInt(req.query.w, 10);-- w = Number.isFinite(w) ? Math.min(1920, Math.max(64, w)) : null;++ // Snap to a fixed width ladder — an unbounded `w` would give every request++ // its own cache key and defeat the cache that IS the rate limit.++ const w = frames.quantizeWidth(req.query.w);const resolved = await frames.resolveSelector(appId, req.params.sel);if (!resolved.msid) return sendCard(404, resolved.label, 'is offline right now');@@@ -234,6 +242,10 @@ router.get('/live/:sel/frame.jpg', async (req, res) => {const out = await frames.getLiveFrame(appId, resolved.msid, w);if (!out.ok) {if (out.reason === 'not_live') return sendCard(404, resolved.label, 'is offline right now');++ if (out.reason === 'busy') {++ res.set('Retry-After', '2');++ return sendCard(503, resolved.label, 'is live — server busy, retry shortly');++ }return sendCard(503, resolved.label, 'is live — frame unavailable, retry shortly');}res.set({@@@ -21,22 +21,95 @@ const CACHE_TTL_MS = 5000;const EXTRACT_TIMEOUT_MS = 10000;const PROBE_TIMEOUT_MS = 6000;++// How far behind the tail to seek. The recorder is still writing the last++// fragment, so back off a little to stay inside flushed data.++const LIVE_EDGE_BACKOFF_SEC = 3;++++// ── Abuse limits ─────────────────────────────────────────────++// This endpoint spawns ffprobe/ffmpeg per miss, so every knob a caller can turn++// is bounded. The 5s cache is only a rate limit if the CACHE KEY is bounded too:++// `w` used to accept any value in 64..1920, i.e. ~1857 distinct keys per slot,++// so a caller walking `?w=` bypassed the cache entirely and span up two++// processes per request. Widths are therefore quantized to a short ladder.++const WIDTH_LADDER = [160, 320, 480, 640, 960, 1280, 1920];++const MAX_CACHE_ENTRIES = 400; // hard cap on cached frames/placeholders++const MAX_CONCURRENT_EXTRACTS = 3; // global ffmpeg ceiling for this endpoint++const RATE_BURST = 30; // per-IP token bucket++const RATE_REFILL_PER_SEC = 3;++const MAX_RATE_BUCKETS = 5000;++const _cache = new Map(); // key → { at, out }const _inFlight = new Map(); // key → Promise<out>++let _activeExtracts = 0;--function _probeDuration(file) {-- return new Promise((resolve) => {++/** Snap a requested width to the nearest allowed rung (null = native size). */++function quantizeWidth(w) {++ const n = parseInt(w, 10);++ if (!Number.isFinite(n)) return null;++ return WIDTH_LADDER.reduce((best, cur) =>++ Math.abs(cur - n) < Math.abs(best - n) ? cur : best, WIDTH_LADDER[0]);++}++++// Per-IP token bucket. In-memory and O(1); the map is swept when it grows past++// MAX_RATE_BUCKETS so a spoofed-IP flood cannot grow it without bound.++const _rateBuckets = new Map(); // ip → { tokens, at }++function rateLimitOk(ip) {++ const now = Date.now();++ const key = String(ip || 'unknown');++ let b = _rateBuckets.get(key);++ if (!b) {++ if (_rateBuckets.size >= MAX_RATE_BUCKETS) {++ const cutoff = now - (RATE_BURST / RATE_REFILL_PER_SEC) * 1000;++ for (const [k, v] of _rateBuckets) { if (v.at < cutoff) _rateBuckets.delete(k); }++ // Still full of active talkers → drop the oldest few.++ while (_rateBuckets.size >= MAX_RATE_BUCKETS) {++ const k = _rateBuckets.keys().next().value;++ if (k === undefined) break;++ _rateBuckets.delete(k);++ }++ }++ b = { tokens: RATE_BURST, at: now };++ _rateBuckets.set(key, b);++ }++ b.tokens = Math.min(RATE_BURST, b.tokens + ((now - b.at) / 1000) * RATE_REFILL_PER_SEC);++ b.at = now;++ if (b.tokens < 1) return false;++ b.tokens -= 1;++ return true;++}++++/**++ * Position of the newest written frame, in seconds.++ *++ * A still-recording file has NO duration in its container header (it is written++ * when the muxer finalizes), so `format=duration` reports "N/A" for exactly the++ * files this service exists to read. Reading the LAST video packet's timestamp++ * gives the real tail instead; `-read_intervals 999999%+#1` seeks to the end and++ * reads a single packet, so it costs ~0.1s even on a multi-GB growing file.++ * Falls back to the header duration for finalized files.++ *++ * @returns {Promise<number>} seconds, or 0 if it cannot be determined++ */++function _probeTailSeconds(file) {++ const run = (args) => new Promise((resolve) => {let out = '';let p;-- try {-- p = spawn('ffprobe', ['-v', 'error', '-show_entries', 'format=duration',-- '-of', 'default=noprint_wrappers=1:nokey=1', file]);-- } catch { return resolve(0); }++ try { p = spawn('ffprobe', args); } catch { return resolve(0); }const to = setTimeout(() => { try { p.kill('SIGKILL'); } catch { /* */ } resolve(0); }, PROBE_TIMEOUT_MS);p.stdout.on('data', (d) => { out += d; });-- p.on('close', () => { clearTimeout(to); resolve(parseFloat(out) || 0); });++ p.on('close', () => {++ clearTimeout(to);++ const last = String(out).trim().split(/\s+/).filter(Boolean).pop();++ const v = parseFloat(last);++ resolve(Number.isFinite(v) && v > 0 ? v : 0);++ });p.on('error', () => { clearTimeout(to); resolve(0); });});++++ return run(['-v', 'error', '-select_streams', 'v:0', '-show_entries', 'packet=pts_time',++ '-of', 'csv=p=0', '-read_intervals', '999999%+#1', file])++ .then((pts) => pts > 0 ? pts : run(['-v', 'error', '-show_entries', 'format=duration',++ '-of', 'default=noprint_wrappers=1:nokey=1', file]));}function _extractFrame(file, seekSeconds, width) {@@@ -68,12 +141,25 @@ async function _grab(appId, managedStreamId, width) {if (!vod || !vod.file_path || !fs.existsSync(vod.file_path)) {return { ok: false, reason: 'not_live' };}-- // Seek near the live edge; a short back-off keeps us inside written fragments.-- const duration = await _probeDuration(vod.file_path);-- let buf = await _extractFrame(vod.file_path, Math.max(0, duration - 2), width);-- if (!buf && duration > 4) buf = await _extractFrame(vod.file_path, Math.max(0, duration / 2), width);-- if (!buf) buf = await _extractFrame(vod.file_path, 0, width);-- return buf ? { ok: true, buf } : { ok: false, reason: 'unavailable' };++ // Global ceiling on concurrent transcodes. Shed load immediately rather than++ // queueing — a queue under attack just turns into unbounded memory + CPU.++ // 'busy' is deliberately NOT cached, so a legitimate caller retries cheaply.++ if (_activeExtracts >= MAX_CONCURRENT_EXTRACTS) return { ok: false, reason: 'busy' };++ _activeExtracts++;++ try {++ // Seek near the live edge; a short back-off keeps us inside written++ // fragments. Previously the tail probe returned 0 for a growing WebM, so++ // this seeked to 0 and served the FIRST frame of the broadcast forever.++ const tail = await _probeTailSeconds(vod.file_path);++ let buf = null;++ if (tail > 0) buf = await _extractFrame(vod.file_path, Math.max(0, tail - LIVE_EDGE_BACKOFF_SEC), width);++ // One fallback only (was three): if the tail is unreadable, take whatever++ // the head has. Each extra attempt is another ffmpeg process per request.++ if (!buf) buf = await _extractFrame(vod.file_path, 0, width);++ return buf ? { ok: true, buf } : { ok: false, reason: 'unavailable' };++ } finally {++ _activeExtracts--;++ }}// ── Selector resolution (slot id / slot slug / @username) ────@@@ -167,37 +253,60 @@ function getLiveFrame(appId, managedStreamId, width = null) {return { ok: false, reason: 'unavailable' };}).then((out) => {-- _cache.set(key, { at: Date.now(), out });++ // Don't cache load-shedding — it is about server state, not this slot.++ if (out.reason !== 'busy') _cacheSet(key, out);_inFlight.delete(key);-- // Opportunistic sweep so dead slots don't accumulate.-- if (_cache.size > 500) {-- const cutoff = Date.now() - CACHE_TTL_MS;-- for (const [k, v] of _cache) { if (v.at < cutoff) _cache.delete(k); }-- }return out;});_inFlight.set(key, p);return p;}++// Bounded cache write: expire first, then evict oldest-inserted, so the map can++// never grow without limit no matter how many distinct keys are requested.++function _cacheSet(key, out) {++ if (_cache.size >= MAX_CACHE_ENTRIES) {++ const cutoff = Date.now() - CACHE_TTL_MS;++ for (const [k, v] of _cache) { if (v.at < cutoff) _cache.delete(k); }++ while (_cache.size >= MAX_CACHE_ENTRIES) {++ const k = _cache.keys().next().value;++ if (k === undefined) break;++ _cache.delete(k);++ }++ }++ _cache.set(key, { at: Date.now(), out });++}++// The endpoint is frame.jpg — dev pipelines feed the bytes straight into image// decoders, so the placeholder must be a real JPEG too. Rasterize the SVG card// via sharp, cached per label (the card for a given name rarely changes).const _cardCache = new Map(); // `${label}|${subtitle}` → { at, buf }const CARD_CACHE_TTL_MS = 60_000;async function offlineCardJpeg(label, subtitle) {++ // The label is caller-controlled (any /live/:sel), so it is both truncated++ // (offlineCardSvg slices to 40) and used in a bounded, expiring cache —++ // otherwise a walk of random names would rasterize unbounded sharp jobs.const key = `${label}|${subtitle}`;const hit = _cardCache.get(key);if (hit && Date.now() - hit.at < CARD_CACHE_TTL_MS) return hit.buf;const buf = await require('sharp')(Buffer.from(offlineCardSvg(label, subtitle))).jpeg({ quality: 88 }).toBuffer();-- if (_cardCache.size > 200) _cardCache.clear();++ if (_cardCache.size >= 200) {++ const cutoff = Date.now() - CARD_CACHE_TTL_MS;++ for (const [k, v] of _cardCache) { if (v.at < cutoff) _cardCache.delete(k); }++ while (_cardCache.size >= 200) {++ const k = _cardCache.keys().next().value;++ if (k === undefined) break;++ _cardCache.delete(k);++ }++ }_cardCache.set(key, { at: Date.now(), buf });return buf;}module.exports = {getLiveFrame, resolveSelector, offlineCardSvg, offlineCardJpeg, CACHE_TTL_MS,++ quantizeWidth, rateLimitOk,// shared with the dev-data API (transcripts / chat insight)APP_INTERNAL_URLS, appLiveStreams: _appLiveStreams,};@@@ -36,12 +36,61 @@ if (!fs.existsSync(THUMB_DIR)) {// ── Generate Thumbnail from Video File (VODs & Clips) ────────const _activeThumbJobs = new Set(); // Dedup concurrent VOD/clip thumbnail generation++// How far behind the live edge to grab a frame for an in-progress recording.++// The recorder is still writing the tail fragment, so back off a few seconds to++// stay inside data that has actually been flushed to disk.++const LIVE_EDGE_BACKOFF_SEC = 3;++++/**++ * Duration of a file whose container header has no usable duration.++ *++ * A recording that is still being written (fragmented mp4, and especially WebM/++ * Matroska) carries NO duration in its header — it is only written when the++ * muxer finalizes the file. `ffprobe -show_format` therefore reports "N/A", and++ * `parseFloat('N/A')` is NaN, which silently fails every `duration > 2` check.++ *++ * Reading the timestamp of the LAST video packet gives the real position instead.++ * `-read_intervals 999999%+#1` makes ffprobe seek to (effectively) the end and++ * read a single packet, so this costs ~0.1s even on a multi-GB growing file++ * rather than decoding it.++ *++ * @returns {Promise<number>} seconds, or 0 when it cannot be determined++ */++function _probeLastPts(videoPath) {++ return new Promise((resolve) => {++ let out = '';++ let p;++ try {++ p = spawn('ffprobe', [++ '-v', 'error',++ '-select_streams', 'v:0',++ '-show_entries', 'packet=pts_time',++ '-of', 'csv=p=0',++ '-read_intervals', '999999%+#1',++ videoPath,++ ]);++ } catch { return resolve(0); }++ const to = setTimeout(() => { try { p.kill('SIGKILL'); } catch { /* */ } resolve(0); }, 10000);++ p.stdout.on('data', (d) => { out += d; });++ p.on('close', () => {++ clearTimeout(to);++ // The interval can yield a few packets; the last non-empty one wins.++ const last = String(out).trim().split(/\s+/).filter(Boolean).pop();++ const v = parseFloat(last);++ resolve(Number.isFinite(v) && v > 0 ? v : 0);++ });++ p.on('error', () => { clearTimeout(to); resolve(0); });++ });++}++/*** Extract a thumbnail frame from a video file using ffmpeg.* @param {string} videoPath – Absolute path (or presigned URL) of the video* @param {string} prefix – Filename prefix ('vod' or 'clip')* @param {number} entityId – DB id (vod or clip id)-- * @param {object} opts – { seekPercent?, seekSeconds? }++ * @param {object} opts – { seekPercent?, seekSeconds?, liveEdge? }++ * liveEdge: the file is an in-progress recording — grab the newest frame++ * available instead of a percentage of a duration the header does not have.* @returns {Promise<string|null>} public thumbnail URL (/t/<file>), or null*/function generateFromVideo(videoPath, prefix, entityId, opts = {}) {@@@ -68,20 +117,39 @@ function generateFromVideo(videoPath, prefix, entityId, opts = {}) {let probeData = '';probe.stdout.on('data', (d) => (probeData += d));-- probe.on('close', (probeCode) => {++ probe.on('close', async (probeCode) => {let seekTime = opts.seekSeconds || 1;-- if (probeCode === 0 && !opts.seekSeconds) {-- try {-- const info = JSON.parse(probeData);-- const duration = parseFloat(info.format?.duration || '0');-- if (duration > 2) {-- seekTime = Math.min(-- duration * ((opts.seekPercent || 10) / 100),-- duration - 0.5-- );-- }-- } catch {}++ if (!opts.seekSeconds) {++ let duration = 0;++ if (probeCode === 0) {++ try {++ const info = JSON.parse(probeData);++ // parseFloat('N/A') is NaN — normalize so the checks below++ // treat "no usable duration" as 0 rather than silently++ // falling through to the default 1-second seek.++ const d = parseFloat(info.format?.duration);++ if (Number.isFinite(d)) duration = d;++ } catch { /* */ }++ }++++ // A still-recording file has no header duration; find the real++ // tail position from the last video packet instead.++ if (opts.liveEdge || duration <= 0) {++ const lastPts = await _probeLastPts(videoPath);++ if (lastPts > 0) duration = lastPts;++ }++++ if (opts.liveEdge) {++ // Newest frame that is safely on disk. Falls back to the++ // start only when the tail could not be located at all.++ seekTime = duration > 0 ? Math.max(0, duration - LIVE_EDGE_BACKOFF_SEC) : 0;++ } else if (duration > 2) {++ seekTime = Math.min(++ duration * ((opts.seekPercent || 10) / 100),++ duration - 0.5++ );++ }}// Extract frame@@@ -112,9 +180,18 @@ function generateFromVideo(videoPath, prefix, entityId, opts = {}) {}).finally(() => _activeThumbJobs.delete(jobKey));}--// ── Generate VOD Thumbnail (~10% in) ─────────────────────────--async function generateVodThumbnail(vodId, filePath) {-- const thumbUrl = await generateFromVideo(filePath, 'vod', vodId, { seekPercent: 10 });++// ── Generate VOD Thumbnail (~10% in, or the live edge while recording) ───────++// While a stream is live, OpenVibe.Live regenerates this VOD's thumbnail every++// ~2 min and uses it as the stream's card image. "10% in" is the wrong frame for++// that: on a growing file the header carries no duration, so every regeneration++// produced the identical 1-second frame and the card looked frozen for the whole++// broadcast. An in-progress recording therefore grabs the live edge instead.++async function generateVodThumbnail(vodId, filePath, opts = {}) {++ const isRecording = opts.liveEdge != null++ ? !!opts.liveEdge++ : !!db.get('SELECT is_recording FROM vods WHERE id = ?', [vodId])?.is_recording;++ const thumbUrl = await generateFromVideo(filePath, 'vod', vodId,++ isRecording ? { liveEdge: true } : { seekPercent: 10 });if (thumbUrl) {// Clean up the previous thumbnail for this VOD_removeOldThumb(db.get('SELECT thumbnail_url FROM vods WHERE id = ?', [vodId])?.thumbnail_url, thumbUrl);
@@ -224,9 +224,17 @@ router.get('/live/:sel/frame.jpg', async (req, res) => {
@@ -224,9 +224,17 @@ router.get('/live/:sel/frame.jpg', async (req, res) => {
};
};
try {
try {
res.set('Access-Control-Allow-Origin', '*');
res.set('Access-Control-Allow-Origin', '*');
+ // Each cache miss spawns ffprobe+ffmpeg, so the endpoint is guarded by a
+ // per-IP token bucket before any work is done.
+ if (!frames.rateLimitOk(req.ip)) {
+ res.set('Retry-After', '2');
+ if (wantJson) return res.status(429).json({ error: 'Too many requests' });
+ return res.status(429).end();
+ }
const appId = String(req.query.app || 'live');
const appId = String(req.query.app || 'live');
- let w = parseInt(req.query.w, 10);
- w = Number.isFinite(w) ? Math.min(1920, Math.max(64, w)) : null;
+ // Snap to a fixed width ladder — an unbounded `w` would give every request
+ // its own cache key and defeat the cache that IS the rate limit.
+ const w = frames.quantizeWidth(req.query.w);
const resolved = await frames.resolveSelector(appId, req.params.sel);
const resolved = await frames.resolveSelector(appId, req.params.sel);
if (!resolved.msid) return sendCard(404, resolved.label, 'is offline right now');
if (!resolved.msid) return sendCard(404, resolved.label, 'is offline right now');
@@ -234,6 +242,10 @@ router.get('/live/:sel/frame.jpg', async (req, res) => {
@@ -234,6 +242,10 @@ router.get('/live/:sel/frame.jpg', async (req, res) => {
const out = await frames.getLiveFrame(appId, resolved.msid, w);
const out = await frames.getLiveFrame(appId, resolved.msid, w);
if (!out.ok) {
if (!out.ok) {
if (out.reason === 'not_live') return sendCard(404, resolved.label, 'is offline right now');
if (out.reason === 'not_live') return sendCard(404, resolved.label, 'is offline right now');
+ if (out.reason === 'busy') {
+ res.set('Retry-After', '2');
+ return sendCard(503, resolved.label, 'is live — server busy, retry shortly');
+ }
return sendCard(503, resolved.label, 'is live — frame unavailable, retry shortly');
return sendCard(503, resolved.label, 'is live — frame unavailable, retry shortly');
}
}
res.set({
res.set({
@@ -21,22 +21,95 @@ const CACHE_TTL_MS = 5000;
@@ -21,22 +21,95 @@ const CACHE_TTL_MS = 5000;
const EXTRACT_TIMEOUT_MS = 10000;
const EXTRACT_TIMEOUT_MS = 10000;
const PROBE_TIMEOUT_MS = 6000;
const PROBE_TIMEOUT_MS = 6000;
+// How far behind the tail to seek. The recorder is still writing the last
+// fragment, so back off a little to stay inside flushed data.
+const LIVE_EDGE_BACKOFF_SEC = 3;
+
+// ── Abuse limits ─────────────────────────────────────────────
+// This endpoint spawns ffprobe/ffmpeg per miss, so every knob a caller can turn
+// is bounded. The 5s cache is only a rate limit if the CACHE KEY is bounded too:
+// `w` used to accept any value in 64..1920, i.e. ~1857 distinct keys per slot,
+// so a caller walking `?w=` bypassed the cache entirely and span up two
+// processes per request. Widths are therefore quantized to a short ladder.
+const WIDTH_LADDER = [160, 320, 480, 640, 960, 1280, 1920];
+const MAX_CACHE_ENTRIES = 400; // hard cap on cached frames/placeholders
+const MAX_CONCURRENT_EXTRACTS = 3; // global ffmpeg ceiling for this endpoint
+const RATE_BURST = 30; // per-IP token bucket
+const RATE_REFILL_PER_SEC = 3;
+const MAX_RATE_BUCKETS = 5000;
+
const _cache = new Map(); // key → { at, out }
const _cache = new Map(); // key → { at, out }
const _inFlight = new Map(); // key → Promise<out>
const _inFlight = new Map(); // key → Promise<out>
+let _activeExtracts = 0;
-function _probeDuration(file) {
- return new Promise((resolve) => {
+/** Snap a requested width to the nearest allowed rung (null = native size). */
+function quantizeWidth(w) {
+ const n = parseInt(w, 10);
+ if (!Number.isFinite(n)) return null;
+ return WIDTH_LADDER.reduce((best, cur) =>
+ Math.abs(cur - n) < Math.abs(best - n) ? cur : best, WIDTH_LADDER[0]);
+}
+
+// Per-IP token bucket. In-memory and O(1); the map is swept when it grows past
+// MAX_RATE_BUCKETS so a spoofed-IP flood cannot grow it without bound.
+const _rateBuckets = new Map(); // ip → { tokens, at }
+function rateLimitOk(ip) {
+ const now = Date.now();
+ const key = String(ip || 'unknown');
+ let b = _rateBuckets.get(key);
+ if (!b) {
+ if (_rateBuckets.size >= MAX_RATE_BUCKETS) {
+ const cutoff = now - (RATE_BURST / RATE_REFILL_PER_SEC) * 1000;
+ for (const [k, v] of _rateBuckets) { if (v.at < cutoff) _rateBuckets.delete(k); }
+ // Still full of active talkers → drop the oldest few.
+ while (_rateBuckets.size >= MAX_RATE_BUCKETS) {
+ const k = _rateBuckets.keys().next().value;
+ if (k === undefined) break;
+ _rateBuckets.delete(k);
+ }
+ }
+ b = { tokens: RATE_BURST, at: now };
+ _rateBuckets.set(key, b);
+ }
+ b.tokens = Math.min(RATE_BURST, b.tokens + ((now - b.at) / 1000) * RATE_REFILL_PER_SEC);
+ b.at = now;
+ if (b.tokens < 1) return false;
+ b.tokens -= 1;
+ return true;
+}
+
+/**
+ * Position of the newest written frame, in seconds.
+ *
+ * A still-recording file has NO duration in its container header (it is written
+ * when the muxer finalizes), so `format=duration` reports "N/A" for exactly the
+ * files this service exists to read. Reading the LAST video packet's timestamp
+ * gives the real tail instead; `-read_intervals 999999%+#1` seeks to the end and
+ * reads a single packet, so it costs ~0.1s even on a multi-GB growing file.
+ * Falls back to the header duration for finalized files.
+ *
+ * @returns {Promise<number>} seconds, or 0 if it cannot be determined
+ */
+function _probeTailSeconds(file) {
+ const run = (args) => new Promise((resolve) => {
let out = '';
let out = '';
let p;
let p;
- try {
- p = spawn('ffprobe', ['-v', 'error', '-show_entries', 'format=duration',
- '-of', 'default=noprint_wrappers=1:nokey=1', file]);
- } catch { return resolve(0); }
+ try { p = spawn('ffprobe', args); } catch { return resolve(0); }
const to = setTimeout(() => { try { p.kill('SIGKILL'); } catch { /* */ } resolve(0); }, PROBE_TIMEOUT_MS);
const to = setTimeout(() => { try { p.kill('SIGKILL'); } catch { /* */ } resolve(0); }, PROBE_TIMEOUT_MS);
p.stdout.on('data', (d) => { out += d; });
p.stdout.on('data', (d) => { out += d; });
- p.on('close', () => { clearTimeout(to); resolve(parseFloat(out) || 0); });
+ p.on('close', () => {
+ clearTimeout(to);
+ const last = String(out).trim().split(/\s+/).filter(Boolean).pop();
+ const v = parseFloat(last);
+ resolve(Number.isFinite(v) && v > 0 ? v : 0);
+ });
p.on('error', () => { clearTimeout(to); resolve(0); });
p.on('error', () => { clearTimeout(to); resolve(0); });
});
});
+
+ return run(['-v', 'error', '-select_streams', 'v:0', '-show_entries', 'packet=pts_time',
+ '-of', 'csv=p=0', '-read_intervals', '999999%+#1', file])
+ .then((pts) => pts > 0 ? pts : run(['-v', 'error', '-show_entries', 'format=duration',
+ '-of', 'default=noprint_wrappers=1:nokey=1', file]));
}
}
function _extractFrame(file, seekSeconds, width) {
function _extractFrame(file, seekSeconds, width) {
@@ -68,12 +141,25 @@ async function _grab(appId, managedStreamId, width) {
@@ -68,12 +141,25 @@ async function _grab(appId, managedStreamId, width) {
if (!vod || !vod.file_path || !fs.existsSync(vod.file_path)) {
if (!vod || !vod.file_path || !fs.existsSync(vod.file_path)) {
return { ok: false, reason: 'not_live' };
return { ok: false, reason: 'not_live' };
}
}
- // Seek near the live edge; a short back-off keeps us inside written fragments.
- const duration = await _probeDuration(vod.file_path);
- let buf = await _extractFrame(vod.file_path, Math.max(0, duration - 2), width);
- if (!buf && duration > 4) buf = await _extractFrame(vod.file_path, Math.max(0, duration / 2), width);
- if (!buf) buf = await _extractFrame(vod.file_path, 0, width);
- return buf ? { ok: true, buf } : { ok: false, reason: 'unavailable' };
+ // Global ceiling on concurrent transcodes. Shed load immediately rather than
+ // queueing — a queue under attack just turns into unbounded memory + CPU.
+ // 'busy' is deliberately NOT cached, so a legitimate caller retries cheaply.
+ if (_activeExtracts >= MAX_CONCURRENT_EXTRACTS) return { ok: false, reason: 'busy' };
+ _activeExtracts++;
+ try {
+ // Seek near the live edge; a short back-off keeps us inside written
+ // fragments. Previously the tail probe returned 0 for a growing WebM, so
+ // this seeked to 0 and served the FIRST frame of the broadcast forever.
+ const tail = await _probeTailSeconds(vod.file_path);
+ let buf = null;
+ if (tail > 0) buf = await _extractFrame(vod.file_path, Math.max(0, tail - LIVE_EDGE_BACKOFF_SEC), width);
+ // One fallback only (was three): if the tail is unreadable, take whatever
+ // the head has. Each extra attempt is another ffmpeg process per request.
+ if (!buf) buf = await _extractFrame(vod.file_path, 0, width);
+ return buf ? { ok: true, buf } : { ok: false, reason: 'unavailable' };
+ } finally {
+ _activeExtracts--;
+ }
}
}
// ── Selector resolution (slot id / slot slug / @username) ────
// ── Selector resolution (slot id / slot slug / @username) ────
@@ -167,37 +253,60 @@ function getLiveFrame(appId, managedStreamId, width = null) {
@@ -167,37 +253,60 @@ function getLiveFrame(appId, managedStreamId, width = null) {
return { ok: false, reason: 'unavailable' };
return { ok: false, reason: 'unavailable' };
})
})
.then((out) => {
.then((out) => {
- _cache.set(key, { at: Date.now(), out });
+ // Don't cache load-shedding — it is about server state, not this slot.
+ if (out.reason !== 'busy') _cacheSet(key, out);
_inFlight.delete(key);
_inFlight.delete(key);
- // Opportunistic sweep so dead slots don't accumulate.
- if (_cache.size > 500) {
- const cutoff = Date.now() - CACHE_TTL_MS;
- for (const [k, v] of _cache) { if (v.at < cutoff) _cache.delete(k); }
- }
return out;
return out;
});
});
_inFlight.set(key, p);
_inFlight.set(key, p);
return p;
return p;
}
}
+// Bounded cache write: expire first, then evict oldest-inserted, so the map can
+// never grow without limit no matter how many distinct keys are requested.
+function _cacheSet(key, out) {
+ if (_cache.size >= MAX_CACHE_ENTRIES) {
+ const cutoff = Date.now() - CACHE_TTL_MS;
+ for (const [k, v] of _cache) { if (v.at < cutoff) _cache.delete(k); }
+ while (_cache.size >= MAX_CACHE_ENTRIES) {
+ const k = _cache.keys().next().value;
+ if (k === undefined) break;
+ _cache.delete(k);
+ }
+ }
+ _cache.set(key, { at: Date.now(), out });
+}
+
// The endpoint is frame.jpg — dev pipelines feed the bytes straight into image
// The endpoint is frame.jpg — dev pipelines feed the bytes straight into image
// decoders, so the placeholder must be a real JPEG too. Rasterize the SVG card
// decoders, so the placeholder must be a real JPEG too. Rasterize the SVG card
// via sharp, cached per label (the card for a given name rarely changes).
// via sharp, cached per label (the card for a given name rarely changes).
const _cardCache = new Map(); // `${label}|${subtitle}` → { at, buf }
const _cardCache = new Map(); // `${label}|${subtitle}` → { at, buf }
const CARD_CACHE_TTL_MS = 60_000;
const CARD_CACHE_TTL_MS = 60_000;
async function offlineCardJpeg(label, subtitle) {
async function offlineCardJpeg(label, subtitle) {
+ // The label is caller-controlled (any /live/:sel), so it is both truncated
+ // (offlineCardSvg slices to 40) and used in a bounded, expiring cache —
+ // otherwise a walk of random names would rasterize unbounded sharp jobs.
const key = `${label}|${subtitle}`;
const key = `${label}|${subtitle}`;
const hit = _cardCache.get(key);
const hit = _cardCache.get(key);
if (hit && Date.now() - hit.at < CARD_CACHE_TTL_MS) return hit.buf;
if (hit && Date.now() - hit.at < CARD_CACHE_TTL_MS) return hit.buf;
const buf = await require('sharp')(Buffer.from(offlineCardSvg(label, subtitle)))
const buf = await require('sharp')(Buffer.from(offlineCardSvg(label, subtitle)))
.jpeg({ quality: 88 }).toBuffer();
.jpeg({ quality: 88 }).toBuffer();
- if (_cardCache.size > 200) _cardCache.clear();
+ if (_cardCache.size >= 200) {
+ const cutoff = Date.now() - CARD_CACHE_TTL_MS;
+ for (const [k, v] of _cardCache) { if (v.at < cutoff) _cardCache.delete(k); }
+ while (_cardCache.size >= 200) {
+ const k = _cardCache.keys().next().value;
+ if (k === undefined) break;
+ _cardCache.delete(k);
+ }
+ }
_cardCache.set(key, { at: Date.now(), buf });
_cardCache.set(key, { at: Date.now(), buf });
return buf;
return buf;
}
}
module.exports = {
module.exports = {
getLiveFrame, resolveSelector, offlineCardSvg, offlineCardJpeg, CACHE_TTL_MS,
getLiveFrame, resolveSelector, offlineCardSvg, offlineCardJpeg, CACHE_TTL_MS,
+ quantizeWidth, rateLimitOk,
// shared with the dev-data API (transcripts / chat insight)
// shared with the dev-data API (transcripts / chat insight)
APP_INTERNAL_URLS, appLiveStreams: _appLiveStreams,
APP_INTERNAL_URLS, appLiveStreams: _appLiveStreams,
};
};
@@ -36,12 +36,61 @@ if (!fs.existsSync(THUMB_DIR)) {
@@ -36,12 +36,61 @@ if (!fs.existsSync(THUMB_DIR)) {
// ── Generate Thumbnail from Video File (VODs & Clips) ────────
// ── Generate Thumbnail from Video File (VODs & Clips) ────────
const _activeThumbJobs = new Set(); // Dedup concurrent VOD/clip thumbnail generation
const _activeThumbJobs = new Set(); // Dedup concurrent VOD/clip thumbnail generation
+// How far behind the live edge to grab a frame for an in-progress recording.
+// The recorder is still writing the tail fragment, so back off a few seconds to
+// stay inside data that has actually been flushed to disk.
+const LIVE_EDGE_BACKOFF_SEC = 3;
+
+/**
+ * Duration of a file whose container header has no usable duration.
+ *
+ * A recording that is still being written (fragmented mp4, and especially WebM/
+ * Matroska) carries NO duration in its header — it is only written when the
+ * muxer finalizes the file. `ffprobe -show_format` therefore reports "N/A", and
+ * `parseFloat('N/A')` is NaN, which silently fails every `duration > 2` check.
+ *
+ * Reading the timestamp of the LAST video packet gives the real position instead.
+ * `-read_intervals 999999%+#1` makes ffprobe seek to (effectively) the end and
+ * read a single packet, so this costs ~0.1s even on a multi-GB growing file
+ * rather than decoding it.
+ *
+ * @returns {Promise<number>} seconds, or 0 when it cannot be determined
+ */
+function _probeLastPts(videoPath) {
+ return new Promise((resolve) => {
+ let out = '';
+ let p;
+ try {
+ p = spawn('ffprobe', [
+ '-v', 'error',
+ '-select_streams', 'v:0',
+ '-show_entries', 'packet=pts_time',
+ '-of', 'csv=p=0',
+ '-read_intervals', '999999%+#1',
+ videoPath,
+ ]);
+ } catch { return resolve(0); }
+ const to = setTimeout(() => { try { p.kill('SIGKILL'); } catch { /* */ } resolve(0); }, 10000);
+ p.stdout.on('data', (d) => { out += d; });
+ p.on('close', () => {
+ clearTimeout(to);
+ // The interval can yield a few packets; the last non-empty one wins.
+ const last = String(out).trim().split(/\s+/).filter(Boolean).pop();
+ const v = parseFloat(last);
+ resolve(Number.isFinite(v) && v > 0 ? v : 0);
+ });
+ p.on('error', () => { clearTimeout(to); resolve(0); });
+ });
+}
+
/**
/**
* Extract a thumbnail frame from a video file using ffmpeg.
* Extract a thumbnail frame from a video file using ffmpeg.
* @param {string} videoPath – Absolute path (or presigned URL) of the video
* @param {string} videoPath – Absolute path (or presigned URL) of the video
* @param {string} prefix – Filename prefix ('vod' or 'clip')
* @param {string} prefix – Filename prefix ('vod' or 'clip')
* @param {number} entityId – DB id (vod or clip id)
* @param {number} entityId – DB id (vod or clip id)
- * @param {object} opts – { seekPercent?, seekSeconds? }
+ * @param {object} opts – { seekPercent?, seekSeconds?, liveEdge? }
+ * liveEdge: the file is an in-progress recording — grab the newest frame
+ * available instead of a percentage of a duration the header does not have.
* @returns {Promise<string|null>} public thumbnail URL (/t/<file>), or null
* @returns {Promise<string|null>} public thumbnail URL (/t/<file>), or null
*/
*/
function generateFromVideo(videoPath, prefix, entityId, opts = {}) {
function generateFromVideo(videoPath, prefix, entityId, opts = {}) {
@@ -68,20 +117,39 @@ function generateFromVideo(videoPath, prefix, entityId, opts = {}) {
@@ -68,20 +117,39 @@ function generateFromVideo(videoPath, prefix, entityId, opts = {}) {
let probeData = '';
let probeData = '';
probe.stdout.on('data', (d) => (probeData += d));
probe.stdout.on('data', (d) => (probeData += d));
- probe.on('close', (probeCode) => {
+ probe.on('close', async (probeCode) => {
let seekTime = opts.seekSeconds || 1;
let seekTime = opts.seekSeconds || 1;
- if (probeCode === 0 && !opts.seekSeconds) {
- try {
- const info = JSON.parse(probeData);
- const duration = parseFloat(info.format?.duration || '0');
- if (duration > 2) {
- seekTime = Math.min(
- duration * ((opts.seekPercent || 10) / 100),
- duration - 0.5
- );
- }
- } catch {}
+ if (!opts.seekSeconds) {
+ let duration = 0;
+ if (probeCode === 0) {
+ try {
+ const info = JSON.parse(probeData);
+ // parseFloat('N/A') is NaN — normalize so the checks below
+ // treat "no usable duration" as 0 rather than silently
+ // falling through to the default 1-second seek.
+ const d = parseFloat(info.format?.duration);
+ if (Number.isFinite(d)) duration = d;
+ } catch { /* */ }
+ }
+
+ // A still-recording file has no header duration; find the real
+ // tail position from the last video packet instead.
+ if (opts.liveEdge || duration <= 0) {
+ const lastPts = await _probeLastPts(videoPath);
+ if (lastPts > 0) duration = lastPts;
+ }
+
+ if (opts.liveEdge) {
+ // Newest frame that is safely on disk. Falls back to the
+ // start only when the tail could not be located at all.
+ seekTime = duration > 0 ? Math.max(0, duration - LIVE_EDGE_BACKOFF_SEC) : 0;
+ } else if (duration > 2) {
+ seekTime = Math.min(
+ duration * ((opts.seekPercent || 10) / 100),
+ duration - 0.5
+ );
+ }
}
}
// Extract frame
// Extract frame
@@ -112,9 +180,18 @@ function generateFromVideo(videoPath, prefix, entityId, opts = {}) {
@@ -112,9 +180,18 @@ function generateFromVideo(videoPath, prefix, entityId, opts = {}) {
}).finally(() => _activeThumbJobs.delete(jobKey));
}).finally(() => _activeThumbJobs.delete(jobKey));
}
}
-// ── Generate VOD Thumbnail (~10% in) ─────────────────────────
-async function generateVodThumbnail(vodId, filePath) {
- const thumbUrl = await generateFromVideo(filePath, 'vod', vodId, { seekPercent: 10 });
+// ── Generate VOD Thumbnail (~10% in, or the live edge while recording) ───────
+// While a stream is live, OpenVibe.Live regenerates this VOD's thumbnail every
+// ~2 min and uses it as the stream's card image. "10% in" is the wrong frame for
+// that: on a growing file the header carries no duration, so every regeneration
+// produced the identical 1-second frame and the card looked frozen for the whole
+// broadcast. An in-progress recording therefore grabs the live edge instead.
+async function generateVodThumbnail(vodId, filePath, opts = {}) {
+ const isRecording = opts.liveEdge != null
+ ? !!opts.liveEdge
+ : !!db.get('SELECT is_recording FROM vods WHERE id = ?', [vodId])?.is_recording;
+ const thumbUrl = await generateFromVideo(filePath, 'vod', vodId,
+ isRecording ? { liveEdge: true } : { seekPercent: 10 });
if (thumbUrl) {
if (thumbUrl) {
// Clean up the previous thumbnail for this VOD
// Clean up the previous thumbnail for this VOD
_removeOldThumb(db.get('SELECT thumbnail_url FROM vods WHERE id = ?', [vodId])?.thumbnail_url, thumbUrl);
_removeOldThumb(db.get('SELECT thumbnail_url FROM vods WHERE id = ?', [vodId])?.thumbnail_url, thumbUrl);