openvibers/openvibe.chat Commits
Chat parity script (roadmap WS-I task 6): scripts/parity.js runs the chat product end to end over the real protocol with two or three people connected and checks what each sees, one scenario per item: join (anonymous and signed in, the user list), send (echo and delivery with one id, history), DM (participants only, no echo, a non-participant refused), /tts (the room's tts line, the request in the channel's queue, the audio frame), soundboard (!sound: announce, audio, queue), ban, timeout, purge (the dashboard's ISO range), slow mode (viewers slowed, moderators exempt), the popout (a live popout and a channel-room popout, as popout-chat.html joins), and reconnect convergence (a reader away while a line is deleted and another written reads ?after_id= and ends with the connected reader's view; a fresh page agrees). Sub-only mode and blocks in public chat do not exist: those two scenarios report the gap and touch nothing. test/parity.test.js runs the same scenarios against Chat with stubs (found by npm test) and adds Live's moderator paths through the bridge (the context-menu ban that disconnects and drops the user from the list, the single-message delete converging in the global feed's cursor read), the DM participant rule on delivery, and the CLI: its dry run prints the plan and opens nothing, and --apply first resolves every account behind the tokens and the stream's owner and runs nothing unless all are named in OV_PARITY_TEST_ACCOUNTS. docs/parity.md: how to run it, the gaps, what was fixed on the way.
Changes
@@@ -261,6 +261,7 @@ node scripts/parity-check.js --tables --live-db /tmp/live-snapshot.db --chat-dbnode scripts/table-authority.js # who writes each staged table (the handoff is Live's)node scripts/mirror-flush.js # rollback helper: push queued mirror rows to Livenode scripts/subscribe-events.js --dry-run # Chat's Events subscriptions (boot creates missing ones)++node scripts/parity.js # chat parity scenarios: a dry run; --apply only on test accounts (docs/parity.md)````/ready` (openvibe-shared/ready shape: `status` ready/degraded/not_ready and `checks`) is 503 only@@@ -288,8 +289,8 @@ server/net/service-auth.js service tokens: client (Chat → others) and guard (oserver/prefs/ chat preferences in the Network user module chat.preferences (routes, cache, migration from Live)server/calls/ moved from Live: the call server (/ws/call), its REST routes, Live's stream hooks (/internal/calls), the calls lifecycleserver/db/ schema.sql, database.js (Live's chat functions, same names and arguments)--scripts/ import-from-live, parity-check, mirror-flush, table-authority, migrate-chat-preferences, subscribe-events--docs/ cutover.md, calls-cutover.md, staged-tables-cutover.md, live-patch.diff, capabilities-proposal/++scripts/ import-from-live, parity-check, parity, mirror-flush, table-authority, migrate-chat-preferences, subscribe-events++docs/ cutover.md, calls-cutover.md, staged-tables-cutover.md, parity.md, live-patch.diff, capabilities-proposal/```## Owns@@@ -343,6 +344,7 @@ Events: `chat.message.created`, `chat.message.deleted`, `chat.dm.created`, `chat- restart Live without losing Chat; restart Chat's delivery plane and resume persisted messages — *seen in production (Live restarted several times on 2026-09-23 while Chat stayed up; messages and a queued outbox row survived Chat restarts); `test/restart-resume.test.js` restarts Chat as a real process (SIGTERM, new process on the same database) and proves stream, global and channel readers resume from their `after_id` cursor with no gap and no duplicate, including Live bridge placeholders that straddle the restart (kept in `bridge_refs`)*- a call row without a working signalling/media path is not parity — *Live's `/ws/call` protocol and call routes run in Chat with a `calls` row per call (`test/calls.test.js`: two signed-in sockets exchange offer/answer/ICE, limits, kick/ban, ringing → active/declined/missed/failed, stream channels); not switched over yet (`docs/calls-cutover.md`)*- a paid TTS request is never duplicated by a retry — *forwarded writes are applied once per idempotency key; paid TTS does not exist yet*++- after a reconnect, deleted and blocked state converge; browser parity (join, send, DM, `/tts`, moderation, popout) — *`scripts/parity.js` and `test/parity.test.js` (`docs/parity.md`): a reader that was away converges on what a connected reader saw (cursor reads carry `deleted_ids`); gaps: sub-only mode does not exist, public chat does not apply blocks*## Launch rulennew file mode 100644@@@ -0,0 +1,81 @@++# Chat parity script (WS-I task 6)++++`scripts/parity.js` runs the chat product end to end over the real protocol (`/ws/chat` and the++REST routes Live's `chat.js` and popout use), with two or three people connected, and checks what++each of them sees. `test/parity.test.js` runs the same scenarios against Chat booted with stubs++(part of `npm test`), and adds what only a test can reach: Live's moderator paths through the++bridge and the CLI's own safety checks.++++| Scenario | What is checked | Kind |++| --- | --- | --- |++| `join` | anonymous and signed-in joins get their `auth` frame; the user list names the signed-in people and counts the anonymous one | write |++| `send` | the sender's echo and every other viewer's copy carry the same stored id; the line is in the stream history | write |++| `dm` | the other participant's socket gets the `dm` frame, the sender's is not echoed, a non-participant gets nothing and is refused the conversation (`dm.isParticipant`) | write |++| `tts` | `/tts` reaches the room as a `tts` line and is a request in the channel's TTS queue (`GET /api/tts/queue`); its audio frame follows when it plays | write |++| `soundboard` | `!<sound>` gives the room the "played !sound" line and the `soundboard-audio` frame, through the queue | write |++| `ban` | `/ban`: everyone sees the notice, the banned person's next line is refused and reaches nobody, `/unban` lifts it | destructive |++| `timeout` | `/timeout B 2`: refused inside the two seconds, talking again after | destructive |++| `purge` | the dashboard's range purge (ISO `from`/`to`): the room gets the `purge` frame, the range is gone from history, the line before it stays | destructive |++| `slow` | `/slow 2`: a viewer's second line inside the window is refused and reaches nobody; the moderator's two lines 1.2 s apart both go out | destructive |++| `subonly` | **gap**: there is no sub-only mode (below) | — |++| `popout` | a live popout (stream join) and a channel-room popout (join by `channelUserId`) read their histories, get the stream's lines, and the channel popout's line reaches the stream; `get-users` answers | write |++| `reconnect` | A stays, B leaves; a line is deleted and another written while B is away; B reconnects and reads `?after_id=<cursor>`: B's view equals A's, and a fresh page agrees | destructive |++| `blocked` | **gap**: public chat does not apply blocks (below) | — |++++## Running it++++```bash++npm test -- parity # the scenarios against Chat with stubs++++# Against a running Chat. Without --apply it only prints the plan (no connection is opened).++OV_PARITY_BASE=http://127.0.0.1:4401 OV_PARITY_STREAM=<live stream of a test channel> \++OV_PARITY_TOKEN_A=… OV_PARITY_TOKEN_B=… OV_PARITY_TOKEN_MOD=… OV_PARITY_TOKEN_STREAMER=… \++OV_PARITY_TEST_ACCOUNTS=parity_a,parity_b,parity_mod,parity_streamer OV_PARITY_SOUND=honk \++node scripts/parity.js [--apply] [--only join,send]++```++++Roles: A and B are viewers (B is the one banned, timed out and reconnecting), MOD moderates the++stream's channel, STREAMER owns it (needed for `purge` only). `OV_PARITY_SOUND` names a channel sound++command (else `soundboard` is skipped); `OV_PARITY_ORIGIN` (default `https://openvibe.live`) is the++WebSocket origin; `OV_PARITY_CHANNEL_USER_ID` saves looking up the owner's id.++++**Safety.** With `--apply` the script first signs each token in and reads the stream's channel++owner, and runs **nothing** (exit 2) unless every one of those accounts is named in++`OV_PARITY_TEST_ACCOUNTS`. Bans, timeouts, slow mode, purges and deletes then only touch dedicated++test accounts and their own channel. Do not run it against a real channel. Tokens are never printed.++Exit 1 when a scenario fails; gaps and skips are reported, not failed.++++## Known gaps (2026-09-25)++++- **Sub-only mode does not exist**, in Chat or in Live: no channel setting, no command, no check++ (followers-only exists). Chat also has no subscriber lookup: subscriptions are Live's. A product++ decision (what "sub" means: Live subscriptions, VIP membership), then a Live context read and a++ rule in `_chatRulesBlock`.++- **Blocks do not apply to public chat.** `network.block.changed` and `dm_blocks` are honoured for++ DMs and calls only. A blocked person's lines reach the blocker live and in every history read,++ before and after a reconnect, and Live's `chat.js` hides nothing either. Needs a decision (hide++ for the blocker only? both ways? not for moderators?), then per-recipient filtering in the++ broadcasts and the history reads.++- **A channel-room popout's users panel is not the channel's.** `getUserList(null)` is everyone++ outside a stream: global chat and every offline channel room. The `popout` scenario notes it.++- **Slow mode lives in memory.** `/slow` is written to the channel's `slow_mode_seconds`, but that++ setting is never read back: a Chat restart turns slow mode off, and the dashboard's value is not++ enforced.++- `/clear` only clears screens; the lines stay in history and return on a reload. A range purge++ reaches only sockets joined to that stream (not a channel-room popout, not the global feed that++ showed the forwarded copies).++++## Fixed while building it++++- Timeouts ended at midnight UTC: `expires_at` (ISO) was compared as text with SQLite's time.++ `live-context.js` now reads it as an instant. (Live's own ban queries still compare text.)++- A ban or timeout took effect only after an unawaited cache refresh, so the target's next line could++ still go out; the refresh now finishes before the moderator is answered, also when one was++ already in flight.++- The dashboard's range purge, its preview and the log filter compared ISO bounds with SQLite++ timestamps as text: a same-day range matched nothing. They use `datetime(?)`.++- Moderators were slowed by slow mode (they could not even turn it off inside the window). The++ room's moderators now keep only the 1 s flood limit.++- A cursor read (`?after_id=`) never said what was deleted under the cursor, so a reader that++ reconnected kept deleted lines. Cursor reads now carry `deleted_ids` (among the room's newest 1000++ rows); openvibe.chat applies them. Live's floating widget (`hydrateWidgetOnly`) should too.nnew file mode 100644@@@ -0,0 +1,644 @@++#!/usr/bin/env node++/**++ * Chat parity (roadmap WS-I task 6): the chat product, end to end over the real protocol, with two++ * or three connected people, checking what each of them sees. One scenario per item: join, send,++ * DM, /tts, soundboard, moderation (ban, timeout, purge), slow mode, sub-only mode, the popout, and++ * reconnect convergence for deleted and blocked messages.++ *++ * test/parity.test.js runs these same scenarios against Chat booted with stubs (and adds what only++ * a test can see). This CLI runs them against a running Chat:++ *++ * OV_PARITY_BASE=http://127.0.0.1:4401 \++ * OV_PARITY_STREAM=<live stream id of a test channel> \++ * OV_PARITY_TOKEN_A=… OV_PARITY_TOKEN_B=… OV_PARITY_TOKEN_MOD=… [OV_PARITY_TOKEN_STREAMER=…] \++ * OV_PARITY_TEST_ACCOUNTS=parity_a,parity_b,parity_mod,parity_streamer \++ * [OV_PARITY_SOUND=honk] [OV_PARITY_ORIGIN=https://openvibe.live] [OV_PARITY_CHANNEL_USER_ID=…] \++ * node scripts/parity.js [--apply] [--only join,send,…]++ *++ * Roles: A and B are viewers (B is the one banned, timed out and reconnecting), MOD moderates the++ * stream's channel (a channel moderator or its owner), STREAMER owns it (the purge; optional).++ *++ * Without --apply this is a dry run: it prints what each scenario would do and as whom, and opens++ * no connection. With --apply it first signs each token in (one WebSocket join each) and reads the++ * stream's channel owner, and runs NOTHING (exit 2) unless every one of those accounts is named in++ * OV_PARITY_TEST_ACCOUNTS: the messages, DMs, bans, timeouts, slow mode and purges then only ever++ * touch dedicated test accounts and their own channel. Never point --apply at a real channel.++ * Tokens are never printed. Exit 0 = no scenario failed (gaps and skips are reported, not failed).++ */++'use strict';++++const assert = require('assert');++const crypto = require('crypto');++++const PACE_MS = 1150; // Chat allows one chat line per second per address and room++const WAIT_MS = 4000; // how long a frame may take to arrive++const QUIET_MS = 400; // how long "nobody got it" is watched++const ROLES = ['a', 'b', 'mod', 'streamer'];++++class Gap extends Error { constructor(reason) { super(reason); this.status = 'gap'; } }++class Skip extends Error { constructor(reason) { super(reason); this.status = 'skip'; } }++++const names = (list) => ((list && list.logged) || []).map((u) => String(u.username).toLowerCase());++const lower = (s) => String(s || '').toLowerCase();++++/**++ * One parity run: who is who, the stream, and the helpers every scenario uses.++ *++ * driver = {++ * connect({ role, stream }) → a socket to /ws/chat (stream = the ?stream= id or null) with++ * .all (frames received), .next(pred, ms) (the first frame matching, not taken before),++ * .none(pred, ms) (true when none arrives), .sendJson(obj), .close();++ * http(method, path, { token, body }) → { status, body, text };++ * address(role) → the address that role's sockets come from (Chat's rate limit key).++ * }++ */++class Parity {++ constructor({ driver, tokens, streamId, channelUserId = null, sound = null, testAccounts = [], log = () => {} }) {++ this.driver = driver;++ this.tokens = tokens;++ this.streamId = Number(streamId);++ this.ownerId = channelUserId ? Number(channelUserId) : null;++ this.owner = null;++ this.sound = sound ? String(sound).replace(/^!/, '').toLowerCase() : null;++ this.testAccounts = testAccounts.map(lower);++ this.log = log;++ this.users = {};++ this.nonce = crypto.randomBytes(3).toString('hex');++ this.sockets = new Set();++ this.lastSend = new Map();++ }++++ /** Sign each token in (a join) and find the stream's channel owner. */++ async resolve() {++ for (const role of ROLES) {++ if (!this.tokens[role]) continue;++ const ws = await this.driver.connect({ role, stream: null });++ try {++ ws.sendJson({ type: 'join', token: this.tokens[role] });++ const auth = await ws.next((m) => m.type === 'auth', WAIT_MS);++ if (!auth.authenticated) throw new Error(`the token for ${role} did not sign in`);++ this.users[role] = { id: auth.user_id, username: auth.core_username, display: auth.username, role: auth.role };++ } finally { ws.close(); }++ }++ const hist = await this.http('GET', `/api/chat/${this.streamId}/history?limit=1`);++ if (hist.status !== 200 || !hist.body || !hist.body.channel) throw new Error(`stream ${this.streamId}: no channel (${hist.status})`);++ this.owner = hist.body.channel;++ if (!this.ownerId) {++ const prof = await this.http('GET', `/api/chat/user/${encodeURIComponent(this.owner)}/profile`);++ if (prof.status !== 200 || !prof.body || !prof.body.id) throw new Error(`no user id for the channel owner (${prof.status}); set OV_PARITY_CHANNEL_USER_ID`);++ this.ownerId = Number(prof.body.id);++ }++ }++++ /** Accounts --apply would touch that are not named as test accounts. */++ untested() {++ const touched = [...Object.values(this.users).map((u) => u.username), this.owner];++ return [...new Set(touched.filter((n) => !this.testAccounts.includes(lower(n))))];++ }++++ text(label) { return `[parity ${this.nonce}] ${label}`; }++ sleep(ms) { return new Promise((r) => setTimeout(r, ms)); }++ gap(reason) { throw new Gap(reason); }++ skip(reason) { throw new Skip(reason); }++++ http(method, path, { as = null, body } = {}) {++ return this.driver.http(method, path, { token: as ? this.tokens[as] : null, body });++ }++++ /**++ * Open /ws/chat and join the way Live's chat.js does (the token only in the join message):++ * stream → ?stream=<id> and join { streamId }; channel → join { channelUserId } (the channel++ * room, as the popout and an offline channel page do). role 'anon' joins without a token.++ */++ async join(role, { stream = true, channel = false } = {}) {++ const ws = await this.driver.connect({ role, stream: stream ? this.streamId : null });++ this.sockets.add(ws);++ ws.on('close', () => this.sockets.delete(ws));++ ws.role = role;++ ws.rateKey = `${this.driver.address(role)}|${stream ? this.streamId : 'global'}`;++ ws.sendJson({++ type: 'join',++ ...(stream ? { streamId: this.streamId } : {}),++ ...(channel ? { channelUserId: this.ownerId } : {}),++ ...(this.tokens[role] ? { token: this.tokens[role] } : {}),++ });++ ws.auth = await ws.next((m) => m.type === 'auth', WAIT_MS);++ return ws;++ }++++ /** Send a chat line, one per PACE_MS per address and room (Chat's flood limit). */++ async say(ws, message) {++ const wait = (this.lastSend.get(ws.rateKey) || 0) + PACE_MS - Date.now();++ if (wait > 0) await this.sleep(wait);++ ws.sendJson({ type: 'chat', message });++ this.lastSend.set(ws.rateKey, Date.now());++ }++++ saw(ws, message, ms = WAIT_MS) { return ws.next((m) => m.type === 'chat' && m.message === message, ms); }++ system(ws, message, ms = WAIT_MS) { return ws.next((m) => m.type === 'system' && m.message === message, ms); }++ async left(ws) {++ if (ws.readyState === 3) return;++ await new Promise((r) => { ws.once('close', r); ws.close(); });++ }++ async closeAll() { await Promise.all([...this.sockets].map((ws) => this.left(ws).catch(() => {}))); }++++ /** Poll fn until it returns something truthy. */++ async until(fn, ms = WAIT_MS) {++ const end = Date.now() + ms;++ for (;;) {++ const v = await fn();++ if (v) return v;++ if (Date.now() > end) return null;++ await this.sleep(150);++ }++ }++++ /** Run one scenario → { key, status: 'pass' | 'gap' | 'skip' | 'fail', note, error }. */++ async run(scenario) {++ const started = Date.now();++ try {++ for (const r of scenario.needs || []) if (!this.tokens[r]) this.skip(`needs OV_PARITY_TOKEN_${r.toUpperCase()}`);++ const note = await scenario.run(this);++ return { key: scenario.key, status: 'pass', note: note || '', ms: Date.now() - started };++ } catch (err) {++ if (err instanceof Gap || err instanceof Skip) return { key: scenario.key, status: err.status, note: err.message, ms: Date.now() - started };++ return { key: scenario.key, status: 'fail', note: err && err.message, error: err, ms: Date.now() - started };++ } finally {++ await this.closeAll();++ }++ }++}++++/**++ * A reader's view of a room, the way chat.js keeps it: the ids of the messages on screen, from a++ * history page, the live frames (chat lines in, delete-messages out) and cursor reads (?after_id=:++ * new rows in, deleted_ids out).++ */++class RoomView {++ constructor(page) {++ this.ids = new Set((page.messages || []).map((m) => m.id));++ this.cursor = page.latest_id || 0;++ }++ frames(list) {++ for (const m of list) {++ if (m.type === 'chat' && Number.isInteger(m.id)) { this.ids.add(m.id); this.cursor = Math.max(this.cursor, m.id); }++ if (m.type === 'delete-messages') for (const id of m.ids || []) this.ids.delete(id);++ }++ return this;++ }++ delta(d) {++ for (const m of d.messages || []) this.ids.add(m.id);++ for (const id of d.deleted_ids || []) this.ids.delete(id);++ this.cursor = Math.max(this.cursor, d.latest_id || 0);++ return this;++ }++ sorted() { return [...this.ids].sort((x, y) => x - y); }++}++++// ── The scenarios ────────────────────────────────────────────────────────────────────────────++// kind: 'write' posts into the test channel or DMs; 'destructive' bans, times out, deletes or++// changes the channel's chat mode. Both run only with --apply and only on test accounts. 'gap': the++// product has no such behaviour yet; the scenario reports why and touches nothing.++++const SCENARIOS = [++ {++ key: 'join', kind: 'write', as: ['anon', 'a', 'b'], needs: ['a', 'b'],++ does: 'anonymous and signed-in sockets join the stream; each gets its auth frame; the user list names A and B',++ async run(p) {++ const anon = await p.join('anon');++ assert.strictEqual(anon.auth.authenticated, false, 'anonymous join');++ assert.match(String(anon.auth.username), /^anon\d+$/, 'an anon number');++ assert.strictEqual(anon.auth.role, 'anon');++ const a = await p.join('a');++ assert.strictEqual(a.auth.authenticated, true, 'signed-in join');++ assert.strictEqual(a.auth.core_username, p.users.a.username);++ assert.strictEqual(a.auth.user_id, p.users.a.id);++ const b = await p.join('b');++ const both = (m) => m.type === 'users-list' && names(m.users).includes(lower(p.users.a.username)) && names(m.users).includes(lower(p.users.b.username));++ anon.sendJson({ type: 'get-users' });++ const list = await anon.next(both, WAIT_MS);++ assert.ok(list.users.anonCount >= 1, 'the anonymous viewer is counted');++ await a.next(both, WAIT_MS); // pushed to everyone in the room, not only on request++ return `anon ${anon.auth.username}; list ${list.users.logged.length} named, ${list.users.anonCount} anon`;++ },++ },++ {++ key: 'send', kind: 'write', as: ['a', 'b', 'anon'], needs: ['a', 'b'],++ does: 'A sends a line: A gets its own echo, B and an anonymous viewer get it with the same id; it is in the stream history',++ async run(p) {++ const a = await p.join('a'); const b = await p.join('b'); const anon = await p.join('anon');++ const text = p.text('hello from A');++ await p.say(a, text);++ const echo = await p.saw(a, text);++ const got = await p.saw(b, text);++ const gotAnon = await p.saw(anon, text);++ assert.ok(Number.isInteger(echo.id), 'the echo carries the stored id');++ assert.strictEqual(got.id, echo.id); assert.strictEqual(gotAnon.id, echo.id);++ assert.strictEqual(got.core_username, p.users.a.username);++ assert.strictEqual(got.user_id, p.users.a.id);++ assert.strictEqual(got.stream_id, p.streamId);++ const hist = await p.http('GET', `/api/chat/${p.streamId}/history?limit=50`);++ assert.ok(hist.body.messages.some((m) => m.id === echo.id && m.message === text), 'in the stream history');++ return `message ${echo.id}`;++ },++ },++ {++ key: 'dm', kind: 'write', as: ['a', 'b', 'mod'], needs: ['a', 'b', 'mod'],++ does: 'A opens a DM with B and writes: B’s socket gets it, A’s socket is not echoed, MOD (not a participant) gets nothing and is refused the conversation',++ async run(p) {++ const a = await p.join('a'); const b = await p.join('b'); const x = await p.join('mod');++ const conv = await p.http('POST', '/api/dm/conversations', { as: 'a', body: { user_ids: [p.users.b.id] } });++ assert.strictEqual(conv.status, 200, conv.text);++ const convId = conv.body.conversation.id;++ assert.deepStrictEqual(conv.body.conversation.participants.map((u) => u.id).sort(), [p.users.a.id, p.users.b.id].sort());++ const text = p.text('a private hello');++ const sent = await p.http('POST', `/api/dm/conversations/${convId}/messages`, { as: 'a', body: { message: text } });++ assert.strictEqual(sent.status, 200, sent.text);++ const got = await b.next((m) => m.type === 'dm' && m.conversation_id === convId && m.message && m.message.message === text);++ assert.strictEqual(got.message.sender_id, p.users.a.id);++ const isIt = (m) => m.type === 'dm' && m.message && m.message.message === text;++ assert.ok(await a.none(isIt, QUIET_MS), 'the sender’s socket is not echoed (the REST answer carries it)');++ assert.ok(await x.none(isIt, QUIET_MS), 'delivered to participants only');++ assert.strictEqual((await p.http('GET', `/api/dm/conversations/${convId}/messages`, { as: 'mod' })).status, 403, 'a non-participant cannot read it');++ const inbox = await p.http('GET', `/api/dm/conversations/${convId}/messages`, { as: 'b' });++ assert.ok(inbox.body.messages.some((m) => m.message === text), 'B reads it');++ return `conversation ${convId}`;++ },++ },++ {++ key: 'tts', kind: 'write', as: ['a', 'b', 'mod'], needs: ['a', 'b', 'mod'],++ does: 'A sends /tts: the room gets the tts line and the request is in the channel’s TTS queue (read as MOD); the audio frame follows when it plays',++ async run(p) {++ const a = await p.join('a'); const b = await p.join('b');++ const text = p.text('read this out');++ await p.say(a, `/tts ${text}`);++ const line = await b.next((m) => m.type === 'tts' && m.message === text);++ assert.strictEqual(line.core_username, p.users.a.username);++ const req = await p.until(async () => {++ const q = await p.http('GET', `/api/tts/queue?stream_id=${p.streamId}&recent=50`, { as: 'mod' });++ assert.strictEqual(q.status, 200, q.text);++ return [q.body.playing, ...q.body.queued, ...q.body.recent].filter(Boolean).find((r) => r.kind === 'tts' && r.label === text);++ });++ assert.ok(req, 'the /tts request is in the room’s queue');++ assert.strictEqual(req.requested_by, p.users.a.display);++ let audio = 'not played yet';++ if (req.state === 'played' || req.state === 'playing') {++ const f = await b.next((m) => m.type === 'tts-audio' && m.message === text, 8000);++ assert.ok(f.audio, 'the audio frame carries audio');++ audio = 'audio frame received';++ }++ return `queue request ${req.id} (${req.state}); ${audio}`;++ },++ },++ {++ key: 'soundboard', kind: 'write', as: ['a', 'b', 'mod'], needs: ['a', 'b', 'mod'],++ does: 'A types !<sound>: the room gets the “played !sound” line and the soundboard-audio frame; the channel’s queue has it',++ async run(p) {++ if (!p.sound) p.skip('no channel sound command (OV_PARITY_SOUND)');++ const a = await p.join('a'); const b = await p.join('b');++ await p.say(a, `!${p.sound}`);++ const isAnnounce = (m) => m.type === 'chat' && m.message_type === 'channel-sound' && m.message === `played !${p.sound}`;++ const ann = await b.next(isAnnounce);++ assert.strictEqual(ann.core_username, p.users.a.username);++ await a.next(isAnnounce);++ const audio = await b.next((m) => m.type === 'soundboard-audio' && m.title === `!${p.sound}`, 8000);++ assert.ok(audio.audio && audio.audio.length > 0, 'the clip is in the frame');++ const q = await p.http('GET', `/api/tts/queue?stream_id=${p.streamId}&recent=50`, { as: 'mod' });++ const req = [q.body.playing, ...q.body.queued, ...q.body.recent].filter(Boolean).find((r) => r.kind === 'channel-sound' && r.label === `!${p.sound}`);++ assert.ok(req, 'the sound went through the room’s queue');++ return `!${p.sound}: announce ${ann.id}, queue request ${req.id} (${req.state})`;++ },++ },++ {++ key: 'ban', kind: 'destructive', as: ['mod', 'a', 'b'], needs: ['a', 'b', 'mod'],++ does: 'MOD /ban B in the stream: everyone sees the notice; B’s next line is refused and reaches nobody; /unban lifts it',++ async run(p) {++ const a = await p.join('a'); const b = await p.join('b'); const mod = await p.join('mod');++ const target = p.users.b.username;++ await p.say(mod, `/ban ${target}`);++ await p.system(a, `${target} has been banned.`);++ await p.system(b, `${target} has been banned.`);++ try {++ const text = p.text('can B still talk?');++ await p.say(b, text);++ await p.system(b, 'You are banned from this chat.');++ assert.ok(await a.none((m) => m.type === 'chat' && m.message === text, QUIET_MS), 'a banned line reaches nobody');++ } finally {++ await p.say(mod, `/unban ${target}`);++ await p.system(mod, `${target} has been unbanned.`);++ }++ const back = p.text('B is back');++ await p.say(b, back);++ await p.saw(a, back);++ return `${target} banned and unbanned`;++ },++ },++ {++ key: 'timeout', kind: 'destructive', as: ['mod', 'a', 'b'], needs: ['a', 'b', 'mod'],++ does: 'MOD /timeout B 2: B’s line inside the 2 s is refused; after it B’s line reaches A',++ async run(p) {++ const a = await p.join('a'); const b = await p.join('b'); const mod = await p.join('mod');++ const target = p.users.b.username;++ const secs = 2;++ await p.say(mod, `/timeout ${target} ${secs}`);++ await p.system(mod, `${target} timed out for ${secs}s.`);++ const t0 = Date.now();++ const during = p.text('timed out?');++ await p.say(b, during);++ await p.system(b, 'You are banned from this chat.');++ assert.ok(await a.none((m) => m.type === 'chat' && m.message === during, QUIET_MS), 'a timed-out line reaches nobody');++ await p.sleep(Math.max(0, t0 + secs * 1000 + 700 - Date.now()));++ const after = p.text('the timeout is over');++ await p.say(b, after);++ await p.saw(a, after);++ return `${target} timed out for ${secs}s, then talked again`;++ },++ },++ {++ key: 'purge', kind: 'destructive', as: ['streamer', 'a', 'b'], needs: ['a', 'b', 'streamer'],++ does: 'STREAMER purges a time range of the stream (the dashboard’s ISO range): A and B get the purge frame; the lines in the range are gone from history, the one before stays',++ async run(p) {++ const a = await p.join('a'); const b = await p.join('b');++ const before = p.text('before the purge');++ await p.say(a, before);++ const kept = await p.saw(b, before);++ // The range starts after `before` and is built from the server's own timestamps, so a++ // client clock that is off does not matter. Rows keep whole seconds.++ const from = new Date(Date.parse(kept.timestamp) + 1500).toISOString();++ await p.sleep(2700);++ const g1 = p.text('purge me (A)'); const g2 = p.text('purge me (B)');++ await p.say(a, g1); const m1 = await p.saw(b, g1);++ await p.say(b, g2); const m2 = await p.saw(a, g2);++ assert.ok(Date.parse(m1.timestamp) >= Date.parse(from) + 1000, 'the purged lines are inside the range');++ const to = new Date(Date.parse(m2.timestamp) + 1500).toISOString();++ const range = { streamId: p.streamId, from, to };++ const preview = await p.http('POST', '/api/chat/admin/purge/preview', { as: 'streamer', body: range });++ assert.strictEqual(preview.status, 200, preview.text);++ assert.ok(preview.body.count >= 2, `the preview counts the range (${preview.body.count})`);++ const r = await p.http('DELETE', '/api/chat/admin/purge', { as: 'streamer', body: range });++ assert.strictEqual(r.status, 200, r.text);++ assert.ok(r.body.deleted >= 2, `purged ${r.body.deleted}`);++ for (const ws of [a, b]) await ws.next((m) => m.type === 'purge' && m.from === from && m.to === to);++ const ids = (await p.http('GET', `/api/chat/${p.streamId}/history?limit=200`)).body.messages.map((m) => m.id);++ assert.ok(!ids.includes(m1.id) && !ids.includes(m2.id), 'the range is gone from history');++ assert.ok(ids.includes(kept.id), 'the line before the range stays');++ return `${r.body.deleted} purged`;++ },++ },++ {++ key: 'slow', kind: 'destructive', as: ['mod', 'a', 'b'], needs: ['a', 'b', 'mod'],++ does: 'MOD /slow 2: A’s second line inside 2 s is refused and reaches nobody; MOD’s two lines 1.2 s apart both go out (moderators are exempt); /slow off',++ async run(p) {++ const a = await p.join('a'); const b = await p.join('b'); const mod = await p.join('mod');++ const secs = 2;++ await p.say(mod, `/slow ${secs}`);++ assert.strictEqual((await a.next((m) => m.type === 'slowmode')).seconds, secs);++ await p.system(a, `Slow mode enabled: ${secs}s between messages`);++ try {++ // Everyone may share one address (the CLI): let the /slow line's window pass first.++ await p.sleep(secs * 1000 + 200);++ const one = p.text('slow one'); const two = p.text('slow two');++ await p.say(a, one);++ await p.saw(b, one);++ await p.say(a, two);++ await p.system(a, 'Slow down! You are sending messages too fast.');++ assert.ok(await b.none((m) => m.type === 'chat' && m.message === two, QUIET_MS), 'a slowed line reaches nobody');++ const m1 = p.text('mod one'); const m2 = p.text('mod two');++ await p.say(mod, m1); await p.saw(b, m1);++ await p.say(mod, m2); await p.saw(b, m2);++ } finally {++ // Turn it off whatever happened; a Chat that slows moderators refuses the first try.++ await p.say(mod, '/slow off');++ const off = await Promise.race([++ p.system(a, 'Slow mode disabled.').then(() => true),++ mod.next((m) => m.type === 'system' && /^Slow down!/.test(m.message)).then(() => false),++ ]).catch(() => false);++ if (!off) {++ await p.sleep(secs * 1000 + 200);++ await p.say(mod, '/slow off');++ await p.system(a, 'Slow mode disabled.');++ }++ }++ return `viewer slowed, moderator exempt (${secs}s)`;++ },++ },++ {++ key: 'subonly', kind: 'gap', as: [], needs: [],++ does: 'sub-only mode: a non-subscriber is refused, a subscriber may talk',++ async run(p) {++ p.gap('no sub-only mode exists: no channel setting, no command and no check in Chat or Live (followers-only exists), and Chat has no subscriber lookup in live-context (subscriptions are Live’s)');++ },++ },++ {++ key: 'popout', kind: 'write', as: ['a', 'b', 'mod'], needs: ['a', 'b', 'mod'],++ does: 'Live’s popout (popout-chat.html + chat.js): a live popout (stream join) and a channel-room popout (join by channelUserId) read their histories, get stream lines, and the channel popout’s line reaches the stream; get-users answers',++ async run(p) {++ const pop = await p.join('a', { stream: true, channel: true });++ const chan = await p.join('b', { stream: false, channel: true });++ const viewer = await p.join('mod');++ assert.strictEqual(chan.auth.authenticated, true);++ const sHist = await p.http('GET', `/api/chat/${p.streamId}/history?limit=500`);++ assert.strictEqual(sHist.status, 200); assert.strictEqual(lower(sHist.body.channel), lower(p.owner));++ assert.strictEqual((await p.http('GET', '/api/chat/global/history?limit=500')).status, 200);++ const cHist = await p.http('GET', `/api/chat/channel/${p.ownerId}/history?limit=500`);++ assert.strictEqual(cHist.status, 200); assert.strictEqual(lower(cHist.body.channel), lower(p.owner));++ const t1 = p.text('to the popouts');++ await p.say(viewer, t1);++ const f1 = await p.saw(pop, t1); await p.saw(chan, t1);++ const t2 = p.text('from the channel popout');++ await p.say(chan, t2);++ const f2 = await p.saw(viewer, t2); await p.saw(pop, t2);++ assert.strictEqual(f2.channel_user_id, p.ownerId);++ const ids = (await p.http('GET', `/api/chat/channel/${p.ownerId}/history?limit=50`)).body.messages.map((m) => m.id);++ assert.ok(ids.includes(f1.id) && ids.includes(f2.id), 'both lines are in the channel history a popout reloads');++ pop.sendJson({ type: 'get-users' });++ await pop.next((m) => m.type === 'users-list' && names(m.users).includes(lower(p.users.a.username)));++ // The channel-room popout's users panel: is it the channel's? (MOD is also in global chat.)++ await p.join('mod', { stream: false });++ chan.sendJson({ type: 'get-users' });++ const list = await chan.next((m) => m.type === 'users-list' && names(m.users).includes(lower(p.users.b.username)) && names(m.users).includes(lower(p.users.mod.username)), 1500).catch(() => null);++ return `live and channel-room popouts${list ? '; GAP: the channel-room popout’s users panel also names people in global chat (getUserList(null) is everyone outside a stream)' : ''}`;++ },++ },++ {++ key: 'reconnect', kind: 'destructive', as: ['a', 'b', 'mod'], needs: ['a', 'b', 'mod'],++ does: 'A stays connected, B leaves; MOD’s line is deleted (MOD’s own delete-all) and A writes while B is away; B reconnects and reads ?after_id=<cursor>: B’s view equals A’s, and a fresh page agrees',++ async run(p) {++ const room = `/api/chat/channel/${p.ownerId}/history`;++ const stayer = await p.join('a'); let leaver = await p.join('b'); const writer = await p.join('mod');++ const viewA = new RoomView((await p.http('GET', `${room}?limit=500`)).body); const fromA = stayer.all.length;++ const viewB = new RoomView((await p.http('GET', `${room}?limit=500`)).body); const fromB = leaver.all.length;++ const doomed = p.text('this line will be deleted');++ await p.say(writer, doomed);++ const m1 = await p.saw(stayer, doomed); await p.saw(leaver, doomed);++ viewB.frames(leaver.all.slice(fromB));++ await p.left(leaver);++ // While B is away: a deletion, and a new line.++ writer.sendJson({ type: 'self-delete-history' });++ await writer.next((m) => m.type === 'self-delete-result');++ await stayer.next((m) => m.type === 'delete-messages' && (m.ids || []).includes(m1.id));++ const meanwhile = p.text('written while B was away');++ await p.say(stayer, meanwhile);++ const m2 = await p.saw(stayer, meanwhile);++ viewA.frames(stayer.all.slice(fromA));++ // B comes back: a new socket, then the cursor read chat.js makes.++ leaver = await p.join('b');++ const d = await p.http('GET', `${room}?after_id=${viewB.cursor}`);++ assert.strictEqual(d.status, 200); assert.strictEqual(d.body.complete, true);++ viewB.delta(d.body);++ assert.ok(!viewB.ids.has(m1.id), 'the line deleted while B was away is gone from B’s view');++ assert.ok(viewB.ids.has(m2.id), 'the line written while B was away is in B’s view');++ assert.deepStrictEqual(viewB.sorted(), viewA.sorted(), 'B converges on what A saw');++ // A fresh page (the main pane's reload) agrees with both, over the rows it covers.++ const fresh = (await p.http('GET', `${room}?limit=500`)).body.messages.map((m) => m.id);++ const lo = Math.min(...fresh);++ assert.deepStrictEqual(fresh.slice().sort((x, y) => x - y), viewA.sorted().filter((id) => id >= lo), 'a fresh page shows the same');++ return `deleted ${m1.id} dropped via deleted_ids; ${m2.id} added via the delta`;++ },++ },++ {++ key: 'blocked', kind: 'gap', as: [], needs: [],++ does: 'reconnect convergence for blocked messages: lines from someone the reader blocked (network blocks) are hidden live and after a reconnect',++ async run(p) {++ p.gap('public chat does not apply blocks: network.block.changed and dm_blocks are honoured for DMs and calls only (server/chat/network-blocks.js, dm.js isBlockedEither); room broadcasts and history pages show a blocked person’s lines to the blocker, live and after a reconnect alike, and Live’s chat.js hides nothing either');++ },++ },++];++++// ── The command line ─────────────────────────────────────────────────────────────────────────++++/** A socket with the frame inbox the scenarios read (the same semantics as test/helpers.js h.ws). */++function openSocket(url, headers) {++ const WebSocket = require('ws');++ return new Promise((resolve, reject) => {++ const ws = new WebSocket(url, { headers });++ const all = [];++ const waiters = [];++ ws.on('message', (d) => {++ let m; try { m = JSON.parse(d.toString()); } catch { return; }++ all.push(m);++ for (const w of [...waiters]) if (w.pred(m)) { waiters.splice(waiters.indexOf(w), 1); clearTimeout(w.timer); w.resolve(m); }++ });++ ws.all = all;++ ws.next = (pred, ms = WAIT_MS) => {++ const hit = all.find((m) => pred(m) && !m.__taken);++ if (hit) { hit.__taken = true; return Promise.resolve(hit); }++ return new Promise((res, rej) => {++ const w = { pred: (m) => { if (pred(m)) { m.__taken = true; return true; } return false; }, resolve: res };++ w.timer = setTimeout(() => { waiters.splice(waiters.indexOf(w), 1); rej(new Error(`timed out waiting (last frames: ${JSON.stringify(all.slice(-8).map((x) => [x.type, x.message]))})`)); }, ms);++ waiters.push(w);++ });++ };++ ws.none = async (pred, ms = QUIET_MS) => { await new Promise((r) => setTimeout(r, ms)); return !all.some(pred); };++ ws.sendJson = (o) => ws.send(JSON.stringify(o));++ ws.once('open', () => resolve(ws));++ ws.once('error', reject);++ ws.once('unexpected-response', (req, res) => reject(new Error(`upgrade refused ${res.statusCode}`)));++ });++}++++function networkDriver({ base, origin }) {++ const wsBase = base.replace(/^http/, 'ws');++ return {++ address: () => 'this machine',++ connect: ({ stream }) => openSocket(`${wsBase}/ws/chat${stream ? `?stream=${stream}` : ''}`, { origin }),++ async http(method, path, { token, body } = {}) {++ const res = await fetch(`${base}${path}`, {++ method,++ headers: { Accept: 'application/json', ...(body !== undefined ? { 'Content-Type': 'application/json' } : {}), ...(token ? { Authorization: `Bearer ${token}` } : {}) },++ body: body !== undefined ? JSON.stringify(body) : undefined,++ signal: AbortSignal.timeout(15000),++ });++ const text = await res.text();++ let json = null; try { json = JSON.parse(text); } catch { /* not JSON */ }++ return { status: res.status, body: json, text };++ },++ };++}++++function parseArgs(argv) {++ const o = { apply: false, only: null };++ for (let i = 0; i < argv.length; i++) {++ const a = argv[i];++ if (a === '--apply') o.apply = true;++ else if (a === '--only') { o.only = String(argv[++i] || '').split(',').map((s) => s.trim()).filter(Boolean); }++ else if (a === '--help' || a === '-h') o.help = true;++ else throw new Error(`unknown argument ${a}`);++ }++ return o;++}++++function configFrom(env) {++ const base = String(env.OV_PARITY_BASE || '').replace(/\/+$/, '');++ return {++ base,++ origin: env.OV_PARITY_ORIGIN || 'https://openvibe.live',++ streamId: parseInt(env.OV_PARITY_STREAM, 10) || null,++ channelUserId: parseInt(env.OV_PARITY_CHANNEL_USER_ID, 10) || null,++ sound: env.OV_PARITY_SOUND || null,++ tokens: { a: env.OV_PARITY_TOKEN_A || null, b: env.OV_PARITY_TOKEN_B || null, mod: env.OV_PARITY_TOKEN_MOD || null, streamer: env.OV_PARITY_TOKEN_STREAMER || null },++ testAccounts: String(env.OV_PARITY_TEST_ACCOUNTS || '').split(',').map((s) => s.trim()).filter(Boolean),++ };++}++++const LABEL = { pass: 'PASS', gap: 'GAP ', skip: 'SKIP', fail: 'FAIL' };++++/** → exit code: 0 ok, 1 a scenario failed, 2 refused (configuration, or accounts not named as test accounts). */++async function main(argv = process.argv.slice(2), env = process.env, log = console.log, { driver = null } = {}) {++ let o;++ try { o = parseArgs(argv); } catch (err) { log(err.message); return 2; }++ if (o.help) { log(require('fs').readFileSync(__filename, 'utf8').split('\n').slice(1, 32).join('\n')); return 0; }++ const cfg = configFrom(env);++ const chosen = SCENARIOS.filter((s) => !o.only || o.only.includes(s.key));++ if (o.only && chosen.length !== o.only.length) { log(`unknown scenario in --only (known: ${SCENARIOS.map((s) => s.key).join(', ')})`); return 2; }++ const problems = [];++ if (!cfg.base && !driver) problems.push('OV_PARITY_BASE is required');++ if (!cfg.streamId) problems.push('OV_PARITY_STREAM is required (a live stream of a test channel)');++ const have = ROLES.map((r) => `${r} ${cfg.tokens[r] ? 'set' : '-'}`).join(', ');++++ if (!o.apply) {++ log('Chat parity: dry run. Nothing is sent and no connection is opened; --apply runs it.');++ log(` base ${cfg.base || '(none)'} origin ${cfg.origin} stream ${cfg.streamId || '(none)'} sound ${cfg.sound ? `!${cfg.sound}` : '(none)'}`);++ log(` tokens: ${have}`);++ log(` test accounts (OV_PARITY_TEST_ACCOUNTS): ${cfg.testAccounts.join(', ') || '(none)'}`);++ for (const p of problems) log(` missing: ${p}`);++ log('');++ for (const s of chosen) {++ const missing = (s.needs || []).filter((r) => !cfg.tokens[r]);++ const would = s.kind === 'gap' ? 'reports a gap, touches nothing'++ : missing.length ? `would skip: needs ${missing.map((r) => `OV_PARITY_TOKEN_${r.toUpperCase()}`).join(', ')}`++ : s.key === 'soundboard' && !cfg.sound ? 'would skip: needs OV_PARITY_SOUND' : `as ${s.as.join(', ')}`;++ log(` ${s.key.padEnd(11)} ${s.kind.padEnd(11)} ${would}`);++ log(` ${s.does}`);++ }++ log('');++ log('With --apply every account behind the tokens, and the stream’s channel owner, must be named in');++ log('OV_PARITY_TEST_ACCOUNTS, or nothing runs.');++ return problems.length ? 2 : 0;++ }++++ if (problems.length) { for (const p of problems) log(p); return 2; }++ const p = new Parity({ driver: driver || networkDriver(cfg), tokens: cfg.tokens, streamId: cfg.streamId, channelUserId: cfg.channelUserId, sound: cfg.sound, testAccounts: cfg.testAccounts, log });++ try { await p.resolve(); } catch (err) { log(`could not resolve the accounts and the stream: ${err.message}`); return 2; }++ const outside = p.untested();++ if (outside.length) {++ log(`refused: ${outside.join(', ')} ${outside.length === 1 ? 'is' : 'are'} not named in OV_PARITY_TEST_ACCOUNTS; nothing was run.`);++ return 2;++ }++ log(`Chat parity against ${cfg.base || 'the given driver'}: stream ${p.streamId} (channel ${p.owner}), run ${p.nonce}`);++ const results = [];++ for (const s of chosen) {++ const r = await p.run(s);++ results.push(r);++ log(` ${LABEL[r.status]} ${s.key.padEnd(11)} ${r.note || ''}`);++ }++ const count = (st) => results.filter((r) => r.status === st).length;++ log(`\n${count('pass')} passed, ${count('gap')} gap(s), ${count('skip')} skipped, ${count('fail')} failed`);++ return count('fail') ? 1 : 0;++}++++module.exports = { SCENARIOS, Parity, RoomView, Gap, Skip, openSocket, networkDriver, main, PACE_MS };++++if (require.main === module) {++ main().then((code) => process.exit(code), (err) => { console.error(err && err.stack || err); process.exit(1); });++}nnew file mode 100644@@@ -0,0 +1,195 @@++'use strict';++/**++ * Chat parity (roadmap WS-I task 6). scripts/parity.js's scenarios (join, send, DM, /tts,++ * soundboard, ban, timeout, purge, slow mode, sub-only mode, the popout, reconnect convergence for++ * deleted and blocked messages) run here against Chat booted with stubs, with two or three people++ * connected, each from their own address: the same code the CLI runs against a running Chat.++ *++ * On top, what only a test can reach: Live's moderator paths through the bridge (the context-menu++ * ban that disconnects, the single-message delete) and the global feed's cursor read (Live's++ * floating widget, openvibe.chat), the DM participant rule on delivery, and the CLI itself: its dry++ * run touches nothing, and --apply refuses to run for an account that is not a named test account.++ *++ * A scenario the product does not support ends as a gap with its reason (sub-only mode, blocks in++ * public chat); nothing here asserts that it works. The matrix is printed at the end.++ */++const assert = require('assert');++const { boot, suite } = require('./helpers');++const parity = require('../scripts/parity');++++const t = suite('parity');++let h, p, streamer, mod, alice, bob, dave, channelId, streamId;++const IPS = { a: '198.51.100.101', b: '198.51.100.102', mod: '198.51.100.103', streamer: '198.51.100.104', anon: '198.51.100.105', dave: '198.51.100.106' };++const results = [];++let bridgeSeq = 0;++const bridge = (ops) => h.http('POST', '/internal/live/calls', {++ token: h.serviceToken(['chat.live_bridge.write', 'chat.message.send']),++ body: { boot: 'parity-live', ops: ops.map((o) => ({ seq: ++bridgeSeq, ...o })) },++});++++/** 8 kHz 8-bit mono WAV of `seconds`. */++function wav(seconds) {++ const n = Math.floor(8000 * seconds);++ const b = Buffer.alloc(44 + n);++ b.write('RIFF', 0); b.writeUInt32LE(36 + n, 4); b.write('WAVE', 8); b.write('fmt ', 12);++ b.writeUInt32LE(16, 16); b.writeUInt16LE(1, 20); b.writeUInt16LE(1, 22); b.writeUInt32LE(8000, 24);++ b.writeUInt32LE(8000, 28); b.writeUInt16LE(1, 32); b.writeUInt16LE(8, 34); b.write('data', 36); b.writeUInt32LE(n, 40);++ b.fill(128, 44);++ return b;++}++++// Each role's sockets come from its own address (through nginx: CF-Connecting-IP).++const driver = {++ address: (role) => IPS[role],++ connect: ({ role, stream }) => h.ws({ ip: IPS[role], stream }),++ http: (method, path, { token, body } = {}) => h.http(method, path, { token, body }),++};++++t('boot: a channel with a moderator, a live stream, a channel sound, TTS that synthesizes', async () => {++ h = await boot();++ streamer = h.addUser('streamer', { role: 'streamer', subject: 'usr_01J9PAR1TY00000000000000S1' });++ mod = h.addUser('moddy', { subject: 'usr_01J9PAR1TY00000000000000M1' });++ alice = h.addUser('alice', { subject: 'usr_01J9PAR1TY00000000000000A1' });++ bob = h.addUser('bob', { subject: 'usr_01J9PAR1TY00000000000000B1' });++ dave = h.addUser('dave');++ channelId = h.addChannel(streamer.id, { moderators: [mod.id] });++ streamId = h.addStream(streamer.id, channelId);++ await h.ctx.sync();++ // TTS without an engine: every utterance is 0.2 s of silence.++ const tts = require('../server/chat/tts-engine');++ const fake = async (text) => ({ audio: wav(0.2).toString('base64'), mimeType: 'audio/wav', engine: 'test', voiceName: 'test', voiceId: 'test', text });++ tts.synthesize = fake;++ tts.synthesizeUserVoice = fake;++ // The streamer's [+] modal uploads a channel sound (converted to MP3).++ const fd = new FormData();++ fd.append('command', '!honk'); fd.append('channel_id', String(streamer.id));++ fd.append('sound', new Blob([wav(0.4)], { type: 'audio/wav' }), 'honk.wav');++ const up = await fetch(`${h.base}/api/sounds`, { method: 'POST', headers: { Authorization: `Bearer ${streamer.token}` }, body: fd });++ assert.strictEqual(up.status, 200, await up.text());++ p = new parity.Parity({++ driver, streamId, sound: 'honk',++ tokens: { a: alice.token, b: bob.token, mod: mod.token, streamer: streamer.token },++ testAccounts: ['alice', 'bob', 'moddy', 'streamer'],++ });++ await p.resolve();++ assert.deepStrictEqual(Object.fromEntries(Object.entries(p.users).map(([r, u]) => [r, u.id])), { a: alice.id, b: bob.id, mod: mod.id, streamer: streamer.id });++ assert.strictEqual(p.owner, 'streamer');++ assert.strictEqual(p.ownerId, streamer.id);++ assert.deepStrictEqual(p.untested(), []);++});++++for (const s of parity.SCENARIOS) {++ t(`${s.key}: ${s.does}`, async () => {++ const r = await p.run(s);++ results.push(r);++ if (r.status === 'fail') throw r.error;++ assert.notStrictEqual(r.status, 'skip', `nothing is skipped here: ${r.note}`);++ });++}++++t('dm: delivery keeps the participant rule (chatServer.sendDm checks dm.isParticipant)', async () => {++ const conv = await h.http('POST', '/api/dm/conversations', { token: alice.token, body: { user_ids: [bob.id] } });++ const convId = conv.body.conversation.id;++ const outsider = await p.join('mod', { stream: false });++ const bobWs = await p.join('b', { stream: false });++ h.chatServer.sendDm(mod.id, { type: 'dm', conversation_id: convId, message: { message: 'not for you' } });++ h.chatServer.sendDm(bob.id, { type: 'dm', conversation_id: convId, message: { message: 'for bob' } });++ await bobWs.next((m) => m.type === 'dm' && m.message.message === 'for bob');++ assert.ok(await outsider.none((m) => m.type === 'dm'), 'a non-participant never receives it');++ await p.closeAll();++});++++t('ban from Live’s context menu (/api/mod/stream-ban → the bridge): the socket is closed, the user list drops them, they cannot talk', async () => {++ const watcher = await p.join('a');++ const daveTokens = { ...p.tokens, dave: dave.token };++ const q = new parity.Parity({ driver, tokens: daveTokens, streamId });++ q.ownerId = streamer.id;++ const daveWs = await q.join('dave');++ const listed = (m) => m.type === 'users-list' && m.users.logged.some((u) => u.username === 'dave');++ await watcher.next(listed);++ // Live writes the bans row, then asks Chat to disconnect (Live server/admin/mod-routes.js).++ h.live.addBan({ stream_id: streamId, user_id: dave.id });++ const r = await bridge([{ op: 'disconnectUser', args: [{ userId: dave.id, ip: null, streamId }] }]);++ assert.strictEqual(r.status, 200, r.text);++ await daveWs.next((m) => m.type === 'system' && m.message === 'You have been banned.');++ await new Promise((res) => (daveWs.readyState === 3 ? res() : daveWs.once('close', res)));++ await watcher.next((m) => m.type === 'users-list' && !m.users.logged.some((u) => u.username === 'dave'));++ const again = await q.join('dave');++ await q.say(again, 'let me back in');++ await q.system(again, 'You are banned from this chat.');++ assert.ok(await watcher.none((m) => m.type === 'chat' && m.message === 'let me back in'));++ await q.closeAll();++ await p.closeAll();++ h.live.clearBans();++ await h.ctx.invalidateBans();++});++++t('reconnect in global chat (Live’s floating widget, openvibe.chat): Live’s moderator delete while away, then ?after_id= converges', async () => {++ const stayer = await p.join('a', { stream: false });++ let leaver = await p.join('b', { stream: false });++ const writer = await p.join('mod', { stream: false });++ const viewA = new parity.RoomView((await h.http('GET', '/api/chat/global/history?limit=500')).body); const fromA = stayer.all.length;++ const viewB = new parity.RoomView((await h.http('GET', '/api/chat/global/history?limit=500')).body); const fromB = leaver.all.length;++ const doomed = p.text('global line a moderator deletes');++ await p.say(writer, doomed);++ const m1 = await p.saw(stayer, doomed); await p.saw(leaver, doomed);++ viewB.frames(leaver.all.slice(fromB));++ await p.left(leaver);++ // Live's /api/mod/delete-message: the row through the bridge, then the frame to the global feed.++ const r = await bridge([++ { op: 'db', args: ['deleteChatMessage', m1.id, mod.id] },++ { op: 'broadcastGlobal', args: [{ type: 'delete-messages', ids: [m1.id] }] },++ ]);++ assert.ok(r.body.results.every((x) => x.ok), r.text);++ await stayer.next((m) => m.type === 'delete-messages' && m.ids.includes(m1.id));++ const later = p.text('global line while B was away');++ await p.say(stayer, later);++ const m2 = await p.saw(stayer, later);++ viewA.frames(stayer.all.slice(fromA));++ leaver = await p.join('b', { stream: false });++ const d = (await h.http('GET', `/api/chat/global/history?after_id=${viewB.cursor}&limit=200`)).body;++ assert.ok(d.deleted_ids.includes(m1.id), 'the delete is in deleted_ids');++ viewB.delta(d);++ assert.ok(!viewB.ids.has(m1.id) && viewB.ids.has(m2.id));++ assert.deepStrictEqual(viewB.sorted(), viewA.sorted(), 'the widget converges on what the connected reader saw');++ await p.closeAll();++});++++t('the CLI: a dry run lists every scenario and opens nothing; --apply refuses accounts that are not named test accounts', async () => {++ const env = {++ OV_PARITY_BASE: h.base, OV_PARITY_STREAM: String(streamId), OV_PARITY_SOUND: 'honk',++ OV_PARITY_TOKEN_A: alice.token, OV_PARITY_TOKEN_B: bob.token, OV_PARITY_TOKEN_MOD: mod.token,++ OV_PARITY_TEST_ACCOUNTS: 'alice,bob',++ };++ const out = [];++ const sockets = h.chatServer.clients.size;++ const requests = h.live.requests.length;++ assert.strictEqual(await parity.main([], env, (l) => out.push(l)), 0);++ const text = out.join('\n');++ assert.match(text, /dry run/);++ for (const s of parity.SCENARIOS) assert.match(text, new RegExp(`\\b${s.key}\\b`), s.key);++ assert.match(text, /purge\s+destructive\s+would skip: needs OV_PARITY_TOKEN_STREAMER/);++ assert.ok(![alice.token, bob.token, mod.token].some((tok) => text.includes(tok)), 'tokens are never printed');++ assert.strictEqual(h.chatServer.clients.size, sockets, 'no connection');++ assert.strictEqual(h.live.requests.length, requests, 'nothing reached Chat');++ // --apply: moddy and the channel owner are not named → nothing runs.++ const before = h.db.get('SELECT COUNT(*) AS n FROM chat_messages').n;++ out.length = 0;++ assert.strictEqual(await parity.main(['--apply'], env, (l) => out.push(l)), 2);++ assert.match(out.join('\n'), /refused: moddy, streamer are not named in OV_PARITY_TEST_ACCOUNTS; nothing was run/);++ assert.strictEqual(h.db.get('SELECT COUNT(*) AS n FROM chat_messages').n, before, 'no message was written');++ // Named: the CLI's own network client runs a scenario end to end.++ out.length = 0;++ assert.strictEqual(await parity.main(['--apply', '--only', 'join,send'], { ...env, OV_PARITY_TEST_ACCOUNTS: 'Alice, bob, moddy, streamer' }, (l) => out.push(l)), 0, out.join('\n'));++ assert.match(out.join('\n'), /PASS join/);++ assert.match(out.join('\n'), /PASS send/);++ assert.match(out.join('\n'), /2 passed, 0 gap\(s\), 0 skipped, 0 failed/);++});++++t('the parity matrix', async () => {++ const width = Math.max(...results.map((r) => r.key.length));++ for (const r of results) console.log(` ${r.key.padEnd(width)} ${r.status.toUpperCase().padEnd(4)} ${r.note}`);++ assert.deepStrictEqual(results.filter((r) => r.status === 'gap').map((r) => r.key), ['subonly', 'blocked'], 'the known gaps (see scripts/parity.js)');++});++++t.run(async () => { if (p) await p.closeAll(); if (h) await h.close(); });
@@ -261,6 +261,7 @@ node scripts/parity-check.js --tables --live-db /tmp/live-snapshot.db --chat-db
@@ -261,6 +261,7 @@ node scripts/parity-check.js --tables --live-db /tmp/live-snapshot.db --chat-db
node scripts/table-authority.js # who writes each staged table (the handoff is Live's)
node scripts/table-authority.js # who writes each staged table (the handoff is Live's)
node scripts/mirror-flush.js # rollback helper: push queued mirror rows to Live
node scripts/mirror-flush.js # rollback helper: push queued mirror rows to Live
node scripts/subscribe-events.js --dry-run # Chat's Events subscriptions (boot creates missing ones)
node scripts/subscribe-events.js --dry-run # Chat's Events subscriptions (boot creates missing ones)
+node scripts/parity.js # chat parity scenarios: a dry run; --apply only on test accounts (docs/parity.md)
```
```
`/ready` (openvibe-shared/ready shape: `status` ready/degraded/not_ready and `checks`) is 503 only
`/ready` (openvibe-shared/ready shape: `status` ready/degraded/not_ready and `checks`) is 503 only
@@ -288,8 +289,8 @@ server/net/service-auth.js service tokens: client (Chat → others) and guard (o
@@ -288,8 +289,8 @@ server/net/service-auth.js service tokens: client (Chat → others) and guard (o
server/prefs/ chat preferences in the Network user module chat.preferences (routes, cache, migration from Live)
server/prefs/ chat preferences in the Network user module chat.preferences (routes, cache, migration from Live)
server/calls/ moved from Live: the call server (/ws/call), its REST routes, Live's stream hooks (/internal/calls), the calls lifecycle
server/calls/ moved from Live: the call server (/ws/call), its REST routes, Live's stream hooks (/internal/calls), the calls lifecycle
server/db/ schema.sql, database.js (Live's chat functions, same names and arguments)
server/db/ schema.sql, database.js (Live's chat functions, same names and arguments)
-scripts/ import-from-live, parity-check, mirror-flush, table-authority, migrate-chat-preferences, subscribe-events
-docs/ cutover.md, calls-cutover.md, staged-tables-cutover.md, live-patch.diff, capabilities-proposal/
+scripts/ import-from-live, parity-check, parity, mirror-flush, table-authority, migrate-chat-preferences, subscribe-events
+docs/ cutover.md, calls-cutover.md, staged-tables-cutover.md, parity.md, live-patch.diff, capabilities-proposal/
```
```
## Owns
## Owns
@@ -343,6 +344,7 @@ Events: `chat.message.created`, `chat.message.deleted`, `chat.dm.created`, `chat
@@ -343,6 +344,7 @@ Events: `chat.message.created`, `chat.message.deleted`, `chat.dm.created`, `chat
- restart Live without losing Chat; restart Chat's delivery plane and resume persisted messages — *seen in production (Live restarted several times on 2026-09-23 while Chat stayed up; messages and a queued outbox row survived Chat restarts); `test/restart-resume.test.js` restarts Chat as a real process (SIGTERM, new process on the same database) and proves stream, global and channel readers resume from their `after_id` cursor with no gap and no duplicate, including Live bridge placeholders that straddle the restart (kept in `bridge_refs`)*
- restart Live without losing Chat; restart Chat's delivery plane and resume persisted messages — *seen in production (Live restarted several times on 2026-09-23 while Chat stayed up; messages and a queued outbox row survived Chat restarts); `test/restart-resume.test.js` restarts Chat as a real process (SIGTERM, new process on the same database) and proves stream, global and channel readers resume from their `after_id` cursor with no gap and no duplicate, including Live bridge placeholders that straddle the restart (kept in `bridge_refs`)*
- a call row without a working signalling/media path is not parity — *Live's `/ws/call` protocol and call routes run in Chat with a `calls` row per call (`test/calls.test.js`: two signed-in sockets exchange offer/answer/ICE, limits, kick/ban, ringing → active/declined/missed/failed, stream channels); not switched over yet (`docs/calls-cutover.md`)*
- a call row without a working signalling/media path is not parity — *Live's `/ws/call` protocol and call routes run in Chat with a `calls` row per call (`test/calls.test.js`: two signed-in sockets exchange offer/answer/ICE, limits, kick/ban, ringing → active/declined/missed/failed, stream channels); not switched over yet (`docs/calls-cutover.md`)*
- a paid TTS request is never duplicated by a retry — *forwarded writes are applied once per idempotency key; paid TTS does not exist yet*
- a paid TTS request is never duplicated by a retry — *forwarded writes are applied once per idempotency key; paid TTS does not exist yet*
+- after a reconnect, deleted and blocked state converge; browser parity (join, send, DM, `/tts`, moderation, popout) — *`scripts/parity.js` and `test/parity.test.js` (`docs/parity.md`): a reader that was away converges on what a connected reader saw (cursor reads carry `deleted_ids`); gaps: sub-only mode does not exist, public chat does not apply blocks*
## Launch rule
## Launch rule
new file mode 100644
new file mode 100644
@@ -0,0 +1,81 @@
@@ -0,0 +1,81 @@
+# Chat parity script (WS-I task 6)
+
+`scripts/parity.js` runs the chat product end to end over the real protocol (`/ws/chat` and the
+REST routes Live's `chat.js` and popout use), with two or three people connected, and checks what
+each of them sees. `test/parity.test.js` runs the same scenarios against Chat booted with stubs
+(part of `npm test`), and adds what only a test can reach: Live's moderator paths through the
+bridge and the CLI's own safety checks.
+
+| Scenario | What is checked | Kind |
+| --- | --- | --- |
+| `join` | anonymous and signed-in joins get their `auth` frame; the user list names the signed-in people and counts the anonymous one | write |
+| `send` | the sender's echo and every other viewer's copy carry the same stored id; the line is in the stream history | write |
+| `dm` | the other participant's socket gets the `dm` frame, the sender's is not echoed, a non-participant gets nothing and is refused the conversation (`dm.isParticipant`) | write |
+| `tts` | `/tts` reaches the room as a `tts` line and is a request in the channel's TTS queue (`GET /api/tts/queue`); its audio frame follows when it plays | write |
+| `soundboard` | `!<sound>` gives the room the "played !sound" line and the `soundboard-audio` frame, through the queue | write |
+| `ban` | `/ban`: everyone sees the notice, the banned person's next line is refused and reaches nobody, `/unban` lifts it | destructive |
+| `timeout` | `/timeout B 2`: refused inside the two seconds, talking again after | destructive |
+| `purge` | the dashboard's range purge (ISO `from`/`to`): the room gets the `purge` frame, the range is gone from history, the line before it stays | destructive |
+| `slow` | `/slow 2`: a viewer's second line inside the window is refused and reaches nobody; the moderator's two lines 1.2 s apart both go out | destructive |
+| `subonly` | **gap**: there is no sub-only mode (below) | — |
+| `popout` | a live popout (stream join) and a channel-room popout (join by `channelUserId`) read their histories, get the stream's lines, and the channel popout's line reaches the stream; `get-users` answers | write |
+| `reconnect` | A stays, B leaves; a line is deleted and another written while B is away; B reconnects and reads `?after_id=<cursor>`: B's view equals A's, and a fresh page agrees | destructive |
+| `blocked` | **gap**: public chat does not apply blocks (below) | — |
+
+## Running it
+
+```bash
+npm test -- parity # the scenarios against Chat with stubs
+
+# Against a running Chat. Without --apply it only prints the plan (no connection is opened).
+OV_PARITY_BASE=http://127.0.0.1:4401 OV_PARITY_STREAM=<live stream of a test channel> \
+OV_PARITY_TOKEN_A=… OV_PARITY_TOKEN_B=… OV_PARITY_TOKEN_MOD=… OV_PARITY_TOKEN_STREAMER=… \
+OV_PARITY_TEST_ACCOUNTS=parity_a,parity_b,parity_mod,parity_streamer OV_PARITY_SOUND=honk \
+node scripts/parity.js [--apply] [--only join,send]
+```
+
+Roles: A and B are viewers (B is the one banned, timed out and reconnecting), MOD moderates the
+stream's channel, STREAMER owns it (needed for `purge` only). `OV_PARITY_SOUND` names a channel sound
+command (else `soundboard` is skipped); `OV_PARITY_ORIGIN` (default `https://openvibe.live`) is the
+WebSocket origin; `OV_PARITY_CHANNEL_USER_ID` saves looking up the owner's id.
+
+**Safety.** With `--apply` the script first signs each token in and reads the stream's channel
+owner, and runs **nothing** (exit 2) unless every one of those accounts is named in
+`OV_PARITY_TEST_ACCOUNTS`. Bans, timeouts, slow mode, purges and deletes then only touch dedicated
+test accounts and their own channel. Do not run it against a real channel. Tokens are never printed.
+Exit 1 when a scenario fails; gaps and skips are reported, not failed.
+
+## Known gaps (2026-09-25)
+
+- **Sub-only mode does not exist**, in Chat or in Live: no channel setting, no command, no check
+ (followers-only exists). Chat also has no subscriber lookup: subscriptions are Live's. A product
+ decision (what "sub" means: Live subscriptions, VIP membership), then a Live context read and a
+ rule in `_chatRulesBlock`.
+- **Blocks do not apply to public chat.** `network.block.changed` and `dm_blocks` are honoured for
+ DMs and calls only. A blocked person's lines reach the blocker live and in every history read,
+ before and after a reconnect, and Live's `chat.js` hides nothing either. Needs a decision (hide
+ for the blocker only? both ways? not for moderators?), then per-recipient filtering in the
+ broadcasts and the history reads.
+- **A channel-room popout's users panel is not the channel's.** `getUserList(null)` is everyone
+ outside a stream: global chat and every offline channel room. The `popout` scenario notes it.
+- **Slow mode lives in memory.** `/slow` is written to the channel's `slow_mode_seconds`, but that
+ setting is never read back: a Chat restart turns slow mode off, and the dashboard's value is not
+ enforced.
+- `/clear` only clears screens; the lines stay in history and return on a reload. A range purge
+ reaches only sockets joined to that stream (not a channel-room popout, not the global feed that
+ showed the forwarded copies).
+
+## Fixed while building it
+
+- Timeouts ended at midnight UTC: `expires_at` (ISO) was compared as text with SQLite's time.
+ `live-context.js` now reads it as an instant. (Live's own ban queries still compare text.)
+- A ban or timeout took effect only after an unawaited cache refresh, so the target's next line could
+ still go out; the refresh now finishes before the moderator is answered, also when one was
+ already in flight.
+- The dashboard's range purge, its preview and the log filter compared ISO bounds with SQLite
+ timestamps as text: a same-day range matched nothing. They use `datetime(?)`.
+- Moderators were slowed by slow mode (they could not even turn it off inside the window). The
+ room's moderators now keep only the 1 s flood limit.
+- A cursor read (`?after_id=`) never said what was deleted under the cursor, so a reader that
+ reconnected kept deleted lines. Cursor reads now carry `deleted_ids` (among the room's newest 1000
+ rows); openvibe.chat applies them. Live's floating widget (`hydrateWidgetOnly`) should too.
new file mode 100644
new file mode 100644
@@ -0,0 +1,644 @@
@@ -0,0 +1,644 @@
+#!/usr/bin/env node
+/**
+ * Chat parity (roadmap WS-I task 6): the chat product, end to end over the real protocol, with two
+ * or three connected people, checking what each of them sees. One scenario per item: join, send,
+ * DM, /tts, soundboard, moderation (ban, timeout, purge), slow mode, sub-only mode, the popout, and
+ * reconnect convergence for deleted and blocked messages.
+ *
+ * test/parity.test.js runs these same scenarios against Chat booted with stubs (and adds what only
+ * a test can see). This CLI runs them against a running Chat:
+ *
+ * OV_PARITY_BASE=http://127.0.0.1:4401 \
+ * OV_PARITY_STREAM=<live stream id of a test channel> \
+ * OV_PARITY_TOKEN_A=… OV_PARITY_TOKEN_B=… OV_PARITY_TOKEN_MOD=… [OV_PARITY_TOKEN_STREAMER=…] \
+ * OV_PARITY_TEST_ACCOUNTS=parity_a,parity_b,parity_mod,parity_streamer \
+ * [OV_PARITY_SOUND=honk] [OV_PARITY_ORIGIN=https://openvibe.live] [OV_PARITY_CHANNEL_USER_ID=…] \
+ * node scripts/parity.js [--apply] [--only join,send,…]
+ *
+ * Roles: A and B are viewers (B is the one banned, timed out and reconnecting), MOD moderates the
+ * stream's channel (a channel moderator or its owner), STREAMER owns it (the purge; optional).
+ *
+ * Without --apply this is a dry run: it prints what each scenario would do and as whom, and opens
+ * no connection. With --apply it first signs each token in (one WebSocket join each) and reads the
+ * stream's channel owner, and runs NOTHING (exit 2) unless every one of those accounts is named in
+ * OV_PARITY_TEST_ACCOUNTS: the messages, DMs, bans, timeouts, slow mode and purges then only ever
+ * touch dedicated test accounts and their own channel. Never point --apply at a real channel.
+ * Tokens are never printed. Exit 0 = no scenario failed (gaps and skips are reported, not failed).
+ */
+'use strict';
+
+const assert = require('assert');
+const crypto = require('crypto');
+
+const PACE_MS = 1150; // Chat allows one chat line per second per address and room
+const WAIT_MS = 4000; // how long a frame may take to arrive
+const QUIET_MS = 400; // how long "nobody got it" is watched
+const ROLES = ['a', 'b', 'mod', 'streamer'];
+
+class Gap extends Error { constructor(reason) { super(reason); this.status = 'gap'; } }
+class Skip extends Error { constructor(reason) { super(reason); this.status = 'skip'; } }
+
+const names = (list) => ((list && list.logged) || []).map((u) => String(u.username).toLowerCase());
+const lower = (s) => String(s || '').toLowerCase();
+
+/**
+ * One parity run: who is who, the stream, and the helpers every scenario uses.
+ *
+ * driver = {
+ * connect({ role, stream }) → a socket to /ws/chat (stream = the ?stream= id or null) with
+ * .all (frames received), .next(pred, ms) (the first frame matching, not taken before),
+ * .none(pred, ms) (true when none arrives), .sendJson(obj), .close();
+ * http(method, path, { token, body }) → { status, body, text };
+ * address(role) → the address that role's sockets come from (Chat's rate limit key).
+ * }
+ */
+class Parity {
+ constructor({ driver, tokens, streamId, channelUserId = null, sound = null, testAccounts = [], log = () => {} }) {
+ this.driver = driver;
+ this.tokens = tokens;
+ this.streamId = Number(streamId);
+ this.ownerId = channelUserId ? Number(channelUserId) : null;
+ this.owner = null;
+ this.sound = sound ? String(sound).replace(/^!/, '').toLowerCase() : null;
+ this.testAccounts = testAccounts.map(lower);
+ this.log = log;
+ this.users = {};
+ this.nonce = crypto.randomBytes(3).toString('hex');
+ this.sockets = new Set();
+ this.lastSend = new Map();
+ }
+
+ /** Sign each token in (a join) and find the stream's channel owner. */
+ async resolve() {
+ for (const role of ROLES) {
+ if (!this.tokens[role]) continue;
+ const ws = await this.driver.connect({ role, stream: null });
+ try {
+ ws.sendJson({ type: 'join', token: this.tokens[role] });
+ const auth = await ws.next((m) => m.type === 'auth', WAIT_MS);
+ if (!auth.authenticated) throw new Error(`the token for ${role} did not sign in`);
+ this.users[role] = { id: auth.user_id, username: auth.core_username, display: auth.username, role: auth.role };
+ } finally { ws.close(); }
+ }
+ const hist = await this.http('GET', `/api/chat/${this.streamId}/history?limit=1`);
+ if (hist.status !== 200 || !hist.body || !hist.body.channel) throw new Error(`stream ${this.streamId}: no channel (${hist.status})`);
+ this.owner = hist.body.channel;
+ if (!this.ownerId) {
+ const prof = await this.http('GET', `/api/chat/user/${encodeURIComponent(this.owner)}/profile`);
+ if (prof.status !== 200 || !prof.body || !prof.body.id) throw new Error(`no user id for the channel owner (${prof.status}); set OV_PARITY_CHANNEL_USER_ID`);
+ this.ownerId = Number(prof.body.id);
+ }
+ }
+
+ /** Accounts --apply would touch that are not named as test accounts. */
+ untested() {
+ const touched = [...Object.values(this.users).map((u) => u.username), this.owner];
+ return [...new Set(touched.filter((n) => !this.testAccounts.includes(lower(n))))];
+ }
+
+ text(label) { return `[parity ${this.nonce}] ${label}`; }
+ sleep(ms) { return new Promise((r) => setTimeout(r, ms)); }
+ gap(reason) { throw new Gap(reason); }
+ skip(reason) { throw new Skip(reason); }
+
+ http(method, path, { as = null, body } = {}) {
+ return this.driver.http(method, path, { token: as ? this.tokens[as] : null, body });
+ }
+
+ /**
+ * Open /ws/chat and join the way Live's chat.js does (the token only in the join message):
+ * stream → ?stream=<id> and join { streamId }; channel → join { channelUserId } (the channel
+ * room, as the popout and an offline channel page do). role 'anon' joins without a token.
+ */
+ async join(role, { stream = true, channel = false } = {}) {
+ const ws = await this.driver.connect({ role, stream: stream ? this.streamId : null });
+ this.sockets.add(ws);
+ ws.on('close', () => this.sockets.delete(ws));
+ ws.role = role;
+ ws.rateKey = `${this.driver.address(role)}|${stream ? this.streamId : 'global'}`;
+ ws.sendJson({
+ type: 'join',
+ ...(stream ? { streamId: this.streamId } : {}),
+ ...(channel ? { channelUserId: this.ownerId } : {}),
+ ...(this.tokens[role] ? { token: this.tokens[role] } : {}),
+ });
+ ws.auth = await ws.next((m) => m.type === 'auth', WAIT_MS);
+ return ws;
+ }
+
+ /** Send a chat line, one per PACE_MS per address and room (Chat's flood limit). */
+ async say(ws, message) {
+ const wait = (this.lastSend.get(ws.rateKey) || 0) + PACE_MS - Date.now();
+ if (wait > 0) await this.sleep(wait);
+ ws.sendJson({ type: 'chat', message });
+ this.lastSend.set(ws.rateKey, Date.now());
+ }
+
+ saw(ws, message, ms = WAIT_MS) { return ws.next((m) => m.type === 'chat' && m.message === message, ms); }
+ system(ws, message, ms = WAIT_MS) { return ws.next((m) => m.type === 'system' && m.message === message, ms); }
+ async left(ws) {
+ if (ws.readyState === 3) return;
+ await new Promise((r) => { ws.once('close', r); ws.close(); });
+ }
+ async closeAll() { await Promise.all([...this.sockets].map((ws) => this.left(ws).catch(() => {}))); }
+
+ /** Poll fn until it returns something truthy. */
+ async until(fn, ms = WAIT_MS) {
+ const end = Date.now() + ms;
+ for (;;) {
+ const v = await fn();
+ if (v) return v;
+ if (Date.now() > end) return null;
+ await this.sleep(150);
+ }
+ }
+
+ /** Run one scenario → { key, status: 'pass' | 'gap' | 'skip' | 'fail', note, error }. */
+ async run(scenario) {
+ const started = Date.now();
+ try {
+ for (const r of scenario.needs || []) if (!this.tokens[r]) this.skip(`needs OV_PARITY_TOKEN_${r.toUpperCase()}`);
+ const note = await scenario.run(this);
+ return { key: scenario.key, status: 'pass', note: note || '', ms: Date.now() - started };
+ } catch (err) {
+ if (err instanceof Gap || err instanceof Skip) return { key: scenario.key, status: err.status, note: err.message, ms: Date.now() - started };
+ return { key: scenario.key, status: 'fail', note: err && err.message, error: err, ms: Date.now() - started };
+ } finally {
+ await this.closeAll();
+ }
+ }
+}
+
+/**
+ * A reader's view of a room, the way chat.js keeps it: the ids of the messages on screen, from a
+ * history page, the live frames (chat lines in, delete-messages out) and cursor reads (?after_id=:
+ * new rows in, deleted_ids out).
+ */
+class RoomView {
+ constructor(page) {
+ this.ids = new Set((page.messages || []).map((m) => m.id));
+ this.cursor = page.latest_id || 0;
+ }
+ frames(list) {
+ for (const m of list) {
+ if (m.type === 'chat' && Number.isInteger(m.id)) { this.ids.add(m.id); this.cursor = Math.max(this.cursor, m.id); }
+ if (m.type === 'delete-messages') for (const id of m.ids || []) this.ids.delete(id);
+ }
+ return this;
+ }
+ delta(d) {
+ for (const m of d.messages || []) this.ids.add(m.id);
+ for (const id of d.deleted_ids || []) this.ids.delete(id);
+ this.cursor = Math.max(this.cursor, d.latest_id || 0);
+ return this;
+ }
+ sorted() { return [...this.ids].sort((x, y) => x - y); }
+}
+
+// ── The scenarios ────────────────────────────────────────────────────────────────────────────
+// kind: 'write' posts into the test channel or DMs; 'destructive' bans, times out, deletes or
+// changes the channel's chat mode. Both run only with --apply and only on test accounts. 'gap': the
+// product has no such behaviour yet; the scenario reports why and touches nothing.
+
+const SCENARIOS = [
+ {
+ key: 'join', kind: 'write', as: ['anon', 'a', 'b'], needs: ['a', 'b'],
+ does: 'anonymous and signed-in sockets join the stream; each gets its auth frame; the user list names A and B',
+ async run(p) {
+ const anon = await p.join('anon');
+ assert.strictEqual(anon.auth.authenticated, false, 'anonymous join');
+ assert.match(String(anon.auth.username), /^anon\d+$/, 'an anon number');
+ assert.strictEqual(anon.auth.role, 'anon');
+ const a = await p.join('a');
+ assert.strictEqual(a.auth.authenticated, true, 'signed-in join');
+ assert.strictEqual(a.auth.core_username, p.users.a.username);
+ assert.strictEqual(a.auth.user_id, p.users.a.id);
+ const b = await p.join('b');
+ const both = (m) => m.type === 'users-list' && names(m.users).includes(lower(p.users.a.username)) && names(m.users).includes(lower(p.users.b.username));
+ anon.sendJson({ type: 'get-users' });
+ const list = await anon.next(both, WAIT_MS);
+ assert.ok(list.users.anonCount >= 1, 'the anonymous viewer is counted');
+ await a.next(both, WAIT_MS); // pushed to everyone in the room, not only on request
+ return `anon ${anon.auth.username}; list ${list.users.logged.length} named, ${list.users.anonCount} anon`;
+ },
+ },
+ {
+ key: 'send', kind: 'write', as: ['a', 'b', 'anon'], needs: ['a', 'b'],
+ does: 'A sends a line: A gets its own echo, B and an anonymous viewer get it with the same id; it is in the stream history',
+ async run(p) {
+ const a = await p.join('a'); const b = await p.join('b'); const anon = await p.join('anon');
+ const text = p.text('hello from A');
+ await p.say(a, text);
+ const echo = await p.saw(a, text);
+ const got = await p.saw(b, text);
+ const gotAnon = await p.saw(anon, text);
+ assert.ok(Number.isInteger(echo.id), 'the echo carries the stored id');
+ assert.strictEqual(got.id, echo.id); assert.strictEqual(gotAnon.id, echo.id);
+ assert.strictEqual(got.core_username, p.users.a.username);
+ assert.strictEqual(got.user_id, p.users.a.id);
+ assert.strictEqual(got.stream_id, p.streamId);
+ const hist = await p.http('GET', `/api/chat/${p.streamId}/history?limit=50`);
+ assert.ok(hist.body.messages.some((m) => m.id === echo.id && m.message === text), 'in the stream history');
+ return `message ${echo.id}`;
+ },
+ },
+ {
+ key: 'dm', kind: 'write', as: ['a', 'b', 'mod'], needs: ['a', 'b', 'mod'],
+ does: 'A opens a DM with B and writes: B’s socket gets it, A’s socket is not echoed, MOD (not a participant) gets nothing and is refused the conversation',
+ async run(p) {
+ const a = await p.join('a'); const b = await p.join('b'); const x = await p.join('mod');
+ const conv = await p.http('POST', '/api/dm/conversations', { as: 'a', body: { user_ids: [p.users.b.id] } });
+ assert.strictEqual(conv.status, 200, conv.text);
+ const convId = conv.body.conversation.id;
+ assert.deepStrictEqual(conv.body.conversation.participants.map((u) => u.id).sort(), [p.users.a.id, p.users.b.id].sort());
+ const text = p.text('a private hello');
+ const sent = await p.http('POST', `/api/dm/conversations/${convId}/messages`, { as: 'a', body: { message: text } });
+ assert.strictEqual(sent.status, 200, sent.text);
+ const got = await b.next((m) => m.type === 'dm' && m.conversation_id === convId && m.message && m.message.message === text);
+ assert.strictEqual(got.message.sender_id, p.users.a.id);
+ const isIt = (m) => m.type === 'dm' && m.message && m.message.message === text;
+ assert.ok(await a.none(isIt, QUIET_MS), 'the sender’s socket is not echoed (the REST answer carries it)');
+ assert.ok(await x.none(isIt, QUIET_MS), 'delivered to participants only');
+ assert.strictEqual((await p.http('GET', `/api/dm/conversations/${convId}/messages`, { as: 'mod' })).status, 403, 'a non-participant cannot read it');
+ const inbox = await p.http('GET', `/api/dm/conversations/${convId}/messages`, { as: 'b' });
+ assert.ok(inbox.body.messages.some((m) => m.message === text), 'B reads it');
+ return `conversation ${convId}`;
+ },
+ },
+ {
+ key: 'tts', kind: 'write', as: ['a', 'b', 'mod'], needs: ['a', 'b', 'mod'],
+ does: 'A sends /tts: the room gets the tts line and the request is in the channel’s TTS queue (read as MOD); the audio frame follows when it plays',
+ async run(p) {
+ const a = await p.join('a'); const b = await p.join('b');
+ const text = p.text('read this out');
+ await p.say(a, `/tts ${text}`);
+ const line = await b.next((m) => m.type === 'tts' && m.message === text);
+ assert.strictEqual(line.core_username, p.users.a.username);
+ const req = await p.until(async () => {
+ const q = await p.http('GET', `/api/tts/queue?stream_id=${p.streamId}&recent=50`, { as: 'mod' });
+ assert.strictEqual(q.status, 200, q.text);
+ return [q.body.playing, ...q.body.queued, ...q.body.recent].filter(Boolean).find((r) => r.kind === 'tts' && r.label === text);
+ });
+ assert.ok(req, 'the /tts request is in the room’s queue');
+ assert.strictEqual(req.requested_by, p.users.a.display);
+ let audio = 'not played yet';
+ if (req.state === 'played' || req.state === 'playing') {
+ const f = await b.next((m) => m.type === 'tts-audio' && m.message === text, 8000);
+ assert.ok(f.audio, 'the audio frame carries audio');
+ audio = 'audio frame received';
+ }
+ return `queue request ${req.id} (${req.state}); ${audio}`;
+ },
+ },
+ {
+ key: 'soundboard', kind: 'write', as: ['a', 'b', 'mod'], needs: ['a', 'b', 'mod'],
+ does: 'A types !<sound>: the room gets the “played !sound” line and the soundboard-audio frame; the channel’s queue has it',
+ async run(p) {
+ if (!p.sound) p.skip('no channel sound command (OV_PARITY_SOUND)');
+ const a = await p.join('a'); const b = await p.join('b');
+ await p.say(a, `!${p.sound}`);
+ const isAnnounce = (m) => m.type === 'chat' && m.message_type === 'channel-sound' && m.message === `played !${p.sound}`;
+ const ann = await b.next(isAnnounce);
+ assert.strictEqual(ann.core_username, p.users.a.username);
+ await a.next(isAnnounce);
+ const audio = await b.next((m) => m.type === 'soundboard-audio' && m.title === `!${p.sound}`, 8000);
+ assert.ok(audio.audio && audio.audio.length > 0, 'the clip is in the frame');
+ const q = await p.http('GET', `/api/tts/queue?stream_id=${p.streamId}&recent=50`, { as: 'mod' });
+ const req = [q.body.playing, ...q.body.queued, ...q.body.recent].filter(Boolean).find((r) => r.kind === 'channel-sound' && r.label === `!${p.sound}`);
+ assert.ok(req, 'the sound went through the room’s queue');
+ return `!${p.sound}: announce ${ann.id}, queue request ${req.id} (${req.state})`;
+ },
+ },
+ {
+ key: 'ban', kind: 'destructive', as: ['mod', 'a', 'b'], needs: ['a', 'b', 'mod'],
+ does: 'MOD /ban B in the stream: everyone sees the notice; B’s next line is refused and reaches nobody; /unban lifts it',
+ async run(p) {
+ const a = await p.join('a'); const b = await p.join('b'); const mod = await p.join('mod');
+ const target = p.users.b.username;
+ await p.say(mod, `/ban ${target}`);
+ await p.system(a, `${target} has been banned.`);
+ await p.system(b, `${target} has been banned.`);
+ try {
+ const text = p.text('can B still talk?');
+ await p.say(b, text);
+ await p.system(b, 'You are banned from this chat.');
+ assert.ok(await a.none((m) => m.type === 'chat' && m.message === text, QUIET_MS), 'a banned line reaches nobody');
+ } finally {
+ await p.say(mod, `/unban ${target}`);
+ await p.system(mod, `${target} has been unbanned.`);
+ }
+ const back = p.text('B is back');
+ await p.say(b, back);
+ await p.saw(a, back);
+ return `${target} banned and unbanned`;
+ },
+ },
+ {
+ key: 'timeout', kind: 'destructive', as: ['mod', 'a', 'b'], needs: ['a', 'b', 'mod'],
+ does: 'MOD /timeout B 2: B’s line inside the 2 s is refused; after it B’s line reaches A',
+ async run(p) {
+ const a = await p.join('a'); const b = await p.join('b'); const mod = await p.join('mod');
+ const target = p.users.b.username;
+ const secs = 2;
+ await p.say(mod, `/timeout ${target} ${secs}`);
+ await p.system(mod, `${target} timed out for ${secs}s.`);
+ const t0 = Date.now();
+ const during = p.text('timed out?');
+ await p.say(b, during);
+ await p.system(b, 'You are banned from this chat.');
+ assert.ok(await a.none((m) => m.type === 'chat' && m.message === during, QUIET_MS), 'a timed-out line reaches nobody');
+ await p.sleep(Math.max(0, t0 + secs * 1000 + 700 - Date.now()));
+ const after = p.text('the timeout is over');
+ await p.say(b, after);
+ await p.saw(a, after);
+ return `${target} timed out for ${secs}s, then talked again`;
+ },
+ },
+ {
+ key: 'purge', kind: 'destructive', as: ['streamer', 'a', 'b'], needs: ['a', 'b', 'streamer'],
+ does: 'STREAMER purges a time range of the stream (the dashboard’s ISO range): A and B get the purge frame; the lines in the range are gone from history, the one before stays',
+ async run(p) {
+ const a = await p.join('a'); const b = await p.join('b');
+ const before = p.text('before the purge');
+ await p.say(a, before);
+ const kept = await p.saw(b, before);
+ // The range starts after `before` and is built from the server's own timestamps, so a
+ // client clock that is off does not matter. Rows keep whole seconds.
+ const from = new Date(Date.parse(kept.timestamp) + 1500).toISOString();
+ await p.sleep(2700);
+ const g1 = p.text('purge me (A)'); const g2 = p.text('purge me (B)');
+ await p.say(a, g1); const m1 = await p.saw(b, g1);
+ await p.say(b, g2); const m2 = await p.saw(a, g2);
+ assert.ok(Date.parse(m1.timestamp) >= Date.parse(from) + 1000, 'the purged lines are inside the range');
+ const to = new Date(Date.parse(m2.timestamp) + 1500).toISOString();
+ const range = { streamId: p.streamId, from, to };
+ const preview = await p.http('POST', '/api/chat/admin/purge/preview', { as: 'streamer', body: range });
+ assert.strictEqual(preview.status, 200, preview.text);
+ assert.ok(preview.body.count >= 2, `the preview counts the range (${preview.body.count})`);
+ const r = await p.http('DELETE', '/api/chat/admin/purge', { as: 'streamer', body: range });
+ assert.strictEqual(r.status, 200, r.text);
+ assert.ok(r.body.deleted >= 2, `purged ${r.body.deleted}`);
+ for (const ws of [a, b]) await ws.next((m) => m.type === 'purge' && m.from === from && m.to === to);
+ const ids = (await p.http('GET', `/api/chat/${p.streamId}/history?limit=200`)).body.messages.map((m) => m.id);
+ assert.ok(!ids.includes(m1.id) && !ids.includes(m2.id), 'the range is gone from history');
+ assert.ok(ids.includes(kept.id), 'the line before the range stays');
+ return `${r.body.deleted} purged`;
+ },
+ },
+ {
+ key: 'slow', kind: 'destructive', as: ['mod', 'a', 'b'], needs: ['a', 'b', 'mod'],
+ does: 'MOD /slow 2: A’s second line inside 2 s is refused and reaches nobody; MOD’s two lines 1.2 s apart both go out (moderators are exempt); /slow off',
+ async run(p) {
+ const a = await p.join('a'); const b = await p.join('b'); const mod = await p.join('mod');
+ const secs = 2;
+ await p.say(mod, `/slow ${secs}`);
+ assert.strictEqual((await a.next((m) => m.type === 'slowmode')).seconds, secs);
+ await p.system(a, `Slow mode enabled: ${secs}s between messages`);
+ try {
+ // Everyone may share one address (the CLI): let the /slow line's window pass first.
+ await p.sleep(secs * 1000 + 200);
+ const one = p.text('slow one'); const two = p.text('slow two');
+ await p.say(a, one);
+ await p.saw(b, one);
+ await p.say(a, two);
+ await p.system(a, 'Slow down! You are sending messages too fast.');
+ assert.ok(await b.none((m) => m.type === 'chat' && m.message === two, QUIET_MS), 'a slowed line reaches nobody');
+ const m1 = p.text('mod one'); const m2 = p.text('mod two');
+ await p.say(mod, m1); await p.saw(b, m1);
+ await p.say(mod, m2); await p.saw(b, m2);
+ } finally {
+ // Turn it off whatever happened; a Chat that slows moderators refuses the first try.
+ await p.say(mod, '/slow off');
+ const off = await Promise.race([
+ p.system(a, 'Slow mode disabled.').then(() => true),
+ mod.next((m) => m.type === 'system' && /^Slow down!/.test(m.message)).then(() => false),
+ ]).catch(() => false);
+ if (!off) {
+ await p.sleep(secs * 1000 + 200);
+ await p.say(mod, '/slow off');
+ await p.system(a, 'Slow mode disabled.');
+ }
+ }
+ return `viewer slowed, moderator exempt (${secs}s)`;
+ },
+ },
+ {
+ key: 'subonly', kind: 'gap', as: [], needs: [],
+ does: 'sub-only mode: a non-subscriber is refused, a subscriber may talk',
+ async run(p) {
+ p.gap('no sub-only mode exists: no channel setting, no command and no check in Chat or Live (followers-only exists), and Chat has no subscriber lookup in live-context (subscriptions are Live’s)');
+ },
+ },
+ {
+ key: 'popout', kind: 'write', as: ['a', 'b', 'mod'], needs: ['a', 'b', 'mod'],
+ does: 'Live’s popout (popout-chat.html + chat.js): a live popout (stream join) and a channel-room popout (join by channelUserId) read their histories, get stream lines, and the channel popout’s line reaches the stream; get-users answers',
+ async run(p) {
+ const pop = await p.join('a', { stream: true, channel: true });
+ const chan = await p.join('b', { stream: false, channel: true });
+ const viewer = await p.join('mod');
+ assert.strictEqual(chan.auth.authenticated, true);
+ const sHist = await p.http('GET', `/api/chat/${p.streamId}/history?limit=500`);
+ assert.strictEqual(sHist.status, 200); assert.strictEqual(lower(sHist.body.channel), lower(p.owner));
+ assert.strictEqual((await p.http('GET', '/api/chat/global/history?limit=500')).status, 200);
+ const cHist = await p.http('GET', `/api/chat/channel/${p.ownerId}/history?limit=500`);
+ assert.strictEqual(cHist.status, 200); assert.strictEqual(lower(cHist.body.channel), lower(p.owner));
+ const t1 = p.text('to the popouts');
+ await p.say(viewer, t1);
+ const f1 = await p.saw(pop, t1); await p.saw(chan, t1);
+ const t2 = p.text('from the channel popout');
+ await p.say(chan, t2);
+ const f2 = await p.saw(viewer, t2); await p.saw(pop, t2);
+ assert.strictEqual(f2.channel_user_id, p.ownerId);
+ const ids = (await p.http('GET', `/api/chat/channel/${p.ownerId}/history?limit=50`)).body.messages.map((m) => m.id);
+ assert.ok(ids.includes(f1.id) && ids.includes(f2.id), 'both lines are in the channel history a popout reloads');
+ pop.sendJson({ type: 'get-users' });
+ await pop.next((m) => m.type === 'users-list' && names(m.users).includes(lower(p.users.a.username)));
+ // The channel-room popout's users panel: is it the channel's? (MOD is also in global chat.)
+ await p.join('mod', { stream: false });
+ chan.sendJson({ type: 'get-users' });
+ const list = await chan.next((m) => m.type === 'users-list' && names(m.users).includes(lower(p.users.b.username)) && names(m.users).includes(lower(p.users.mod.username)), 1500).catch(() => null);
+ return `live and channel-room popouts${list ? '; GAP: the channel-room popout’s users panel also names people in global chat (getUserList(null) is everyone outside a stream)' : ''}`;
+ },
+ },
+ {
+ key: 'reconnect', kind: 'destructive', as: ['a', 'b', 'mod'], needs: ['a', 'b', 'mod'],
+ does: 'A stays connected, B leaves; MOD’s line is deleted (MOD’s own delete-all) and A writes while B is away; B reconnects and reads ?after_id=<cursor>: B’s view equals A’s, and a fresh page agrees',
+ async run(p) {
+ const room = `/api/chat/channel/${p.ownerId}/history`;
+ const stayer = await p.join('a'); let leaver = await p.join('b'); const writer = await p.join('mod');
+ const viewA = new RoomView((await p.http('GET', `${room}?limit=500`)).body); const fromA = stayer.all.length;
+ const viewB = new RoomView((await p.http('GET', `${room}?limit=500`)).body); const fromB = leaver.all.length;
+ const doomed = p.text('this line will be deleted');
+ await p.say(writer, doomed);
+ const m1 = await p.saw(stayer, doomed); await p.saw(leaver, doomed);
+ viewB.frames(leaver.all.slice(fromB));
+ await p.left(leaver);
+ // While B is away: a deletion, and a new line.
+ writer.sendJson({ type: 'self-delete-history' });
+ await writer.next((m) => m.type === 'self-delete-result');
+ await stayer.next((m) => m.type === 'delete-messages' && (m.ids || []).includes(m1.id));
+ const meanwhile = p.text('written while B was away');
+ await p.say(stayer, meanwhile);
+ const m2 = await p.saw(stayer, meanwhile);
+ viewA.frames(stayer.all.slice(fromA));
+ // B comes back: a new socket, then the cursor read chat.js makes.
+ leaver = await p.join('b');
+ const d = await p.http('GET', `${room}?after_id=${viewB.cursor}`);
+ assert.strictEqual(d.status, 200); assert.strictEqual(d.body.complete, true);
+ viewB.delta(d.body);
+ assert.ok(!viewB.ids.has(m1.id), 'the line deleted while B was away is gone from B’s view');
+ assert.ok(viewB.ids.has(m2.id), 'the line written while B was away is in B’s view');
+ assert.deepStrictEqual(viewB.sorted(), viewA.sorted(), 'B converges on what A saw');
+ // A fresh page (the main pane's reload) agrees with both, over the rows it covers.
+ const fresh = (await p.http('GET', `${room}?limit=500`)).body.messages.map((m) => m.id);
+ const lo = Math.min(...fresh);
+ assert.deepStrictEqual(fresh.slice().sort((x, y) => x - y), viewA.sorted().filter((id) => id >= lo), 'a fresh page shows the same');
+ return `deleted ${m1.id} dropped via deleted_ids; ${m2.id} added via the delta`;
+ },
+ },
+ {
+ key: 'blocked', kind: 'gap', as: [], needs: [],
+ does: 'reconnect convergence for blocked messages: lines from someone the reader blocked (network blocks) are hidden live and after a reconnect',
+ async run(p) {
+ p.gap('public chat does not apply blocks: network.block.changed and dm_blocks are honoured for DMs and calls only (server/chat/network-blocks.js, dm.js isBlockedEither); room broadcasts and history pages show a blocked person’s lines to the blocker, live and after a reconnect alike, and Live’s chat.js hides nothing either');
+ },
+ },
+];
+
+// ── The command line ─────────────────────────────────────────────────────────────────────────
+
+/** A socket with the frame inbox the scenarios read (the same semantics as test/helpers.js h.ws). */
+function openSocket(url, headers) {
+ const WebSocket = require('ws');
+ return new Promise((resolve, reject) => {
+ const ws = new WebSocket(url, { headers });
+ const all = [];
+ const waiters = [];
+ ws.on('message', (d) => {
+ let m; try { m = JSON.parse(d.toString()); } catch { return; }
+ all.push(m);
+ for (const w of [...waiters]) if (w.pred(m)) { waiters.splice(waiters.indexOf(w), 1); clearTimeout(w.timer); w.resolve(m); }
+ });
+ ws.all = all;
+ ws.next = (pred, ms = WAIT_MS) => {
+ const hit = all.find((m) => pred(m) && !m.__taken);
+ if (hit) { hit.__taken = true; return Promise.resolve(hit); }
+ return new Promise((res, rej) => {
+ const w = { pred: (m) => { if (pred(m)) { m.__taken = true; return true; } return false; }, resolve: res };
+ w.timer = setTimeout(() => { waiters.splice(waiters.indexOf(w), 1); rej(new Error(`timed out waiting (last frames: ${JSON.stringify(all.slice(-8).map((x) => [x.type, x.message]))})`)); }, ms);
+ waiters.push(w);
+ });
+ };
+ ws.none = async (pred, ms = QUIET_MS) => { await new Promise((r) => setTimeout(r, ms)); return !all.some(pred); };
+ ws.sendJson = (o) => ws.send(JSON.stringify(o));
+ ws.once('open', () => resolve(ws));
+ ws.once('error', reject);
+ ws.once('unexpected-response', (req, res) => reject(new Error(`upgrade refused ${res.statusCode}`)));
+ });
+}
+
+function networkDriver({ base, origin }) {
+ const wsBase = base.replace(/^http/, 'ws');
+ return {
+ address: () => 'this machine',
+ connect: ({ stream }) => openSocket(`${wsBase}/ws/chat${stream ? `?stream=${stream}` : ''}`, { origin }),
+ async http(method, path, { token, body } = {}) {
+ const res = await fetch(`${base}${path}`, {
+ method,
+ headers: { Accept: 'application/json', ...(body !== undefined ? { 'Content-Type': 'application/json' } : {}), ...(token ? { Authorization: `Bearer ${token}` } : {}) },
+ body: body !== undefined ? JSON.stringify(body) : undefined,
+ signal: AbortSignal.timeout(15000),
+ });
+ const text = await res.text();
+ let json = null; try { json = JSON.parse(text); } catch { /* not JSON */ }
+ return { status: res.status, body: json, text };
+ },
+ };
+}
+
+function parseArgs(argv) {
+ const o = { apply: false, only: null };
+ for (let i = 0; i < argv.length; i++) {
+ const a = argv[i];
+ if (a === '--apply') o.apply = true;
+ else if (a === '--only') { o.only = String(argv[++i] || '').split(',').map((s) => s.trim()).filter(Boolean); }
+ else if (a === '--help' || a === '-h') o.help = true;
+ else throw new Error(`unknown argument ${a}`);
+ }
+ return o;
+}
+
+function configFrom(env) {
+ const base = String(env.OV_PARITY_BASE || '').replace(/\/+$/, '');
+ return {
+ base,
+ origin: env.OV_PARITY_ORIGIN || 'https://openvibe.live',
+ streamId: parseInt(env.OV_PARITY_STREAM, 10) || null,
+ channelUserId: parseInt(env.OV_PARITY_CHANNEL_USER_ID, 10) || null,
+ sound: env.OV_PARITY_SOUND || null,
+ tokens: { a: env.OV_PARITY_TOKEN_A || null, b: env.OV_PARITY_TOKEN_B || null, mod: env.OV_PARITY_TOKEN_MOD || null, streamer: env.OV_PARITY_TOKEN_STREAMER || null },
+ testAccounts: String(env.OV_PARITY_TEST_ACCOUNTS || '').split(',').map((s) => s.trim()).filter(Boolean),
+ };
+}
+
+const LABEL = { pass: 'PASS', gap: 'GAP ', skip: 'SKIP', fail: 'FAIL' };
+
+/** → exit code: 0 ok, 1 a scenario failed, 2 refused (configuration, or accounts not named as test accounts). */
+async function main(argv = process.argv.slice(2), env = process.env, log = console.log, { driver = null } = {}) {
+ let o;
+ try { o = parseArgs(argv); } catch (err) { log(err.message); return 2; }
+ if (o.help) { log(require('fs').readFileSync(__filename, 'utf8').split('\n').slice(1, 32).join('\n')); return 0; }
+ const cfg = configFrom(env);
+ const chosen = SCENARIOS.filter((s) => !o.only || o.only.includes(s.key));
+ if (o.only && chosen.length !== o.only.length) { log(`unknown scenario in --only (known: ${SCENARIOS.map((s) => s.key).join(', ')})`); return 2; }
+ const problems = [];
+ if (!cfg.base && !driver) problems.push('OV_PARITY_BASE is required');
+ if (!cfg.streamId) problems.push('OV_PARITY_STREAM is required (a live stream of a test channel)');
+ const have = ROLES.map((r) => `${r} ${cfg.tokens[r] ? 'set' : '-'}`).join(', ');
+
+ if (!o.apply) {
+ log('Chat parity: dry run. Nothing is sent and no connection is opened; --apply runs it.');
+ log(` base ${cfg.base || '(none)'} origin ${cfg.origin} stream ${cfg.streamId || '(none)'} sound ${cfg.sound ? `!${cfg.sound}` : '(none)'}`);
+ log(` tokens: ${have}`);
+ log(` test accounts (OV_PARITY_TEST_ACCOUNTS): ${cfg.testAccounts.join(', ') || '(none)'}`);
+ for (const p of problems) log(` missing: ${p}`);
+ log('');
+ for (const s of chosen) {
+ const missing = (s.needs || []).filter((r) => !cfg.tokens[r]);
+ const would = s.kind === 'gap' ? 'reports a gap, touches nothing'
+ : missing.length ? `would skip: needs ${missing.map((r) => `OV_PARITY_TOKEN_${r.toUpperCase()}`).join(', ')}`
+ : s.key === 'soundboard' && !cfg.sound ? 'would skip: needs OV_PARITY_SOUND' : `as ${s.as.join(', ')}`;
+ log(` ${s.key.padEnd(11)} ${s.kind.padEnd(11)} ${would}`);
+ log(` ${s.does}`);
+ }
+ log('');
+ log('With --apply every account behind the tokens, and the stream’s channel owner, must be named in');
+ log('OV_PARITY_TEST_ACCOUNTS, or nothing runs.');
+ return problems.length ? 2 : 0;
+ }
+
+ if (problems.length) { for (const p of problems) log(p); return 2; }
+ const p = new Parity({ driver: driver || networkDriver(cfg), tokens: cfg.tokens, streamId: cfg.streamId, channelUserId: cfg.channelUserId, sound: cfg.sound, testAccounts: cfg.testAccounts, log });
+ try { await p.resolve(); } catch (err) { log(`could not resolve the accounts and the stream: ${err.message}`); return 2; }
+ const outside = p.untested();
+ if (outside.length) {
+ log(`refused: ${outside.join(', ')} ${outside.length === 1 ? 'is' : 'are'} not named in OV_PARITY_TEST_ACCOUNTS; nothing was run.`);
+ return 2;
+ }
+ log(`Chat parity against ${cfg.base || 'the given driver'}: stream ${p.streamId} (channel ${p.owner}), run ${p.nonce}`);
+ const results = [];
+ for (const s of chosen) {
+ const r = await p.run(s);
+ results.push(r);
+ log(` ${LABEL[r.status]} ${s.key.padEnd(11)} ${r.note || ''}`);
+ }
+ const count = (st) => results.filter((r) => r.status === st).length;
+ log(`\n${count('pass')} passed, ${count('gap')} gap(s), ${count('skip')} skipped, ${count('fail')} failed`);
+ return count('fail') ? 1 : 0;
+}
+
+module.exports = { SCENARIOS, Parity, RoomView, Gap, Skip, openSocket, networkDriver, main, PACE_MS };
+
+if (require.main === module) {
+ main().then((code) => process.exit(code), (err) => { console.error(err && err.stack || err); process.exit(1); });
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,195 @@
@@ -0,0 +1,195 @@
+'use strict';
+/**
+ * Chat parity (roadmap WS-I task 6). scripts/parity.js's scenarios (join, send, DM, /tts,
+ * soundboard, ban, timeout, purge, slow mode, sub-only mode, the popout, reconnect convergence for
+ * deleted and blocked messages) run here against Chat booted with stubs, with two or three people
+ * connected, each from their own address: the same code the CLI runs against a running Chat.
+ *
+ * On top, what only a test can reach: Live's moderator paths through the bridge (the context-menu
+ * ban that disconnects, the single-message delete) and the global feed's cursor read (Live's
+ * floating widget, openvibe.chat), the DM participant rule on delivery, and the CLI itself: its dry
+ * run touches nothing, and --apply refuses to run for an account that is not a named test account.
+ *
+ * A scenario the product does not support ends as a gap with its reason (sub-only mode, blocks in
+ * public chat); nothing here asserts that it works. The matrix is printed at the end.
+ */
+const assert = require('assert');
+const { boot, suite } = require('./helpers');
+const parity = require('../scripts/parity');
+
+const t = suite('parity');
+let h, p, streamer, mod, alice, bob, dave, channelId, streamId;
+const IPS = { a: '198.51.100.101', b: '198.51.100.102', mod: '198.51.100.103', streamer: '198.51.100.104', anon: '198.51.100.105', dave: '198.51.100.106' };
+const results = [];
+let bridgeSeq = 0;
+const bridge = (ops) => h.http('POST', '/internal/live/calls', {
+ token: h.serviceToken(['chat.live_bridge.write', 'chat.message.send']),
+ body: { boot: 'parity-live', ops: ops.map((o) => ({ seq: ++bridgeSeq, ...o })) },
+});
+
+/** 8 kHz 8-bit mono WAV of `seconds`. */
+function wav(seconds) {
+ const n = Math.floor(8000 * seconds);
+ const b = Buffer.alloc(44 + n);
+ b.write('RIFF', 0); b.writeUInt32LE(36 + n, 4); b.write('WAVE', 8); b.write('fmt ', 12);
+ b.writeUInt32LE(16, 16); b.writeUInt16LE(1, 20); b.writeUInt16LE(1, 22); b.writeUInt32LE(8000, 24);
+ b.writeUInt32LE(8000, 28); b.writeUInt16LE(1, 32); b.writeUInt16LE(8, 34); b.write('data', 36); b.writeUInt32LE(n, 40);
+ b.fill(128, 44);
+ return b;
+}
+
+// Each role's sockets come from its own address (through nginx: CF-Connecting-IP).
+const driver = {
+ address: (role) => IPS[role],
+ connect: ({ role, stream }) => h.ws({ ip: IPS[role], stream }),
+ http: (method, path, { token, body } = {}) => h.http(method, path, { token, body }),
+};
+
+t('boot: a channel with a moderator, a live stream, a channel sound, TTS that synthesizes', async () => {
+ h = await boot();
+ streamer = h.addUser('streamer', { role: 'streamer', subject: 'usr_01J9PAR1TY00000000000000S1' });
+ mod = h.addUser('moddy', { subject: 'usr_01J9PAR1TY00000000000000M1' });
+ alice = h.addUser('alice', { subject: 'usr_01J9PAR1TY00000000000000A1' });
+ bob = h.addUser('bob', { subject: 'usr_01J9PAR1TY00000000000000B1' });
+ dave = h.addUser('dave');
+ channelId = h.addChannel(streamer.id, { moderators: [mod.id] });
+ streamId = h.addStream(streamer.id, channelId);
+ await h.ctx.sync();
+ // TTS without an engine: every utterance is 0.2 s of silence.
+ const tts = require('../server/chat/tts-engine');
+ const fake = async (text) => ({ audio: wav(0.2).toString('base64'), mimeType: 'audio/wav', engine: 'test', voiceName: 'test', voiceId: 'test', text });
+ tts.synthesize = fake;
+ tts.synthesizeUserVoice = fake;
+ // The streamer's [+] modal uploads a channel sound (converted to MP3).
+ const fd = new FormData();
+ fd.append('command', '!honk'); fd.append('channel_id', String(streamer.id));
+ fd.append('sound', new Blob([wav(0.4)], { type: 'audio/wav' }), 'honk.wav');
+ const up = await fetch(`${h.base}/api/sounds`, { method: 'POST', headers: { Authorization: `Bearer ${streamer.token}` }, body: fd });
+ assert.strictEqual(up.status, 200, await up.text());
+ p = new parity.Parity({
+ driver, streamId, sound: 'honk',
+ tokens: { a: alice.token, b: bob.token, mod: mod.token, streamer: streamer.token },
+ testAccounts: ['alice', 'bob', 'moddy', 'streamer'],
+ });
+ await p.resolve();
+ assert.deepStrictEqual(Object.fromEntries(Object.entries(p.users).map(([r, u]) => [r, u.id])), { a: alice.id, b: bob.id, mod: mod.id, streamer: streamer.id });
+ assert.strictEqual(p.owner, 'streamer');
+ assert.strictEqual(p.ownerId, streamer.id);
+ assert.deepStrictEqual(p.untested(), []);
+});
+
+for (const s of parity.SCENARIOS) {
+ t(`${s.key}: ${s.does}`, async () => {
+ const r = await p.run(s);
+ results.push(r);
+ if (r.status === 'fail') throw r.error;
+ assert.notStrictEqual(r.status, 'skip', `nothing is skipped here: ${r.note}`);
+ });
+}
+
+t('dm: delivery keeps the participant rule (chatServer.sendDm checks dm.isParticipant)', async () => {
+ const conv = await h.http('POST', '/api/dm/conversations', { token: alice.token, body: { user_ids: [bob.id] } });
+ const convId = conv.body.conversation.id;
+ const outsider = await p.join('mod', { stream: false });
+ const bobWs = await p.join('b', { stream: false });
+ h.chatServer.sendDm(mod.id, { type: 'dm', conversation_id: convId, message: { message: 'not for you' } });
+ h.chatServer.sendDm(bob.id, { type: 'dm', conversation_id: convId, message: { message: 'for bob' } });
+ await bobWs.next((m) => m.type === 'dm' && m.message.message === 'for bob');
+ assert.ok(await outsider.none((m) => m.type === 'dm'), 'a non-participant never receives it');
+ await p.closeAll();
+});
+
+t('ban from Live’s context menu (/api/mod/stream-ban → the bridge): the socket is closed, the user list drops them, they cannot talk', async () => {
+ const watcher = await p.join('a');
+ const daveTokens = { ...p.tokens, dave: dave.token };
+ const q = new parity.Parity({ driver, tokens: daveTokens, streamId });
+ q.ownerId = streamer.id;
+ const daveWs = await q.join('dave');
+ const listed = (m) => m.type === 'users-list' && m.users.logged.some((u) => u.username === 'dave');
+ await watcher.next(listed);
+ // Live writes the bans row, then asks Chat to disconnect (Live server/admin/mod-routes.js).
+ h.live.addBan({ stream_id: streamId, user_id: dave.id });
+ const r = await bridge([{ op: 'disconnectUser', args: [{ userId: dave.id, ip: null, streamId }] }]);
+ assert.strictEqual(r.status, 200, r.text);
+ await daveWs.next((m) => m.type === 'system' && m.message === 'You have been banned.');
+ await new Promise((res) => (daveWs.readyState === 3 ? res() : daveWs.once('close', res)));
+ await watcher.next((m) => m.type === 'users-list' && !m.users.logged.some((u) => u.username === 'dave'));
+ const again = await q.join('dave');
+ await q.say(again, 'let me back in');
+ await q.system(again, 'You are banned from this chat.');
+ assert.ok(await watcher.none((m) => m.type === 'chat' && m.message === 'let me back in'));
+ await q.closeAll();
+ await p.closeAll();
+ h.live.clearBans();
+ await h.ctx.invalidateBans();
+});
+
+t('reconnect in global chat (Live’s floating widget, openvibe.chat): Live’s moderator delete while away, then ?after_id= converges', async () => {
+ const stayer = await p.join('a', { stream: false });
+ let leaver = await p.join('b', { stream: false });
+ const writer = await p.join('mod', { stream: false });
+ const viewA = new parity.RoomView((await h.http('GET', '/api/chat/global/history?limit=500')).body); const fromA = stayer.all.length;
+ const viewB = new parity.RoomView((await h.http('GET', '/api/chat/global/history?limit=500')).body); const fromB = leaver.all.length;
+ const doomed = p.text('global line a moderator deletes');
+ await p.say(writer, doomed);
+ const m1 = await p.saw(stayer, doomed); await p.saw(leaver, doomed);
+ viewB.frames(leaver.all.slice(fromB));
+ await p.left(leaver);
+ // Live's /api/mod/delete-message: the row through the bridge, then the frame to the global feed.
+ const r = await bridge([
+ { op: 'db', args: ['deleteChatMessage', m1.id, mod.id] },
+ { op: 'broadcastGlobal', args: [{ type: 'delete-messages', ids: [m1.id] }] },
+ ]);
+ assert.ok(r.body.results.every((x) => x.ok), r.text);
+ await stayer.next((m) => m.type === 'delete-messages' && m.ids.includes(m1.id));
+ const later = p.text('global line while B was away');
+ await p.say(stayer, later);
+ const m2 = await p.saw(stayer, later);
+ viewA.frames(stayer.all.slice(fromA));
+ leaver = await p.join('b', { stream: false });
+ const d = (await h.http('GET', `/api/chat/global/history?after_id=${viewB.cursor}&limit=200`)).body;
+ assert.ok(d.deleted_ids.includes(m1.id), 'the delete is in deleted_ids');
+ viewB.delta(d);
+ assert.ok(!viewB.ids.has(m1.id) && viewB.ids.has(m2.id));
+ assert.deepStrictEqual(viewB.sorted(), viewA.sorted(), 'the widget converges on what the connected reader saw');
+ await p.closeAll();
+});
+
+t('the CLI: a dry run lists every scenario and opens nothing; --apply refuses accounts that are not named test accounts', async () => {
+ const env = {
+ OV_PARITY_BASE: h.base, OV_PARITY_STREAM: String(streamId), OV_PARITY_SOUND: 'honk',
+ OV_PARITY_TOKEN_A: alice.token, OV_PARITY_TOKEN_B: bob.token, OV_PARITY_TOKEN_MOD: mod.token,
+ OV_PARITY_TEST_ACCOUNTS: 'alice,bob',
+ };
+ const out = [];
+ const sockets = h.chatServer.clients.size;
+ const requests = h.live.requests.length;
+ assert.strictEqual(await parity.main([], env, (l) => out.push(l)), 0);
+ const text = out.join('\n');
+ assert.match(text, /dry run/);
+ for (const s of parity.SCENARIOS) assert.match(text, new RegExp(`\\b${s.key}\\b`), s.key);
+ assert.match(text, /purge\s+destructive\s+would skip: needs OV_PARITY_TOKEN_STREAMER/);
+ assert.ok(![alice.token, bob.token, mod.token].some((tok) => text.includes(tok)), 'tokens are never printed');
+ assert.strictEqual(h.chatServer.clients.size, sockets, 'no connection');
+ assert.strictEqual(h.live.requests.length, requests, 'nothing reached Chat');
+ // --apply: moddy and the channel owner are not named → nothing runs.
+ const before = h.db.get('SELECT COUNT(*) AS n FROM chat_messages').n;
+ out.length = 0;
+ assert.strictEqual(await parity.main(['--apply'], env, (l) => out.push(l)), 2);
+ assert.match(out.join('\n'), /refused: moddy, streamer are not named in OV_PARITY_TEST_ACCOUNTS; nothing was run/);
+ assert.strictEqual(h.db.get('SELECT COUNT(*) AS n FROM chat_messages').n, before, 'no message was written');
+ // Named: the CLI's own network client runs a scenario end to end.
+ out.length = 0;
+ assert.strictEqual(await parity.main(['--apply', '--only', 'join,send'], { ...env, OV_PARITY_TEST_ACCOUNTS: 'Alice, bob, moddy, streamer' }, (l) => out.push(l)), 0, out.join('\n'));
+ assert.match(out.join('\n'), /PASS join/);
+ assert.match(out.join('\n'), /PASS send/);
+ assert.match(out.join('\n'), /2 passed, 0 gap\(s\), 0 skipped, 0 failed/);
+});
+
+t('the parity matrix', async () => {
+ const width = Math.max(...results.map((r) => r.key.length));
+ for (const r of results) console.log(` ${r.key.padEnd(width)} ${r.status.toUpperCase().padEnd(4)} ${r.note}`);
+ assert.deepStrictEqual(results.filter((r) => r.status === 'gap').map((r) => r.key), ['subonly', 'blocked'], 'the known gaps (see scripts/parity.js)');
+});
+
+t.run(async () => { if (p) await p.closeAll(); if (h) await h.close(); });