openvibers/openvibe.live Commits

Deploys are announced as OpenVibe.Events events: the deploy notice queues live.release.deployed (subject release <head sha>; payload service, release, commit, previous, commit_count, commits[] up to 40, deployed_at, notes_url; visibility internal) through Live's outbox in the SAME transaction that records the commits as announced and, with local chat, stores the rolling chat row, so there is never a notice without its event or the other way round; the chat notice itself is unchanged (local row, or OpenVibe.Chat over the bridge). stream-events.js exposes enqueue()/kick() so Live's other events share the one outbox. openvibe-contracts v0.29.0 → v0.30.1 (no contract for this type yet: a live.* contract addition). test/release-events.test.js: envelope valid against event-envelope@1, published, silent without new code, rolled back when the outbox insert fails, bridge path kept

OP OpenVibers committed 23 Sept 2026 · e73a92d004 · 6 files, +231 −15


Changes

ddiff --git a/package-lock.json b/package-lock.json
iindex 35e4b691..baceea4d 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -25,7 +25,7 @@
"multer": "^1.4.5-lts.1",
"node-media-server": "^2.7.1",
"onnxruntime-node": "^1.20.0",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.29.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.30.1",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.4.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.3.0",
"sanitize-html": "^2.17.7",
@@@ -2814,9 +2814,9 @@
}
},
"node_modules/openvibe-contracts": {
-- "version": "0.29.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.29.0",
-- "integrity": "sha512-IJhCi0qY7AuvFUt3+QMLkA0vi4WbUecIyQ3rdT7mF0U+pMTDxHZm2XHxpjfF0t7XPhoRCfgnKRZ4M7q7vLlkIA==",
++ "version": "0.30.1",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.30.1",
++ "integrity": "sha512-JAOna8MA/+4iABEZ5qnzGjW27z9UgAHDF1Pj4VE+R3tb1UVyRUVZX06yuE+BggTdhZPGZRq5mFFIyPNHp8N3GA==",
"license": "AGPL-3.0-only",
"dependencies": {
"ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex 027046b4..6ce653e4 100644
---- a/package.json
++++ b/package.json
@@@ -43,7 +43,7 @@
"multer": "^1.4.5-lts.1",
"node-media-server": "^2.7.1",
"onnxruntime-node": "^1.20.0",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.29.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.30.1",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.4.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.3.0",
"sanitize-html": "^2.17.7",
ddiff --git a/server/chat/deploy-notice.js b/server/chat/deploy-notice.js
iindex 33f88114..a2ba4ff9 100644
---- a/server/chat/deploy-notice.js
++++ b/server/chat/deploy-notice.js
@@@ -15,6 +15,11 @@
// updated_at }. Clients render times from ISO timestamps, so they are always right.
// • The live broadcast carries the row id; clients replace the card with that id instead of
// appending, so reconnects and repeats can never duplicate it.
++// • The deploy is also a durable OpenVibe.Events event, live.release.deployed
++// (server/events/release-events.js), queued in the SAME transaction that records the commits
++// as announced (and, without the Chat service, stores the chat row). The chat notice itself
++// still goes to OpenVibe.Chat over the bridge (chat-remote deployNotice) until Chat consumes
++// the event; then that hop can go.
// ═══════════════════════════════════════════════════════════════
const { execFile } = require('child_process');
const path = require('path');
@@@ -41,15 +46,16 @@ const parseLog = (raw) => raw.trim().split('\n').filter(Boolean).map((line) => {
/** Commits on HEAD that have not been announced yet (newest first). */
async function newCommits(db) {
const head = (await git(['rev-parse', 'HEAD'], 3000)).trim();
-- if (!/^[0-9a-f]{40}$/.test(head)) return { head: '', commits: [] };
++ if (!/^[0-9a-f]{40}$/.test(head)) return { head: '', previous: null, commits: [] };
const last = String(db.getSetting(SETTING) || '').trim();
-- if (last === head) return { head, commits: [] };
++ const previous = /^[0-9a-f]{40}$/.test(last) ? last : null;
++ if (last === head) return { head, previous, commits: [] };
const fmt = '--pretty=format:%H%x1f%h%x1f%aI%x1f%s';
let raw = '';
if (/^[0-9a-f]{40}$/.test(last)) raw = await git(['--no-pager', 'log', fmt, '-n', String(MAX_COMMITS), `${last}..${head}`]);
// First run, or history was rewritten: announce the current commit only, never a backlog.
if (!raw.trim()) raw = await git(['--no-pager', 'log', fmt, '-n', '1', head]);
-- return { head, commits: parseLog(raw) };
++ return { head, previous, commits: parseLog(raw) };
}
const plainText = (meta) => `🚀 ${meta.commits.length} update${meta.commits.length === 1 ? '' : 's'} shipped: ${meta.commits.slice(0, 3).map(c => c.subject).join(' · ')}${meta.commits.length > 3 ? ` · and ${meta.commits.length - 3} more` : ''}`;
@@@ -76,23 +82,40 @@ function persist(db, commits) {
/**
* Announce this boot's new commits. Safe to call once after the chat server is up.
-- * @returns {Promise<{ announced: number }>}
++ * @returns {Promise<{ announced: number, event_id?: string|null }>}
*/
async function announce({ db, chatServer, log = console }) {
-- const { head, commits } = await newCommits(db);
++ const { head, previous, commits } = await newCommits(db);
if (!head || !commits.length) return { announced: 0 };
++ // Live's outbox exists before the transaction (idempotent; null while Events is off).
++ let outbox = null;
++ try { outbox = require('../events/stream-events'); outbox.init(); } catch (err) { log.warn('[Deploy notice] Events outbox unavailable:', err.message); outbox = null; }
++ let eventId = null;
++ // Recording the commits as announced and queueing live.release.deployed are one commit: the
++ // event exists if and only if this deploy counts as announced.
++ const recordDeploy = () => {
++ db.setSetting(SETTING, head);
++ if (outbox) {
++ const env = require('../events/release-events').record({ head, previous, commits });
++ eventId = env ? env.event_id : null;
++ }
++ };
++ const inTransaction = (fn) => db.getDb().transaction(fn)();
++
// CHAT_AUTHORITY=chat: Live still decides what shipped; OpenVibe.Chat stores the rolling
// message and shows it (its own copy of this module), so hand it the commits.
if (chatServer && chatServer.remote) {
chatServer.deployNotice(commits);
-- try { db.setSetting(SETTING, head); } catch (err) { log.warn('[Deploy notice] not recorded:', err.message); }
-- return { announced: commits.length };
++ try { inTransaction(recordDeploy); } catch (err) { log.warn('[Deploy notice] not recorded:', err.message); return { announced: commits.length, event_id: null }; }
++ if (outbox) outbox.kick();
++ return { announced: commits.length, event_id: eventId };
}
let saved;
-- try { saved = persist(db, commits); db.setSetting(SETTING, head); }
++ try { saved = inTransaction(() => { const s = persist(db, commits); recordDeploy(); return s; }); }
catch (err) { log.warn('[Deploy notice] not saved:', err.message); return { announced: 0 }; }
++ if (outbox) outbox.kick();
const payload = JSON.stringify({ type: 'update', kind: 'deploy', id: saved.id, fresh: commits.map(c => c.hash), url: '/updates', timestamp: saved.meta.updated_at, ...saved.meta });
const sent = new WeakSet();
@@@ -106,7 +129,7 @@ async function announce({ db, chatServer, log = console }) {
return n;
};
ATTEMPTS_MS.forEach((ms, i) => { const t = setTimeout(() => { const n = push(); if (n) log.log(`[Deploy notice] ${commits.length} commit(s) → ${n} client(s) (pass ${i + 1})`); }, ms); if (t.unref) t.unref(); });
-- return { announced: commits.length };
++ return { announced: commits.length, event_id: eventId };
}
/**
ddiff --git a/server/events/release-events.js b/server/events/release-events.js
nnew file mode 100644
iindex 00000000..e05a188b
---- /dev/null
++++ b/server/events/release-events.js
@@@ -0,0 +1,55 @@
++'use strict';
++/**
++ * live.release.deployed — Live announces a deploy that shipped new commits as a durable
++ * OpenVibe.Events event (roadmap Wave 3 exit: "the Live chat deploy notice goes through Events").
++ *
++ * Emitted by server/chat/deploy-notice.js on the boot that first runs new commits, in the SAME
++ * transaction that records them as announced (site setting deploy_last_announced), through Live's
++ * outbox (server/events/stream-events.js). A restart without new code emits nothing; a crash before
++ * that commit re-announces on the next boot, with the same subject (the head commit), so a consumer
++ * keys on subject.id to fold repeats.
++ *
++ * subject { type: 'release', id: <head commit, 40 hex> }
++ * payload { service: 'live', release: <short sha>, commit: <head>, previous: <last announced
++ * head or null>, commit_count, commits: [{ hash, short, subject, date }] (newest
++ * first, at most 40), deployed_at, notes_url }
++ *
++ * Visibility is `internal`: the commit subjects are public in chat and on /updates already, but
++ * the consumer (the chat notice) decides what is shown. No contract in openvibe-contracts v0.30.1
++ * defines this type yet; it is listed as a contract addition for the `live.*` namespace Live owns.
++ */
++const EVENT_TYPE = 'live.release.deployed';
++const NOTES_URL = 'https://openvibe.live/updates';
++
++function envelopeFor({ head, previous = null, commits = [], deployedAt = new Date().toISOString() }) {
++ if (!/^[0-9a-f]{40}$/.test(String(head || ''))) throw new TypeError('head must be a full commit sha');
++ const list = commits.slice(0, 40).map((c) => ({ hash: c.hash, short: c.short, subject: String(c.subject || '').slice(0, 200), date: c.date || null }));
++ const headShort = (list.find((c) => c.hash === head) || {}).short;
++ return {
++ event_type: EVENT_TYPE,
++ actor: { type: 'service', id: 'live' },
++ subject: { type: 'release', id: head },
++ visibility: 'internal',
++ priority: 'low',
++ payload: {
++ service: 'live',
++ release: /^[0-9a-f]{7,40}$/.test(String(headShort || '')) ? headShort : head.slice(0, 12),
++ commit: head,
++ previous: /^[0-9a-f]{40}$/.test(String(previous || '')) ? previous : null,
++ commit_count: commits.length,
++ commits: list,
++ deployed_at: deployedAt,
++ notes_url: NOTES_URL,
++ },
++ };
++}
++
++/**
++ * Queue the event INSIDE the caller's transaction. Returns the envelope, or null when Live's
++ * outbox is off (no EVENTS_URL). Throws if the outbox insert fails (the caller's change rolls back).
++ */
++function record(args) {
++ return require('./stream-events').enqueue(envelopeFor(args));
++}
++
++module.exports = { EVENT_TYPE, envelopeFor, record };
ddiff --git a/server/events/stream-events.js b/server/events/stream-events.js
iindex 17b4e6d9..86606c09 100644
---- a/server/events/stream-events.js
++++ b/server/events/stream-events.js
@@@ -10,6 +10,9 @@
* token (audience openvibe.events, capability events.event.publish). If Events or Network is down
* the rows wait and are retried with backoff; going live never waits on either.
*
++ * The same outbox carries Live's other events (enqueue()): live.release.deployed from the deploy
++ * notice (server/events/release-events.js).
++ *
* Off unless EVENTS_URL and OV_OAUTH_CLIENT_SECRET are set (EVENTS_PUBLISH=off disables it).
* Payloads carry public channel facts only (the stream is already listed publicly); consumers such
* as Network's go-live notifications decide who hears about it.
@@@ -100,6 +103,22 @@ function init({ eventsUrl = EVENTS_URL, clientSecret = CLIENT_SECRET, fetchImpl,
return outbox;
}
++/**
++ * Queue another Live event in the outbox. MUST run inside the transaction that makes the change it
++ * describes (the SDK outbox refuses otherwise); throws when the insert fails so the change rolls
++ * back with it. Returns the envelope, or null while publishing is off. Call init() before the
++ * transaction (it is idempotent), and kick() after the commit.
++ */
++function enqueue(envelope) {
++ if (!outbox) return null;
++ const env = outbox.enqueue(envelope);
++ stats.queued++;
++ return env;
++}
++
++/** Wake the relay once the transaction that queued an event has committed. */
++function kick() { if (outbox) setImmediate(() => outbox && outbox.kick()); }
++
function status() {
if (!outbox) return { enabled: false };
return { enabled: true, pending: outbox.pending(), rejected: outbox.rejected(), queued_since_boot: stats.queued, last_error: stats.lastError };
@@@ -107,4 +126,4 @@ function status() {
function _reset() { if (outbox) outbox.stop(); outbox = null; db.onStreamLifecycle(null); stats.queued = 0; stats.lastError = null; }
--module.exports = { init, status, envelopeFor, _reset };
++module.exports = { init, enqueue, kick, status, envelopeFor, _reset };
ddiff --git a/test/release-events.test.js b/test/release-events.test.js
nnew file mode 100644
iindex 00000000..64340e27
---- /dev/null
++++ b/test/release-events.test.js
@@@ -0,0 +1,119 @@
++'use strict';
++
++// Deploys as events (roadmap Wave 3 exit): the deploy notice queues live.release.deployed in the
++// same transaction that records the commits as announced (and stores the chat row), publishes it
++// through Live's outbox, keeps the chat notice working (local row, or the Chat bridge), says
++// nothing on a restart without new code, and rolls the notice back if the event cannot be queued.
++
++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-release-events-'));
++process.env.DB_PATH = path.join(tmp, 'live.db');
++
++const published = [];
++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' && req.method === 'POST') {
++ const parsed = JSON.parse(body);
++ const list = parsed.events || [parsed];
++ const results = list.map((e) => { published.push(e); return { event_id: e.event_id, seq: published.length, duplicate: false }; });
++ return res.end(JSON.stringify(parsed.events ? { results } : results[0]));
++ }
++ res.statusCode = 404; res.end('{}');
++ });
++});
++
++(async () => {
++ await new Promise((r) => stub.listen(0, '127.0.0.1', r));
++ const base = `http://127.0.0.1:${stub.address().port}`;
++ process.env.OV_NETWORK_INTERNAL_URL = base;
++
++ const contracts = require('openvibe-contracts');
++ const db = require('../server/db/database');
++ db.initDb();
++ const raw = db.getDb();
++ const dn = require('../server/chat/deploy-notice');
++ const releaseEvents = require('../server/events/release-events');
++ const streamEvents = require('../server/events/stream-events');
++ const quiet = { log() {}, warn() {} };
++ const outboxRows = () => raw.prepare('SELECT event_id, envelope FROM event_outbox ORDER BY id').all().map(r => JSON.parse(r.envelope)).filter(e => e.event_type === releaseEvents.EVENT_TYPE);
++ const deployRows = () => raw.prepare("SELECT id, metadata FROM chat_messages WHERE message_type = 'system' AND metadata LIKE '%\"kind\":\"deploy\"%'").all();
++
++ // The envelope is a valid event-envelope@1.
++ const head = (await new Promise(r => require('child_process').execFile('git', ['rev-parse', 'HEAD'], { cwd: path.join(__dirname, '..') }, (e, o) => r(String(o).trim()))));
++ assert.match(head, /^[0-9a-f]{40}$/);
++ const sample = releaseEvents.envelopeFor({ head, previous: null, commits: [{ hash: head, short: head.slice(0, 7), subject: 'x', date: '2026-09-23T00:00:00Z' }] });
++ const full = { ...sample, event_id: 'evt_01JAB2C3D4E5F6G7H8J9K0MNPQ', version: 1, source: 'live', timestamp: new Date().toISOString() };
++ contracts.assertValid('events.event-envelope@1', full);
++ assert.strictEqual(sample.payload.release, head.slice(0, 7));
++ assert.throws(() => releaseEvents.envelopeFor({ head: 'nope' }), /full commit sha/);
++
++ // Events off: the notice works exactly as before and nothing is queued.
++ let r = await dn.announce({ db, chatServer: { remote: false, clients: new Map() }, log: quiet });
++ assert.ok(r.announced >= 1);
++ assert.strictEqual(r.event_id, null);
++ assert.strictEqual(db.getSetting(dn.SETTING), head);
++ assert.strictEqual(deployRows().length, 1);
++
++ // Events on: a new deploy (the setting points elsewhere) queues and publishes the event.
++ const outbox = streamEvents.init({ eventsUrl: base, clientSecret: 's3cret', intervalMs: 50 });
++ assert.ok(outbox);
++ db.setSetting(dn.SETTING, '');
++ r = await dn.announce({ db, chatServer: { remote: false, clients: new Map() }, log: quiet });
++ assert.ok(r.announced >= 1);
++ assert.match(r.event_id, /^evt_[0-9A-HJKMNP-TV-Z]{26}$/);
++ let rows = outboxRows();
++ assert.strictEqual(rows.length, 1);
++ assert.strictEqual(rows[0].event_id, r.event_id);
++ assert.deepStrictEqual(rows[0].subject, { type: 'release', id: head });
++ assert.strictEqual(rows[0].payload.commit, head);
++ assert.strictEqual(rows[0].payload.service, 'live');
++ assert.ok(rows[0].payload.commits.length >= 1 && rows[0].payload.commits[0].hash === head);
++ contracts.assertValid('events.event-envelope@1', rows[0]);
++ await outbox.flush();
++ const sent = published.find(e => e.event_id === r.event_id);
++ assert.ok(sent, 'published to Events');
++ assert.strictEqual(sent.source, 'live');
++ assert.strictEqual(sent.event_type, 'live.release.deployed');
++
++ // A restart with no new code: no notice, no event.
++ r = await dn.announce({ db, chatServer: { remote: false, clients: new Map() }, log: quiet });
++ assert.strictEqual(r.announced, 0);
++ assert.strictEqual(outboxRows().length, 1);
++
++ // The event cannot be queued: nothing is recorded (no chat row, setting unchanged), so the next
++ // boot announces again — never a notice without its event.
++ db.setSetting(dn.SETTING, '');
++ const chatBefore = deployRows().map(x => x.metadata).join('|');
++ raw.exec("CREATE TEMP TRIGGER outbox_boom BEFORE INSERT ON event_outbox BEGIN SELECT RAISE(ABORT, 'outbox insert failed'); END");
++ r = await dn.announce({ db, chatServer: { remote: false, clients: new Map() }, log: quiet });
++ raw.exec('DROP TRIGGER temp.outbox_boom');
++ assert.strictEqual(r.announced, 0);
++ assert.strictEqual(db.getSetting(dn.SETTING), '');
++ assert.strictEqual(deployRows().map(x => x.metadata).join('|'), chatBefore, 'the chat row rolled back too');
++ assert.strictEqual(outboxRows().length, 1);
++
++ // CHAT_AUTHORITY=chat: the chat notice still goes over the bridge, and the event is queued.
++ const bridged = [];
++ r = await dn.announce({ db, chatServer: { remote: true, deployNotice: (commits) => bridged.push(commits) }, log: quiet });
++ assert.ok(r.announced >= 1);
++ assert.strictEqual(bridged.length, 1);
++ assert.strictEqual(bridged[0][0].hash, head);
++ assert.strictEqual(db.getSetting(dn.SETTING), head);
++ rows = outboxRows();
++ assert.strictEqual(rows.length, 2);
++ assert.strictEqual(rows[1].event_id, r.event_id);
++
++ streamEvents._reset();
++ stub.close();
++ console.log('release events: live.release.deployed queued with the announcement, published, chat notice kept — all checks passed');
++ process.exit(0);
++})().catch((err) => { console.error(err); process.exit(1); });