openvibers/openvibe.chat Commits

Outbox: openvibe-sdk/events (SDK 0.38.1), and keep the migration order safe (#59)

OP OpenVibers committed 10 Oct 2026 · 381f5a8608 · 25 files, +527 −208

* Outbox: openvibe-sdk/events (SDK 0.38.1), and keep the migration order safe

Plan T1, the kernel: Chat's hand-rolled events outbox (`server/events/outbox.js`, table `events_outbox`, its own relay) gives way to `openvibe-sdk/events` `createServiceOutbox`, on SDK **0.38.1**.

**The outbox:**
- `server/events/service-outbox.js` holds Chat's one SDK outbox. `server/index.js` configures it once at boot and starts and stops it where the relay ran. `emit` builds exactly the envelope the old `enqueue` built (event_id, version 1, source chat, actor, timestamp, priority, visibility) and is called by DMs, rooms and `database.js` inside the change's transaction, so an event exists if and only if its change committed. `eventTypes` is the manifest's six, and every envelope is validated against `events.event-envelope@1`.
- The SDK relay is leased (FOR UPDATE SKIP LOCKED) and rejects events one at a time; the old relay retried a refused batch forever. Production's `chat.env` has `EVENTS_URL` and `OV_OAUTH_CLIENT_SECRET`.
- **Expand migration `0008_sdk_outbox.sql`:** creates `service_outbox` with the SDK's DDL and copies every unsent `events_outbox` row, converting the ISO and `ov_now()` timestamps to milliseconds. `events_outbox` stays for a rollback; a contract migration drops it after the N-1 window.

**Migration order, which needed a fix:** Chat#58's `0008_drop_live_mirror_outbox.sql` (a contract after 0006) is deleted here. It has run nowhere: production holds it until 2026-10-16, and fresh databases never reached it. openvibe-sdk/db keeps order, so a held contract holds every migration after it. Left in place, it would have held this release's outbox table back in production for six days while the code wrote events to it, breaking chat writes. The drop comes back as a new contract migration once 0006's window has passed (it then applies at once and blocks nothing). `pg-schema` and the README say the table stays until then.

**SDK 0.38.1** applies contracts on a fresh database, so Chat's test databases now carry 0005 to 0008. Before, they stopped at 0004 behind the 0005 contract, and 0006/0007 had never run in a test.

`ov test` (files covering the change): 49 passed. The outbox suite covers commit-or-nothing, Events down then up (each event published once, then marked sent) and a row copied by the migration being relayed. `service-outbox-db.test.js` covers the timestamp conversion.

* N-1: re-record from 143e52e (production's schema, 0001-0007; the 10-05 recording predates 0005-0007)

Changes

ddiff --git a/README.md b/README.md
iindex 1f3f994..317a2db 100644
---- a/README.md
++++ b/README.md
@@@ -68,7 +68,8 @@ before changing behaviour.** Browser JavaScript does not change; nginx routes th
`DATABASE_URL` outside production it runs on an embedded PGlite in `data/pglite` (`CHAT_PGLITE_DIR`).
`migrations/0001_initial.sql` is the whole schema (timestamps stay SQLite-format text through
`ov_now()`/`datetime()`); `0006_stop_live_mirror_triggers.sql` dropped the twelve read-mirror capture
-- triggers, and migration 0008 dropped the idle `live_mirror_outbox` table.
++ triggers, and `live_mirror_outbox` stays, idle, until a contract migration drops it (after 0008's
++ SDK outbox: a held contract would hold every migration after it).
`VALKEY_URL`/`VALKEY_PREFIX` put the per-actor rate-limit counters on Valkey instead of this process.
- **The six chat tables** (`channel_moderators`, `channel_moderation_settings`, `emotes`, `user_tags`,
`chat_ai_summaries`, `chat_timeline_events`): Chat's own since the C-04 cutover — Chat is their only
@@@ -120,7 +121,9 @@ before changing behaviour.** Browser JavaScript does not change; nginx routes th
| Emote image bytes | **OpenVibe.Media** (namespace `chat`) | `server/media/client.js` (openvibe-sdk `createObjectsClient`), Chat's service token (`media.object.upload` / `.delete`); the row keeps `media_url` + `media_asset_id` |
| Chat-AI summaries (global / per-chatter insight + timeline) | **OpenVibe.AI** (namespace `chat.*`) | `server/ai/client.js` (openvibe-sdk `createAiClient`), Chat's service token (`ai.run.create` / `ai.run.read`); workflows `chat.global` / `chat.profile`; extractive fallback when AI does not answer |
| Identity | **OpenVibe.Network** | user tokens are resolved by Live (its account links); service tokens from `/oauth/token` |
--| Events | **OpenVibe.Events** | `events_outbox` → `POST /api/v1/events` when `EVENTS_URL` is set (`events.event.publish`); Chat's subscriptions deliver to `POST /internal/events` (`server/events/consumer.js`, `events.subscription.manage`) |
++| Events | **OpenVibe.Events** | `service_outbox` (openvibe-sdk outbox) → `POST /api/v1/events` when `EVENTS_URL` and `OV_OAUTH_CLIENT_SECRET` are set (`events.event.publish`); Chat's subscriptions deliver to `POST /internal/events` (`server/events/consumer.js`, `events.subscription.manage`) |
++
++Migration 0008 copies pending `events_outbox` rows into `service_outbox`. The old table stays through the N-1 rollback window; a later contract migration removes it.
| A person's chat preferences | **OpenVibe.Network** user module `chat.preferences` (Chat owns the namespace) | `server/prefs/` → `GET/PUT/DELETE /internal/modules/chat.preferences/:subject` (`network.modules.read` / `.write`), cached per person |
| Member badges | **OpenVibe.VIP** (Billing holds the entitlement) | `server/vip/badges.js` → `POST /api/v1/entitlements/check` with `product: 'chat'` (`vip.entitlement.check`), behind the shared product cache |
@@@ -427,7 +430,7 @@ server/app.js Live's guards for these routes: CORS, /api rate limit
server/live-context.js the only module that talks to Live (interface in its header)
server/chat/ moved from Live: chat-server, dm, dm-routes, routes, history-store, tts-*, sounds-*, soundboard, moderation-utils, word-filter, deploy-notice; plus the T3 APIs: emotes-routes, channel-mod-routes, internal-moderation (Live's read API), alert-sounds
server/auth/ token resolution through Live; the chat subset of Live's permissions
--server/events/outbox.js events.event-envelope@1 outbox and relay
++server/events/service-outbox.js Chat envelopes and openvibe-sdk outbox wiring
server/events/consumer.js POST /internal/events: Chat's Events subscriptions (live.release.deployed, network.module.updated)
server/events/subscriptions.js creates them at boot when missing; list/disable/enable for scripts/subscribe-events.js
server/net/service-auth.js service tokens: client (Chat → others) and guard (others → Chat)
@@@ -572,6 +575,6 @@ Part of the [OpenVibe network](https://openvibe.network). Built in the open by [
<!-- versions:start -->
- openvibe-contracts: v0.129.0
--- openvibe-sdk: v0.38.0
++- openvibe-sdk: v0.38.1
- openvibe-shared: v3.0.0
<!-- versions:end -->
ddiff --git a/STATUS.json b/STATUS.json
iindex 2dae737..9c73ea6 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -8,7 +8,7 @@
"port": 4400,
"domain": "openvibe.chat (Chat's own site: global chat, messages, rooms, settings; the service manifest records it live since 2026-09-24) and openvibe.live (/ws/chat, /api/chat/, /api/dm/, /api/tts/, /api/sounds via an nginx include in openvibe.live's vhost)",
"contracts": "openvibe-contracts v0.129.0",
-- "sdk": "openvibe-sdk v0.38.0",
++ "sdk": "openvibe-sdk v0.38.1",
"shared": "openvibe-shared v3.0.0",
"capabilities": [
"chat.message.send",
@@@ -63,7 +63,7 @@
"chat preferences in Network user modules",
"VIP subscriber badges",
"network blocks, revocation, account merge, export and deletion from Network events",
-- "outbox events and chat.room.message.* on the realtime plane",
++ "openvibe-sdk service_outbox events and chat.room.message.* on the realtime plane",
"N-1 compatibility tests",
"per-actor limits (openvibe-sdk/limits)"
],
ddiff --git a/migrations/0008_drop_live_mirror_outbox.sql b/migrations/0008_drop_live_mirror_outbox.sql
ddeleted file mode 100644
iindex ae6c105..0000000
---- a/migrations/0008_drop_live_mirror_outbox.sql
++++ /dev/null
@@@ -1,7 +0,0 @@
---- phase: contract
---- after: 0006
----
---- The Live read mirror is retired (2026-10-05, Live #31): 0006_stop_live_mirror_triggers.sql dropped the twelve
---- capture triggers, and nothing has written or read live_mirror_outbox since. Its last rows (frozen before
---- 2026-10-07) never had a drain: they only ever fed Live's mirror, which no longer exists.
--DROP TABLE IF EXISTS live_mirror_outbox;
ddiff --git a/migrations/0008_sdk_outbox.sql b/migrations/0008_sdk_outbox.sql
nnew file mode 100644
iindex 0000000..67763fd
---- /dev/null
++++ b/migrations/0008_sdk_outbox.sql
@@@ -0,0 +1,27 @@
++-- phase: expand
++--
++-- SDK outbox rows use a JSONB envelope and millisecond timestamps. Keep events_outbox through the
++-- N-1 window; a later contract migration drops it after rollback no longer needs the old relay.
++CREATE TABLE IF NOT EXISTS service_outbox (
++ 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 service_outbox_due ON service_outbox (next_attempt_at, id) WHERE sent_at IS NULL AND rejected_at IS NULL;
++CREATE INDEX IF NOT EXISTS service_outbox_sent ON service_outbox (sent_at) WHERE sent_at IS NOT NULL;
++
++-- The old writer stores ISO 8601 UTC. Accept ov_now()'s UTC SQLite format as well.
++INSERT INTO service_outbox (event_id, envelope, created_at, attempts)
++SELECT event_id, event::jsonb,
++ COALESCE((EXTRACT(EPOCH FROM ov_ts(created_at) AT TIME ZONE 'UTC') * 1000)::bigint, 0),
++ attempts
++FROM events_outbox WHERE sent_at IS NULL
++ON CONFLICT (event_id) DO NOTHING;
ddiff --git a/package-lock.json b/package-lock.json
iindex 992e1b6..5c021b7 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -17,7 +17,7 @@
"iovalkey": "^0.4.0",
"multer": "^2.0.2",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.129.0",
-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.38.0",
++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.38.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v3.0.0",
"pg": "^8.23.0",
"ws": "^8.18.0"
@@@ -1036,9 +1036,9 @@
}
},
"node_modules/openvibe-sdk": {
-- "version": "0.38.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.38.0",
-- "integrity": "sha512-/Z+1kB5l4Hff0E1O1Zo+DcXaZX93wrmISOPP75sB7NqVUZVeKuJgjYu6vBwG8ufJjlk0vD3ZJ0ntBLI/3IKdpg==",
++ "version": "0.38.1",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.38.1",
++ "integrity": "sha512-Pn3nji9Nc+4rMRjhwO9AhhR80OW7cwLo/NLObiVR/4TlcOxyk/V8DLzGHQmFjNxqg0xLBj6MOiXYQR2qAxEycg==",
"license": "MIT",
"engines": {
"node": ">=20"
ddiff --git a/package.json b/package.json
iindex 84ed817..b86dfae 100644
---- a/package.json
++++ b/package.json
@@@ -29,7 +29,7 @@
"iovalkey": "^0.4.0",
"multer": "^2.0.2",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.129.0",
-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.38.0",
++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.38.1",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v3.0.0",
"pg": "^8.23.0",
"ws": "^8.18.0"
ddiff --git a/server/app.js b/server/app.js
iindex 8b55829..baf09c4 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -167,7 +167,7 @@ function createApp({ chatServer, relay, events = null, callServer = null }) {
connections: chatServer.getTotalConnections(),
live: { last_sync_at: ctx.stats.lastSyncAt, last_success_at: ctx.stats.lastSuccessAt, failures: ctx.stats.failures, last_error: ctx.stats.lastError },
events: {
-- enabled: !!config.events.url,
++ ...(relay ? await relay.status().catch((err) => ({ enabled: relay.enabled, pending: null, rejected: null, last_error: err.message })) : { enabled: false, pending: 0, rejected: 0, last_error: null }),
consumer: events ? { enabled: events.enabled, ...events.stats } : null,
},
});
ddiff --git a/server/chat/dm.js b/server/chat/dm.js
iindex bd8dbdf..2e26457 100644
---- a/server/chat/dm.js
++++ b/server/chat/dm.js
@@@ -17,7 +17,7 @@
* isBlockedEither is asked: new conversations, group invites, 1:1 messages, calls.
*/
const db = require('../db/database');
--const outbox = require('../events/outbox');
++const outbox = require('../events/service-outbox');
const networkBlocks = require('./network-blocks');
// ── Schema & Migrations ──────────────────────────────────────
@@@ -176,7 +176,7 @@ async function sendMessage(conversationId, senderId, text) {
await markRead(conversationId, senderId);
// The event names who is in the conversation, never what was said.
const participants = await db.all('SELECT user_id, subject_id FROM dm_participants WHERE conversation_id = ?', [conversationId]);
-- await outbox.enqueue({
++ await outbox.emit({
event_type: 'chat.dm.created',
visibility: 'subject',
actorSubject: senderSubject,
ddiff --git a/server/config.js b/server/config.js
iindex 03a49e1..88d690e 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -120,7 +120,7 @@ module.exports = {
maxEntries: int(process.env.CHAT_PREFS_CACHE_MAX, 5000),
},
-- // OpenVibe.Events — the outbox relays only when EVENTS_URL is set.
++ // OpenVibe.Events — the SDK outbox relays only with EVENTS_URL and the OAuth client secret.
events: {
url: strip(process.env.EVENTS_URL || ''),
intervalMs: int(process.env.EVENTS_RELAY_INTERVAL_MS, 5000),
ddiff --git a/server/db/database.js b/server/db/database.js
iindex 2eb113c..6329552 100644
---- a/server/db/database.js
++++ b/server/db/database.js
@@@ -8,7 +8,7 @@
* server/live-context.js maintains, never from Live's database.
* - New rows record the author's Network subject (subject_id columns) next to the Live id.
* - Writes that are events (a chat message, a DM, a moderation action) add their envelope to
-- * events_outbox in the same transaction (server/events/outbox.js).
++ * service_outbox in the same transaction (server/events/service-outbox.js).
*/
'use strict';
@@@ -137,7 +137,7 @@ async function subjectFor(userId) {
try { return (await get('SELECT subject_id FROM ctx_users WHERE id = ?', [userId]))?.subject_id || null; } catch { return null; }
}
--function _outbox() { return require('../events/outbox'); }
++function _outbox() { return require('../events/service-outbox'); }
function _ctx() { return require('../live-context'); }
// Ids per chat.message.deleted event (OpenVibe.Events takes up to 1000 per redaction directive).
@@@ -153,7 +153,7 @@ async function _announceDeleted(ids) {
const list = [...new Set((ids || []).map(Number).filter((n) => Number.isInteger(n) && n > 0))];
for (let i = 0; i < list.length; i += DELETED_EVENT_IDS) {
const part = list.slice(i, i + DELETED_EVENT_IDS);
-- await _outbox().enqueue({
++ await _outbox().emit({
event_type: 'chat.message.deleted',
visibility: 'public',
subject: { type: 'chat_message', id: String(part[0]) },
@@@ -196,7 +196,7 @@ async function saveChatMessage({ stream_id, channel_user_id, user_id, anon_id, u
[stream_id, chanUid, user_id || null, anon_id || null, username, message, message_type || 'chat', is_global ? 1 : 0, reply_to_id || null, source_platform || null, auto_delete_at || null, metaStr, subject]
);
res.first_chat = !!(chatterKey && chanUid) && await recordFirstChat(chatterKey, chanUid);
-- await _outbox().enqueue({
++ await _outbox().emit({
event_type: 'chat.message.created',
visibility: 'public',
actorSubject: subject,
@@@ -674,7 +674,7 @@ async function logModerationAction({ scope_type, scope_id, actor_user_id, target
INSERT INTO moderation_actions (scope_type, scope_id, actor_user_id, target_user_id, action_type, details, actor_subject_id)
VALUES (?, ?, ?, ?, ?, ?, ?)
`, [scope_type || 'site', scope_id || null, actor_user_id || null, target_user_id || null, action_type, JSON.stringify(details || {}), actorSubject]);
-- await _outbox().enqueue({
++ await _outbox().emit({
event_type: 'chat.moderation.action',
visibility: 'internal',
actorSubject,
ddiff --git a/server/events/outbox.js b/server/events/outbox.js
ddeleted file mode 100644
iindex 7eacb0a..0000000
---- a/server/events/outbox.js
++++ /dev/null
@@@ -1,119 +0,0 @@
--/**
-- * Transactional outbox (events.event-envelope@1).
-- *
-- * enqueue() runs inside the same SQLite transaction as the change it announces (a chat message,
-- * a DM, a moderation action), so an event exists exactly when its effect does:
-- * chat.message.created visibility public — a message in a public room (global/channel/stream)
-- * chat.message.deleted visibility public — ids of deleted messages (moderation, the author,
-- * a purge, auto-delete); payload.redacts makes
-- * OpenVibe.Events tombstone their chat.message.created
-- * chat.dm.created visibility subject — a DM; the payload names participants, never the text
-- * chat.moderation.action visibility internal — ban/timeout/clear/slow/deletes, from chat or Live
-- *
-- * The relay publishes unsent rows to OpenVibe.Events (POST /api/v1/events, service token for
-- * audience openvibe.events with events.event.publish) and runs only when EVENTS_URL is set;
-- * without it rows accumulate and are relayed later. Events deduplicates on event_id, so a retry
-- * after a lost response never publishes twice.
-- */
--'use strict';
--
--const { ids, validate, serviceAuth } = require('openvibe-contracts');
--
--const SOURCE = 'chat';
--const SERVICE_ACTOR = { type: 'service', id: SOURCE };
--const USER_SUBJECT = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
--
--function actorFor(subjectId) {
-- return subjectId && USER_SUBJECT.test(subjectId) ? { type: 'user', id: subjectId } : SERVICE_ACTOR;
--}
--
--/** Add one envelope to the outbox. Call inside the transaction that makes the change. */
--async function enqueue({ event_type, subject, payload, visibility = 'internal', priority = 'important', actorSubject = null }) {
-- const db = require('../db/database');
-- const ms = Date.now();
-- const env = {
-- event_id: ids.newId('event', ms),
-- event_type,
-- version: 1,
-- source: SOURCE,
-- actor: actorFor(actorSubject),
-- timestamp: new Date(ms).toISOString(),
-- priority,
-- visibility,
-- subject,
-- payload: payload || {},
-- };
-- const v = validate('events.event-envelope@1', env);
-- if (!v.valid) throw new Error(`outbox: invalid envelope for ${event_type}: ${v.errors.map((e) => `${e.path} ${e.message}`).join('; ')}`);
-- await db.run('INSERT INTO events_outbox (event_id, event_type, event, created_at) VALUES (?, ?, ?, ?)', [env.event_id, event_type, JSON.stringify(env), env.timestamp]);
-- return env;
--}
--
--/** Relay loop. Returns { start, stop, flush }. */
--function createRelay({ config, fetchImpl = globalThis.fetch, tokenClient, log = console } = {}) {
-- const db = require('../db/database');
-- const tokens = tokenClient || serviceAuth.createTokenClient({
-- tokenUrl: `${config.networkInternalUrl}/oauth/token`,
-- clientId: config.oauth.clientId,
-- clientSecret: config.oauth.clientSecret,
-- audience: 'openvibe.events',
-- scope: 'events.event.publish',
-- fetchImpl,
-- });
-- let timer = null;
-- let busy = false;
--
-- async function markFailed(rows, message) {
-- const mark = db.getDb().prepare('UPDATE events_outbox SET attempts = attempts + 1, last_error = ? WHERE seq = ?');
-- await db.tx(async () => { for (const r of rows) await mark.run(String(message).slice(0, 500), r.seq); });
-- log.warn(`[Chat] outbox relay: ${rows.length} event(s) not published: ${message}`);
-- }
--
-- async function flush() {
-- if (!config.events.url) return { sent: 0, disabled: true };
-- if (busy) return { sent: 0, busy: true };
-- busy = true;
-- let sent = 0;
-- try {
-- for (;;) {
-- const rows = await db.all('SELECT seq, event_id, event FROM events_outbox WHERE sent_at IS NULL ORDER BY seq LIMIT 100');
-- if (!rows.length) break;
-- let res;
-- try {
-- res = await fetchImpl(`${config.events.url}/api/v1/events`, {
-- method: 'POST',
-- headers: { 'Content-Type': 'application/json', ...(await tokens.authHeaders()) },
-- body: JSON.stringify({ events: rows.map((r) => JSON.parse(r.event)) }),
-- signal: AbortSignal.timeout(10000),
-- });
-- } catch (e) {
-- await markFailed(rows, e.message);
-- break;
-- }
-- if (res.status === 401) tokens.invalidate();
-- if (!res.ok) {
-- const text = await res.text().catch(() => '');
-- await markFailed(rows, `${res.status} ${text.slice(0, 300)}`);
-- break;
-- }
-- const at = new Date().toISOString();
-- const mark = db.getDb().prepare('UPDATE events_outbox SET sent_at = ?, attempts = attempts + 1, last_error = NULL WHERE seq = ?');
-- await db.tx(async () => { for (const r of rows) await mark.run(at, r.seq); });
-- sent += rows.length;
-- }
-- } finally {
-- busy = false;
-- }
-- return { sent };
-- }
--
-- function start() {
-- if (timer || !config.events.url) return;
-- timer = setInterval(() => { flush().catch((e) => log.warn('[Chat] outbox relay:', e.message)); }, config.events.intervalMs);
-- if (timer.unref) timer.unref();
-- }
-- function stop() { if (timer) clearInterval(timer); timer = null; }
-- return { start, stop, flush };
--}
--
--module.exports = { enqueue, createRelay, actorFor };
ddiff --git a/server/events/service-outbox.js b/server/events/service-outbox.js
nnew file mode 100644
iindex 0000000..b8be3fa
---- /dev/null
++++ b/server/events/service-outbox.js
@@@ -0,0 +1,53 @@
++/**
++ * Chat's one SDK events outbox. Producers call emit inside the transaction that makes the change.
++ */
++'use strict';
++
++const { ids, validate } = require('openvibe-contracts');
++const { createServiceOutbox } = require('openvibe-sdk/events');
++const db = require('../db/database');
++
++const SOURCE = 'chat';
++const SERVICE_ACTOR = { type: 'service', id: SOURCE };
++const USER_SUBJECT = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
++const EVENT_TYPES = [
++ 'chat.message.created', 'chat.message.deleted', 'chat.dm.created',
++ 'chat.moderation.action', 'chat.room.message.created', 'chat.room.message.deleted',
++];
++let instance = null;
++
++function actorFor(subjectId) {
++ return subjectId && USER_SUBJECT.test(subjectId) ? { type: 'user', id: subjectId } : SERVICE_ACTOR;
++}
++
++function configure({ config, log = console }) {
++ if (instance) throw new Error('service outbox is already configured');
++ instance = createServiceOutbox({
++ db: db.getDb(), source: SOURCE, eventsUrl: config.events.url,
++ networkInternalUrl: config.networkInternalUrl,
++ clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret,
++ intervalMs: config.events.intervalMs, log, table: 'service_outbox',
++ eventTypes: EVENT_TYPES,
++ validate: (env) => validate('events.event-envelope@1', env),
++ });
++ return instance;
++}
++
++function current() {
++ if (!instance) throw new Error('service outbox is not configured');
++ return instance;
++}
++
++/** Preserve Chat's envelope fields and defaults across the SDK migration. */
++async function emit({ event_type, subject, payload, visibility = 'internal', priority = 'important', actorSubject = null }) {
++ const ms = Date.now();
++ return current().emit({
++ event_id: ids.newId('event', ms), event_type, version: 1, source: SOURCE,
++ actor: actorFor(actorSubject), timestamp: new Date(ms).toISOString(),
++ priority, visibility, subject, payload: payload || {},
++ });
++}
++
++function reset() { instance = null; }
++
++module.exports = { configure, current, emit, reset, actorFor };
ddiff --git a/server/index.js b/server/index.js
iindex 92bfa31..762dbf9 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -17,7 +17,7 @@ const db = require('./db/database');
const ctx = require('./live-context');
const chatServer = require('./chat/chat-server');
const callServer = require('./calls/call-server');
--const { createRelay } = require('./events/outbox');
++const serviceOutbox = require('./events/service-outbox');
const { createEventsConsumer } = require('./events/consumer');
const subscriptions = require('./events/subscriptions');
const chatAi = require('./ai/chat-ai');
@@@ -53,7 +53,7 @@ async function start() {
const server = http.createServer();
chatServer.init(server);
callServer.init(server);
-- const relay = createRelay({ config });
++ const relay = serviceOutbox.configure({ config });
const events = createEventsConsumer({ chatServer, secrets: config.events.secrets });
const { app, handleUpgrade } = createApp({ chatServer, relay, events, callServer });
server.on('request', app);
@@@ -86,7 +86,7 @@ async function start() {
timestamp: new Date().toISOString(),
}),
() => ctx.stop(),
-- () => relay.stop(),
++ async () => { await relay.stop(); serviceOutbox.reset(); },
() => events.stop(),
() => chatAi.stop(),
() => subs.stop(),
ddiff --git a/server/rooms/rooms.js b/server/rooms/rooms.js
iindex c832aa1..d888002 100644
---- a/server/rooms/rooms.js
++++ b/server/rooms/rooms.js
@@@ -213,7 +213,7 @@ function rateOk(key, now = Date.now()) {
const REDACT_CHUNK = 500;
async function announceMessage(room, user, id, message, now) {
if (room.visibility !== 'public') return;
-- await require('../events/outbox').enqueue({
++ await require('../events/service-outbox').emit({
event_type: 'chat.room.message.created',
visibility: 'public',
actorSubject: user.subject_id || null,
@@@ -233,7 +233,7 @@ async function announceDeleted(room, ids) {
const list = [...new Set(ids.map(Number).filter((n) => Number.isInteger(n) && n > 0))];
for (let i = 0; i < list.length; i += REDACT_CHUNK) {
const part = list.slice(i, i + REDACT_CHUNK);
-- await require('../events/outbox').enqueue({
++ await require('../events/service-outbox').emit({
event_type: 'chat.room.message.deleted',
visibility: 'public',
subject: { type: 'chat_room_message', id: String(part[0]) },
ddiff --git a/test/fixtures/n-1/client.json b/test/fixtures/n-1/client.json
iindex 415a93b..1d72eb6 100644
---- a/test/fixtures/n-1/client.json
++++ b/test/fixtures/n-1/client.json
@@@ -1,32 +1,32 @@
{
"service": "chat",
-- "release": "360bb4ebcaa8906d47187623d447effd18bade95",
-- "recorded_at": "2026-10-05T18:52:56.885Z",
++ "release": "143e52e371b2a659c0a407635f60fd256a4e619a",
++ "recorded_at": "2026-10-10T16:02:05.083Z",
"note": "N-1 fixture: written by `npm run n-1:record`, replayed by test/n-1.test.js",
"clients": {
-- "live": "ee2ad948a4c25b9af9c27be16e731025afa6c2d3"
++ "live": "92f003c2a75833950b875e8c314a59b55e9d9502"
},
"manifest": {
"service": "chat",
-- "release": "360bb4ebcaa8",
-- "released_at": "2026-10-05T18:18:22.000Z",
-- "booted_at": "2026-10-05T18:52:51.633Z",
-- "contracts_version": "0.96.0",
++ "release": "143e52e371b2",
++ "released_at": "2026-10-10T13:50:03.000Z",
++ "booted_at": "2026-10-10T16:01:59.563Z",
++ "contracts_version": "0.129.0",
"packages": {
-- "openvibe-shared": "2.5.0",
-- "openvibe-sdk": "0.26.0",
-- "openvibe-contracts": "0.96.0"
++ "openvibe-shared": "3.0.0",
++ "openvibe-sdk": "0.38.1",
++ "openvibe-contracts": "0.129.0"
},
"min_client_release": null,
"mixed_version_window_hours": 24,
"components": {
"shell": {
"kind": "script",
-- "version": "360bb4ebcaa8"
++ "version": "143e52e371b2"
},
"server": {
"kind": "server",
-- "version": "360bb4ebcaa8"
++ "version": "143e52e371b2"
}
},
"assets": {},
@@@ -67,6 +67,60 @@
"OpenVibe.Live:public/js/dashboard.js"
]
},
++ {
++ "method": "GET",
++ "path": "/api/chat/ai/global",
++ "auth": "anon",
++ "status": 200,
++ "kind": "json",
++ "reads": {
++ "insight": "null"
++ },
++ "from": [
++ "OpenVibe.Live:public/js/app-chatpage.js"
++ ]
++ },
++ {
++ "method": "GET",
++ "path": "/api/chat/ai/global",
++ "auth": "user",
++ "status": 200,
++ "kind": "json",
++ "reads": {
++ "insight": "null"
++ },
++ "from": [
++ "OpenVibe.Live:public/js/app-chatpage.js"
++ ]
++ },
++ {
++ "method": "GET",
++ "path": "/api/chat/ai/timeline",
++ "auth": "anon",
++ "status": 200,
++ "kind": "json",
++ "reads": {
++ "events": "array",
++ "hasMore": "boolean"
++ },
++ "from": [
++ "OpenVibe.Live:public/js/app-chatpage.js"
++ ]
++ },
++ {
++ "method": "GET",
++ "path": "/api/chat/ai/timeline",
++ "auth": "user",
++ "status": 200,
++ "kind": "json",
++ "reads": {
++ "events": "array",
++ "hasMore": "boolean"
++ },
++ "from": [
++ "OpenVibe.Live:public/js/app-chatpage.js"
++ ]
++ },
{
"method": "GET",
"path": "/api/chat/anon/n1fan",
@@@ -173,6 +227,117 @@
"OpenVibe.Live:public/js/chat.js"
]
},
++ {
++ "method": "GET",
++ "path": "/api/chat/channels/moderation/mine",
++ "auth": "anon",
++ "status": 401,
++ "kind": "json",
++ "from": [
++ "OpenVibe.Live:public/js/app-channel.js",
++ "OpenVibe.Live:public/js/channel-uploads.js",
++ "OpenVibe.Live:public/js/dashboard-moderation.js"
++ ]
++ },
++ {
++ "method": "GET",
++ "path": "/api/chat/channels/moderation/mine",
++ "auth": "user",
++ "status": 200,
++ "kind": "json",
++ "reads": {
++ "channels": "array",
++ "channels[]": "object",
++ "channels[].id": "number",
++ "channels[].user_id": "number",
++ "channels[].title": "string",
++ "channels[].moderation_settings": "object",
++ "channels[].moderation_settings.channel_id": "number",
++ "channels[].moderation_settings.slow_mode_seconds": "number",
++ "channels[].moderation_settings.followers_only": "number",
++ "channels[].moderation_settings.allow_anonymous": "number",
++ "channels[].moderation_settings.links_allowed": "number",
++ "channels[].moderation_settings.gifs_enabled": "number",
++ "channels[].moderation_settings.account_age_gate_hours": "number",
++ "channels[].moderation_settings.caps_percentage_limit": "number",
++ "channels[].moderation_settings.aggressive_filter": "number",
++ "channels[].moderation_settings.max_message_length": "number",
++ "channels[].moderation_settings.tts_max_length": "number",
++ "channels[].moderation_settings.slur_filter_enabled": "number",
++ "channels[].moderation_settings.slur_filter_use_builtin": "number",
++ "channels[].moderation_settings.slur_filter_terms": "string",
++ "channels[].moderation_settings.slur_filter_regexes": "string",
++ "channels[].moderation_settings.slur_filter_nudge_message": "string",
++ "channels[].moderation_settings.slur_filter_disabled_categories": "string",
++ "channels[].moderation_settings.ip_approval_mode": "number",
++ "channels[].moderation_settings.soundboard_enabled": "number",
++ "channels[].moderation_settings.soundboard_allow_pitch": "number",
++ "channels[].moderation_settings.soundboard_allow_speed": "number",
++ "channels[].moderation_settings.soundboard_banned_ids": "string",
++ "channels[].moderation_settings.viewer_auto_delete_enabled": "number",
++ "channels[].moderation_settings.viewer_delete_all_enabled": "number",
++ "channels[].moderation_settings.custom_emotes_enabled": "number",
++ "channels[].moderation_settings.custom_sounds_enabled": "number",
++ "channels[].moderation_settings.max_sound_seconds": "number",
++ "channels[].moderation_settings.uploads_mods_only": "number",
++ "channels[].moderation_settings.mods_can_edit_about": "number",
++ "channels[].moderation_settings.emote_scale": "number",
++ "channels[].moderation_settings.emote_size_min": "number",
++ "channels[].moderation_settings.emote_size_max": "number",
++ "channels[].moderation_settings.sounds_mods_only": "number",
++ "channels[].moderation_settings.sound_min_speed": "number",
++ "channels[].moderation_settings.sound_max_speed": "number",
++ "channels[].moderation_settings.sound_min_pitch_cents": "number",
++ "channels[].moderation_settings.sound_max_pitch_cents": "number",
++ "channels[].moderation_settings.sub_only": "number",
++ "channels[].moderators": "array"
++ },
++ "from": [
++ "OpenVibe.Live:public/js/app-channel.js",
++ "OpenVibe.Live:public/js/channel-uploads.js",
++ "OpenVibe.Live:public/js/dashboard-moderation.js"
++ ]
++ },
++ {
++ "method": "GET",
++ "path": "/api/chat/channels/1/moderation/chat-search",
++ "auth": "anon",
++ "status": 401,
++ "kind": "json",
++ "from": [
++ "OpenVibe.Live:public/js/dashboard-moderation.js"
++ ]
++ },
++ {
++ "method": "GET",
++ "path": "/api/chat/channels/1/moderation/chat-search",
++ "auth": "user",
++ "status": 404,
++ "kind": "json",
++ "from": [
++ "OpenVibe.Live:public/js/dashboard-moderation.js"
++ ]
++ },
++ {
++ "method": "GET",
++ "path": "/api/chat/channels/1/moderation/logs?limit=12",
++ "auth": "anon",
++ "status": 401,
++ "kind": "json",
++ "from": [
++ "OpenVibe.Live:public/js/dashboard-moderation.js"
++ ]
++ },
++ {
++ "method": "GET",
++ "path": "/api/chat/channels/1/moderation/logs?limit=12",
++ "auth": "user",
++ "status": 404,
++ "kind": "json",
++ "from": [
++ "OpenVibe.Live:public/js/dashboard-moderation.js"
++ ]
++ },
{
"method": "GET",
"path": "/api/chat/filters/friendly",
@@@ -1374,6 +1539,17 @@
"page /"
]
},
++ {
++ "method": "GET",
++ "path": "/shared/items.css?v=0705cd3fd62e",
++ "auth": "anon",
++ "accept": "html",
++ "status": 200,
++ "kind": "text",
++ "from": [
++ "page /"
++ ]
++ },
{
"method": "GET",
"path": "/shared/theme-loader.js?v=ee4a6754d0f7",
@@@ -1387,7 +1563,7 @@
},
{
"method": "GET",
-- "path": "/shared/navbar.js?v=017a740747b4",
++ "path": "/shared/navbar.js?v=8a1438c1d7b3",
"auth": "anon",
"accept": "html",
"status": 200,
@@@ -1398,7 +1574,7 @@
},
{
"method": "GET",
-- "path": "/shared/footer.js?v=a85ab893be88",
++ "path": "/shared/footer.js?v=0eafb8254aa5",
"auth": "anon",
"accept": "html",
"status": 200,
@@@ -1409,7 +1585,7 @@
},
{
"method": "GET",
-- "path": "/web/chat.js?v=64ef4815d2",
++ "path": "/web/chat.js?v=ea6b3ee678",
"auth": "anon",
"accept": "html",
"status": 200,
@@@ -1547,7 +1723,7 @@
},
{
"method": "GET",
-- "path": "/shared/shipped.js?v=18e4e7ae4ec3",
++ "path": "/shared/shipped.js?v=bed628bacc75",
"auth": "anon",
"accept": "html",
"status": 200,
@@@ -1576,6 +1752,46 @@
"OpenVibe.Live:public/js/dashboard.js"
]
},
++ {
++ "method": "PUT",
++ "path": "/api/chat/channels/1/moderation",
++ "auth": "anon",
++ "status": 401,
++ "kind": "json",
++ "from": [
++ "OpenVibe.Live:public/js/app-channel.js"
++ ]
++ },
++ {
++ "method": "POST",
++ "path": "/api/chat/channels/1/moderation/messages/1/delete",
++ "auth": "anon",
++ "status": 401,
++ "kind": "json",
++ "from": [
++ "OpenVibe.Live:public/js/dashboard-moderation.js"
++ ]
++ },
++ {
++ "method": "POST",
++ "path": "/api/chat/channels/1/mods",
++ "auth": "anon",
++ "status": 401,
++ "kind": "json",
++ "from": [
++ "OpenVibe.Live:public/js/dashboard-moderation.js"
++ ]
++ },
++ {
++ "method": "DELETE",
++ "path": "/api/chat/channels/1/mods/1002",
++ "auth": "anon",
++ "status": 401,
++ "kind": "json",
++ "from": [
++ "OpenVibe.Live:public/js/dashboard-moderation.js"
++ ]
++ },
{
"method": "POST",
"path": "/api/chat/rooms",
ddiff --git a/test/fixtures/n-1/worker.json b/test/fixtures/n-1/worker.json
iindex 936e625..58b1fab 100644
---- a/test/fixtures/n-1/worker.json
++++ b/test/fixtures/n-1/worker.json
@@@ -1,7 +1,7 @@
{
"service": "chat",
-- "release": "360bb4ebcaa8906d47187623d447effd18bade95",
-- "recorded_at": "2026-10-05T18:52:56.885Z",
++ "release": "143e52e371b2a659c0a407635f60fd256a4e619a",
++ "recorded_at": "2026-10-10T16:02:05.083Z",
"note": "N-1 fixture: written by `npm run n-1:record`, replayed by test/n-1.test.js",
"engine": "postgresql",
"migrations": [
@@@ -20,6 +20,18 @@
{
"file": "0004_channel_sound_media.sql",
"sql": "-- phase: expand\n-- Where Live's asset sync put each channel sound on OpenVibe.Media (plan T3 J4b, N5): Live reads the\n-- sounds still to upload (GET /internal/chat/sounds?pending_asset=1) and writes the asset back\n-- (POST /internal/chat/sounds/asset), instead of using its mirrored copy of channel_sounds.\n-- Both stay NULL until the sound is uploaded.\nALTER TABLE channel_sounds ADD COLUMN IF NOT EXISTS media_url text COLLATE \"C\";\nALTER TABLE channel_sounds ADD COLUMN IF NOT EXISTS media_asset_id bigint;\nCREATE INDEX IF NOT EXISTS idx_channel_sounds_pending_asset ON channel_sounds(id) WHERE media_asset_id IS NULL;\n"
++ },
++ {
++ "file": "0005_drop_bridge_tables.sql",
++ "sql": "-- phase: contract\n-- after: 0001\n--\n-- The Live chat bridge is retired (T3 J3a): Live #20/#21 route every producer through Chat's own\n-- ingress (`/internal/chat/*`, docs/chat-ingress.md), Chat no longer mounts the bridge routes, and\n-- nothing reads or writes bridge_applied (the bridge's idempotency ledger for Live's outbox\n-- deliveries) or bridge_refs (its map from a Live boot's placeholder ids to the real chat_messages\n-- ids) any more. docs/cutover.md recorded the cutover with them kept as data until this migration\n-- (J3b). deploy_releases.bridge_at / first_via stay as data: they record how a release was applied.\nDROP TABLE IF EXISTS bridge_applied;\nDROP TABLE IF EXISTS bridge_refs;\n"
++ },
++ {
++ "file": "0006_stop_live_mirror_triggers.sql",
++ "sql": "-- phase: expand\n--\n-- The Live read mirror is retired (2026-10-05): Live #31 reads every chat table through Chat's internal reads\n-- (docs/chat-ingress.md) and this release deletes the sender (server/bridge/live-mirror.js). The capture\n-- triggers 0001_initial.sql put on twelve tables would otherwise keep queueing a row in live_mirror_outbox for\n-- every chat write with nothing left to drain it. Dropping them is safe for the previous release too: its sender\n-- only ever fed Live's mirror, which nothing reads any more. The (now idle) live_mirror_outbox table itself\n-- is dropped by a later contract migration.\nDROP TRIGGER IF EXISTS mirror_chat_messages ON chat_messages;\nDROP TRIGGER IF EXISTS mirror_dm_conversations ON dm_conversations;\nDROP TRIGGER IF EXISTS mirror_dm_participants ON dm_participants;\nDROP TRIGGER IF EXISTS mirror_dm_messages ON dm_messages;\nDROP TRIGGER IF EXISTS mirror_dm_blocks ON dm_blocks;\nDROP TRIGGER IF EXISTS mirror_tts_voice_overrides ON tts_voice_overrides;\nDROP TRIGGER IF EXISTS mirror_channel_sounds ON channel_sounds;\nDROP TRIGGER IF EXISTS mirror_relay_users ON relay_users;\nDROP TRIGGER IF EXISTS mirror_hidden_relay_users ON hidden_relay_users;\nDROP TRIGGER IF EXISTS mirror_pending_ip_messages ON pending_ip_messages;\nDROP TRIGGER IF EXISTS mirror_stream_first_chats ON stream_first_chats;\nDROP TRIGGER IF EXISTS mirror_moderation_actions ON moderation_actions;\n\nDROP FUNCTION IF EXISTS mirror_chat_messages();\nDROP FUNCTION IF EXISTS mirror_dm_conversations();\nDROP FUNCTION IF EXISTS mirror_dm_participants();\nDROP FUNCTION IF EXISTS mirror_dm_messages();\nDROP FUNCTION IF EXISTS mirror_dm_blocks();\nDROP FUNCTION IF EXISTS mirror_tts_voice_overrides();\nDROP FUNCTION IF EXISTS mirror_channel_sounds();\nDROP FUNCTION IF EXISTS mirror_relay_users();\nDROP FUNCTION IF EXISTS mirror_hidden_relay_users();\nDROP FUNCTION IF EXISTS mirror_pending_ip_messages();\nDROP FUNCTION IF EXISTS mirror_stream_first_chats();\nDROP FUNCTION IF EXISTS mirror_moderation_actions();\n"
++ },
++ {
++ "file": "0007_ticket_conversations.sql",
++ "sql": "-- phase: expand\n-- OpenVibe.Help's per-ticket conversation (capability chat.ticket.write,\n-- chat.ticket-conversation@1, added in openvibe-contracts 0.100.0). One conversation per\n-- (ticket id, calling service); two services on one ticket get two conversations and neither\n-- can read the other's. Messages carry the author's kind, subject and body (plain text, never\n-- HTML) and the author's own created_at; Chat orders by its own id (recorded_at is when it\n-- arrived). Service tokens are loopback-only and never granted to an app or an agent.\nCREATE TABLE IF NOT EXISTS ticket_conversations (\n id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,\n ticket_id text COLLATE \"C\" NOT NULL,\n service text COLLATE \"C\" NOT NULL,\n created_at text COLLATE \"C\" NOT NULL DEFAULT ov_now(),\n UNIQUE (ticket_id, service)\n);\nCREATE TABLE IF NOT EXISTS ticket_messages (\n id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,\n conversation_id bigint NOT NULL REFERENCES ticket_conversations(id) ON DELETE CASCADE,\n author_kind text COLLATE \"C\" NOT NULL CHECK (author_kind IN ('person', 'agent', 'staff')),\n author text COLLATE \"C\" NOT NULL,\n body text COLLATE \"C\" NOT NULL, -- the author's text, as sent; Chat never renders it\n created_at text COLLATE \"C\" NOT NULL, -- the author's clock (contract), not Chat's ordering\n recorded_at text COLLATE \"C\" NOT NULL DEFAULT ov_now()\n);\nCREATE INDEX IF NOT EXISTS idx_ticket_messages_conversation ON ticket_messages(conversation_id, id);\n"
}
],
"ledger": {
@@@ -53,8 +65,6 @@
"DELETE FROM emotes WHERE user_id = ANY($1)",
"DELETE FROM hidden_relay_users WHERE id = $1",
"DELETE FROM hidden_relay_users WHERE platform = $1 AND external_username = $2 AND channel_id IS NOT DISTINCT FROM $3",
-- "DELETE FROM live_mirror_outbox",
-- "DELETE FROM live_mirror_outbox WHERE seq <= $1",
"DELETE FROM network_blocks WHERE blocked_subject = $1 AND blocker_subject = $2",
"DELETE FROM network_blocks WHERE blocker_subject = $1 AND blocked_subject = $2",
"DELETE FROM network_blocks WHERE blocker_subject = ANY($1) OR blocked_subject = ANY($1)",
@@@ -111,6 +121,8 @@
"INSERT INTO rooms (slug, name, topic, kind, visibility, owner_id, owner_subject, join_role) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)",
"INSERT INTO rooms (slug, name, topic, kind, visibility, owner_id, owner_subject, join_role) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING id",
"INSERT INTO stream_first_chats (chatter_key, channel_user_id) VALUES ($1, $2) ON CONFLICT DO NOTHING",
++ "INSERT INTO ticket_conversations (ticket_id, service) VALUES ($1, $2) ON CONFLICT DO NOTHING",
++ "INSERT INTO ticket_messages (conversation_id, author_kind, author, body, created_at) VALUES ($1, $2, $3, $4, $5)",
"INSERT INTO token_revocations (subject_id, valid_after_ms, reason, updated_at) VALUES ($1, $2, $3, $4) ON CONFLICT(subject_id) DO UPDATE SET valid_after_ms = excluded.valid_after_ms, reason = excluded.reason, updated_at = excluded.updated_at WHERE excluded.valid_after_ms > token_revocations.valid_after_ms",
"INSERT INTO tts_voice_overrides (identity_key, voice, pitch, speed, gap, set_by, updated_at) VALUES ($1, $2, $3, $4, $5, $6, ov_now()) ON CONFLICT(identity_key) DO UPDATE SET voice=excluded.voice, pitch=excluded.pitch, speed=excluded.speed, gap=excluded.gap, set_by=excluded.set_by, updated_at=ov_now()",
"INSERT INTO user_tags (user_id, tag_id, source) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING",
@@@ -141,18 +153,13 @@
"SELECT * FROM ctx_users",
"SELECT * FROM ctx_users WHERE id = $1",
"SELECT * FROM ctx_users WHERE subject_id = $1 LIMIT 2",
-- "SELECT * FROM dm_blocks WHERE id = $1",
"SELECT * FROM dm_conversations WHERE id = $1",
"SELECT * FROM dm_messages WHERE id = $1",
-- "SELECT * FROM dm_participants WHERE id = $1",
"SELECT * FROM emotes WHERE code = $1 AND ((channel_owner_id = $2) OR (channel_owner_id IS NULL AND user_id = $3)) LIMIT 1",
"SELECT * FROM emotes WHERE id = $1",
"SELECT * FROM emotes WHERE user_id = $1 ORDER BY code",
-- "SELECT * FROM hidden_relay_users WHERE id = $1",
-- "SELECT * FROM moderation_actions WHERE id = $1",
"SELECT * FROM pending_ip_messages WHERE channel_id = $1 AND status = 'pending' ORDER BY created_at ASC, id ASC LIMIT $2",
"SELECT * FROM pending_ip_messages WHERE id = $1",
-- "SELECT * FROM relay_users WHERE id = $1",
"SELECT * FROM relay_users WHERE platform = $1 AND username = $2",
"SELECT * FROM room_attachments WHERE room_id = $1 AND service = $2 AND resource = $3",
"SELECT * FROM room_attachments WHERE room_id = $1 ORDER BY created_at, service, resource",
@@@ -160,10 +167,10 @@
"SELECT * FROM rooms WHERE id = $1",
"SELECT * FROM rooms WHERE slug = $1 AND archived_at IS NULL",
"SELECT * FROM rooms WHERE visibility = 'public' AND archived_at IS NULL ORDER BY COALESCE(last_message_at, created_at) DESC LIMIT $1",
-- "SELECT * FROM stream_first_chats WHERE chatter_key = $1 AND channel_user_id = $2",
"SELECT * FROM tts_voice_overrides WHERE identity_key = $1",
"SELECT * FROM user_tags WHERE user_id = $1 AND tag_id = $2",
"SELECT 1 AS one FROM audio_requests WHERE room = $1 AND state = 'playing'",
++ "SELECT 1 AS present FROM stream_first_chats WHERE chatter_key = $1 AND channel_user_id = $2",
"SELECT 1 FROM audio_requests WHERE room = $1 AND dedupe_key = $2",
"SELECT 1 FROM channel_moderation_settings WHERE channel_id = $1",
"SELECT 1 FROM channel_moderators WHERE user_id = $1 AND channel_id = $2",
@@@ -188,7 +195,6 @@
"SELECT COUNT(*) AS n FROM audio_requests WHERE room = $1 AND identity_key = $2 AND state IN ('queued', 'playing')",
"SELECT COUNT(*) AS n FROM audio_requests WHERE room = $1 AND state IN ('queued', 'playing')",
"SELECT COUNT(*) AS n FROM channel_sounds WHERE channel_owner_id = $1",
-- "SELECT COUNT(*) AS n FROM live_mirror_outbox",
"SELECT COUNT(*) AS n FROM moderation_actions WHERE actor_subject_id = ANY($1) OR target_user_id = ANY($2) OR actor_user_id = ANY($2)",
"SELECT COUNT(*) AS n FROM room_members WHERE room_id = $1 AND role != 'blocked'",
"SELECT COUNT(*) AS n FROM room_members m JOIN rooms r ON r.id = m.room_id AND r.archived_at IS NULL JOIN room_messages x ON x.room_id = m.room_id AND x.id > m.last_read_id AND x.is_deleted = 0 AND x.user_id != m.user_id WHERE m.user_id = $1 AND m.role != 'blocked'",
@@@ -205,6 +211,7 @@
"SELECT MAX(id) AS id FROM chat_messages",
"SELECT MAX(id) AS id FROM room_messages WHERE room_id = $1",
"SELECT MAX(id) AS m FROM chat_messages",
++ "SELECT MAX(id) AS max_id FROM ticket_messages WHERE conversation_id = $1",
"SELECT MIN(timestamp) AS t FROM chat_messages WHERE anon_id = $1 AND user_id IS NULL AND is_deleted = 0",
"SELECT blocked_subject AS o FROM network_blocks WHERE blocker_subject = $1",
"SELECT blocker_subject AS o FROM network_blocks WHERE blocked_subject = $1",
@@@ -252,10 +259,12 @@
"SELECT id FROM user_tags WHERE user_id = $1 AND tag_id = $2",
"SELECT id, blocked_id AS o FROM dm_blocks WHERE blocker_id = $1",
"SELECT id, blocker_id AS o FROM dm_blocks WHERE blocked_id = $1",
++ "SELECT id, created_at FROM ticket_conversations WHERE ticket_id = $1 AND service = $2",
"SELECT id, message_type, metadata FROM chat_messages WHERE is_deleted = 0 ORDER BY id DESC LIMIT 1",
"SELECT id, state FROM calls WHERE state IN ('pending', 'ringing', 'active')",
"SELECT id, stream_id FROM chat_messages WHERE is_deleted = 0 AND auto_delete_at IS NOT NULL AND auto_delete_at <= ov_now() ORDER BY auto_delete_at ASC LIMIT $1",
"SELECT id, stream_id, channel_user_id FROM chat_messages WHERE id = $1 AND is_deleted = 0",
++ "SELECT id, ts, label, detail FROM chat_timeline_events WHERE scope = $1 AND subject_id = $2 ORDER BY ts DESC, id DESC LIMIT $3",
"SELECT id, username, display_name, avatar_url, profile_color FROM ctx_users WHERE id != $1 AND is_banned = 0 AND id NOT IN (SELECT blocked_id FROM dm_blocks WHERE blocker_id = $2) AND id NOT IN (SELECT blocker_id FROM dm_blocks WHERE blocked_id = $3) AND COALESCE(subject_id, '') NOT IN (SELECT blocked_subject FROM network_blocks WHERE blocker_subject = $4 AND active = 1) AND COALESCE(subject_id, '') NOT IN (SELECT blocker_subject FROM network_blocks WHERE blocked_subject = $5 AND active = 1) AND (username ILIKE $6 OR display_name ILIKE $7) LIMIT $8",
"SELECT m.*, u.username, u.display_name, u.avatar_url, u.profile_color FROM dm_messages m JOIN ctx_users u ON u.id = m.sender_id WHERE m.conversation_id = $1 AND m.id < $2 ORDER BY m.created_at DESC LIMIT $3",
"SELECT m.*, u.username, u.display_name, u.avatar_url, u.profile_color FROM dm_messages m JOIN ctx_users u ON u.id = m.sender_id WHERE m.conversation_id = $1 AND m.id > $2 ORDER BY m.created_at ASC LIMIT $3",
@@@ -278,7 +287,7 @@
"SELECT s.*, u.username, u.display_name, u.avatar_url, u.profile_color, ms.slug AS managed_stream_slug, ms.title AS managed_stream_title FROM ctx_streams s LEFT JOIN ctx_users u ON s.user_id = u.id LEFT JOIN ctx_managed_streams ms ON s.managed_stream_id = ms.id WHERE s.user_id = $1 ORDER BY s.created_at DESC LIMIT $2",
"SELECT sent_at FROM account_data_events WHERE id = $1",
"SELECT seq, event_id, event FROM events_outbox WHERE sent_at IS NULL ORDER BY seq LIMIT 100",
-- "SELECT seq, tbl, op, pk FROM live_mirror_outbox ORDER BY seq LIMIT $1",
++ "SELECT slug, name, visibility, owner_subject, created_at FROM rooms WHERE archived_at IS NULL AND slug > $1 COLLATE \"C\" ORDER BY slug COLLATE \"C\" LIMIT $2",
"SELECT subject_id FROM ctx_users WHERE id = $1",
"SELECT table_name FROM information_schema.columns WHERE table_schema = current_schema() AND column_name = 'id'",
"SELECT u.id FROM network_blocks b JOIN ctx_users u ON u.subject_id = b.blocked_subject WHERE b.blocker_subject = $1 AND b.active = 1",
ddiff --git a/test/message-deleted.test.js b/test/message-deleted.test.js
iindex b9e7b32..2b1c63f 100644
---- a/test/message-deleted.test.js
++++ b/test/message-deleted.test.js
@@@ -26,10 +26,10 @@ let h, events, eventsDb, eventsPort, admin, alice, streamer, streamId;
/** Outbox chat.message.deleted envelopes after `afterSeq`. */
async function deletions(afterSeq = 0) {
-- return (await h.db.all("SELECT seq, event FROM events_outbox WHERE event_type = 'chat.message.deleted' AND seq > ? ORDER BY seq", [afterSeq]))
-- .map((r) => JSON.parse(r.event));
++ return (await h.db.all("SELECT id, envelope FROM service_outbox WHERE envelope->>'event_type' = 'chat.message.deleted' AND id > ? ORDER BY id", [afterSeq]))
++ .map((r) => r.envelope);
}
--const outboxSeq = async () => (await h.db.get('SELECT COALESCE(MAX(seq), 0) AS s FROM events_outbox')).s;
++const outboxSeq = async () => (await h.db.get('SELECT COALESCE(MAX(id), 0) AS s FROM service_outbox')).s;
async function say(message, over = {}) {
return Number((await h.db.saveChatMessage({ stream_id: null, user_id: null, anon_id: 'anon7', username: 'anon7', message, message_type: 'chat', is_global: 1, ...over })).lastInsertRowid);
}
@@@ -207,7 +207,7 @@ const userJwt = (subjectId) => serviceAuth.signServiceToken({
let firstSeq, msgId;
t('e2e: the relay publishes the message, then its deletion; replay serves only the tombstone', async () => {
if (!haveEvents) return;
-- await h.db.run('UPDATE events_outbox SET sent_at = ? WHERE sent_at IS NULL', [new Date().toISOString()]); // earlier tests' rows
++ await h.db.run('UPDATE service_outbox SET sent_at = ? WHERE sent_at IS NULL', [Date.now()]); // earlier tests' rows
msgId = await say(SECRET, { user_id: alice.id, anon_id: null, username: 'alice' });
assert.strictEqual((await h.eventsRelay.flush()).sent, 1);
firstSeq = await events.store.lastSeq();
ddiff --git a/test/outbox.test.js b/test/outbox.test.js
iindex 17b76ad..152878d 100644
---- a/test/outbox.test.js
++++ b/test/outbox.test.js
@@@ -6,6 +6,9 @@
*/
const assert = require('assert');
const http = require('http');
++const fs = require('fs');
++const path = require('path');
++const { ids } = require('openvibe-contracts');
const { validate } = require('openvibe-contracts');
const { boot, suite } = require('./helpers');
@@@ -25,13 +28,20 @@ t('boot with a stub OpenVibe.Events', async () => {
const r = serviceAuth.verifyServiceToken(String(req.headers.authorization || '').slice(7), { publicKey: h.keys.publicKey, issuer: h.ISS, audience: 'openvibe.events' });
if (!r.ok || !r.claims.cap.includes('events.event.publish')) { res.statusCode = 401; return res.end('{}'); }
const body = JSON.parse(raw);
-- received.push(...body.events);
++ const batch = Array.isArray(body.events) ? body.events : [body];
++ received.push(...batch);
res.statusCode = 201;
-- res.end(JSON.stringify({ accepted: body.events.length }));
++ res.end(JSON.stringify(Array.isArray(body.events) ? { results: batch.map((e, i) => ({ event_id: e.event_id, seq: i + 1 })) } : { event_id: body.event_id, seq: received.length }));
});
});
eventsPort = await new Promise((r) => events.listen(0, '127.0.0.1', () => r(events.address().port)));
h = await boot({ env: { EVENTS_URL: `http://127.0.0.1:${eventsPort}`, EVENTS_RELAY_INTERVAL_MS: '3600000' } });
++ await h.eventsRelay.stop();
++ const status = await h.eventsRelay.status();
++ assert.strictEqual(status.enabled, true);
++ assert.strictEqual(Number(status.pending), 0);
++ assert.strictEqual(Number(status.rejected), 0);
++ assert.strictEqual(status.last_error, null);
alice = h.addUser('alice', { subject: 'usr_01J9AAAAAAAAAAAAAAAAAAAAAA' });
bob = h.addUser('bob');
await h.ctx.sync();
@@@ -43,15 +53,15 @@ t('a REST fallback message, a DM and a moderation action each leave one envelope
const conv = await h.http('POST', '/api/dm/conversations', { token: alice.token, body: { user_ids: [bob.id] } });
await h.http('POST', `/api/dm/conversations/${conv.body.conversation.id}/messages`, { token: alice.token, body: { message: 'dm text' } });
await h.db.logModerationAction({ scope_type: 'site', actor_user_id: alice.id, target_user_id: bob.id, action_type: 'site_timeout', details: { duration: 60 } });
-- const rows = await h.db.all('SELECT event_type, event, sent_at FROM events_outbox ORDER BY seq');
-- assert.deepStrictEqual(rows.map((r) => r.event_type), ['chat.message.created', 'chat.dm.created', 'chat.moderation.action']);
++ const rows = await h.db.all('SELECT envelope, sent_at FROM service_outbox ORDER BY id');
++ assert.deepStrictEqual(rows.map((r) => r.envelope.event_type), ['chat.message.created', 'chat.dm.created', 'chat.moderation.action']);
for (const r of rows) {
-- const env = JSON.parse(r.event);
++ const env = r.envelope;
const v = validate('events.event-envelope@1', env);
assert.ok(v.valid, JSON.stringify(v.errors));
assert.strictEqual(r.sent_at, null);
}
-- const [msg, dm, mod] = rows.map((r) => JSON.parse(r.event));
++ const [msg, dm, mod] = rows.map((r) => r.envelope);
assert.strictEqual(msg.visibility, 'public');
assert.deepStrictEqual(msg.payload.room, { type: 'global' });
assert.strictEqual(msg.payload.text, 'rest hello');
@@@ -64,18 +74,50 @@ t('a REST fallback message, a DM and a moderation action each leave one envelope
assert.deepStrictEqual(dm.actor, { type: 'user', id: 'usr_01J9AAAAAAAAAAAAAAAAAAAAAA' });
});
++t('an event and its change roll back together', async () => {
++ const before = (await h.db.get('SELECT COUNT(*) AS n FROM service_outbox')).n;
++ await assert.rejects(h.db.transaction(async () => {
++ await h.db.run('INSERT INTO chat_meta (key, value) VALUES (?, ?)', ['outbox-rollback', 'changed']);
++ await require('../server/events/service-outbox').emit({
++ event_type: 'chat.moderation.action',
++ subject: { type: 'moderation_action', id: 'rollback' },
++ payload: { action: 'test' },
++ });
++ throw new Error('roll back');
++ }), /roll back/);
++ assert.strictEqual((await h.db.get('SELECT COUNT(*) AS n FROM service_outbox')).n, before);
++ assert.ok(!(await h.db.get('SELECT value FROM chat_meta WHERE key = ?', ['outbox-rollback'])));
++});
++
t('Events down: rows wait with the error; back up: published once, marked sent', async () => {
eventsMode = 'down';
-- await h.eventsRelay.flush();
-- const waiting = await h.db.all('SELECT attempts, last_error, sent_at FROM events_outbox');
-- assert.ok(waiting.every((r) => r.attempts === 1 && /503/.test(r.last_error) && r.sent_at === null));
++ await h.eventsRelay.outbox.flush();
++ const waiting = await h.db.all('SELECT attempts, last_error, sent_at FROM service_outbox');
++ assert.ok(waiting.every((r) => r.attempts >= 1 && r.last_error && r.sent_at === null));
++ await h.db.run('UPDATE service_outbox SET next_attempt_at = 0 WHERE sent_at IS NULL');
eventsMode = 'ok';
-- const r = await h.eventsRelay.flush();
++ const r = await h.eventsRelay.outbox.flush();
assert.strictEqual(r.sent, 3);
assert.strictEqual(received.length, 3);
-- assert.ok((await h.db.all('SELECT sent_at FROM events_outbox')).every((x) => x.sent_at));
-- assert.strictEqual((await h.eventsRelay.flush()).sent, 0, 'nothing twice');
++ assert.ok((await h.db.all('SELECT sent_at FROM service_outbox')).every((x) => x.sent_at));
++ assert.strictEqual((await h.eventsRelay.outbox.flush()).sent, 0, 'nothing twice');
assert.ok(h.tokenRequests.some((x) => x.audience === 'openvibe.events' && x.scope === 'events.event.publish'));
});
++t('a pending row copied by the migration is relayed by the SDK', async () => {
++ const original = (await h.db.get('SELECT envelope FROM service_outbox ORDER BY id LIMIT 1')).envelope;
++ const envelope = { ...original, event_id: ids.newId('event') };
++ await h.db.run('INSERT INTO events_outbox (event_id, event_type, event, created_at) VALUES (?, ?, ?, ?)',
++ [envelope.event_id, envelope.event_type, JSON.stringify(envelope), envelope.timestamp]);
++ // The harness ran the migration as the owner; re-run its copy step, the only DML in it (the runtime role has
++ // no CREATE on the CI containers' schema).
++ const migration = fs.readFileSync(path.join(__dirname, '..', 'migrations', '0008_sdk_outbox.sql'), 'utf8');
++ await h.db.getDb().query(migration.slice(migration.indexOf('INSERT INTO service_outbox')));
++ const copied = await h.db.get('SELECT envelope, created_at FROM service_outbox WHERE event_id = ?', [envelope.event_id]);
++ assert.deepStrictEqual(copied.envelope, envelope);
++ assert.strictEqual(Number(copied.created_at), Date.parse(envelope.timestamp));
++ assert.strictEqual((await h.eventsRelay.outbox.flush()).sent, 1);
++ assert.ok(received.some((e) => e.event_id === envelope.event_id));
++});
++
t.run(async () => { if (h) await h.close(); events.close(); });
ddiff --git a/test/pg-schema.test.js b/test/pg-schema.test.js
iindex dd45fa5..80276a7 100644
---- a/test/pg-schema.test.js
++++ b/test/pg-schema.test.js
@@@ -3,10 +3,11 @@
* The PostgreSQL schema (plan T3, decision 1). migrations/0001_initial.sql opens a real PostgreSQL
* (PGlite in-process, or the containers with OV_TEST_STORE=pg) and is checked here: every Chat table
* exists, the keys/indexes the queries rely on are present, and the retired Live read mirror captures
-- * nothing any more (0006 dropped its twelve triggers; 0008 dropped live_mirror_outbox).
++ * nothing any more (0006 dropped its twelve triggers; live_mirror_outbox stays until a contract migration).
*/
const assert = require('assert');
const path = require('path');
++const fs = require('fs');
const { createTestDb } = require('openvibe-sdk/testing');
const { createDb } = require('openvibe-sdk/db');
const { suite } = require('./helpers');
@@@ -24,14 +25,14 @@ const TABLES = [
'chat_messages', 'dm_conversations', 'dm_participants', 'dm_messages', 'dm_blocks', 'tts_voice_overrides',
'channel_sounds', 'relay_users', 'hidden_relay_users', 'pending_ip_messages', 'stream_first_chats', 'moderation_actions',
'channel_moderators', 'channel_moderation_settings', 'emotes', 'user_tags', 'chat_ai_summaries', 'chat_timeline_events',
-- 'ctx_users', 'ctx_streams', 'ctx_managed_streams', 'ctx_channels', 'ctx_sync', 'events_outbox',
++ 'ctx_users', 'ctx_streams', 'ctx_managed_streams', 'ctx_channels', 'ctx_sync', 'events_outbox', 'live_mirror_outbox', 'service_outbox',
'chat_ingress_applied', 'audio_requests', 'import_hold', 'import_runs', 'chat_meta', 'deploy_releases',
'chat_event_inbox', 'calls', 'rooms', 'room_members', 'room_messages', 'room_attachments', 'token_revocations',
'network_blocks', 'account_data_events', 'ticket_conversations', 'ticket_messages',
];
const INDEXES = [
'idx_chat_ts_deleted', 'idx_chat_page_live', 'idx_chat_channel_user_ts', 'idx_chat_stream_ts', 'idx_emotes_channel_code', 'idx_chat_tl_dedup',
-- 'idx_calls_open', 'idx_ctx_users_subject', 'idx_ctx_users_username', 'idx_audio_requests_room_state', 'idx_events_outbox_unsent',
++ 'idx_calls_open', 'idx_ctx_users_subject', 'idx_ctx_users_username', 'idx_audio_requests_room_state', 'idx_events_outbox_unsent', 'service_outbox_due', 'service_outbox_sent',
'idx_room_messages_room', 'idx_room_members_user', 'idx_network_blocks_blocked', 'idx_mod_actions_created',
];
const MIRRORED = ['chat_messages', 'dm_conversations', 'dm_participants', 'dm_messages', 'dm_blocks', 'tts_voice_overrides',
@@@ -59,7 +60,7 @@ t('every table exists', async () => {
// 0005 (contract) drops the retired Live chat bridge: the tables exist on an N-1 database, but
// this release migrates them away (ADR-028).
t('the retired bridge tables are gone', async () => {
-- for (const name of ['bridge_applied', 'bridge_refs', 'live_mirror_outbox']) {
++ for (const name of ['bridge_applied', 'bridge_refs']) {
assert.strictEqual((await get(`SELECT to_regclass('${name}') AS t`)).t, null, `${name} is dropped`);
}
});
@@@ -70,6 +71,32 @@ t('every index exists', async () => {
for (const name of INDEXES) assert.ok(have.has(name), `index ${name} exists`);
});
++t('SDK migration copies pending legacy envelopes and preserves UTC milliseconds', async () => {
++ const env = { event_id: 'evt_01J9AAAAAAAAAAAAAAAAAAAAAA', event_type: 'chat.message.created',
++ source: 'chat', timestamp: '2026-09-29T10:00:00.123Z' };
++ await db.query(`INSERT INTO events_outbox (event_id, event_type, event, created_at, attempts)
++ VALUES ($1, $2, $3, $4, $5)`, [env.event_id, env.event_type, JSON.stringify(env), env.timestamp, 2]);
++ await db.query(`INSERT INTO events_outbox (event_id, event_type, event, created_at)
++ VALUES ($1, $2, $3, $4)`, ['evt_01J9BBBBBBBBBBBBBBBBBBBBBB', env.event_type,
++ JSON.stringify({ ...env, event_id: 'evt_01J9BBBBBBBBBBBBBBBBBBBBBB' }), 'bad-date']);
++ await db.query(`INSERT INTO events_outbox (event_id, event_type, event, created_at, sent_at)
++ VALUES ($1, $2, $3, $4, $5)`, ['evt_01J9CCCCCCCCCCCCCCCCCCCCCC', env.event_type,
++ JSON.stringify({ ...env, event_id: 'evt_01J9CCCCCCCCCCCCCCCCCCCCCC' }), env.timestamp, env.timestamp]);
++ // The test database already ran the migration as the owner; re-run its copy step, the only DML in it (the CI
++ // containers give the runtime role no CREATE on the schema).
++ const full = fs.readFileSync(path.join(MIGRATIONS, '0008_sdk_outbox.sql'), 'utf8');
++ const sql = full.slice(full.indexOf('INSERT INTO service_outbox'));
++ assert.match(sql, /^INSERT INTO service_outbox[\s\S]*ON CONFLICT \(event_id\) DO NOTHING;/);
++ await db.query(sql);
++ await db.query(sql);
++ const copied = await db.maybe('SELECT envelope, created_at, attempts FROM service_outbox WHERE event_id = $1', [env.event_id]);
++ assert.deepStrictEqual(copied.envelope, env);
++ assert.strictEqual(Number(copied.created_at), Date.parse(env.timestamp));
++ assert.strictEqual(copied.attempts, 2);
++ assert.strictEqual(Number(await db.value('SELECT created_at FROM service_outbox WHERE event_id = $1', ['evt_01J9BBBBBBBBBBBBBBBBBBBBBB'])), 0);
++ assert.strictEqual(await db.maybe('SELECT id FROM service_outbox WHERE event_id = $1', ['evt_01J9CCCCCCCCCCCCCCCCCCCCCC']), null);
++});
++
t('the SQLite-compat functions exist', async () => {
const r = await get(`SELECT ov_now() AS now, datetime('2026-09-29 10:00:00', '+1 hour') AS later, json_valid('{}') AS j`);
assert.match(r.now, /^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$/, 'ov_now is SQLite-format text');
@@@ -82,10 +109,12 @@ t('relay_users has a real identity id', async () => {
assert.strictEqual(r.is_identity, 'YES', 'id is an identity column');
});
--t('the retired Live mirror is gone (0006 dropped its triggers, 0008 its outbox)', async () => {
++t('the retired Live mirror captures nothing (0006 dropped its triggers)', async () => {
const triggers = await all("SELECT tgname FROM pg_trigger WHERE NOT tgisinternal AND tgname LIKE 'mirror\\_%'");
assert.deepStrictEqual(triggers, [], 'no mirror_* capture trigger remains');
-- await db.prepare('INSERT INTO chat_messages (message) VALUES (?)').run('hello'); // a Chat write needs no mirror table
++ const before = (await get('SELECT COUNT(*) AS n FROM live_mirror_outbox')).n;
++ await db.prepare('INSERT INTO chat_messages (message) VALUES (?)').run('hello');
++ assert.strictEqual((await get('SELECT COUNT(*) AS n FROM live_mirror_outbox')).n, before, 'a Chat write queues nothing');
});
t('cleanup', async () => { await close(); });
ddiff --git a/test/ready.test.js b/test/ready.test.js
iindex 16ba532..3a83ee6 100644
---- a/test/ready.test.js
++++ b/test/ready.test.js
@@@ -46,6 +46,10 @@ t('healthy: 200 ready with a db and a live_sync check', async () => {
assert.strictEqual(body.checks.live_sync.detail.threshold_ms, STALE_MS);
assert.ok(body.live.last_success_at, 'last successful sync reported');
assert.strictEqual(body.service, 'chat');
++ assert.deepStrictEqual(body.events.enabled, false);
++ assert.strictEqual(Number(body.events.pending), 0);
++ assert.strictEqual(Number(body.events.rejected), 0);
++ assert.strictEqual(body.events.last_error, null);
});
t('Live failing: passes keep running but do not count; degraded (still 200) after the threshold', async () => {
ddiff --git a/test/rooms.test.js b/test/rooms.test.js
iindex 51efb05..c0190e1 100644
---- a/test/rooms.test.js
++++ b/test/rooms.test.js
@@@ -127,8 +127,8 @@ t('realtime plane (WS-I task 9): public rooms announce messages and redactions,
const dan = h.addUser('dan', { subject: ids.newId('user') });
await h.ctx.upsertUser(h.live.users.get(dan.id));
const db = require('../server/db/database');
-- const since = (await db.get('SELECT MAX(seq) AS n FROM events_outbox') || {}).n || 0;
-- const newEvents = async () => (await db.all('SELECT event FROM events_outbox WHERE seq > ? ORDER BY seq', [since])).map((r) => JSON.parse(r.event)).filter((e) => /^chat\.room\./.test(e.event_type));
++ const since = (await db.get('SELECT MAX(id) AS n FROM service_outbox') || {}).n || 0;
++ const newEvents = async () => (await db.all('SELECT envelope FROM service_outbox WHERE id > ? ORDER BY id', [since])).map((r) => r.envelope).filter((e) => /^chat\.room\./.test(e.event_type));
let r = await api('POST', '/', dan, { name: 'Open Plaza' });
assert.strictEqual(r.status, 201, r.text);
r = await api('POST', '/open-plaza/messages', dan, { message: 'hello plaza' });
ddiff --git a/test/security-private.test.js b/test/security-private.test.js
iindex a648e57..e77841a 100644
---- a/test/security-private.test.js
++++ b/test/security-private.test.js
@@@ -122,7 +122,7 @@ t('WebSocket: a stranger cannot join the private room, and hears nothing posted
t('events outbox: nothing about a private room or a DM leaves Chat', async () => {
// (The staff member's own log searches above are audited with the words they typed: their query, not the room's content.)
-- const rows = JSON.stringify((await h.db.all('SELECT * FROM events_outbox')).filter((r) => !/"action_type":"chat_log_search"/.test(r.event)));
++ const rows = JSON.stringify((await h.db.all('SELECT * FROM service_outbox')).filter((r) => !/"action_type":"chat_log_search"/.test(JSON.stringify(r.envelope))));
for (const w of ['live-private-words', 'live-dm-words', SECRET.roomWords, SECRET.callWords, SECRET.dmWords, SECRET.roomName, SECRET.topic]) assert.ok(!rows.includes(w), `outbox carries "${w}"`);
});
ddiff --git a/test/security-secrets.test.js b/test/security-secrets.test.js
iindex 8a9362d..28f4c8c 100644
---- a/test/security-secrets.test.js
++++ b/test/security-secrets.test.js
@@@ -113,7 +113,7 @@ t('internal routes with a wrong, missing or other-audience service token, and th
});
t('the events outbox and the log lines carry no secret', async () => {
-- const rows = JSON.stringify(await h.db.all('SELECT * FROM events_outbox'));
++ const rows = JSON.stringify(await h.db.all('SELECT * FROM service_outbox'));
for (const [k, v] of Object.entries({ ...ENV, ...SETTINGS })) {
assert.ok(!rows.includes(v), `outbox carries ${k}`);
const line = logs.find((l) => l.includes(v));
ddiff --git a/test/service-outbox-db.test.js b/test/service-outbox-db.test.js
nnew file mode 100644
iindex 0000000..9632458
---- /dev/null
++++ b/test/service-outbox-db.test.js
@@@ -0,0 +1,62 @@
++'use strict';
++/** Chat's SDK outbox joins the ambient transaction and keeps rows while publishing is disabled. */
++const assert = require('assert');
++const path = require('path');
++const { createTestDb } = require('openvibe-sdk/testing');
++const { createDb } = require('openvibe-sdk/db');
++const { suite } = require('./helpers');
++const database = require('../server/db/database');
++const serviceOutbox = require('../server/events/service-outbox');
++
++const t = suite('service-outbox-db');
++let store;
++
++const event = () => ({
++ event_type: 'chat.moderation.action',
++ subject: { type: 'moderation_action', id: '1' },
++ payload: { action_id: 1, action_type: 'test' },
++});
++
++t('configure the SDK outbox with a migrated database', async () => {
++ const migrations = path.join(__dirname, '..', 'migrations');
++ store = await createTestDb({ migrations, service: 'chat' });
++ const owner = store.directUrl ? createDb({ url: store.directUrl, service: 'chat-test-migrate', max: 1 }) : store.db;
++ try { await owner.migrate({ dir: migrations, windowDays: 0 }); }
++ finally { if (owner !== store.db) await owner.close(); }
++ database.setDb(store.db);
++ serviceOutbox.configure({ config: {
++ events: { url: '', intervalMs: 5000 }, networkInternalUrl: 'http://127.0.0.1:4000',
++ oauth: { clientId: 'chat', clientSecret: '' },
++ } });
++});
++
++t('rolled-back changes leave no event; committed changes leave one', async () => {
++ await assert.rejects(store.db.tx(async () => {
++ await store.db.query('INSERT INTO chat_meta (key, value) VALUES ($1, $2)', ['outbox-test', 'rolled-back']);
++ await serviceOutbox.emit(event());
++ throw new Error('abort');
++ }), /abort/);
++ assert.strictEqual(await store.db.value('SELECT count(*) FROM service_outbox'), 0);
++ assert.strictEqual(await store.db.maybe('SELECT value FROM chat_meta WHERE key = $1', ['outbox-test']), null);
++
++ await store.db.tx(async () => {
++ await store.db.query('INSERT INTO chat_meta (key, value) VALUES ($1, $2)', ['outbox-test', 'committed']);
++ await serviceOutbox.emit(event());
++ });
++ const rows = await store.db.many('SELECT envelope, sent_at FROM service_outbox');
++ assert.strictEqual(rows.length, 1);
++ assert.strictEqual(rows[0].envelope.event_type, 'chat.moderation.action');
++ assert.deepStrictEqual(rows[0].envelope.actor, { type: 'service', id: 'chat' });
++ assert.strictEqual(rows[0].sent_at, null);
++ const status = await serviceOutbox.current().status();
++ assert.strictEqual(status.enabled, false);
++ assert.strictEqual(Number(status.pending), 1);
++ await serviceOutbox.current().kick();
++ assert.strictEqual((await store.db.maybe('SELECT sent_at FROM service_outbox')).sent_at, null);
++});
++
++t.run(async () => {
++ serviceOutbox.reset();
++ database.setDb(null);
++ if (store) await store.close();
++});
ddiff --git a/test/ws-protocol.test.js b/test/ws-protocol.test.js
iindex 10d390e..d9926bd 100644
---- a/test/ws-protocol.test.js
++++ b/test/ws-protocol.test.js
@@@ -125,7 +125,7 @@ t('chat into a stream room: room delivery, global cross-feed, persistence, outbo
assert.strictEqual(row.channel_user_id, streamer.id);
assert.strictEqual(row.subject_id, 'usr_01J9AAAAAAAAAAAAAAAAAAAAAA');
// Outbox envelope in the same transaction.
-- const ev = (await h.db.all("SELECT event FROM events_outbox WHERE event_type = 'chat.message.created'")).map((r) => JSON.parse(r.event)).find((e) => e.payload.message_id === msgId);
++ const ev = (await h.db.all("SELECT envelope FROM service_outbox WHERE envelope->>'event_type' = 'chat.message.created'")).map((r) => r.envelope).find((e) => e.payload.message_id === msgId);
assert.ok(ev, 'chat.message.created in the outbox');
assert.strictEqual(ev.visibility, 'public');
assert.deepStrictEqual(ev.actor, { type: 'user', id: 'usr_01J9AAAAAAAAAAAAAAAAAAAAAA' });
@@@ -202,7 +202,7 @@ t('DM delivery keeps the participant check (REST send → live socket; a non-par
assert.strictEqual((await h.http('GET', `/api/dm/conversations/${convId}/messages`, { token: eve.token })).status, 403);
// Offline notification went to Live; the DM event is subject-scoped and carries no text.
assert.ok(h.live.effects.some((e) => e.name === 'notify/dm' && e.body.recipient_ids.includes(bob.id)));
-- const ev = (await h.db.all("SELECT event FROM events_outbox WHERE event_type = 'chat.dm.created'")).map((r) => JSON.parse(r.event)).pop();
++ const ev = (await h.db.all("SELECT envelope FROM service_outbox WHERE envelope->>'event_type' = 'chat.dm.created'")).map((r) => r.envelope).pop();
assert.strictEqual(ev.visibility, 'subject');
assert.ok(!JSON.stringify(ev).includes('secret hello'), 'no DM text in the event');
eveWs.close();
@@@ -259,7 +259,7 @@ t('moderation: only moderators; /ban goes to Live and takes effect at once', asy
// Every action is logged (and announced as an internal moderation event).
const actions = (await h.db.all('SELECT action_type FROM moderation_actions ORDER BY id')).map((r) => r.action_type);
for (const a of ['channel_ban', 'channel_unban', 'channel_timeout', 'slowmode_update', 'clear_chat']) assert.ok(actions.includes(a), a);
-- const modEv = (await h.db.all("SELECT event FROM events_outbox WHERE event_type = 'chat.moderation.action'")).map((r) => JSON.parse(r.event));
++ const modEv = (await h.db.all("SELECT envelope FROM service_outbox WHERE envelope->>'event_type' = 'chat.moderation.action'")).map((r) => r.envelope);
assert.ok(modEv.length >= 5 && modEv.every((e) => e.visibility === 'internal'));
aliceWs.sendJson({ type: 'chat', message: 'slow one' });
await bobWs.next((m) => m.type === 'chat' && m.message === 'slow one');