openvibers/openvibe.media Commits
Media's Events outbox row is written in the state change's transaction: webhooks.announce() runs the change (VOD ready/quarantine/delete, clip ready/failed, object ready + invariant row) and events.record() in ONE SQLite transaction, then wakes the relay and sends the webhook, so an outcome is never lost (committed without its event) or a phantom (event for a rolled-back change); a failed outbox insert rolls the change back. The thumbnail is generated before the ready commit so the payload still carries it. Webhook bodies carry event_id = the durable event's id, so an app reading both paths (Live's webhook → Events transition) handles each outcome once. Lifecycle events are priority important (roadmap §6.3; storage.recovered stays low). test/outbox-transaction.test.js covers announce, outbox-insert failure, change failure, and the clip-jobs and finalize call sites
Changes
@@@ -245,10 +245,18 @@ directories are shared across apps) and responses carry a `note` saying so.### Webhooks (outbound)`POST` to the app's `webhook_url` with body--`{ "event": "vod.ready"|"vod.failed"|"clip.ready"|"clip.failed"|"media.object.uploaded", "app_id", "data" }`++`{ "event": "vod.ready"|"vod.failed"|"clip.ready"|"clip.failed"|"media.object.uploaded"|"storage.alert"|"storage.recovered", "app_id", "data", "event_id"? }`and header `X-OVMedia-Signature: sha256=<hex hmac-sha256 of the raw body withthe app's webhook_secret>`. 3 attempts with backoff, 10 s timeout.++Each outcome is also a durable OpenVibe.Events event (`media.vod.ready|failed`,++`media.clip.ready|failed`, `media.object.uploaded`, `media.storage.alert|recovered`;++`server/events.js`). The outbox row is written in the **same SQLite transaction** as the state++change it describes (`webhooks.announce()`), so an outcome is never lost or announced for a++rolled-back change; the webhook follows the commit and carries `event_id` = that event's id++(absent when the outbox is off), so an app that reads both paths (Live during its webhook →++Events transition) handles each outcome once. Webhooks stay until every consumer reads Events.++## Health, readiness and metrics| path | behavior |@@@ -10,10 +10,14 @@* media.object.uploaded → media.object.uploaded (subject object <id>)* storage.alert | storage.recovered → media.storage.alert | media.storage.recovered*-- * The event is enqueued right after the state change commits (its own SQLite transaction, a local-- * insert): the loss window is a process crash between those two statements. Delivery from the-- * outbox on is at least once. Every event is `internal` visibility: a VOD's own visibility decides-- * who may see it, and consumers (Live, Search) apply it.++ * The outbox row is written INSIDE the SQLite transaction that makes the state change it describes++ * (record(), called from webhooks.announce()): the event exists if and only if the change committed,++ * so a crash can neither lose an outcome nor announce one that rolled back. The webhook for the same++ * outcome is sent after the commit and carries the envelope's event_id, so a consumer receiving it by++ * both paths (Live during its webhook → Events transition) handles it once. Delivery from the outbox++ * on is at least once. Every event is `internal` visibility: a VOD's own visibility decides who may++ * see it, and consumers (Live, Search) apply it. Lifecycle outcomes are `important` (roadmap §6.3);++ * only storage.recovered is `low`.** Off unless EVENTS_URL and OV_OAUTH_CLIENT_SECRET are set (EVENTS_PUBLISH=off disables it).*/@@@ -69,8 +73,13 @@ function slim(appId, data) {return { app_id: appId || null, ...rest };}--/** Queue the durable twin of a webhook event. Never throws; returns the envelope or null. */--function emit(webhookEvent, appId, data) {++/**++ * Queue the durable event for an outcome. MUST run inside the transaction that makes the change++ * (the SDK outbox refuses otherwise), and throws if the insert fails, so the change rolls back with++ * it. Returns the envelope (with event_id), or null when the outbox is off, the tenant is a sandbox++ * or the webhook event has no durable twin.++ */++function record(webhookEvent, appId, data) {if (!outbox) return null;// Developer-project sandbox tenants (ADR-014) produce no platform events: sandbox activity must// never reach production consumers.@@@ -82,25 +91,21 @@ function emit(webhookEvent, appId, data) {const subject = subjectType === 'storage'? { type: 'storage', id: String((data && data.kind) || 'storage') }: { type: subjectType, id: String(id == null ? 'unknown' : id) };-- try {-- let env = null;-- db.getDb().transaction(() => {-- env = outbox.enqueue({-- event_type: eventType,-- actor: { type: 'service', id: 'media' },-- subject,-- visibility: 'internal',-- priority: /failed|alert/.test(eventType) ? 'important' : 'low',-- payload: slim(appId, data),-- });-- })();-- stats.queued++;-- setImmediate(() => outbox && outbox.kick());-- return env;-- } catch (err) {-- console.warn(`[Events] ${eventType} not queued:`, err.message);-- return null;-- }++ const env = outbox.enqueue({++ event_type: eventType,++ actor: { type: 'service', id: 'media' },++ subject,++ visibility: 'internal',++ priority: eventType === 'media.storage.recovered' ? 'low' : 'important',++ payload: slim(appId, data),++ });++ stats.queued++;++ return env;++}++++/** Wake the relay once the transaction that queued events has committed. */++function kick() {++ if (outbox) setImmediate(() => outbox && outbox.kick());}function status() {@@@ -110,4 +115,4 @@ function status() {function _reset() { if (outbox) outbox.stop(); outbox = null; stats.queued = 0; stats.lastError = null; }--module.exports = { init, emit, status, TYPES, _reset };++module.exports = { init, record, kick, status, TYPES, _reset };@@@ -37,7 +37,7 @@ const model = require('./model');const invariant = require('./invariant');const signing = require('./signing');const { tenantAuth, tenantCors, tenantPath } = require('../auth');--const { sendWebhook } = require('../webhooks');++const { announce } = require('../webhooks');const MB = 1024 * 1024;const upload = tenantAuth({ capability: 'media.object.upload' });@@@ -259,11 +259,15 @@ router.post('/:id/complete', contentAuth, tenantCors, (req, res) => {return problem(res, 422, 'media.invariant.public_object_too_large', `Public playback objects are limited to ${config.objects.publicMaxMb} MB`);}if (!quotaCheck(req, res, Number(obj.size_bytes) || 0, obj.id)) return;-- model.updateObject(obj.id, { lifecycle_status: 'ready' });-- const done = model.getObject(obj.id);-- invariant.record(done);-- const body = model.objectPublic(done);-- sendWebhook(req.appId, 'media.object.uploaded', body).catch(() => {});++ // The ready transition, its invariant row and the media.object.uploaded event commit++ // together; the webhook follows the commit.++ const { data: body } = announce(req.appId, 'media.object.uploaded', {++ change: () => {++ model.updateObject(obj.id, { lifecycle_status: 'ready' });++ invariant.record(model.getObject(obj.id));++ },++ payload: () => model.objectPublic(model.getObject(obj.id)),++ });res.json(body);} catch (err) {console.error('[Objects] complete error:', err.message);@@@ -3,7 +3,7 @@** recutClip(clipId, { reason }) the ONE code path that (re)cuts an existing clip row:* resolves the source, cuts, updates the row, thumbnails,-- * fires the webhook. Records cut_error / cut_attempts and++ * commits the outcome with its event, fires the webhook. Records cut_error / cut_attempts and* schedules an automatic retry on failure.* start() sweeper: every 3 min, failed clips that still have* attempts left are re-cut. From the second attempt on, a@@@ -19,7 +19,7 @@ const db = require('../db/database');const config = require('../config');const cutter = require('./clip-cutter');const vodStorage = require('./vod-storage');--const { sendWebhook } = require('../webhooks');++const { announce } = require('../webhooks');const MAX_ATTEMPTS = 4;const BACKOFF_MIN = [2, 10, 30, 90]; // minutes before attempt 2, 3, 4, …@@@ -75,12 +75,16 @@ async function _recutClip(clipId, { reason = 'recut' } = {}) {const duration = Math.max(1, (Number(clip.end_time) || 0) - startTime);const cut = await cutter.cutClipFile({ source: source.value, startTime, duration });if (!cut.ok) return fail(clipId, clip, attempt, cut.error);-- db.run("UPDATE clips SET file_path = ?, duration_seconds = ?, end_time = ?, status = 'ready', cut_error = NULL, cut_next_at = NULL WHERE id = ?",-- [cut.filePath, cut.duration, startTime + cut.duration, clipId]);-- require('../objects/model').safeSync('clip', clipId);++ // Thumbnail first (the clip.ready payload carries it), then the ready transition and its event++ // in one transaction (webhooks.announce), then the webhook.try { await require('../thumbnails/thumbnail-service').generateClipThumbnail(clipId, cut.filePath); } catch { /* */ }++ announce(clip.app_id, 'clip.ready', {++ change: () => db.run("UPDATE clips SET file_path = ?, duration_seconds = ?, end_time = ?, status = 'ready', cut_error = NULL, cut_next_at = NULL WHERE id = ?",++ [cut.filePath, cut.duration, startTime + cut.duration, clipId]),++ payload: () => _clipPublic(db.getClipById(clipId)),++ });++ require('../objects/model').safeSync('clip', clipId);console.log(`[Clips] Clip ${clipId} ${reason} OK from vod ${clip.vod_id} (${startTime.toFixed(1)}-${(startTime + cut.duration).toFixed(1)}s, attempt ${attempt})`);-- sendWebhook(clip.app_id, 'clip.ready', _clipPublic(db.getClipById(clipId))).catch(() => {});return { ok: true };}@@@ -92,10 +96,12 @@ function fail(clipId, clip, attempt, error) {const more = attempt < MAX_ATTEMPTS && !hopeless;const mins = BACKOFF_MIN[Math.min(attempt - 1, BACKOFF_MIN.length - 1)];const nextAt = more ? new Date(Date.now() + mins * 60000).toISOString().replace('T', ' ').slice(0, 19) : null;-- db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_next_at = ? WHERE id = ?", [msg, nextAt, clipId]);++ announce(clip.app_id, 'clip.failed', {++ change: () => db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_next_at = ? WHERE id = ?", [msg, nextAt, clipId]),++ payload: () => _clipPublic(db.getClipById(clipId)),++ });require('../objects/model').safeSync('clip', clipId);console.warn(`[Clips] Clip ${clipId} attempt ${attempt}/${MAX_ATTEMPTS} failed: ${msg}${more ? ` — retry in ${mins} min` : ' — giving up'}`);-- sendWebhook(clip.app_id, 'clip.failed', _clipPublic(db.getClipById(clipId))).catch(() => {});return { ok: false, error: msg, retry_at: nextAt };}@@@ -22,7 +22,7 @@ const db = require('../db/database');const tools = require('./media-tools');const cutter = require('./clip-cutter');const { tenantAuth, tenantCors } = require('../auth');--const { sendWebhook } = require('../webhooks');++const { announce } = require('../webhooks');const objects = require('../objects/model');const router = express.Router({ mergeParams: true });@@@ -205,22 +205,28 @@ router.post('/', tenantAuth({ allowUser: true }), clipUpload.single('video'), as(async () => {const cut = await cutter.cutClipFile({ source: source.value, startTime, duration });if (cut.ok) {-- db.run('UPDATE clips SET file_path = ?, duration_seconds = ?, end_time = ?, status = ? WHERE id = ?',-- [cut.filePath, cut.duration, startTime + cut.duration, 'ready', clipId]);-- objects.safeSync('clip', clipId);++ // Thumbnail first so the clip.ready payload carries it; then the ready transition++ // and its event commit together (webhooks.announce) and the webhook goes out.try {await require('../thumbnails/thumbnail-service').generateClipThumbnail(clipId, cut.filePath);} catch { /* */ }++ announce(appId, 'clip.ready', {++ change: () => db.run('UPDATE clips SET file_path = ?, duration_seconds = ?, end_time = ?, status = ? WHERE id = ?',++ [cut.filePath, cut.duration, startTime + cut.duration, 'ready', clipId]),++ payload: () => clipPublic(db.getClipById(clipId)),++ });++ objects.safeSync('clip', clipId);console.log(`[Clips] Clip ${clipId} cut from vod ${vodId} (${startTime.toFixed(1)}-${(startTime + cut.duration).toFixed(1)}s)`);-- sendWebhook(appId, 'clip.ready', clipPublic(db.getClipById(clipId))).catch(() => {});} else {// First attempt failed: record why and let the retry sweeper take it from here// (attempt 2 pulls a cloud-stored VOD back to local disk first).const nextAt = new Date(Date.now() + 2 * 60000).toISOString().replace('T', ' ').slice(0, 19);-- db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_attempts = 1, cut_next_at = ? WHERE id = ?", [String(cut.error || 'cut failed').slice(0, 500), nextAt, clipId]);++ announce(appId, 'clip.failed', {++ change: () => db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_attempts = 1, cut_next_at = ? WHERE id = ?", [String(cut.error || 'cut failed').slice(0, 500), nextAt, clipId]),++ payload: () => clipPublic(db.getClipById(clipId)),++ });objects.safeSync('clip', clipId);console.warn(`[Clips] Clip ${clipId} failed: ${cut.error} — auto-retry in 2 min`);-- sendWebhook(appId, 'clip.failed', clipPublic(db.getClipById(clipId))).catch(() => {});}})().catch(err => console.error('[Clips] Background cut error:', err.message));@@@ -4,7 +4,8 @@* Ported from the predecessor's finalizeVodRecording/_doFinalize, keyed by* vodId (there is no stream registry here — apps own stream state):* merge pending chunk segments → seekable remux → probe → master recovery-- * (truncated-webm rebuild) → DB update → thumbnail → webhook vod.ready.++ * (truncated-webm rebuild) → thumbnail → DB update + vod.ready event (one++ * transaction) → webhook vod.ready.** Guarded against double-invocation; refuses to finalize while the recorder* still holds the file open (stops it gracefully instead — its ffmpeg exit@@@ -18,7 +19,7 @@ const { spawn } = require('child_process');const db = require('../db/database');const config = require('../config');const tools = require('./media-tools');--const { sendWebhook } = require('../webhooks');++const { announce } = require('../webhooks');/** Absolute-ize a stored Media-relative URL (/t/…, /api/thumbnails/…). */function _absUrl(u) {@@@ -102,10 +103,18 @@ function rebuildWebmFromMaster(masterPath, webmPath) {});}--function _webhookForVod(vod, event) {++/**++ * The state change for a VOD outcome and its event/webhook, committed together (webhooks.announce).++ * `row()` gives the row the payload describes (read inside the transaction). Returns the payload,++ * or null when the transaction failed (then nothing changed and nothing was announced).++ */++function _commitVodOutcome(vod, event, change, row = () => db.getVodById(vod.id)) {try {-- sendWebhook(vod.app_id, event, vodPublic(vod)).catch(() => {});-- } catch { /* non-critical */ }++ return announce(vod.app_id, event, { change, payload: () => vodPublic(row()) }).data;++ } catch (err) {++ console.error(`[VOD] vod ${vod.id}: ${event} not committed:`, err.message);++ return null;++ }}/**@@@ -164,8 +173,7 @@ async function _doFinalize(vodId, opts) {// process died before ffmpeg opened the file). Nothing to review — delete// the row so a 0:00 ghost never reaches listings.console.warn(`[VOD] vod ${vodId}: no recording file — deleting empty row`);-- _webhookForVod(vod, 'vod.failed');-- try { db.run('DELETE FROM vods WHERE id = ?', [vodId]); } catch { /* */ }++ _commitVodOutcome(vod, 'vod.failed', () => db.run('DELETE FROM vods WHERE id = ?', [vodId]), () => vod);return null;}@@@ -182,11 +190,11 @@ async function _doFinalize(vodId, opts) {// ready, and a 0:00 ghost must never reach listings.if (tools.getFileSizeSafe(filePath) === 0) {console.warn(`[VOD] vod ${vodId}: zero-byte recording — deleting empty recording`);-- _webhookForVod({ ...vod, health_status: 'zero_byte', is_public: 0 }, 'vod.failed');try { fs.unlinkSync(filePath); } catch { /* */ }try { tools.cleanupSeekableFile(filePath); } catch { /* */ }try { if (vod.master_file_path && fs.existsSync(vod.master_file_path)) fs.unlinkSync(vod.master_file_path); } catch { /* */ }-- try { db.run('DELETE FROM vods WHERE id = ?', [vodId]); } catch { /* */ }++ _commitVodOutcome(vod, 'vod.failed', () => db.run('DELETE FROM vods WHERE id = ?', [vodId]),++ () => ({ ...vod, health_status: 'zero_byte', is_public: 0 }));return null;}@@@ -244,22 +252,18 @@ async function _doFinalize(vodId, opts) {if (opts.ffmpegCorrupted) {console.warn(`[VOD] Finalized VOD ${vodId} marked corrupt by FFmpeg diagnostics; quarantining without deletion`);-- db.run(`UPDATE vods SET is_recording = 0, health_status = ?, health_issues_json = ?, probe_duration_seconds = ?, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,-- ['corrupt', JSON.stringify(['ffmpeg-corruption-detected']), durationSeconds > 0 ? durationSeconds : 0, probeFormatJson, vodId]);-- const failed = db.getVodById(vodId);-- _webhookForVod(failed, 'vod.failed');-- return failed;++ _commitVodOutcome(vod, 'vod.failed', () => db.run(`UPDATE vods SET is_recording = 0, health_status = ?, health_issues_json = ?, probe_duration_seconds = ?, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,++ ['corrupt', JSON.stringify(['ffmpeg-corruption-detected']), durationSeconds > 0 ? durationSeconds : 0, probeFormatJson, vodId]));++ return db.getVodById(vodId);}// Very short recordings are quarantined for review instead of deleted.const MIN_VOD_SECONDS = parseInt(process.env.MIN_VOD_SECONDS || '2', 10);if (durationSeconds < MIN_VOD_SECONDS) {console.log(`[VOD] Quarantining short vod ${vodId}: duration ${durationSeconds}s`);-- db.run(`UPDATE vods SET is_recording = 0, health_status = ?, health_issues_json = ?, probe_duration_seconds = ?, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,-- ['needs_review', JSON.stringify(['short_duration']), durationSeconds > 0 ? durationSeconds : 0, probeFormatJson, vodId]);-- const failed = db.getVodById(vodId);-- _webhookForVod(failed, 'vod.failed');-- return failed;++ _commitVodOutcome(vod, 'vod.failed', () => db.run(`UPDATE vods SET is_recording = 0, health_status = ?, health_issues_json = ?, probe_duration_seconds = ?, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,++ ['needs_review', JSON.stringify(['short_duration']), durationSeconds > 0 ? durationSeconds : 0, probeFormatJson, vodId]));++ return db.getVodById(vodId);}let stat;@@@ -270,11 +274,6 @@ async function _doFinalize(vodId, opts) {db.run('UPDATE vods SET is_recording = 0 WHERE id = ?', [vodId]);return null;}-- db.run('UPDATE vods SET is_recording = 0, duration_seconds = ?, file_size = ?, probe_duration_seconds = ?, probe_format_json = ?, health_status = ? WHERE id = ?',-- [durationSeconds, stat.size, durationSeconds, probeFormatJson, 'ok', vodId]);---- console.log(`[VOD] Finalized: vod ${vodId}, ${durationSeconds}s, ${(stat.size / 1024 / 1024).toFixed(1)}MB`);--// The lossless .master.mkv archive is only a fallback. Delete it ONLY once// the served webm is confirmed complete; if recovery failed, KEEP it.if (masterPath) {@@@ -296,18 +295,24 @@ async function _doFinalize(vodId, opts) {}}-- // Thumbnail, then webhook vod.ready (thumbnail awaited so the payload has it;-- // failure is non-fatal).++ // Thumbnail BEFORE the ready commit, so the vod.ready payload has it (failure is non-fatal).++ // The row is still is_recording = 1 here, so ask for the 10%-in frame, not the live edge.try {const thumbService = require('../thumbnails/thumbnail-service');-- await thumbService.generateVodThumbnail(vodId, filePath);++ await thumbService.generateVodThumbnail(vodId, filePath, { liveEdge: false });} catch (err) {console.warn(`[VOD] Thumbnail generation failed for vod ${vodId}:`, err.message);}-- const finalVod = db.getVodById(vodId);-- _webhookForVod(finalVod, 'vod.ready');-- return finalVod;++ // The ready transition and the vod.ready event commit together (then the webhook goes out).++ // A failed commit throws, as the plain UPDATE did before.++ announce(vod.app_id, 'vod.ready', {++ change: () => db.run('UPDATE vods SET is_recording = 0, duration_seconds = ?, file_size = ?, probe_duration_seconds = ?, probe_format_json = ?, health_status = ? WHERE id = ?',++ [durationSeconds, stat.size, durationSeconds, probeFormatJson, 'ok', vodId]),++ payload: () => vodPublic(db.getVodById(vodId)),++ });++ console.log(`[VOD] Finalized: vod ${vodId}, ${durationSeconds}s, ${(stat.size / 1024 / 1024).toFixed(1)}MB`);++ return db.getVodById(vodId);}module.exports = { finalizeVod, isFinalizing, vodPublic, rebuildWebmFromMaster, _absUrl };@@@ -751,12 +751,20 @@ async function emitStorageEvent(event, kind, data, settings = getSettings()) {alertLastSentAt.set(cooldownKey, Date.now());const payload = { kind, ...data, at: new Date().toISOString() };(event === 'storage.alert' ? console.error : console.warn)(`[VodStorage] ${event} (${kind}): ${JSON.stringify(data)}`);-- try { require('../events').emit(event, null, payload); } catch { /* never blocks the alert */ }++ // No database state changes here (the cooldown is in memory): the outbox row is the whole++ // commit, queued once for every app; each app's webhook carries its event_id.++ let eventId = null;++ try {++ const events = require('../events');++ const env = db.getDb().transaction(() => events.record(event, null, payload))();++ eventId = env ? env.event_id : null;++ events.kick();++ } catch (err) { console.warn(`[VodStorage] ${event} not queued for Events:`, err.message); }let apps = [];try { apps = db.listApps().filter(a => a.webhook_url); } catch { apps = []; }let webhooks;try { webhooks = require('../webhooks'); } catch { return false; }-- await Promise.all(apps.map(app => webhooks.sendWebhook(app.app_id, event, payload).catch(() => false)));++ await Promise.all(apps.map(app => webhooks.sendWebhook(app.app_id, event, payload, { eventId }).catch(() => false)));return true;}@@@ -5,7 +5,12 @@* `X-OVMedia-Signature: sha256=<hmac>` — HMAC-SHA256 of the raw body using the* app's webhook_secret (per CONTRACTS.md). Fire-and-forget with small retries.*-- * Events: vod.ready | vod.failed | clip.ready | clip.failed++ * Events: vod.ready | vod.failed | clip.ready | clip.failed | media.object.uploaded |++ * storage.alert | storage.recovered++ *++ * Outcomes go through announce(): the state change and its durable OpenVibe.Events outbox row++ * commit in one SQLite transaction, then the webhook is sent with `event_id` = that event's id++ * (absent when the outbox is off), so a consumer that also reads Events dedupes the pair.*/'use strict';@@@ -35,24 +40,24 @@ async function _post(url, rawBody, signature) {/*** Send an event to an app's webhook. Silently no-ops when the app has no-- * webhook_url. Never throws.++ * webhook_url. Never throws. Queues nothing: the durable event is written by++ * announce() (or the caller) inside the state change's transaction.* @param {string|object} appOrId app_id or apps row* @param {string} event e.g. 'vod.ready'* @param {object} data event payload++ * @param {object} [opts]++ * @param {string} [opts.eventId] the OpenVibe.Events event_id of the same outcome*/--async function sendWebhook(appOrId, event, data) {-- // The durable twin (OpenVibe.Events) is queued whether or not this app has a webhook URL.-- // Storage alerts go to every app, so server/vod/vod-storage.js queues those once itself.-- if (!String(event).startsWith('storage.')) {-- try { require('./events').emit(event, typeof appOrId === 'string' ? appOrId : appOrId && appOrId.app_id, data); } catch { /* never blocks a webhook */ }-- }++async function sendWebhook(appOrId, event, data, { eventId = null } = {}) {let app = appOrId;if (typeof appOrId === 'string') {try { app = db.getApp(appOrId); } catch { app = null; }}if (!app || !app.webhook_url) return false;-- const rawBody = JSON.stringify({ event, app_id: app.app_id, data });++ const body = { event, app_id: app.app_id, data };++ if (eventId) body.event_id = eventId;++ const rawBody = JSON.stringify(body);const signature = sign(app.webhook_secret, rawBody);for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {@@@ -70,4 +75,27 @@ async function sendWebhook(appOrId, event, data) {return false;}--module.exports = { sendWebhook, sign };++/**++ * Commit an outcome and announce it. `change()` (synchronous DB writes, optional) and the outbox++ * row for `event` run in ONE SQLite transaction, so the durable event exists if and only if the++ * change committed: never lost after a commit, never announced for a rollback. `payload()` is read++ * inside that transaction (the row as committed). After the commit the relay is woken and the++ * app's webhook is sent with the same event_id. Returns { data, eventId }; throws (nothing changed,++ * nothing queued, no webhook) when the transaction fails.++ */++function announce(appId, event, { change = null, payload }) {++ const events = require('./events');++ let data = null;++ let env = null;++ db.getDb().transaction(() => {++ if (change) change();++ data = typeof payload === 'function' ? payload() : payload;++ env = events.record(event, appId, data);++ })();++ events.kick();++ const eventId = env ? env.event_id : null;++ sendWebhook(appId, event, data, { eventId }).catch(() => {});++ return { data, eventId };++}++++module.exports = { sendWebhook, announce, sign };@@@ -1,7 +1,7 @@'use strict';--// Media → OpenVibe.Events (roadmap Wave 3): webhook outcomes are also queued in the outbox and++// Media → OpenVibe.Events (roadmap Wave 3): announced outcomes are queued in the outbox and// relayed with a service token; storage alerts are queued once, not once per app; disabled--// without EVENTS_URL. Stub Network token endpoint + stub Events.++// without EVENTS_URL. (Transactionality: test/outbox-transaction.test.js.) Stub Network token endpoint + stub Events.const assert = require('assert');const fs = require('fs');const os = require('os');@@@ -41,18 +41,21 @@ const stub = http.createServer((req, res) => {const db = require('../server/db/database');db.getDb();const events = require('../server/events');-- const { sendWebhook } = require('../server/webhooks');++ const { sendWebhook, announce } = require('../server/webhooks');++ const inTx = (fn) => db.getDb().transaction(fn)();assert.strictEqual(events.init({ eventsUrl: '', clientSecret: '' }), null, 'off without EVENTS_URL');-- assert.strictEqual(events.emit('vod.ready', 'live', { id: 1 }), null);++ assert.strictEqual(inTx(() => events.record('vod.ready', 'live', { id: 1 })), null);const outbox = events.init({ eventsUrl: base, clientSecret: 's', networkUrl: base, intervalMs: 50 });assert.ok(outbox);-- // A webhook to an app with no webhook URL still queues the durable event.-- await sendWebhook('no-such-app', 'vod.ready', { id: 42, title: 'Stream', meta: { big: 'x'.repeat(100000) }, ai_overview: 'long text' });-- await sendWebhook('no-such-app', 'clip.failed', { id: 7, status: 'failed' });-- await sendWebhook('no-such-app', 'unknown.event', { id: 1 });++ // An app with no webhook URL still gets the durable event.++ announce('no-such-app', 'vod.ready', { payload: { id: 42, title: 'Stream', meta: { big: 'x'.repeat(100000) }, ai_overview: 'long text' } });++ announce('no-such-app', 'clip.failed', { payload: { id: 7, status: 'failed' } });++ assert.strictEqual(announce('no-such-app', 'unknown.event', { payload: { id: 1 } }).eventId, null, 'no durable twin');++ // A bare webhook queues nothing (the event is written inside the state change's transaction).++ await sendWebhook('no-such-app', 'vod.ready', { id: 43 });await outbox.flush();assert.strictEqual(published.length, 2);assert.strictEqual(published[0].event_type, 'media.vod.ready');@@@ -61,6 +64,7 @@ const stub = http.createServer((req, res) => {assert.strictEqual(published[0].visibility, 'internal');assert.strictEqual(published[0].payload.app_id, 'no-such-app');assert.ok(!('meta' in published[0].payload) && !('ai_overview' in published[0].payload), 'free text stays behind the API');++ assert.strictEqual(published[0].priority, 'important', 'lifecycle outcomes are important (§6.3)');assert.strictEqual(published[1].event_type, 'media.clip.failed');assert.strictEqual(published[1].priority, 'important');@@@ -68,7 +72,7 @@ const stub = http.createServer((req, res) => {await sendWebhook('no-such-app', 'storage.alert', { kind: 'disk' });await outbox.flush();assert.strictEqual(published.length, 2);-- events.emit('storage.alert', null, { kind: 'disk', free_gb: 3 });++ inTx(() => events.record('storage.alert', null, { kind: 'disk', free_gb: 3 }));await outbox.flush();assert.strictEqual(published.length, 3);assert.deepStrictEqual(published[2].subject, { type: 'storage', id: 'disk' });nnew file mode 100644@@@ -0,0 +1,152 @@++'use strict';++// Roadmap Wave 3 (audit item 7): Media's outbox row is written in the SAME SQLite transaction as++// the state change it describes. An outcome is never lost (the change committed, the event did++// not) and never a phantom (the event exists, the change rolled back). The webhook for the same++// outcome follows the commit and carries the event's id, so a consumer reading both paths++// handles it once. Covers webhooks.announce() and the real call sites for clip.failed++// (clip-jobs) and vod.failed (finalize), with an outbox insert that fails and a change that fails.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const http = require('http');++const crypto = require('crypto');++++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-outbox-tx-'));++const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };++process.env.DB_PATH = path.join(tmp, 'media.db');++process.env.VOD_PATH = dir('vods');++process.env.CLIPS_PATH = dir('clips');++process.env.FILES_PATH = dir('files');++process.env.THUMBNAILS_PATH = dir('thumbnails');++process.env.PASTES_PATH = dir('pastes');++process.env.OBJECTS_PATH = dir('objects');++++const hooks = [];++const stub = http.createServer((req, res) => {++ let body = '';++ req.on('data', (c) => { body += c; });++ req.on('end', () => {++ res.setHeader('Content-Type', 'application/json');++ if (req.url === '/oauth/token') return res.end(JSON.stringify({ access_token: 'tok', token_type: 'Bearer', expires_in: 300 }));++ if (req.url === '/api/v1/events') {++ const parsed = JSON.parse(body);++ const list = parsed.events || [parsed];++ const results = list.map((e, i) => ({ event_id: e.event_id, seq: i + 1, duplicate: false }));++ return res.end(JSON.stringify(parsed.events ? { results } : results[0]));++ }++ if (req.url === '/hook') {++ hooks.push({ body, signature: req.headers['x-ovmedia-signature'] });++ return res.end('{"ok":true}');++ }++ res.statusCode = 404; res.end('{}');++ });++});++++const waitFor = async (fn, ms = 3000) => {++ const end = Date.now() + ms;++ while (Date.now() < end) { if (fn()) return true; await new Promise((r) => setTimeout(r, 20)); }++ return false;++};++++(async () => {++ await new Promise((r) => stub.listen(0, '127.0.0.1', r));++ const base = `http://127.0.0.1:${stub.address().port}`;++ const db = require('../server/db/database');++ const raw = db.getDb();++ db.upsertApp({ app_id: 'live', api_key: 'k', webhook_url: `${base}/hook`, webhook_secret: 'hook-secret' });++++ const events = require('../server/events');++ const { announce } = require('../server/webhooks');++ // A long relay interval: the test reads the outbox table, the relay is not under test here.++ assert.ok(events.init({ eventsUrl: base, clientSecret: 's', networkUrl: base, intervalMs: 60000 }));++ const outboxRows = () => raw.prepare('SELECT event_id, envelope FROM event_outbox ORDER BY id').all().map((r) => ({ id: r.event_id, env: JSON.parse(r.envelope) }));++ const failOutboxInserts = (on) => raw.exec(on++ ? "CREATE TEMP TRIGGER outbox_boom BEFORE INSERT ON event_outbox BEGIN SELECT RAISE(ABORT, 'outbox insert failed'); END"++ : 'DROP TRIGGER IF EXISTS temp.outbox_boom');++++ raw.prepare(`INSERT INTO vods (id, app_id, user_id, title, file_path, is_public, duration_seconds)++ VALUES (10, 'live', 5, 'Source', ?, 1, 60)`).run(path.join(process.env.VOD_PATH, 'missing.webm'));++ const clipId = Number(db.createClip({ app_id: 'live', vod_id: 10, user_id: 5, channel_user_id: 5, title: 'A clip', file_path: '', start_time: 1, end_time: 5, duration_seconds: 4, status: 'processing' }).lastInsertRowid);++ const { clipPublic } = require('../server/vod/clips-routes');++ const clipStatus = () => raw.prepare('SELECT status FROM clips WHERE id = ?').get(clipId).status;++++ // 1. The change and its event commit together; the webhook carries the same event_id.++ const out = announce('live', 'clip.ready', {++ change: () => db.run("UPDATE clips SET status = 'ready' WHERE id = ?", [clipId]),++ payload: () => clipPublic(db.getClipById(clipId)),++ });++ assert.strictEqual(clipStatus(), 'ready');++ let rows = outboxRows();++ assert.strictEqual(rows.length, 1);++ assert.strictEqual(rows[0].env.event_type, 'media.clip.ready');++ assert.deepStrictEqual(rows[0].env.subject, { type: 'clip', id: String(clipId) });++ assert.strictEqual(rows[0].env.payload.status, 'ready', 'payload is the row as committed');++ assert.strictEqual(out.eventId, rows[0].id);++ assert.ok(await waitFor(() => hooks.length === 1), 'webhook sent after the commit');++ const hook = JSON.parse(hooks[0].body);++ assert.strictEqual(hook.event, 'clip.ready');++ assert.strictEqual(hook.event_id, rows[0].id, 'the webhook names the durable event');++ const sig = 'sha256=' + crypto.createHmac('sha256', 'hook-secret').update(hooks[0].body).digest('hex');++ assert.strictEqual(hooks[0].signature, sig, 'the signature covers event_id');++++ // 2. No phantom: a change that fails leaves no event and sends no webhook.++ assert.throws(() => announce('live', 'clip.failed', {++ change: () => { db.run("UPDATE clips SET status = 'failed' WHERE id = ?", [clipId]); throw new Error('disk full'); },++ payload: () => clipPublic(db.getClipById(clipId)),++ }), /disk full/);++ assert.strictEqual(clipStatus(), 'ready', 'the change rolled back');++ assert.strictEqual(outboxRows().length, 1, 'no event for a rolled-back change');++++ // 3. Never lost: an outbox insert that fails rolls the change back with it.++ failOutboxInserts(true);++ assert.throws(() => announce('live', 'clip.failed', {++ change: () => db.run("UPDATE clips SET status = 'failed' WHERE id = ?", [clipId]),++ payload: () => clipPublic(db.getClipById(clipId)),++ }), /outbox insert failed/);++ failOutboxInserts(false);++ assert.strictEqual(clipStatus(), 'ready', 'no committed change without its event');++++ // 4. Outside a transaction the outbox refuses (nothing can enqueue "after the commit" again).++ assert.throws(() => events.record('clip.ready', 'live', { id: clipId }), /inside the transaction/);++++ // 5. Real call site: a re-cut whose source is gone commits status=failed with media.clip.failed.++ const clipJobs = require('../server/vod/clip-jobs');++ db.run("UPDATE clips SET status = 'failed', cut_attempts = 1 WHERE id = ?", [clipId]);++ failOutboxInserts(true);++ await assert.rejects(clipJobs.recutClip(clipId, { reason: 'test' }), /outbox insert failed/);++ failOutboxInserts(false);++ assert.strictEqual(raw.prepare('SELECT cut_error FROM clips WHERE id = ?').get(clipId).cut_error, null, 'the failure was not recorded without its event');++ assert.strictEqual(outboxRows().length, 1);++ const r = await clipJobs.recutClip(clipId, { reason: 'test' });++ assert.strictEqual(r.ok, false);++ rows = outboxRows();++ assert.strictEqual(rows.length, 2);++ assert.strictEqual(rows[1].env.event_type, 'media.clip.failed');++ assert.strictEqual(rows[1].env.payload.status, 'failed');++ assert.match(raw.prepare('SELECT cut_error FROM clips WHERE id = ?').get(clipId).cut_error, /unavailable/);++++ // 6. Real call site: finalizing a VOD with no recording file deletes the row and queues++ // media.vod.failed in one transaction; if the event cannot be queued the row stays.++ const { finalizeVod } = require('../server/vod/finalize');++ raw.prepare(`INSERT INTO vods (id, app_id, user_id, title, file_path, is_public, is_recording)++ VALUES (20, 'live', 5, 'Ghost', ?, 1, 1)`).run(path.join(process.env.VOD_PATH, 'never-written.webm'));++ failOutboxInserts(true);++ await finalizeVod(20);++ failOutboxInserts(false);++ assert.ok(raw.prepare('SELECT 1 FROM vods WHERE id = 20').get(), 'not deleted without its vod.failed event');++ assert.strictEqual(outboxRows().length, 2);++ await finalizeVod(20);++ assert.ok(!raw.prepare('SELECT 1 FROM vods WHERE id = 20').get(), 'deleted');++ rows = outboxRows();++ assert.strictEqual(rows.length, 3);++ assert.strictEqual(rows[2].env.event_type, 'media.vod.failed');++ assert.deepStrictEqual(rows[2].env.subject, { type: 'vod', id: '20' });++ assert.strictEqual(rows[2].env.priority, 'important');++ assert.ok(await waitFor(() => hooks.some((h) => JSON.parse(h.body).event_id === rows[2].id)), 'vod.failed webhook names the event');++++ events._reset();++ stub.close();++ console.log('✅ media outbox: state change and event commit together (no lost or phantom outcome), webhook carries event_id');++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });
@@ -245,10 +245,18 @@ directories are shared across apps) and responses carry a `note` saying so.
@@ -245,10 +245,18 @@ directories are shared across apps) and responses carry a `note` saying so.
### Webhooks (outbound)
### Webhooks (outbound)
`POST` to the app's `webhook_url` with body
`POST` to the app's `webhook_url` with body
-`{ "event": "vod.ready"|"vod.failed"|"clip.ready"|"clip.failed"|"media.object.uploaded", "app_id", "data" }`
+`{ "event": "vod.ready"|"vod.failed"|"clip.ready"|"clip.failed"|"media.object.uploaded"|"storage.alert"|"storage.recovered", "app_id", "data", "event_id"? }`
and header `X-OVMedia-Signature: sha256=<hex hmac-sha256 of the raw body with
and header `X-OVMedia-Signature: sha256=<hex hmac-sha256 of the raw body with
the app's webhook_secret>`. 3 attempts with backoff, 10 s timeout.
the app's webhook_secret>`. 3 attempts with backoff, 10 s timeout.
+Each outcome is also a durable OpenVibe.Events event (`media.vod.ready|failed`,
+`media.clip.ready|failed`, `media.object.uploaded`, `media.storage.alert|recovered`;
+`server/events.js`). The outbox row is written in the **same SQLite transaction** as the state
+change it describes (`webhooks.announce()`), so an outcome is never lost or announced for a
+rolled-back change; the webhook follows the commit and carries `event_id` = that event's id
+(absent when the outbox is off), so an app that reads both paths (Live during its webhook →
+Events transition) handles each outcome once. Webhooks stay until every consumer reads Events.
+
## Health, readiness and metrics
## Health, readiness and metrics
| path | behavior |
| path | behavior |
@@ -10,10 +10,14 @@
@@ -10,10 +10,14 @@
* media.object.uploaded → media.object.uploaded (subject object <id>)
* media.object.uploaded → media.object.uploaded (subject object <id>)
* storage.alert | storage.recovered → media.storage.alert | media.storage.recovered
* storage.alert | storage.recovered → media.storage.alert | media.storage.recovered
*
*
- * The event is enqueued right after the state change commits (its own SQLite transaction, a local
- * insert): the loss window is a process crash between those two statements. Delivery from the
- * outbox on is at least once. Every event is `internal` visibility: a VOD's own visibility decides
- * who may see it, and consumers (Live, Search) apply it.
+ * The outbox row is written INSIDE the SQLite transaction that makes the state change it describes
+ * (record(), called from webhooks.announce()): the event exists if and only if the change committed,
+ * so a crash can neither lose an outcome nor announce one that rolled back. The webhook for the same
+ * outcome is sent after the commit and carries the envelope's event_id, so a consumer receiving it by
+ * both paths (Live during its webhook → Events transition) handles it once. Delivery from the outbox
+ * on is at least once. Every event is `internal` visibility: a VOD's own visibility decides who may
+ * see it, and consumers (Live, Search) apply it. Lifecycle outcomes are `important` (roadmap §6.3);
+ * only storage.recovered is `low`.
*
*
* Off unless EVENTS_URL and OV_OAUTH_CLIENT_SECRET are set (EVENTS_PUBLISH=off disables it).
* Off unless EVENTS_URL and OV_OAUTH_CLIENT_SECRET are set (EVENTS_PUBLISH=off disables it).
*/
*/
@@ -69,8 +73,13 @@ function slim(appId, data) {
@@ -69,8 +73,13 @@ function slim(appId, data) {
return { app_id: appId || null, ...rest };
return { app_id: appId || null, ...rest };
}
}
-/** Queue the durable twin of a webhook event. Never throws; returns the envelope or null. */
-function emit(webhookEvent, appId, data) {
+/**
+ * Queue the durable event for an outcome. MUST run inside the transaction that makes the change
+ * (the SDK outbox refuses otherwise), and throws if the insert fails, so the change rolls back with
+ * it. Returns the envelope (with event_id), or null when the outbox is off, the tenant is a sandbox
+ * or the webhook event has no durable twin.
+ */
+function record(webhookEvent, appId, data) {
if (!outbox) return null;
if (!outbox) return null;
// Developer-project sandbox tenants (ADR-014) produce no platform events: sandbox activity must
// Developer-project sandbox tenants (ADR-014) produce no platform events: sandbox activity must
// never reach production consumers.
// never reach production consumers.
@@ -82,25 +91,21 @@ function emit(webhookEvent, appId, data) {
@@ -82,25 +91,21 @@ function emit(webhookEvent, appId, data) {
const subject = subjectType === 'storage'
const subject = subjectType === 'storage'
? { type: 'storage', id: String((data && data.kind) || 'storage') }
? { type: 'storage', id: String((data && data.kind) || 'storage') }
: { type: subjectType, id: String(id == null ? 'unknown' : id) };
: { type: subjectType, id: String(id == null ? 'unknown' : id) };
- try {
- let env = null;
- db.getDb().transaction(() => {
- env = outbox.enqueue({
- event_type: eventType,
- actor: { type: 'service', id: 'media' },
- subject,
- visibility: 'internal',
- priority: /failed|alert/.test(eventType) ? 'important' : 'low',
- payload: slim(appId, data),
- });
- })();
- stats.queued++;
- setImmediate(() => outbox && outbox.kick());
- return env;
- } catch (err) {
- console.warn(`[Events] ${eventType} not queued:`, err.message);
- return null;
- }
+ const env = outbox.enqueue({
+ event_type: eventType,
+ actor: { type: 'service', id: 'media' },
+ subject,
+ visibility: 'internal',
+ priority: eventType === 'media.storage.recovered' ? 'low' : 'important',
+ payload: slim(appId, data),
+ });
+ stats.queued++;
+ return env;
+}
+
+/** Wake the relay once the transaction that queued events has committed. */
+function kick() {
+ if (outbox) setImmediate(() => outbox && outbox.kick());
}
}
function status() {
function status() {
@@ -110,4 +115,4 @@ function status() {
@@ -110,4 +115,4 @@ function status() {
function _reset() { if (outbox) outbox.stop(); outbox = null; stats.queued = 0; stats.lastError = null; }
function _reset() { if (outbox) outbox.stop(); outbox = null; stats.queued = 0; stats.lastError = null; }
-module.exports = { init, emit, status, TYPES, _reset };
+module.exports = { init, record, kick, status, TYPES, _reset };
@@ -37,7 +37,7 @@ const model = require('./model');
@@ -37,7 +37,7 @@ const model = require('./model');
const invariant = require('./invariant');
const invariant = require('./invariant');
const signing = require('./signing');
const signing = require('./signing');
const { tenantAuth, tenantCors, tenantPath } = require('../auth');
const { tenantAuth, tenantCors, tenantPath } = require('../auth');
-const { sendWebhook } = require('../webhooks');
+const { announce } = require('../webhooks');
const MB = 1024 * 1024;
const MB = 1024 * 1024;
const upload = tenantAuth({ capability: 'media.object.upload' });
const upload = tenantAuth({ capability: 'media.object.upload' });
@@ -259,11 +259,15 @@ router.post('/:id/complete', contentAuth, tenantCors, (req, res) => {
@@ -259,11 +259,15 @@ router.post('/:id/complete', contentAuth, tenantCors, (req, res) => {
return problem(res, 422, 'media.invariant.public_object_too_large', `Public playback objects are limited to ${config.objects.publicMaxMb} MB`);
return problem(res, 422, 'media.invariant.public_object_too_large', `Public playback objects are limited to ${config.objects.publicMaxMb} MB`);
}
}
if (!quotaCheck(req, res, Number(obj.size_bytes) || 0, obj.id)) return;
if (!quotaCheck(req, res, Number(obj.size_bytes) || 0, obj.id)) return;
- model.updateObject(obj.id, { lifecycle_status: 'ready' });
- const done = model.getObject(obj.id);
- invariant.record(done);
- const body = model.objectPublic(done);
- sendWebhook(req.appId, 'media.object.uploaded', body).catch(() => {});
+ // The ready transition, its invariant row and the media.object.uploaded event commit
+ // together; the webhook follows the commit.
+ const { data: body } = announce(req.appId, 'media.object.uploaded', {
+ change: () => {
+ model.updateObject(obj.id, { lifecycle_status: 'ready' });
+ invariant.record(model.getObject(obj.id));
+ },
+ payload: () => model.objectPublic(model.getObject(obj.id)),
+ });
res.json(body);
res.json(body);
} catch (err) {
} catch (err) {
console.error('[Objects] complete error:', err.message);
console.error('[Objects] complete error:', err.message);
@@ -3,7 +3,7 @@
@@ -3,7 +3,7 @@
*
*
* recutClip(clipId, { reason }) the ONE code path that (re)cuts an existing clip row:
* recutClip(clipId, { reason }) the ONE code path that (re)cuts an existing clip row:
* resolves the source, cuts, updates the row, thumbnails,
* resolves the source, cuts, updates the row, thumbnails,
- * fires the webhook. Records cut_error / cut_attempts and
+ * commits the outcome with its event, fires the webhook. Records cut_error / cut_attempts and
* schedules an automatic retry on failure.
* schedules an automatic retry on failure.
* start() sweeper: every 3 min, failed clips that still have
* start() sweeper: every 3 min, failed clips that still have
* attempts left are re-cut. From the second attempt on, a
* attempts left are re-cut. From the second attempt on, a
@@ -19,7 +19,7 @@ const db = require('../db/database');
@@ -19,7 +19,7 @@ const db = require('../db/database');
const config = require('../config');
const config = require('../config');
const cutter = require('./clip-cutter');
const cutter = require('./clip-cutter');
const vodStorage = require('./vod-storage');
const vodStorage = require('./vod-storage');
-const { sendWebhook } = require('../webhooks');
+const { announce } = require('../webhooks');
const MAX_ATTEMPTS = 4;
const MAX_ATTEMPTS = 4;
const BACKOFF_MIN = [2, 10, 30, 90]; // minutes before attempt 2, 3, 4, …
const BACKOFF_MIN = [2, 10, 30, 90]; // minutes before attempt 2, 3, 4, …
@@ -75,12 +75,16 @@ async function _recutClip(clipId, { reason = 'recut' } = {}) {
@@ -75,12 +75,16 @@ async function _recutClip(clipId, { reason = 'recut' } = {}) {
const duration = Math.max(1, (Number(clip.end_time) || 0) - startTime);
const duration = Math.max(1, (Number(clip.end_time) || 0) - startTime);
const cut = await cutter.cutClipFile({ source: source.value, startTime, duration });
const cut = await cutter.cutClipFile({ source: source.value, startTime, duration });
if (!cut.ok) return fail(clipId, clip, attempt, cut.error);
if (!cut.ok) return fail(clipId, clip, attempt, cut.error);
- db.run("UPDATE clips SET file_path = ?, duration_seconds = ?, end_time = ?, status = 'ready', cut_error = NULL, cut_next_at = NULL WHERE id = ?",
- [cut.filePath, cut.duration, startTime + cut.duration, clipId]);
- require('../objects/model').safeSync('clip', clipId);
+ // Thumbnail first (the clip.ready payload carries it), then the ready transition and its event
+ // in one transaction (webhooks.announce), then the webhook.
try { await require('../thumbnails/thumbnail-service').generateClipThumbnail(clipId, cut.filePath); } catch { /* */ }
try { await require('../thumbnails/thumbnail-service').generateClipThumbnail(clipId, cut.filePath); } catch { /* */ }
+ announce(clip.app_id, 'clip.ready', {
+ change: () => db.run("UPDATE clips SET file_path = ?, duration_seconds = ?, end_time = ?, status = 'ready', cut_error = NULL, cut_next_at = NULL WHERE id = ?",
+ [cut.filePath, cut.duration, startTime + cut.duration, clipId]),
+ payload: () => _clipPublic(db.getClipById(clipId)),
+ });
+ require('../objects/model').safeSync('clip', clipId);
console.log(`[Clips] Clip ${clipId} ${reason} OK from vod ${clip.vod_id} (${startTime.toFixed(1)}-${(startTime + cut.duration).toFixed(1)}s, attempt ${attempt})`);
console.log(`[Clips] Clip ${clipId} ${reason} OK from vod ${clip.vod_id} (${startTime.toFixed(1)}-${(startTime + cut.duration).toFixed(1)}s, attempt ${attempt})`);
- sendWebhook(clip.app_id, 'clip.ready', _clipPublic(db.getClipById(clipId))).catch(() => {});
return { ok: true };
return { ok: true };
}
}
@@ -92,10 +96,12 @@ function fail(clipId, clip, attempt, error) {
@@ -92,10 +96,12 @@ function fail(clipId, clip, attempt, error) {
const more = attempt < MAX_ATTEMPTS && !hopeless;
const more = attempt < MAX_ATTEMPTS && !hopeless;
const mins = BACKOFF_MIN[Math.min(attempt - 1, BACKOFF_MIN.length - 1)];
const mins = BACKOFF_MIN[Math.min(attempt - 1, BACKOFF_MIN.length - 1)];
const nextAt = more ? new Date(Date.now() + mins * 60000).toISOString().replace('T', ' ').slice(0, 19) : null;
const nextAt = more ? new Date(Date.now() + mins * 60000).toISOString().replace('T', ' ').slice(0, 19) : null;
- db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_next_at = ? WHERE id = ?", [msg, nextAt, clipId]);
+ announce(clip.app_id, 'clip.failed', {
+ change: () => db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_next_at = ? WHERE id = ?", [msg, nextAt, clipId]),
+ payload: () => _clipPublic(db.getClipById(clipId)),
+ });
require('../objects/model').safeSync('clip', clipId);
require('../objects/model').safeSync('clip', clipId);
console.warn(`[Clips] Clip ${clipId} attempt ${attempt}/${MAX_ATTEMPTS} failed: ${msg}${more ? ` — retry in ${mins} min` : ' — giving up'}`);
console.warn(`[Clips] Clip ${clipId} attempt ${attempt}/${MAX_ATTEMPTS} failed: ${msg}${more ? ` — retry in ${mins} min` : ' — giving up'}`);
- sendWebhook(clip.app_id, 'clip.failed', _clipPublic(db.getClipById(clipId))).catch(() => {});
return { ok: false, error: msg, retry_at: nextAt };
return { ok: false, error: msg, retry_at: nextAt };
}
}
@@ -22,7 +22,7 @@ const db = require('../db/database');
@@ -22,7 +22,7 @@ const db = require('../db/database');
const tools = require('./media-tools');
const tools = require('./media-tools');
const cutter = require('./clip-cutter');
const cutter = require('./clip-cutter');
const { tenantAuth, tenantCors } = require('../auth');
const { tenantAuth, tenantCors } = require('../auth');
-const { sendWebhook } = require('../webhooks');
+const { announce } = require('../webhooks');
const objects = require('../objects/model');
const objects = require('../objects/model');
const router = express.Router({ mergeParams: true });
const router = express.Router({ mergeParams: true });
@@ -205,22 +205,28 @@ router.post('/', tenantAuth({ allowUser: true }), clipUpload.single('video'), as
@@ -205,22 +205,28 @@ router.post('/', tenantAuth({ allowUser: true }), clipUpload.single('video'), as
(async () => {
(async () => {
const cut = await cutter.cutClipFile({ source: source.value, startTime, duration });
const cut = await cutter.cutClipFile({ source: source.value, startTime, duration });
if (cut.ok) {
if (cut.ok) {
- db.run('UPDATE clips SET file_path = ?, duration_seconds = ?, end_time = ?, status = ? WHERE id = ?',
- [cut.filePath, cut.duration, startTime + cut.duration, 'ready', clipId]);
- objects.safeSync('clip', clipId);
+ // Thumbnail first so the clip.ready payload carries it; then the ready transition
+ // and its event commit together (webhooks.announce) and the webhook goes out.
try {
try {
await require('../thumbnails/thumbnail-service').generateClipThumbnail(clipId, cut.filePath);
await require('../thumbnails/thumbnail-service').generateClipThumbnail(clipId, cut.filePath);
} catch { /* */ }
} catch { /* */ }
+ announce(appId, 'clip.ready', {
+ change: () => db.run('UPDATE clips SET file_path = ?, duration_seconds = ?, end_time = ?, status = ? WHERE id = ?',
+ [cut.filePath, cut.duration, startTime + cut.duration, 'ready', clipId]),
+ payload: () => clipPublic(db.getClipById(clipId)),
+ });
+ objects.safeSync('clip', clipId);
console.log(`[Clips] Clip ${clipId} cut from vod ${vodId} (${startTime.toFixed(1)}-${(startTime + cut.duration).toFixed(1)}s)`);
console.log(`[Clips] Clip ${clipId} cut from vod ${vodId} (${startTime.toFixed(1)}-${(startTime + cut.duration).toFixed(1)}s)`);
- sendWebhook(appId, 'clip.ready', clipPublic(db.getClipById(clipId))).catch(() => {});
} else {
} else {
// First attempt failed: record why and let the retry sweeper take it from here
// First attempt failed: record why and let the retry sweeper take it from here
// (attempt 2 pulls a cloud-stored VOD back to local disk first).
// (attempt 2 pulls a cloud-stored VOD back to local disk first).
const nextAt = new Date(Date.now() + 2 * 60000).toISOString().replace('T', ' ').slice(0, 19);
const nextAt = new Date(Date.now() + 2 * 60000).toISOString().replace('T', ' ').slice(0, 19);
- db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_attempts = 1, cut_next_at = ? WHERE id = ?", [String(cut.error || 'cut failed').slice(0, 500), nextAt, clipId]);
+ announce(appId, 'clip.failed', {
+ change: () => db.run("UPDATE clips SET status = 'failed', cut_error = ?, cut_attempts = 1, cut_next_at = ? WHERE id = ?", [String(cut.error || 'cut failed').slice(0, 500), nextAt, clipId]),
+ payload: () => clipPublic(db.getClipById(clipId)),
+ });
objects.safeSync('clip', clipId);
objects.safeSync('clip', clipId);
console.warn(`[Clips] Clip ${clipId} failed: ${cut.error} — auto-retry in 2 min`);
console.warn(`[Clips] Clip ${clipId} failed: ${cut.error} — auto-retry in 2 min`);
- sendWebhook(appId, 'clip.failed', clipPublic(db.getClipById(clipId))).catch(() => {});
}
}
})().catch(err => console.error('[Clips] Background cut error:', err.message));
})().catch(err => console.error('[Clips] Background cut error:', err.message));
@@ -4,7 +4,8 @@
@@ -4,7 +4,8 @@
* Ported from the predecessor's finalizeVodRecording/_doFinalize, keyed by
* Ported from the predecessor's finalizeVodRecording/_doFinalize, keyed by
* vodId (there is no stream registry here — apps own stream state):
* vodId (there is no stream registry here — apps own stream state):
* merge pending chunk segments → seekable remux → probe → master recovery
* merge pending chunk segments → seekable remux → probe → master recovery
- * (truncated-webm rebuild) → DB update → thumbnail → webhook vod.ready.
+ * (truncated-webm rebuild) → thumbnail → DB update + vod.ready event (one
+ * transaction) → webhook vod.ready.
*
*
* Guarded against double-invocation; refuses to finalize while the recorder
* Guarded against double-invocation; refuses to finalize while the recorder
* still holds the file open (stops it gracefully instead — its ffmpeg exit
* still holds the file open (stops it gracefully instead — its ffmpeg exit
@@ -18,7 +19,7 @@ const { spawn } = require('child_process');
@@ -18,7 +19,7 @@ const { spawn } = require('child_process');
const db = require('../db/database');
const db = require('../db/database');
const config = require('../config');
const config = require('../config');
const tools = require('./media-tools');
const tools = require('./media-tools');
-const { sendWebhook } = require('../webhooks');
+const { announce } = require('../webhooks');
/** Absolute-ize a stored Media-relative URL (/t/…, /api/thumbnails/…). */
/** Absolute-ize a stored Media-relative URL (/t/…, /api/thumbnails/…). */
function _absUrl(u) {
function _absUrl(u) {
@@ -102,10 +103,18 @@ function rebuildWebmFromMaster(masterPath, webmPath) {
@@ -102,10 +103,18 @@ function rebuildWebmFromMaster(masterPath, webmPath) {
});
});
}
}
-function _webhookForVod(vod, event) {
+/**
+ * The state change for a VOD outcome and its event/webhook, committed together (webhooks.announce).
+ * `row()` gives the row the payload describes (read inside the transaction). Returns the payload,
+ * or null when the transaction failed (then nothing changed and nothing was announced).
+ */
+function _commitVodOutcome(vod, event, change, row = () => db.getVodById(vod.id)) {
try {
try {
- sendWebhook(vod.app_id, event, vodPublic(vod)).catch(() => {});
- } catch { /* non-critical */ }
+ return announce(vod.app_id, event, { change, payload: () => vodPublic(row()) }).data;
+ } catch (err) {
+ console.error(`[VOD] vod ${vod.id}: ${event} not committed:`, err.message);
+ return null;
+ }
}
}
/**
/**
@@ -164,8 +173,7 @@ async function _doFinalize(vodId, opts) {
@@ -164,8 +173,7 @@ async function _doFinalize(vodId, opts) {
// process died before ffmpeg opened the file). Nothing to review — delete
// process died before ffmpeg opened the file). Nothing to review — delete
// the row so a 0:00 ghost never reaches listings.
// the row so a 0:00 ghost never reaches listings.
console.warn(`[VOD] vod ${vodId}: no recording file — deleting empty row`);
console.warn(`[VOD] vod ${vodId}: no recording file — deleting empty row`);
- _webhookForVod(vod, 'vod.failed');
- try { db.run('DELETE FROM vods WHERE id = ?', [vodId]); } catch { /* */ }
+ _commitVodOutcome(vod, 'vod.failed', () => db.run('DELETE FROM vods WHERE id = ?', [vodId]), () => vod);
return null;
return null;
}
}
@@ -182,11 +190,11 @@ async function _doFinalize(vodId, opts) {
@@ -182,11 +190,11 @@ async function _doFinalize(vodId, opts) {
// ready, and a 0:00 ghost must never reach listings.
// ready, and a 0:00 ghost must never reach listings.
if (tools.getFileSizeSafe(filePath) === 0) {
if (tools.getFileSizeSafe(filePath) === 0) {
console.warn(`[VOD] vod ${vodId}: zero-byte recording — deleting empty recording`);
console.warn(`[VOD] vod ${vodId}: zero-byte recording — deleting empty recording`);
- _webhookForVod({ ...vod, health_status: 'zero_byte', is_public: 0 }, 'vod.failed');
try { fs.unlinkSync(filePath); } catch { /* */ }
try { fs.unlinkSync(filePath); } catch { /* */ }
try { tools.cleanupSeekableFile(filePath); } catch { /* */ }
try { tools.cleanupSeekableFile(filePath); } catch { /* */ }
try { if (vod.master_file_path && fs.existsSync(vod.master_file_path)) fs.unlinkSync(vod.master_file_path); } catch { /* */ }
try { if (vod.master_file_path && fs.existsSync(vod.master_file_path)) fs.unlinkSync(vod.master_file_path); } catch { /* */ }
- try { db.run('DELETE FROM vods WHERE id = ?', [vodId]); } catch { /* */ }
+ _commitVodOutcome(vod, 'vod.failed', () => db.run('DELETE FROM vods WHERE id = ?', [vodId]),
+ () => ({ ...vod, health_status: 'zero_byte', is_public: 0 }));
return null;
return null;
}
}
@@ -244,22 +252,18 @@ async function _doFinalize(vodId, opts) {
@@ -244,22 +252,18 @@ async function _doFinalize(vodId, opts) {
if (opts.ffmpegCorrupted) {
if (opts.ffmpegCorrupted) {
console.warn(`[VOD] Finalized VOD ${vodId} marked corrupt by FFmpeg diagnostics; quarantining without deletion`);
console.warn(`[VOD] Finalized VOD ${vodId} marked corrupt by FFmpeg diagnostics; quarantining without deletion`);
- db.run(`UPDATE vods SET is_recording = 0, health_status = ?, health_issues_json = ?, probe_duration_seconds = ?, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,
- ['corrupt', JSON.stringify(['ffmpeg-corruption-detected']), durationSeconds > 0 ? durationSeconds : 0, probeFormatJson, vodId]);
- const failed = db.getVodById(vodId);
- _webhookForVod(failed, 'vod.failed');
- return failed;
+ _commitVodOutcome(vod, 'vod.failed', () => db.run(`UPDATE vods SET is_recording = 0, health_status = ?, health_issues_json = ?, probe_duration_seconds = ?, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,
+ ['corrupt', JSON.stringify(['ffmpeg-corruption-detected']), durationSeconds > 0 ? durationSeconds : 0, probeFormatJson, vodId]));
+ return db.getVodById(vodId);
}
}
// Very short recordings are quarantined for review instead of deleted.
// Very short recordings are quarantined for review instead of deleted.
const MIN_VOD_SECONDS = parseInt(process.env.MIN_VOD_SECONDS || '2', 10);
const MIN_VOD_SECONDS = parseInt(process.env.MIN_VOD_SECONDS || '2', 10);
if (durationSeconds < MIN_VOD_SECONDS) {
if (durationSeconds < MIN_VOD_SECONDS) {
console.log(`[VOD] Quarantining short vod ${vodId}: duration ${durationSeconds}s`);
console.log(`[VOD] Quarantining short vod ${vodId}: duration ${durationSeconds}s`);
- db.run(`UPDATE vods SET is_recording = 0, health_status = ?, health_issues_json = ?, probe_duration_seconds = ?, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,
- ['needs_review', JSON.stringify(['short_duration']), durationSeconds > 0 ? durationSeconds : 0, probeFormatJson, vodId]);
- const failed = db.getVodById(vodId);
- _webhookForVod(failed, 'vod.failed');
- return failed;
+ _commitVodOutcome(vod, 'vod.failed', () => db.run(`UPDATE vods SET is_recording = 0, health_status = ?, health_issues_json = ?, probe_duration_seconds = ?, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,
+ ['needs_review', JSON.stringify(['short_duration']), durationSeconds > 0 ? durationSeconds : 0, probeFormatJson, vodId]));
+ return db.getVodById(vodId);
}
}
let stat;
let stat;
@@ -270,11 +274,6 @@ async function _doFinalize(vodId, opts) {
@@ -270,11 +274,6 @@ async function _doFinalize(vodId, opts) {
db.run('UPDATE vods SET is_recording = 0 WHERE id = ?', [vodId]);
db.run('UPDATE vods SET is_recording = 0 WHERE id = ?', [vodId]);
return null;
return null;
}
}
- db.run('UPDATE vods SET is_recording = 0, duration_seconds = ?, file_size = ?, probe_duration_seconds = ?, probe_format_json = ?, health_status = ? WHERE id = ?',
- [durationSeconds, stat.size, durationSeconds, probeFormatJson, 'ok', vodId]);
-
- console.log(`[VOD] Finalized: vod ${vodId}, ${durationSeconds}s, ${(stat.size / 1024 / 1024).toFixed(1)}MB`);
-
// The lossless .master.mkv archive is only a fallback. Delete it ONLY once
// The lossless .master.mkv archive is only a fallback. Delete it ONLY once
// the served webm is confirmed complete; if recovery failed, KEEP it.
// the served webm is confirmed complete; if recovery failed, KEEP it.
if (masterPath) {
if (masterPath) {
@@ -296,18 +295,24 @@ async function _doFinalize(vodId, opts) {
@@ -296,18 +295,24 @@ async function _doFinalize(vodId, opts) {
}
}
}
}
- // Thumbnail, then webhook vod.ready (thumbnail awaited so the payload has it;
- // failure is non-fatal).
+ // Thumbnail BEFORE the ready commit, so the vod.ready payload has it (failure is non-fatal).
+ // The row is still is_recording = 1 here, so ask for the 10%-in frame, not the live edge.
try {
try {
const thumbService = require('../thumbnails/thumbnail-service');
const thumbService = require('../thumbnails/thumbnail-service');
- await thumbService.generateVodThumbnail(vodId, filePath);
+ await thumbService.generateVodThumbnail(vodId, filePath, { liveEdge: false });
} catch (err) {
} catch (err) {
console.warn(`[VOD] Thumbnail generation failed for vod ${vodId}:`, err.message);
console.warn(`[VOD] Thumbnail generation failed for vod ${vodId}:`, err.message);
}
}
- const finalVod = db.getVodById(vodId);
- _webhookForVod(finalVod, 'vod.ready');
- return finalVod;
+ // The ready transition and the vod.ready event commit together (then the webhook goes out).
+ // A failed commit throws, as the plain UPDATE did before.
+ announce(vod.app_id, 'vod.ready', {
+ change: () => db.run('UPDATE vods SET is_recording = 0, duration_seconds = ?, file_size = ?, probe_duration_seconds = ?, probe_format_json = ?, health_status = ? WHERE id = ?',
+ [durationSeconds, stat.size, durationSeconds, probeFormatJson, 'ok', vodId]),
+ payload: () => vodPublic(db.getVodById(vodId)),
+ });
+ console.log(`[VOD] Finalized: vod ${vodId}, ${durationSeconds}s, ${(stat.size / 1024 / 1024).toFixed(1)}MB`);
+ return db.getVodById(vodId);
}
}
module.exports = { finalizeVod, isFinalizing, vodPublic, rebuildWebmFromMaster, _absUrl };
module.exports = { finalizeVod, isFinalizing, vodPublic, rebuildWebmFromMaster, _absUrl };
@@ -751,12 +751,20 @@ async function emitStorageEvent(event, kind, data, settings = getSettings()) {
@@ -751,12 +751,20 @@ async function emitStorageEvent(event, kind, data, settings = getSettings()) {
alertLastSentAt.set(cooldownKey, Date.now());
alertLastSentAt.set(cooldownKey, Date.now());
const payload = { kind, ...data, at: new Date().toISOString() };
const payload = { kind, ...data, at: new Date().toISOString() };
(event === 'storage.alert' ? console.error : console.warn)(`[VodStorage] ${event} (${kind}): ${JSON.stringify(data)}`);
(event === 'storage.alert' ? console.error : console.warn)(`[VodStorage] ${event} (${kind}): ${JSON.stringify(data)}`);
- try { require('../events').emit(event, null, payload); } catch { /* never blocks the alert */ }
+ // No database state changes here (the cooldown is in memory): the outbox row is the whole
+ // commit, queued once for every app; each app's webhook carries its event_id.
+ let eventId = null;
+ try {
+ const events = require('../events');
+ const env = db.getDb().transaction(() => events.record(event, null, payload))();
+ eventId = env ? env.event_id : null;
+ events.kick();
+ } catch (err) { console.warn(`[VodStorage] ${event} not queued for Events:`, err.message); }
let apps = [];
let apps = [];
try { apps = db.listApps().filter(a => a.webhook_url); } catch { apps = []; }
try { apps = db.listApps().filter(a => a.webhook_url); } catch { apps = []; }
let webhooks;
let webhooks;
try { webhooks = require('../webhooks'); } catch { return false; }
try { webhooks = require('../webhooks'); } catch { return false; }
- await Promise.all(apps.map(app => webhooks.sendWebhook(app.app_id, event, payload).catch(() => false)));
+ await Promise.all(apps.map(app => webhooks.sendWebhook(app.app_id, event, payload, { eventId }).catch(() => false)));
return true;
return true;
}
}
@@ -5,7 +5,12 @@
@@ -5,7 +5,12 @@
* `X-OVMedia-Signature: sha256=<hmac>` — HMAC-SHA256 of the raw body using the
* `X-OVMedia-Signature: sha256=<hmac>` — HMAC-SHA256 of the raw body using the
* app's webhook_secret (per CONTRACTS.md). Fire-and-forget with small retries.
* app's webhook_secret (per CONTRACTS.md). Fire-and-forget with small retries.
*
*
- * Events: vod.ready | vod.failed | clip.ready | clip.failed
+ * Events: vod.ready | vod.failed | clip.ready | clip.failed | media.object.uploaded |
+ * storage.alert | storage.recovered
+ *
+ * Outcomes go through announce(): the state change and its durable OpenVibe.Events outbox row
+ * commit in one SQLite transaction, then the webhook is sent with `event_id` = that event's id
+ * (absent when the outbox is off), so a consumer that also reads Events dedupes the pair.
*/
*/
'use strict';
'use strict';
@@ -35,24 +40,24 @@ async function _post(url, rawBody, signature) {
@@ -35,24 +40,24 @@ async function _post(url, rawBody, signature) {
/**
/**
* Send an event to an app's webhook. Silently no-ops when the app has no
* Send an event to an app's webhook. Silently no-ops when the app has no
- * webhook_url. Never throws.
+ * webhook_url. Never throws. Queues nothing: the durable event is written by
+ * announce() (or the caller) inside the state change's transaction.
* @param {string|object} appOrId app_id or apps row
* @param {string|object} appOrId app_id or apps row
* @param {string} event e.g. 'vod.ready'
* @param {string} event e.g. 'vod.ready'
* @param {object} data event payload
* @param {object} data event payload
+ * @param {object} [opts]
+ * @param {string} [opts.eventId] the OpenVibe.Events event_id of the same outcome
*/
*/
-async function sendWebhook(appOrId, event, data) {
- // The durable twin (OpenVibe.Events) is queued whether or not this app has a webhook URL.
- // Storage alerts go to every app, so server/vod/vod-storage.js queues those once itself.
- if (!String(event).startsWith('storage.')) {
- try { require('./events').emit(event, typeof appOrId === 'string' ? appOrId : appOrId && appOrId.app_id, data); } catch { /* never blocks a webhook */ }
- }
+async function sendWebhook(appOrId, event, data, { eventId = null } = {}) {
let app = appOrId;
let app = appOrId;
if (typeof appOrId === 'string') {
if (typeof appOrId === 'string') {
try { app = db.getApp(appOrId); } catch { app = null; }
try { app = db.getApp(appOrId); } catch { app = null; }
}
}
if (!app || !app.webhook_url) return false;
if (!app || !app.webhook_url) return false;
- const rawBody = JSON.stringify({ event, app_id: app.app_id, data });
+ const body = { event, app_id: app.app_id, data };
+ if (eventId) body.event_id = eventId;
+ const rawBody = JSON.stringify(body);
const signature = sign(app.webhook_secret, rawBody);
const signature = sign(app.webhook_secret, rawBody);
for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
@@ -70,4 +75,27 @@ async function sendWebhook(appOrId, event, data) {
@@ -70,4 +75,27 @@ async function sendWebhook(appOrId, event, data) {
return false;
return false;
}
}
-module.exports = { sendWebhook, sign };
+/**
+ * Commit an outcome and announce it. `change()` (synchronous DB writes, optional) and the outbox
+ * row for `event` run in ONE SQLite transaction, so the durable event exists if and only if the
+ * change committed: never lost after a commit, never announced for a rollback. `payload()` is read
+ * inside that transaction (the row as committed). After the commit the relay is woken and the
+ * app's webhook is sent with the same event_id. Returns { data, eventId }; throws (nothing changed,
+ * nothing queued, no webhook) when the transaction fails.
+ */
+function announce(appId, event, { change = null, payload }) {
+ const events = require('./events');
+ let data = null;
+ let env = null;
+ db.getDb().transaction(() => {
+ if (change) change();
+ data = typeof payload === 'function' ? payload() : payload;
+ env = events.record(event, appId, data);
+ })();
+ events.kick();
+ const eventId = env ? env.event_id : null;
+ sendWebhook(appId, event, data, { eventId }).catch(() => {});
+ return { data, eventId };
+}
+
+module.exports = { sendWebhook, announce, sign };
@@ -1,7 +1,7 @@
@@ -1,7 +1,7 @@
'use strict';
'use strict';
-// Media → OpenVibe.Events (roadmap Wave 3): webhook outcomes are also queued in the outbox and
+// Media → OpenVibe.Events (roadmap Wave 3): announced outcomes are queued in the outbox and
// relayed with a service token; storage alerts are queued once, not once per app; disabled
// relayed with a service token; storage alerts are queued once, not once per app; disabled
-// without EVENTS_URL. Stub Network token endpoint + stub Events.
+// without EVENTS_URL. (Transactionality: test/outbox-transaction.test.js.) Stub Network token endpoint + stub Events.
const assert = require('assert');
const assert = require('assert');
const fs = require('fs');
const fs = require('fs');
const os = require('os');
const os = require('os');
@@ -41,18 +41,21 @@ const stub = http.createServer((req, res) => {
@@ -41,18 +41,21 @@ const stub = http.createServer((req, res) => {
const db = require('../server/db/database');
const db = require('../server/db/database');
db.getDb();
db.getDb();
const events = require('../server/events');
const events = require('../server/events');
- const { sendWebhook } = require('../server/webhooks');
+ const { sendWebhook, announce } = require('../server/webhooks');
+ const inTx = (fn) => db.getDb().transaction(fn)();
assert.strictEqual(events.init({ eventsUrl: '', clientSecret: '' }), null, 'off without EVENTS_URL');
assert.strictEqual(events.init({ eventsUrl: '', clientSecret: '' }), null, 'off without EVENTS_URL');
- assert.strictEqual(events.emit('vod.ready', 'live', { id: 1 }), null);
+ assert.strictEqual(inTx(() => events.record('vod.ready', 'live', { id: 1 })), null);
const outbox = events.init({ eventsUrl: base, clientSecret: 's', networkUrl: base, intervalMs: 50 });
const outbox = events.init({ eventsUrl: base, clientSecret: 's', networkUrl: base, intervalMs: 50 });
assert.ok(outbox);
assert.ok(outbox);
- // A webhook to an app with no webhook URL still queues the durable event.
- await sendWebhook('no-such-app', 'vod.ready', { id: 42, title: 'Stream', meta: { big: 'x'.repeat(100000) }, ai_overview: 'long text' });
- await sendWebhook('no-such-app', 'clip.failed', { id: 7, status: 'failed' });
- await sendWebhook('no-such-app', 'unknown.event', { id: 1 });
+ // An app with no webhook URL still gets the durable event.
+ announce('no-such-app', 'vod.ready', { payload: { id: 42, title: 'Stream', meta: { big: 'x'.repeat(100000) }, ai_overview: 'long text' } });
+ announce('no-such-app', 'clip.failed', { payload: { id: 7, status: 'failed' } });
+ assert.strictEqual(announce('no-such-app', 'unknown.event', { payload: { id: 1 } }).eventId, null, 'no durable twin');
+ // A bare webhook queues nothing (the event is written inside the state change's transaction).
+ await sendWebhook('no-such-app', 'vod.ready', { id: 43 });
await outbox.flush();
await outbox.flush();
assert.strictEqual(published.length, 2);
assert.strictEqual(published.length, 2);
assert.strictEqual(published[0].event_type, 'media.vod.ready');
assert.strictEqual(published[0].event_type, 'media.vod.ready');
@@ -61,6 +64,7 @@ const stub = http.createServer((req, res) => {
@@ -61,6 +64,7 @@ const stub = http.createServer((req, res) => {
assert.strictEqual(published[0].visibility, 'internal');
assert.strictEqual(published[0].visibility, 'internal');
assert.strictEqual(published[0].payload.app_id, 'no-such-app');
assert.strictEqual(published[0].payload.app_id, 'no-such-app');
assert.ok(!('meta' in published[0].payload) && !('ai_overview' in published[0].payload), 'free text stays behind the API');
assert.ok(!('meta' in published[0].payload) && !('ai_overview' in published[0].payload), 'free text stays behind the API');
+ assert.strictEqual(published[0].priority, 'important', 'lifecycle outcomes are important (§6.3)');
assert.strictEqual(published[1].event_type, 'media.clip.failed');
assert.strictEqual(published[1].event_type, 'media.clip.failed');
assert.strictEqual(published[1].priority, 'important');
assert.strictEqual(published[1].priority, 'important');
@@ -68,7 +72,7 @@ const stub = http.createServer((req, res) => {
@@ -68,7 +72,7 @@ const stub = http.createServer((req, res) => {
await sendWebhook('no-such-app', 'storage.alert', { kind: 'disk' });
await sendWebhook('no-such-app', 'storage.alert', { kind: 'disk' });
await outbox.flush();
await outbox.flush();
assert.strictEqual(published.length, 2);
assert.strictEqual(published.length, 2);
- events.emit('storage.alert', null, { kind: 'disk', free_gb: 3 });
+ inTx(() => events.record('storage.alert', null, { kind: 'disk', free_gb: 3 }));
await outbox.flush();
await outbox.flush();
assert.strictEqual(published.length, 3);
assert.strictEqual(published.length, 3);
assert.deepStrictEqual(published[2].subject, { type: 'storage', id: 'disk' });
assert.deepStrictEqual(published[2].subject, { type: 'storage', id: 'disk' });
new file mode 100644
new file mode 100644
@@ -0,0 +1,152 @@
@@ -0,0 +1,152 @@
+'use strict';
+// Roadmap Wave 3 (audit item 7): Media's outbox row is written in the SAME SQLite transaction as
+// the state change it describes. An outcome is never lost (the change committed, the event did
+// not) and never a phantom (the event exists, the change rolled back). The webhook for the same
+// outcome follows the commit and carries the event's id, so a consumer reading both paths
+// handles it once. Covers webhooks.announce() and the real call sites for clip.failed
+// (clip-jobs) and vod.failed (finalize), with an outbox insert that fails and a change that fails.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const http = require('http');
+const crypto = require('crypto');
+
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-outbox-tx-'));
+const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
+process.env.DB_PATH = path.join(tmp, 'media.db');
+process.env.VOD_PATH = dir('vods');
+process.env.CLIPS_PATH = dir('clips');
+process.env.FILES_PATH = dir('files');
+process.env.THUMBNAILS_PATH = dir('thumbnails');
+process.env.PASTES_PATH = dir('pastes');
+process.env.OBJECTS_PATH = dir('objects');
+
+const hooks = [];
+const stub = http.createServer((req, res) => {
+ let body = '';
+ req.on('data', (c) => { body += c; });
+ req.on('end', () => {
+ res.setHeader('Content-Type', 'application/json');
+ if (req.url === '/oauth/token') return res.end(JSON.stringify({ access_token: 'tok', token_type: 'Bearer', expires_in: 300 }));
+ if (req.url === '/api/v1/events') {
+ const parsed = JSON.parse(body);
+ const list = parsed.events || [parsed];
+ const results = list.map((e, i) => ({ event_id: e.event_id, seq: i + 1, duplicate: false }));
+ return res.end(JSON.stringify(parsed.events ? { results } : results[0]));
+ }
+ if (req.url === '/hook') {
+ hooks.push({ body, signature: req.headers['x-ovmedia-signature'] });
+ return res.end('{"ok":true}');
+ }
+ res.statusCode = 404; res.end('{}');
+ });
+});
+
+const waitFor = async (fn, ms = 3000) => {
+ const end = Date.now() + ms;
+ while (Date.now() < end) { if (fn()) return true; await new Promise((r) => setTimeout(r, 20)); }
+ return false;
+};
+
+(async () => {
+ await new Promise((r) => stub.listen(0, '127.0.0.1', r));
+ const base = `http://127.0.0.1:${stub.address().port}`;
+ const db = require('../server/db/database');
+ const raw = db.getDb();
+ db.upsertApp({ app_id: 'live', api_key: 'k', webhook_url: `${base}/hook`, webhook_secret: 'hook-secret' });
+
+ const events = require('../server/events');
+ const { announce } = require('../server/webhooks');
+ // A long relay interval: the test reads the outbox table, the relay is not under test here.
+ assert.ok(events.init({ eventsUrl: base, clientSecret: 's', networkUrl: base, intervalMs: 60000 }));
+ const outboxRows = () => raw.prepare('SELECT event_id, envelope FROM event_outbox ORDER BY id').all().map((r) => ({ id: r.event_id, env: JSON.parse(r.envelope) }));
+ const failOutboxInserts = (on) => raw.exec(on
+ ? "CREATE TEMP TRIGGER outbox_boom BEFORE INSERT ON event_outbox BEGIN SELECT RAISE(ABORT, 'outbox insert failed'); END"
+ : 'DROP TRIGGER IF EXISTS temp.outbox_boom');
+
+ raw.prepare(`INSERT INTO vods (id, app_id, user_id, title, file_path, is_public, duration_seconds)
+ VALUES (10, 'live', 5, 'Source', ?, 1, 60)`).run(path.join(process.env.VOD_PATH, 'missing.webm'));
+ const clipId = Number(db.createClip({ app_id: 'live', vod_id: 10, user_id: 5, channel_user_id: 5, title: 'A clip', file_path: '', start_time: 1, end_time: 5, duration_seconds: 4, status: 'processing' }).lastInsertRowid);
+ const { clipPublic } = require('../server/vod/clips-routes');
+ const clipStatus = () => raw.prepare('SELECT status FROM clips WHERE id = ?').get(clipId).status;
+
+ // 1. The change and its event commit together; the webhook carries the same event_id.
+ const out = announce('live', 'clip.ready', {
+ change: () => db.run("UPDATE clips SET status = 'ready' WHERE id = ?", [clipId]),
+ payload: () => clipPublic(db.getClipById(clipId)),
+ });
+ assert.strictEqual(clipStatus(), 'ready');
+ let rows = outboxRows();
+ assert.strictEqual(rows.length, 1);
+ assert.strictEqual(rows[0].env.event_type, 'media.clip.ready');
+ assert.deepStrictEqual(rows[0].env.subject, { type: 'clip', id: String(clipId) });
+ assert.strictEqual(rows[0].env.payload.status, 'ready', 'payload is the row as committed');
+ assert.strictEqual(out.eventId, rows[0].id);
+ assert.ok(await waitFor(() => hooks.length === 1), 'webhook sent after the commit');
+ const hook = JSON.parse(hooks[0].body);
+ assert.strictEqual(hook.event, 'clip.ready');
+ assert.strictEqual(hook.event_id, rows[0].id, 'the webhook names the durable event');
+ const sig = 'sha256=' + crypto.createHmac('sha256', 'hook-secret').update(hooks[0].body).digest('hex');
+ assert.strictEqual(hooks[0].signature, sig, 'the signature covers event_id');
+
+ // 2. No phantom: a change that fails leaves no event and sends no webhook.
+ assert.throws(() => announce('live', 'clip.failed', {
+ change: () => { db.run("UPDATE clips SET status = 'failed' WHERE id = ?", [clipId]); throw new Error('disk full'); },
+ payload: () => clipPublic(db.getClipById(clipId)),
+ }), /disk full/);
+ assert.strictEqual(clipStatus(), 'ready', 'the change rolled back');
+ assert.strictEqual(outboxRows().length, 1, 'no event for a rolled-back change');
+
+ // 3. Never lost: an outbox insert that fails rolls the change back with it.
+ failOutboxInserts(true);
+ assert.throws(() => announce('live', 'clip.failed', {
+ change: () => db.run("UPDATE clips SET status = 'failed' WHERE id = ?", [clipId]),
+ payload: () => clipPublic(db.getClipById(clipId)),
+ }), /outbox insert failed/);
+ failOutboxInserts(false);
+ assert.strictEqual(clipStatus(), 'ready', 'no committed change without its event');
+
+ // 4. Outside a transaction the outbox refuses (nothing can enqueue "after the commit" again).
+ assert.throws(() => events.record('clip.ready', 'live', { id: clipId }), /inside the transaction/);
+
+ // 5. Real call site: a re-cut whose source is gone commits status=failed with media.clip.failed.
+ const clipJobs = require('../server/vod/clip-jobs');
+ db.run("UPDATE clips SET status = 'failed', cut_attempts = 1 WHERE id = ?", [clipId]);
+ failOutboxInserts(true);
+ await assert.rejects(clipJobs.recutClip(clipId, { reason: 'test' }), /outbox insert failed/);
+ failOutboxInserts(false);
+ assert.strictEqual(raw.prepare('SELECT cut_error FROM clips WHERE id = ?').get(clipId).cut_error, null, 'the failure was not recorded without its event');
+ assert.strictEqual(outboxRows().length, 1);
+ const r = await clipJobs.recutClip(clipId, { reason: 'test' });
+ assert.strictEqual(r.ok, false);
+ rows = outboxRows();
+ assert.strictEqual(rows.length, 2);
+ assert.strictEqual(rows[1].env.event_type, 'media.clip.failed');
+ assert.strictEqual(rows[1].env.payload.status, 'failed');
+ assert.match(raw.prepare('SELECT cut_error FROM clips WHERE id = ?').get(clipId).cut_error, /unavailable/);
+
+ // 6. Real call site: finalizing a VOD with no recording file deletes the row and queues
+ // media.vod.failed in one transaction; if the event cannot be queued the row stays.
+ const { finalizeVod } = require('../server/vod/finalize');
+ raw.prepare(`INSERT INTO vods (id, app_id, user_id, title, file_path, is_public, is_recording)
+ VALUES (20, 'live', 5, 'Ghost', ?, 1, 1)`).run(path.join(process.env.VOD_PATH, 'never-written.webm'));
+ failOutboxInserts(true);
+ await finalizeVod(20);
+ failOutboxInserts(false);
+ assert.ok(raw.prepare('SELECT 1 FROM vods WHERE id = 20').get(), 'not deleted without its vod.failed event');
+ assert.strictEqual(outboxRows().length, 2);
+ await finalizeVod(20);
+ assert.ok(!raw.prepare('SELECT 1 FROM vods WHERE id = 20').get(), 'deleted');
+ rows = outboxRows();
+ assert.strictEqual(rows.length, 3);
+ assert.strictEqual(rows[2].env.event_type, 'media.vod.failed');
+ assert.deepStrictEqual(rows[2].env.subject, { type: 'vod', id: '20' });
+ assert.strictEqual(rows[2].env.priority, 'important');
+ assert.ok(await waitFor(() => hooks.some((h) => JSON.parse(h.body).event_id === rows[2].id)), 'vod.failed webhook names the event');
+
+ events._reset();
+ stub.close();
+ console.log('✅ media outbox: state change and event commit together (no lost or phantom outcome), webhook carries event_id');
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });