openvibers/openvibe.live Commits

PIPs in VODs: while a stream is live, when each PIP was on screen is kept (stream_history kind 'pip'); GET /api/vods/:id answers pips, each its stream's public VOD in both videos' seconds (#392)

OP OpenVibers committed 17 days ago · 3f6abf42dd · 4 files, +240 −0

Changes

ddiff --git a/server/index.js b/server/index.js
iindex ce25972b..46cf0646 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -1460,6 +1460,8 @@ async function start() {
try { require('./ai/clip-title').start(); } catch (e) { console.warn('[AI] clip-title job not started:', e.message); }
// A stream recorded in parts is one video on Media (streaming/recordings.js): synced, closed, backfilled.
try { require('./streaming/recordings').start(); } catch (e) { console.warn('[Recordings] sweep not started:', e.message); }
++ // When each picture-in-picture was on screen, so a VOD draws it again (streaming/pip-history.js).
++ try { require('./streaming/pip-history').start(); } catch (e) { console.warn('[PiP history] not started:', e.message); }
try { require('./ai/easter-egg-job').start(); } catch (e) { console.warn('[AI] easter-egg job not started:', e.message); }
// The Rundown's desk replaced the Arena's jobs (the drama desk, the beef listener and judges, 2026-10-19).
try { require('./rundown/desk').start(); } catch (e) { console.warn('[Rundown] desk not started:', e.message); }
ddiff --git a/server/media-proxy/vods.js b/server/media-proxy/vods.js
iindex 146ba7dc..6e283938 100644
---- a/server/media-proxy/vods.js
++++ b/server/media-proxy/vods.js
@@@ -588,6 +588,8 @@ router.get('/:id', optionalAuth, async (req, res) => {
// One video over a stream recorded in parts (streaming/recordings.js; OpenVibe.Media docs/recordings.md).
try { await presentRecording(vod); } catch (err) { console.warn('[VOD] recording:', err.message); }
++ // The streams that were drawn over it, to play over it again (streaming/pip-history.js).
++ try { vod.pips = await pipsForVod(vod, stream); } catch (err) { console.warn('[VOD] pips:', err.message); vod.pips = []; }
const clips = co?.clips || (Array.isArray(co) ? co : []);
try {
@@@ -932,6 +934,66 @@ async function clipPlace(stream, startTime, endTime) {
* Until then, and for every later part, it answers as a part with `recording` naming the whole (`vod_id` of its first
* part and `start_sec`, where this part starts in it), so a page can send people to the whole video at that second.
*/
++/**
++ * The picture-in-pictures that were on screen during this VOD (streaming/pip-history.js), each as its stream's public
++ * VOD to play over this one (owner, 2026-10-26: "the VODs system should support and handle the PIPs a streamer had
++ * configured during that vod"): [{ vod_id, title, username, from, to, offset, x, y, w, muted }]. `from` and `to` are
++ * this VOD's seconds (its whole recording's, when it answers for one), `offset` the PIP VOD's second at `from`, so a
++ * player shows it while from <= t < to at offset + (t - from). A PIP whose stream has no public recording is left out.
++ */
++async function pipsForVod(vod, stream) {
++ if (!vod.stream_id || !stream) return [];
++ const spans = await require('../streaming/pip-history').spansOf(vod.stream_id);
++ if (!spans.length) return [];
++ const history = require('../streaming/stream-history');
++ const recordings = require('../streaming/recordings');
++ const hostStart = history.epochMs(stream.started_at);
++ if (hostStart == null) return [];
++ const whole = vod.recording && vod.recording.joined && Number(vod.recording.vod_id) === Number(vod.id) ? vod.recording.parts : null;
++ const hostTime = (sec) => (whole ? recordings.toRecordingTime(whole, sec) : sec - (Number(vod.stream_offset_sec) || 0));
++ const duration = Number(vod.duration_seconds) || Infinity;
++ const vods = await require('./lookups').publicVodsForStreams(spans.map((sp) => sp.source_stream_id), { budgetMs: 3000 });
++ // Each source stream once: when it started, and its stream seconds → its VOD's seconds (whole recording or one part).
++ const sources = new Map();
++ const sourceOf = async (streamId, sv) => {
++ if (sources.has(streamId)) return sources.get(streamId);
++ let out = null;
++ try {
++ const s = await db.getStreamById(streamId);
++ const startMs = history.epochMs(s && s.started_at);
++ if (startMs != null) {
++ if (sv.recording && Number(sv.recording.parts) > 1) {
++ const parts = await recordings.presentParts(await media.getRecording(recordings.keyOf(streamId)));
++ out = { startMs, toVod: (sec) => recordings.toRecordingTime(parts, sec) };
++ } else {
++ const part = await history.partForVod(sv.id).catch(() => null);
++ const at = part ? Number(part.at) || 0 : 0;
++ out = { startMs, toVod: (sec) => sec - at };
++ }
++ }
++ } catch { out = null; }
++ sources.set(streamId, out);
++ return out;
++ };
++ const round = (n) => Math.round(n * 100) / 100;
++ const out = [];
++ for (const sp of spans) {
++ const sv = vods.get(sp.source_stream_id);
++ if (!sv) continue;
++ const start = hostTime(sp.at), end = hostTime(sp.end);
++ if (start == null || end == null) continue;
++ const from = Math.max(0, start), to = Math.min(duration, end);
++ if (to - from < 1) continue;
++ const src = await sourceOf(sp.source_stream_id, sv);
++ const atSource = src && src.toVod((hostStart + sp.at * 1000 - src.startMs) / 1000);
++ if (atSource == null) continue;
++ out.push({ vod_id: Number(sv.id), title: sp.title || sv.title || null, username: sp.username || sv.username || null,
++ from: round(from), to: round(to), offset: round(Math.max(0, atSource + (from - start))),
++ x: sp.x, y: sp.y, w: sp.w, muted: sp.muted });
++ }
++ return out.slice(0, 200);
++}
++
async function presentRecording(vod) {
const rec = vod.recording;
if (!rec || !Array.isArray(rec.parts) || rec.parts.length < 2) { delete vod.recording; return; }
@@@ -1017,3 +1079,4 @@ router.post('/clips/:id/trim', requireAuth, (req, res) => {
module.exports = router;
module.exports.isGhostVod = isGhostVod;
module.exports.chunkOpensVod = chunkOpensVod;
++module.exports.pipsForVod = pipsForVod;
ddiff --git a/server/streaming/pip-history.js b/server/streaming/pip-history.js
nnew file mode 100644
iindex 00000000..70a39611
---- /dev/null
++++ b/server/streaming/pip-history.js
@@@ -0,0 +1,90 @@
++/**
++ * pip-history.js — when each picture-in-picture was on screen during a live stream, so its VOD can draw it again
++ * (owner, 2026-10-26: "the VODs system should support and handle the PIPs a streamer had configured during that vod").
++ *
++ * Every TICK_MS, for each live stream whose slot lists PIPs (stream_pips), each PIP whose stream is live is a span in
++ * stream_history, kind 'pip', in the host stream's time:
++ * at_sec when it came on screen: its stream's start, or the tick before it was first seen (a PIP added mid-stream)
++ * end_sec the last tick it was seen (a span stops growing when its stream ends or the streamer removes it)
++ * detail { source_stream_id, source_msid, title, username, x, y, w, muted }: what to draw and where it started
++ * A span grows while the same stream stays in the same place; a new stream of the source (a restart) or a new place
++ * starts a new span. Server/media-proxy/vods.js turns a VOD's spans into the PIP VODs to play over it.
++ */
++'use strict';
++
++const db = require('../db/database');
++const history = require('./stream-history');
++
++const TICK_MS = 30 * 1000;
++let started = false;
++
++const near = (a, b) => Math.abs(Number(a) - Number(b)) < 0.005;
++const parse = (s) => { try { return JSON.parse(s || '{}'); } catch { return {}; } };
++
++/** One pass: extend or open a span for each PIP live over each live stream. */
++async function tick(nowMs = Date.now()) {
++ let hosts = [];
++ try {
++ hosts = await db.all(`SELECT s.id, s.started_at, s.managed_stream_id FROM streams s
++ WHERE s.is_live = 1 AND s.managed_stream_id IS NOT NULL
++ AND EXISTS (SELECT 1 FROM stream_pips p WHERE p.managed_stream_id = s.managed_stream_id)`);
++ } catch (err) {
++ console.warn('[PiP history] hosts not read:', err.message);
++ return 0;
++ }
++ let written = 0;
++ for (const host of hosts) {
++ const start = history.epochMs(host.started_at);
++ if (start == null) continue;
++ const nowSec = Math.max(0, Math.round((nowMs - start) / 100) / 10);
++ const live = (await db.getPipsForManagedStream(host.managed_stream_id)).filter((p) => p.live && p.stream_id);
++ if (!live.length) continue;
++ const open = await db.all(`SELECT id, at_sec, end_sec, detail FROM stream_history
++ WHERE stream_id = ? AND kind = 'pip' AND end_sec >= ? ORDER BY id DESC LIMIT 50`, [host.id, nowSec - (TICK_MS * 1.5) / 1000]);
++ for (const pip of live) {
++ const g = pip.defaults || {};
++ const row = open.find((r) => {
++ const d = parse(r.detail);
++ return Number(d.source_stream_id) === pip.stream_id && near(d.x, g.x) && near(d.y, g.y) && near(d.w, g.w);
++ });
++ if (row) {
++ await db.run('UPDATE stream_history SET end_sec = ? WHERE id = ?', [nowSec, row.id]);
++ written += 1;
++ continue;
++ }
++ let sourceStart = null;
++ try { sourceStart = history.epochMs((await db.get('SELECT started_at FROM streams WHERE id = ?', [pip.stream_id]))?.started_at); } catch { /* */ }
++ const sourceSec = sourceStart == null ? 0 : (sourceStart - start) / 1000;
++ const at = Math.max(0, Math.round(Math.max(sourceSec, nowSec - TICK_MS / 1000) * 10) / 10);
++ await db.run(`INSERT INTO stream_history (stream_id, kind, at_sec, end_sec, detail) VALUES (?, 'pip', ?, ?, ?)`, [host.id, Math.min(at, nowSec), nowSec, JSON.stringify({
++ source_stream_id: pip.stream_id, source_msid: pip.source_msid, title: pip.title, username: pip.username,
++ x: g.x, y: g.y, w: g.w, muted: pip.muted,
++ })]);
++ written += 1;
++ }
++ }
++ return written;
++}
++
++/** A stream's PIP spans, oldest first: [{ at, end, source_stream_id, source_msid, title, username, x, y, w, muted }]. */
++async function spansOf(streamId) {
++ let rows = [];
++ try { rows = await db.all(`SELECT at_sec, end_sec, detail FROM stream_history WHERE stream_id = ? AND kind = 'pip' ORDER BY at_sec ASC, id ASC LIMIT 500`, [streamId]); } catch { return []; }
++ return rows.map((r) => {
++ const d = parse(r.detail);
++ return {
++ at: Number(r.at_sec), end: r.end_sec == null ? Number(r.at_sec) : Number(r.end_sec),
++ source_stream_id: Number(d.source_stream_id) || null, source_msid: Number(d.source_msid) || null,
++ title: d.title || null, username: d.username || null,
++ x: Number(d.x), y: Number(d.y), w: Number(d.w), muted: d.muted !== false,
++ };
++ }).filter((s) => s.source_stream_id && s.end > s.at);
++}
++
++function start() {
++ if (started || require('../drill').enabled) return;
++ started = true;
++ require('../utils/jobs').every('pip-history', TICK_MS, () => tick(), { initialDelayMs: 20 * 1000, jitterMs: 2000 });
++}
++
++module.exports = { tick, spansOf, start, TICK_MS };
ddiff --git a/test/pip-vods.test.js b/test/pip-vods.test.js
nnew file mode 100644
iindex 00000000..55dedf19
---- /dev/null
++++ b/test/pip-vods.test.js
@@@ -0,0 +1,85 @@
++'use strict';
++/**
++ * PIPs in VODs (owner, 2026-10-26: "the VODs system should support and handle the PIPs a streamer had configured during
++ * that vod"). While a stream is live, server/streaming/pip-history.js keeps when each PIP was on screen (stream_history,
++ * kind 'pip'): a span from when its stream started (or the tick before it was seen) to the last tick it was seen, a new
++ * span for a new place or a restarted source. GET /api/vods/:id answers `pips`: each span as the PIP stream's public VOD,
++ * in this VOD's seconds (`from`, `to`) and the PIP VOD's (`offset` at `from`), for openvibe.video to play over it.
++ *
++ * node --import ./test/helpers/pg-preload.mjs test/pip-vods.test.js
++ */
++const assert = require('assert');
++
++process.env.NODE_ENV = 'test';
++console.warn = () => {};
++
++const db = require('../server/db/database');
++const pipHistory = require('../server/streaming/pip-history');
++const lookups = require('../server/media-proxy/lookups');
++const media = require('../server/media-client');
++
++const at = (ms) => new Date(ms).toISOString().replace('T', ' ').slice(0, 19);
++
++(async () => {
++ await db.initDb();
++ await db.run("INSERT INTO users (id, username, password_hash) OVERRIDING SYSTEM VALUE VALUES (1, 'host', 'x'), (2, 'cohost', 'x')");
++ await db.run(`INSERT INTO managed_streams (id, user_id, title, stream_key) OVERRIDING SYSTEM VALUE VALUES
++ (10, 1, 'Screen', 'k10'), (11, 1, 'Webcam', 'k11'), (12, 2, 'Co-host', 'k12')`);
++ await db.setPipsForManagedStream(10, [{ source_msid: 11, x: 0.73, y: 0.7, w: 0.25 }, { source_msid: 12, x: 0.02, y: 0.02, w: 0.18 }]);
++ const T0 = Date.UTC(2026, 9, 27, 12, 0, 0);
++ await db.run('INSERT INTO streams (id, user_id, managed_stream_id, is_live, started_at) OVERRIDING SYSTEM VALUE VALUES (600, 1, 10, 1, ?)', [at(T0)]);
++ // The webcam starts two minutes in; the co-host is offline for now.
++ await db.run('INSERT INTO streams (id, user_id, managed_stream_id, is_live, started_at) OVERRIDING SYSTEM VALUE VALUES (601, 1, 11, 1, ?)', [at(T0 + 120000)]);
++
++ // ── the recorder ──
++ assert.strictEqual(await pipHistory.tick(T0 + 130000), 1, 'one live PIP: one span');
++ await pipHistory.tick(T0 + 160000);
++ let spans = await pipHistory.spansOf(600);
++ assert.deepStrictEqual(spans.map((s) => [s.at, s.end, s.source_stream_id, s.title]), [[120, 160, 601, 'Webcam']], 'from the webcam\'s start, grown by the next tick');
++ // The streamer moves it: a new span from the tick before.
++ await db.setPipsForManagedStream(10, [{ source_msid: 11, x: 0.02, y: 0.7, w: 0.25 }, { source_msid: 12, x: 0.02, y: 0.02, w: 0.18 }]);
++ await pipHistory.tick(T0 + 190000);
++ // The co-host goes live (a PIP added or a source starting mid-stream starts at its own start).
++ await db.run('INSERT INTO streams (id, user_id, managed_stream_id, is_live, started_at) OVERRIDING SYSTEM VALUE VALUES (602, 2, 12, 1, ?)', [at(T0 + 200000)]);
++ await pipHistory.tick(T0 + 220000);
++ // The webcam ends: its span stops growing.
++ await db.run('UPDATE streams SET is_live = 0 WHERE id = 601');
++ await pipHistory.tick(T0 + 250000);
++ spans = await pipHistory.spansOf(600);
++ assert.deepStrictEqual(spans.map((s) => [s.at, s.end, s.source_stream_id, s.x]),
++ [[120, 160, 601, 0.73], [160, 220, 601, 0.02], [200, 250, 602, 0.02]], 'a new place is a new span; an ended source stops growing');
++ assert.strictEqual(await pipHistory.tick(T0 + 280000), 1, 'the co-host is still on screen at 280 s');
++ await db.run('UPDATE streams SET is_live = 0 WHERE id = 600');
++ assert.strictEqual(await pipHistory.tick(T0 + 310000), 0, 'nothing is written for a stream that is not live');
++ console.log('ok: the recorder keeps when each PIP was on screen, a span per place and per source stream');
++
++ // ── the VOD's pips ──
++ const { pipsForVod } = require('../server/media-proxy/vods');
++ // The webcam has a public one-part VOD; the co-host's recording is two parts (Media joins them), the second from 260 s.
++ lookups.publicVodsForStreams = async (ids) => new Map(ids.filter((id) => id !== 999).map((id) => [id, id === 601 ? { id: 9001, title: 'Webcam' } : { id: 9002, title: 'Co-host', recording: { parts: 2 } }]));
++ media.getRecording = async (key) => {
++ assert.strictEqual(key, 'stream:602');
++ return { id: 'rec_1', vod_id: 9002, duration_ms: 400000, parts: [
++ { vod_id: 9002, start_ms: 0, duration_ms: 50000, stream_offset_ms: 0 },
++ { vod_id: 9003, start_ms: 50000, duration_ms: 350000, stream_offset_ms: 60000 }] };
++ };
++ const host = await db.getStreamById(600);
++ // A VOD of the host's second part, from stream second 100 (300 s long).
++ const part = await pipsForVod({ id: 9000, stream_id: 600, duration_seconds: 300, stream_offset_sec: 100 }, host);
++ assert.deepStrictEqual(part.map((p) => [p.vod_id, p.from, p.to, p.offset, p.x]),
++ [[9001, 20, 60, 0, 0.73], [9001, 60, 120, 40, 0.02], [9002, 100, 180, 0, 0.02]],
++ "in this VOD's seconds; each PIP VOD at its own second (the webcam 40 s in when it moved)");
++ // A VOD that starts after the webcam span began: clipped to its start, the PIP VOD's offset moved with it.
++ const late = await pipsForVod({ id: 9004, stream_id: 600, duration_seconds: 30, stream_offset_sec: 140 }, host);
++ assert.deepStrictEqual(late.map((p) => [p.vod_id, p.from, p.to, p.offset]), [[9001, 0, 20, 20], [9001, 20, 30, 40]], 'clipped to the VOD, offsets moved with it');
++ // The whole recording of the host (its first part answers for it): recording time, from the parts.
++ const whole = await pipsForVod({ id: 9010, stream_id: 600, duration_seconds: 500, recording: { joined: true, vod_id: 9010,
++ parts: [{ vod_id: 9010, start_sec: 0, duration_sec: 90, stream_offset_sec: 0 }, { vod_id: 9011, start_sec: 90, duration_sec: 400, stream_offset_sec: 100 }] } }, host);
++ assert.deepStrictEqual(whole.map((p) => [p.from, p.to]), [[110, 150], [150, 210], [190, 270]], 'in the recording\'s time (stream 120 → 90 + 20)');
++ // No public recording for a source: that PIP is left out.
++ lookups.publicVodsForStreams = async () => new Map();
++ assert.deepStrictEqual(await pipsForVod({ id: 9000, stream_id: 600, duration_seconds: 300, stream_offset_sec: 100 }, host), []);
++ assert.deepStrictEqual(await pipsForVod({ id: 9100, stream_id: null }, null), [], 'a VOD with no stream has none');
++ console.log("ok: a VOD's pips are its PIP streams' public VODs, in both videos' seconds, for a part, a late part and a whole recording");
++ process.exit(0);
++})().catch((e) => { console.error(e); process.exit(1); });