openvibers/openvibe.media Commits
Events media.object.deleted and media.object.visibility_changed through the outbox, so consumers (Search, Live, Community) learn that an object went away or stopped being public (Media emitted neither). Payloads are minimal: media.object.deleted { object_id, app_id, kind, legacy_ref, deleted_at }, media.object.visibility_changed { object_id, app_id, kind, legacy_ref, visibility, previous_visibility, changed_at }; subject object <id>, internal, important, actor service:media, times ISO 8601 UTC, legacy_ref null for native objects, never the title, owner, storage keys, sizes or metadata. Events only, no webhook; sandbox tenants produce none. Triggers on media_objects stage one media_object_changes row per deletion (lifecycle to deleted) and per visibility change (not for an object that is deleted) inside the transaction that makes it, whichever path made it: the v1 visibility routes through projection sync, the inherited row-delete triggers, a v2 soft delete, operator SQL. recordObjectChanges() turns staged rows into outbox envelopes and removes them, and must run inside a transaction: Media calls it at the end of its own transactions (projection sync, announce(), soft delete), so those events commit with their change and a rolled-back change has none; the relay drains the rest on its interval. With the outbox off the staged rows are dropped hourly. The outbox test now expects the media.object.deleted that a finalize row delete queues with media.vod.failed. test/object-events.test.js covers each path, the payload shape, rollback, the drain and sandboxes. Contract schemas for both types are being added in OpenVibe.Contracts v0.34.0 (payload shape in the C1 handoff).
Changes
@@@ -282,6 +282,17 @@ rolled-back change; the webhook follows the commit and carries `event_id` = that(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.++Two object events go to OpenVibe.Events only (no webhook): **`media.object.deleted`**++`{ object_id, app_id, kind, legacy_ref, deleted_at }` and **`media.object.visibility_changed`**++`{ object_id, app_id, kind, legacy_ref, visibility, previous_visibility, changed_at }` (subject++`object <id>`, `important`, times ISO 8601 UTC, `legacy_ref` null for native objects). Nothing++else about the object is in them (no title, owner, keys or sizes). Triggers on `media_objects`++stage every deletion and visibility change in the transaction that makes it, whichever path made it++(a v1 visibility change, a row delete, a v2 soft delete, operator SQL); the staged rows become outbox++envelopes inside Media's own transactions (projection sync, `announce()`, soft delete), so those++commit with their change, and the relay drains anything else. No visibility event for an object++that is deleted; none for sandbox tenants. `test/object-events.test.js`.++## Restore drills (`MEDIA_DRILL=1`)`ovhost drill media` (OpenVibe.Host `docs/restore-drills.md`) restores `media.db` from the latest@@@ -135,6 +135,16 @@ function ensureObjectTriggers() {CREATE TRIGGER IF NOT EXISTS trg_media_objects_hold_delete BEFORE DELETE ON media_objectsWHEN ${held('OLD.id')}BEGIN SELECT RAISE(ABORT, 'media object is under a retention hold'); END;`);++ // media.object.visibility_changed / media.object.deleted: staged in the changing transaction,++ // whichever path changed the object (projection sync, soft delete, the row-delete triggers above,++ // operator SQL); server/events.js turns the rows into outbox envelopes.++ database.exec(`++ CREATE TRIGGER IF NOT EXISTS trg_media_objects_visibility_event AFTER UPDATE OF visibility ON media_objects++ WHEN OLD.visibility IS NOT NEW.visibility AND NEW.lifecycle_status != 'deleted'++ BEGIN INSERT INTO media_object_changes (object_id, change, previous_visibility, visibility) VALUES (NEW.id, 'visibility_changed', OLD.visibility, NEW.visibility); END;++ CREATE TRIGGER IF NOT EXISTS trg_media_objects_deleted_event AFTER UPDATE OF lifecycle_status ON media_objects++ WHEN NEW.lifecycle_status = 'deleted' AND OLD.lifecycle_status != 'deleted'++ BEGIN INSERT INTO media_object_changes (object_id, change, previous_visibility) VALUES (NEW.id, 'deleted', OLD.visibility); END;`);}/** Keep a projected row's media_object current. Never throws — the legacy write already happened. */@@@ -360,6 +360,19 @@ CREATE TABLE IF NOT EXISTS media_invariant_violations (resolved_at DATETIME);++-- Object changes waiting to become events (server/events.js recordObjectChanges): media_objects++-- triggers (database.js ensureObjectTriggers) write one row per visibility change and per deletion,++-- in the transaction that makes the change, whatever path made it; the rows become++-- media.object.visibility_changed / media.object.deleted outbox envelopes and are removed.++CREATE TABLE IF NOT EXISTS media_object_changes (++ id INTEGER PRIMARY KEY AUTOINCREMENT,++ object_id TEXT NOT NULL,++ change TEXT NOT NULL CHECK(change IN ('deleted', 'visibility_changed')),++ previous_visibility TEXT,++ visibility TEXT,++ changed_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))++);++-- Scheduled copy verification (server/objects/verify-job.js). One row per ready object: when it was-- last checked and what that check found; the least recently verified objects go first in each run.-- Verification only records and restores copies — it never deletes bytes or rows.@@@ -12,6 +12,9 @@* job state changes (server/jobs/queue.js) → media.job.proposed | queued | started | retrying |* succeeded | failed | cancelled (subject job <id>; Events only, no webhook;* payload queue.jobEvent, not the tenant's params/result/error)++ * an object deleted / its visibility changed → media.object.deleted | media.object.visibility_changed++ * (subject object <id>; Events only; payload object id, tenant,++ * kind, legacy_ref and the times/visibilities, nothing else)** 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,@@@ -46,6 +49,7 @@ const JOB_TRANSITIONS = {};let outbox = null;++let drainTimer = null;const stats = { queued: 0, lastError: null };function init({@@@ -58,7 +62,14 @@ function init({if (outbox) return outbox;// A restore drill (MEDIA_DRILL) relays nothing: its outbox rows describe a restored copy.if (require('./drill').enabled) return null;-- if (process.env.EVENTS_PUBLISH === 'off' || !eventsUrl || !clientSecret) return null;++ if (process.env.EVENTS_PUBLISH === 'off' || !eventsUrl || !clientSecret) {++ // No outbox: object changes staged by the triggers are not events here; drop them hourly.++ if (!drainTimer) {++ drainTimer = setInterval(() => { try { discardObjectChanges(); } catch { /* next hour */ } }, 60 * 60 * 1000);++ if (drainTimer.unref) drainTimer.unref();++ }++ return null;++ }const tokens = createServiceTokenClient({ tokenUrl: `${networkUrl}/oauth/token`, clientId, clientSecret, fetch: fetchImpl });const client = createClient({ baseUrls: { events: eventsUrl }, tokenProvider: tokens, fetch: fetchImpl, retries: 0 });outbox = createOutbox(db.getDb(), {@@@ -74,6 +85,11 @@ function init({outbox.start();const prune = setInterval(() => { try { outbox.prune(); } catch { /* next time */ } }, 6 * 60 * 60 * 1000);if (prune.unref) prune.unref();++ // Object changes that no Media transaction recorded (a row deleted outside one, operator SQL).++ if (drainTimer) clearInterval(drainTimer);++ drainTimer = setInterval(() => { try { drainObjectChanges(); } catch (err) { console.warn('[Events] object changes:', err.message); } }, intervalMs);++ if (drainTimer.unref) drainTimer.unref();++ try { drainObjectChanges(); } catch { /* the timer retries */ }console.log(`[Events] media outcomes → ${eventsUrl} (${outbox.pending()} pending)`);return outbox;}@@@ -145,6 +161,66 @@ function initWriter() {return outbox;}++const isoTime = (v) => {++ if (!v) return new Date().toISOString();++ const s = String(v);++ const d = new Date(/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}(\.\d+)?$/.test(s) ? `${s.replace(' ', 'T')}Z` : s);++ return Number.isNaN(d.getTime()) ? new Date().toISOString() : d.toISOString();++};++++/**++ * The event for one staged object change: [event_type, payload]. Minimal on purpose: which object++ * (id, tenant, kind, the inherited row it projects) and what changed. Never its title, metadata,++ * owner, storage keys or size: consumers GET the object with a tenant-scoped token when they need more.++ */++function objectChangeEvent(change, obj) {++ const base = { object_id: obj.id, app_id: obj.app_id, kind: obj.kind, legacy_ref: obj.legacy_ref || null };++ if (change.change === 'deleted') return ['media.object.deleted', { ...base, deleted_at: isoTime(change.changed_at) }];++ return ['media.object.visibility_changed', {++ ...base, visibility: change.visibility, previous_visibility: change.previous_visibility, changed_at: isoTime(change.changed_at),++ }];++}++++/**++ * media.object.deleted / media.object.visibility_changed. The media_objects triggers++ * (server/db/database.js) stage one media_object_changes row per change inside the transaction that++ * makes it, whatever path made it. This turns the staged rows into outbox envelopes and removes++ * them; it MUST run inside a transaction (throws otherwise), so each event and the removal of its++ * staged row commit together. Media calls it at the end of its own transactions (projection sync,++ * announce(), soft delete), so those events commit with their change; the relay's drain takes the++ * rest. With the outbox off it does nothing (the rows wait, or the service drops them hourly).++ * Returns how many rows it turned into events (sandbox tenants' rows are removed without one).++ */++function recordObjectChanges({ limit = 1000 } = {}) {++ if (!outbox) return 0;++ const raw = db.getDb();++ if (!raw.inTransaction) throw new Error('recordObjectChanges() must run inside the transaction that changed the objects');++ const rows = raw.prepare('SELECT * FROM media_object_changes ORDER BY id LIMIT ?').all(limit);++ let n = 0;++ for (const c of rows) {++ const obj = raw.prepare('SELECT id, app_id, kind, legacy_ref FROM media_objects WHERE id = ?').get(c.object_id);++ if (obj) {++ const [type, payload] = objectChangeEvent(c, obj);++ if (enqueue(type, obj.app_id, { type: 'object', id: obj.id }, payload, 'important')) n++;++ }++ raw.prepare('DELETE FROM media_object_changes WHERE id = ?').run(c.id);++ }++ return n;++}++++/** Drain staged object changes in a transaction of their own (the relay's timer). */++function drainObjectChanges() {++ if (!outbox) return 0;++ let n = 0;++ db.getDb().transaction(() => { n = recordObjectChanges(); })();++ if (n) kick();++ return n;++}++++function discardObjectChanges() {++ return db.run('DELETE FROM media_object_changes').changes;++}++/** Wake the relay once the transaction that queued events has committed. */function kick() {if (outbox) setImmediate(() => outbox && outbox.kick());@@@ -155,6 +231,11 @@ function status() {return { enabled: true, pending: outbox.pending(), rejected: outbox.rejected(), queued_since_boot: stats.queued, last_error: stats.lastError };}--function _reset() { if (outbox) outbox.stop(); outbox = null; stats.queued = 0; stats.lastError = null; }++function _reset() {++ if (outbox) outbox.stop();++ outbox = null;++ if (drainTimer) { clearInterval(drainTimer); drainTimer = null; }++ stats.queued = 0; stats.lastError = null;++}--module.exports = { init, initWriter, record, recordJob, kick, status, TYPES, JOB_TRANSITIONS, _reset };++module.exports = { init, initWriter, record, recordJob, recordObjectChanges, drainObjectChanges, discardObjectChanges, objectChangeEvent, kick, status, TYPES, JOB_TRANSITIONS, _reset };@@@ -409,8 +409,16 @@ function syncThumbnail(parentRow, parentObjectId, parentKind) {return { ...r, locations: [local.state] };}++/** One projection in one transaction, with the object-change events it caused (server/events.js). */function _syncTx(fn, arg) {-- return db.getDb().transaction(() => fn(arg))();++ const events = require('../events');++ const out = db.getDb().transaction(() => {++ const r = fn(arg);++ events.recordObjectChanges();++ return r;++ })();++ events.kick();++ return out;}/** Re-project one inherited row: kind vod|clip|file|paste, id = row id (file key for files). */@@@ -465,8 +473,13 @@ function softDelete(obj, { by = null } = {}) {md.pre_delete_status = obj.lifecycle_status;md.retention_until = new Date(Date.now() + config.objects.retentionDays * 864e5).toISOString();if (by) md.deleted_by = by;-- db.run(`UPDATE media_objects SET lifecycle_status = 'deleted', deleted_at = CURRENT_TIMESTAMP, metadata = ?, updated_at = CURRENT_TIMESTAMP-- WHERE id = ?`, [JSON.stringify(md), obj.id]);++ const events = require('../events');++ db.getDb().transaction(() => {++ db.run(`UPDATE media_objects SET lifecycle_status = 'deleted', deleted_at = CURRENT_TIMESTAMP, metadata = ?, updated_at = CURRENT_TIMESTAMP++ WHERE id = ?`, [JSON.stringify(md), obj.id]);++ events.recordObjectChanges(); // media.object.deleted commits with the delete++ })();++ events.kick();return getObject(obj.id);}@@@ -93,6 +93,7 @@ function announce(appId, event, { change = null, payload }) {if (change) change();data = typeof payload === 'function' ? payload() : payload;env = events.record(event, appId, data);++ events.recordObjectChanges(); // a row the change deleted: media.object.deleted, same transaction})();events.kick();const eventId = env ? env.event_id : null;nnew file mode 100644@@@ -0,0 +1,138 @@++'use strict';++// media.object.deleted / media.object.visibility_changed (server/events.js, the media_objects triggers++// in server/db/database.js): every deletion and visibility change of an object, whichever path makes++// it, becomes one event through Media's outbox; the events Media's own transactions cause commit in++// those transactions (projection sync, announce(), soft delete); a rolled-back change has none; the++// payload is minimal (identity and the change, no tenant data); sandbox tenants produce none.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const http = require('http');++++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-objevents-'));++const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };++Object.assign(process.env, {++ DB_PATH: path.join(tmp, 'media.db'), VOD_PATH: dir('vods'), CLIPS_PATH: dir('clips'), FILES_PATH: dir('files'),++ THUMBNAILS_PATH: dir('thumbnails'), PASTES_PATH: dir('pastes'), OBJECTS_PATH: dir('objects'),++});++++const stub = http.createServer((req, res) => {++ req.resume();++ 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 }));++ res.statusCode = 503; res.end('{}'); // Events down: rows stay in the outbox for the test to read++ });++});++++const ISO = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$/;++const DELETED_KEYS = ['app_id', 'deleted_at', 'kind', 'legacy_ref', 'object_id'];++const VIS_KEYS = ['app_id', 'changed_at', 'kind', 'legacy_ref', 'object_id', 'previous_visibility', 'visibility'];++++(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 model = require('../server/objects/model');++ const events = require('../server/events');++ db.upsertApp({ app_id: 'live', api_key: 'live-key-objevents' });++ const sandbox = db.ensureProjectTenant('prj_01TESTOBJEVENTS000000000A', 'sandbox', 1024).app_id;++ const raw = db.getDb();++ const staged = () => raw.prepare('SELECT COUNT(*) AS n FROM media_object_changes').get().n;++++ // With the outbox off, changes wait staged (and the service drops them hourly).++ const v0 = Number(db.createVod({ app_id: 'live', title: 'before the outbox', file_path: path.join(tmp, 'none.mp4') }).lastInsertRowid);++ model.sync('vod', v0);++ db.setVodVisibility(v0, 'unlisted');++ assert.strictEqual(staged(), 1, 'staged by the trigger');++ assert.strictEqual(events.discardObjectChanges(), 1);++++ assert.ok(events.init({ eventsUrl: base, clientSecret: 's', networkUrl: base, intervalMs: 3600000 }));++ const outbox = () => raw.prepare('SELECT envelope FROM event_outbox ORDER BY id').all().map((r) => JSON.parse(r.envelope));++ const objectEvents = () => outbox().filter((e) => e.event_type.startsWith('media.object.'));++++ // ── 1. A visibility change through the v1 route's path (projection sync): one event, same transaction ──++ const vodId = Number(db.createVod({ app_id: 'live', user_id: 7, title: 'Secret title', file_path: path.join(tmp, 'v.mp4'), meta: { a: 1 } }).lastInsertRowid);++ model.sync('vod', vodId);++ const obj = model.getObject(db.get('SELECT object_id FROM vods WHERE id = ?', [vodId]).object_id);++ assert.strictEqual(objectEvents().length, 0, 'creating an object is not one of these events');++ db.setVodVisibility(vodId, 'private');++ let evs = objectEvents();++ assert.strictEqual(evs.length, 1);++ assert.strictEqual(staged(), 0, 'recorded in the sync transaction, nothing left staged');++ const e = evs[0];++ assert.strictEqual(e.event_type, 'media.object.visibility_changed');++ assert.deepStrictEqual(e.subject, { type: 'object', id: obj.id });++ assert.deepStrictEqual([e.source, e.visibility, e.priority, e.actor], ['media', 'internal', 'important', { type: 'service', id: 'media' }]);++ assert.deepStrictEqual(Object.keys(e.payload).sort(), VIS_KEYS, 'exactly the minimal payload');++ assert.deepStrictEqual([e.payload.object_id, e.payload.app_id, e.payload.kind, e.payload.legacy_ref, e.payload.visibility, e.payload.previous_visibility],++ [obj.id, 'live', 'vod', `legacy:live:vod:${vodId}`, 'private', 'public']);++ assert.ok(ISO.test(e.payload.changed_at));++ assert.ok(!JSON.stringify(e.payload).includes('Secret title') && !('owner_user_id' in e.payload), 'no tenant data');++ db.setVodVisibility(vodId, 'private');++ model.sync('vod', vodId);++ assert.strictEqual(objectEvents().length, 1, 'no event when nothing changed');++ console.log('✅ visibility change: one media.object.visibility_changed, minimal payload, committed with the change');++++ // ── 2. A rolled-back change has no event ──++ assert.throws(() => raw.transaction(() => {++ raw.prepare("UPDATE media_objects SET visibility = 'public' WHERE id = ?").run(obj.id);++ events.recordObjectChanges();++ throw new Error('rollback');++ })(), /rollback/);++ assert.strictEqual(objectEvents().length, 1);++ assert.strictEqual(staged(), 0);++ assert.throws(() => events.recordObjectChanges(), /inside the transaction/);++ console.log('✅ a rolled-back change leaves neither an event nor a staged row');++++ // ── 3. A row deleted inside announce(): media.object.deleted in the same transaction ──++ const { announce } = require('../server/webhooks');++ announce('live', 'vod.failed', { change: () => db.run('DELETE FROM vods WHERE id = ?', [vodId]), payload: { id: vodId } });++ evs = objectEvents();++ assert.strictEqual(evs.length, 2);++ assert.strictEqual(evs[1].event_type, 'media.object.deleted');++ assert.deepStrictEqual(Object.keys(evs[1].payload).sort(), DELETED_KEYS);++ assert.deepStrictEqual([evs[1].payload.object_id, evs[1].payload.kind, evs[1].payload.legacy_ref], [obj.id, 'vod', `legacy:live:vod:${vodId}`]);++ assert.ok(ISO.test(evs[1].payload.deleted_at));++ const all = outbox();++ assert.strictEqual(all[all.length - 2].event_type, 'media.vod.failed', 'queued together with the outcome');++ console.log('✅ a vods row deleted in announce(): media.object.deleted with media.vod.failed, one transaction');++++ // ── 4. A native object soft-deleted through the v2 path ──++ const nativeId = model.createObject({ app_id: 'live', kind: 'file', visibility: 'unlisted', lifecycle_status: 'ready' });++ model.softDelete(model.getObject(nativeId), { by: 'test' });++ evs = objectEvents();++ assert.deepStrictEqual([evs[2].event_type, evs[2].payload.object_id, evs[2].payload.legacy_ref, evs[2].payload.kind], ['media.object.deleted', nativeId, null, 'file']);++ raw.prepare("UPDATE media_objects SET visibility = 'public' WHERE id = ?").run(nativeId);++ assert.strictEqual(staged(), 0, 'a deleted object\'s visibility is not an event');++ console.log('✅ soft delete of a native object: media.object.deleted');++++ // ── 5. Any other path (a row deleted outside a transaction, operator SQL): staged, then drained ──++ const clipId = Number(db.createClip({ app_id: 'live', vod_id: null, title: 'c', file_path: path.join(tmp, 'c.webm') }).lastInsertRowid);++ const clipObj = db.get('SELECT object_id FROM clips WHERE id = ?', [clipId]).object_id;++ const before = objectEvents().length;++ db.run('DELETE FROM clips WHERE id = ?', [clipId]);++ assert.strictEqual(staged(), 1, 'the trigger staged it with the delete');++ assert.strictEqual(objectEvents().length, before);++ assert.strictEqual(events.drainObjectChanges(), 1);++ evs = objectEvents();++ assert.deepStrictEqual([evs[evs.length - 1].event_type, evs[evs.length - 1].payload.object_id, evs[evs.length - 1].payload.kind], ['media.object.deleted', clipObj, 'clip']);++ assert.strictEqual(staged(), 0);++ console.log('✅ a change made outside Media\'s transactions is staged by the trigger and drained into the outbox');++++ // ── 6. Sandbox tenants produce no platform events ──++ const sb = model.createObject({ app_id: sandbox, kind: 'file', visibility: 'private', lifecycle_status: 'ready' });++ const n = objectEvents().length;++ raw.transaction(() => { raw.prepare("UPDATE media_objects SET visibility = 'unlisted' WHERE id = ?").run(sb); events.recordObjectChanges(); })();++ assert.strictEqual(objectEvents().length, n, 'none for a sandbox');++ assert.strictEqual(staged(), 0);++ console.log('✅ sandbox tenants: no events');++++ events._reset();++ stub.close();++ db.close();++ fs.rmSync(tmp, { recursive: true, force: true });++ console.log('\n✅ All object event tests passed');++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });@@@ -139,7 +139,11 @@ const waitFor = async (fn, ms = 3000) => {await finalizeVod(20);assert.ok(!raw.prepare('SELECT 1 FROM vods WHERE id = 20').get(), 'deleted');rows = outboxRows();-- assert.strictEqual(rows.length, 3);++ // The row's object is marked deleted by the row-delete trigger in the same transaction, so its++ // media.object.deleted is queued with media.vod.failed.++ assert.strictEqual(rows.length, 4);++ assert.strictEqual(rows[3].env.event_type, 'media.object.deleted');++ assert.strictEqual(rows[3].env.payload.legacy_ref, 'legacy:live:vod:20');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');
@@ -282,6 +282,17 @@ rolled-back change; the webhook follows the commit and carries `event_id` = that
@@ -282,6 +282,17 @@ rolled-back change; the webhook follows the commit and carries `event_id` = that
(absent when the outbox is off), so an app that reads both paths (Live during its webhook →
(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.
Events transition) handles each outcome once. Webhooks stay until every consumer reads Events.
+Two object events go to OpenVibe.Events only (no webhook): **`media.object.deleted`**
+`{ object_id, app_id, kind, legacy_ref, deleted_at }` and **`media.object.visibility_changed`**
+`{ object_id, app_id, kind, legacy_ref, visibility, previous_visibility, changed_at }` (subject
+`object <id>`, `important`, times ISO 8601 UTC, `legacy_ref` null for native objects). Nothing
+else about the object is in them (no title, owner, keys or sizes). Triggers on `media_objects`
+stage every deletion and visibility change in the transaction that makes it, whichever path made it
+(a v1 visibility change, a row delete, a v2 soft delete, operator SQL); the staged rows become outbox
+envelopes inside Media's own transactions (projection sync, `announce()`, soft delete), so those
+commit with their change, and the relay drains anything else. No visibility event for an object
+that is deleted; none for sandbox tenants. `test/object-events.test.js`.
+
## Restore drills (`MEDIA_DRILL=1`)
## Restore drills (`MEDIA_DRILL=1`)
`ovhost drill media` (OpenVibe.Host `docs/restore-drills.md`) restores `media.db` from the latest
`ovhost drill media` (OpenVibe.Host `docs/restore-drills.md`) restores `media.db` from the latest
@@ -135,6 +135,16 @@ function ensureObjectTriggers() {
@@ -135,6 +135,16 @@ function ensureObjectTriggers() {
CREATE TRIGGER IF NOT EXISTS trg_media_objects_hold_delete BEFORE DELETE ON media_objects
CREATE TRIGGER IF NOT EXISTS trg_media_objects_hold_delete BEFORE DELETE ON media_objects
WHEN ${held('OLD.id')}
WHEN ${held('OLD.id')}
BEGIN SELECT RAISE(ABORT, 'media object is under a retention hold'); END;`);
BEGIN SELECT RAISE(ABORT, 'media object is under a retention hold'); END;`);
+ // media.object.visibility_changed / media.object.deleted: staged in the changing transaction,
+ // whichever path changed the object (projection sync, soft delete, the row-delete triggers above,
+ // operator SQL); server/events.js turns the rows into outbox envelopes.
+ database.exec(`
+ CREATE TRIGGER IF NOT EXISTS trg_media_objects_visibility_event AFTER UPDATE OF visibility ON media_objects
+ WHEN OLD.visibility IS NOT NEW.visibility AND NEW.lifecycle_status != 'deleted'
+ BEGIN INSERT INTO media_object_changes (object_id, change, previous_visibility, visibility) VALUES (NEW.id, 'visibility_changed', OLD.visibility, NEW.visibility); END;
+ CREATE TRIGGER IF NOT EXISTS trg_media_objects_deleted_event AFTER UPDATE OF lifecycle_status ON media_objects
+ WHEN NEW.lifecycle_status = 'deleted' AND OLD.lifecycle_status != 'deleted'
+ BEGIN INSERT INTO media_object_changes (object_id, change, previous_visibility) VALUES (NEW.id, 'deleted', OLD.visibility); END;`);
}
}
/** Keep a projected row's media_object current. Never throws — the legacy write already happened. */
/** Keep a projected row's media_object current. Never throws — the legacy write already happened. */
@@ -360,6 +360,19 @@ CREATE TABLE IF NOT EXISTS media_invariant_violations (
@@ -360,6 +360,19 @@ CREATE TABLE IF NOT EXISTS media_invariant_violations (
resolved_at DATETIME
resolved_at DATETIME
);
);
+-- Object changes waiting to become events (server/events.js recordObjectChanges): media_objects
+-- triggers (database.js ensureObjectTriggers) write one row per visibility change and per deletion,
+-- in the transaction that makes the change, whatever path made it; the rows become
+-- media.object.visibility_changed / media.object.deleted outbox envelopes and are removed.
+CREATE TABLE IF NOT EXISTS media_object_changes (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ object_id TEXT NOT NULL,
+ change TEXT NOT NULL CHECK(change IN ('deleted', 'visibility_changed')),
+ previous_visibility TEXT,
+ visibility TEXT,
+ changed_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
+);
+
-- Scheduled copy verification (server/objects/verify-job.js). One row per ready object: when it was
-- Scheduled copy verification (server/objects/verify-job.js). One row per ready object: when it was
-- last checked and what that check found; the least recently verified objects go first in each run.
-- last checked and what that check found; the least recently verified objects go first in each run.
-- Verification only records and restores copies — it never deletes bytes or rows.
-- Verification only records and restores copies — it never deletes bytes or rows.
@@ -12,6 +12,9 @@
@@ -12,6 +12,9 @@
* job state changes (server/jobs/queue.js) → media.job.proposed | queued | started | retrying |
* job state changes (server/jobs/queue.js) → media.job.proposed | queued | started | retrying |
* succeeded | failed | cancelled (subject job <id>; Events only, no webhook;
* succeeded | failed | cancelled (subject job <id>; Events only, no webhook;
* payload queue.jobEvent, not the tenant's params/result/error)
* payload queue.jobEvent, not the tenant's params/result/error)
+ * an object deleted / its visibility changed → media.object.deleted | media.object.visibility_changed
+ * (subject object <id>; Events only; payload object id, tenant,
+ * kind, legacy_ref and the times/visibilities, nothing else)
*
*
* The outbox row is written INSIDE the SQLite transaction that makes the state change it describes
* 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,
* (record(), called from webhooks.announce()): the event exists if and only if the change committed,
@@ -46,6 +49,7 @@ const JOB_TRANSITIONS = {
@@ -46,6 +49,7 @@ const JOB_TRANSITIONS = {
};
};
let outbox = null;
let outbox = null;
+let drainTimer = null;
const stats = { queued: 0, lastError: null };
const stats = { queued: 0, lastError: null };
function init({
function init({
@@ -58,7 +62,14 @@ function init({
@@ -58,7 +62,14 @@ function init({
if (outbox) return outbox;
if (outbox) return outbox;
// A restore drill (MEDIA_DRILL) relays nothing: its outbox rows describe a restored copy.
// A restore drill (MEDIA_DRILL) relays nothing: its outbox rows describe a restored copy.
if (require('./drill').enabled) return null;
if (require('./drill').enabled) return null;
- if (process.env.EVENTS_PUBLISH === 'off' || !eventsUrl || !clientSecret) return null;
+ if (process.env.EVENTS_PUBLISH === 'off' || !eventsUrl || !clientSecret) {
+ // No outbox: object changes staged by the triggers are not events here; drop them hourly.
+ if (!drainTimer) {
+ drainTimer = setInterval(() => { try { discardObjectChanges(); } catch { /* next hour */ } }, 60 * 60 * 1000);
+ if (drainTimer.unref) drainTimer.unref();
+ }
+ return null;
+ }
const tokens = createServiceTokenClient({ tokenUrl: `${networkUrl}/oauth/token`, clientId, clientSecret, fetch: fetchImpl });
const tokens = createServiceTokenClient({ tokenUrl: `${networkUrl}/oauth/token`, clientId, clientSecret, fetch: fetchImpl });
const client = createClient({ baseUrls: { events: eventsUrl }, tokenProvider: tokens, fetch: fetchImpl, retries: 0 });
const client = createClient({ baseUrls: { events: eventsUrl }, tokenProvider: tokens, fetch: fetchImpl, retries: 0 });
outbox = createOutbox(db.getDb(), {
outbox = createOutbox(db.getDb(), {
@@ -74,6 +85,11 @@ function init({
@@ -74,6 +85,11 @@ function init({
outbox.start();
outbox.start();
const prune = setInterval(() => { try { outbox.prune(); } catch { /* next time */ } }, 6 * 60 * 60 * 1000);
const prune = setInterval(() => { try { outbox.prune(); } catch { /* next time */ } }, 6 * 60 * 60 * 1000);
if (prune.unref) prune.unref();
if (prune.unref) prune.unref();
+ // Object changes that no Media transaction recorded (a row deleted outside one, operator SQL).
+ if (drainTimer) clearInterval(drainTimer);
+ drainTimer = setInterval(() => { try { drainObjectChanges(); } catch (err) { console.warn('[Events] object changes:', err.message); } }, intervalMs);
+ if (drainTimer.unref) drainTimer.unref();
+ try { drainObjectChanges(); } catch { /* the timer retries */ }
console.log(`[Events] media outcomes → ${eventsUrl} (${outbox.pending()} pending)`);
console.log(`[Events] media outcomes → ${eventsUrl} (${outbox.pending()} pending)`);
return outbox;
return outbox;
}
}
@@ -145,6 +161,66 @@ function initWriter() {
@@ -145,6 +161,66 @@ function initWriter() {
return outbox;
return outbox;
}
}
+const isoTime = (v) => {
+ if (!v) return new Date().toISOString();
+ const s = String(v);
+ const d = new Date(/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}(\.\d+)?$/.test(s) ? `${s.replace(' ', 'T')}Z` : s);
+ return Number.isNaN(d.getTime()) ? new Date().toISOString() : d.toISOString();
+};
+
+/**
+ * The event for one staged object change: [event_type, payload]. Minimal on purpose: which object
+ * (id, tenant, kind, the inherited row it projects) and what changed. Never its title, metadata,
+ * owner, storage keys or size: consumers GET the object with a tenant-scoped token when they need more.
+ */
+function objectChangeEvent(change, obj) {
+ const base = { object_id: obj.id, app_id: obj.app_id, kind: obj.kind, legacy_ref: obj.legacy_ref || null };
+ if (change.change === 'deleted') return ['media.object.deleted', { ...base, deleted_at: isoTime(change.changed_at) }];
+ return ['media.object.visibility_changed', {
+ ...base, visibility: change.visibility, previous_visibility: change.previous_visibility, changed_at: isoTime(change.changed_at),
+ }];
+}
+
+/**
+ * media.object.deleted / media.object.visibility_changed. The media_objects triggers
+ * (server/db/database.js) stage one media_object_changes row per change inside the transaction that
+ * makes it, whatever path made it. This turns the staged rows into outbox envelopes and removes
+ * them; it MUST run inside a transaction (throws otherwise), so each event and the removal of its
+ * staged row commit together. Media calls it at the end of its own transactions (projection sync,
+ * announce(), soft delete), so those events commit with their change; the relay's drain takes the
+ * rest. With the outbox off it does nothing (the rows wait, or the service drops them hourly).
+ * Returns how many rows it turned into events (sandbox tenants' rows are removed without one).
+ */
+function recordObjectChanges({ limit = 1000 } = {}) {
+ if (!outbox) return 0;
+ const raw = db.getDb();
+ if (!raw.inTransaction) throw new Error('recordObjectChanges() must run inside the transaction that changed the objects');
+ const rows = raw.prepare('SELECT * FROM media_object_changes ORDER BY id LIMIT ?').all(limit);
+ let n = 0;
+ for (const c of rows) {
+ const obj = raw.prepare('SELECT id, app_id, kind, legacy_ref FROM media_objects WHERE id = ?').get(c.object_id);
+ if (obj) {
+ const [type, payload] = objectChangeEvent(c, obj);
+ if (enqueue(type, obj.app_id, { type: 'object', id: obj.id }, payload, 'important')) n++;
+ }
+ raw.prepare('DELETE FROM media_object_changes WHERE id = ?').run(c.id);
+ }
+ return n;
+}
+
+/** Drain staged object changes in a transaction of their own (the relay's timer). */
+function drainObjectChanges() {
+ if (!outbox) return 0;
+ let n = 0;
+ db.getDb().transaction(() => { n = recordObjectChanges(); })();
+ if (n) kick();
+ return n;
+}
+
+function discardObjectChanges() {
+ return db.run('DELETE FROM media_object_changes').changes;
+}
+
/** Wake the relay once the transaction that queued events has committed. */
/** Wake the relay once the transaction that queued events has committed. */
function kick() {
function kick() {
if (outbox) setImmediate(() => outbox && outbox.kick());
if (outbox) setImmediate(() => outbox && outbox.kick());
@@ -155,6 +231,11 @@ function status() {
@@ -155,6 +231,11 @@ function status() {
return { enabled: true, pending: outbox.pending(), rejected: outbox.rejected(), queued_since_boot: stats.queued, last_error: stats.lastError };
return { enabled: true, pending: outbox.pending(), rejected: outbox.rejected(), queued_since_boot: stats.queued, last_error: stats.lastError };
}
}
-function _reset() { if (outbox) outbox.stop(); outbox = null; stats.queued = 0; stats.lastError = null; }
+function _reset() {
+ if (outbox) outbox.stop();
+ outbox = null;
+ if (drainTimer) { clearInterval(drainTimer); drainTimer = null; }
+ stats.queued = 0; stats.lastError = null;
+}
-module.exports = { init, initWriter, record, recordJob, kick, status, TYPES, JOB_TRANSITIONS, _reset };
+module.exports = { init, initWriter, record, recordJob, recordObjectChanges, drainObjectChanges, discardObjectChanges, objectChangeEvent, kick, status, TYPES, JOB_TRANSITIONS, _reset };
@@ -409,8 +409,16 @@ function syncThumbnail(parentRow, parentObjectId, parentKind) {
@@ -409,8 +409,16 @@ function syncThumbnail(parentRow, parentObjectId, parentKind) {
return { ...r, locations: [local.state] };
return { ...r, locations: [local.state] };
}
}
+/** One projection in one transaction, with the object-change events it caused (server/events.js). */
function _syncTx(fn, arg) {
function _syncTx(fn, arg) {
- return db.getDb().transaction(() => fn(arg))();
+ const events = require('../events');
+ const out = db.getDb().transaction(() => {
+ const r = fn(arg);
+ events.recordObjectChanges();
+ return r;
+ })();
+ events.kick();
+ return out;
}
}
/** Re-project one inherited row: kind vod|clip|file|paste, id = row id (file key for files). */
/** Re-project one inherited row: kind vod|clip|file|paste, id = row id (file key for files). */
@@ -465,8 +473,13 @@ function softDelete(obj, { by = null } = {}) {
@@ -465,8 +473,13 @@ function softDelete(obj, { by = null } = {}) {
md.pre_delete_status = obj.lifecycle_status;
md.pre_delete_status = obj.lifecycle_status;
md.retention_until = new Date(Date.now() + config.objects.retentionDays * 864e5).toISOString();
md.retention_until = new Date(Date.now() + config.objects.retentionDays * 864e5).toISOString();
if (by) md.deleted_by = by;
if (by) md.deleted_by = by;
- db.run(`UPDATE media_objects SET lifecycle_status = 'deleted', deleted_at = CURRENT_TIMESTAMP, metadata = ?, updated_at = CURRENT_TIMESTAMP
- WHERE id = ?`, [JSON.stringify(md), obj.id]);
+ const events = require('../events');
+ db.getDb().transaction(() => {
+ db.run(`UPDATE media_objects SET lifecycle_status = 'deleted', deleted_at = CURRENT_TIMESTAMP, metadata = ?, updated_at = CURRENT_TIMESTAMP
+ WHERE id = ?`, [JSON.stringify(md), obj.id]);
+ events.recordObjectChanges(); // media.object.deleted commits with the delete
+ })();
+ events.kick();
return getObject(obj.id);
return getObject(obj.id);
}
}
@@ -93,6 +93,7 @@ function announce(appId, event, { change = null, payload }) {
@@ -93,6 +93,7 @@ function announce(appId, event, { change = null, payload }) {
if (change) change();
if (change) change();
data = typeof payload === 'function' ? payload() : payload;
data = typeof payload === 'function' ? payload() : payload;
env = events.record(event, appId, data);
env = events.record(event, appId, data);
+ events.recordObjectChanges(); // a row the change deleted: media.object.deleted, same transaction
})();
})();
events.kick();
events.kick();
const eventId = env ? env.event_id : null;
const eventId = env ? env.event_id : null;
new file mode 100644
new file mode 100644
@@ -0,0 +1,138 @@
@@ -0,0 +1,138 @@
+'use strict';
+// media.object.deleted / media.object.visibility_changed (server/events.js, the media_objects triggers
+// in server/db/database.js): every deletion and visibility change of an object, whichever path makes
+// it, becomes one event through Media's outbox; the events Media's own transactions cause commit in
+// those transactions (projection sync, announce(), soft delete); a rolled-back change has none; the
+// payload is minimal (identity and the change, no tenant data); sandbox tenants produce none.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const http = require('http');
+
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-objevents-'));
+const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
+Object.assign(process.env, {
+ DB_PATH: path.join(tmp, 'media.db'), VOD_PATH: dir('vods'), CLIPS_PATH: dir('clips'), FILES_PATH: dir('files'),
+ THUMBNAILS_PATH: dir('thumbnails'), PASTES_PATH: dir('pastes'), OBJECTS_PATH: dir('objects'),
+});
+
+const stub = http.createServer((req, res) => {
+ req.resume();
+ 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 }));
+ res.statusCode = 503; res.end('{}'); // Events down: rows stay in the outbox for the test to read
+ });
+});
+
+const ISO = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$/;
+const DELETED_KEYS = ['app_id', 'deleted_at', 'kind', 'legacy_ref', 'object_id'];
+const VIS_KEYS = ['app_id', 'changed_at', 'kind', 'legacy_ref', 'object_id', 'previous_visibility', 'visibility'];
+
+(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 model = require('../server/objects/model');
+ const events = require('../server/events');
+ db.upsertApp({ app_id: 'live', api_key: 'live-key-objevents' });
+ const sandbox = db.ensureProjectTenant('prj_01TESTOBJEVENTS000000000A', 'sandbox', 1024).app_id;
+ const raw = db.getDb();
+ const staged = () => raw.prepare('SELECT COUNT(*) AS n FROM media_object_changes').get().n;
+
+ // With the outbox off, changes wait staged (and the service drops them hourly).
+ const v0 = Number(db.createVod({ app_id: 'live', title: 'before the outbox', file_path: path.join(tmp, 'none.mp4') }).lastInsertRowid);
+ model.sync('vod', v0);
+ db.setVodVisibility(v0, 'unlisted');
+ assert.strictEqual(staged(), 1, 'staged by the trigger');
+ assert.strictEqual(events.discardObjectChanges(), 1);
+
+ assert.ok(events.init({ eventsUrl: base, clientSecret: 's', networkUrl: base, intervalMs: 3600000 }));
+ const outbox = () => raw.prepare('SELECT envelope FROM event_outbox ORDER BY id').all().map((r) => JSON.parse(r.envelope));
+ const objectEvents = () => outbox().filter((e) => e.event_type.startsWith('media.object.'));
+
+ // ── 1. A visibility change through the v1 route's path (projection sync): one event, same transaction ──
+ const vodId = Number(db.createVod({ app_id: 'live', user_id: 7, title: 'Secret title', file_path: path.join(tmp, 'v.mp4'), meta: { a: 1 } }).lastInsertRowid);
+ model.sync('vod', vodId);
+ const obj = model.getObject(db.get('SELECT object_id FROM vods WHERE id = ?', [vodId]).object_id);
+ assert.strictEqual(objectEvents().length, 0, 'creating an object is not one of these events');
+ db.setVodVisibility(vodId, 'private');
+ let evs = objectEvents();
+ assert.strictEqual(evs.length, 1);
+ assert.strictEqual(staged(), 0, 'recorded in the sync transaction, nothing left staged');
+ const e = evs[0];
+ assert.strictEqual(e.event_type, 'media.object.visibility_changed');
+ assert.deepStrictEqual(e.subject, { type: 'object', id: obj.id });
+ assert.deepStrictEqual([e.source, e.visibility, e.priority, e.actor], ['media', 'internal', 'important', { type: 'service', id: 'media' }]);
+ assert.deepStrictEqual(Object.keys(e.payload).sort(), VIS_KEYS, 'exactly the minimal payload');
+ assert.deepStrictEqual([e.payload.object_id, e.payload.app_id, e.payload.kind, e.payload.legacy_ref, e.payload.visibility, e.payload.previous_visibility],
+ [obj.id, 'live', 'vod', `legacy:live:vod:${vodId}`, 'private', 'public']);
+ assert.ok(ISO.test(e.payload.changed_at));
+ assert.ok(!JSON.stringify(e.payload).includes('Secret title') && !('owner_user_id' in e.payload), 'no tenant data');
+ db.setVodVisibility(vodId, 'private');
+ model.sync('vod', vodId);
+ assert.strictEqual(objectEvents().length, 1, 'no event when nothing changed');
+ console.log('✅ visibility change: one media.object.visibility_changed, minimal payload, committed with the change');
+
+ // ── 2. A rolled-back change has no event ──
+ assert.throws(() => raw.transaction(() => {
+ raw.prepare("UPDATE media_objects SET visibility = 'public' WHERE id = ?").run(obj.id);
+ events.recordObjectChanges();
+ throw new Error('rollback');
+ })(), /rollback/);
+ assert.strictEqual(objectEvents().length, 1);
+ assert.strictEqual(staged(), 0);
+ assert.throws(() => events.recordObjectChanges(), /inside the transaction/);
+ console.log('✅ a rolled-back change leaves neither an event nor a staged row');
+
+ // ── 3. A row deleted inside announce(): media.object.deleted in the same transaction ──
+ const { announce } = require('../server/webhooks');
+ announce('live', 'vod.failed', { change: () => db.run('DELETE FROM vods WHERE id = ?', [vodId]), payload: { id: vodId } });
+ evs = objectEvents();
+ assert.strictEqual(evs.length, 2);
+ assert.strictEqual(evs[1].event_type, 'media.object.deleted');
+ assert.deepStrictEqual(Object.keys(evs[1].payload).sort(), DELETED_KEYS);
+ assert.deepStrictEqual([evs[1].payload.object_id, evs[1].payload.kind, evs[1].payload.legacy_ref], [obj.id, 'vod', `legacy:live:vod:${vodId}`]);
+ assert.ok(ISO.test(evs[1].payload.deleted_at));
+ const all = outbox();
+ assert.strictEqual(all[all.length - 2].event_type, 'media.vod.failed', 'queued together with the outcome');
+ console.log('✅ a vods row deleted in announce(): media.object.deleted with media.vod.failed, one transaction');
+
+ // ── 4. A native object soft-deleted through the v2 path ──
+ const nativeId = model.createObject({ app_id: 'live', kind: 'file', visibility: 'unlisted', lifecycle_status: 'ready' });
+ model.softDelete(model.getObject(nativeId), { by: 'test' });
+ evs = objectEvents();
+ assert.deepStrictEqual([evs[2].event_type, evs[2].payload.object_id, evs[2].payload.legacy_ref, evs[2].payload.kind], ['media.object.deleted', nativeId, null, 'file']);
+ raw.prepare("UPDATE media_objects SET visibility = 'public' WHERE id = ?").run(nativeId);
+ assert.strictEqual(staged(), 0, 'a deleted object\'s visibility is not an event');
+ console.log('✅ soft delete of a native object: media.object.deleted');
+
+ // ── 5. Any other path (a row deleted outside a transaction, operator SQL): staged, then drained ──
+ const clipId = Number(db.createClip({ app_id: 'live', vod_id: null, title: 'c', file_path: path.join(tmp, 'c.webm') }).lastInsertRowid);
+ const clipObj = db.get('SELECT object_id FROM clips WHERE id = ?', [clipId]).object_id;
+ const before = objectEvents().length;
+ db.run('DELETE FROM clips WHERE id = ?', [clipId]);
+ assert.strictEqual(staged(), 1, 'the trigger staged it with the delete');
+ assert.strictEqual(objectEvents().length, before);
+ assert.strictEqual(events.drainObjectChanges(), 1);
+ evs = objectEvents();
+ assert.deepStrictEqual([evs[evs.length - 1].event_type, evs[evs.length - 1].payload.object_id, evs[evs.length - 1].payload.kind], ['media.object.deleted', clipObj, 'clip']);
+ assert.strictEqual(staged(), 0);
+ console.log('✅ a change made outside Media\'s transactions is staged by the trigger and drained into the outbox');
+
+ // ── 6. Sandbox tenants produce no platform events ──
+ const sb = model.createObject({ app_id: sandbox, kind: 'file', visibility: 'private', lifecycle_status: 'ready' });
+ const n = objectEvents().length;
+ raw.transaction(() => { raw.prepare("UPDATE media_objects SET visibility = 'unlisted' WHERE id = ?").run(sb); events.recordObjectChanges(); })();
+ assert.strictEqual(objectEvents().length, n, 'none for a sandbox');
+ assert.strictEqual(staged(), 0);
+ console.log('✅ sandbox tenants: no events');
+
+ events._reset();
+ stub.close();
+ db.close();
+ fs.rmSync(tmp, { recursive: true, force: true });
+ console.log('\n✅ All object event tests passed');
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });
@@ -139,7 +139,11 @@ const waitFor = async (fn, ms = 3000) => {
@@ -139,7 +139,11 @@ const waitFor = async (fn, ms = 3000) => {
await finalizeVod(20);
await finalizeVod(20);
assert.ok(!raw.prepare('SELECT 1 FROM vods WHERE id = 20').get(), 'deleted');
assert.ok(!raw.prepare('SELECT 1 FROM vods WHERE id = 20').get(), 'deleted');
rows = outboxRows();
rows = outboxRows();
- assert.strictEqual(rows.length, 3);
+ // The row's object is marked deleted by the row-delete trigger in the same transaction, so its
+ // media.object.deleted is queued with media.vod.failed.
+ assert.strictEqual(rows.length, 4);
+ assert.strictEqual(rows[3].env.event_type, 'media.object.deleted');
+ assert.strictEqual(rows[3].env.payload.legacy_ref, 'legacy:live:vod:20');
assert.strictEqual(rows[2].env.event_type, 'media.vod.failed');
assert.strictEqual(rows[2].env.event_type, 'media.vod.failed');
assert.deepStrictEqual(rows[2].env.subject, { type: 'vod', id: '20' });
assert.deepStrictEqual(rows[2].env.subject, { type: 'vod', id: '20' });
assert.strictEqual(rows[2].env.priority, 'important');
assert.strictEqual(rows[2].env.priority, 'important');