openvibers/openvibe.live Commits
Keep the chat bridge writers and outbox until Chat owns delivery (#18)
Changes
@@@ -23,6 +23,13 @@ with the same WebSocket protocol and REST paths. `CHAT_AUTHORITY=chat` switchesin `chat_bridge_outbox` (and are applied here if Live boots without the flag again — rollback).- Synchronous reads (`getTotalConnections`, viewer counts, slow modes, a connected user's IP) come fromChat's presence snapshot, polled every 3 s.++- **The bridge writers stay** (T3 J2 deferred). Live still generates the messages Chat delivers —++ AI viewer replies, relays, donations, `/api/mod` deletes, deploy notices — and alert sounds are++ requested from Live (`playAlertSound`). Chat does not yet consume those from an event or generate++ them itself, so `chat-remote.js` and its `chat_bridge_outbox` must keep forwarding them: removing++ the writers first sends nothing to Chat and viewers lose the messages (review 2026-10-02, PR #12).++ Retire them only once Chat owns each path, and drop the outbox only after any unacknowledged++ `op = 'db'` rows are delivered (`test/chat-context.test.js` locks the paths).- Chat reads Live data and asks for side effects on `/internal/chat-context/*` and`/internal/chat-effects/*` (`server/chat/live-context-routes.js`), with Network service tokens(`live.chat_context.read`, `live.chat_effects.write`, `live.chat_mirror.write`,@@@ -17,6 +17,12 @@* loses none;* - synchronous reads (connection counts, viewer counts, slow modes, a connected user's address)* come from a presence snapshot polled from Chat's GET /internal/live/presence every few seconds.++ *++ * Do NOT remove these writers (T3 J2) yet: Live still generates the messages Chat delivers (AI viewer++ * replies, relays, donations, /api/mod deletes, deploy notices) and requests its alert sounds, and Chat++ * does not yet consume or generate them on its own, so dropping this forwarding sends nothing to Chat and++ * viewers lose them. Retire a path only once Chat owns it; drop chat_bridge_outbox only after its++ * unacknowledged op = 'db' rows have been delivered. Covered by test/chat-context.test.js.*/const crypto = require('crypto');const db = require('../db/database');@@@ -334,6 +334,17 @@ const sleep = (ms) => new Promise((r) => setTimeout(r, ms));for (const [ws] of chatServer.clients) { if (ws.readyState === 1) ws.send(JSON.stringify({ type: 'delete-messages', ids: [1] })); }await chatServer.flush();assert.strictEqual(bridgeCalls.at(-1).ops[0].op, 'broadcastAllRaw');++ // Every viewer-facing path Chat does not yet own itself must keep reaching it over the bridge++ // (review 2026-10-02, PR #12): alert sounds, the /api/mod per-socket reply, deploy notices and++ // user-change invalidations. A durable write waits in the outbox until Chat acknowledges it.++ chatServer.playAlertSound(streamer, streamId, 'donation');++ chatServer.sendToConn({ remote: true }, { type: 'mod-action' });++ chatServer.deployNotice([{ hash: 'deadbeef' }]);++ chatServer.userChanged(viewer);++ assert.strictEqual(d.prepare('SELECT COUNT(*) AS n FROM chat_bridge_outbox').get().n, 1, 'a durable write waits in the outbox');++ await chatServer.flush();++ assert.deepStrictEqual(bridgeCalls.at(-1).ops.map((o) => o.op), ['playAlertSound', 'sendToConn', 'deployNotice', 'userChanged']);++ assert.strictEqual(d.prepare('SELECT COUNT(*) AS n FROM chat_bridge_outbox').get().n, 0, 'Chat acknowledged it');// Synchronous reads come from Chat's presence snapshot.await sleep(200);assert.strictEqual(chatServer.getTotalConnections(), 7);
@@ -23,6 +23,13 @@ with the same WebSocket protocol and REST paths. `CHAT_AUTHORITY=chat` switches
@@ -23,6 +23,13 @@ with the same WebSocket protocol and REST paths. `CHAT_AUTHORITY=chat` switches
in `chat_bridge_outbox` (and are applied here if Live boots without the flag again — rollback).
in `chat_bridge_outbox` (and are applied here if Live boots without the flag again — rollback).
- Synchronous reads (`getTotalConnections`, viewer counts, slow modes, a connected user's IP) come from
- Synchronous reads (`getTotalConnections`, viewer counts, slow modes, a connected user's IP) come from
Chat's presence snapshot, polled every 3 s.
Chat's presence snapshot, polled every 3 s.
+- **The bridge writers stay** (T3 J2 deferred). Live still generates the messages Chat delivers —
+ AI viewer replies, relays, donations, `/api/mod` deletes, deploy notices — and alert sounds are
+ requested from Live (`playAlertSound`). Chat does not yet consume those from an event or generate
+ them itself, so `chat-remote.js` and its `chat_bridge_outbox` must keep forwarding them: removing
+ the writers first sends nothing to Chat and viewers lose the messages (review 2026-10-02, PR #12).
+ Retire them only once Chat owns each path, and drop the outbox only after any unacknowledged
+ `op = 'db'` rows are delivered (`test/chat-context.test.js` locks the paths).
- Chat reads Live data and asks for side effects on `/internal/chat-context/*` and
- Chat reads Live data and asks for side effects on `/internal/chat-context/*` and
`/internal/chat-effects/*` (`server/chat/live-context-routes.js`), with Network service tokens
`/internal/chat-effects/*` (`server/chat/live-context-routes.js`), with Network service tokens
(`live.chat_context.read`, `live.chat_effects.write`, `live.chat_mirror.write`,
(`live.chat_context.read`, `live.chat_effects.write`, `live.chat_mirror.write`,
@@ -17,6 +17,12 @@
@@ -17,6 +17,12 @@
* loses none;
* loses none;
* - synchronous reads (connection counts, viewer counts, slow modes, a connected user's address)
* - synchronous reads (connection counts, viewer counts, slow modes, a connected user's address)
* come from a presence snapshot polled from Chat's GET /internal/live/presence every few seconds.
* come from a presence snapshot polled from Chat's GET /internal/live/presence every few seconds.
+ *
+ * Do NOT remove these writers (T3 J2) yet: Live still generates the messages Chat delivers (AI viewer
+ * replies, relays, donations, /api/mod deletes, deploy notices) and requests its alert sounds, and Chat
+ * does not yet consume or generate them on its own, so dropping this forwarding sends nothing to Chat and
+ * viewers lose them. Retire a path only once Chat owns it; drop chat_bridge_outbox only after its
+ * unacknowledged op = 'db' rows have been delivered. Covered by test/chat-context.test.js.
*/
*/
const crypto = require('crypto');
const crypto = require('crypto');
const db = require('../db/database');
const db = require('../db/database');
@@ -334,6 +334,17 @@ const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
@@ -334,6 +334,17 @@ const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
for (const [ws] of chatServer.clients) { if (ws.readyState === 1) ws.send(JSON.stringify({ type: 'delete-messages', ids: [1] })); }
for (const [ws] of chatServer.clients) { if (ws.readyState === 1) ws.send(JSON.stringify({ type: 'delete-messages', ids: [1] })); }
await chatServer.flush();
await chatServer.flush();
assert.strictEqual(bridgeCalls.at(-1).ops[0].op, 'broadcastAllRaw');
assert.strictEqual(bridgeCalls.at(-1).ops[0].op, 'broadcastAllRaw');
+ // Every viewer-facing path Chat does not yet own itself must keep reaching it over the bridge
+ // (review 2026-10-02, PR #12): alert sounds, the /api/mod per-socket reply, deploy notices and
+ // user-change invalidations. A durable write waits in the outbox until Chat acknowledges it.
+ chatServer.playAlertSound(streamer, streamId, 'donation');
+ chatServer.sendToConn({ remote: true }, { type: 'mod-action' });
+ chatServer.deployNotice([{ hash: 'deadbeef' }]);
+ chatServer.userChanged(viewer);
+ assert.strictEqual(d.prepare('SELECT COUNT(*) AS n FROM chat_bridge_outbox').get().n, 1, 'a durable write waits in the outbox');
+ await chatServer.flush();
+ assert.deepStrictEqual(bridgeCalls.at(-1).ops.map((o) => o.op), ['playAlertSound', 'sendToConn', 'deployNotice', 'userChanged']);
+ assert.strictEqual(d.prepare('SELECT COUNT(*) AS n FROM chat_bridge_outbox').get().n, 0, 'Chat acknowledged it');
// Synchronous reads come from Chat's presence snapshot.
// Synchronous reads come from Chat's presence snapshot.
await sleep(200);
await sleep(200);
assert.strictEqual(chatServer.getTotalConnections(), 7);
assert.strictEqual(chatServer.getTotalConnections(), 7);