openvibers/openvibe.mediahub Commits

A record-only recording from OpenRestream becomes its owner's private video (plan D1) (#114)

OP OpenVibers committed 21 days ago · 03cfbc3cea · 7 files, +155 −4

Changes

ddiff --git a/migrations/0018_upload_vod_unique.sql b/migrations/0018_upload_vod_unique.sql
nnew file mode 100644
iindex 0000000..c710e39
---- /dev/null
++++ b/migrations/0018_upload_vod_unique.sql
@@@ -0,0 +1,5 @@
++-- phase: expand
++-- One upload per Media VOD. Recordings OpenRestream makes for a record-only stream (plan 2026-10-22 §3 D1) are filed here
++-- when Media says they are ready, and that delivery can come twice; the index makes the second one change nothing.
++-- Production had 8 uploads and 8 distinct VODs when this was added (2026-10-23).
++CREATE UNIQUE INDEX IF NOT EXISTS video_uploads_vod ON video_uploads (media_vod_id) WHERE media_vod_id IS NOT NULL;
ddiff --git a/server/account.js b/server/account.js
iindex 2f373fd..5f209ae 100644
---- a/server/account.js
++++ b/server/account.js
@@@ -53,7 +53,7 @@ const TABLES = [
{ table: 'mh_sync_keys', subject: 'owner', value: AS_REQUESTER, file: 'sync_keys.json', columns: ['id', 'name', 'prefix', 'created_at', 'last_used_at', 'revoked_at'] },
];
--function createAccount({ db, config, media, log = console }) {
++function createAccount({ db, config, media, log = console, onMediaVodReady = null }) {
/** The object ids of a deletion, read before the rows go (extraErase runs after them). */
let pendingObjects = [];
let pendingVods = [];
@@@ -108,6 +108,13 @@ function createAccount({ db, config, media, log = console }) {
/** One envelope → an outcome (see openvibe-sdk/account-data); throws so Events redelivers. */
async function apply(ev, { send }) {
++ // Media's ready VODs: a record-only recording OpenRestream made in this tenant becomes its owner's private video
++ // (server/video/uploads.js importRecording). Every other VOD is acknowledged and left alone.
++ if (ev && ev.event_type === 'media.vod.ready') {
++ if (ev.source !== 'media' || !onMediaVodReady) return 'ignored:source';
++ const row = await onMediaVodReady(ev.payload && typeof ev.payload === 'object' ? ev.payload : {});
++ return row ? 'filed' : 'ignored:not_record_only';
++ }
if (ev && ev.event_type === 'network.account.deleted' && ev.source === 'network') {
const payload = ev.payload && typeof ev.payload === 'object' ? ev.payload : {};
// Read the media objects first: the erase deletes the rows that name them.
ddiff --git a/server/app.js b/server/app.js
iindex fc12c65..c97b196 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -69,7 +69,8 @@ async function createApp(opts = {}) {
const identity = createIdentity({ config, fetchImpl, log });
const drive = createService({ s, media, identity, config, log });
const uploads = createUploads({ s, media, config, service: drive, log });
-- const account = createAccount({ db: s.db, config, media, log });
++ // videoUploads is made below; the account consumer only calls it when an event arrives.
++ const account = createAccount({ db: s.db, config, media, log, onMediaVodReady: (vod) => videoUploads.importRecording(vod) });
const playlists = createPlaylists({ s, config, log });
// MediaHub's own events (server/events-outbox.js): written with the change, relayed to OpenVibe.Events.
const eventsOutbox = require('./events-outbox').createEventsOutbox({ s, config, log, fetchImpl: opts.fetchImpl || null });
ddiff --git a/server/events-consumer.js b/server/events-consumer.js
iindex bc2fe07..7827d55 100644
---- a/server/events-consumer.js
++++ b/server/events-consumer.js
@@@ -12,7 +12,10 @@
* scope events.subscription.manage.
*/
const { serviceAuth } = require('openvibe-contracts');
--const { TOPICS } = require('openvibe-sdk/account-data');
++const { TOPICS: ACCOUNT_TOPICS } = require('openvibe-sdk/account-data');
++// Beside account export and deletion: Media's ready VODs, for the recordings of OpenRestream's record-only streams
++// (server/video/uploads.js importRecording; plan 2026-10-22 §3 D1).
++const TOPICS = Object.freeze([...ACCOUNT_TOPICS, 'media.vod.ready']);
/** Create any missing subscription for TOPICS at Events (idempotent; retried in the background at boot). */
function startSubscriptions({ config, port, secret, eventsUrl = config.events.url, fetchImpl = globalThis.fetch, log = console }) {
ddiff --git a/server/video/upload-routes.js b/server/video/upload-routes.js
iindex a679419..92bfec5 100644
---- a/server/video/upload-routes.js
++++ b/server/video/upload-routes.js
@@@ -215,6 +215,8 @@ ${cuts.length ? html`<ul class="ovv-cuts">${cuts.map(cutLine)}</ul>` : ''}
}, 400));
r.get('/me/videos', mustSignIn('/me/videos', 'Sign in to see the videos you uploaded.'), async (req, res) => {
++ // A record-only recording arrives without its owner's channel name: it is filled in the first time they look.
++ if (req.viewer && req.viewer.username) await up.fillUsername(ownerOf(req), req.viewer.username).catch(() => 0);
const rows = await up.mine(ownerOf(req));
await Promise.all(rows.filter((x) => x.status === 'processing').map((x) => up.refresh(x).catch(() => null)));
const fresh = await up.mine(ownerOf(req));
ddiff --git a/server/video/uploads.js b/server/video/uploads.js
iindex 3fc8bf9..06651b0 100644
---- a/server/video/uploads.js
++++ b/server/video/uploads.js
@@@ -13,6 +13,8 @@
* its size is not known ahead, so parts_expected stays 0 until complete()
* chunk(row, n, buffer) the n-th piece of a recording (any size up to 9 MB), in order; the size and the person's
* quota are checked as it grows
++ * importRecording(vod) a recording OpenRestream made for a record-only stream, filed as its owner's private video
++ * when Media says it is ready (media.vod.ready; plan 2026-10-22 §3 D1)
*
* Rules: MP4 (QuickTime and M4V are the same container and go to Media as MP4), WebM and Matroska; at most
* VIDEO_UPLOAD_MAX_BYTES (2 GB) a file, VIDEO_UPLOAD_QUOTA_BYTES (20 GB) a person, VIDEO_UPLOADS_PER_DAY (10) a day.
@@@ -182,6 +184,40 @@ function createUploads({ s, media, config, log = console, events = null }) {
return now;
}
++ /**
++ * A recording OpenRestream made for a record-only stream ("Record only, to my videos" on openre.stream): Media's VOD
++ * in this tenant, ready, becomes its owner's private video here, like a browser recording, with the AI edit on its
++ * page. Only a VOD of this tenant whose meta says so is taken (source openre, record_only, the owner's subject):
++ * MediaHub and OpenRestream are the only services that may write to this tenant. A VOD already filed changes
++ * nothing (one upload per VOD, migration 0018). It counts toward the person's storage like any video, but is never
++ * refused for it: the session was already recorded. → the row, or null when the VOD is not such a recording.
++ */
++ async function importRecording(vod) {
++ const meta = vod && vod.meta && typeof vod.meta === 'object' ? vod.meta : {};
++ if (!vod || String(vod.app_id) !== String(config.media.app) || meta.source !== 'openre' || meta.record_only !== true) return null;
++ const subject = String(meta.owner_subject || '');
++ const vodId = Number(vod.id);
++ if (!/^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(subject) || !Number.isSafeInteger(vodId) || vodId < 1) return null;
++ const known = await s.db.maybe('SELECT * FROM video_uploads WHERE media_vod_id = $1', [vodId]);
++ if (known) return known;
++ const now = s.iso();
++ const title = cleanLine(vod.title, 120) || `Recording, ${now.slice(0, 16).replace('T', ' ')} UTC`;
++ const size = Number(vod.file_size) > 0 ? Math.round(Number(vod.file_size)) : 0;
++ const row = await s.db.maybe(`INSERT INTO video_uploads (id, owner, owner_username, title, description, visibility, media_vod_id, content_type, size_bytes, part_size, parts_expected, status, created_at, updated_at)
++ VALUES ($1, $2, NULL, $3, '', 'private', $4, 'video/mp4', $5, 0, 0, 'processing', $6, $6)
++ ON CONFLICT (media_vod_id) WHERE media_vod_id IS NOT NULL DO NOTHING RETURNING *`,
++ [s.newId('vup'), `user:${subject}`, title, vodId, size, now]);
++ if (!row) return await s.db.maybe('SELECT * FROM video_uploads WHERE media_vod_id = $1', [vodId]);
++ log.log(`[MediaHub] a record-only recording is ${row.id} (Media VOD ${vodId})`);
++ return await absorb(row, vod);
++ }
++
++ /** The owner's channel name on rows filed without one (record-only recordings), once they are signed in to see them. */
++ async function fillUsername(owner, username) {
++ if (!owner || !username) return 0;
++ return await s.db.exec('UPDATE video_uploads SET owner_username = $2 WHERE owner = $1 AND owner_username IS NULL', [owner, String(username).slice(0, 64)]);
++ }
++
/** A row still processing: ask Media how it is going (no more than once every 15 seconds per row). */
const lastAsked = new Map();
async function refresh(row) {
@@@ -224,7 +260,7 @@ function createUploads({ s, media, config, log = console, events = null }) {
return {
PART_SIZE, REASONS, UploadError, kindOf, addressId, canSee,
-- byId, bySeq, start, part, complete, refresh, update, remove, report, startRecording, chunk,
++ byId, bySeq, start, part, complete, refresh, update, remove, report, startRecording, chunk, importRecording, fillUsername,
mine: (owner, limit = 60) => s.db.many("SELECT * FROM video_uploads WHERE owner = $1 AND state <> 'removed' ORDER BY id DESC LIMIT $2", [owner, limit]),
latest: (limit = 12) => s.db.many("SELECT * FROM video_uploads WHERE visibility = 'public' AND status = 'ready' AND state = 'active' ORDER BY id DESC LIMIT $1", [limit]),
channel: (username, limit = 48) => s.db.many("SELECT * FROM video_uploads WHERE lower(owner_username) = lower($1) AND visibility = 'public' AND status = 'ready' AND state = 'active' ORDER BY id DESC LIMIT $2", [String(username), limit]),
ddiff --git a/test/record-only-import.test.js b/test/record-only-import.test.js
nnew file mode 100644
iindex 0000000..be75710
---- /dev/null
++++ b/test/record-only-import.test.js
@@@ -0,0 +1,97 @@
++'use strict';
++/**
++ * Record only, to my videos (plan 2026-10-22 §3 D1): OpenRestream records a record-only stream into this tenant as a
++ * private Media VOD whose meta names the owner; when Media says it is ready (media.vod.ready, signed on
++ * POST /internal/events) it becomes that person's private video here (server/video/uploads.js importRecording).
++ * - only a VOD of this tenant marked record-only by OpenRestream, with an owner subject, is filed; anything else,
++ * and anything not from Media, is acknowledged and left alone;
++ * - it is private and ready, plays for its owner only, and a redelivery files nothing twice;
++ * - its owner's channel name is filled in when they first open their videos;
++ * - MediaHub subscribes to media.vod.ready beside the account topics.
++ */
++const assert = require('assert');
++const http = require('http');
++const { signDeliveryHeaders } = require('openvibe-sdk/events');
++const { boot, check, done } = require('./helpers/boot');
++
++const SECRET = 'r'.repeat(48);
++
++function vget(base, path, cookie) {
++ return new Promise((resolve, reject) => {
++ const u = new URL(base + path);
++ const req = http.request({ hostname: u.hostname, port: u.port, path: u.pathname, headers: { Host: 'openvibe.video', ...(cookie ? { cookie } : {}) } }, (res) => {
++ const chunks = [];
++ res.on('data', (c) => chunks.push(c));
++ res.on('end', () => resolve({ status: res.statusCode, text: Buffer.concat(chunks).toString('utf8') }));
++ });
++ req.on('error', reject);
++ req.end();
++ });
++}
++
++(async () => {
++ const t = await boot({ env: { MEDIAHUB_EVENTS_SECRET: SECRET, MEDIAHUB_VIDEO_URL: 'https://openvibe.video' } });
++ const kim = t.network.addUser('kim');
++ const sam = t.network.addUser('sam');
++ const db = t.ctx.s.db;
++
++ // The VOD OpenRestream asked Media for (server/store/recordings.js vodFields), finished.
++ const meta = { source: 'openre', openre_session_id: 'ses_01JABCDEFGHJKMNPQRSTVWXYZ0', openre_stream_id: 'str_01JABCDEFGHJKMNPQRSTVWXYZ0', protocol: 'rtmp', mode: 'vod', record_only: true, owner_subject: kim.subject };
++ t.media.vods.set(900, { id: 900, title: 'Morning take', visibility: 'private', user_id: null, meta, bytes: Buffer.alloc(4096, 1), types: [], status: 'ready' });
++ const ready = (over = {}) => ({
++ id: 900, app_id: 'media-hub', title: 'Morning take', user_id: null, visibility: 'private', is_recording: false,
++ duration_seconds: 42.5, file_size: 4096, file_url: 'https://openvibe.media/v/vod-900.mp4', thumbnail_url: 'https://openvibe.media/t/vod-900.jpg', meta, ...over,
++ });
++ const deliver = (payload, { source = 'media', type = 'media.vod.ready' } = {}) => {
++ const event = { event_id: `evt_${Math.random().toString(16).slice(2)}`, event_type: type, source, at: new Date().toISOString(), payload };
++ const body = Buffer.from(JSON.stringify({ event }));
++ return t.get('/internal/events', { method: 'POST', body, headers: { 'content-type': 'application/json', ...signDeliveryHeaders(body, SECRET) } });
++ };
++ const rows = () => db.many('SELECT * FROM video_uploads ORDER BY created_at');
++
++ await check('a ready record-only VOD of this tenant becomes its owner\'s private, ready video', async () => {
++ const r = await deliver(ready());
++ assert.strictEqual(r.status, 200, r.text.slice(0, 300));
++ assert.strictEqual(r.json().outcome, 'filed');
++ const [row] = await rows();
++ assert.deepStrictEqual([row.owner, row.visibility, row.status, Number(row.media_vod_id), row.title, Number(row.size_bytes), row.owner_username],
++ [`user:${kim.subject}`, 'private', 'ready', 900, 'Morning take', 4096, null]);
++ assert.strictEqual(Number(row.duration_seconds), 42.5);
++ });
++
++ await check('a redelivery files nothing twice', async () => {
++ assert.strictEqual((await deliver(ready())).json().outcome, 'filed');
++ assert.strictEqual((await rows()).length, 1);
++ });
++
++ await check('any other VOD, or one not from Media, is acknowledged and left alone', async () => {
++ const live = await deliver(ready({ id: 901, app_id: 'live', meta: { source: 'openre', mode: 'vod' } }));
++ assert.deepStrictEqual([live.status, live.json().outcome], [200, 'ignored:not_record_only']);
++ assert.strictEqual((await deliver(ready({ id: 902, meta: { ...meta, record_only: false } }))).json().outcome, 'ignored:not_record_only');
++ assert.strictEqual((await deliver(ready({ id: 903, meta: { source: 'media-hub', record_only: true, owner_subject: kim.subject } }))).json().outcome, 'ignored:not_record_only');
++ assert.strictEqual((await deliver(ready({ id: 904, meta: { ...meta, owner_subject: 'kim' } }))).json().outcome, 'ignored:not_record_only');
++ assert.strictEqual((await deliver(ready({ id: 905 }), { source: 'network' })).json().outcome, 'ignored:source');
++ assert.strictEqual((await rows()).length, 1);
++ });
++
++ await check('it plays for its owner only, and their channel name is filled in when they open their videos', async () => {
++ const [row] = await rows();
++ assert.strictEqual((await vget(t.base, `/u/${row.id}`)).status, 404, 'private: not for the signed-out');
++ assert.strictEqual((await vget(t.base, `/u/${row.id}`, t.signIn(sam))).status, 404, 'nor for anyone else');
++ const page = await vget(t.base, `/u/${row.id}`, t.signIn(kim));
++ assert.strictEqual(page.status, 200);
++ assert.match(page.text, /Morning take/);
++ const mine = await vget(t.base, '/me/videos', t.signIn(kim));
++ assert.match(mine.text, new RegExp(`/u/${row.id}`));
++ assert.strictEqual((await db.maybe('SELECT owner_username FROM video_uploads WHERE id = $1', [row.id])).owner_username, 'kim');
++ });
++
++ await check('MediaHub subscribes to media.vod.ready beside the account topics', async () => {
++ const { TOPICS } = require('../server/events-consumer');
++ assert.ok(TOPICS.includes('media.vod.ready'));
++ assert.ok(TOPICS.includes('network.account.deleted'));
++ });
++
++ await t.close();
++ done();
++})().catch((err) => { console.error(err); process.exit(1); });