openvibers/openvibe.events Commits
One inbox (plan T7, ADR-042 decision 9): lib/client.js re-exports openvibe-sdk's PostgreSQL outbox and inbox (createPgOutbox, createPgInbox, outboxSchema, inboxSchema) and its own SQLite copies are gone, with better-sqlite3; the README shows the async, transaction-handle usage. test/outbox-inbox.test.js runs producer and consumer on embedded PostgreSQL: a rolled-back enqueue publishes nothing, a resend is a duplicate, a poisoned row is isolated, a down Events is retried after backoff, and a consumer crash before or after commit plus redelivery and replay yields exactly one effect. npm test and test:pg 18/18.
Changes
@@@ -169,27 +169,28 @@ Quotas recorded in Network (`dev_quotas`) are not read yet: that needs `network.const events = require('openvibe-events');const tokenClient = contracts.serviceAuth.createTokenClient({ tokenUrl: `${NETWORK}/oauth/token`, clientId, clientSecret, audience: 'openvibe.events' });--// Producer: transactional outbox in the service's own better-sqlite3 database (a service on PostgreSQL uses--// openvibe-sdk/events createPgOutbox / createPgInbox instead: same envelope, receipts and relay, async)--const outbox = events.createOutbox(db, { publisher: events.createPublisher({ eventsUrl: 'http://127.0.0.1:4300', tokenClient }) });--outbox.ensureSchema();--db.transaction(() => {-- db.prepare('UPDATE vods SET status = ? WHERE id = ?').run('ready', id);-- outbox.enqueue({ event_type: 'media.vod.ready', source: 'media', actor, subject: { type: 'vod', id, revision }, payload });--})(); // the event exists if and only if the change committed++// Producer: transactional outbox in the service's own PostgreSQL (openvibe-sdk/db; the table from events.outboxSchema()++// goes in one of the service's migrations)++const outbox = events.createPgOutbox(db, { events: events.createPublisher({ eventsUrl: 'http://127.0.0.1:4300', tokenClient }) });++await db.tx(async (t) => {++ await t.query('UPDATE vods SET status = $1 WHERE id = $2', ['ready', id]);++ await outbox.enqueue(t, { event_type: 'media.vod.ready', source: 'media', actor, subject: { type: 'vod', id, revision }, payload });++}); // the event exists if and only if the change committedoutbox.start(); // relay: publishes pending rows, marks them sent, backs off on failure// Consumer: signed webhook (v2: signature plus a 300 s replay window) + exactly-once effects in the consumer's own database--const inbox = events.createInbox(db);--inbox.ensureSchema();--app.post('/internal/events', express.json({ verify: (req, _res, buf) => { req.rawBody = buf; } }), (req, res) => {++const inbox = events.createPgInbox(db); // the table from events.inboxSchema()++app.post('/internal/events', express.json({ verify: (req, _res, buf) => { req.rawBody = buf; } }), async (req, res) => {if (!events.verifyDeliveryV2(req.rawBody, req.headers, SECRET)) return res.sendStatus(401);-- inbox.once('live', req.body.event.event_id, () => { /* synchronous db writes */ });++ await inbox.once('live', req.body.event.event_id, async (t) => { /* writes on t */ });res.sendStatus(204);});```--`enqueue()` refuses to run outside a transaction. `once()` records the receipt and runs the handler in one SQLite transaction, so a crash before commit leaves nothing behind and the redelivery runs it again, and a crash after commit makes the redelivery a no-op.++`enqueue(t, …)` takes the transaction handle, so it cannot run outside one. `once()` records the receipt and runs the handler++in one transaction, so a crash before commit leaves nothing behind and the redelivery runs it again, and a crash after++commit makes the redelivery a no-op. The outbox and inbox are openvibe-sdk's PostgreSQL kits (ADR-042): this package++re-exports them and keeps no SQLite copies.## Realtime (SSE)@@@ -4,23 +4,26 @@** const events = require('openvibe-events');*-- * // Producer: transactional outbox (better-sqlite3), relayed to OpenVibe.Events++ * // Producer: transactional outbox in the service's own PostgreSQL (openvibe-sdk/db), relayed to OpenVibe.Events* const publisher = events.createPublisher({ eventsUrl: 'http://127.0.0.1:4300', tokenClient });-- * const outbox = events.createOutbox(db, { publisher });-- * outbox.ensureSchema();-- * db.transaction(() => {-- * db.prepare('UPDATE vods SET status = ? WHERE id = ?').run('ready', id);-- * outbox.enqueue({ event_type: 'media.vod.ready', source: 'media', actor, subject, payload });-- * })();++ * const outbox = events.createPgOutbox(db, { events: publisher }); // table from events.outboxSchema() in a migration++ * await db.tx(async (t) => {++ * await t.query('UPDATE vods SET status = $1 WHERE id = $2', ['ready', id]);++ * await outbox.enqueue(t, { event_type: 'media.vod.ready', source: 'media', actor, subject, payload });++ * });* outbox.start();** // Consumer: signed webhook (v2: signature + 5-minute replay window) + exactly-once effects-- * app.post('/internal/events', express.json({ verify: (req, _res, buf) => { req.rawBody = buf; } }), (req, res) => {++ * const inbox = events.createPgInbox(db); // table from events.inboxSchema()++ * app.post('/internal/events', express.json({ verify: (req, _res, buf) => { req.rawBody = buf; } }), async (req, res) => {* if (!events.verifyDeliveryV2(req.rawBody, req.headers, SECRET)) return res.sendStatus(401);-- * inbox.once('live', req.body.event.event_id, () => { ...db writes... });++ * await inbox.once('live', req.body.event.event_id, async (t) => { ...writes on t... });* res.sendStatus(204);* });*++ * The outbox and inbox are openvibe-sdk's PostgreSQL kits (ADR-042 decision 9): this package no longer carries its++ * own SQLite copies.++ ** `tokenClient` is anything with `authHeaders()` (openvibe-contracts serviceAuth.createTokenClient* with audience 'openvibe.events').*/@@@ -181,194 +184,8 @@ function createPublisher({ eventsUrl, tokenClient, fetchImpl = globalThis.fetch,return { publish, prepare: prepareEnvelope };}--// ── Transactional outbox ───────────────────────────────────----/**-- * createOutbox(db, { publisher, table = 'event_outbox', batchSize = 50, intervalMs = 1000, now })-- *-- * ensureSchema() create the outbox table (idempotent)-- * enqueue(envelope) INSIDE the caller's db transaction: the event exists iff the change commits-- * flush() publish due rows once (batch; a rejected batch is retried row by row so one-- * bad row cannot block the rest); resolves { sent, failed, rejected }-- * start() / stop() relay loop (setTimeout chain, never overlapping)-- * pending() count of rows not yet accepted by Events-- *-- * At-least-once: a crash between publish and mark-sent republishes the same event_id, which-- * OpenVibe.Events answers as a duplicate.-- */--function createOutbox(db, { publisher, table = 'event_outbox', batchSize = 50, intervalMs = 1000, backoffMs = [1000, 5000, 30000, 120000, 600000], now = () => Date.now(), onError = null, allowOutsideTransaction = false } = {}) {-- if (!db || typeof db.prepare !== 'function') throw new TypeError('a better-sqlite3 database is required');-- if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(table)) throw new TypeError('bad outbox table name');-- let stmts = null;-- let timer = null;-- let running = false;-- let flushing = null;---- function ensureSchema() {-- db.exec(`CREATE TABLE IF NOT EXISTS ${table} (-- id INTEGER PRIMARY KEY AUTOINCREMENT,-- event_id TEXT NOT NULL UNIQUE,-- envelope TEXT NOT NULL,-- traceparent TEXT,-- created_at INTEGER NOT NULL,-- attempts INTEGER NOT NULL DEFAULT 0,-- next_attempt_at INTEGER NOT NULL DEFAULT 0,-- sent_at INTEGER,-- seq INTEGER,-- rejected_at INTEGER,-- last_error TEXT-- );-- CREATE INDEX IF NOT EXISTS idx_${table}_due ON ${table}(sent_at, rejected_at, next_attempt_at);`);-- stmts = null;-- }---- function q() {-- if (!stmts) {-- stmts = {-- insert: db.prepare(`INSERT INTO ${table} (event_id, envelope, traceparent, created_at, next_attempt_at) VALUES (?, ?, ?, ?, 0)`),-- due: db.prepare(`SELECT * FROM ${table} WHERE sent_at IS NULL AND rejected_at IS NULL AND next_attempt_at <= ? ORDER BY id LIMIT ?`),-- sent: db.prepare(`UPDATE ${table} SET sent_at = ?, seq = ?, attempts = attempts + 1, last_error = NULL WHERE id = ?`),-- failed: db.prepare(`UPDATE ${table} SET attempts = attempts + 1, next_attempt_at = ?, last_error = ? WHERE id = ?`),-- rejected: db.prepare(`UPDATE ${table} SET rejected_at = ?, attempts = attempts + 1, last_error = ? WHERE id = ?`),-- pending: db.prepare(`SELECT COUNT(*) AS n FROM ${table} WHERE sent_at IS NULL AND rejected_at IS NULL`),-- prune: db.prepare(`DELETE FROM ${table} WHERE sent_at IS NOT NULL AND sent_at < ?`),-- };-- }-- return stmts;-- }---- function enqueue(envelope, { traceparent } = {}) {-- if (!allowOutsideTransaction && !db.inTransaction) {-- throw new Error('outbox.enqueue() must run inside the transaction that makes the change');-- }-- const env = prepareEnvelope(envelope, { traceparent, now: now() });-- q().insert.run(env.event_id, JSON.stringify(env), traceparent || null, now());-- return env;-- }---- function backoff(attempts) {-- return backoffMs[Math.min(attempts, backoffMs.length - 1)];-- }---- function markFailure(row, err) {-- const message = String((err && err.message) || err).slice(0, 500);-- if (err && err.permanent) q().rejected.run(now(), message, row.id);-- else q().failed.run(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 => JSON.parse(r.envelope));-- try {-- const out = rows.length === 1-- ? { results: [await publisher.publish(envelopes[0], { traceparent: rows[0].traceparent || undefined })] }-- : await publisher.publish(envelopes, { traceparent: rows[0].traceparent || undefined });-- const byId = new Map((out.results || []).map(r => [r.event_id, r]));-- db.transaction(() => { for (const row of rows) q().sent.run(now(), byId.get(row.event_id)?.seq ?? null, row.id); })();-- stats.sent += rows.length;-- } catch (err) {-- if (rows.length > 1 && err.permanent) {-- // One bad envelope rejects the whole batch: isolate it.-- 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) markFailure(row, err);-- if (err.permanent) stats.rejected += rows.length; else stats.failed += rows.length;-- }-- }-- return stats;-- }++// ── Outbox and inbox: openvibe-sdk's PostgreSQL kits ─────────-- async function doFlush() {-- const total = { sent: 0, failed: 0, rejected: 0 };-- for (;;) {-- const rows = q().due.all(now(), batchSize);-- 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;-- }---- /** Publish due rows. Concurrent callers share the flush in progress. */-- 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);-- timer.unref?.();-- }---- return {-- ensureSchema,-- enqueue,-- flush,-- start() { if (!running) { running = true; schedule(0); } },-- stop() { running = false; clearTimeout(timer); timer = null; return flushing || Promise.resolve(); },-- /** Wake the relay now (e.g. right after the transaction commits). */-- kick() { schedule(0); },-- pending: () => q().pending.get().n,-- /** Delete sent rows older than `olderThanMs`. */-- prune(olderThanMs = 7 * 24 * 60 * 60 * 1000) { return q().prune.run(now() - olderThanMs).changes; },-- };--}----// ── Inbox (exactly-once effects) ───────────────────────────----/**-- * createInbox(db, { table = 'idempotency_receipts' })-- * ensureSchema()-- * once(consumer, eventId, fn) -> { duplicate: true } | { duplicate: false, result }-- *-- * fn runs inside a better-sqlite3 transaction together with the receipt insert: either both commit-- * (the effect happened exactly once) or neither does (fn threw; the redelivery will try again).-- * fn must be synchronous and make its effects through this same database.-- */--function createInbox(db, { table = 'idempotency_receipts', now = () => Date.now() } = {}) {-- if (!db || typeof db.prepare !== 'function') throw new TypeError('a better-sqlite3 database is required');-- if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(table)) throw new TypeError('bad inbox table name');-- let claim = null;---- function ensureSchema() {-- db.exec(`CREATE TABLE IF NOT EXISTS ${table} (-- consumer TEXT NOT NULL,-- event_id TEXT NOT NULL,-- processed_at INTEGER NOT NULL,-- PRIMARY KEY (consumer, event_id)-- )`);-- claim = null;-- }---- function once(consumer, eventId, fn) {-- if (!consumer || !eventId) throw new TypeError('consumer and eventId are required');-- if (!claim) claim = db.prepare(`INSERT OR IGNORE INTO ${table} (consumer, event_id, processed_at) VALUES (?, ?, ?)`);-- return db.transaction(() => {-- if (claim.run(String(consumer), String(eventId), now()).changes === 0) return { duplicate: true };-- const result = fn();-- if (result && typeof result.then === 'function') {-- throw new TypeError('inbox.once(): fn must be synchronous (it runs inside a SQLite transaction)');-- }-- return { duplicate: false, result };-- })();-- }---- function seen(consumer, eventId) {-- return Boolean(db.prepare(`SELECT 1 FROM ${table} WHERE consumer = ? AND event_id = ?`).get(String(consumer), String(eventId)));-- }---- return { ensureSchema, once, seen };--}++const { createPgOutbox, createPgInbox, outboxSchema, inboxSchema } = require('openvibe-sdk/events');--module.exports = { sign, verifyDelivery, signV2, verifyDeliveryV2, prepareEnvelope, createPublisher, createOutbox, createInbox, PublishError };++module.exports = { sign, verifyDelivery, signV2, verifyDeliveryV2, prepareEnvelope, createPublisher, createPgOutbox, createPgInbox, outboxSchema, inboxSchema, PublishError };@@@ -9,7 +9,6 @@"version": "0.1.0","license": "AGPL-3.0-only","dependencies": {-- "better-sqlite3": "^11.10.0","dotenv": "^16.4.7","express": "^4.21.2","iovalkey": "^0.4.0",@@@ -108,7 +107,9 @@"url": "https://feross.org/support"}],-- "license": "MIT"++ "license": "MIT",++ "optional": true,++ "peer": true},"node_modules/better-sqlite3": {"version": "11.10.0",@@@ -116,6 +117,8 @@"integrity": "sha512-EwhOpyXiOEL/lKzHz9AW1msWFNzGc/z+LzeB3/jnFJpxu+th2yqvzsSWas1v9jgs9+xiXJcD5A8CJxAG2TaghQ==","hasInstallScript": true,"license": "MIT",++ "optional": true,++ "peer": true,"dependencies": {"bindings": "^1.5.0","prebuild-install": "^7.1.1"@@@ -126,6 +129,8 @@"resolved": "https://registry.npmjs.org/bindings/-/bindings-1.5.0.tgz","integrity": "sha512-p2q/t/mhvuOj/UeLlV6566GD/guowlr0hHxClI0W9m7MWYkL1F0hLo+0Aexs9HSPCtR1SXQ0TD3MMKrXZajbiQ==","license": "MIT",++ "optional": true,++ "peer": true,"dependencies": {"file-uri-to-path": "1.0.0"}@@@ -135,6 +140,8 @@"resolved": "https://registry.npmjs.org/bl/-/bl-4.1.0.tgz","integrity": "sha512-1W07cM9gS6DcLperZfFSj+bWLtaPGSOHWhPiGzXmvVJbRLdG82sH/Kn8EtW1VqWVA54AKf2h5k5BbnIbwF3h6w==","license": "MIT",++ "optional": true,++ "peer": true,"dependencies": {"buffer": "^5.5.0","inherits": "^2.0.4",@@@ -184,6 +191,8 @@}],"license": "MIT",++ "optional": true,++ "peer": true,"dependencies": {"base64-js": "^1.3.1","ieee754": "^1.1.13"@@@ -237,7 +246,9 @@"version": "1.1.4","resolved": "https://registry.npmjs.org/chownr/-/chownr-1.1.4.tgz","integrity": "sha512-jJ0bqzaylmJtVnNgzTeSOs8DPavpbYgEr/b0YL8/2GO3xJEhInFmhKMUnEJQjZumK7KXGFhUy89PrsJWlakBVg==",-- "license": "ISC"++ "license": "ISC",++ "optional": true,++ "peer": true},"node_modules/cluster-key-slot": {"version": "1.1.2",@@@ -298,6 +309,8 @@"resolved": "https://registry.npmjs.org/decompress-response/-/decompress-response-6.0.0.tgz","integrity": "sha512-aW35yZM6Bb/4oJlZncMH2LCoZtJXTRxES17vE3hoRiowU2kWHaJKFkSBDnDR+cm9J+9QhXmREyIfv0pji9ejCQ==","license": "MIT",++ "optional": true,++ "peer": true,"dependencies": {"mimic-response": "^3.1.0"},@@@ -313,6 +326,8 @@"resolved": "https://registry.npmjs.org/deep-extend/-/deep-extend-0.6.0.tgz","integrity": "sha512-LOHxIOaPYdHlJRtCQfDIVZtfw/ufM8+rVj649RIHzcm/vGwQRXFt6OPqIFWsm2XEMrNIEtWR64sY1LEKD2vAOA==","license": "MIT",++ "optional": true,++ "peer": true,"engines": {"node": ">=4.0.0"}@@@ -350,6 +365,8 @@"resolved": "https://registry.npmjs.org/detect-libc/-/detect-libc-2.1.2.tgz","integrity": "sha512-Btj2BOOO83o3WyH59e8MgXsxEQVcarkUOpEYrubB0urwnN10yQ364rsiByU11nZlqWYZm05i/of7io4mzihBtQ==","license": "Apache-2.0",++ "optional": true,++ "peer": true,"engines": {"node": ">=8"}@@@ -409,6 +426,8 @@"resolved": "https://registry.npmjs.org/end-of-stream/-/end-of-stream-1.4.5.tgz","integrity": "sha512-ooEGc6HP26xXq/N+GCGOT0JKCLDGrq2bQUZrQ7gyrJiZANJ/8YDTxTpQBXGMn+WbIQXNVpyWymm7KYVICQnyOg==","license": "MIT",++ "optional": true,++ "peer": true,"dependencies": {"once": "^1.4.0"}@@@ -463,6 +482,8 @@"resolved": "https://registry.npmjs.org/expand-template/-/expand-template-2.0.3.tgz","integrity": "sha512-XYfuKMvj4O35f/pOXLObndIRvyQ+/+6AhODh+OKWj9S9498pHHn/IMszH+gt0fBCRWMNfk1ZSp5x3AifmnI2vg==","license": "(MIT OR WTFPL)",++ "optional": true,++ "peer": true,"engines": {"node": ">=6"}@@@ -539,7 +560,9 @@"version": "1.0.0","resolved": "https://registry.npmjs.org/file-uri-to-path/-/file-uri-to-path-1.0.0.tgz","integrity": "sha512-0Zt+s3L7Vf1biwWZ29aARiVYLx7iMGnEUl9x33fbB/j3jR81u/O2LbqK+Bm1CDSNDKVtJ/YjwY7TUd5SkeLQLw==",-- "license": "MIT"++ "license": "MIT",++ "optional": true,++ "peer": true},"node_modules/finalhandler": {"version": "1.3.2",@@@ -581,7 +604,9 @@"version": "1.0.0","resolved": "https://registry.npmjs.org/fs-constants/-/fs-constants-1.0.0.tgz","integrity": "sha512-y6OAwoSIf7FyjMIv94u+b5rdheZEjzR63GTyZJm5qh4Bi+2YgwLCcI/fPFZkL5PSixOt6ZNKm+w+Hfp/Bciwow==",-- "license": "MIT"++ "license": "MIT",++ "optional": true,++ "peer": true},"node_modules/function-bind": {"version": "1.1.2",@@@ -633,7 +658,9 @@"version": "0.0.0","resolved": "https://registry.npmjs.org/github-from-package/-/github-from-package-0.0.0.tgz","integrity": "sha512-SyHy3T1v2NUXn29OsWdxmK6RwHD+vkj3v8en8AOBZ1wBQ/hCAQ5bAQTD02kW4W9tUp/3Qh6J8r9EvntiyCmOOw==",-- "license": "MIT"++ "license": "MIT",++ "optional": true,++ "peer": true},"node_modules/gopd": {"version": "1.2.0",@@@ -721,7 +748,9 @@"url": "https://feross.org/support"}],-- "license": "BSD-3-Clause"++ "license": "BSD-3-Clause",++ "optional": true,++ "peer": true},"node_modules/inherits": {"version": "2.0.4",@@@ -733,7 +762,9 @@"version": "1.3.8","resolved": "https://registry.npmjs.org/ini/-/ini-1.3.8.tgz","integrity": "sha512-JV/yugV2uzW5iMRSiZAyDtQd+nxtUnjeLt0acNdw98kKLrvuRVyB80tsREOE7yvGVgalhZ6RNXCmEHkUKBKxew==",-- "license": "ISC"++ "license": "ISC",++ "optional": true,++ "peer": true},"node_modules/iovalkey": {"version": "0.4.0",@@@ -970,6 +1001,8 @@"resolved": "https://registry.npmjs.org/mimic-response/-/mimic-response-3.1.0.tgz","integrity": "sha512-z0yWI+4FDrrweS8Zmt4Ej5HdJmky15+L2e6Wgn3+iK5fWzb6T3fhNFq2+MeTRb064c6Wr4N/wv0DzQTjNzHNGQ==","license": "MIT",++ "optional": true,++ "peer": true,"engines": {"node": ">=10"},@@@ -982,6 +1015,8 @@"resolved": "https://registry.npmjs.org/minimist/-/minimist-1.2.8.tgz","integrity": "sha512-2yyAR8qBkN3YuheJanUpWC5U3bb5osDywNB8RzDVlDwDHbocAJveqqj1u8+SVD7jkWT4yvsHCpWqqWqAxb0zCA==","license": "MIT",++ "optional": true,++ "peer": true,"funding": {"url": "https://github.com/sponsors/ljharb"}@@@ -990,7 +1025,9 @@"version": "0.5.3","resolved": "https://registry.npmjs.org/mkdirp-classic/-/mkdirp-classic-0.5.3.tgz","integrity": "sha512-gKLcREMhtuZRwRAfqP3RFW+TK4JqApVBtOIftVgjuABpAtpxhPGaDcfvbhNvD0B8iD1oUr/txX35NjcaY6Ns/A==",-- "license": "MIT"++ "license": "MIT",++ "optional": true,++ "peer": true},"node_modules/ms": {"version": "2.0.0",@@@ -1002,7 +1039,9 @@"version": "2.0.0","resolved": "https://registry.npmjs.org/napi-build-utils/-/napi-build-utils-2.0.0.tgz","integrity": "sha512-GEbrYkbfF7MoNaoh2iGG84Mnf/WZfB0GdGEsM8wz7Expx/LlWf5U8t9nvJKXSp3qr5IsEbK04cBGhol/KwOsWA==",-- "license": "MIT"++ "license": "MIT",++ "optional": true,++ "peer": true},"node_modules/negotiator": {"version": "0.6.3",@@@ -1018,6 +1057,8 @@"resolved": "https://registry.npmjs.org/node-abi/-/node-abi-3.96.0.tgz","integrity": "sha512-rebQ/lz7i0EkoLzUVSrKRzA69zMkwLp95kKMWoMDkkM00Suxz0D7zEQPwRml5fQum24mj7bPvmlgLAmu2JCiYg==","license": "MIT",++ "optional": true,++ "peer": true,"dependencies": {"semver": "^7.3.5"},@@@ -1054,6 +1095,8 @@"resolved": "https://registry.npmjs.org/once/-/once-1.4.0.tgz","integrity": "sha512-lNaJgI+2Q5URQBkccEKHTQOPaXdUxnZZElQTZY0MFUAuaEqe1E+Nyvgdz/aIyNi6Z9MzO5dv1H8n58/GELp3+w==","license": "ISC",++ "optional": true,++ "peer": true,"dependencies": {"wrappy": "1"}@@@ -1261,6 +1304,8 @@"integrity": "sha512-8Mf2cbV7x1cXPUILADGI3wuhfqWvtiLA1iclTDbFRZkgRQS0NqsPZphna9V+HyTEadheuPmjaJMsbzKQFOzLug==","deprecated": "No longer maintained. Please contact the author of the relevant native addon; alternatives are available.","license": "MIT",++ "optional": true,++ "peer": true,"dependencies": {"detect-libc": "^2.0.0","expand-template": "^2.0.3",@@@ -1304,6 +1349,8 @@"resolved": "https://registry.npmjs.org/pump/-/pump-3.0.4.tgz","integrity": "sha512-VS7sjc6KR7e1ukRFhQSY5LM2uBWAUPiOPa/A3mkKmiMwSmRFUITt0xuj+/lesgnCv+dPIEYlkzrcyXgquIHMcA==","license": "MIT",++ "optional": true,++ "peer": true,"dependencies": {"end-of-stream": "^1.1.0","once": "^1.3.1"@@@ -1354,6 +1401,8 @@"resolved": "https://registry.npmjs.org/rc/-/rc-1.2.8.tgz","integrity": "sha512-y3bGgqKj3QBdxLbLkomlohkvsA8gdAiUQlSBJnBhfn+BPxg4bc62d8TcBW15wavDfgexCgccckhcZvywyQYPOw==","license": "(BSD-2-Clause OR MIT OR Apache-2.0)",++ "optional": true,++ "peer": true,"dependencies": {"deep-extend": "^0.6.0","ini": "~1.3.0",@@@ -1369,6 +1418,8 @@"resolved": "https://registry.npmjs.org/readable-stream/-/readable-stream-3.6.2.tgz","integrity": "sha512-9u/sniCrY3D5WdsERHzHE4G2YCXqoG5FTHUiCC4SIbr6XcLZBY05ya9EKjYek9O5xOAwjGq+1JdGBAS7Q9ScoA==","license": "MIT",++ "optional": true,++ "peer": true,"dependencies": {"inherits": "^2.0.3","string_decoder": "^1.1.1",@@@ -1587,7 +1638,9 @@"url": "https://feross.org/support"}],-- "license": "MIT"++ "license": "MIT",++ "optional": true,++ "peer": true},"node_modules/simple-get": {"version": "4.0.1",@@@ -1608,6 +1661,8 @@}],"license": "MIT",++ "optional": true,++ "peer": true,"dependencies": {"decompress-response": "^6.0.0","once": "^1.3.1",@@@ -1643,6 +1698,8 @@"resolved": "https://registry.npmjs.org/string_decoder/-/string_decoder-1.3.0.tgz","integrity": "sha512-hkRX8U1WjJFd8LsDJ2yQ/wWWxaopEsABU1XfkM8A+j0+85JAGppt16cr1Whg6KIbb4okU6Mql6BOj+uup/wKeA==","license": "MIT",++ "optional": true,++ "peer": true,"dependencies": {"safe-buffer": "~5.2.0"}@@@ -1652,6 +1709,8 @@"resolved": "https://registry.npmjs.org/strip-json-comments/-/strip-json-comments-2.0.1.tgz","integrity": "sha512-4gB8na07fecVVkOI6Rs4e7T6NOTki5EmL7TUduTs6bu3EdnSycntVJ4re8kgZA+wx9IueI2Y11bfbgwtzuE0KQ==","license": "MIT",++ "optional": true,++ "peer": true,"engines": {"node": ">=0.10.0"}@@@ -1661,6 +1720,8 @@"resolved": "https://registry.npmjs.org/tar-fs/-/tar-fs-2.1.5.tgz","integrity": "sha512-OboTd8mmMhZDNPV+UjQcK9yKAatXu2aJ+r1w4im1Otd4M4fl2hwvdoXUxIYHFTHWK/3y3FarBP70v3vwmGlOxw==","license": "MIT",++ "optional": true,++ "peer": true,"dependencies": {"chownr": "^1.1.1","mkdirp-classic": "^0.5.2",@@@ -1673,6 +1734,8 @@"resolved": "https://registry.npmjs.org/tar-stream/-/tar-stream-2.2.0.tgz","integrity": "sha512-ujeqbceABgwMZxEJnk2HDY2DlnUZ+9oEcb1KzTVfYHio0UE6dG71n60d8D2I4qNvleWrrXpmjpt7vZeF1LnMZQ==","license": "MIT",++ "optional": true,++ "peer": true,"dependencies": {"bl": "^4.0.3","end-of-stream": "^1.4.1",@@@ -1698,6 +1761,8 @@"resolved": "https://registry.npmjs.org/tunnel-agent/-/tunnel-agent-0.6.0.tgz","integrity": "sha512-McnNiV1l8RYeY8tBgEpuodCC1mLUdbSN+CYBL7kJsJNInOP8UjDDEwdk6Mw60vdLLrr5NHKZhMAOSrR2NZuQ+w==","license": "Apache-2.0",++ "optional": true,++ "peer": true,"dependencies": {"safe-buffer": "^5.0.1"},@@@ -1731,7 +1796,9 @@"version": "1.0.2","resolved": "https://registry.npmjs.org/util-deprecate/-/util-deprecate-1.0.2.tgz","integrity": "sha512-EPD5q1uXyFxJpCrLnCc1nHnq3gOa6DZBocAIiI2TaSCA7VCJ1UJDMagCzIkXNsUYfD1daK//LTEQ8xiIbrHtcw==",-- "license": "MIT"++ "license": "MIT",++ "optional": true,++ "peer": true},"node_modules/utils-merge": {"version": "1.0.1",@@@ -1755,7 +1822,9 @@"version": "1.0.2","resolved": "https://registry.npmjs.org/wrappy/-/wrappy-1.0.2.tgz","integrity": "sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ==",-- "license": "ISC"++ "license": "ISC",++ "optional": true,++ "peer": true},"node_modules/xtend": {"version": "4.0.2",@@@ -21,7 +21,6 @@"license": "AGPL-3.0-only","repository": "github:OpenVibers/OpenVibe.Events","dependencies": {-- "better-sqlite3": "^11.10.0","dotenv": "^16.4.7","express": "^4.21.2","iovalkey": "^0.4.0",@@@ -125,12 +125,12 @@ async function subscriber(respond = () => 204) {const server = nodeHttp.createServer((req, res) => {const chunks = [];req.on('data', c => chunks.push(c));-- req.on('end', () => {++ req.on('end', async () => {const rawBody = Buffer.concat(chunks);const call = { headers: req.headers, rawBody, body: JSON.parse(rawBody.toString('utf8') || 'null') };calls.push(call);let status;-- try { status = respond(call, calls.length); } catch (err) { status = 500; call.error = err; }++ try { status = await respond(call, calls.length); } catch (err) { status = 500; call.error = err; } // respond may be async (a consumer's inbox)if (status === 'destroy') { req.socket.destroy(); return; }res.statusCode = status;res.end();@@@ -1,11 +1,10 @@'use strict';const assert = require('assert');--const path = require('path');const nodeHttp = require('http');--const Database = require('better-sqlite3');++const { createDb, sql } = require('openvibe-sdk/db');const { serviceAuth } = require('openvibe-contracts');const events = require('..');--const { boot, request, serviceToken, envelope, subscriber, suite, tmpDir } = require('./helpers');++const { boot, request, serviceToken, envelope, subscriber, suite } = require('./helpers');const t = suite('outbox-inbox');let h;@@@ -54,130 +53,132 @@ t('publisher fills event_id, timestamp and trace (from traceparent)', async () =await assert.rejects(publisher.publish(envelope('media')), (err) => err.status === 403 && err.code === 'events.source_mismatch' && err.permanent);});++// The producer's own database: embedded PostgreSQL, like a service's (openvibe-sdk/db).++async function producerDb(extra = '') {++ const db = createDb({ pglite: true });++ await db.query(`${extra}${events.outboxSchema()}${events.inboxSchema()}`);++ return db;++}++t('outbox: enqueue inside a rolled-back transaction publishes nothing; committed publishes exactly once', async () => {-- const db = new Database(path.join(tmpDir(), 'producer.db'));-- db.exec('CREATE TABLE vods (id INTEGER PRIMARY KEY, status TEXT)');-- const outbox = events.createOutbox(db, { publisher: events.createPublisher({ eventsUrl: h.base, tokenClient }) });-- outbox.ensureSchema();---- assert.throws(() => outbox.enqueue(envelope()), /inside the transaction/);---- let rolledBack;-- assert.throws(() => db.transaction(() => {-- db.prepare('INSERT INTO vods (id, status) VALUES (1, ?)').run('ready');-- rolledBack = outbox.enqueue(envelope('live', { event_type: 'live.vod.ready' }));-- throw new Error('domain write failed');-- })(), /domain write failed/);-- assert.strictEqual(outbox.pending(), 0);-- let r = await outbox.flush();-- assert.deepStrictEqual(r, { sent: 0, failed: 0, rejected: 0 });-- assert.strictEqual(await countIn(rolledBack.event_id), 0, 'rolled back: never published');---- let committed;-- db.transaction(() => {-- db.prepare('INSERT INTO vods (id, status) VALUES (2, ?)').run('ready');-- committed = outbox.enqueue(envelope('live', { event_type: 'live.vod.ready' }));-- })();-- assert.strictEqual(outbox.pending(), 1);-- r = await outbox.flush();-- assert.strictEqual(r.sent, 1);-- r = await outbox.flush();-- assert.strictEqual(r.sent, 0, 'sent rows are not sent again');-- assert.strictEqual(await countIn(committed.event_id), 1);---- // Relay crashed after Events accepted but before the row was marked: at-least-once resend is a duplicate.-- db.prepare('UPDATE event_outbox SET sent_at = NULL WHERE event_id = ?').run(committed.event_id);-- r = await outbox.flush();-- assert.strictEqual(r.sent, 1);-- assert.strictEqual(await countIn(committed.event_id), 1, 'still exactly one event');-- assert.ok(db.prepare('SELECT seq FROM event_outbox WHERE event_id = ?').get(committed.event_id).seq > 0);-- db.close();++ const db = await producerDb('CREATE TABLE vods (id bigint PRIMARY KEY, status text);');++ const outbox = events.createPgOutbox(db, { events: events.createPublisher({ eventsUrl: h.base, tokenClient }) });++ try {++ await assert.rejects(outbox.enqueue(null, envelope()), /transaction handle/);++++ let rolledBack;++ await assert.rejects(db.tx(async (t) => {++ await t.exec(sql`INSERT INTO vods (id, status) VALUES (1, 'ready')`);++ rolledBack = await outbox.enqueue(t, envelope('live', { event_type: 'live.vod.ready' }));++ throw new Error('domain write failed');++ }), /domain write failed/);++ assert.strictEqual(await outbox.pending(), 0);++ let r = await outbox.flush();++ assert.deepStrictEqual(r, { sent: 0, failed: 0, rejected: 0 });++ assert.strictEqual(await countIn(rolledBack.event_id), 0, 'rolled back: never published');++++ const committed = await db.tx(async (t) => {++ await t.exec(sql`INSERT INTO vods (id, status) VALUES (2, 'ready')`);++ return outbox.enqueue(t, envelope('live', { event_type: 'live.vod.ready' }));++ });++ assert.strictEqual(await outbox.pending(), 1);++ r = await outbox.flush();++ assert.strictEqual(r.sent, 1);++ r = await outbox.flush();++ assert.strictEqual(r.sent, 0, 'sent rows are not sent again');++ assert.strictEqual(await countIn(committed.event_id), 1);++++ // Relay crashed after Events accepted but before the row was marked: at-least-once resend is a duplicate.++ await db.query('UPDATE event_outbox SET sent_at = NULL, next_attempt_at = 0 WHERE event_id = $1', [committed.event_id]);++ r = await outbox.flush();++ assert.strictEqual(r.sent, 1);++ assert.strictEqual(await countIn(committed.event_id), 1, 'still exactly one event');++ assert.ok(Number((await db.one('SELECT seq FROM event_outbox WHERE event_id = $1', [committed.event_id])).seq) > 0);++ } finally { await db.close(); }});t('outbox: a poisoned row is isolated and rejected; a down Events is retried later', async () => {-- const db = new Database(path.join(tmpDir(), 'producer2.db'));++ const db = await producerDb();let clock = Date.now();-- const outbox = events.createOutbox(db, { publisher: events.createPublisher({ eventsUrl: h.base, tokenClient }), now: () => clock });-- outbox.ensureSchema();-- let good; let bad;-- db.transaction(() => {-- good = outbox.enqueue(envelope());-- bad = outbox.enqueue(envelope('media')); // live may not publish as media-- })();-- const r = await outbox.flush();-- assert.deepStrictEqual(r, { sent: 1, failed: 0, rejected: 1 });-- assert.strictEqual(await countIn(good.event_id), 1);-- assert.ok(db.prepare('SELECT rejected_at FROM event_outbox WHERE event_id = ?').get(bad.event_id).rejected_at);---- const down = events.createOutbox(db, { publisher: events.createPublisher({ eventsUrl: 'http://127.0.0.1:9', tokenClient }), now: () => clock });-- let later;-- db.transaction(() => { later = down.enqueue(envelope()); })();-- const d = await down.flush();-- assert.deepStrictEqual(d, { sent: 0, failed: 1, rejected: 0 });-- assert.strictEqual(down.pending(), 1);-- const row = db.prepare('SELECT * FROM event_outbox WHERE event_id = ?').get(later.event_id);-- assert.strictEqual(row.next_attempt_at, clock + 1000, 'backoff after the first failure');-- // Events is back: the regular relay picks it up once due.-- clock += 1000;-- assert.strictEqual((await outbox.flush()).sent, 1);-- assert.strictEqual(await countIn(later.event_id), 1);-- db.close();++ const outbox = events.createPgOutbox(db, { events: events.createPublisher({ eventsUrl: h.base, tokenClient }), now: () => clock });++ try {++ const [good, bad] = await db.tx(async (t) => [++ await outbox.enqueue(t, envelope()),++ await outbox.enqueue(t, envelope('media')), // live may not publish as media++ ]);++ const r = await outbox.flush();++ assert.deepStrictEqual(r, { sent: 1, failed: 0, rejected: 1 });++ assert.strictEqual(await countIn(good.event_id), 1);++ assert.ok((await db.one('SELECT rejected_at FROM event_outbox WHERE event_id = $1', [bad.event_id])).rejected_at);++++ const down = events.createPgOutbox(db, { events: events.createPublisher({ eventsUrl: 'http://127.0.0.1:9', tokenClient }), now: () => clock });++ const later = await db.tx((t) => down.enqueue(t, envelope()));++ const d = await down.flush();++ assert.deepStrictEqual(d, { sent: 0, failed: 1, rejected: 0 });++ assert.strictEqual(await down.pending(), 1);++ const row = await db.one('SELECT * FROM event_outbox WHERE event_id = $1', [later.event_id]);++ assert.strictEqual(Number(row.next_attempt_at), clock + 1000, 'backoff after the first failure');++ // Events is back: the regular relay picks it up once due.++ clock += 1000;++ assert.strictEqual((await outbox.flush()).sent, 1);++ assert.strictEqual(await countIn(later.event_id), 1);++ } finally { await db.close(); }});--t('consumer crash mid-processing + redelivery + replay -> exactly one effect (createInbox)', async () => {-- const consumerDb = new Database(path.join(tmpDir(), 'consumer.db'));-- consumerDb.exec('CREATE TABLE credits (id INTEGER PRIMARY KEY AUTOINCREMENT, event_id TEXT, amount INTEGER)');-- const inbox = events.createInbox(consumerDb);-- inbox.ensureSchema();++t('consumer crash mid-processing + redelivery + replay -> exactly one effect (createPgInbox)', async () => {++ const consumerDb = await producerDb('CREATE TABLE credits (id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, event_id text, amount integer);');++ const inbox = events.createPgInbox(consumerDb);let secret = null;let mode = 'crash-after-commit';-- const stub = await subscriber((call) => {++ const stub = await subscriber(async (call) => {if (!events.verifyDeliveryV2(call.rawBody, call.headers, secret, { now: h.clock.now() })) return 401;const ev = call.body.event;if (mode === 'crash-before-commit') {mode = 'ok';try {-- inbox.once('media', ev.event_id, () => {-- consumerDb.prepare('INSERT INTO credits (event_id, amount) VALUES (?, ?)').run(ev.event_id, ev.payload.amount);++ await inbox.once('media', ev.event_id, async (t) => {++ await t.query('INSERT INTO credits (event_id, amount) VALUES ($1, $2)', [ev.event_id, ev.payload.amount]);throw new Error('process died mid-transaction');});} catch { /* the transaction rolled back */ }return 'destroy';}-- inbox.once('media', ev.event_id, () => {-- consumerDb.prepare('INSERT INTO credits (event_id, amount) VALUES (?, ?)').run(ev.event_id, ev.payload.amount);++ await inbox.once('media', ev.event_id, async (t) => {++ await t.query('INSERT INTO credits (event_id, amount) VALUES ($1, $2)', [ev.event_id, ev.payload.amount]);});if (mode === 'crash-after-commit') { mode = 'ok'; return 'destroy'; } // effect committed, no response sentreturn 204;});-- const media = serviceToken('media', ['events.subscription.manage']);-- const sub = (await request(h.base, 'POST', '/api/v1/subscriptions', { token: media, body: { topic_pattern: 'live.tip.*', endpoint: stub.url } })).body;-- secret = sub.secret;-- const live = serviceToken('live', ['events.event.publish']);-- const effects = (id) => consumerDb.prepare('SELECT COUNT(*) AS n FROM credits WHERE event_id = ?').get(id).n;---- for (const first of ['crash-after-commit', 'crash-before-commit']) {-- mode = first;-- const env = envelope('live', { event_type: 'live.tip.sent', payload: { amount: 5 } });-- await request(h.base, 'POST', '/api/v1/events', { token: live, body: env });-- await h.worker.drain();-- const d = await h.store.getDelivery(env.event_id, sub.id);-- assert.strictEqual(d.status, 'failed', `${first}: the crashed attempt is retried`);-- assert.strictEqual(effects(env.event_id), first === 'crash-after-commit' ? 1 : 0);-- h.clock.advance(1000);-- await h.worker.drain();-- assert.strictEqual((await h.store.getDelivery(env.event_id, sub.id)).status, 'delivered');-- assert.strictEqual(effects(env.event_id), 1, `${first}: exactly one effect after redelivery`);---- const rp = await request(h.base, 'POST', '/api/v1/deliveries/replay', { token: admin, body: { subscription_id: sub.id, event_ids: [env.event_id] } });-- assert.strictEqual(rp.body.queued, 1);-- await h.worker.drain();-- assert.strictEqual(effects(env.event_id), 1, `${first}: replay does not repeat the effect`);++ try {++ const media = serviceToken('media', ['events.subscription.manage']);++ const sub = (await request(h.base, 'POST', '/api/v1/subscriptions', { token: media, body: { topic_pattern: 'live.tip.*', endpoint: stub.url } })).body;++ secret = sub.secret;++ const live = serviceToken('live', ['events.event.publish']);++ const effects = async (id) => Number((await consumerDb.one('SELECT COUNT(*) AS n FROM credits WHERE event_id = $1', [id])).n);++++ for (const first of ['crash-after-commit', 'crash-before-commit']) {++ mode = first;++ const env = envelope('live', { event_type: 'live.tip.sent', payload: { amount: 5 } });++ await request(h.base, 'POST', '/api/v1/events', { token: live, body: env });++ await h.worker.drain();++ const d = await h.store.getDelivery(env.event_id, sub.id);++ assert.strictEqual(d.status, 'failed', `${first}: the crashed attempt is retried`);++ assert.strictEqual(await effects(env.event_id), first === 'crash-after-commit' ? 1 : 0);++ h.clock.advance(1000);++ await h.worker.drain();++ assert.strictEqual((await h.store.getDelivery(env.event_id, sub.id)).status, 'delivered');++ assert.strictEqual(await effects(env.event_id), 1, `${first}: exactly one effect after redelivery`);++++ const rp = await request(h.base, 'POST', '/api/v1/deliveries/replay', { token: admin, body: { subscription_id: sub.id, event_ids: [env.event_id] } });++ assert.strictEqual(rp.body.queued, 1);++ await h.worker.drain();++ assert.strictEqual(await effects(env.event_id), 1, `${first}: replay does not repeat the effect`);++ }++ assert.strictEqual(await inbox.seen('media', 'evt_x'), false);++ } finally {++ await stub.close();++ await consumerDb.close();}-- assert.throws(() => inbox.once('media', 'evt_x', () => Promise.resolve()), /synchronous/);-- assert.strictEqual(inbox.seen('media', 'evt_x'), false, 'an async fn is rolled back, not recorded');-- await stub.close();-- consumerDb.close();});t('stop', async () => { await h.stop(); network.close(); });
@@ -169,27 +169,28 @@ Quotas recorded in Network (`dev_quotas`) are not read yet: that needs `network.
@@ -169,27 +169,28 @@ Quotas recorded in Network (`dev_quotas`) are not read yet: that needs `network.
const events = require('openvibe-events');
const events = require('openvibe-events');
const tokenClient = contracts.serviceAuth.createTokenClient({ tokenUrl: `${NETWORK}/oauth/token`, clientId, clientSecret, audience: 'openvibe.events' });
const tokenClient = contracts.serviceAuth.createTokenClient({ tokenUrl: `${NETWORK}/oauth/token`, clientId, clientSecret, audience: 'openvibe.events' });
-// Producer: transactional outbox in the service's own better-sqlite3 database (a service on PostgreSQL uses
-// openvibe-sdk/events createPgOutbox / createPgInbox instead: same envelope, receipts and relay, async)
-const outbox = events.createOutbox(db, { publisher: events.createPublisher({ eventsUrl: 'http://127.0.0.1:4300', tokenClient }) });
-outbox.ensureSchema();
-db.transaction(() => {
- db.prepare('UPDATE vods SET status = ? WHERE id = ?').run('ready', id);
- outbox.enqueue({ event_type: 'media.vod.ready', source: 'media', actor, subject: { type: 'vod', id, revision }, payload });
-})(); // the event exists if and only if the change committed
+// Producer: transactional outbox in the service's own PostgreSQL (openvibe-sdk/db; the table from events.outboxSchema()
+// goes in one of the service's migrations)
+const outbox = events.createPgOutbox(db, { events: events.createPublisher({ eventsUrl: 'http://127.0.0.1:4300', tokenClient }) });
+await db.tx(async (t) => {
+ await t.query('UPDATE vods SET status = $1 WHERE id = $2', ['ready', id]);
+ await outbox.enqueue(t, { event_type: 'media.vod.ready', source: 'media', actor, subject: { type: 'vod', id, revision }, payload });
+}); // the event exists if and only if the change committed
outbox.start(); // relay: publishes pending rows, marks them sent, backs off on failure
outbox.start(); // relay: publishes pending rows, marks them sent, backs off on failure
// Consumer: signed webhook (v2: signature plus a 300 s replay window) + exactly-once effects in the consumer's own database
// Consumer: signed webhook (v2: signature plus a 300 s replay window) + exactly-once effects in the consumer's own database
-const inbox = events.createInbox(db);
-inbox.ensureSchema();
-app.post('/internal/events', express.json({ verify: (req, _res, buf) => { req.rawBody = buf; } }), (req, res) => {
+const inbox = events.createPgInbox(db); // the table from events.inboxSchema()
+app.post('/internal/events', express.json({ verify: (req, _res, buf) => { req.rawBody = buf; } }), async (req, res) => {
if (!events.verifyDeliveryV2(req.rawBody, req.headers, SECRET)) return res.sendStatus(401);
if (!events.verifyDeliveryV2(req.rawBody, req.headers, SECRET)) return res.sendStatus(401);
- inbox.once('live', req.body.event.event_id, () => { /* synchronous db writes */ });
+ await inbox.once('live', req.body.event.event_id, async (t) => { /* writes on t */ });
res.sendStatus(204);
res.sendStatus(204);
});
});
```
```
-`enqueue()` refuses to run outside a transaction. `once()` records the receipt and runs the handler in one SQLite transaction, so a crash before commit leaves nothing behind and the redelivery runs it again, and a crash after commit makes the redelivery a no-op.
+`enqueue(t, …)` takes the transaction handle, so it cannot run outside one. `once()` records the receipt and runs the handler
+in one transaction, so a crash before commit leaves nothing behind and the redelivery runs it again, and a crash after
+commit makes the redelivery a no-op. The outbox and inbox are openvibe-sdk's PostgreSQL kits (ADR-042): this package
+re-exports them and keeps no SQLite copies.
## Realtime (SSE)
## Realtime (SSE)
@@ -4,23 +4,26 @@
@@ -4,23 +4,26 @@
*
*
* const events = require('openvibe-events');
* const events = require('openvibe-events');
*
*
- * // Producer: transactional outbox (better-sqlite3), relayed to OpenVibe.Events
+ * // Producer: transactional outbox in the service's own PostgreSQL (openvibe-sdk/db), relayed to OpenVibe.Events
* const publisher = events.createPublisher({ eventsUrl: 'http://127.0.0.1:4300', tokenClient });
* const publisher = events.createPublisher({ eventsUrl: 'http://127.0.0.1:4300', tokenClient });
- * const outbox = events.createOutbox(db, { publisher });
- * outbox.ensureSchema();
- * db.transaction(() => {
- * db.prepare('UPDATE vods SET status = ? WHERE id = ?').run('ready', id);
- * outbox.enqueue({ event_type: 'media.vod.ready', source: 'media', actor, subject, payload });
- * })();
+ * const outbox = events.createPgOutbox(db, { events: publisher }); // table from events.outboxSchema() in a migration
+ * await db.tx(async (t) => {
+ * await t.query('UPDATE vods SET status = $1 WHERE id = $2', ['ready', id]);
+ * await outbox.enqueue(t, { event_type: 'media.vod.ready', source: 'media', actor, subject, payload });
+ * });
* outbox.start();
* outbox.start();
*
*
* // Consumer: signed webhook (v2: signature + 5-minute replay window) + exactly-once effects
* // Consumer: signed webhook (v2: signature + 5-minute replay window) + exactly-once effects
- * app.post('/internal/events', express.json({ verify: (req, _res, buf) => { req.rawBody = buf; } }), (req, res) => {
+ * const inbox = events.createPgInbox(db); // table from events.inboxSchema()
+ * app.post('/internal/events', express.json({ verify: (req, _res, buf) => { req.rawBody = buf; } }), async (req, res) => {
* if (!events.verifyDeliveryV2(req.rawBody, req.headers, SECRET)) return res.sendStatus(401);
* if (!events.verifyDeliveryV2(req.rawBody, req.headers, SECRET)) return res.sendStatus(401);
- * inbox.once('live', req.body.event.event_id, () => { ...db writes... });
+ * await inbox.once('live', req.body.event.event_id, async (t) => { ...writes on t... });
* res.sendStatus(204);
* res.sendStatus(204);
* });
* });
*
*
+ * The outbox and inbox are openvibe-sdk's PostgreSQL kits (ADR-042 decision 9): this package no longer carries its
+ * own SQLite copies.
+ *
* `tokenClient` is anything with `authHeaders()` (openvibe-contracts serviceAuth.createTokenClient
* `tokenClient` is anything with `authHeaders()` (openvibe-contracts serviceAuth.createTokenClient
* with audience 'openvibe.events').
* with audience 'openvibe.events').
*/
*/
@@ -181,194 +184,8 @@ function createPublisher({ eventsUrl, tokenClient, fetchImpl = globalThis.fetch,
@@ -181,194 +184,8 @@ function createPublisher({ eventsUrl, tokenClient, fetchImpl = globalThis.fetch,
return { publish, prepare: prepareEnvelope };
return { publish, prepare: prepareEnvelope };
}
}
-// ── Transactional outbox ───────────────────────────────────
-
-/**
- * createOutbox(db, { publisher, table = 'event_outbox', batchSize = 50, intervalMs = 1000, now })
- *
- * ensureSchema() create the outbox table (idempotent)
- * enqueue(envelope) INSIDE the caller's db transaction: the event exists iff the change commits
- * flush() publish due rows once (batch; a rejected batch is retried row by row so one
- * bad row cannot block the rest); resolves { sent, failed, rejected }
- * start() / stop() relay loop (setTimeout chain, never overlapping)
- * pending() count of rows not yet accepted by Events
- *
- * At-least-once: a crash between publish and mark-sent republishes the same event_id, which
- * OpenVibe.Events answers as a duplicate.
- */
-function createOutbox(db, { publisher, table = 'event_outbox', batchSize = 50, intervalMs = 1000, backoffMs = [1000, 5000, 30000, 120000, 600000], now = () => Date.now(), onError = null, allowOutsideTransaction = false } = {}) {
- if (!db || typeof db.prepare !== 'function') throw new TypeError('a better-sqlite3 database is required');
- if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(table)) throw new TypeError('bad outbox table name');
- let stmts = null;
- let timer = null;
- let running = false;
- let flushing = null;
-
- function ensureSchema() {
- db.exec(`CREATE TABLE IF NOT EXISTS ${table} (
- id INTEGER PRIMARY KEY AUTOINCREMENT,
- event_id TEXT NOT NULL UNIQUE,
- envelope TEXT NOT NULL,
- traceparent TEXT,
- created_at INTEGER NOT NULL,
- attempts INTEGER NOT NULL DEFAULT 0,
- next_attempt_at INTEGER NOT NULL DEFAULT 0,
- sent_at INTEGER,
- seq INTEGER,
- rejected_at INTEGER,
- last_error TEXT
- );
- CREATE INDEX IF NOT EXISTS idx_${table}_due ON ${table}(sent_at, rejected_at, next_attempt_at);`);
- stmts = null;
- }
-
- function q() {
- if (!stmts) {
- stmts = {
- insert: db.prepare(`INSERT INTO ${table} (event_id, envelope, traceparent, created_at, next_attempt_at) VALUES (?, ?, ?, ?, 0)`),
- due: db.prepare(`SELECT * FROM ${table} WHERE sent_at IS NULL AND rejected_at IS NULL AND next_attempt_at <= ? ORDER BY id LIMIT ?`),
- sent: db.prepare(`UPDATE ${table} SET sent_at = ?, seq = ?, attempts = attempts + 1, last_error = NULL WHERE id = ?`),
- failed: db.prepare(`UPDATE ${table} SET attempts = attempts + 1, next_attempt_at = ?, last_error = ? WHERE id = ?`),
- rejected: db.prepare(`UPDATE ${table} SET rejected_at = ?, attempts = attempts + 1, last_error = ? WHERE id = ?`),
- pending: db.prepare(`SELECT COUNT(*) AS n FROM ${table} WHERE sent_at IS NULL AND rejected_at IS NULL`),
- prune: db.prepare(`DELETE FROM ${table} WHERE sent_at IS NOT NULL AND sent_at < ?`),
- };
- }
- return stmts;
- }
-
- function enqueue(envelope, { traceparent } = {}) {
- if (!allowOutsideTransaction && !db.inTransaction) {
- throw new Error('outbox.enqueue() must run inside the transaction that makes the change');
- }
- const env = prepareEnvelope(envelope, { traceparent, now: now() });
- q().insert.run(env.event_id, JSON.stringify(env), traceparent || null, now());
- return env;
- }
-
- function backoff(attempts) {
- return backoffMs[Math.min(attempts, backoffMs.length - 1)];
- }
-
- function markFailure(row, err) {
- const message = String((err && err.message) || err).slice(0, 500);
- if (err && err.permanent) q().rejected.run(now(), message, row.id);
- else q().failed.run(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 => JSON.parse(r.envelope));
- try {
- const out = rows.length === 1
- ? { results: [await publisher.publish(envelopes[0], { traceparent: rows[0].traceparent || undefined })] }
- : await publisher.publish(envelopes, { traceparent: rows[0].traceparent || undefined });
- const byId = new Map((out.results || []).map(r => [r.event_id, r]));
- db.transaction(() => { for (const row of rows) q().sent.run(now(), byId.get(row.event_id)?.seq ?? null, row.id); })();
- stats.sent += rows.length;
- } catch (err) {
- if (rows.length > 1 && err.permanent) {
- // One bad envelope rejects the whole batch: isolate it.
- 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) markFailure(row, err);
- if (err.permanent) stats.rejected += rows.length; else stats.failed += rows.length;
- }
- }
- return stats;
- }
+// ── Outbox and inbox: openvibe-sdk's PostgreSQL kits ─────────
- async function doFlush() {
- const total = { sent: 0, failed: 0, rejected: 0 };
- for (;;) {
- const rows = q().due.all(now(), batchSize);
- 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;
- }
-
- /** Publish due rows. Concurrent callers share the flush in progress. */
- 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);
- timer.unref?.();
- }
-
- return {
- ensureSchema,
- enqueue,
- flush,
- start() { if (!running) { running = true; schedule(0); } },
- stop() { running = false; clearTimeout(timer); timer = null; return flushing || Promise.resolve(); },
- /** Wake the relay now (e.g. right after the transaction commits). */
- kick() { schedule(0); },
- pending: () => q().pending.get().n,
- /** Delete sent rows older than `olderThanMs`. */
- prune(olderThanMs = 7 * 24 * 60 * 60 * 1000) { return q().prune.run(now() - olderThanMs).changes; },
- };
-}
-
-// ── Inbox (exactly-once effects) ───────────────────────────
-
-/**
- * createInbox(db, { table = 'idempotency_receipts' })
- * ensureSchema()
- * once(consumer, eventId, fn) -> { duplicate: true } | { duplicate: false, result }
- *
- * fn runs inside a better-sqlite3 transaction together with the receipt insert: either both commit
- * (the effect happened exactly once) or neither does (fn threw; the redelivery will try again).
- * fn must be synchronous and make its effects through this same database.
- */
-function createInbox(db, { table = 'idempotency_receipts', now = () => Date.now() } = {}) {
- if (!db || typeof db.prepare !== 'function') throw new TypeError('a better-sqlite3 database is required');
- if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(table)) throw new TypeError('bad inbox table name');
- let claim = null;
-
- function ensureSchema() {
- db.exec(`CREATE TABLE IF NOT EXISTS ${table} (
- consumer TEXT NOT NULL,
- event_id TEXT NOT NULL,
- processed_at INTEGER NOT NULL,
- PRIMARY KEY (consumer, event_id)
- )`);
- claim = null;
- }
-
- function once(consumer, eventId, fn) {
- if (!consumer || !eventId) throw new TypeError('consumer and eventId are required');
- if (!claim) claim = db.prepare(`INSERT OR IGNORE INTO ${table} (consumer, event_id, processed_at) VALUES (?, ?, ?)`);
- return db.transaction(() => {
- if (claim.run(String(consumer), String(eventId), now()).changes === 0) return { duplicate: true };
- const result = fn();
- if (result && typeof result.then === 'function') {
- throw new TypeError('inbox.once(): fn must be synchronous (it runs inside a SQLite transaction)');
- }
- return { duplicate: false, result };
- })();
- }
-
- function seen(consumer, eventId) {
- return Boolean(db.prepare(`SELECT 1 FROM ${table} WHERE consumer = ? AND event_id = ?`).get(String(consumer), String(eventId)));
- }
-
- return { ensureSchema, once, seen };
-}
+const { createPgOutbox, createPgInbox, outboxSchema, inboxSchema } = require('openvibe-sdk/events');
-module.exports = { sign, verifyDelivery, signV2, verifyDeliveryV2, prepareEnvelope, createPublisher, createOutbox, createInbox, PublishError };
+module.exports = { sign, verifyDelivery, signV2, verifyDeliveryV2, prepareEnvelope, createPublisher, createPgOutbox, createPgInbox, outboxSchema, inboxSchema, PublishError };
@@ -9,7 +9,6 @@
@@ -9,7 +9,6 @@
"version": "0.1.0",
"version": "0.1.0",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
- "better-sqlite3": "^11.10.0",
"dotenv": "^16.4.7",
"dotenv": "^16.4.7",
"express": "^4.21.2",
"express": "^4.21.2",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
@@ -108,7 +107,9 @@
@@ -108,7 +107,9 @@
"url": "https://feross.org/support"
"url": "https://feross.org/support"
}
}
],
],
- "license": "MIT"
+ "license": "MIT",
+ "optional": true,
+ "peer": true
},
},
"node_modules/better-sqlite3": {
"node_modules/better-sqlite3": {
"version": "11.10.0",
"version": "11.10.0",
@@ -116,6 +117,8 @@
@@ -116,6 +117,8 @@
"integrity": "sha512-EwhOpyXiOEL/lKzHz9AW1msWFNzGc/z+LzeB3/jnFJpxu+th2yqvzsSWas1v9jgs9+xiXJcD5A8CJxAG2TaghQ==",
"integrity": "sha512-EwhOpyXiOEL/lKzHz9AW1msWFNzGc/z+LzeB3/jnFJpxu+th2yqvzsSWas1v9jgs9+xiXJcD5A8CJxAG2TaghQ==",
"hasInstallScript": true,
"hasInstallScript": true,
"license": "MIT",
"license": "MIT",
+ "optional": true,
+ "peer": true,
"dependencies": {
"dependencies": {
"bindings": "^1.5.0",
"bindings": "^1.5.0",
"prebuild-install": "^7.1.1"
"prebuild-install": "^7.1.1"
@@ -126,6 +129,8 @@
@@ -126,6 +129,8 @@
"resolved": "https://registry.npmjs.org/bindings/-/bindings-1.5.0.tgz",
"resolved": "https://registry.npmjs.org/bindings/-/bindings-1.5.0.tgz",
"integrity": "sha512-p2q/t/mhvuOj/UeLlV6566GD/guowlr0hHxClI0W9m7MWYkL1F0hLo+0Aexs9HSPCtR1SXQ0TD3MMKrXZajbiQ==",
"integrity": "sha512-p2q/t/mhvuOj/UeLlV6566GD/guowlr0hHxClI0W9m7MWYkL1F0hLo+0Aexs9HSPCtR1SXQ0TD3MMKrXZajbiQ==",
"license": "MIT",
"license": "MIT",
+ "optional": true,
+ "peer": true,
"dependencies": {
"dependencies": {
"file-uri-to-path": "1.0.0"
"file-uri-to-path": "1.0.0"
}
}
@@ -135,6 +140,8 @@
@@ -135,6 +140,8 @@
"resolved": "https://registry.npmjs.org/bl/-/bl-4.1.0.tgz",
"resolved": "https://registry.npmjs.org/bl/-/bl-4.1.0.tgz",
"integrity": "sha512-1W07cM9gS6DcLperZfFSj+bWLtaPGSOHWhPiGzXmvVJbRLdG82sH/Kn8EtW1VqWVA54AKf2h5k5BbnIbwF3h6w==",
"integrity": "sha512-1W07cM9gS6DcLperZfFSj+bWLtaPGSOHWhPiGzXmvVJbRLdG82sH/Kn8EtW1VqWVA54AKf2h5k5BbnIbwF3h6w==",
"license": "MIT",
"license": "MIT",
+ "optional": true,
+ "peer": true,
"dependencies": {
"dependencies": {
"buffer": "^5.5.0",
"buffer": "^5.5.0",
"inherits": "^2.0.4",
"inherits": "^2.0.4",
@@ -184,6 +191,8 @@
@@ -184,6 +191,8 @@
}
}
],
],
"license": "MIT",
"license": "MIT",
+ "optional": true,
+ "peer": true,
"dependencies": {
"dependencies": {
"base64-js": "^1.3.1",
"base64-js": "^1.3.1",
"ieee754": "^1.1.13"
"ieee754": "^1.1.13"
@@ -237,7 +246,9 @@
@@ -237,7 +246,9 @@
"version": "1.1.4",
"version": "1.1.4",
"resolved": "https://registry.npmjs.org/chownr/-/chownr-1.1.4.tgz",
"resolved": "https://registry.npmjs.org/chownr/-/chownr-1.1.4.tgz",
"integrity": "sha512-jJ0bqzaylmJtVnNgzTeSOs8DPavpbYgEr/b0YL8/2GO3xJEhInFmhKMUnEJQjZumK7KXGFhUy89PrsJWlakBVg==",
"integrity": "sha512-jJ0bqzaylmJtVnNgzTeSOs8DPavpbYgEr/b0YL8/2GO3xJEhInFmhKMUnEJQjZumK7KXGFhUy89PrsJWlakBVg==",
- "license": "ISC"
+ "license": "ISC",
+ "optional": true,
+ "peer": true
},
},
"node_modules/cluster-key-slot": {
"node_modules/cluster-key-slot": {
"version": "1.1.2",
"version": "1.1.2",
@@ -298,6 +309,8 @@
@@ -298,6 +309,8 @@
"resolved": "https://registry.npmjs.org/decompress-response/-/decompress-response-6.0.0.tgz",
"resolved": "https://registry.npmjs.org/decompress-response/-/decompress-response-6.0.0.tgz",
"integrity": "sha512-aW35yZM6Bb/4oJlZncMH2LCoZtJXTRxES17vE3hoRiowU2kWHaJKFkSBDnDR+cm9J+9QhXmREyIfv0pji9ejCQ==",
"integrity": "sha512-aW35yZM6Bb/4oJlZncMH2LCoZtJXTRxES17vE3hoRiowU2kWHaJKFkSBDnDR+cm9J+9QhXmREyIfv0pji9ejCQ==",
"license": "MIT",
"license": "MIT",
+ "optional": true,
+ "peer": true,
"dependencies": {
"dependencies": {
"mimic-response": "^3.1.0"
"mimic-response": "^3.1.0"
},
},
@@ -313,6 +326,8 @@
@@ -313,6 +326,8 @@
"resolved": "https://registry.npmjs.org/deep-extend/-/deep-extend-0.6.0.tgz",
"resolved": "https://registry.npmjs.org/deep-extend/-/deep-extend-0.6.0.tgz",
"integrity": "sha512-LOHxIOaPYdHlJRtCQfDIVZtfw/ufM8+rVj649RIHzcm/vGwQRXFt6OPqIFWsm2XEMrNIEtWR64sY1LEKD2vAOA==",
"integrity": "sha512-LOHxIOaPYdHlJRtCQfDIVZtfw/ufM8+rVj649RIHzcm/vGwQRXFt6OPqIFWsm2XEMrNIEtWR64sY1LEKD2vAOA==",
"license": "MIT",
"license": "MIT",
+ "optional": true,
+ "peer": true,
"engines": {
"engines": {
"node": ">=4.0.0"
"node": ">=4.0.0"
}
}
@@ -350,6 +365,8 @@
@@ -350,6 +365,8 @@
"resolved": "https://registry.npmjs.org/detect-libc/-/detect-libc-2.1.2.tgz",
"resolved": "https://registry.npmjs.org/detect-libc/-/detect-libc-2.1.2.tgz",
"integrity": "sha512-Btj2BOOO83o3WyH59e8MgXsxEQVcarkUOpEYrubB0urwnN10yQ364rsiByU11nZlqWYZm05i/of7io4mzihBtQ==",
"integrity": "sha512-Btj2BOOO83o3WyH59e8MgXsxEQVcarkUOpEYrubB0urwnN10yQ364rsiByU11nZlqWYZm05i/of7io4mzihBtQ==",
"license": "Apache-2.0",
"license": "Apache-2.0",
+ "optional": true,
+ "peer": true,
"engines": {
"engines": {
"node": ">=8"
"node": ">=8"
}
}
@@ -409,6 +426,8 @@
@@ -409,6 +426,8 @@
"resolved": "https://registry.npmjs.org/end-of-stream/-/end-of-stream-1.4.5.tgz",
"resolved": "https://registry.npmjs.org/end-of-stream/-/end-of-stream-1.4.5.tgz",
"integrity": "sha512-ooEGc6HP26xXq/N+GCGOT0JKCLDGrq2bQUZrQ7gyrJiZANJ/8YDTxTpQBXGMn+WbIQXNVpyWymm7KYVICQnyOg==",
"integrity": "sha512-ooEGc6HP26xXq/N+GCGOT0JKCLDGrq2bQUZrQ7gyrJiZANJ/8YDTxTpQBXGMn+WbIQXNVpyWymm7KYVICQnyOg==",
"license": "MIT",
"license": "MIT",
+ "optional": true,
+ "peer": true,
"dependencies": {
"dependencies": {
"once": "^1.4.0"
"once": "^1.4.0"
}
}
@@ -463,6 +482,8 @@
@@ -463,6 +482,8 @@
"resolved": "https://registry.npmjs.org/expand-template/-/expand-template-2.0.3.tgz",
"resolved": "https://registry.npmjs.org/expand-template/-/expand-template-2.0.3.tgz",
"integrity": "sha512-XYfuKMvj4O35f/pOXLObndIRvyQ+/+6AhODh+OKWj9S9498pHHn/IMszH+gt0fBCRWMNfk1ZSp5x3AifmnI2vg==",
"integrity": "sha512-XYfuKMvj4O35f/pOXLObndIRvyQ+/+6AhODh+OKWj9S9498pHHn/IMszH+gt0fBCRWMNfk1ZSp5x3AifmnI2vg==",
"license": "(MIT OR WTFPL)",
"license": "(MIT OR WTFPL)",
+ "optional": true,
+ "peer": true,
"engines": {
"engines": {
"node": ">=6"
"node": ">=6"
}
}
@@ -539,7 +560,9 @@
@@ -539,7 +560,9 @@
"version": "1.0.0",
"version": "1.0.0",
"resolved": "https://registry.npmjs.org/file-uri-to-path/-/file-uri-to-path-1.0.0.tgz",
"resolved": "https://registry.npmjs.org/file-uri-to-path/-/file-uri-to-path-1.0.0.tgz",
"integrity": "sha512-0Zt+s3L7Vf1biwWZ29aARiVYLx7iMGnEUl9x33fbB/j3jR81u/O2LbqK+Bm1CDSNDKVtJ/YjwY7TUd5SkeLQLw==",
"integrity": "sha512-0Zt+s3L7Vf1biwWZ29aARiVYLx7iMGnEUl9x33fbB/j3jR81u/O2LbqK+Bm1CDSNDKVtJ/YjwY7TUd5SkeLQLw==",
- "license": "MIT"
+ "license": "MIT",
+ "optional": true,
+ "peer": true
},
},
"node_modules/finalhandler": {
"node_modules/finalhandler": {
"version": "1.3.2",
"version": "1.3.2",
@@ -581,7 +604,9 @@
@@ -581,7 +604,9 @@
"version": "1.0.0",
"version": "1.0.0",
"resolved": "https://registry.npmjs.org/fs-constants/-/fs-constants-1.0.0.tgz",
"resolved": "https://registry.npmjs.org/fs-constants/-/fs-constants-1.0.0.tgz",
"integrity": "sha512-y6OAwoSIf7FyjMIv94u+b5rdheZEjzR63GTyZJm5qh4Bi+2YgwLCcI/fPFZkL5PSixOt6ZNKm+w+Hfp/Bciwow==",
"integrity": "sha512-y6OAwoSIf7FyjMIv94u+b5rdheZEjzR63GTyZJm5qh4Bi+2YgwLCcI/fPFZkL5PSixOt6ZNKm+w+Hfp/Bciwow==",
- "license": "MIT"
+ "license": "MIT",
+ "optional": true,
+ "peer": true
},
},
"node_modules/function-bind": {
"node_modules/function-bind": {
"version": "1.1.2",
"version": "1.1.2",
@@ -633,7 +658,9 @@
@@ -633,7 +658,9 @@
"version": "0.0.0",
"version": "0.0.0",
"resolved": "https://registry.npmjs.org/github-from-package/-/github-from-package-0.0.0.tgz",
"resolved": "https://registry.npmjs.org/github-from-package/-/github-from-package-0.0.0.tgz",
"integrity": "sha512-SyHy3T1v2NUXn29OsWdxmK6RwHD+vkj3v8en8AOBZ1wBQ/hCAQ5bAQTD02kW4W9tUp/3Qh6J8r9EvntiyCmOOw==",
"integrity": "sha512-SyHy3T1v2NUXn29OsWdxmK6RwHD+vkj3v8en8AOBZ1wBQ/hCAQ5bAQTD02kW4W9tUp/3Qh6J8r9EvntiyCmOOw==",
- "license": "MIT"
+ "license": "MIT",
+ "optional": true,
+ "peer": true
},
},
"node_modules/gopd": {
"node_modules/gopd": {
"version": "1.2.0",
"version": "1.2.0",
@@ -721,7 +748,9 @@
@@ -721,7 +748,9 @@
"url": "https://feross.org/support"
"url": "https://feross.org/support"
}
}
],
],
- "license": "BSD-3-Clause"
+ "license": "BSD-3-Clause",
+ "optional": true,
+ "peer": true
},
},
"node_modules/inherits": {
"node_modules/inherits": {
"version": "2.0.4",
"version": "2.0.4",
@@ -733,7 +762,9 @@
@@ -733,7 +762,9 @@
"version": "1.3.8",
"version": "1.3.8",
"resolved": "https://registry.npmjs.org/ini/-/ini-1.3.8.tgz",
"resolved": "https://registry.npmjs.org/ini/-/ini-1.3.8.tgz",
"integrity": "sha512-JV/yugV2uzW5iMRSiZAyDtQd+nxtUnjeLt0acNdw98kKLrvuRVyB80tsREOE7yvGVgalhZ6RNXCmEHkUKBKxew==",
"integrity": "sha512-JV/yugV2uzW5iMRSiZAyDtQd+nxtUnjeLt0acNdw98kKLrvuRVyB80tsREOE7yvGVgalhZ6RNXCmEHkUKBKxew==",
- "license": "ISC"
+ "license": "ISC",
+ "optional": true,
+ "peer": true
},
},
"node_modules/iovalkey": {
"node_modules/iovalkey": {
"version": "0.4.0",
"version": "0.4.0",
@@ -970,6 +1001,8 @@
@@ -970,6 +1001,8 @@
"resolved": "https://registry.npmjs.org/mimic-response/-/mimic-response-3.1.0.tgz",
"resolved": "https://registry.npmjs.org/mimic-response/-/mimic-response-3.1.0.tgz",
"integrity": "sha512-z0yWI+4FDrrweS8Zmt4Ej5HdJmky15+L2e6Wgn3+iK5fWzb6T3fhNFq2+MeTRb064c6Wr4N/wv0DzQTjNzHNGQ==",
"integrity": "sha512-z0yWI+4FDrrweS8Zmt4Ej5HdJmky15+L2e6Wgn3+iK5fWzb6T3fhNFq2+MeTRb064c6Wr4N/wv0DzQTjNzHNGQ==",
"license": "MIT",
"license": "MIT",
+ "optional": true,
+ "peer": true,
"engines": {
"engines": {
"node": ">=10"
"node": ">=10"
},
},
@@ -982,6 +1015,8 @@
@@ -982,6 +1015,8 @@
"resolved": "https://registry.npmjs.org/minimist/-/minimist-1.2.8.tgz",
"resolved": "https://registry.npmjs.org/minimist/-/minimist-1.2.8.tgz",
"integrity": "sha512-2yyAR8qBkN3YuheJanUpWC5U3bb5osDywNB8RzDVlDwDHbocAJveqqj1u8+SVD7jkWT4yvsHCpWqqWqAxb0zCA==",
"integrity": "sha512-2yyAR8qBkN3YuheJanUpWC5U3bb5osDywNB8RzDVlDwDHbocAJveqqj1u8+SVD7jkWT4yvsHCpWqqWqAxb0zCA==",
"license": "MIT",
"license": "MIT",
+ "optional": true,
+ "peer": true,
"funding": {
"funding": {
"url": "https://github.com/sponsors/ljharb"
"url": "https://github.com/sponsors/ljharb"
}
}
@@ -990,7 +1025,9 @@
@@ -990,7 +1025,9 @@
"version": "0.5.3",
"version": "0.5.3",
"resolved": "https://registry.npmjs.org/mkdirp-classic/-/mkdirp-classic-0.5.3.tgz",
"resolved": "https://registry.npmjs.org/mkdirp-classic/-/mkdirp-classic-0.5.3.tgz",
"integrity": "sha512-gKLcREMhtuZRwRAfqP3RFW+TK4JqApVBtOIftVgjuABpAtpxhPGaDcfvbhNvD0B8iD1oUr/txX35NjcaY6Ns/A==",
"integrity": "sha512-gKLcREMhtuZRwRAfqP3RFW+TK4JqApVBtOIftVgjuABpAtpxhPGaDcfvbhNvD0B8iD1oUr/txX35NjcaY6Ns/A==",
- "license": "MIT"
+ "license": "MIT",
+ "optional": true,
+ "peer": true
},
},
"node_modules/ms": {
"node_modules/ms": {
"version": "2.0.0",
"version": "2.0.0",
@@ -1002,7 +1039,9 @@
@@ -1002,7 +1039,9 @@
"version": "2.0.0",
"version": "2.0.0",
"resolved": "https://registry.npmjs.org/napi-build-utils/-/napi-build-utils-2.0.0.tgz",
"resolved": "https://registry.npmjs.org/napi-build-utils/-/napi-build-utils-2.0.0.tgz",
"integrity": "sha512-GEbrYkbfF7MoNaoh2iGG84Mnf/WZfB0GdGEsM8wz7Expx/LlWf5U8t9nvJKXSp3qr5IsEbK04cBGhol/KwOsWA==",
"integrity": "sha512-GEbrYkbfF7MoNaoh2iGG84Mnf/WZfB0GdGEsM8wz7Expx/LlWf5U8t9nvJKXSp3qr5IsEbK04cBGhol/KwOsWA==",
- "license": "MIT"
+ "license": "MIT",
+ "optional": true,
+ "peer": true
},
},
"node_modules/negotiator": {
"node_modules/negotiator": {
"version": "0.6.3",
"version": "0.6.3",
@@ -1018,6 +1057,8 @@
@@ -1018,6 +1057,8 @@
"resolved": "https://registry.npmjs.org/node-abi/-/node-abi-3.96.0.tgz",
"resolved": "https://registry.npmjs.org/node-abi/-/node-abi-3.96.0.tgz",
"integrity": "sha512-rebQ/lz7i0EkoLzUVSrKRzA69zMkwLp95kKMWoMDkkM00Suxz0D7zEQPwRml5fQum24mj7bPvmlgLAmu2JCiYg==",
"integrity": "sha512-rebQ/lz7i0EkoLzUVSrKRzA69zMkwLp95kKMWoMDkkM00Suxz0D7zEQPwRml5fQum24mj7bPvmlgLAmu2JCiYg==",
"license": "MIT",
"license": "MIT",
+ "optional": true,
+ "peer": true,
"dependencies": {
"dependencies": {
"semver": "^7.3.5"
"semver": "^7.3.5"
},
},
@@ -1054,6 +1095,8 @@
@@ -1054,6 +1095,8 @@
"resolved": "https://registry.npmjs.org/once/-/once-1.4.0.tgz",
"resolved": "https://registry.npmjs.org/once/-/once-1.4.0.tgz",
"integrity": "sha512-lNaJgI+2Q5URQBkccEKHTQOPaXdUxnZZElQTZY0MFUAuaEqe1E+Nyvgdz/aIyNi6Z9MzO5dv1H8n58/GELp3+w==",
"integrity": "sha512-lNaJgI+2Q5URQBkccEKHTQOPaXdUxnZZElQTZY0MFUAuaEqe1E+Nyvgdz/aIyNi6Z9MzO5dv1H8n58/GELp3+w==",
"license": "ISC",
"license": "ISC",
+ "optional": true,
+ "peer": true,
"dependencies": {
"dependencies": {
"wrappy": "1"
"wrappy": "1"
}
}
@@ -1261,6 +1304,8 @@
@@ -1261,6 +1304,8 @@
"integrity": "sha512-8Mf2cbV7x1cXPUILADGI3wuhfqWvtiLA1iclTDbFRZkgRQS0NqsPZphna9V+HyTEadheuPmjaJMsbzKQFOzLug==",
"integrity": "sha512-8Mf2cbV7x1cXPUILADGI3wuhfqWvtiLA1iclTDbFRZkgRQS0NqsPZphna9V+HyTEadheuPmjaJMsbzKQFOzLug==",
"deprecated": "No longer maintained. Please contact the author of the relevant native addon; alternatives are available.",
"deprecated": "No longer maintained. Please contact the author of the relevant native addon; alternatives are available.",
"license": "MIT",
"license": "MIT",
+ "optional": true,
+ "peer": true,
"dependencies": {
"dependencies": {
"detect-libc": "^2.0.0",
"detect-libc": "^2.0.0",
"expand-template": "^2.0.3",
"expand-template": "^2.0.3",
@@ -1304,6 +1349,8 @@
@@ -1304,6 +1349,8 @@
"resolved": "https://registry.npmjs.org/pump/-/pump-3.0.4.tgz",
"resolved": "https://registry.npmjs.org/pump/-/pump-3.0.4.tgz",
"integrity": "sha512-VS7sjc6KR7e1ukRFhQSY5LM2uBWAUPiOPa/A3mkKmiMwSmRFUITt0xuj+/lesgnCv+dPIEYlkzrcyXgquIHMcA==",
"integrity": "sha512-VS7sjc6KR7e1ukRFhQSY5LM2uBWAUPiOPa/A3mkKmiMwSmRFUITt0xuj+/lesgnCv+dPIEYlkzrcyXgquIHMcA==",
"license": "MIT",
"license": "MIT",
+ "optional": true,
+ "peer": true,
"dependencies": {
"dependencies": {
"end-of-stream": "^1.1.0",
"end-of-stream": "^1.1.0",
"once": "^1.3.1"
"once": "^1.3.1"
@@ -1354,6 +1401,8 @@
@@ -1354,6 +1401,8 @@
"resolved": "https://registry.npmjs.org/rc/-/rc-1.2.8.tgz",
"resolved": "https://registry.npmjs.org/rc/-/rc-1.2.8.tgz",
"integrity": "sha512-y3bGgqKj3QBdxLbLkomlohkvsA8gdAiUQlSBJnBhfn+BPxg4bc62d8TcBW15wavDfgexCgccckhcZvywyQYPOw==",
"integrity": "sha512-y3bGgqKj3QBdxLbLkomlohkvsA8gdAiUQlSBJnBhfn+BPxg4bc62d8TcBW15wavDfgexCgccckhcZvywyQYPOw==",
"license": "(BSD-2-Clause OR MIT OR Apache-2.0)",
"license": "(BSD-2-Clause OR MIT OR Apache-2.0)",
+ "optional": true,
+ "peer": true,
"dependencies": {
"dependencies": {
"deep-extend": "^0.6.0",
"deep-extend": "^0.6.0",
"ini": "~1.3.0",
"ini": "~1.3.0",
@@ -1369,6 +1418,8 @@
@@ -1369,6 +1418,8 @@
"resolved": "https://registry.npmjs.org/readable-stream/-/readable-stream-3.6.2.tgz",
"resolved": "https://registry.npmjs.org/readable-stream/-/readable-stream-3.6.2.tgz",
"integrity": "sha512-9u/sniCrY3D5WdsERHzHE4G2YCXqoG5FTHUiCC4SIbr6XcLZBY05ya9EKjYek9O5xOAwjGq+1JdGBAS7Q9ScoA==",
"integrity": "sha512-9u/sniCrY3D5WdsERHzHE4G2YCXqoG5FTHUiCC4SIbr6XcLZBY05ya9EKjYek9O5xOAwjGq+1JdGBAS7Q9ScoA==",
"license": "MIT",
"license": "MIT",
+ "optional": true,
+ "peer": true,
"dependencies": {
"dependencies": {
"inherits": "^2.0.3",
"inherits": "^2.0.3",
"string_decoder": "^1.1.1",
"string_decoder": "^1.1.1",
@@ -1587,7 +1638,9 @@
@@ -1587,7 +1638,9 @@
"url": "https://feross.org/support"
"url": "https://feross.org/support"
}
}
],
],
- "license": "MIT"
+ "license": "MIT",
+ "optional": true,
+ "peer": true
},
},
"node_modules/simple-get": {
"node_modules/simple-get": {
"version": "4.0.1",
"version": "4.0.1",
@@ -1608,6 +1661,8 @@
@@ -1608,6 +1661,8 @@
}
}
],
],
"license": "MIT",
"license": "MIT",
+ "optional": true,
+ "peer": true,
"dependencies": {
"dependencies": {
"decompress-response": "^6.0.0",
"decompress-response": "^6.0.0",
"once": "^1.3.1",
"once": "^1.3.1",
@@ -1643,6 +1698,8 @@
@@ -1643,6 +1698,8 @@
"resolved": "https://registry.npmjs.org/string_decoder/-/string_decoder-1.3.0.tgz",
"resolved": "https://registry.npmjs.org/string_decoder/-/string_decoder-1.3.0.tgz",
"integrity": "sha512-hkRX8U1WjJFd8LsDJ2yQ/wWWxaopEsABU1XfkM8A+j0+85JAGppt16cr1Whg6KIbb4okU6Mql6BOj+uup/wKeA==",
"integrity": "sha512-hkRX8U1WjJFd8LsDJ2yQ/wWWxaopEsABU1XfkM8A+j0+85JAGppt16cr1Whg6KIbb4okU6Mql6BOj+uup/wKeA==",
"license": "MIT",
"license": "MIT",
+ "optional": true,
+ "peer": true,
"dependencies": {
"dependencies": {
"safe-buffer": "~5.2.0"
"safe-buffer": "~5.2.0"
}
}
@@ -1652,6 +1709,8 @@
@@ -1652,6 +1709,8 @@
"resolved": "https://registry.npmjs.org/strip-json-comments/-/strip-json-comments-2.0.1.tgz",
"resolved": "https://registry.npmjs.org/strip-json-comments/-/strip-json-comments-2.0.1.tgz",
"integrity": "sha512-4gB8na07fecVVkOI6Rs4e7T6NOTki5EmL7TUduTs6bu3EdnSycntVJ4re8kgZA+wx9IueI2Y11bfbgwtzuE0KQ==",
"integrity": "sha512-4gB8na07fecVVkOI6Rs4e7T6NOTki5EmL7TUduTs6bu3EdnSycntVJ4re8kgZA+wx9IueI2Y11bfbgwtzuE0KQ==",
"license": "MIT",
"license": "MIT",
+ "optional": true,
+ "peer": true,
"engines": {
"engines": {
"node": ">=0.10.0"
"node": ">=0.10.0"
}
}
@@ -1661,6 +1720,8 @@
@@ -1661,6 +1720,8 @@
"resolved": "https://registry.npmjs.org/tar-fs/-/tar-fs-2.1.5.tgz",
"resolved": "https://registry.npmjs.org/tar-fs/-/tar-fs-2.1.5.tgz",
"integrity": "sha512-OboTd8mmMhZDNPV+UjQcK9yKAatXu2aJ+r1w4im1Otd4M4fl2hwvdoXUxIYHFTHWK/3y3FarBP70v3vwmGlOxw==",
"integrity": "sha512-OboTd8mmMhZDNPV+UjQcK9yKAatXu2aJ+r1w4im1Otd4M4fl2hwvdoXUxIYHFTHWK/3y3FarBP70v3vwmGlOxw==",
"license": "MIT",
"license": "MIT",
+ "optional": true,
+ "peer": true,
"dependencies": {
"dependencies": {
"chownr": "^1.1.1",
"chownr": "^1.1.1",
"mkdirp-classic": "^0.5.2",
"mkdirp-classic": "^0.5.2",
@@ -1673,6 +1734,8 @@
@@ -1673,6 +1734,8 @@
"resolved": "https://registry.npmjs.org/tar-stream/-/tar-stream-2.2.0.tgz",
"resolved": "https://registry.npmjs.org/tar-stream/-/tar-stream-2.2.0.tgz",
"integrity": "sha512-ujeqbceABgwMZxEJnk2HDY2DlnUZ+9oEcb1KzTVfYHio0UE6dG71n60d8D2I4qNvleWrrXpmjpt7vZeF1LnMZQ==",
"integrity": "sha512-ujeqbceABgwMZxEJnk2HDY2DlnUZ+9oEcb1KzTVfYHio0UE6dG71n60d8D2I4qNvleWrrXpmjpt7vZeF1LnMZQ==",
"license": "MIT",
"license": "MIT",
+ "optional": true,
+ "peer": true,
"dependencies": {
"dependencies": {
"bl": "^4.0.3",
"bl": "^4.0.3",
"end-of-stream": "^1.4.1",
"end-of-stream": "^1.4.1",
@@ -1698,6 +1761,8 @@
@@ -1698,6 +1761,8 @@
"resolved": "https://registry.npmjs.org/tunnel-agent/-/tunnel-agent-0.6.0.tgz",
"resolved": "https://registry.npmjs.org/tunnel-agent/-/tunnel-agent-0.6.0.tgz",
"integrity": "sha512-McnNiV1l8RYeY8tBgEpuodCC1mLUdbSN+CYBL7kJsJNInOP8UjDDEwdk6Mw60vdLLrr5NHKZhMAOSrR2NZuQ+w==",
"integrity": "sha512-McnNiV1l8RYeY8tBgEpuodCC1mLUdbSN+CYBL7kJsJNInOP8UjDDEwdk6Mw60vdLLrr5NHKZhMAOSrR2NZuQ+w==",
"license": "Apache-2.0",
"license": "Apache-2.0",
+ "optional": true,
+ "peer": true,
"dependencies": {
"dependencies": {
"safe-buffer": "^5.0.1"
"safe-buffer": "^5.0.1"
},
},
@@ -1731,7 +1796,9 @@
@@ -1731,7 +1796,9 @@
"version": "1.0.2",
"version": "1.0.2",
"resolved": "https://registry.npmjs.org/util-deprecate/-/util-deprecate-1.0.2.tgz",
"resolved": "https://registry.npmjs.org/util-deprecate/-/util-deprecate-1.0.2.tgz",
"integrity": "sha512-EPD5q1uXyFxJpCrLnCc1nHnq3gOa6DZBocAIiI2TaSCA7VCJ1UJDMagCzIkXNsUYfD1daK//LTEQ8xiIbrHtcw==",
"integrity": "sha512-EPD5q1uXyFxJpCrLnCc1nHnq3gOa6DZBocAIiI2TaSCA7VCJ1UJDMagCzIkXNsUYfD1daK//LTEQ8xiIbrHtcw==",
- "license": "MIT"
+ "license": "MIT",
+ "optional": true,
+ "peer": true
},
},
"node_modules/utils-merge": {
"node_modules/utils-merge": {
"version": "1.0.1",
"version": "1.0.1",
@@ -1755,7 +1822,9 @@
@@ -1755,7 +1822,9 @@
"version": "1.0.2",
"version": "1.0.2",
"resolved": "https://registry.npmjs.org/wrappy/-/wrappy-1.0.2.tgz",
"resolved": "https://registry.npmjs.org/wrappy/-/wrappy-1.0.2.tgz",
"integrity": "sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ==",
"integrity": "sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ==",
- "license": "ISC"
+ "license": "ISC",
+ "optional": true,
+ "peer": true
},
},
"node_modules/xtend": {
"node_modules/xtend": {
"version": "4.0.2",
"version": "4.0.2",
@@ -21,7 +21,6 @@
@@ -21,7 +21,6 @@
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"repository": "github:OpenVibers/OpenVibe.Events",
"repository": "github:OpenVibers/OpenVibe.Events",
"dependencies": {
"dependencies": {
- "better-sqlite3": "^11.10.0",
"dotenv": "^16.4.7",
"dotenv": "^16.4.7",
"express": "^4.21.2",
"express": "^4.21.2",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
@@ -125,12 +125,12 @@ async function subscriber(respond = () => 204) {
@@ -125,12 +125,12 @@ async function subscriber(respond = () => 204) {
const server = nodeHttp.createServer((req, res) => {
const server = nodeHttp.createServer((req, res) => {
const chunks = [];
const chunks = [];
req.on('data', c => chunks.push(c));
req.on('data', c => chunks.push(c));
- req.on('end', () => {
+ req.on('end', async () => {
const rawBody = Buffer.concat(chunks);
const rawBody = Buffer.concat(chunks);
const call = { headers: req.headers, rawBody, body: JSON.parse(rawBody.toString('utf8') || 'null') };
const call = { headers: req.headers, rawBody, body: JSON.parse(rawBody.toString('utf8') || 'null') };
calls.push(call);
calls.push(call);
let status;
let status;
- try { status = respond(call, calls.length); } catch (err) { status = 500; call.error = err; }
+ try { status = await respond(call, calls.length); } catch (err) { status = 500; call.error = err; } // respond may be async (a consumer's inbox)
if (status === 'destroy') { req.socket.destroy(); return; }
if (status === 'destroy') { req.socket.destroy(); return; }
res.statusCode = status;
res.statusCode = status;
res.end();
res.end();
@@ -1,11 +1,10 @@
@@ -1,11 +1,10 @@
'use strict';
'use strict';
const assert = require('assert');
const assert = require('assert');
-const path = require('path');
const nodeHttp = require('http');
const nodeHttp = require('http');
-const Database = require('better-sqlite3');
+const { createDb, sql } = require('openvibe-sdk/db');
const { serviceAuth } = require('openvibe-contracts');
const { serviceAuth } = require('openvibe-contracts');
const events = require('..');
const events = require('..');
-const { boot, request, serviceToken, envelope, subscriber, suite, tmpDir } = require('./helpers');
+const { boot, request, serviceToken, envelope, subscriber, suite } = require('./helpers');
const t = suite('outbox-inbox');
const t = suite('outbox-inbox');
let h;
let h;
@@ -54,130 +53,132 @@ t('publisher fills event_id, timestamp and trace (from traceparent)', async () =
@@ -54,130 +53,132 @@ t('publisher fills event_id, timestamp and trace (from traceparent)', async () =
await assert.rejects(publisher.publish(envelope('media')), (err) => err.status === 403 && err.code === 'events.source_mismatch' && err.permanent);
await assert.rejects(publisher.publish(envelope('media')), (err) => err.status === 403 && err.code === 'events.source_mismatch' && err.permanent);
});
});
+// The producer's own database: embedded PostgreSQL, like a service's (openvibe-sdk/db).
+async function producerDb(extra = '') {
+ const db = createDb({ pglite: true });
+ await db.query(`${extra}${events.outboxSchema()}${events.inboxSchema()}`);
+ return db;
+}
+
t('outbox: enqueue inside a rolled-back transaction publishes nothing; committed publishes exactly once', async () => {
t('outbox: enqueue inside a rolled-back transaction publishes nothing; committed publishes exactly once', async () => {
- const db = new Database(path.join(tmpDir(), 'producer.db'));
- db.exec('CREATE TABLE vods (id INTEGER PRIMARY KEY, status TEXT)');
- const outbox = events.createOutbox(db, { publisher: events.createPublisher({ eventsUrl: h.base, tokenClient }) });
- outbox.ensureSchema();
-
- assert.throws(() => outbox.enqueue(envelope()), /inside the transaction/);
-
- let rolledBack;
- assert.throws(() => db.transaction(() => {
- db.prepare('INSERT INTO vods (id, status) VALUES (1, ?)').run('ready');
- rolledBack = outbox.enqueue(envelope('live', { event_type: 'live.vod.ready' }));
- throw new Error('domain write failed');
- })(), /domain write failed/);
- assert.strictEqual(outbox.pending(), 0);
- let r = await outbox.flush();
- assert.deepStrictEqual(r, { sent: 0, failed: 0, rejected: 0 });
- assert.strictEqual(await countIn(rolledBack.event_id), 0, 'rolled back: never published');
-
- let committed;
- db.transaction(() => {
- db.prepare('INSERT INTO vods (id, status) VALUES (2, ?)').run('ready');
- committed = outbox.enqueue(envelope('live', { event_type: 'live.vod.ready' }));
- })();
- assert.strictEqual(outbox.pending(), 1);
- r = await outbox.flush();
- assert.strictEqual(r.sent, 1);
- r = await outbox.flush();
- assert.strictEqual(r.sent, 0, 'sent rows are not sent again');
- assert.strictEqual(await countIn(committed.event_id), 1);
-
- // Relay crashed after Events accepted but before the row was marked: at-least-once resend is a duplicate.
- db.prepare('UPDATE event_outbox SET sent_at = NULL WHERE event_id = ?').run(committed.event_id);
- r = await outbox.flush();
- assert.strictEqual(r.sent, 1);
- assert.strictEqual(await countIn(committed.event_id), 1, 'still exactly one event');
- assert.ok(db.prepare('SELECT seq FROM event_outbox WHERE event_id = ?').get(committed.event_id).seq > 0);
- db.close();
+ const db = await producerDb('CREATE TABLE vods (id bigint PRIMARY KEY, status text);');
+ const outbox = events.createPgOutbox(db, { events: events.createPublisher({ eventsUrl: h.base, tokenClient }) });
+ try {
+ await assert.rejects(outbox.enqueue(null, envelope()), /transaction handle/);
+
+ let rolledBack;
+ await assert.rejects(db.tx(async (t) => {
+ await t.exec(sql`INSERT INTO vods (id, status) VALUES (1, 'ready')`);
+ rolledBack = await outbox.enqueue(t, envelope('live', { event_type: 'live.vod.ready' }));
+ throw new Error('domain write failed');
+ }), /domain write failed/);
+ assert.strictEqual(await outbox.pending(), 0);
+ let r = await outbox.flush();
+ assert.deepStrictEqual(r, { sent: 0, failed: 0, rejected: 0 });
+ assert.strictEqual(await countIn(rolledBack.event_id), 0, 'rolled back: never published');
+
+ const committed = await db.tx(async (t) => {
+ await t.exec(sql`INSERT INTO vods (id, status) VALUES (2, 'ready')`);
+ return outbox.enqueue(t, envelope('live', { event_type: 'live.vod.ready' }));
+ });
+ assert.strictEqual(await outbox.pending(), 1);
+ r = await outbox.flush();
+ assert.strictEqual(r.sent, 1);
+ r = await outbox.flush();
+ assert.strictEqual(r.sent, 0, 'sent rows are not sent again');
+ assert.strictEqual(await countIn(committed.event_id), 1);
+
+ // Relay crashed after Events accepted but before the row was marked: at-least-once resend is a duplicate.
+ await db.query('UPDATE event_outbox SET sent_at = NULL, next_attempt_at = 0 WHERE event_id = $1', [committed.event_id]);
+ r = await outbox.flush();
+ assert.strictEqual(r.sent, 1);
+ assert.strictEqual(await countIn(committed.event_id), 1, 'still exactly one event');
+ assert.ok(Number((await db.one('SELECT seq FROM event_outbox WHERE event_id = $1', [committed.event_id])).seq) > 0);
+ } finally { await db.close(); }
});
});
t('outbox: a poisoned row is isolated and rejected; a down Events is retried later', async () => {
t('outbox: a poisoned row is isolated and rejected; a down Events is retried later', async () => {
- const db = new Database(path.join(tmpDir(), 'producer2.db'));
+ const db = await producerDb();
let clock = Date.now();
let clock = Date.now();
- const outbox = events.createOutbox(db, { publisher: events.createPublisher({ eventsUrl: h.base, tokenClient }), now: () => clock });
- outbox.ensureSchema();
- let good; let bad;
- db.transaction(() => {
- good = outbox.enqueue(envelope());
- bad = outbox.enqueue(envelope('media')); // live may not publish as media
- })();
- const r = await outbox.flush();
- assert.deepStrictEqual(r, { sent: 1, failed: 0, rejected: 1 });
- assert.strictEqual(await countIn(good.event_id), 1);
- assert.ok(db.prepare('SELECT rejected_at FROM event_outbox WHERE event_id = ?').get(bad.event_id).rejected_at);
-
- const down = events.createOutbox(db, { publisher: events.createPublisher({ eventsUrl: 'http://127.0.0.1:9', tokenClient }), now: () => clock });
- let later;
- db.transaction(() => { later = down.enqueue(envelope()); })();
- const d = await down.flush();
- assert.deepStrictEqual(d, { sent: 0, failed: 1, rejected: 0 });
- assert.strictEqual(down.pending(), 1);
- const row = db.prepare('SELECT * FROM event_outbox WHERE event_id = ?').get(later.event_id);
- assert.strictEqual(row.next_attempt_at, clock + 1000, 'backoff after the first failure');
- // Events is back: the regular relay picks it up once due.
- clock += 1000;
- assert.strictEqual((await outbox.flush()).sent, 1);
- assert.strictEqual(await countIn(later.event_id), 1);
- db.close();
+ const outbox = events.createPgOutbox(db, { events: events.createPublisher({ eventsUrl: h.base, tokenClient }), now: () => clock });
+ try {
+ const [good, bad] = await db.tx(async (t) => [
+ await outbox.enqueue(t, envelope()),
+ await outbox.enqueue(t, envelope('media')), // live may not publish as media
+ ]);
+ const r = await outbox.flush();
+ assert.deepStrictEqual(r, { sent: 1, failed: 0, rejected: 1 });
+ assert.strictEqual(await countIn(good.event_id), 1);
+ assert.ok((await db.one('SELECT rejected_at FROM event_outbox WHERE event_id = $1', [bad.event_id])).rejected_at);
+
+ const down = events.createPgOutbox(db, { events: events.createPublisher({ eventsUrl: 'http://127.0.0.1:9', tokenClient }), now: () => clock });
+ const later = await db.tx((t) => down.enqueue(t, envelope()));
+ const d = await down.flush();
+ assert.deepStrictEqual(d, { sent: 0, failed: 1, rejected: 0 });
+ assert.strictEqual(await down.pending(), 1);
+ const row = await db.one('SELECT * FROM event_outbox WHERE event_id = $1', [later.event_id]);
+ assert.strictEqual(Number(row.next_attempt_at), clock + 1000, 'backoff after the first failure');
+ // Events is back: the regular relay picks it up once due.
+ clock += 1000;
+ assert.strictEqual((await outbox.flush()).sent, 1);
+ assert.strictEqual(await countIn(later.event_id), 1);
+ } finally { await db.close(); }
});
});
-t('consumer crash mid-processing + redelivery + replay -> exactly one effect (createInbox)', async () => {
- const consumerDb = new Database(path.join(tmpDir(), 'consumer.db'));
- consumerDb.exec('CREATE TABLE credits (id INTEGER PRIMARY KEY AUTOINCREMENT, event_id TEXT, amount INTEGER)');
- const inbox = events.createInbox(consumerDb);
- inbox.ensureSchema();
+t('consumer crash mid-processing + redelivery + replay -> exactly one effect (createPgInbox)', async () => {
+ const consumerDb = await producerDb('CREATE TABLE credits (id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, event_id text, amount integer);');
+ const inbox = events.createPgInbox(consumerDb);
let secret = null;
let secret = null;
let mode = 'crash-after-commit';
let mode = 'crash-after-commit';
- const stub = await subscriber((call) => {
+ const stub = await subscriber(async (call) => {
if (!events.verifyDeliveryV2(call.rawBody, call.headers, secret, { now: h.clock.now() })) return 401;
if (!events.verifyDeliveryV2(call.rawBody, call.headers, secret, { now: h.clock.now() })) return 401;
const ev = call.body.event;
const ev = call.body.event;
if (mode === 'crash-before-commit') {
if (mode === 'crash-before-commit') {
mode = 'ok';
mode = 'ok';
try {
try {
- inbox.once('media', ev.event_id, () => {
- consumerDb.prepare('INSERT INTO credits (event_id, amount) VALUES (?, ?)').run(ev.event_id, ev.payload.amount);
+ await inbox.once('media', ev.event_id, async (t) => {
+ await t.query('INSERT INTO credits (event_id, amount) VALUES ($1, $2)', [ev.event_id, ev.payload.amount]);
throw new Error('process died mid-transaction');
throw new Error('process died mid-transaction');
});
});
} catch { /* the transaction rolled back */ }
} catch { /* the transaction rolled back */ }
return 'destroy';
return 'destroy';
}
}
- inbox.once('media', ev.event_id, () => {
- consumerDb.prepare('INSERT INTO credits (event_id, amount) VALUES (?, ?)').run(ev.event_id, ev.payload.amount);
+ await inbox.once('media', ev.event_id, async (t) => {
+ await t.query('INSERT INTO credits (event_id, amount) VALUES ($1, $2)', [ev.event_id, ev.payload.amount]);
});
});
if (mode === 'crash-after-commit') { mode = 'ok'; return 'destroy'; } // effect committed, no response sent
if (mode === 'crash-after-commit') { mode = 'ok'; return 'destroy'; } // effect committed, no response sent
return 204;
return 204;
});
});
- const media = serviceToken('media', ['events.subscription.manage']);
- const sub = (await request(h.base, 'POST', '/api/v1/subscriptions', { token: media, body: { topic_pattern: 'live.tip.*', endpoint: stub.url } })).body;
- secret = sub.secret;
- const live = serviceToken('live', ['events.event.publish']);
- const effects = (id) => consumerDb.prepare('SELECT COUNT(*) AS n FROM credits WHERE event_id = ?').get(id).n;
-
- for (const first of ['crash-after-commit', 'crash-before-commit']) {
- mode = first;
- const env = envelope('live', { event_type: 'live.tip.sent', payload: { amount: 5 } });
- await request(h.base, 'POST', '/api/v1/events', { token: live, body: env });
- await h.worker.drain();
- const d = await h.store.getDelivery(env.event_id, sub.id);
- assert.strictEqual(d.status, 'failed', `${first}: the crashed attempt is retried`);
- assert.strictEqual(effects(env.event_id), first === 'crash-after-commit' ? 1 : 0);
- h.clock.advance(1000);
- await h.worker.drain();
- assert.strictEqual((await h.store.getDelivery(env.event_id, sub.id)).status, 'delivered');
- assert.strictEqual(effects(env.event_id), 1, `${first}: exactly one effect after redelivery`);
-
- const rp = await request(h.base, 'POST', '/api/v1/deliveries/replay', { token: admin, body: { subscription_id: sub.id, event_ids: [env.event_id] } });
- assert.strictEqual(rp.body.queued, 1);
- await h.worker.drain();
- assert.strictEqual(effects(env.event_id), 1, `${first}: replay does not repeat the effect`);
+ try {
+ const media = serviceToken('media', ['events.subscription.manage']);
+ const sub = (await request(h.base, 'POST', '/api/v1/subscriptions', { token: media, body: { topic_pattern: 'live.tip.*', endpoint: stub.url } })).body;
+ secret = sub.secret;
+ const live = serviceToken('live', ['events.event.publish']);
+ const effects = async (id) => Number((await consumerDb.one('SELECT COUNT(*) AS n FROM credits WHERE event_id = $1', [id])).n);
+
+ for (const first of ['crash-after-commit', 'crash-before-commit']) {
+ mode = first;
+ const env = envelope('live', { event_type: 'live.tip.sent', payload: { amount: 5 } });
+ await request(h.base, 'POST', '/api/v1/events', { token: live, body: env });
+ await h.worker.drain();
+ const d = await h.store.getDelivery(env.event_id, sub.id);
+ assert.strictEqual(d.status, 'failed', `${first}: the crashed attempt is retried`);
+ assert.strictEqual(await effects(env.event_id), first === 'crash-after-commit' ? 1 : 0);
+ h.clock.advance(1000);
+ await h.worker.drain();
+ assert.strictEqual((await h.store.getDelivery(env.event_id, sub.id)).status, 'delivered');
+ assert.strictEqual(await effects(env.event_id), 1, `${first}: exactly one effect after redelivery`);
+
+ const rp = await request(h.base, 'POST', '/api/v1/deliveries/replay', { token: admin, body: { subscription_id: sub.id, event_ids: [env.event_id] } });
+ assert.strictEqual(rp.body.queued, 1);
+ await h.worker.drain();
+ assert.strictEqual(await effects(env.event_id), 1, `${first}: replay does not repeat the effect`);
+ }
+ assert.strictEqual(await inbox.seen('media', 'evt_x'), false);
+ } finally {
+ await stub.close();
+ await consumerDb.close();
}
}
- assert.throws(() => inbox.once('media', 'evt_x', () => Promise.resolve()), /synchronous/);
- assert.strictEqual(inbox.seen('media', 'evt_x'), false, 'an async fn is rolled back, not recorded');
- await stub.close();
- consumerDb.close();
});
});
t('stop', async () => { await h.stop(); network.close(); });
t('stop', async () => { await h.stop(); network.close(); });