openvibers/openvibe.sdk Commits
SDK 0.15.0: the event outbox and inbox on PostgreSQL (ADR-004 on ADR-035), in openvibe-sdk/events. createPgOutbox(db, { events }) takes an openvibe-sdk/db handle. enqueue(t, envelope) requires the transaction handle, so the event exists if and only if the change commits. The relay claims due rows with a lease (FOR UPDATE SKIP LOCKED), so any number of processes and hosts relay one table without sending an event twice, and a dead relay's rows return after the lease. Backoff, per-row permanent refusals, the kept seq, pending, rejected and prune are as before. createPgInbox(db).once(consumer, eventId, async (t) => …) runs a handler once in one transaction with its receipt; a failed handler leaves none. outboxSchema() and inboxSchema() give the DDL for service migrations. The SQLite versions stay for services not yet moved. The db slow-query log is off by default on PGlite (its first query includes the WASM start). test/pg-outbox.test.js runs on PGlite and through PgBouncer, where three relays sharing one table send 60 events with none twice. npm test 36/36 with the containers.
Changes
@@@ -3,6 +3,18 @@All notable changes to `openvibe-sdk`. The package follows semver; while it is `0.x`, a minorrelease may change an API and says so here.++## 0.15.0 (2026-09-28)++++**The event outbox and inbox on PostgreSQL** (ADR-004 on ADR-035), in `openvibe-sdk/events`. The better-sqlite3 versions stay for services that have not moved yet.++- **`createPgOutbox(db, { events })`:** takes an openvibe-sdk/db handle.++ - `enqueue(t, envelope)` takes the transaction handle, so the event exists if and only if the change commits.++ - The relay claims due rows with a lease (`FOR UPDATE SKIP LOCKED`), so any number of processes and hosts relay one table without sending an event twice. A relay that dies leaves its rows to the next one once the lease expires.++ - Otherwise it behaves as before: backoff on transient failures, permanent refusals isolated per row, the seq kept, `pending`, `rejected` and `prune`.++- **`createPgInbox(db)`:** `once(consumer, eventId, async (t) => …)` runs the handler once, in one transaction with its receipt, and a failed handler leaves no receipt.++- **`outboxSchema()` and `inboxSchema()`:** the DDL to put in a service migration, since the runtime role cannot create tables.++- **Quieter tests:** the slow-query log is off by default on PGlite, whose first query includes its WASM start.++- **Tests:** `test/pg-outbox.test.js` runs on PGlite, and through PgBouncer with three relays sharing one table (60 events, none sent twice).++## 0.14.0 (2026-09-28)**The async data layer and Valkey modules** (ADR-035 and its amendment; roadmap WS-X2): every service moves to PostgreSQL 18 and Valkey, async-first.@@@ -3,7 +3,7 @@"stage": "alpha","code": true,"wave": "W2",-- "version": "0.14.0",++ "version": "0.15.0","plan": "OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026), §3.2; roadmap §30","deployed": false,"deployment": "none: a package released as git tags; services and apps pin a tag",@@@ -21,7 +21,8 @@"mock platform for tests (openvibe-sdk/testing)","ESM entries, TypeScript types and a browser bundle","db (ADR-035): the async PostgreSQL data layer (node-postgres through PgBouncer; PGlite in tests) with safe sql fragments, retrying transactions and savepoints, versioned expand/migrate/contract migrations, and importSqlite for the one-time SQLite move with verification",-- "valkey, cache, queue, pubsub and a shared limits store on Valkey 9, each with an in-process twin for tests"++ "valkey, cache, queue, pubsub and a shared limits store on Valkey 9, each with an in-process twin for tests",++ "events: the PostgreSQL outbox (lease-claimed, several relays per table) and inbox (handler in its transaction)"],"updated": "2026-09-28"}@@@ -1,4 +1,4 @@--// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.14.0, MIT). Do not edit.++// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.15.0, MIT). Do not edit.// One self-contained ES module: core at the top level plus the browser-safe subpaths as namespaces// (auth, registry, modules, realtime, media, community, search, chat, ai, jobs, tools, projects). No server-only code: no service tokens, JWT verification, events, identity./* eslint-disable */@@@ -40,7 +40,7 @@ const { paginate, offsetPager } = __load(7);const semver = __load(6);/** The package version (test/version.test.js keeps it equal to package.json). */--const SDK_VERSION = '0.14.0';++const SDK_VERSION = '0.15.0';module.exports = {createClient,@@@ -3,6 +3,7 @@ import cjs from '../src/events.js';export const {createEventsClient, verifyDelivery, signDelivery, parseDelivery, createOutbox, createInbox,-- projectKey, appSource, createAppEvents, verifyDeliveryV2, signDeliveryV2, signDeliveryHeaders,++ createPgOutbox, createPgInbox, outboxSchema, inboxSchema, projectKey, appSource,++ createAppEvents, verifyDeliveryV2, signDeliveryV2, signDeliveryHeaders,} = cjs;export default cjs;@@@ -1,12 +1,12 @@{"name": "openvibe-sdk",-- "version": "0.14.0",++ "version": "0.15.0","lockfileVersion": 3,"requires": true,"packages": {"": {"name": "openvibe-sdk",-- "version": "0.14.0",++ "version": "0.15.0","license": "MIT","devDependencies": {"@electric-sql/pglite": "^0.5.8",@@@ -1,6 +1,6 @@{"name": "openvibe-sdk",-- "version": "0.14.0",++ "version": "0.15.0","description": "Supported browser and server clients for the OpenVibe platform: auth (incl. developer apps), registry discovery, identity, user modules, events, realtime, media, community, search, chat, AI runs, Tools (registry, run API and jobs), VIP memberships and developer projects, the async PostgreSQL data layer (db) and the Valkey cache, limits store, queue and pub/sub, with deadlines, safe retries, problem errors and trace propagation.","main": "index.js","types": "types/index.d.ts",@@@ -8,7 +8,7 @@ const { paginate, offsetPager } = require('./paginate');const semver = require('./semver');/** The package version (test/version.test.js keeps it equal to package.json). */--const SDK_VERSION = '0.14.0';++const SDK_VERSION = '0.15.0';module.exports = {createClient,@@@ -85,7 +85,7 @@ function queryApi(run, self) {function createDb(o = {}) {const log = o.log || console;const service = o.service || process.env.OV_SERVICE || 'service';-- const slowMs = o.slowMs == null ? 500 : o.slowMs;++ const slowMs = o.slowMs == null ? (o.pglite ? Infinity : 500) : o.slowMs; // PGlite's first query includes its WASM startconst stats = { queries: 0, errors: 0, slow: 0, retries: 0, tx: 0 };let hist = null;if (o.registry && typeof o.registry.histogram === 'function') {@@@ -355,9 +355,10 @@ function createAppEvents(client, { projectId, appId, onBehalfOf, baseUrl } = {})};}--const { createOutbox, createInbox } = require('./outbox');++const { createOutbox, createInbox, createPgOutbox, createPgInbox, outboxSchema, inboxSchema } = require('./outbox');module.exports = {createEventsClient, verifyDelivery, signDelivery, parseDelivery, createOutbox, createInbox,++ createPgOutbox, createPgInbox, outboxSchema, inboxSchema,projectKey, appSource, createAppEvents, verifyDeliveryV2, signDeliveryV2, signDeliveryHeaders,};@@@ -1,7 +1,8 @@'use strict';/**-- * openvibe-sdk/events (server): transactional outbox and inbox on the service's own better-sqlite3-- * database (ADR-004). No dependency: the caller passes its database handle.++ * openvibe-sdk/events (server): transactional outbox and inbox (ADR-004), on the service's own database:++ * createOutbox/createInbox on better-sqlite3 (services not yet moved), createPgOutbox/createPgInbox on an++ * openvibe-sdk/db handle (PostgreSQL, ADR-035; below). No dependency: the caller passes its database handle.** const events = createEventsClient(client, { source: 'live' });* const outbox = createOutbox(db, { events });@@@ -207,4 +208,168 @@ function createInbox(db, { table = 'idempotency_receipts', now = () => Date.now(return { ensureSchema, once, seen };}--module.exports = { createOutbox, createInbox, isPermanent };++// ── PostgreSQL (ADR-035): the same outbox and inbox on an openvibe-sdk/db handle ─────────────────++++/** The outbox table's DDL, for a service migration (the runtime role cannot create tables). */++function outboxSchema(table = 'event_outbox') {++ if (!TABLE_RE.test(table)) throw new TypeError('bad outbox table name');++ return `CREATE TABLE IF NOT EXISTS ${table} (++ id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,++ event_id text NOT NULL UNIQUE,++ envelope jsonb NOT NULL,++ traceparent text,++ created_at bigint NOT NULL,++ attempts integer NOT NULL DEFAULT 0,++ next_attempt_at bigint NOT NULL DEFAULT 0,++ sent_at bigint,++ seq bigint,++ rejected_at bigint,++ last_error text++);++CREATE INDEX IF NOT EXISTS ${table}_due ON ${table} (next_attempt_at, id) WHERE sent_at IS NULL AND rejected_at IS NULL;++CREATE INDEX IF NOT EXISTS ${table}_sent ON ${table} (sent_at) WHERE sent_at IS NOT NULL;`;++}++++/** The inbox table's DDL, for a service migration. */++function inboxSchema(table = 'idempotency_receipts') {++ if (!TABLE_RE.test(table)) throw new TypeError('bad inbox table name');++ return `CREATE TABLE IF NOT EXISTS ${table} (++ consumer text NOT NULL,++ event_id text NOT NULL,++ processed_at bigint NOT NULL,++ PRIMARY KEY (consumer, event_id)++);`;++}++++/**++ * createPgOutbox(db, { events }) on an openvibe-sdk/db handle. enqueue takes the caller's transaction handle,++ * so the event exists if and only if the change commits:++ *++ * await db.tx(async (t) => {++ * await t.exec(sql`UPDATE streams SET is_live = true WHERE id = ${id}`);++ * await outbox.enqueue(t, { event_type: 'live.stream.started', actor, subject, payload });++ * });++ * outbox.kick();++ *++ * The relay claims due rows with a short lease (FOR UPDATE SKIP LOCKED), so any number of processes and hosts++ * relay one table without double-sending; a relay that dies mid-batch leaves its rows to the next after the++ * lease. Delivery stays at least once (Events answers a repeated event_id as a duplicate).++ */++function createPgOutbox(db, {++ events, table = 'event_outbox', batchSize = 50, intervalMs = 1000, leaseMs = 60000,++ backoffMs = [1000, 5000, 30000, 120000, 600000], now = () => Date.now(), onError = null,++} = {}) {++ if (!db || typeof db.query !== 'function' || typeof db.tx !== 'function') throw new TypeError('an openvibe-sdk/db handle is required');++ if (!events || typeof events.publish !== 'function' || typeof events.prepare !== 'function') {++ throw new TypeError('events must be an openvibe-sdk events client (createEventsClient)');++ }++ if (!TABLE_RE.test(table)) throw new TypeError('bad outbox table name');++ let timer = null; let running = false; let flushing = null;++ const backoff = (attempts) => backoffMs[Math.min(attempts, backoffMs.length - 1)];++++ async function enqueue(t, envelope, { traceparent } = {}) {++ if (!t || typeof t.query !== 'function' || typeof t.tx !== 'function') throw new TypeError('outbox.enqueue(t, …): pass the transaction handle db.tx gives you');++ const m = traceparent && TRACEPARENT_RE.exec(traceparent);++ const env = events.prepare(envelope, { traceId: m ? m[1] : undefined, now: now() });++ await t.query(`INSERT INTO ${table} (event_id, envelope, traceparent, created_at) VALUES ($1, $2, $3, $4)`, [env.event_id, JSON.stringify(env), m ? traceparent : null, now()]);++ return env;++ }++++ async function claim() {++ const t = now();++ return db.many(`UPDATE ${table} SET next_attempt_at = $1++ WHERE id IN (SELECT id FROM ${table} WHERE sent_at IS NULL AND rejected_at IS NULL AND next_attempt_at <= $2++ ORDER BY id LIMIT $3 FOR UPDATE SKIP LOCKED)++ RETURNING id, event_id, envelope, traceparent, attempts`, [t + leaseMs, t, batchSize]);++ }++++ async function markFailure(row, err) {++ const message = String((err && err.message) || err).slice(0, 500);++ if (isPermanent(err)) await db.query(`UPDATE ${table} SET rejected_at = $1, attempts = attempts + 1, last_error = $2 WHERE id = $3`, [now(), message, row.id]);++ else await db.query(`UPDATE ${table} SET attempts = attempts + 1, next_attempt_at = $1, last_error = $2 WHERE id = $3`, [now() + backoff(row.attempts), message, row.id]);++ if (onError) { try { onError(err, row); } catch { /* reporting must not break the relay */ } }++ }++++ async function publishRows(rows) {++ const stats = { sent: 0, failed: 0, rejected: 0 };++ const envelopes = rows.map((r) => (typeof r.envelope === 'string' ? JSON.parse(r.envelope) : r.envelope));++ const traceparent = rows[0].traceparent || undefined;++ try {++ const out = rows.length === 1++ ? { results: [await events.publish(envelopes[0], { traceparent })] }++ : await events.publish(envelopes, { traceparent });++ const byId = new Map(((out && out.results) || []).map((r) => [r.event_id, r]));++ await db.query(`UPDATE ${table} AS o SET sent_at = $1, seq = v.seq, attempts = o.attempts + 1, last_error = NULL++ FROM unnest($2::bigint[], $3::bigint[]) AS v(id, seq) WHERE o.id = v.id`,++ [now(), rows.map((r) => r.id), rows.map((r) => (byId.get(r.event_id) && byId.get(r.event_id).seq) ?? null)]);++ stats.sent += rows.length;++ } catch (err) {++ if (rows.length > 1 && isPermanent(err)) {++ for (const row of rows) { const s = await publishRows([row]); stats.sent += s.sent; stats.failed += s.failed; stats.rejected += s.rejected; }++ } else {++ for (const row of rows) await markFailure(row, err);++ if (isPermanent(err)) stats.rejected += rows.length; else stats.failed += rows.length;++ }++ }++ return stats;++ }++++ async function doFlush() {++ const total = { sent: 0, failed: 0, rejected: 0 };++ for (;;) {++ const rows = await claim();++ if (!rows.length) break;++ const s = await publishRows(rows);++ total.sent += s.sent; total.failed += s.failed; total.rejected += s.rejected;++ if (s.failed || rows.length < batchSize) break;++ }++ return total;++ }++ function flush() { if (!flushing) flushing = doFlush().finally(() => { flushing = null; }); return flushing; }++ function schedule(ms) {++ if (!running) return;++ clearTimeout(timer);++ timer = setTimeout(async () => {++ timer = null;++ try { await flush(); } catch (err) { if (onError) { try { onError(err); } catch { /* ignore */ } } }++ schedule(intervalMs);++ }, ms);++ if (timer.unref) timer.unref();++ }++ return {++ schema: () => outboxSchema(table),++ /** Create the table where the handle may (tests, PGlite); services put schema() in a migration. */++ ensureSchema: () => db.query(outboxSchema(table)),++ enqueue, flush,++ start() { if (!running) { running = true; schedule(0); } },++ stop() { running = false; clearTimeout(timer); timer = null; return flushing || Promise.resolve(); },++ kick() { schedule(0); },++ pending: () => db.value(`SELECT count(*) FROM ${table} WHERE sent_at IS NULL AND rejected_at IS NULL`),++ rejected: () => db.value(`SELECT count(*) FROM ${table} WHERE rejected_at IS NOT NULL`),++ prune: (olderThanMs = 7 * 24 * 60 * 60 * 1000) => db.exec(`DELETE FROM ${table} WHERE sent_at IS NOT NULL AND sent_at < $1`, [now() - olderThanMs]),++ };++}++++/**++ * createPgInbox(db): run fn once per (consumer, eventId), in one transaction with its receipt.++ * const r = await inbox.once('network', event.event_id, async (t) => { … writes with t … }); // { duplicate, result }++ */++function createPgInbox(db, { table = 'idempotency_receipts', now = () => Date.now() } = {}) {++ if (!db || typeof db.tx !== 'function') throw new TypeError('an openvibe-sdk/db handle is required');++ if (!TABLE_RE.test(table)) throw new TypeError('bad inbox table name');++ return {++ schema: () => inboxSchema(table),++ ensureSchema: () => db.query(inboxSchema(table)),++ async once(consumer, eventId, fn) {++ if (!consumer || !eventId) throw new TypeError('consumer and eventId are required');++ return db.tx(async (t) => {++ const n = await t.exec(`INSERT INTO ${table} (consumer, event_id, processed_at) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING`, [String(consumer), String(eventId), now()]);++ if (n === 0) return { duplicate: true };++ return { duplicate: false, result: await fn(t) };++ });++ },++ seen: async (consumer, eventId) => Boolean(await db.maybe(`SELECT 1 FROM ${table} WHERE consumer = $1 AND event_id = $2`, [String(consumer), String(eventId)])),++ };++}++++module.exports = { createOutbox, createInbox, createPgOutbox, createPgInbox, outboxSchema, inboxSchema, isPermanent };nnew file mode 100644@@@ -0,0 +1,123 @@++'use strict';++/**++ * The PostgreSQL outbox and inbox (ADR-004 on ADR-035): enqueue needs the transaction handle, so a rolled-back++ * change leaves no event; the relay publishes once, backs off on a transient failure, rejects only a permanently++ * refused row, and several relays on one table never send an event twice (FOR UPDATE SKIP LOCKED with a lease);++ * the inbox runs a handler once per event in the handler's transaction. PGlite always; PostgreSQL through++ * PgBouncer when OV_TEST_PG_URL is set.++ */++const assert = require('node:assert/strict');++const { run } = require('./helpers');++const { createClient } = require('../src/core');++const { createServiceTokenClient } = require('../src/auth');++const { createEventsClient, createPgOutbox, createPgInbox, outboxSchema, inboxSchema } = require('../src/events');++const { createMockPlatform } = require('../src/testing');++const { createDb, sql } = require('../src/db');++++const actor = { type: 'service', id: 'live' };++const subject = { type: 'stream', id: '12', revision: 1 };++++function eventsClient() {++ const platform = createMockPlatform({ clients: { live: { secret: 'live-secret', grants: [['events.event.publish', 'openvibe.events']] } } });++ const tokens = createServiceTokenClient({ clientId: 'live', clientSecret: 'live-secret', fetch: platform.fetch });++ const client = createClient({ fetch: platform.fetch, tokenProvider: tokens, retries: 0 });++ return { platform, events: createEventsClient(client, { source: 'live' }) };++}++++function cases(label, open) {++ return [++ [`${label}: enqueue needs the transaction; a rollback leaves no event; a commit publishes once`, async () => {++ const { db, done } = await open();++ try {++ const { platform, events } = eventsClient();++ const outbox = createPgOutbox(db, { events });++ await assert.rejects(outbox.enqueue(null, { event_type: 'live.stream.started', actor, subject }), /transaction handle/);++ await assert.rejects(db.tx(async (t) => {++ await t.exec(sql`INSERT INTO streams (id, is_live) VALUES (1, true)`);++ await outbox.enqueue(t, { event_type: 'live.stream.started', actor, subject });++ throw new Error('boom');++ }), /boom/);++ assert.equal(await outbox.pending(), 0);++ const env = await db.tx(async (t) => {++ await t.exec(sql`INSERT INTO streams (id, is_live) VALUES (2, true)`);++ return outbox.enqueue(t, { event_type: 'live.stream.started', actor, subject, payload: { stream_id: 2 } }, { traceparent: '00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01' });++ });++ assert.equal(env.trace_id, '0af7651916cd43dd8448eb211c80319c');++ assert.deepEqual(await outbox.flush(), { sent: 1, failed: 0, rejected: 0 });++ assert.equal(await outbox.pending(), 0);++ assert.equal(platform.state.events.length, 1);++ assert.equal(platform.state.events[0].event.event_id, env.event_id);++ await outbox.flush();++ assert.equal(platform.state.events.length, 1, 'nothing republished');++ assert.ok((await db.value(sql`SELECT seq FROM event_outbox WHERE event_id = ${env.event_id}`)) >= 1, 'the seq Events assigned is kept');++ } finally { await done(); }++ }],++ [`${label}: transient failures back off; a permanent refusal rejects only its row`, async () => {++ const { db, done } = await open();++ try {++ const { platform, events } = eventsClient();++ let t = 1_000_000; let fail = true;++ const flaky = { prepare: events.prepare, publish: (...a) => (fail ? Promise.reject(Object.assign(new Error('down'), { status: 503 })) : events.publish(...a)) };++ const outbox = createPgOutbox(db, { events: flaky, now: () => t });++ await db.tx(async (tx) => { for (let i = 0; i < 3; i++) await outbox.enqueue(tx, { event_type: 'live.stream.started', actor, subject: { type: 'stream', id: String(i) } }); });++ assert.deepEqual(await outbox.flush(), { sent: 0, failed: 3, rejected: 0 });++ fail = false;++ assert.deepEqual(await outbox.flush(), { sent: 0, failed: 0, rejected: 0 }, 'not due yet');++ t += 1000;++ assert.deepEqual(await outbox.flush(), { sent: 3, failed: 0, rejected: 0 });++ assert.equal(platform.state.events.length, 3);++ const picky = { prepare: events.prepare, publish: (env) => (Array.isArray(env) || env.subject.id === 'bad' ? Promise.reject(Object.assign(new Error('invalid'), { status: 422 })) : events.publish(env)) };++ const o2 = createPgOutbox(db, { events: picky, now: () => t });++ await db.tx(async (tx) => { await o2.enqueue(tx, { event_type: 'live.stream.started', actor, subject: { type: 'stream', id: 'bad' } }); await o2.enqueue(tx, { event_type: 'live.stream.started', actor, subject: { type: 'stream', id: 'ok' } }); });++ assert.deepEqual(await o2.flush(), { sent: 1, failed: 0, rejected: 1 });++ assert.equal(await o2.rejected(), 1);++ } finally { await done(); }++ }],++ [`${label}: the inbox runs a handler once, inside its transaction`, async () => {++ const { db, done } = await open();++ try {++ const inbox = createPgInbox(db);++ let runs = 0;++ const a = await inbox.once('network', 'evt_1', async (t) => { runs++; await t.exec(sql`INSERT INTO streams (id, is_live) VALUES (7, false)`); return 'ok'; });++ const b = await inbox.once('network', 'evt_1', async () => { runs++; });++ assert.deepEqual([a, b, runs], [{ duplicate: false, result: 'ok' }, { duplicate: true }, 1]);++ await assert.rejects(inbox.once('network', 'evt_2', async () => { throw new Error('fail'); }), /fail/);++ assert.equal(await inbox.seen('network', 'evt_2'), false, 'a failed handler leaves no receipt: it runs again next time');++ } finally { await done(); }++ }],++ ];++}++++const schema = `CREATE TABLE streams (id bigint PRIMARY KEY, is_live boolean NOT NULL);\n${outboxSchema()}\n${inboxSchema()}`;++const tests = cases('pglite', async () => {++ const db = createDb({ pglite: true });++ await db.query(schema);++ return { db, done: () => db.close() };++});++++if (process.env.OV_TEST_PG_URL && process.env.OV_TEST_PG_DIRECT_URL) {++ const pgOpen = async () => {++ const owner = createDb({ url: process.env.OV_TEST_PG_DIRECT_URL });++ await owner.query('DROP TABLE IF EXISTS streams, event_outbox, idempotency_receipts CASCADE');++ await owner.query(schema);++ await owner.close();++ const db = createDb({ url: process.env.OV_TEST_PG_URL, max: 6 });++ return { db, done: () => db.close() };++ };++ tests.push(...cases('postgresql+pgbouncer', pgOpen));++ tests.push(['postgresql+pgbouncer: three relays on one table send each event exactly once', async () => {++ const { db, done } = await pgOpen();++ try {++ const { platform, events } = eventsClient();++ const slow = { prepare: events.prepare, publish: async (...a) => { await new Promise((r) => setTimeout(r, 30)); return events.publish(...a); } };++ const relays = [0, 1, 2].map(() => createPgOutbox(db, { events: slow, batchSize: 7 }));++ await db.tx(async (t) => { for (let i = 0; i < 60; i++) await relays[0].enqueue(t, { event_type: 'live.stream.started', actor, subject: { type: 'stream', id: String(i) } }); });++ const results = await Promise.all(relays.map(async (r) => { let sent = 0; for (;;) { const s = await r.flush(); sent += s.sent; if (!s.sent) break; } return sent; }));++ assert.equal(results.reduce((a, b) => a + b, 0), 60);++ assert.equal(platform.state.events.length, 60, 'no event sent twice');++ assert.ok(results.filter((n) => n > 0).length >= 2, `the work was shared: ${results}`);++ } finally { await done(); }++ }]);++} else console.log('pg outbox through PgBouncer: skipped (OV_TEST_PG_URL not set; scripts/test-services.sh up)');++++run(tests);@@@ -144,3 +144,33 @@ export interface Inbox {seen(consumer: string, eventId: string): boolean;}export declare function createInbox(db: SqliteDatabase, opts?: { table?: string; now?: () => number }): Inbox;++++/** The PostgreSQL outbox (ADR-035) on an openvibe-sdk/db handle; several relays may share one table. */++export interface PgOutbox {++ schema(): string;++ ensureSchema(): Promise<unknown>;++ /** Pass the transaction handle db.tx gives you: the event exists if and only if the change commits. */++ enqueue(t: import('./db').Tx, envelope: EventInput, opts?: { traceparent?: string }): Promise<EventEnvelope>;++ flush(): Promise<FlushStats>;++ start(): void;++ stop(): Promise<unknown>;++ kick(): void;++ pending(): Promise<number | null>;++ rejected(): Promise<number | null>;++ prune(olderThanMs?: number): Promise<number>;++}++export declare function createPgOutbox(db: import('./db').Db, opts: {++ events: Pick<EventsClient, 'publish' | 'prepare'>;++ table?: string; batchSize?: number; intervalMs?: number; leaseMs?: number; backoffMs?: number[];++ now?: () => number; onError?: (err: unknown, row?: unknown) => void;++}): PgOutbox;++export interface PgInbox {++ schema(): string;++ ensureSchema(): Promise<unknown>;++ once<T>(consumer: string, eventId: string, fn: (t: import('./db').Tx) => Promise<T>): Promise<{ duplicate: true } | { duplicate: false; result: T }>;++ seen(consumer: string, eventId: string): Promise<boolean>;++}++export declare function createPgInbox(db: import('./db').Db, opts?: { table?: string; now?: () => number }): PgInbox;++/** DDL for a service migration. */++export declare function outboxSchema(table?: string): string;++export declare function inboxSchema(table?: string): string;
@@ -3,6 +3,18 @@
@@ -3,6 +3,18 @@
All notable changes to `openvibe-sdk`. The package follows semver; while it is `0.x`, a minor
All notable changes to `openvibe-sdk`. The package follows semver; while it is `0.x`, a minor
release may change an API and says so here.
release may change an API and says so here.
+## 0.15.0 (2026-09-28)
+
+**The event outbox and inbox on PostgreSQL** (ADR-004 on ADR-035), in `openvibe-sdk/events`. The better-sqlite3 versions stay for services that have not moved yet.
+- **`createPgOutbox(db, { events })`:** takes an openvibe-sdk/db handle.
+ - `enqueue(t, envelope)` takes the transaction handle, so the event exists if and only if the change commits.
+ - The relay claims due rows with a lease (`FOR UPDATE SKIP LOCKED`), so any number of processes and hosts relay one table without sending an event twice. A relay that dies leaves its rows to the next one once the lease expires.
+ - Otherwise it behaves as before: backoff on transient failures, permanent refusals isolated per row, the seq kept, `pending`, `rejected` and `prune`.
+- **`createPgInbox(db)`:** `once(consumer, eventId, async (t) => …)` runs the handler once, in one transaction with its receipt, and a failed handler leaves no receipt.
+- **`outboxSchema()` and `inboxSchema()`:** the DDL to put in a service migration, since the runtime role cannot create tables.
+- **Quieter tests:** the slow-query log is off by default on PGlite, whose first query includes its WASM start.
+- **Tests:** `test/pg-outbox.test.js` runs on PGlite, and through PgBouncer with three relays sharing one table (60 events, none sent twice).
+
## 0.14.0 (2026-09-28)
## 0.14.0 (2026-09-28)
**The async data layer and Valkey modules** (ADR-035 and its amendment; roadmap WS-X2): every service moves to PostgreSQL 18 and Valkey, async-first.
**The async data layer and Valkey modules** (ADR-035 and its amendment; roadmap WS-X2): every service moves to PostgreSQL 18 and Valkey, async-first.
@@ -3,7 +3,7 @@
@@ -3,7 +3,7 @@
"stage": "alpha",
"stage": "alpha",
"code": true,
"code": true,
"wave": "W2",
"wave": "W2",
- "version": "0.14.0",
+ "version": "0.15.0",
"plan": "OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026), §3.2; roadmap §30",
"plan": "OpenVibe End-to-End Realignment & Implementation Plan, revision 3 (20 Sep 2026), §3.2; roadmap §30",
"deployed": false,
"deployed": false,
"deployment": "none: a package released as git tags; services and apps pin a tag",
"deployment": "none: a package released as git tags; services and apps pin a tag",
@@ -21,7 +21,8 @@
@@ -21,7 +21,8 @@
"mock platform for tests (openvibe-sdk/testing)",
"mock platform for tests (openvibe-sdk/testing)",
"ESM entries, TypeScript types and a browser bundle",
"ESM entries, TypeScript types and a browser bundle",
"db (ADR-035): the async PostgreSQL data layer (node-postgres through PgBouncer; PGlite in tests) with safe sql fragments, retrying transactions and savepoints, versioned expand/migrate/contract migrations, and importSqlite for the one-time SQLite move with verification",
"db (ADR-035): the async PostgreSQL data layer (node-postgres through PgBouncer; PGlite in tests) with safe sql fragments, retrying transactions and savepoints, versioned expand/migrate/contract migrations, and importSqlite for the one-time SQLite move with verification",
- "valkey, cache, queue, pubsub and a shared limits store on Valkey 9, each with an in-process twin for tests"
+ "valkey, cache, queue, pubsub and a shared limits store on Valkey 9, each with an in-process twin for tests",
+ "events: the PostgreSQL outbox (lease-claimed, several relays per table) and inbox (handler in its transaction)"
],
],
"updated": "2026-09-28"
"updated": "2026-09-28"
}
}
@@ -1,4 +1,4 @@
@@ -1,4 +1,4 @@
-// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.14.0, MIT). Do not edit.
+// Generated by scripts/browser-bundle.js from browser.js (openvibe-sdk 0.15.0, MIT). Do not edit.
// One self-contained ES module: core at the top level plus the browser-safe subpaths as namespaces
// One self-contained ES module: core at the top level plus the browser-safe subpaths as namespaces
// (auth, registry, modules, realtime, media, community, search, chat, ai, jobs, tools, projects). No server-only code: no service tokens, JWT verification, events, identity.
// (auth, registry, modules, realtime, media, community, search, chat, ai, jobs, tools, projects). No server-only code: no service tokens, JWT verification, events, identity.
/* eslint-disable */
/* eslint-disable */
@@ -40,7 +40,7 @@ const { paginate, offsetPager } = __load(7);
@@ -40,7 +40,7 @@ const { paginate, offsetPager } = __load(7);
const semver = __load(6);
const semver = __load(6);
/** The package version (test/version.test.js keeps it equal to package.json). */
/** The package version (test/version.test.js keeps it equal to package.json). */
-const SDK_VERSION = '0.14.0';
+const SDK_VERSION = '0.15.0';
module.exports = {
module.exports = {
createClient,
createClient,
@@ -3,6 +3,7 @@ import cjs from '../src/events.js';
@@ -3,6 +3,7 @@ import cjs from '../src/events.js';
export const {
export const {
createEventsClient, verifyDelivery, signDelivery, parseDelivery, createOutbox, createInbox,
createEventsClient, verifyDelivery, signDelivery, parseDelivery, createOutbox, createInbox,
- projectKey, appSource, createAppEvents, verifyDeliveryV2, signDeliveryV2, signDeliveryHeaders,
+ createPgOutbox, createPgInbox, outboxSchema, inboxSchema, projectKey, appSource,
+ createAppEvents, verifyDeliveryV2, signDeliveryV2, signDeliveryHeaders,
} = cjs;
} = cjs;
export default cjs;
export default cjs;
@@ -1,12 +1,12 @@
@@ -1,12 +1,12 @@
{
{
"name": "openvibe-sdk",
"name": "openvibe-sdk",
- "version": "0.14.0",
+ "version": "0.15.0",
"lockfileVersion": 3,
"lockfileVersion": 3,
"requires": true,
"requires": true,
"packages": {
"packages": {
"": {
"": {
"name": "openvibe-sdk",
"name": "openvibe-sdk",
- "version": "0.14.0",
+ "version": "0.15.0",
"license": "MIT",
"license": "MIT",
"devDependencies": {
"devDependencies": {
"@electric-sql/pglite": "^0.5.8",
"@electric-sql/pglite": "^0.5.8",
@@ -1,6 +1,6 @@
@@ -1,6 +1,6 @@
{
{
"name": "openvibe-sdk",
"name": "openvibe-sdk",
- "version": "0.14.0",
+ "version": "0.15.0",
"description": "Supported browser and server clients for the OpenVibe platform: auth (incl. developer apps), registry discovery, identity, user modules, events, realtime, media, community, search, chat, AI runs, Tools (registry, run API and jobs), VIP memberships and developer projects, the async PostgreSQL data layer (db) and the Valkey cache, limits store, queue and pub/sub, with deadlines, safe retries, problem errors and trace propagation.",
"description": "Supported browser and server clients for the OpenVibe platform: auth (incl. developer apps), registry discovery, identity, user modules, events, realtime, media, community, search, chat, AI runs, Tools (registry, run API and jobs), VIP memberships and developer projects, the async PostgreSQL data layer (db) and the Valkey cache, limits store, queue and pub/sub, with deadlines, safe retries, problem errors and trace propagation.",
"main": "index.js",
"main": "index.js",
"types": "types/index.d.ts",
"types": "types/index.d.ts",
@@ -8,7 +8,7 @@ const { paginate, offsetPager } = require('./paginate');
@@ -8,7 +8,7 @@ const { paginate, offsetPager } = require('./paginate');
const semver = require('./semver');
const semver = require('./semver');
/** The package version (test/version.test.js keeps it equal to package.json). */
/** The package version (test/version.test.js keeps it equal to package.json). */
-const SDK_VERSION = '0.14.0';
+const SDK_VERSION = '0.15.0';
module.exports = {
module.exports = {
createClient,
createClient,
@@ -85,7 +85,7 @@ function queryApi(run, self) {
@@ -85,7 +85,7 @@ function queryApi(run, self) {
function createDb(o = {}) {
function createDb(o = {}) {
const log = o.log || console;
const log = o.log || console;
const service = o.service || process.env.OV_SERVICE || 'service';
const service = o.service || process.env.OV_SERVICE || 'service';
- const slowMs = o.slowMs == null ? 500 : o.slowMs;
+ const slowMs = o.slowMs == null ? (o.pglite ? Infinity : 500) : o.slowMs; // PGlite's first query includes its WASM start
const stats = { queries: 0, errors: 0, slow: 0, retries: 0, tx: 0 };
const stats = { queries: 0, errors: 0, slow: 0, retries: 0, tx: 0 };
let hist = null;
let hist = null;
if (o.registry && typeof o.registry.histogram === 'function') {
if (o.registry && typeof o.registry.histogram === 'function') {
@@ -355,9 +355,10 @@ function createAppEvents(client, { projectId, appId, onBehalfOf, baseUrl } = {})
@@ -355,9 +355,10 @@ function createAppEvents(client, { projectId, appId, onBehalfOf, baseUrl } = {})
};
};
}
}
-const { createOutbox, createInbox } = require('./outbox');
+const { createOutbox, createInbox, createPgOutbox, createPgInbox, outboxSchema, inboxSchema } = require('./outbox');
module.exports = {
module.exports = {
createEventsClient, verifyDelivery, signDelivery, parseDelivery, createOutbox, createInbox,
createEventsClient, verifyDelivery, signDelivery, parseDelivery, createOutbox, createInbox,
+ createPgOutbox, createPgInbox, outboxSchema, inboxSchema,
projectKey, appSource, createAppEvents, verifyDeliveryV2, signDeliveryV2, signDeliveryHeaders,
projectKey, appSource, createAppEvents, verifyDeliveryV2, signDeliveryV2, signDeliveryHeaders,
};
};
@@ -1,7 +1,8 @@
@@ -1,7 +1,8 @@
'use strict';
'use strict';
/**
/**
- * openvibe-sdk/events (server): transactional outbox and inbox on the service's own better-sqlite3
- * database (ADR-004). No dependency: the caller passes its database handle.
+ * openvibe-sdk/events (server): transactional outbox and inbox (ADR-004), on the service's own database:
+ * createOutbox/createInbox on better-sqlite3 (services not yet moved), createPgOutbox/createPgInbox on an
+ * openvibe-sdk/db handle (PostgreSQL, ADR-035; below). No dependency: the caller passes its database handle.
*
*
* const events = createEventsClient(client, { source: 'live' });
* const events = createEventsClient(client, { source: 'live' });
* const outbox = createOutbox(db, { events });
* const outbox = createOutbox(db, { events });
@@ -207,4 +208,168 @@ function createInbox(db, { table = 'idempotency_receipts', now = () => Date.now(
@@ -207,4 +208,168 @@ function createInbox(db, { table = 'idempotency_receipts', now = () => Date.now(
return { ensureSchema, once, seen };
return { ensureSchema, once, seen };
}
}
-module.exports = { createOutbox, createInbox, isPermanent };
+// ── PostgreSQL (ADR-035): the same outbox and inbox on an openvibe-sdk/db handle ─────────────────
+
+/** The outbox table's DDL, for a service migration (the runtime role cannot create tables). */
+function outboxSchema(table = 'event_outbox') {
+ if (!TABLE_RE.test(table)) throw new TypeError('bad outbox table name');
+ return `CREATE TABLE IF NOT EXISTS ${table} (
+ id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+ event_id text NOT NULL UNIQUE,
+ envelope jsonb NOT NULL,
+ traceparent text,
+ created_at bigint NOT NULL,
+ attempts integer NOT NULL DEFAULT 0,
+ next_attempt_at bigint NOT NULL DEFAULT 0,
+ sent_at bigint,
+ seq bigint,
+ rejected_at bigint,
+ last_error text
+);
+CREATE INDEX IF NOT EXISTS ${table}_due ON ${table} (next_attempt_at, id) WHERE sent_at IS NULL AND rejected_at IS NULL;
+CREATE INDEX IF NOT EXISTS ${table}_sent ON ${table} (sent_at) WHERE sent_at IS NOT NULL;`;
+}
+
+/** The inbox table's DDL, for a service migration. */
+function inboxSchema(table = 'idempotency_receipts') {
+ if (!TABLE_RE.test(table)) throw new TypeError('bad inbox table name');
+ return `CREATE TABLE IF NOT EXISTS ${table} (
+ consumer text NOT NULL,
+ event_id text NOT NULL,
+ processed_at bigint NOT NULL,
+ PRIMARY KEY (consumer, event_id)
+);`;
+}
+
+/**
+ * createPgOutbox(db, { events }) on an openvibe-sdk/db handle. enqueue takes the caller's transaction handle,
+ * so the event exists if and only if the change commits:
+ *
+ * await db.tx(async (t) => {
+ * await t.exec(sql`UPDATE streams SET is_live = true WHERE id = ${id}`);
+ * await outbox.enqueue(t, { event_type: 'live.stream.started', actor, subject, payload });
+ * });
+ * outbox.kick();
+ *
+ * The relay claims due rows with a short lease (FOR UPDATE SKIP LOCKED), so any number of processes and hosts
+ * relay one table without double-sending; a relay that dies mid-batch leaves its rows to the next after the
+ * lease. Delivery stays at least once (Events answers a repeated event_id as a duplicate).
+ */
+function createPgOutbox(db, {
+ events, table = 'event_outbox', batchSize = 50, intervalMs = 1000, leaseMs = 60000,
+ backoffMs = [1000, 5000, 30000, 120000, 600000], now = () => Date.now(), onError = null,
+} = {}) {
+ if (!db || typeof db.query !== 'function' || typeof db.tx !== 'function') throw new TypeError('an openvibe-sdk/db handle is required');
+ if (!events || typeof events.publish !== 'function' || typeof events.prepare !== 'function') {
+ throw new TypeError('events must be an openvibe-sdk events client (createEventsClient)');
+ }
+ if (!TABLE_RE.test(table)) throw new TypeError('bad outbox table name');
+ let timer = null; let running = false; let flushing = null;
+ const backoff = (attempts) => backoffMs[Math.min(attempts, backoffMs.length - 1)];
+
+ async function enqueue(t, envelope, { traceparent } = {}) {
+ if (!t || typeof t.query !== 'function' || typeof t.tx !== 'function') throw new TypeError('outbox.enqueue(t, …): pass the transaction handle db.tx gives you');
+ const m = traceparent && TRACEPARENT_RE.exec(traceparent);
+ const env = events.prepare(envelope, { traceId: m ? m[1] : undefined, now: now() });
+ await t.query(`INSERT INTO ${table} (event_id, envelope, traceparent, created_at) VALUES ($1, $2, $3, $4)`, [env.event_id, JSON.stringify(env), m ? traceparent : null, now()]);
+ return env;
+ }
+
+ async function claim() {
+ const t = now();
+ return db.many(`UPDATE ${table} SET next_attempt_at = $1
+ WHERE id IN (SELECT id FROM ${table} WHERE sent_at IS NULL AND rejected_at IS NULL AND next_attempt_at <= $2
+ ORDER BY id LIMIT $3 FOR UPDATE SKIP LOCKED)
+ RETURNING id, event_id, envelope, traceparent, attempts`, [t + leaseMs, t, batchSize]);
+ }
+
+ async function markFailure(row, err) {
+ const message = String((err && err.message) || err).slice(0, 500);
+ if (isPermanent(err)) await db.query(`UPDATE ${table} SET rejected_at = $1, attempts = attempts + 1, last_error = $2 WHERE id = $3`, [now(), message, row.id]);
+ else await db.query(`UPDATE ${table} SET attempts = attempts + 1, next_attempt_at = $1, last_error = $2 WHERE id = $3`, [now() + backoff(row.attempts), message, row.id]);
+ if (onError) { try { onError(err, row); } catch { /* reporting must not break the relay */ } }
+ }
+
+ async function publishRows(rows) {
+ const stats = { sent: 0, failed: 0, rejected: 0 };
+ const envelopes = rows.map((r) => (typeof r.envelope === 'string' ? JSON.parse(r.envelope) : r.envelope));
+ const traceparent = rows[0].traceparent || undefined;
+ try {
+ const out = rows.length === 1
+ ? { results: [await events.publish(envelopes[0], { traceparent })] }
+ : await events.publish(envelopes, { traceparent });
+ const byId = new Map(((out && out.results) || []).map((r) => [r.event_id, r]));
+ await db.query(`UPDATE ${table} AS o SET sent_at = $1, seq = v.seq, attempts = o.attempts + 1, last_error = NULL
+ FROM unnest($2::bigint[], $3::bigint[]) AS v(id, seq) WHERE o.id = v.id`,
+ [now(), rows.map((r) => r.id), rows.map((r) => (byId.get(r.event_id) && byId.get(r.event_id).seq) ?? null)]);
+ stats.sent += rows.length;
+ } catch (err) {
+ if (rows.length > 1 && isPermanent(err)) {
+ for (const row of rows) { const s = await publishRows([row]); stats.sent += s.sent; stats.failed += s.failed; stats.rejected += s.rejected; }
+ } else {
+ for (const row of rows) await markFailure(row, err);
+ if (isPermanent(err)) stats.rejected += rows.length; else stats.failed += rows.length;
+ }
+ }
+ return stats;
+ }
+
+ async function doFlush() {
+ const total = { sent: 0, failed: 0, rejected: 0 };
+ for (;;) {
+ const rows = await claim();
+ if (!rows.length) break;
+ const s = await publishRows(rows);
+ total.sent += s.sent; total.failed += s.failed; total.rejected += s.rejected;
+ if (s.failed || rows.length < batchSize) break;
+ }
+ return total;
+ }
+ function flush() { if (!flushing) flushing = doFlush().finally(() => { flushing = null; }); return flushing; }
+ function schedule(ms) {
+ if (!running) return;
+ clearTimeout(timer);
+ timer = setTimeout(async () => {
+ timer = null;
+ try { await flush(); } catch (err) { if (onError) { try { onError(err); } catch { /* ignore */ } } }
+ schedule(intervalMs);
+ }, ms);
+ if (timer.unref) timer.unref();
+ }
+ return {
+ schema: () => outboxSchema(table),
+ /** Create the table where the handle may (tests, PGlite); services put schema() in a migration. */
+ ensureSchema: () => db.query(outboxSchema(table)),
+ enqueue, flush,
+ start() { if (!running) { running = true; schedule(0); } },
+ stop() { running = false; clearTimeout(timer); timer = null; return flushing || Promise.resolve(); },
+ kick() { schedule(0); },
+ pending: () => db.value(`SELECT count(*) FROM ${table} WHERE sent_at IS NULL AND rejected_at IS NULL`),
+ rejected: () => db.value(`SELECT count(*) FROM ${table} WHERE rejected_at IS NOT NULL`),
+ prune: (olderThanMs = 7 * 24 * 60 * 60 * 1000) => db.exec(`DELETE FROM ${table} WHERE sent_at IS NOT NULL AND sent_at < $1`, [now() - olderThanMs]),
+ };
+}
+
+/**
+ * createPgInbox(db): run fn once per (consumer, eventId), in one transaction with its receipt.
+ * const r = await inbox.once('network', event.event_id, async (t) => { … writes with t … }); // { duplicate, result }
+ */
+function createPgInbox(db, { table = 'idempotency_receipts', now = () => Date.now() } = {}) {
+ if (!db || typeof db.tx !== 'function') throw new TypeError('an openvibe-sdk/db handle is required');
+ if (!TABLE_RE.test(table)) throw new TypeError('bad inbox table name');
+ return {
+ schema: () => inboxSchema(table),
+ ensureSchema: () => db.query(inboxSchema(table)),
+ async once(consumer, eventId, fn) {
+ if (!consumer || !eventId) throw new TypeError('consumer and eventId are required');
+ return db.tx(async (t) => {
+ const n = await t.exec(`INSERT INTO ${table} (consumer, event_id, processed_at) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING`, [String(consumer), String(eventId), now()]);
+ if (n === 0) return { duplicate: true };
+ return { duplicate: false, result: await fn(t) };
+ });
+ },
+ seen: async (consumer, eventId) => Boolean(await db.maybe(`SELECT 1 FROM ${table} WHERE consumer = $1 AND event_id = $2`, [String(consumer), String(eventId)])),
+ };
+}
+
+module.exports = { createOutbox, createInbox, createPgOutbox, createPgInbox, outboxSchema, inboxSchema, isPermanent };
new file mode 100644
new file mode 100644
@@ -0,0 +1,123 @@
@@ -0,0 +1,123 @@
+'use strict';
+/**
+ * The PostgreSQL outbox and inbox (ADR-004 on ADR-035): enqueue needs the transaction handle, so a rolled-back
+ * change leaves no event; the relay publishes once, backs off on a transient failure, rejects only a permanently
+ * refused row, and several relays on one table never send an event twice (FOR UPDATE SKIP LOCKED with a lease);
+ * the inbox runs a handler once per event in the handler's transaction. PGlite always; PostgreSQL through
+ * PgBouncer when OV_TEST_PG_URL is set.
+ */
+const assert = require('node:assert/strict');
+const { run } = require('./helpers');
+const { createClient } = require('../src/core');
+const { createServiceTokenClient } = require('../src/auth');
+const { createEventsClient, createPgOutbox, createPgInbox, outboxSchema, inboxSchema } = require('../src/events');
+const { createMockPlatform } = require('../src/testing');
+const { createDb, sql } = require('../src/db');
+
+const actor = { type: 'service', id: 'live' };
+const subject = { type: 'stream', id: '12', revision: 1 };
+
+function eventsClient() {
+ const platform = createMockPlatform({ clients: { live: { secret: 'live-secret', grants: [['events.event.publish', 'openvibe.events']] } } });
+ const tokens = createServiceTokenClient({ clientId: 'live', clientSecret: 'live-secret', fetch: platform.fetch });
+ const client = createClient({ fetch: platform.fetch, tokenProvider: tokens, retries: 0 });
+ return { platform, events: createEventsClient(client, { source: 'live' }) };
+}
+
+function cases(label, open) {
+ return [
+ [`${label}: enqueue needs the transaction; a rollback leaves no event; a commit publishes once`, async () => {
+ const { db, done } = await open();
+ try {
+ const { platform, events } = eventsClient();
+ const outbox = createPgOutbox(db, { events });
+ await assert.rejects(outbox.enqueue(null, { event_type: 'live.stream.started', actor, subject }), /transaction handle/);
+ await assert.rejects(db.tx(async (t) => {
+ await t.exec(sql`INSERT INTO streams (id, is_live) VALUES (1, true)`);
+ await outbox.enqueue(t, { event_type: 'live.stream.started', actor, subject });
+ throw new Error('boom');
+ }), /boom/);
+ assert.equal(await outbox.pending(), 0);
+ const env = await db.tx(async (t) => {
+ await t.exec(sql`INSERT INTO streams (id, is_live) VALUES (2, true)`);
+ return outbox.enqueue(t, { event_type: 'live.stream.started', actor, subject, payload: { stream_id: 2 } }, { traceparent: '00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01' });
+ });
+ assert.equal(env.trace_id, '0af7651916cd43dd8448eb211c80319c');
+ assert.deepEqual(await outbox.flush(), { sent: 1, failed: 0, rejected: 0 });
+ assert.equal(await outbox.pending(), 0);
+ assert.equal(platform.state.events.length, 1);
+ assert.equal(platform.state.events[0].event.event_id, env.event_id);
+ await outbox.flush();
+ assert.equal(platform.state.events.length, 1, 'nothing republished');
+ assert.ok((await db.value(sql`SELECT seq FROM event_outbox WHERE event_id = ${env.event_id}`)) >= 1, 'the seq Events assigned is kept');
+ } finally { await done(); }
+ }],
+ [`${label}: transient failures back off; a permanent refusal rejects only its row`, async () => {
+ const { db, done } = await open();
+ try {
+ const { platform, events } = eventsClient();
+ let t = 1_000_000; let fail = true;
+ const flaky = { prepare: events.prepare, publish: (...a) => (fail ? Promise.reject(Object.assign(new Error('down'), { status: 503 })) : events.publish(...a)) };
+ const outbox = createPgOutbox(db, { events: flaky, now: () => t });
+ await db.tx(async (tx) => { for (let i = 0; i < 3; i++) await outbox.enqueue(tx, { event_type: 'live.stream.started', actor, subject: { type: 'stream', id: String(i) } }); });
+ assert.deepEqual(await outbox.flush(), { sent: 0, failed: 3, rejected: 0 });
+ fail = false;
+ assert.deepEqual(await outbox.flush(), { sent: 0, failed: 0, rejected: 0 }, 'not due yet');
+ t += 1000;
+ assert.deepEqual(await outbox.flush(), { sent: 3, failed: 0, rejected: 0 });
+ assert.equal(platform.state.events.length, 3);
+ const picky = { prepare: events.prepare, publish: (env) => (Array.isArray(env) || env.subject.id === 'bad' ? Promise.reject(Object.assign(new Error('invalid'), { status: 422 })) : events.publish(env)) };
+ const o2 = createPgOutbox(db, { events: picky, now: () => t });
+ await db.tx(async (tx) => { await o2.enqueue(tx, { event_type: 'live.stream.started', actor, subject: { type: 'stream', id: 'bad' } }); await o2.enqueue(tx, { event_type: 'live.stream.started', actor, subject: { type: 'stream', id: 'ok' } }); });
+ assert.deepEqual(await o2.flush(), { sent: 1, failed: 0, rejected: 1 });
+ assert.equal(await o2.rejected(), 1);
+ } finally { await done(); }
+ }],
+ [`${label}: the inbox runs a handler once, inside its transaction`, async () => {
+ const { db, done } = await open();
+ try {
+ const inbox = createPgInbox(db);
+ let runs = 0;
+ const a = await inbox.once('network', 'evt_1', async (t) => { runs++; await t.exec(sql`INSERT INTO streams (id, is_live) VALUES (7, false)`); return 'ok'; });
+ const b = await inbox.once('network', 'evt_1', async () => { runs++; });
+ assert.deepEqual([a, b, runs], [{ duplicate: false, result: 'ok' }, { duplicate: true }, 1]);
+ await assert.rejects(inbox.once('network', 'evt_2', async () => { throw new Error('fail'); }), /fail/);
+ assert.equal(await inbox.seen('network', 'evt_2'), false, 'a failed handler leaves no receipt: it runs again next time');
+ } finally { await done(); }
+ }],
+ ];
+}
+
+const schema = `CREATE TABLE streams (id bigint PRIMARY KEY, is_live boolean NOT NULL);\n${outboxSchema()}\n${inboxSchema()}`;
+const tests = cases('pglite', async () => {
+ const db = createDb({ pglite: true });
+ await db.query(schema);
+ return { db, done: () => db.close() };
+});
+
+if (process.env.OV_TEST_PG_URL && process.env.OV_TEST_PG_DIRECT_URL) {
+ const pgOpen = async () => {
+ const owner = createDb({ url: process.env.OV_TEST_PG_DIRECT_URL });
+ await owner.query('DROP TABLE IF EXISTS streams, event_outbox, idempotency_receipts CASCADE');
+ await owner.query(schema);
+ await owner.close();
+ const db = createDb({ url: process.env.OV_TEST_PG_URL, max: 6 });
+ return { db, done: () => db.close() };
+ };
+ tests.push(...cases('postgresql+pgbouncer', pgOpen));
+ tests.push(['postgresql+pgbouncer: three relays on one table send each event exactly once', async () => {
+ const { db, done } = await pgOpen();
+ try {
+ const { platform, events } = eventsClient();
+ const slow = { prepare: events.prepare, publish: async (...a) => { await new Promise((r) => setTimeout(r, 30)); return events.publish(...a); } };
+ const relays = [0, 1, 2].map(() => createPgOutbox(db, { events: slow, batchSize: 7 }));
+ await db.tx(async (t) => { for (let i = 0; i < 60; i++) await relays[0].enqueue(t, { event_type: 'live.stream.started', actor, subject: { type: 'stream', id: String(i) } }); });
+ const results = await Promise.all(relays.map(async (r) => { let sent = 0; for (;;) { const s = await r.flush(); sent += s.sent; if (!s.sent) break; } return sent; }));
+ assert.equal(results.reduce((a, b) => a + b, 0), 60);
+ assert.equal(platform.state.events.length, 60, 'no event sent twice');
+ assert.ok(results.filter((n) => n > 0).length >= 2, `the work was shared: ${results}`);
+ } finally { await done(); }
+ }]);
+} else console.log('pg outbox through PgBouncer: skipped (OV_TEST_PG_URL not set; scripts/test-services.sh up)');
+
+run(tests);
@@ -144,3 +144,33 @@ export interface Inbox {
@@ -144,3 +144,33 @@ export interface Inbox {
seen(consumer: string, eventId: string): boolean;
seen(consumer: string, eventId: string): boolean;
}
}
export declare function createInbox(db: SqliteDatabase, opts?: { table?: string; now?: () => number }): Inbox;
export declare function createInbox(db: SqliteDatabase, opts?: { table?: string; now?: () => number }): Inbox;
+
+/** The PostgreSQL outbox (ADR-035) on an openvibe-sdk/db handle; several relays may share one table. */
+export interface PgOutbox {
+ schema(): string;
+ ensureSchema(): Promise<unknown>;
+ /** Pass the transaction handle db.tx gives you: the event exists if and only if the change commits. */
+ enqueue(t: import('./db').Tx, envelope: EventInput, opts?: { traceparent?: string }): Promise<EventEnvelope>;
+ flush(): Promise<FlushStats>;
+ start(): void;
+ stop(): Promise<unknown>;
+ kick(): void;
+ pending(): Promise<number | null>;
+ rejected(): Promise<number | null>;
+ prune(olderThanMs?: number): Promise<number>;
+}
+export declare function createPgOutbox(db: import('./db').Db, opts: {
+ events: Pick<EventsClient, 'publish' | 'prepare'>;
+ table?: string; batchSize?: number; intervalMs?: number; leaseMs?: number; backoffMs?: number[];
+ now?: () => number; onError?: (err: unknown, row?: unknown) => void;
+}): PgOutbox;
+export interface PgInbox {
+ schema(): string;
+ ensureSchema(): Promise<unknown>;
+ once<T>(consumer: string, eventId: string, fn: (t: import('./db').Tx) => Promise<T>): Promise<{ duplicate: true } | { duplicate: false; result: T }>;
+ seen(consumer: string, eventId: string): Promise<boolean>;
+}
+export declare function createPgInbox(db: import('./db').Db, opts?: { table?: string; now?: () => number }): PgInbox;
+/** DDL for a service migration. */
+export declare function outboxSchema(table?: string): string;
+export declare function inboxSchema(table?: string): string;