openvibers/openrestream Commits
Migration from a Live DB snapshot (new keys only, held/excluded rows recorded, idempotent) with the per-channel cutover checklist; recording, relay, destination and UI tests
Changes
nnew file mode 100644@@@ -0,0 +1,48 @@++#!/usr/bin/env node++'use strict';++/**++ * Import Live's stream slots and restream destinations into OpenRe, from a read-only snapshot of++ * Live's database, and print the per-channel RTMP cutover checklist.++ *++ * node scripts/migrate-from-live.js --live-db /path/to/live-snapshot.db # dry run++ * node scripts/migrate-from-live.js --live-db /path/to/live-snapshot.db --apply # write++ * ... [--slots 12,31] [--checklist /path/to/checklist.md]++ *++ * Take the snapshot with `sqlite3 /opt/openvibe.live/data/openvibe.db ".backup /tmp/live-snapshot.db"`++ * (never point this at the live file). Uses the OpenRe environment (.env or OPENRE_ENV_FILE) for++ * its own database and OPENRE_SECRETS_KEY. Prints no secret: old keys are not imported and the new++ * keys are never shown (the broadcaster rotates to get one). Safe to re-run.++ */++const fs = require('fs');++const path = require('path');++const Database = require('better-sqlite3');++++function arg(name) {++ const i = process.argv.indexOf(`--${name}`);++ return i > 0 ? process.argv[i + 1] : null;++}++++function main() {++ require('dotenv').config({ path: process.env.OPENRE_ENV_FILE || path.join(process.cwd(), '.env') });++ const { load } = require('../server/config');++ const { openRuntime } = require('../server/store');++ const { migrate, checklist } = require('../server/migrate-live');++ const liveDbPath = arg('live-db');++ if (!liveDbPath) { console.error('usage: migrate-from-live.js --live-db <snapshot.db> [--apply] [--slots 1,2] [--checklist out.md]'); process.exit(2); }++ const apply = process.argv.includes('--apply');++ const config = load();++ if (apply && !config.secretsKey) { console.error('OPENRE_SECRETS_KEY is required to import destinations'); process.exit(2); }++ const liveDb = new Database(liveDbPath, { readonly: true, fileMustExist: true });++ const rt = openRuntime({ config, log: { log() {}, warn: console.warn, error: console.error } });++ const onlySlots = arg('slots') ? arg('slots').split(',').map(Number).filter(Number.isFinite) : null;++ const report = migrate({ liveDb, rt, apply, onlySlots });++ const r = config.rtmp;++ const md = checklist(report, { openreUrl: config.baseUrl, rtmpUrl: `rtmp://${r.publicHost}${r.publicPort === 1935 ? '' : `:${r.publicPort}`}/live` });++ const out = arg('checklist');++ if (out) { fs.writeFileSync(out, md + '\n'); console.log(`checklist written to ${out}`); } else console.log(md);++ console.error(`[migrate] ${apply ? 'applied' : 'dry run'}: ${JSON.stringify(report.counts)}`);++ liveDb.close();++ rt.db.close();++}++++if (require.main === module) main();@@@ -37,6 +37,8 @@ CREATE TABLE IF NOT EXISTS stream_definitions (CREATE INDEX IF NOT EXISTS idx_definitions_owner ON stream_definitions(owner_subject, state);-- Typed references to other services' entities (plan §29.3): never a foreign key into them.++-- A reference to a stream-like entity (e.g. live:managed_stream) belongs to one definition; owner++-- references (live:user, live:channel) may repeat across a person's definitions.CREATE TABLE IF NOT EXISTS external_refs (definition_id TEXT NOT NULL REFERENCES stream_definitions(id) ON DELETE CASCADE,service TEXT NOT NULL,@@@ -44,9 +46,9 @@ CREATE TABLE IF NOT EXISTS external_refs (ref_id TEXT NOT NULL,label TEXT,created_at INTEGER NOT NULL,-- PRIMARY KEY (service, type, ref_id)++ PRIMARY KEY (definition_id, service, type, ref_id));--CREATE INDEX IF NOT EXISTS idx_external_refs_definition ON external_refs(definition_id);++CREATE INDEX IF NOT EXISTS idx_external_refs_ref ON external_refs(service, type, ref_id);-- Ingest keys are stored as SHA-256 of a 256-bit random secret; the secret is shown once.CREATE TABLE IF NOT EXISTS ingest_keys (nnew file mode 100644@@@ -0,0 +1,229 @@++'use strict';++/**++ * Import OpenVibe.Live stream slots (managed_streams) and their restream destinations from a++ * read-only snapshot of Live's database into OpenRe (scripts/migrate-from-live.js is the CLI).++ *++ * Rules (ADR-009, plan §8.1):++ * - Live's database is opened read-only; nothing is written to it.++ * - Stream keys are NEVER imported. Every imported slot gets a new OpenRe key that nobody has++ * seen; the streamer gets a usable key by rotating (Live's "Regenerate" once the slot is++ * switched to OpenRe, or openre.stream). Every key that existed before the cutover is dead++ * at the cutover.++ * - Every source row is imported, held with a reason, or excluded with a reason, and recorded in++ * migration_map. Re-running is idempotent: imported rows are skipped, held rows are retried.++ * - Ownership is the canonical subject (linked_accounts.subject_id); Live's integer ids travel++ * only as typed references (live:managed_stream, live:user) for Media's 'live' tenant and the++ * Live mirror.++ */++++const { validateDestinationUrl } = require('./destination-url');++++function columns(db, table) {++ try { return new Set(db.prepare(`PRAGMA table_info(${table})`).all().map(c => c.name)); } catch { return new Set(); }++}++++function recordingModeOf(ms, channel) {++ let vod = !channel ? true : Boolean(channel.vod_recording_enabled) && !channel.force_vod_recording_disabled;++ if (vod && ms.slot_vod_recording_enabled === 0) vod = false;++ if (vod) return 'vod';++ return ms.slot_clip_recording_enabled === 0 ? 'none' : 'clips';++}++++function visibilityOf(ms, channel) {++ const v = ms.default_vod_visibility || (channel && channel.default_vod_visibility) || 'public';++ return ['public', 'unlisted', 'private'].includes(v) ? v : 'public';++}++++/**++ * migrate({ liveDb, rt, apply, onlySlots }) -> report++ * report = { slots: [...], destinations: [...], channels: Map username → {...}, counts }++ */++function migrate({ liveDb, rt, apply = false, onlySlots = null, now = () => Date.now() }) {++ const { db, store, config } = rt;++ const msCols = columns(liveDb, 'managed_streams');++ if (!msCols.size) throw new Error('the Live snapshot has no managed_streams table');++ const rdCols = columns(liveDb, 'restream_destinations');++ const laCols = columns(liveDb, 'linked_accounts');++ const chCols = columns(liveDb, 'channels');++++ const mapGet = db.prepare("SELECT * FROM migration_map WHERE source_system = 'live' AND source_type = ? AND source_id = ?");++ const mapPut = db.prepare(`INSERT INTO migration_map (source_system, source_type, source_id, target_type, target_id, status, reason, imported_at)++ VALUES ('live', @source_type, @source_id, @target_type, @target_id, @status, @reason, @at)++ ON CONFLICT(source_system, source_type, source_id) DO UPDATE SET target_type = @target_type, target_id = @target_id, status = @status, reason = @reason, imported_at = @at`);++ const record = (row) => { if (apply) mapPut.run({ target_type: null, target_id: null, reason: null, ...row, at: now() }); };++++ const slots = liveDb.prepare(`SELECT ms.*, u.username, u.display_name, u.is_banned FROM managed_streams ms JOIN users u ON u.id = ms.user_id ORDER BY ms.user_id, ms.id`).all()++ .filter(ms => !onlySlots || onlySlots.includes(Number(ms.id)));++ const subjectOf = laCols.has('subject_id')++ ? liveDb.prepare("SELECT subject_id FROM linked_accounts WHERE service = 'network' AND user_id = ? AND subject_id IS NOT NULL")++ : null;++ const channelOf = chCols.size ? liveDb.prepare('SELECT * FROM channels WHERE user_id = ?') : null;++ const destsOfSlot = rdCols.has('managed_stream_id') ? liveDb.prepare('SELECT * FROM restream_destinations WHERE managed_stream_id = ? ORDER BY id') : null;++ const unboundOfUser = rdCols.size++ ? liveDb.prepare(`SELECT * FROM restream_destinations WHERE user_id = ? ${rdCols.has('managed_stream_id') ? 'AND managed_stream_id IS NULL' : ''} ORDER BY id`)++ : null;++ const slotsOfUser = liveDb.prepare('SELECT COUNT(*) AS n FROM managed_streams WHERE user_id = ?');++++ const report = { apply, slots: [], destinations: [], channels: new Map(), counts: { imported: 0, held: 0, excluded: 0, skipped: 0 } };++ const channel = (ms) => {++ if (!report.channels.has(ms.user_id)) report.channels.set(ms.user_id, { user_id: ms.user_id, username: ms.username, display_name: ms.display_name, slots: [], notes: [] });++ return report.channels.get(ms.user_id);++ };++ const count = (status) => { report.counts[status] = (report.counts[status] || 0) + 1; };++++ for (const ms of slots) {++ const ch = channel(ms);++ const prior = mapGet.get('managed_stream', String(ms.id));++ const entry = { live_id: ms.id, slug: ms.slug || null, title: ms.title, protocol: ms.protocol, streaming_method: ms.streaming_method || null, definition_id: null, status: null, reason: null, destinations: [] };++ ch.slots.push(entry);++++ if (prior && prior.status === 'imported' && store.definitions.get(prior.target_id)) {++ entry.definition_id = prior.target_id;++ entry.status = 'imported';++ entry.reason = 'already imported';++ count('skipped');++ } else {++ const subject = subjectOf ? (subjectOf.get(ms.user_id) || {}).subject_id : null;++ if (!subject) {++ entry.status = 'held';++ entry.reason = `Live user ${ms.user_id} has no canonical subject yet (it is recorded when they next sign in to Live); the slot is imported on a later run`;++ record({ source_type: 'managed_stream', source_id: String(ms.id), status: 'held', reason: entry.reason });++ count('held');++ } else if (ms.is_banned) {++ entry.status = 'excluded';++ entry.reason = 'the Live account is banned';++ record({ source_type: 'managed_stream', source_id: String(ms.id), status: 'excluded', reason: entry.reason });++ count('excluded');++ } else {++ const chRow = channelOf ? channelOf.get(ms.user_id) : null;++ const fields = {++ owner_subject: subject,++ title: ms.title || `${ms.display_name || ms.username}'s stream`,++ description: ms.description || '',++ protocols: ['rtmp'],++ recording_mode: recordingModeOf(ms, chRow),++ recording_visibility: visibilityOf(ms, chRow),++ mirror_to_live: true,++ // live:user is what Media's 'live' tenant files the VOD under (Live-local ids).++ external_refs: [++ { service: 'live', type: 'managed_stream', id: String(ms.id), label: ms.slug || ms.title || null },++ { service: 'live', type: 'user', id: String(ms.user_id), label: ms.username },++ ],++ created_by: 'svc:openre-migration',++ };++ entry.recording_mode = fields.recording_mode;++ if (apply) {++ const existing = store.definitions.findByRef('live', 'managed_stream', String(ms.id));++ const def = existing || store.definitions.create(fields).definition; // the new key is discarded unseen++ entry.definition_id = def.id;++ record({ source_type: 'managed_stream', source_id: String(ms.id), target_type: 'stream_definition', target_id: def.id, status: 'imported', reason: 'new key issued, old key not imported' });++ }++ entry.status = 'imported';++ entry.reason = 'new key issued, old key not imported';++ count('imported');++ }++ }++++ // Destinations: the slot's own, plus unbound ones when the user has exactly one slot.++ const own = destsOfSlot ? destsOfSlot.all(ms.id) : [];++ const unbound = unboundOfUser ? unboundOfUser.all(ms.user_id) : [];++ const oneSlot = slotsOfUser.get(ms.user_id).n === 1;++ const candidates = [...own.map(d => ({ d, unbound: false })), ...unbound.map(d => ({ d, unbound: true }))];++ for (const { d, unbound: isUnbound } of candidates) {++ const di = { live_id: d.id, platform: d.platform, name: d.name || d.platform, status: null, reason: null, destination_id: null, oauth_linked: Boolean(d.connection_id) };++ entry.destinations.push(di);++ report.destinations.push(di);++ const priorD = mapGet.get('restream_destination', String(d.id));++ if (priorD && priorD.target_id && store.outputs.destinationRow(priorD.target_id)) {++ di.status = priorD.status; di.reason = priorD.reason || 'already imported'; di.destination_id = priorD.target_id; count('skipped');++ continue;++ }++ if (isUnbound && !oneSlot) {++ di.status = 'held';++ di.reason = 'not bound to a slot and the user has several slots (Live pushed unbound destinations from every slot); bind it in Live or add it on openre.stream';++ record({ source_type: 'restream_destination', source_id: String(d.id), status: 'held', reason: di.reason });++ count('held');++ continue;++ }++ if (entry.status !== 'imported') {++ di.status = 'held';++ di.reason = `its slot is ${entry.status}`;++ record({ source_type: 'restream_destination', source_id: String(d.id), status: 'held', reason: di.reason });++ count('held');++ continue;++ }++ const v = validateDestinationUrl(d.server_url, { allowPrivate: config.outputs.allowPrivateHosts });++ const hold = !v.ok ? `server URL refused by OpenRe's destination rules: ${v.error}` : null;++ const input = {++ platform: ['youtube', 'twitch', 'kick', 'custom'].includes(d.platform) ? d.platform : 'custom',++ name: d.name || null,++ server_url: v.ok ? v.value : 'rtmp://held.invalid/app',++ stream_key: d.stream_key || '',++ srt_passphrase: d.srt_passphrase || '',++ srt_latency_ms: d.srt_latency_ms == null ? undefined : d.srt_latency_ms,++ enabled: Boolean(d.enabled),++ auto_start: Boolean(d.auto_start),++ quality_preset: ['auto', 'low', 'medium', 'high', 'ultra', 'source'].includes(d.quality_preset) ? d.quality_preset : 'auto',++ custom_video_bitrate: d.custom_video_bitrate == null ? undefined : d.custom_video_bitrate,++ custom_audio_bitrate: d.custom_audio_bitrate == null ? undefined : d.custom_audio_bitrate,++ custom_fps: d.custom_fps == null ? undefined : d.custom_fps,++ custom_encoder_preset: d.custom_encoder_preset || undefined,++ };++ if (apply) {++ let created;++ if (hold) {++ // Keep the (sealed) key so the owner only has to fix the URL; the row stays++ // disabled until they do. The original URL is kept in the hold reason.++ created = store.outputs.createDestination(entry.definition_id, { ...input, server_url: 'rtmp://held.invalid/app', enabled: false }, { hold_reason: `${hold} (was ${String(d.server_url || '').replace(/\/[^/]*$/, '/…')})` });++ } else {++ created = store.outputs.createDestination(entry.definition_id, input);++ }++ di.destination_id = created.id;++ record({ source_type: 'restream_destination', source_id: String(d.id), target_type: 'destination', target_id: created.id, status: hold ? 'held' : 'imported', reason: hold || null });++ }++ di.status = hold ? 'held' : 'imported';++ di.reason = hold;++ count(hold ? 'held' : 'imported');++ }++ if (own.some(d => d.connection_id) || unbound.some(d => d.connection_id)) {++ ch.notes.push('OAuth-linked destination(s): Live refreshed their ingest key (and created the YouTube broadcast) on every go-live; OpenRe pushes to the key stored at import. Check the platform key is a persistent one, or re-enter it on openre.stream after the cutover.');++ }++ }++ return report;++}++++/** Markdown checklist per channel, for the lead and the broadcaster. Contains no secret. */++function checklist(report, { openreUrl = 'https://openre.stream', rtmpUrl } = {}) {++ const lines = [];++ lines.push(`# OpenRe RTMP cutover checklist${report.apply ? '' : ' (dry run: nothing was written)'}`);++ lines.push('');++ lines.push(`Imported ${report.counts.imported}, held ${report.counts.held}, excluded ${report.counts.excluded}, already done ${report.counts.skipped}.`);++ lines.push('');++ for (const ch of report.channels.values()) {++ lines.push(`## @${ch.username} (Live user ${ch.user_id})`);++ lines.push('');++ for (const s of ch.slots) {++ lines.push(`### Slot ${s.live_id}${s.slug ? ` "${s.slug}"` : ''}: ${s.title || ''} — ${s.status}${s.definition_id ? ` → ${s.definition_id}` : ''}`);++ if (s.reason && s.status !== 'imported') lines.push(`- ${s.status}: ${s.reason}`);++ if (s.status === 'imported') {++ if (s.protocol !== 'rtmp' && !['rtmp', 'obs'].includes(String(s.streaming_method || '').toLowerCase())) {++ lines.push(`- Slot protocol is ${s.protocol}${s.streaming_method ? ` / ${s.streaming_method}` : ''}: only RTMP moves to OpenRe; leave this slot on Live unless the broadcaster streams it with OBS/RTMP.`);++ }++ lines.push(`- [ ] Maintenance window agreed with @${ch.username} (the slot must be offline when it is switched).`);++ lines.push(`- [ ] Destinations reviewed on ${openreUrl}/streams/${s.definition_id || '<id>'}: ${s.destinations.length ? s.destinations.map(d => `${d.name} (${d.platform}) ${d.status}${d.reason ? `: ${d.reason}` : ''}`).join('; ') : 'none'}`);++ lines.push(`- [ ] Recording: ${s.recording_mode || 'as imported'} (from the slot's VOD/clip settings on Live).`);++ lines.push(`- [ ] Switch: \`PUT https://openvibe.live/api/admin/openre/managed/${s.live_id}/ingest-authority {"authority":"openre"}\` (rotates Live's own key for the slot at the same time).`);++ lines.push(`- [ ] Broadcaster presses "Regenerate stream key" on Live's Go Live page (or rotates on openre.stream) and pastes the new server${rtmpUrl ? ` \`${rtmpUrl}\`` : ''} and key into OBS. The old key no longer works anywhere.`);++ lines.push('- [ ] Test broadcast: the stream shows on the Live channel page, restreams go live, a VOD appears in Media when it ends.');++ lines.push(`- [ ] Rollback if needed: \`{"authority":"live"}\` on the same route; the broadcaster regenerates the key on Live.`);++ }++ lines.push('');++ }++ lines.push(`- [ ] Personal key: rotate @${ch.username}'s personal stream key on Live (users.stream_key is not migrated; any leaked copy must die too).`);++ for (const n of ch.notes) lines.push(`- Note: ${n}`);++ lines.push('');++ }++ return lines.join('\n');++}++++module.exports = { migrate, checklist, recordingModeOf };@@@ -16,6 +16,8 @@ const RECORDING_MODES = Object.freeze(['vod', 'clips', 'none']);const VISIBILITIES = Object.freeze(['public', 'unlisted', 'private']);const REF_SERVICE_RE = /^[a-z][a-z0-9-]{1,39}$/;const REF_TYPE_RE = /^[a-z][a-z0-9_]{1,39}$/;++// Owner-level reference types may be shared by several definitions of the same person.++const SHARED_REF_TYPES = new Set(['user', 'channel']);class StoreError extends Error {constructor(status, code, detail) { super(detail); this.status = status; this.code = code; }@@@ -56,7 +58,8 @@ function createDefinitions({ db, config, events, clock }) {get: db.prepare('SELECT * FROM stream_definitions WHERE id = ?'),refs: db.prepare('SELECT service, type, ref_id AS id, label FROM external_refs WHERE definition_id = ? ORDER BY created_at, service, type'),insertRef: db.prepare('INSERT INTO external_refs (definition_id, service, type, ref_id, label, created_at) VALUES (?, ?, ?, ?, ?, ?)'),-- byRef: db.prepare('SELECT definition_id FROM external_refs WHERE service = ? AND type = ? AND ref_id = ?'),++ byRef: db.prepare(`SELECT r.definition_id FROM external_refs r JOIN stream_definitions d ON d.id = r.definition_id++ WHERE r.service = ? AND r.type = ? AND r.ref_id = ? ORDER BY (d.state = 'archived'), d.created_at LIMIT 1`),insertKey: db.prepare(`INSERT INTO ingest_keys (id, definition_id, key_hash, hint, status, created_by, created_at)VALUES (?, ?, ?, ?, 'active', ?, ?)`),activeKeys: db.prepare("SELECT * FROM ingest_keys WHERE definition_id = ? AND status = 'active'"),@@@ -141,7 +144,9 @@ function createDefinitions({ db, config, events, clock }) {const id = newId('stream', now());return db.transaction(() => {for (const r of refs) {-- if (q.byRef.get(r.service, r.type, r.id)) throw new StoreError(409, 'openre.ref_taken', `${r.service}:${r.type}:${r.id} already belongs to another stream definition`);++ if (SHARED_REF_TYPES.has(r.type)) continue;++ const taken = q.byRef.get(r.service, r.type, r.id);++ if (taken && row(taken.definition_id).state !== 'archived') throw new StoreError(409, 'openre.ref_taken', `${r.service}:${r.type}:${r.id} already belongs to another stream definition`);}q.insert.run({ id, owner_subject: input.owner_subject, description: '', ...fields, now: now() });for (const r of refs) q.insertRef.run(id, r.service, r.type, r.id, r.label, now());nnew file mode 100644@@@ -0,0 +1,96 @@++'use strict';++// Destination URL rules (SSRF), ffmpeg argument parity with Live's restream manager, and the++// standalone UI (server-rendered, CSRF-protected, key shown once).++const assert = require('assert');++const { validateDestinationUrl, checkResolvedHost, isPrivateAddress } = require('../server/destination-url');++const ff = require('../workers/restream/ffmpeg-args');++const { bootApi, request, userToken, suite, OWNER, OTHER } = require('./helpers');++const { csrfFor } = require('../server/ui/routes');++++const t = suite('destinations-ui');++++t('destination URLs: live-streaming schemes with public hosts only', async () => {++ const ok = (u) => validateDestinationUrl(u).ok;++ assert.ok(ok('rtmp://a.rtmp.youtube.com/live2'));++ assert.ok(ok('rtmps://live.twitch.tv/app'));++ assert.ok(ok('srt://ingest.example.com:9000?latency=200'));++ for (const bad of ['file:///etc/passwd', 'http://example.com/x', '/tmp/out.flv', 'rtmp://user:pass@host/app', 'rtmp://127.0.0.1/app',++ 'rtmp://10.0.0.5/app', 'rtmp://169.254.169.254/latest', 'rtmp://[::1]/app', 'rtmp://localhost/app', 'rtmp://nas.local/app', 'rtmp://host/app x']) {++ assert.ok(!ok(bad), bad);++ }++ assert.ok(validateDestinationUrl('rtmp://127.0.0.1/app', { allowPrivate: true }).ok);++ assert.ok(isPrivateAddress('100.64.1.1') && isPrivateAddress('::ffff:192.168.1.1') && !isPrivateAddress('8.8.8.8'));++ const rebinding = await checkResolvedHost('evil.example', { lookup: async () => [{ address: '8.8.8.8' }, { address: '10.1.2.3' }] });++ assert.strictEqual(rebinding.ok, false, 'any private answer refuses the host');++ assert.ok((await checkResolvedHost('good.example', { lookup: async () => [{ address: '8.8.8.8' }] })).ok);++});++++t('ffmpeg arguments match Live: codec copy from HTTP-FLV, platform URL fixes, SRT options, redaction', () => {++ assert.strictEqual(ff.buildDestUrl({ platform: 'twitch', server_url: 'rtmp://live.twitch.tv/app/', stream_key: 'live_1' }), 'rtmps://live.twitch.tv/app/live_1');++ assert.strictEqual(ff.buildDestUrl({ platform: 'kick', server_url: 'rtmps://x.global-contribute.live-video.net', stream_key: 'sk' }), 'rtmps://x.global-contribute.live-video.net/app/sk');++ const srt = new URL(ff.buildDestUrl({ platform: 'custom', server_url: 'srt://srt.example:9000', stream_key: 'sid', srt_latency_ms: 200, srt_passphrase: 'passphrase123' }));++ assert.strictEqual(srt.searchParams.get('streamid'), 'sid');++ assert.strictEqual(srt.searchParams.get('latency'), '200000');++ assert.strictEqual(srt.searchParams.get('mode'), 'caller');++ assert.strictEqual(srt.searchParams.get('pkt_size'), '1316');++ const args = ff.withProgress(ff.rtmpCopyArgs('http://127.0.0.1:1/live/ses_x.flv', 'rtmp://dest/app/key'));++ assert.deepStrictEqual(args, ['-hide_banner', '-progress', 'pipe:1', '-stats_period', '1', '-loglevel', 'warning', '-rw_timeout', '10000000',++ '-i', 'http://127.0.0.1:1/live/ses_x.flv', '-c', 'copy', '-fflags', '+nobuffer+discardcorrupt', '-muxdelay', '0', '-muxpreload', '0',++ '-flush_packets', '1', '-max_muxing_queue_size', '4096', '-rtmp_live', 'live', '-f', 'flv', '-flvflags', 'no_duration_filesize', 'rtmp://dest/app/key']);++ assert.deepStrictEqual(ff.outputArgs('srt://h:1?streamid=a').slice(-5), ['-f', 'mpegts', '-mpegts_flags', '+resend_headers', 'srt://h:1?streamid=a']);++ assert.strictEqual(ff.redactUrl('rtmps://live.twitch.tv/app/live_123456789'), 'rtmps://live.twitch.tv/app/****6789');++ assert.ok(!ff.redactUrl('srt://h:1?streamid=secret&passphrase=pw12345678').includes('secret'));++ assert.match(ff.friendlyError('Connection refused', 'twitch'), /Twitch's ingest server/);++ assert.match(ff.friendlyError('NetStream.Publish.BadName', 'youtube'), /Youtube rejected the stream key/);++});++++let api;++const cookie = `ov_token=${userToken({ subjectId: OWNER })}`;++++t('the UI works without JavaScript: landing, sign-in prompt, streams', async () => {++ api = await bootApi();++ const home = await request(api.base, 'GET', '/');++ assert.strictEqual(home.status, 200);++ assert.match(home.text, /OpenRe\.Stream/);++ assert.match(home.text, /theme-loader\.js/);++ assert.match(home.text, /<noscript><nav/);++ const signin = await request(api.base, 'GET', '/streams');++ assert.strictEqual(signin.status, 401);++ assert.match(signin.text, /Sign in with OpenVibe/);++ const mine = await request(api.base, 'GET', '/streams', { cookie });++ assert.strictEqual(mine.status, 200);++ assert.match(mine.text, /New stream/);++});++++t('forms need the CSRF token; creating a stream shows the key once', async () => {++ const noCsrf = await request(api.base, 'POST', '/streams', { cookie, form: { title: 'Nope' } });++ assert.strictEqual(noCsrf.status, 403);++ const token = cookie.slice('ov_token='.length);++ const created = await request(api.base, 'POST', '/streams', { cookie, form: { title: 'Form stream', recording_mode: 'none', recording_visibility: 'public', _csrf: csrfFor(token) } });++ assert.strictEqual(created.status, 200);++ assert.strictEqual(created.headers.get('cache-control'), 'no-store');++ const key = /(ork_[A-Za-z0-9_-]{43})/.exec(created.text)[1];++ const def = api.rt.store.definitions.list({ owner_subject: OWNER })[0];++ assert.strictEqual(api.rt.store.definitions.resolveIngestKey(key, 'rtmp').definition.id, def.id);++ const page = await request(api.base, 'GET', `/streams/${def.id}`, { cookie });++ assert.strictEqual(page.status, 200);++ assert.ok(!page.text.includes(key), 'the stream page never shows the key again');++ assert.match(page.text, new RegExp(`ork_…${key.slice(-4)}`));++ // Destination add through the form; the key is write-only.++ const add = await request(api.base, 'POST', `/streams/${def.id}/destinations`, { cookie, form: { platform: 'twitch', name: 'T', server_url: 'rtmps://live.twitch.tv/app', stream_key: 'live_abcdef_9876', enabled: '1', auto_start: '1', _csrf: csrfFor(token) } });++ assert.strictEqual(add.status, 303);++ const after = await request(api.base, 'GET', `/streams/${def.id}`, { cookie });++ assert.ok(!after.text.includes('live_abcdef_9876'));++ assert.match(after.text, /\*\*\*\*9876/);++ // Someone else cannot open it.++ const other = await request(api.base, 'GET', `/streams/${def.id}`, { cookie: `ov_token=${userToken({ subjectId: OTHER })}` });++ assert.strictEqual(other.status, 404);++ // Rotation through the form shows the new key once.++ const rotated = await request(api.base, 'POST', `/streams/${def.id}/rotate`, { cookie, form: { grace_seconds: '0', _csrf: csrfFor(token) } });++ const key2 = /(ork_[A-Za-z0-9_-]{43})/.exec(rotated.text)[1];++ assert.notStrictEqual(key2, key);++ assert.strictEqual(api.rt.store.definitions.resolveIngestKey(key, 'rtmp').error, 'revoked_key');++ await api.close();++});++++t.run();nnew file mode 100644@@@ -0,0 +1,154 @@++'use strict';++// Migration from a Live DB snapshot: read-only, new keys (never the old ones), destinations++// sealed, held/excluded rows recorded with reasons, idempotent re-runs, and the checklist.++const assert = require('assert');++const fs = require('fs');++const path = require('path');++const { execFileSync } = require('child_process');++const Database = require('better-sqlite3');++const { runtime, tmpDir, testEnv, suite, ROOT } = require('./helpers');++const { migrate, checklist } = require('../server/migrate-live');++const { hashIngestKey } = require('../server/secrets');++++const t = suite('migration');++const SUBJECT_A = 'usr_01J00000000000000000000A01';++const SUBJECT_B = 'usr_01J00000000000000000000B01';++++function liveSnapshot(dir) {++ const file = path.join(dir, 'live-snapshot.db');++ const db = new Database(file);++ db.exec(`++ CREATE TABLE users (id INTEGER PRIMARY KEY, username TEXT, display_name TEXT, password_hash TEXT, stream_key TEXT, is_banned INTEGER DEFAULT 0);++ CREATE TABLE channels (id INTEGER PRIMARY KEY, user_id INTEGER, vod_recording_enabled INTEGER DEFAULT 1, force_vod_recording_disabled INTEGER DEFAULT 0, default_vod_visibility TEXT DEFAULT 'public');++ CREATE TABLE linked_accounts (id INTEGER PRIMARY KEY, user_id INTEGER, service TEXT, service_user_id TEXT, subject_id TEXT);++ CREATE TABLE managed_streams (id INTEGER PRIMARY KEY, user_id INTEGER, slug TEXT, title TEXT, description TEXT, protocol TEXT, stream_key TEXT,++ streaming_method TEXT, default_vod_visibility TEXT, slot_vod_recording_enabled INTEGER DEFAULT 1, slot_clip_recording_enabled INTEGER DEFAULT 1);++ CREATE TABLE restream_destinations (id INTEGER PRIMARY KEY, user_id INTEGER, managed_stream_id INTEGER, platform TEXT, name TEXT, server_url TEXT,++ stream_key TEXT, enabled INTEGER, auto_start INTEGER, quality_preset TEXT, connection_id INTEGER, srt_latency_ms INTEGER, srt_passphrase TEXT,++ custom_video_bitrate INTEGER, custom_audio_bitrate INTEGER, custom_fps INTEGER, custom_encoder_preset TEXT);++ INSERT INTO users VALUES (1, 'japaneseoldguy', 'JOG', 'x', 'personalkey111', 0), (2, 'nosubject', 'NS', 'x', 'k2', 0), (3, 'banned', 'B', 'x', 'k3', 1), (4, 'twoslots', 'TS', 'x', 'k4', 0);++ INSERT INTO channels (user_id, vod_recording_enabled, default_vod_visibility) VALUES (1, 1, 'unlisted'), (4, 0, 'public');++ INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (1, 'network', '57', '${SUBJECT_A}'), (3, 'network', '58', 'usr_01J00000000000000000000C01'), (4, 'network', '59', '${SUBJECT_B}');++ INSERT INTO managed_streams VALUES++ (10, 1, 'main', 'Morning walk', '', 'rtmp', 'a3f9c2leakedkey0000000000000000000000000', 'obs', NULL, 1, 1),++ (11, 2, NULL, 'No subject yet', '', 'rtmp', 'b000000000000000000000000000000000000000', NULL, NULL, 1, 1),++ (12, 3, NULL, 'Banned', '', 'rtmp', 'c000000000000000000000000000000000000000', NULL, NULL, 1, 1),++ (13, 4, 'a', 'Slot A', '', 'webrtc', 'd000000000000000000000000000000000000000', 'browser', NULL, 1, 0),++ (14, 4, 'b', 'Slot B', '', 'rtmp', 'e000000000000000000000000000000000000000', 'obs', 'private', 1, 1);++ INSERT INTO restream_destinations VALUES++ (100, 1, 10, 'twitch', 'Twitch', 'rtmps://live.twitch.tv/app', 'live_twitchsecret_1', 1, 1, 'auto', NULL, NULL, NULL, NULL, NULL, NULL, NULL),++ (101, 1, 10, 'youtube', 'YT', 'rtmp://a.rtmp.youtube.com/live2', 'yt-secret-2222', 1, 0, 'high', 7, NULL, NULL, 4500, NULL, NULL, NULL),++ (102, 1, 10, 'custom', 'LAN box', 'rtmp://192.168.1.20/live', 'lan-secret-333', 1, 1, 'auto', NULL, NULL, NULL, NULL, NULL, NULL, NULL),++ (103, 4, NULL, 'kick', 'Kick', 'rtmps://fa723fc1b171.global-contribute.live-video.net', 'kick-secret-44', 1, 1, 'auto', NULL, NULL, NULL, NULL, NULL, NULL, NULL);++ `);++ db.close();++ return file;++}++++t('dry run writes nothing', () => {++ const dir = tmpDir();++ const rt = runtime({ dir });++ const live = new Database(liveSnapshot(dir), { readonly: true });++ const report = migrate({ liveDb: live, rt, apply: false });++ assert.strictEqual(rt.db.prepare('SELECT COUNT(*) AS n FROM stream_definitions').get().n, 0);++ assert.strictEqual(rt.db.prepare('SELECT COUNT(*) AS n FROM migration_map').get().n, 0);++ assert.ok(report.counts.imported > 0);++ assert.match(checklist(report), /dry run/);++});++++let dir;++let rt;++let liveFile;++t('apply: slots with a subject are imported with new keys; the rest are held or excluded with reasons', () => {++ dir = tmpDir();++ rt = runtime({ dir, env: { OPENRE_DEST_ALLOW_PRIVATE: '0' } });++ liveFile = liveSnapshot(dir);++ const before = fs.readFileSync(liveFile);++ const live = new Database(liveFile, { readonly: true });++ const report = migrate({ liveDb: live, rt, apply: true });++ live.close();++ assert.ok(before.equals(fs.readFileSync(liveFile)), 'the Live snapshot is untouched');++++ const map = Object.fromEntries(rt.db.prepare('SELECT source_type, source_id, status, reason, target_id FROM migration_map').all().map(r => [`${r.source_type}:${r.source_id}`, r]));++ assert.strictEqual(map['managed_stream:10'].status, 'imported');++ assert.strictEqual(map['managed_stream:11'].status, 'held');++ assert.match(map['managed_stream:11'].reason, /no canonical subject/);++ assert.strictEqual(map['managed_stream:12'].status, 'excluded');++ assert.strictEqual(map['managed_stream:13'].status, 'imported');++ assert.strictEqual(map['managed_stream:14'].status, 'imported');++++ const def10 = rt.store.definitions.get(map['managed_stream:10'].target_id);++ assert.strictEqual(def10.owner_subject, SUBJECT_A);++ assert.strictEqual(def10.recording_visibility, 'unlisted', 'from the channel default');++ assert.strictEqual(def10.recording_mode, 'vod');++ assert.deepStrictEqual(def10.external_refs.map(r => `${r.service}:${r.type}:${r.id}`), ['live:managed_stream:10', 'live:user:1']);++ assert.strictEqual(rt.store.definitions.get(map['managed_stream:13'].target_id).recording_mode, 'none', 'VOD off on the channel, clips off on the slot');++ assert.strictEqual(rt.store.definitions.get(map['managed_stream:14'].target_id).recording_mode, 'clips');++ assert.strictEqual(rt.store.definitions.get(map['managed_stream:14'].target_id).recording_visibility, 'private');++++ // Old keys: none of them authenticates, and none is stored in any form.++ for (const old of ['a3f9c2leakedkey0000000000000000000000000', 'personalkey111']) {++ assert.strictEqual(rt.db.prepare('SELECT COUNT(*) AS n FROM ingest_keys WHERE key_hash = ?').get(hashIngestKey(old)).n, 0);++ assert.ok(rt.store.definitions.resolveIngestKey(old, 'rtmp').error);++ }++ assert.strictEqual(rt.db.prepare("SELECT COUNT(*) AS n FROM ingest_keys WHERE status = 'active'").get().n, 3, 'one new key per imported slot');++++ // Destinations: sealed, LAN one held, OAuth-linked flagged, unbound one held (two slots).++ assert.strictEqual(map['restream_destination:100'].status, 'imported');++ assert.strictEqual(map['restream_destination:101'].status, 'imported');++ assert.strictEqual(map['restream_destination:102'].status, 'held');++ assert.match(map['restream_destination:102'].reason, /private/);++ assert.strictEqual(map['restream_destination:103'].status, 'held');++ assert.match(map['restream_destination:103'].reason, /several slots/);++ const d100 = rt.store.outputs.destinationForWorker(map['restream_destination:100'].target_id);++ assert.strictEqual(d100.stream_key, 'live_twitchsecret_1');++ assert.ok(!JSON.stringify(rt.db.prepare('SELECT * FROM destinations').all()).includes('live_twitchsecret_1'), 'sealed at rest');++ const d101 = rt.store.outputs.getDestination(map['restream_destination:101'].target_id);++ assert.strictEqual(d101.auto_start, false);++ assert.strictEqual(d101.custom_video_bitrate, 4500);++ const d102 = rt.store.outputs.getDestination(map['restream_destination:102'].target_id);++ assert.strictEqual(d102.enabled, false);++ assert.ok(d102.hold_reason);++++ const md = checklist(report, { rtmpUrl: 'rtmp://ingest.openre.stream:1936/live' });++ assert.match(md, /## @japaneseoldguy/);++ assert.match(md, /ingest-authority/);++ assert.match(md, /OAuth-linked/);++ assert.match(md, /only RTMP moves to OpenRe/, 'the WebRTC slot is flagged');++ for (const secret of ['a3f9c2leakedkey', 'live_twitchsecret_1', 'yt-secret-2222', 'personalkey111', 'kick-secret-44']) assert.ok(!md.includes(secret), `no ${secret} in the checklist`);++});++++t('re-running is idempotent; a held slot is imported once its user has a subject', () => {++ const counts = () => ({++ defs: rt.db.prepare('SELECT COUNT(*) AS n FROM stream_definitions').get().n,++ dests: rt.db.prepare('SELECT COUNT(*) AS n FROM destinations').get().n,++ keys: rt.db.prepare('SELECT COUNT(*) AS n FROM ingest_keys').get().n,++ });++ const before = counts();++ let live = new Database(liveFile, { readonly: true });++ const again = migrate({ liveDb: live, rt, apply: true });++ live.close();++ assert.deepStrictEqual(counts(), before);++ assert.strictEqual(again.counts.imported, 0);++ const w = new Database(liveFile);++ w.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (2, 'network', '60', 'usr_01J00000000000000000000D01')").run();++ w.close();++ live = new Database(liveFile, { readonly: true });++ migrate({ liveDb: live, rt, apply: true });++ live.close();++ assert.strictEqual(counts().defs, before.defs + 1);++ assert.strictEqual(rt.db.prepare("SELECT status FROM migration_map WHERE source_type = 'managed_stream' AND source_id = '11'").get().status, 'imported');++});++++t('the CLI runs a dry run against a snapshot and prints the checklist', () => {++ const d = tmpDir();++ const file = liveSnapshot(d);++ const env = testEnv(d);++ const out = execFileSync(process.execPath, [path.join(ROOT, 'scripts', 'migrate-from-live.js'), '--live-db', file], {++ env: { ...process.env, ...env, OPENRE_ENV_FILE: '/nonexistent' }, cwd: ROOT, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'],++ });++ assert.match(out, /OpenRe RTMP cutover checklist \(dry run/);++ assert.match(out, /@twoslots/);++});++++t.run();nnew file mode 100644@@@ -0,0 +1,155 @@++'use strict';++// Recording requests (Media contract as Live's recorder uses it) and the event relay (outbox →++// OpenVibe.Events with OpenRe's service token), with stub Media / Network / Events servers.++const assert = require('assert');++const http = require('http');++const { runtime, manualClock, suite, OWNER, silent, outboxTypes } = require('./helpers');++const { createCoordinator } = require('../server/coordinator');++const { createMediaClient } = require('../server/media-client');++++const t = suite('recording-events');++++function stub(handler) {++ return new Promise((resolve) => {++ const s = http.createServer((req, res) => {++ let body = '';++ req.on('data', (c) => { body += c; });++ req.on('end', () => { let parsed = null; try { parsed = body ? JSON.parse(body) : null; } catch { parsed = null; } handler(req, res, parsed, body); });++ });++ s.listen(0, '127.0.0.1', () => resolve({ url: `http://127.0.0.1:${s.address().port}`, close: () => new Promise(r => { s.closeAllConnections(); s.close(r); }) }));++ });++}++const reply = (res, status, obj) => { res.statusCode = status; res.setHeader('Content-Type', 'application/json'); res.end(JSON.stringify(obj || {})); };++++async function setup({ media: mediaHandler, recording_mode = 'vod' }) {++ const calls = [];++ const media = await stub((req, res, body) => { calls.push({ method: req.method, url: req.url, body, auth: req.headers.authorization }); mediaHandler(req, res, body, calls); });++ const clock = manualClock();++ const rt = runtime({ clock, env: { MEDIA_URL: media.url, MEDIA_API_KEY: 'k-live', OPENRE_RECORDING_START_DELAY_MS: '0' } });++ const client = createMediaClient({ config: rt.config });++ const coordinator = createCoordinator({ rt, media: client, log: silent });++ const { definition, key } = rt.store.definitions.create({ owner_subject: OWNER, title: 'Rec', recording_mode, external_refs: [{ service: 'live', type: 'user', id: '8' }] });++ const w = rt.store.workers.register({ kind: 'rtmp-ingest', endpoints: { rtmpPlayPort: 19390, flvPort: 19391 } });++ rt.store.workers.ready(w.id);++ const a = rt.store.sessions.admit({ definition, key, protocol: 'rtmp', worker: rt.store.workers.get(w.id) });++ rt.store.sessions.transition(a.session.id, 'live');++ return { rt, clock, coordinator, session: a.session, calls, media };++}++++t('a live session gets one recording request; finalize when it ends (Media finalises)', async () => {++ const { rt, coordinator, session, calls, media } = await setup({++ media: (req, res) => {++ if (req.url === '/api/v1/live/vods') return reply(res, 201, { id: 501 });++ if (req.url === '/api/v1/live/vods/501/ingest/rtmp') return reply(res, 202, { status: 'recording' });++ if (req.url === '/api/v1/live/vods/501/finalize') return reply(res, 200, {});++ return reply(res, 404);++ },++ });++ await coordinator.tick();++ await coordinator.tick();++ assert.deepStrictEqual(calls.map(c => `${c.method} ${c.url}`), ['POST /api/v1/live/vods', 'POST /api/v1/live/vods/501/ingest/rtmp']);++ assert.strictEqual(calls[0].auth, 'Bearer k-live');++ assert.strictEqual(calls[0].body.user_id, 8);++ assert.strictEqual(calls[0].body.visibility, 'public');++ assert.strictEqual(calls[1].body.rtmp_url, `rtmp://127.0.0.1:19390/live/${session.id}`);++ assert.strictEqual(rt.store.recordings.bySession(session.id).state, 'recording');++ assert.deepStrictEqual(outboxTypes(rt.db).filter(x => x.startsWith('openre.recording')), ['openre.recording.requested']);++ rt.store.sessions.finish(session.id, { reason: 'publisher_disconnected' });++ await coordinator.tick();++ assert.strictEqual(calls.pop().url, '/api/v1/live/vods/501/finalize');++ assert.strictEqual(rt.store.recordings.bySession(session.id).state, 'finalized');++ await media.close();++});++++t('Media refusing for disk space: the empty VOD shell is deleted and the request retried later', async () => {++ let refuse = true;++ const { rt, clock, coordinator, session, calls, media } = await setup({++ media: (req, res) => {++ if (req.method === 'POST' && req.url === '/api/v1/live/vods') return reply(res, 201, { id: 600 + calls.length });++ if (req.url.endsWith('/ingest/rtmp')) return refuse ? reply(res, 409, { error: 'Disk critically low — recording refused' }) : reply(res, 202, {});++ if (req.method === 'DELETE') return reply(res, 200, {});++ return reply(res, 404);++ },++ });++ await coordinator.tick();++ assert.deepStrictEqual(calls.map(c => c.method), ['POST', 'POST', 'DELETE'], 'create, ingest refused, shell deleted');++ const rec = rt.store.recordings.bySession(session.id);++ assert.strictEqual(rec.state, 'pending');++ assert.match(rec.last_error, /Disk/);++ assert.ok(rec.next_attempt_at >= clock.now() + 5 * 60 * 1000, 'disk refusals wait five minutes');++ refuse = false;++ clock.advance(5 * 60 * 1000 + 1);++ rt.store.workers.heartbeat(session.worker_id); // the ingest worker is alive throughout++ await coordinator.tick();++ assert.strictEqual(rt.store.recordings.bySession(session.id).state, 'recording');++ await media.close();++});++++t('clips-only recordings are deleted after finalising; a session that ends first is never requested', async () => {++ const { rt, coordinator, session, calls, media } = await setup({++ recording_mode: 'clips',++ media: (req, res) => {++ if (req.url === '/api/v1/live/vods') return reply(res, 201, { id: 700 });++ if (req.method === 'DELETE') return reply(res, 200, {});++ return reply(res, req.url.endsWith('/finalize') ? 200 : 202, {});++ },++ });++ await coordinator.tick();++ assert.strictEqual(calls[0].body.clips_only, true);++ rt.store.sessions.finish(session.id, { reason: 'x' });++ await coordinator.tick();++ assert.deepStrictEqual(calls.slice(-2).map(c => `${c.method} ${c.url}`), ['POST /api/v1/live/vods/700/finalize', 'DELETE /api/v1/live/vods/700']);++ await media.close();++++ const s2 = await setup({ media: (req, res) => reply(res, 500) });++ s2.rt.store.sessions.finish(s2.session.id, { reason: 'gone' });++ s2.rt.db.prepare("INSERT INTO recordings (id, session_id, mode, state, created_at, updated_at) VALUES ('rec_01J0000000000000000000000A', ?, 'vod', 'pending', 0, 0)").run(s2.session.id);++ await s2.coordinator.tick();++ assert.strictEqual(s2.rt.store.recordings.bySession(s2.session.id).state, 'cancelled');++ assert.strictEqual(s2.calls.length, 0);++ await s2.media.close();++});++++t('the relay publishes outbox rows with a Network service token; Events down means rows wait', async () => {++ let down = true;++ const got = [];++ const events = await stub((req, res, body) => {++ if (down) return reply(res, 503, { code: 'events.unavailable' });++ assert.strictEqual(req.headers.authorization, 'Bearer svc-token');++ const list = body.events || [body];++ list.forEach(e => got.push(e));++ return reply(res, 200, body.events ? { results: list.map((e, i) => ({ event_id: e.event_id, seq: i + 1 })) } : { event_id: list[0].event_id, seq: 1 });++ });++ const network = await stub((req, res, body, raw) => {++ const p = new URLSearchParams(raw);++ assert.strictEqual(p.get('grant_type'), 'client_credentials');++ assert.strictEqual(p.get('client_id'), 'openre');++ assert.strictEqual(p.get('audience'), 'openvibe.events');++ reply(res, 200, { access_token: 'svc-token', token_type: 'Bearer', expires_in: 300 });++ });++ const rt = runtime({ env: { EVENTS_URL: events.url, OV_NETWORK_INTERNAL_URL: network.url, OV_OAUTH_CLIENT_SECRET: 's3cret' } });++ assert.strictEqual(rt.events.configured, true);++ const { definition } = rt.store.definitions.create({ owner_subject: OWNER });++ rt.store.definitions.rotateKey(definition.id, { rotated_by: OWNER });++ await rt.events.outbox.flush();++ assert.strictEqual(rt.events.outbox.pending(), 1, 'kept while Events is down');++ down = false;++ rt.db.prepare('UPDATE event_outbox SET next_attempt_at = 0').run();++ await rt.events.outbox.flush();++ assert.strictEqual(rt.events.outbox.pending(), 0);++ assert.strictEqual(got[0].event_type, 'openre.key.rotated');++ assert.strictEqual(got[0].source, 'openre');++ await events.close();++ await network.close();++});++++t('without EVENTS_URL the relay is off but every row is still written with its change', () => {++ const rt = runtime();++ assert.strictEqual(rt.events.configured, false);++ assert.strictEqual(rt.events.start(), false);++ const { definition } = rt.store.definitions.create({ owner_subject: OWNER });++ rt.store.definitions.rotateKey(definition.id);++ assert.strictEqual(rt.events.status().pending, 1);++});++++t.run();
new file mode 100644
new file mode 100644
@@ -0,0 +1,48 @@
@@ -0,0 +1,48 @@
+#!/usr/bin/env node
+'use strict';
+/**
+ * Import Live's stream slots and restream destinations into OpenRe, from a read-only snapshot of
+ * Live's database, and print the per-channel RTMP cutover checklist.
+ *
+ * node scripts/migrate-from-live.js --live-db /path/to/live-snapshot.db # dry run
+ * node scripts/migrate-from-live.js --live-db /path/to/live-snapshot.db --apply # write
+ * ... [--slots 12,31] [--checklist /path/to/checklist.md]
+ *
+ * Take the snapshot with `sqlite3 /opt/openvibe.live/data/openvibe.db ".backup /tmp/live-snapshot.db"`
+ * (never point this at the live file). Uses the OpenRe environment (.env or OPENRE_ENV_FILE) for
+ * its own database and OPENRE_SECRETS_KEY. Prints no secret: old keys are not imported and the new
+ * keys are never shown (the broadcaster rotates to get one). Safe to re-run.
+ */
+const fs = require('fs');
+const path = require('path');
+const Database = require('better-sqlite3');
+
+function arg(name) {
+ const i = process.argv.indexOf(`--${name}`);
+ return i > 0 ? process.argv[i + 1] : null;
+}
+
+function main() {
+ require('dotenv').config({ path: process.env.OPENRE_ENV_FILE || path.join(process.cwd(), '.env') });
+ const { load } = require('../server/config');
+ const { openRuntime } = require('../server/store');
+ const { migrate, checklist } = require('../server/migrate-live');
+ const liveDbPath = arg('live-db');
+ if (!liveDbPath) { console.error('usage: migrate-from-live.js --live-db <snapshot.db> [--apply] [--slots 1,2] [--checklist out.md]'); process.exit(2); }
+ const apply = process.argv.includes('--apply');
+ const config = load();
+ if (apply && !config.secretsKey) { console.error('OPENRE_SECRETS_KEY is required to import destinations'); process.exit(2); }
+ const liveDb = new Database(liveDbPath, { readonly: true, fileMustExist: true });
+ const rt = openRuntime({ config, log: { log() {}, warn: console.warn, error: console.error } });
+ const onlySlots = arg('slots') ? arg('slots').split(',').map(Number).filter(Number.isFinite) : null;
+ const report = migrate({ liveDb, rt, apply, onlySlots });
+ const r = config.rtmp;
+ const md = checklist(report, { openreUrl: config.baseUrl, rtmpUrl: `rtmp://${r.publicHost}${r.publicPort === 1935 ? '' : `:${r.publicPort}`}/live` });
+ const out = arg('checklist');
+ if (out) { fs.writeFileSync(out, md + '\n'); console.log(`checklist written to ${out}`); } else console.log(md);
+ console.error(`[migrate] ${apply ? 'applied' : 'dry run'}: ${JSON.stringify(report.counts)}`);
+ liveDb.close();
+ rt.db.close();
+}
+
+if (require.main === module) main();
@@ -37,6 +37,8 @@ CREATE TABLE IF NOT EXISTS stream_definitions (
@@ -37,6 +37,8 @@ CREATE TABLE IF NOT EXISTS stream_definitions (
CREATE INDEX IF NOT EXISTS idx_definitions_owner ON stream_definitions(owner_subject, state);
CREATE INDEX IF NOT EXISTS idx_definitions_owner ON stream_definitions(owner_subject, state);
-- Typed references to other services' entities (plan §29.3): never a foreign key into them.
-- Typed references to other services' entities (plan §29.3): never a foreign key into them.
+-- A reference to a stream-like entity (e.g. live:managed_stream) belongs to one definition; owner
+-- references (live:user, live:channel) may repeat across a person's definitions.
CREATE TABLE IF NOT EXISTS external_refs (
CREATE TABLE IF NOT EXISTS external_refs (
definition_id TEXT NOT NULL REFERENCES stream_definitions(id) ON DELETE CASCADE,
definition_id TEXT NOT NULL REFERENCES stream_definitions(id) ON DELETE CASCADE,
service TEXT NOT NULL,
service TEXT NOT NULL,
@@ -44,9 +46,9 @@ CREATE TABLE IF NOT EXISTS external_refs (
@@ -44,9 +46,9 @@ CREATE TABLE IF NOT EXISTS external_refs (
ref_id TEXT NOT NULL,
ref_id TEXT NOT NULL,
label TEXT,
label TEXT,
created_at INTEGER NOT NULL,
created_at INTEGER NOT NULL,
- PRIMARY KEY (service, type, ref_id)
+ PRIMARY KEY (definition_id, service, type, ref_id)
);
);
-CREATE INDEX IF NOT EXISTS idx_external_refs_definition ON external_refs(definition_id);
+CREATE INDEX IF NOT EXISTS idx_external_refs_ref ON external_refs(service, type, ref_id);
-- Ingest keys are stored as SHA-256 of a 256-bit random secret; the secret is shown once.
-- Ingest keys are stored as SHA-256 of a 256-bit random secret; the secret is shown once.
CREATE TABLE IF NOT EXISTS ingest_keys (
CREATE TABLE IF NOT EXISTS ingest_keys (
new file mode 100644
new file mode 100644
@@ -0,0 +1,229 @@
@@ -0,0 +1,229 @@
+'use strict';
+/**
+ * Import OpenVibe.Live stream slots (managed_streams) and their restream destinations from a
+ * read-only snapshot of Live's database into OpenRe (scripts/migrate-from-live.js is the CLI).
+ *
+ * Rules (ADR-009, plan §8.1):
+ * - Live's database is opened read-only; nothing is written to it.
+ * - Stream keys are NEVER imported. Every imported slot gets a new OpenRe key that nobody has
+ * seen; the streamer gets a usable key by rotating (Live's "Regenerate" once the slot is
+ * switched to OpenRe, or openre.stream). Every key that existed before the cutover is dead
+ * at the cutover.
+ * - Every source row is imported, held with a reason, or excluded with a reason, and recorded in
+ * migration_map. Re-running is idempotent: imported rows are skipped, held rows are retried.
+ * - Ownership is the canonical subject (linked_accounts.subject_id); Live's integer ids travel
+ * only as typed references (live:managed_stream, live:user) for Media's 'live' tenant and the
+ * Live mirror.
+ */
+
+const { validateDestinationUrl } = require('./destination-url');
+
+function columns(db, table) {
+ try { return new Set(db.prepare(`PRAGMA table_info(${table})`).all().map(c => c.name)); } catch { return new Set(); }
+}
+
+function recordingModeOf(ms, channel) {
+ let vod = !channel ? true : Boolean(channel.vod_recording_enabled) && !channel.force_vod_recording_disabled;
+ if (vod && ms.slot_vod_recording_enabled === 0) vod = false;
+ if (vod) return 'vod';
+ return ms.slot_clip_recording_enabled === 0 ? 'none' : 'clips';
+}
+
+function visibilityOf(ms, channel) {
+ const v = ms.default_vod_visibility || (channel && channel.default_vod_visibility) || 'public';
+ return ['public', 'unlisted', 'private'].includes(v) ? v : 'public';
+}
+
+/**
+ * migrate({ liveDb, rt, apply, onlySlots }) -> report
+ * report = { slots: [...], destinations: [...], channels: Map username → {...}, counts }
+ */
+function migrate({ liveDb, rt, apply = false, onlySlots = null, now = () => Date.now() }) {
+ const { db, store, config } = rt;
+ const msCols = columns(liveDb, 'managed_streams');
+ if (!msCols.size) throw new Error('the Live snapshot has no managed_streams table');
+ const rdCols = columns(liveDb, 'restream_destinations');
+ const laCols = columns(liveDb, 'linked_accounts');
+ const chCols = columns(liveDb, 'channels');
+
+ const mapGet = db.prepare("SELECT * FROM migration_map WHERE source_system = 'live' AND source_type = ? AND source_id = ?");
+ const mapPut = db.prepare(`INSERT INTO migration_map (source_system, source_type, source_id, target_type, target_id, status, reason, imported_at)
+ VALUES ('live', @source_type, @source_id, @target_type, @target_id, @status, @reason, @at)
+ ON CONFLICT(source_system, source_type, source_id) DO UPDATE SET target_type = @target_type, target_id = @target_id, status = @status, reason = @reason, imported_at = @at`);
+ const record = (row) => { if (apply) mapPut.run({ target_type: null, target_id: null, reason: null, ...row, at: now() }); };
+
+ const slots = liveDb.prepare(`SELECT ms.*, u.username, u.display_name, u.is_banned FROM managed_streams ms JOIN users u ON u.id = ms.user_id ORDER BY ms.user_id, ms.id`).all()
+ .filter(ms => !onlySlots || onlySlots.includes(Number(ms.id)));
+ const subjectOf = laCols.has('subject_id')
+ ? liveDb.prepare("SELECT subject_id FROM linked_accounts WHERE service = 'network' AND user_id = ? AND subject_id IS NOT NULL")
+ : null;
+ const channelOf = chCols.size ? liveDb.prepare('SELECT * FROM channels WHERE user_id = ?') : null;
+ const destsOfSlot = rdCols.has('managed_stream_id') ? liveDb.prepare('SELECT * FROM restream_destinations WHERE managed_stream_id = ? ORDER BY id') : null;
+ const unboundOfUser = rdCols.size
+ ? liveDb.prepare(`SELECT * FROM restream_destinations WHERE user_id = ? ${rdCols.has('managed_stream_id') ? 'AND managed_stream_id IS NULL' : ''} ORDER BY id`)
+ : null;
+ const slotsOfUser = liveDb.prepare('SELECT COUNT(*) AS n FROM managed_streams WHERE user_id = ?');
+
+ const report = { apply, slots: [], destinations: [], channels: new Map(), counts: { imported: 0, held: 0, excluded: 0, skipped: 0 } };
+ const channel = (ms) => {
+ if (!report.channels.has(ms.user_id)) report.channels.set(ms.user_id, { user_id: ms.user_id, username: ms.username, display_name: ms.display_name, slots: [], notes: [] });
+ return report.channels.get(ms.user_id);
+ };
+ const count = (status) => { report.counts[status] = (report.counts[status] || 0) + 1; };
+
+ for (const ms of slots) {
+ const ch = channel(ms);
+ const prior = mapGet.get('managed_stream', String(ms.id));
+ const entry = { live_id: ms.id, slug: ms.slug || null, title: ms.title, protocol: ms.protocol, streaming_method: ms.streaming_method || null, definition_id: null, status: null, reason: null, destinations: [] };
+ ch.slots.push(entry);
+
+ if (prior && prior.status === 'imported' && store.definitions.get(prior.target_id)) {
+ entry.definition_id = prior.target_id;
+ entry.status = 'imported';
+ entry.reason = 'already imported';
+ count('skipped');
+ } else {
+ const subject = subjectOf ? (subjectOf.get(ms.user_id) || {}).subject_id : null;
+ if (!subject) {
+ entry.status = 'held';
+ entry.reason = `Live user ${ms.user_id} has no canonical subject yet (it is recorded when they next sign in to Live); the slot is imported on a later run`;
+ record({ source_type: 'managed_stream', source_id: String(ms.id), status: 'held', reason: entry.reason });
+ count('held');
+ } else if (ms.is_banned) {
+ entry.status = 'excluded';
+ entry.reason = 'the Live account is banned';
+ record({ source_type: 'managed_stream', source_id: String(ms.id), status: 'excluded', reason: entry.reason });
+ count('excluded');
+ } else {
+ const chRow = channelOf ? channelOf.get(ms.user_id) : null;
+ const fields = {
+ owner_subject: subject,
+ title: ms.title || `${ms.display_name || ms.username}'s stream`,
+ description: ms.description || '',
+ protocols: ['rtmp'],
+ recording_mode: recordingModeOf(ms, chRow),
+ recording_visibility: visibilityOf(ms, chRow),
+ mirror_to_live: true,
+ // live:user is what Media's 'live' tenant files the VOD under (Live-local ids).
+ external_refs: [
+ { service: 'live', type: 'managed_stream', id: String(ms.id), label: ms.slug || ms.title || null },
+ { service: 'live', type: 'user', id: String(ms.user_id), label: ms.username },
+ ],
+ created_by: 'svc:openre-migration',
+ };
+ entry.recording_mode = fields.recording_mode;
+ if (apply) {
+ const existing = store.definitions.findByRef('live', 'managed_stream', String(ms.id));
+ const def = existing || store.definitions.create(fields).definition; // the new key is discarded unseen
+ entry.definition_id = def.id;
+ record({ source_type: 'managed_stream', source_id: String(ms.id), target_type: 'stream_definition', target_id: def.id, status: 'imported', reason: 'new key issued, old key not imported' });
+ }
+ entry.status = 'imported';
+ entry.reason = 'new key issued, old key not imported';
+ count('imported');
+ }
+ }
+
+ // Destinations: the slot's own, plus unbound ones when the user has exactly one slot.
+ const own = destsOfSlot ? destsOfSlot.all(ms.id) : [];
+ const unbound = unboundOfUser ? unboundOfUser.all(ms.user_id) : [];
+ const oneSlot = slotsOfUser.get(ms.user_id).n === 1;
+ const candidates = [...own.map(d => ({ d, unbound: false })), ...unbound.map(d => ({ d, unbound: true }))];
+ for (const { d, unbound: isUnbound } of candidates) {
+ const di = { live_id: d.id, platform: d.platform, name: d.name || d.platform, status: null, reason: null, destination_id: null, oauth_linked: Boolean(d.connection_id) };
+ entry.destinations.push(di);
+ report.destinations.push(di);
+ const priorD = mapGet.get('restream_destination', String(d.id));
+ if (priorD && priorD.target_id && store.outputs.destinationRow(priorD.target_id)) {
+ di.status = priorD.status; di.reason = priorD.reason || 'already imported'; di.destination_id = priorD.target_id; count('skipped');
+ continue;
+ }
+ if (isUnbound && !oneSlot) {
+ di.status = 'held';
+ di.reason = 'not bound to a slot and the user has several slots (Live pushed unbound destinations from every slot); bind it in Live or add it on openre.stream';
+ record({ source_type: 'restream_destination', source_id: String(d.id), status: 'held', reason: di.reason });
+ count('held');
+ continue;
+ }
+ if (entry.status !== 'imported') {
+ di.status = 'held';
+ di.reason = `its slot is ${entry.status}`;
+ record({ source_type: 'restream_destination', source_id: String(d.id), status: 'held', reason: di.reason });
+ count('held');
+ continue;
+ }
+ const v = validateDestinationUrl(d.server_url, { allowPrivate: config.outputs.allowPrivateHosts });
+ const hold = !v.ok ? `server URL refused by OpenRe's destination rules: ${v.error}` : null;
+ const input = {
+ platform: ['youtube', 'twitch', 'kick', 'custom'].includes(d.platform) ? d.platform : 'custom',
+ name: d.name || null,
+ server_url: v.ok ? v.value : 'rtmp://held.invalid/app',
+ stream_key: d.stream_key || '',
+ srt_passphrase: d.srt_passphrase || '',
+ srt_latency_ms: d.srt_latency_ms == null ? undefined : d.srt_latency_ms,
+ enabled: Boolean(d.enabled),
+ auto_start: Boolean(d.auto_start),
+ quality_preset: ['auto', 'low', 'medium', 'high', 'ultra', 'source'].includes(d.quality_preset) ? d.quality_preset : 'auto',
+ custom_video_bitrate: d.custom_video_bitrate == null ? undefined : d.custom_video_bitrate,
+ custom_audio_bitrate: d.custom_audio_bitrate == null ? undefined : d.custom_audio_bitrate,
+ custom_fps: d.custom_fps == null ? undefined : d.custom_fps,
+ custom_encoder_preset: d.custom_encoder_preset || undefined,
+ };
+ if (apply) {
+ let created;
+ if (hold) {
+ // Keep the (sealed) key so the owner only has to fix the URL; the row stays
+ // disabled until they do. The original URL is kept in the hold reason.
+ created = store.outputs.createDestination(entry.definition_id, { ...input, server_url: 'rtmp://held.invalid/app', enabled: false }, { hold_reason: `${hold} (was ${String(d.server_url || '').replace(/\/[^/]*$/, '/…')})` });
+ } else {
+ created = store.outputs.createDestination(entry.definition_id, input);
+ }
+ di.destination_id = created.id;
+ record({ source_type: 'restream_destination', source_id: String(d.id), target_type: 'destination', target_id: created.id, status: hold ? 'held' : 'imported', reason: hold || null });
+ }
+ di.status = hold ? 'held' : 'imported';
+ di.reason = hold;
+ count(hold ? 'held' : 'imported');
+ }
+ if (own.some(d => d.connection_id) || unbound.some(d => d.connection_id)) {
+ ch.notes.push('OAuth-linked destination(s): Live refreshed their ingest key (and created the YouTube broadcast) on every go-live; OpenRe pushes to the key stored at import. Check the platform key is a persistent one, or re-enter it on openre.stream after the cutover.');
+ }
+ }
+ return report;
+}
+
+/** Markdown checklist per channel, for the lead and the broadcaster. Contains no secret. */
+function checklist(report, { openreUrl = 'https://openre.stream', rtmpUrl } = {}) {
+ const lines = [];
+ lines.push(`# OpenRe RTMP cutover checklist${report.apply ? '' : ' (dry run: nothing was written)'}`);
+ lines.push('');
+ lines.push(`Imported ${report.counts.imported}, held ${report.counts.held}, excluded ${report.counts.excluded}, already done ${report.counts.skipped}.`);
+ lines.push('');
+ for (const ch of report.channels.values()) {
+ lines.push(`## @${ch.username} (Live user ${ch.user_id})`);
+ lines.push('');
+ for (const s of ch.slots) {
+ lines.push(`### Slot ${s.live_id}${s.slug ? ` "${s.slug}"` : ''}: ${s.title || ''} — ${s.status}${s.definition_id ? ` → ${s.definition_id}` : ''}`);
+ if (s.reason && s.status !== 'imported') lines.push(`- ${s.status}: ${s.reason}`);
+ if (s.status === 'imported') {
+ if (s.protocol !== 'rtmp' && !['rtmp', 'obs'].includes(String(s.streaming_method || '').toLowerCase())) {
+ lines.push(`- Slot protocol is ${s.protocol}${s.streaming_method ? ` / ${s.streaming_method}` : ''}: only RTMP moves to OpenRe; leave this slot on Live unless the broadcaster streams it with OBS/RTMP.`);
+ }
+ lines.push(`- [ ] Maintenance window agreed with @${ch.username} (the slot must be offline when it is switched).`);
+ lines.push(`- [ ] Destinations reviewed on ${openreUrl}/streams/${s.definition_id || '<id>'}: ${s.destinations.length ? s.destinations.map(d => `${d.name} (${d.platform}) ${d.status}${d.reason ? `: ${d.reason}` : ''}`).join('; ') : 'none'}`);
+ lines.push(`- [ ] Recording: ${s.recording_mode || 'as imported'} (from the slot's VOD/clip settings on Live).`);
+ lines.push(`- [ ] Switch: \`PUT https://openvibe.live/api/admin/openre/managed/${s.live_id}/ingest-authority {"authority":"openre"}\` (rotates Live's own key for the slot at the same time).`);
+ lines.push(`- [ ] Broadcaster presses "Regenerate stream key" on Live's Go Live page (or rotates on openre.stream) and pastes the new server${rtmpUrl ? ` \`${rtmpUrl}\`` : ''} and key into OBS. The old key no longer works anywhere.`);
+ lines.push('- [ ] Test broadcast: the stream shows on the Live channel page, restreams go live, a VOD appears in Media when it ends.');
+ lines.push(`- [ ] Rollback if needed: \`{"authority":"live"}\` on the same route; the broadcaster regenerates the key on Live.`);
+ }
+ lines.push('');
+ }
+ lines.push(`- [ ] Personal key: rotate @${ch.username}'s personal stream key on Live (users.stream_key is not migrated; any leaked copy must die too).`);
+ for (const n of ch.notes) lines.push(`- Note: ${n}`);
+ lines.push('');
+ }
+ return lines.join('\n');
+}
+
+module.exports = { migrate, checklist, recordingModeOf };
@@ -16,6 +16,8 @@ const RECORDING_MODES = Object.freeze(['vod', 'clips', 'none']);
@@ -16,6 +16,8 @@ const RECORDING_MODES = Object.freeze(['vod', 'clips', 'none']);
const VISIBILITIES = Object.freeze(['public', 'unlisted', 'private']);
const VISIBILITIES = Object.freeze(['public', 'unlisted', 'private']);
const REF_SERVICE_RE = /^[a-z][a-z0-9-]{1,39}$/;
const REF_SERVICE_RE = /^[a-z][a-z0-9-]{1,39}$/;
const REF_TYPE_RE = /^[a-z][a-z0-9_]{1,39}$/;
const REF_TYPE_RE = /^[a-z][a-z0-9_]{1,39}$/;
+// Owner-level reference types may be shared by several definitions of the same person.
+const SHARED_REF_TYPES = new Set(['user', 'channel']);
class StoreError extends Error {
class StoreError extends Error {
constructor(status, code, detail) { super(detail); this.status = status; this.code = code; }
constructor(status, code, detail) { super(detail); this.status = status; this.code = code; }
@@ -56,7 +58,8 @@ function createDefinitions({ db, config, events, clock }) {
@@ -56,7 +58,8 @@ function createDefinitions({ db, config, events, clock }) {
get: db.prepare('SELECT * FROM stream_definitions WHERE id = ?'),
get: db.prepare('SELECT * FROM stream_definitions WHERE id = ?'),
refs: db.prepare('SELECT service, type, ref_id AS id, label FROM external_refs WHERE definition_id = ? ORDER BY created_at, service, type'),
refs: db.prepare('SELECT service, type, ref_id AS id, label FROM external_refs WHERE definition_id = ? ORDER BY created_at, service, type'),
insertRef: db.prepare('INSERT INTO external_refs (definition_id, service, type, ref_id, label, created_at) VALUES (?, ?, ?, ?, ?, ?)'),
insertRef: db.prepare('INSERT INTO external_refs (definition_id, service, type, ref_id, label, created_at) VALUES (?, ?, ?, ?, ?, ?)'),
- byRef: db.prepare('SELECT definition_id FROM external_refs WHERE service = ? AND type = ? AND ref_id = ?'),
+ byRef: db.prepare(`SELECT r.definition_id FROM external_refs r JOIN stream_definitions d ON d.id = r.definition_id
+ WHERE r.service = ? AND r.type = ? AND r.ref_id = ? ORDER BY (d.state = 'archived'), d.created_at LIMIT 1`),
insertKey: db.prepare(`INSERT INTO ingest_keys (id, definition_id, key_hash, hint, status, created_by, created_at)
insertKey: db.prepare(`INSERT INTO ingest_keys (id, definition_id, key_hash, hint, status, created_by, created_at)
VALUES (?, ?, ?, ?, 'active', ?, ?)`),
VALUES (?, ?, ?, ?, 'active', ?, ?)`),
activeKeys: db.prepare("SELECT * FROM ingest_keys WHERE definition_id = ? AND status = 'active'"),
activeKeys: db.prepare("SELECT * FROM ingest_keys WHERE definition_id = ? AND status = 'active'"),
@@ -141,7 +144,9 @@ function createDefinitions({ db, config, events, clock }) {
@@ -141,7 +144,9 @@ function createDefinitions({ db, config, events, clock }) {
const id = newId('stream', now());
const id = newId('stream', now());
return db.transaction(() => {
return db.transaction(() => {
for (const r of refs) {
for (const r of refs) {
- if (q.byRef.get(r.service, r.type, r.id)) throw new StoreError(409, 'openre.ref_taken', `${r.service}:${r.type}:${r.id} already belongs to another stream definition`);
+ if (SHARED_REF_TYPES.has(r.type)) continue;
+ const taken = q.byRef.get(r.service, r.type, r.id);
+ if (taken && row(taken.definition_id).state !== 'archived') throw new StoreError(409, 'openre.ref_taken', `${r.service}:${r.type}:${r.id} already belongs to another stream definition`);
}
}
q.insert.run({ id, owner_subject: input.owner_subject, description: '', ...fields, now: now() });
q.insert.run({ id, owner_subject: input.owner_subject, description: '', ...fields, now: now() });
for (const r of refs) q.insertRef.run(id, r.service, r.type, r.id, r.label, now());
for (const r of refs) q.insertRef.run(id, r.service, r.type, r.id, r.label, now());
new file mode 100644
new file mode 100644
@@ -0,0 +1,96 @@
@@ -0,0 +1,96 @@
+'use strict';
+// Destination URL rules (SSRF), ffmpeg argument parity with Live's restream manager, and the
+// standalone UI (server-rendered, CSRF-protected, key shown once).
+const assert = require('assert');
+const { validateDestinationUrl, checkResolvedHost, isPrivateAddress } = require('../server/destination-url');
+const ff = require('../workers/restream/ffmpeg-args');
+const { bootApi, request, userToken, suite, OWNER, OTHER } = require('./helpers');
+const { csrfFor } = require('../server/ui/routes');
+
+const t = suite('destinations-ui');
+
+t('destination URLs: live-streaming schemes with public hosts only', async () => {
+ const ok = (u) => validateDestinationUrl(u).ok;
+ assert.ok(ok('rtmp://a.rtmp.youtube.com/live2'));
+ assert.ok(ok('rtmps://live.twitch.tv/app'));
+ assert.ok(ok('srt://ingest.example.com:9000?latency=200'));
+ for (const bad of ['file:///etc/passwd', 'http://example.com/x', '/tmp/out.flv', 'rtmp://user:pass@host/app', 'rtmp://127.0.0.1/app',
+ 'rtmp://10.0.0.5/app', 'rtmp://169.254.169.254/latest', 'rtmp://[::1]/app', 'rtmp://localhost/app', 'rtmp://nas.local/app', 'rtmp://host/app x']) {
+ assert.ok(!ok(bad), bad);
+ }
+ assert.ok(validateDestinationUrl('rtmp://127.0.0.1/app', { allowPrivate: true }).ok);
+ assert.ok(isPrivateAddress('100.64.1.1') && isPrivateAddress('::ffff:192.168.1.1') && !isPrivateAddress('8.8.8.8'));
+ const rebinding = await checkResolvedHost('evil.example', { lookup: async () => [{ address: '8.8.8.8' }, { address: '10.1.2.3' }] });
+ assert.strictEqual(rebinding.ok, false, 'any private answer refuses the host');
+ assert.ok((await checkResolvedHost('good.example', { lookup: async () => [{ address: '8.8.8.8' }] })).ok);
+});
+
+t('ffmpeg arguments match Live: codec copy from HTTP-FLV, platform URL fixes, SRT options, redaction', () => {
+ assert.strictEqual(ff.buildDestUrl({ platform: 'twitch', server_url: 'rtmp://live.twitch.tv/app/', stream_key: 'live_1' }), 'rtmps://live.twitch.tv/app/live_1');
+ assert.strictEqual(ff.buildDestUrl({ platform: 'kick', server_url: 'rtmps://x.global-contribute.live-video.net', stream_key: 'sk' }), 'rtmps://x.global-contribute.live-video.net/app/sk');
+ const srt = new URL(ff.buildDestUrl({ platform: 'custom', server_url: 'srt://srt.example:9000', stream_key: 'sid', srt_latency_ms: 200, srt_passphrase: 'passphrase123' }));
+ assert.strictEqual(srt.searchParams.get('streamid'), 'sid');
+ assert.strictEqual(srt.searchParams.get('latency'), '200000');
+ assert.strictEqual(srt.searchParams.get('mode'), 'caller');
+ assert.strictEqual(srt.searchParams.get('pkt_size'), '1316');
+ const args = ff.withProgress(ff.rtmpCopyArgs('http://127.0.0.1:1/live/ses_x.flv', 'rtmp://dest/app/key'));
+ assert.deepStrictEqual(args, ['-hide_banner', '-progress', 'pipe:1', '-stats_period', '1', '-loglevel', 'warning', '-rw_timeout', '10000000',
+ '-i', 'http://127.0.0.1:1/live/ses_x.flv', '-c', 'copy', '-fflags', '+nobuffer+discardcorrupt', '-muxdelay', '0', '-muxpreload', '0',
+ '-flush_packets', '1', '-max_muxing_queue_size', '4096', '-rtmp_live', 'live', '-f', 'flv', '-flvflags', 'no_duration_filesize', 'rtmp://dest/app/key']);
+ assert.deepStrictEqual(ff.outputArgs('srt://h:1?streamid=a').slice(-5), ['-f', 'mpegts', '-mpegts_flags', '+resend_headers', 'srt://h:1?streamid=a']);
+ assert.strictEqual(ff.redactUrl('rtmps://live.twitch.tv/app/live_123456789'), 'rtmps://live.twitch.tv/app/****6789');
+ assert.ok(!ff.redactUrl('srt://h:1?streamid=secret&passphrase=pw12345678').includes('secret'));
+ assert.match(ff.friendlyError('Connection refused', 'twitch'), /Twitch's ingest server/);
+ assert.match(ff.friendlyError('NetStream.Publish.BadName', 'youtube'), /Youtube rejected the stream key/);
+});
+
+let api;
+const cookie = `ov_token=${userToken({ subjectId: OWNER })}`;
+
+t('the UI works without JavaScript: landing, sign-in prompt, streams', async () => {
+ api = await bootApi();
+ const home = await request(api.base, 'GET', '/');
+ assert.strictEqual(home.status, 200);
+ assert.match(home.text, /OpenRe\.Stream/);
+ assert.match(home.text, /theme-loader\.js/);
+ assert.match(home.text, /<noscript><nav/);
+ const signin = await request(api.base, 'GET', '/streams');
+ assert.strictEqual(signin.status, 401);
+ assert.match(signin.text, /Sign in with OpenVibe/);
+ const mine = await request(api.base, 'GET', '/streams', { cookie });
+ assert.strictEqual(mine.status, 200);
+ assert.match(mine.text, /New stream/);
+});
+
+t('forms need the CSRF token; creating a stream shows the key once', async () => {
+ const noCsrf = await request(api.base, 'POST', '/streams', { cookie, form: { title: 'Nope' } });
+ assert.strictEqual(noCsrf.status, 403);
+ const token = cookie.slice('ov_token='.length);
+ const created = await request(api.base, 'POST', '/streams', { cookie, form: { title: 'Form stream', recording_mode: 'none', recording_visibility: 'public', _csrf: csrfFor(token) } });
+ assert.strictEqual(created.status, 200);
+ assert.strictEqual(created.headers.get('cache-control'), 'no-store');
+ const key = /(ork_[A-Za-z0-9_-]{43})/.exec(created.text)[1];
+ const def = api.rt.store.definitions.list({ owner_subject: OWNER })[0];
+ assert.strictEqual(api.rt.store.definitions.resolveIngestKey(key, 'rtmp').definition.id, def.id);
+ const page = await request(api.base, 'GET', `/streams/${def.id}`, { cookie });
+ assert.strictEqual(page.status, 200);
+ assert.ok(!page.text.includes(key), 'the stream page never shows the key again');
+ assert.match(page.text, new RegExp(`ork_…${key.slice(-4)}`));
+ // Destination add through the form; the key is write-only.
+ const add = await request(api.base, 'POST', `/streams/${def.id}/destinations`, { cookie, form: { platform: 'twitch', name: 'T', server_url: 'rtmps://live.twitch.tv/app', stream_key: 'live_abcdef_9876', enabled: '1', auto_start: '1', _csrf: csrfFor(token) } });
+ assert.strictEqual(add.status, 303);
+ const after = await request(api.base, 'GET', `/streams/${def.id}`, { cookie });
+ assert.ok(!after.text.includes('live_abcdef_9876'));
+ assert.match(after.text, /\*\*\*\*9876/);
+ // Someone else cannot open it.
+ const other = await request(api.base, 'GET', `/streams/${def.id}`, { cookie: `ov_token=${userToken({ subjectId: OTHER })}` });
+ assert.strictEqual(other.status, 404);
+ // Rotation through the form shows the new key once.
+ const rotated = await request(api.base, 'POST', `/streams/${def.id}/rotate`, { cookie, form: { grace_seconds: '0', _csrf: csrfFor(token) } });
+ const key2 = /(ork_[A-Za-z0-9_-]{43})/.exec(rotated.text)[1];
+ assert.notStrictEqual(key2, key);
+ assert.strictEqual(api.rt.store.definitions.resolveIngestKey(key, 'rtmp').error, 'revoked_key');
+ await api.close();
+});
+
+t.run();
new file mode 100644
new file mode 100644
@@ -0,0 +1,154 @@
@@ -0,0 +1,154 @@
+'use strict';
+// Migration from a Live DB snapshot: read-only, new keys (never the old ones), destinations
+// sealed, held/excluded rows recorded with reasons, idempotent re-runs, and the checklist.
+const assert = require('assert');
+const fs = require('fs');
+const path = require('path');
+const { execFileSync } = require('child_process');
+const Database = require('better-sqlite3');
+const { runtime, tmpDir, testEnv, suite, ROOT } = require('./helpers');
+const { migrate, checklist } = require('../server/migrate-live');
+const { hashIngestKey } = require('../server/secrets');
+
+const t = suite('migration');
+const SUBJECT_A = 'usr_01J00000000000000000000A01';
+const SUBJECT_B = 'usr_01J00000000000000000000B01';
+
+function liveSnapshot(dir) {
+ const file = path.join(dir, 'live-snapshot.db');
+ const db = new Database(file);
+ db.exec(`
+ CREATE TABLE users (id INTEGER PRIMARY KEY, username TEXT, display_name TEXT, password_hash TEXT, stream_key TEXT, is_banned INTEGER DEFAULT 0);
+ CREATE TABLE channels (id INTEGER PRIMARY KEY, user_id INTEGER, vod_recording_enabled INTEGER DEFAULT 1, force_vod_recording_disabled INTEGER DEFAULT 0, default_vod_visibility TEXT DEFAULT 'public');
+ CREATE TABLE linked_accounts (id INTEGER PRIMARY KEY, user_id INTEGER, service TEXT, service_user_id TEXT, subject_id TEXT);
+ CREATE TABLE managed_streams (id INTEGER PRIMARY KEY, user_id INTEGER, slug TEXT, title TEXT, description TEXT, protocol TEXT, stream_key TEXT,
+ streaming_method TEXT, default_vod_visibility TEXT, slot_vod_recording_enabled INTEGER DEFAULT 1, slot_clip_recording_enabled INTEGER DEFAULT 1);
+ CREATE TABLE restream_destinations (id INTEGER PRIMARY KEY, user_id INTEGER, managed_stream_id INTEGER, platform TEXT, name TEXT, server_url TEXT,
+ stream_key TEXT, enabled INTEGER, auto_start INTEGER, quality_preset TEXT, connection_id INTEGER, srt_latency_ms INTEGER, srt_passphrase TEXT,
+ custom_video_bitrate INTEGER, custom_audio_bitrate INTEGER, custom_fps INTEGER, custom_encoder_preset TEXT);
+ INSERT INTO users VALUES (1, 'japaneseoldguy', 'JOG', 'x', 'personalkey111', 0), (2, 'nosubject', 'NS', 'x', 'k2', 0), (3, 'banned', 'B', 'x', 'k3', 1), (4, 'twoslots', 'TS', 'x', 'k4', 0);
+ INSERT INTO channels (user_id, vod_recording_enabled, default_vod_visibility) VALUES (1, 1, 'unlisted'), (4, 0, 'public');
+ INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (1, 'network', '57', '${SUBJECT_A}'), (3, 'network', '58', 'usr_01J00000000000000000000C01'), (4, 'network', '59', '${SUBJECT_B}');
+ INSERT INTO managed_streams VALUES
+ (10, 1, 'main', 'Morning walk', '', 'rtmp', 'a3f9c2leakedkey0000000000000000000000000', 'obs', NULL, 1, 1),
+ (11, 2, NULL, 'No subject yet', '', 'rtmp', 'b000000000000000000000000000000000000000', NULL, NULL, 1, 1),
+ (12, 3, NULL, 'Banned', '', 'rtmp', 'c000000000000000000000000000000000000000', NULL, NULL, 1, 1),
+ (13, 4, 'a', 'Slot A', '', 'webrtc', 'd000000000000000000000000000000000000000', 'browser', NULL, 1, 0),
+ (14, 4, 'b', 'Slot B', '', 'rtmp', 'e000000000000000000000000000000000000000', 'obs', 'private', 1, 1);
+ INSERT INTO restream_destinations VALUES
+ (100, 1, 10, 'twitch', 'Twitch', 'rtmps://live.twitch.tv/app', 'live_twitchsecret_1', 1, 1, 'auto', NULL, NULL, NULL, NULL, NULL, NULL, NULL),
+ (101, 1, 10, 'youtube', 'YT', 'rtmp://a.rtmp.youtube.com/live2', 'yt-secret-2222', 1, 0, 'high', 7, NULL, NULL, 4500, NULL, NULL, NULL),
+ (102, 1, 10, 'custom', 'LAN box', 'rtmp://192.168.1.20/live', 'lan-secret-333', 1, 1, 'auto', NULL, NULL, NULL, NULL, NULL, NULL, NULL),
+ (103, 4, NULL, 'kick', 'Kick', 'rtmps://fa723fc1b171.global-contribute.live-video.net', 'kick-secret-44', 1, 1, 'auto', NULL, NULL, NULL, NULL, NULL, NULL, NULL);
+ `);
+ db.close();
+ return file;
+}
+
+t('dry run writes nothing', () => {
+ const dir = tmpDir();
+ const rt = runtime({ dir });
+ const live = new Database(liveSnapshot(dir), { readonly: true });
+ const report = migrate({ liveDb: live, rt, apply: false });
+ assert.strictEqual(rt.db.prepare('SELECT COUNT(*) AS n FROM stream_definitions').get().n, 0);
+ assert.strictEqual(rt.db.prepare('SELECT COUNT(*) AS n FROM migration_map').get().n, 0);
+ assert.ok(report.counts.imported > 0);
+ assert.match(checklist(report), /dry run/);
+});
+
+let dir;
+let rt;
+let liveFile;
+t('apply: slots with a subject are imported with new keys; the rest are held or excluded with reasons', () => {
+ dir = tmpDir();
+ rt = runtime({ dir, env: { OPENRE_DEST_ALLOW_PRIVATE: '0' } });
+ liveFile = liveSnapshot(dir);
+ const before = fs.readFileSync(liveFile);
+ const live = new Database(liveFile, { readonly: true });
+ const report = migrate({ liveDb: live, rt, apply: true });
+ live.close();
+ assert.ok(before.equals(fs.readFileSync(liveFile)), 'the Live snapshot is untouched');
+
+ const map = Object.fromEntries(rt.db.prepare('SELECT source_type, source_id, status, reason, target_id FROM migration_map').all().map(r => [`${r.source_type}:${r.source_id}`, r]));
+ assert.strictEqual(map['managed_stream:10'].status, 'imported');
+ assert.strictEqual(map['managed_stream:11'].status, 'held');
+ assert.match(map['managed_stream:11'].reason, /no canonical subject/);
+ assert.strictEqual(map['managed_stream:12'].status, 'excluded');
+ assert.strictEqual(map['managed_stream:13'].status, 'imported');
+ assert.strictEqual(map['managed_stream:14'].status, 'imported');
+
+ const def10 = rt.store.definitions.get(map['managed_stream:10'].target_id);
+ assert.strictEqual(def10.owner_subject, SUBJECT_A);
+ assert.strictEqual(def10.recording_visibility, 'unlisted', 'from the channel default');
+ assert.strictEqual(def10.recording_mode, 'vod');
+ assert.deepStrictEqual(def10.external_refs.map(r => `${r.service}:${r.type}:${r.id}`), ['live:managed_stream:10', 'live:user:1']);
+ assert.strictEqual(rt.store.definitions.get(map['managed_stream:13'].target_id).recording_mode, 'none', 'VOD off on the channel, clips off on the slot');
+ assert.strictEqual(rt.store.definitions.get(map['managed_stream:14'].target_id).recording_mode, 'clips');
+ assert.strictEqual(rt.store.definitions.get(map['managed_stream:14'].target_id).recording_visibility, 'private');
+
+ // Old keys: none of them authenticates, and none is stored in any form.
+ for (const old of ['a3f9c2leakedkey0000000000000000000000000', 'personalkey111']) {
+ assert.strictEqual(rt.db.prepare('SELECT COUNT(*) AS n FROM ingest_keys WHERE key_hash = ?').get(hashIngestKey(old)).n, 0);
+ assert.ok(rt.store.definitions.resolveIngestKey(old, 'rtmp').error);
+ }
+ assert.strictEqual(rt.db.prepare("SELECT COUNT(*) AS n FROM ingest_keys WHERE status = 'active'").get().n, 3, 'one new key per imported slot');
+
+ // Destinations: sealed, LAN one held, OAuth-linked flagged, unbound one held (two slots).
+ assert.strictEqual(map['restream_destination:100'].status, 'imported');
+ assert.strictEqual(map['restream_destination:101'].status, 'imported');
+ assert.strictEqual(map['restream_destination:102'].status, 'held');
+ assert.match(map['restream_destination:102'].reason, /private/);
+ assert.strictEqual(map['restream_destination:103'].status, 'held');
+ assert.match(map['restream_destination:103'].reason, /several slots/);
+ const d100 = rt.store.outputs.destinationForWorker(map['restream_destination:100'].target_id);
+ assert.strictEqual(d100.stream_key, 'live_twitchsecret_1');
+ assert.ok(!JSON.stringify(rt.db.prepare('SELECT * FROM destinations').all()).includes('live_twitchsecret_1'), 'sealed at rest');
+ const d101 = rt.store.outputs.getDestination(map['restream_destination:101'].target_id);
+ assert.strictEqual(d101.auto_start, false);
+ assert.strictEqual(d101.custom_video_bitrate, 4500);
+ const d102 = rt.store.outputs.getDestination(map['restream_destination:102'].target_id);
+ assert.strictEqual(d102.enabled, false);
+ assert.ok(d102.hold_reason);
+
+ const md = checklist(report, { rtmpUrl: 'rtmp://ingest.openre.stream:1936/live' });
+ assert.match(md, /## @japaneseoldguy/);
+ assert.match(md, /ingest-authority/);
+ assert.match(md, /OAuth-linked/);
+ assert.match(md, /only RTMP moves to OpenRe/, 'the WebRTC slot is flagged');
+ for (const secret of ['a3f9c2leakedkey', 'live_twitchsecret_1', 'yt-secret-2222', 'personalkey111', 'kick-secret-44']) assert.ok(!md.includes(secret), `no ${secret} in the checklist`);
+});
+
+t('re-running is idempotent; a held slot is imported once its user has a subject', () => {
+ const counts = () => ({
+ defs: rt.db.prepare('SELECT COUNT(*) AS n FROM stream_definitions').get().n,
+ dests: rt.db.prepare('SELECT COUNT(*) AS n FROM destinations').get().n,
+ keys: rt.db.prepare('SELECT COUNT(*) AS n FROM ingest_keys').get().n,
+ });
+ const before = counts();
+ let live = new Database(liveFile, { readonly: true });
+ const again = migrate({ liveDb: live, rt, apply: true });
+ live.close();
+ assert.deepStrictEqual(counts(), before);
+ assert.strictEqual(again.counts.imported, 0);
+ const w = new Database(liveFile);
+ w.prepare("INSERT INTO linked_accounts (user_id, service, service_user_id, subject_id) VALUES (2, 'network', '60', 'usr_01J00000000000000000000D01')").run();
+ w.close();
+ live = new Database(liveFile, { readonly: true });
+ migrate({ liveDb: live, rt, apply: true });
+ live.close();
+ assert.strictEqual(counts().defs, before.defs + 1);
+ assert.strictEqual(rt.db.prepare("SELECT status FROM migration_map WHERE source_type = 'managed_stream' AND source_id = '11'").get().status, 'imported');
+});
+
+t('the CLI runs a dry run against a snapshot and prints the checklist', () => {
+ const d = tmpDir();
+ const file = liveSnapshot(d);
+ const env = testEnv(d);
+ const out = execFileSync(process.execPath, [path.join(ROOT, 'scripts', 'migrate-from-live.js'), '--live-db', file], {
+ env: { ...process.env, ...env, OPENRE_ENV_FILE: '/nonexistent' }, cwd: ROOT, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'],
+ });
+ assert.match(out, /OpenRe RTMP cutover checklist \(dry run/);
+ assert.match(out, /@twoslots/);
+});
+
+t.run();
new file mode 100644
new file mode 100644
@@ -0,0 +1,155 @@
@@ -0,0 +1,155 @@
+'use strict';
+// Recording requests (Media contract as Live's recorder uses it) and the event relay (outbox →
+// OpenVibe.Events with OpenRe's service token), with stub Media / Network / Events servers.
+const assert = require('assert');
+const http = require('http');
+const { runtime, manualClock, suite, OWNER, silent, outboxTypes } = require('./helpers');
+const { createCoordinator } = require('../server/coordinator');
+const { createMediaClient } = require('../server/media-client');
+
+const t = suite('recording-events');
+
+function stub(handler) {
+ return new Promise((resolve) => {
+ const s = http.createServer((req, res) => {
+ let body = '';
+ req.on('data', (c) => { body += c; });
+ req.on('end', () => { let parsed = null; try { parsed = body ? JSON.parse(body) : null; } catch { parsed = null; } handler(req, res, parsed, body); });
+ });
+ s.listen(0, '127.0.0.1', () => resolve({ url: `http://127.0.0.1:${s.address().port}`, close: () => new Promise(r => { s.closeAllConnections(); s.close(r); }) }));
+ });
+}
+const reply = (res, status, obj) => { res.statusCode = status; res.setHeader('Content-Type', 'application/json'); res.end(JSON.stringify(obj || {})); };
+
+async function setup({ media: mediaHandler, recording_mode = 'vod' }) {
+ const calls = [];
+ const media = await stub((req, res, body) => { calls.push({ method: req.method, url: req.url, body, auth: req.headers.authorization }); mediaHandler(req, res, body, calls); });
+ const clock = manualClock();
+ const rt = runtime({ clock, env: { MEDIA_URL: media.url, MEDIA_API_KEY: 'k-live', OPENRE_RECORDING_START_DELAY_MS: '0' } });
+ const client = createMediaClient({ config: rt.config });
+ const coordinator = createCoordinator({ rt, media: client, log: silent });
+ const { definition, key } = rt.store.definitions.create({ owner_subject: OWNER, title: 'Rec', recording_mode, external_refs: [{ service: 'live', type: 'user', id: '8' }] });
+ const w = rt.store.workers.register({ kind: 'rtmp-ingest', endpoints: { rtmpPlayPort: 19390, flvPort: 19391 } });
+ rt.store.workers.ready(w.id);
+ const a = rt.store.sessions.admit({ definition, key, protocol: 'rtmp', worker: rt.store.workers.get(w.id) });
+ rt.store.sessions.transition(a.session.id, 'live');
+ return { rt, clock, coordinator, session: a.session, calls, media };
+}
+
+t('a live session gets one recording request; finalize when it ends (Media finalises)', async () => {
+ const { rt, coordinator, session, calls, media } = await setup({
+ media: (req, res) => {
+ if (req.url === '/api/v1/live/vods') return reply(res, 201, { id: 501 });
+ if (req.url === '/api/v1/live/vods/501/ingest/rtmp') return reply(res, 202, { status: 'recording' });
+ if (req.url === '/api/v1/live/vods/501/finalize') return reply(res, 200, {});
+ return reply(res, 404);
+ },
+ });
+ await coordinator.tick();
+ await coordinator.tick();
+ assert.deepStrictEqual(calls.map(c => `${c.method} ${c.url}`), ['POST /api/v1/live/vods', 'POST /api/v1/live/vods/501/ingest/rtmp']);
+ assert.strictEqual(calls[0].auth, 'Bearer k-live');
+ assert.strictEqual(calls[0].body.user_id, 8);
+ assert.strictEqual(calls[0].body.visibility, 'public');
+ assert.strictEqual(calls[1].body.rtmp_url, `rtmp://127.0.0.1:19390/live/${session.id}`);
+ assert.strictEqual(rt.store.recordings.bySession(session.id).state, 'recording');
+ assert.deepStrictEqual(outboxTypes(rt.db).filter(x => x.startsWith('openre.recording')), ['openre.recording.requested']);
+ rt.store.sessions.finish(session.id, { reason: 'publisher_disconnected' });
+ await coordinator.tick();
+ assert.strictEqual(calls.pop().url, '/api/v1/live/vods/501/finalize');
+ assert.strictEqual(rt.store.recordings.bySession(session.id).state, 'finalized');
+ await media.close();
+});
+
+t('Media refusing for disk space: the empty VOD shell is deleted and the request retried later', async () => {
+ let refuse = true;
+ const { rt, clock, coordinator, session, calls, media } = await setup({
+ media: (req, res) => {
+ if (req.method === 'POST' && req.url === '/api/v1/live/vods') return reply(res, 201, { id: 600 + calls.length });
+ if (req.url.endsWith('/ingest/rtmp')) return refuse ? reply(res, 409, { error: 'Disk critically low — recording refused' }) : reply(res, 202, {});
+ if (req.method === 'DELETE') return reply(res, 200, {});
+ return reply(res, 404);
+ },
+ });
+ await coordinator.tick();
+ assert.deepStrictEqual(calls.map(c => c.method), ['POST', 'POST', 'DELETE'], 'create, ingest refused, shell deleted');
+ const rec = rt.store.recordings.bySession(session.id);
+ assert.strictEqual(rec.state, 'pending');
+ assert.match(rec.last_error, /Disk/);
+ assert.ok(rec.next_attempt_at >= clock.now() + 5 * 60 * 1000, 'disk refusals wait five minutes');
+ refuse = false;
+ clock.advance(5 * 60 * 1000 + 1);
+ rt.store.workers.heartbeat(session.worker_id); // the ingest worker is alive throughout
+ await coordinator.tick();
+ assert.strictEqual(rt.store.recordings.bySession(session.id).state, 'recording');
+ await media.close();
+});
+
+t('clips-only recordings are deleted after finalising; a session that ends first is never requested', async () => {
+ const { rt, coordinator, session, calls, media } = await setup({
+ recording_mode: 'clips',
+ media: (req, res) => {
+ if (req.url === '/api/v1/live/vods') return reply(res, 201, { id: 700 });
+ if (req.method === 'DELETE') return reply(res, 200, {});
+ return reply(res, req.url.endsWith('/finalize') ? 200 : 202, {});
+ },
+ });
+ await coordinator.tick();
+ assert.strictEqual(calls[0].body.clips_only, true);
+ rt.store.sessions.finish(session.id, { reason: 'x' });
+ await coordinator.tick();
+ assert.deepStrictEqual(calls.slice(-2).map(c => `${c.method} ${c.url}`), ['POST /api/v1/live/vods/700/finalize', 'DELETE /api/v1/live/vods/700']);
+ await media.close();
+
+ const s2 = await setup({ media: (req, res) => reply(res, 500) });
+ s2.rt.store.sessions.finish(s2.session.id, { reason: 'gone' });
+ s2.rt.db.prepare("INSERT INTO recordings (id, session_id, mode, state, created_at, updated_at) VALUES ('rec_01J0000000000000000000000A', ?, 'vod', 'pending', 0, 0)").run(s2.session.id);
+ await s2.coordinator.tick();
+ assert.strictEqual(s2.rt.store.recordings.bySession(s2.session.id).state, 'cancelled');
+ assert.strictEqual(s2.calls.length, 0);
+ await s2.media.close();
+});
+
+t('the relay publishes outbox rows with a Network service token; Events down means rows wait', async () => {
+ let down = true;
+ const got = [];
+ const events = await stub((req, res, body) => {
+ if (down) return reply(res, 503, { code: 'events.unavailable' });
+ assert.strictEqual(req.headers.authorization, 'Bearer svc-token');
+ const list = body.events || [body];
+ list.forEach(e => got.push(e));
+ return reply(res, 200, body.events ? { results: list.map((e, i) => ({ event_id: e.event_id, seq: i + 1 })) } : { event_id: list[0].event_id, seq: 1 });
+ });
+ const network = await stub((req, res, body, raw) => {
+ const p = new URLSearchParams(raw);
+ assert.strictEqual(p.get('grant_type'), 'client_credentials');
+ assert.strictEqual(p.get('client_id'), 'openre');
+ assert.strictEqual(p.get('audience'), 'openvibe.events');
+ reply(res, 200, { access_token: 'svc-token', token_type: 'Bearer', expires_in: 300 });
+ });
+ const rt = runtime({ env: { EVENTS_URL: events.url, OV_NETWORK_INTERNAL_URL: network.url, OV_OAUTH_CLIENT_SECRET: 's3cret' } });
+ assert.strictEqual(rt.events.configured, true);
+ const { definition } = rt.store.definitions.create({ owner_subject: OWNER });
+ rt.store.definitions.rotateKey(definition.id, { rotated_by: OWNER });
+ await rt.events.outbox.flush();
+ assert.strictEqual(rt.events.outbox.pending(), 1, 'kept while Events is down');
+ down = false;
+ rt.db.prepare('UPDATE event_outbox SET next_attempt_at = 0').run();
+ await rt.events.outbox.flush();
+ assert.strictEqual(rt.events.outbox.pending(), 0);
+ assert.strictEqual(got[0].event_type, 'openre.key.rotated');
+ assert.strictEqual(got[0].source, 'openre');
+ await events.close();
+ await network.close();
+});
+
+t('without EVENTS_URL the relay is off but every row is still written with its change', () => {
+ const rt = runtime();
+ assert.strictEqual(rt.events.configured, false);
+ assert.strictEqual(rt.events.start(), false);
+ const { definition } = rt.store.definitions.create({ owner_subject: OWNER });
+ rt.store.definitions.rotateKey(definition.id);
+ assert.strictEqual(rt.events.status().pending, 1);
+});
+
+t.run();