openvibers/openvibe.chat Commits
chat.message.deleted: every delete of public-room messages (one message by a moderator or through Live's bridge, a user's / anon's / relay user's history from self-delete or /api/mod, a time-range purge, the auto-delete sweep) adds a public chat.message.deleted to the outbox in the same transaction, one per 500 ids, carrying only message_ids (never the text, the author or who deleted it) and payload.redacts { subject_type: chat_message, subject_ids }, so OpenVibe.Events turns the stored chat.message.created into a tombstone on every read path. Return values of the delete functions are unchanged. test/message-deleted.test.js covers every path, chunking and rollback, and, with a sibling OpenVibe.Events checkout, the relay end to end: anonymous, signed-in and service SSE replay and pull return the tombstone at an unchanged seq, and only svc:chat can redact Chat's events
Changes
@@@ -76,8 +76,16 @@ Stays in Live (decided with evidence, `docs/cutover.md`): `media_requests`, `med| --- | --- | --- || `chat.message.created` | public | every stored message in a public room (global, channel, stream) || `chat.dm.created` | subject | every DM; the payload names the participants, never the text |++| `chat.message.deleted` | public | every deletion of public-room messages: one message (moderation), a user's, an anon's or a relay user's history (self-delete, `/api/mod` purge), a time-range purge, and the auto-delete sweep. Only ids (`message_ids`, at most 500 per event), never the text, the author or who deleted it || `chat.moderation.action` | internal | every moderation log row (chat commands and Live's `/api/mod`) |++`chat.message.deleted` also carries `payload.redacts: { subject_type: "chat_message", subject_ids }`,++which makes OpenVibe.Events rewrite the stored `chat.message.created` of each id into a tombstone++(no text, no anon id, no author) on every read path, public SSE replay included. Every delete runs++in one transaction with its outbox row, and the relay publishes in outbox order, so the deletion++always follows the message it removes. Messages deleted before this existed are redacted by++OpenVibe.Events' `scripts/redact-backfill.js`.++## Running it```bash@@@ -137,8 +145,8 @@ Introduced here and registered in `openvibe-contracts` v0.13.0:`live.chat_effects.write`, `live.chat_mirror.write` (owner live). Planned families:`chat.room.*`, `chat.message.*`, `chat.dm.*`, `chat.moderation.*`, `chat.tts.*`, `chat.call.*`.--Events: `chat.message.created`, `chat.dm.created`, `chat.moderation.action` (produced);--planned `chat.message.deleted`, `chat.room.updated`, `chat.call.*`, `chat.tts.queued|played|failed`.++Events: `chat.message.created`, `chat.message.deleted`, `chat.dm.created`, `chat.moderation.action`++(produced); planned `chat.room.updated`, `chat.call.*`, `chat.tts.queued|played|failed`.## Depends on@@@ -11,7 +11,7 @@"capabilities": ["chat.live_bridge.write", "chat.presence.read"],"capabilitiesUsed": ["live.chat_context.read", "live.chat_effects.write", "live.chat_mirror.write", "events.event.publish"],"capabilitiesRegistered": true,-- "events": ["chat.message.created", "chat.dm.created", "chat.moderation.action"],++ "events": ["chat.message.created", "chat.message.deleted", "chat.dm.created", "chat.moderation.action"],"liveAuthorityFlag": "CHAT_AUTHORITY=chat is set in production live.env since the cutover (2026-09-23 02:03 UTC); Live's chat tables are a read mirror Chat keeps current","deployed": true,"deployment": "openvibe-ovh /opt/openvibe.chat, unit openvibe-chat, 127.0.0.1:4400 (loopback); public only through openvibe.live's nginx include for the chat paths; release 45f2b1f; EVENTS_URL set (chat.message.created reaches OpenVibe.Events); restore drill passed 2026-09-23",@@@ -146,6 +146,31 @@ function subjectFor(userId) {function _outbox() { return require('../events/outbox'); }function _ctx() { return require('../live-context'); }++// Ids per chat.message.deleted event (OpenVibe.Events takes up to 1000 per redaction directive).++const DELETED_EVENT_IDS = 500;++++/**++ * Announce deleted messages: one chat.message.deleted per DELETED_EVENT_IDS ids, in the caller's++ * transaction. Public like chat.message.created, and it carries only ids (never the text, the++ * author or who deleted it). payload.redacts asks OpenVibe.Events to turn the stored++ * chat.message.created of each id into a tombstone, so the text stops being replayable there.++ */++function _announceDeleted(ids) {++ const list = [...new Set((ids || []).map(Number).filter((n) => Number.isInteger(n) && n > 0))];++ for (let i = 0; i < list.length; i += DELETED_EVENT_IDS) {++ const part = list.slice(i, i + DELETED_EVENT_IDS);++ _outbox().enqueue({++ event_type: 'chat.message.deleted',++ visibility: 'public',++ subject: { type: 'chat_message', id: String(part[0]) },++ payload: {++ message_ids: part,++ redacts: { subject_type: 'chat_message', subject_ids: part.map(String) },++ },++ });++ }++}++// ── Chat messages ─────────────────────────────────────────────function saveChatMessage({ stream_id, channel_user_id, user_id, anon_id, username, message, message_type, is_global, reply_to_id, source_platform, auto_delete_at, metadata }) {@@@ -271,13 +296,18 @@ function mergeChatMessageMetadata(id, patch) {}/**-- * Soft-delete a chat message by ID. Sets is_deleted=1 and records who deleted it.++ * Soft-delete a chat message by ID. Sets is_deleted=1 and records who deleted it. Every delete below++ * also announces chat.message.deleted in the same transaction (_announceDeleted).*/function deleteChatMessage(id, deletedBy = null) {-- return run(-- 'UPDATE chat_messages SET is_deleted = 1, deleted_by = ?, deleted_at = CURRENT_TIMESTAMP WHERE id = ?',-- [deletedBy, id]-- );++ return transaction(() => {++ const res = run(++ 'UPDATE chat_messages SET is_deleted = 1, deleted_by = ?, deleted_at = CURRENT_TIMESTAMP WHERE id = ?',++ [deletedBy, id]++ );++ if (res.changes) _announceDeleted([id]);++ return res;++ });}/**@@@ -289,15 +319,12 @@ function deleteUserChatMessages(userId, { streamId = null, deletedBy = null } =? 'user_id = ? AND stream_id = ? AND is_deleted = 0': 'user_id = ? AND is_deleted = 0';const params = streamId ? [userId, streamId] : [userId];-- const messages = all(`SELECT id FROM chat_messages WHERE ${condition}`, params);-- const ids = messages.map(m => m.id);-- if (ids.length === 0) return [];-- const placeholders = ids.map(() => '?').join(',');-- run(-- `UPDATE chat_messages SET is_deleted = 1, deleted_by = ?, deleted_at = CURRENT_TIMESTAMP WHERE id IN (${placeholders})`,-- [deletedBy, ...ids]-- );-- return ids;++ return transaction(() => {++ const ids = all(`SELECT id FROM chat_messages WHERE ${condition}`, params).map(m => m.id);++ if (ids.length === 0) return [];++ _softDeleteIds(ids, deletedBy);++ return ids;++ });}/**@@@ -308,15 +335,12 @@ function deleteAnonChatMessages(anonId, { streamId = null, deletedBy = null } =? 'anon_id = ? AND stream_id = ? AND is_deleted = 0': 'anon_id = ? AND is_deleted = 0';const params = streamId ? [anonId, streamId] : [anonId];-- const messages = all(`SELECT id FROM chat_messages WHERE ${condition}`, params);-- const ids = messages.map(m => m.id);-- if (ids.length === 0) return [];-- const placeholders = ids.map(() => '?').join(',');-- run(-- `UPDATE chat_messages SET is_deleted = 1, deleted_by = ?, deleted_at = CURRENT_TIMESTAMP WHERE id IN (${placeholders})`,-- [deletedBy, ...ids]-- );-- return ids;++ return transaction(() => {++ const ids = all(`SELECT id FROM chat_messages WHERE ${condition}`, params).map(m => m.id);++ if (ids.length === 0) return [];++ _softDeleteIds(ids, deletedBy);++ return ids;++ });}/**@@@ -327,15 +351,24 @@ function deleteRelayUserMessages(username, { streamId = null, deletedBy = null }? 'username = ? AND stream_id = ? AND is_deleted = 0': 'username = ? AND is_deleted = 0';const params = streamId ? [username, streamId] : [username];-- const messages = all(`SELECT id FROM chat_messages WHERE ${condition}`, params);-- const ids = messages.map(m => m.id);-- if (ids.length === 0) return [];-- const placeholders = ids.map(() => '?').join(',');-- run(-- `UPDATE chat_messages SET is_deleted = 1, deleted_by = ?, deleted_at = CURRENT_TIMESTAMP WHERE id IN (${placeholders})`,-- [deletedBy, ...ids]-- );-- return ids;++ return transaction(() => {++ const ids = all(`SELECT id FROM chat_messages WHERE ${condition}`, params).map(m => m.id);++ if (ids.length === 0) return [];++ _softDeleteIds(ids, deletedBy);++ return ids;++ });++}++++/** Mark `ids` deleted (in chunks, under SQLite's variable limit) and announce them. In a transaction. */++function _softDeleteIds(ids, deletedBy) {++ for (let i = 0; i < ids.length; i += DELETED_EVENT_IDS) {++ const part = ids.slice(i, i + DELETED_EVENT_IDS);++ run(++ `UPDATE chat_messages SET is_deleted = 1, deleted_by = ?, deleted_at = CURRENT_TIMESTAMP WHERE id IN (${part.map(() => '?').join(',')})`,++ [deletedBy, ...part]++ );++ }++ _announceDeleted(ids);}function deleteExpiredChatMessages(limit = 500) {@@@ -357,29 +390,33 @@ function deleteExpiredChatMessages(limit = 500) {const ids = rows.map(row => row.id);const placeholders = ids.map(() => '?').join(',');-- run(-- `UPDATE chat_messages-- SET is_deleted = 1, deleted_at = CURRENT_TIMESTAMP-- WHERE id IN (${placeholders})`,-- ids-- );++ transaction(() => {++ run(++ `UPDATE chat_messages++ SET is_deleted = 1, deleted_at = CURRENT_TIMESTAMP++ WHERE id IN (${placeholders})`,++ ids++ );++ _announceDeleted(ids);++ });return rows;}function deleteChatMessagesByTimeRange(streamId, fromTime, toTime, deletedBy) {-- if (streamId) {-- return run(-- `UPDATE chat_messages SET is_deleted = 1, deleted_by = ?, deleted_at = CURRENT_TIMESTAMP-- WHERE stream_id = ? AND timestamp >= ? AND timestamp <= ? AND is_deleted = 0`,-- [deletedBy, streamId, fromTime, toTime]++ // Global chat is is_global = 1. The ids are read first, in the same transaction, to announce them.++ const where = streamId++ ? 'stream_id = ? AND timestamp >= ? AND timestamp <= ? AND is_deleted = 0'++ : 'is_global = 1 AND timestamp >= ? AND timestamp <= ? AND is_deleted = 0';++ const params = streamId ? [streamId, fromTime, toTime] : [fromTime, toTime];++ return transaction(() => {++ const ids = all(`SELECT id FROM chat_messages WHERE ${where}`, params).map(m => m.id);++ const res = run(++ `UPDATE chat_messages SET is_deleted = 1, deleted_by = ?, deleted_at = CURRENT_TIMESTAMP WHERE ${where}`,++ [deletedBy, ...params]);-- }-- // Global chat (is_global = 1)-- return run(-- `UPDATE chat_messages SET is_deleted = 1, deleted_by = ?, deleted_at = CURRENT_TIMESTAMP-- WHERE is_global = 1 AND timestamp >= ? AND timestamp <= ? AND is_deleted = 0`,-- [deletedBy, fromTime, toTime]-- );++ _announceDeleted(ids);++ return res;++ });}function countChatMessagesByTimeRange(streamId, fromTime, toTime) {@@@ -4,6 +4,9 @@* enqueue() runs inside the same SQLite transaction as the change it announces (a chat message,* a DM, a moderation action), so an event exists exactly when its effect does:* chat.message.created visibility public — a message in a public room (global/channel/stream)++ * chat.message.deleted visibility public — ids of deleted messages (moderation, the author,++ * a purge, auto-delete); payload.redacts makes++ * OpenVibe.Events tombstone their chat.message.created* chat.dm.created visibility subject — a DM; the payload names participants, never the text* chat.moderation.action visibility internal — ban/timeout/clear/slow/deletes, from chat or Live*nnew file mode 100644@@@ -0,0 +1,249 @@++'use strict';++/**++ * chat.message.deleted: every way a public-room message is deleted (one message by a moderator or++ * through Live's bridge, a user's / anon's / relay user's history, a time-range purge, the++ * auto-delete sweep) adds one outbox envelope per 500 ids, in the same transaction, carrying only++ * ids and payload.redacts. The relay publishes it after the message's chat.message.created.++ *++ * End to end (when a sibling OpenVibe.Events checkout with redaction is present, or++ * OPENVIBE_EVENTS_DIR points at one): Chat's relay publishes to a real Events, and after the++ * deletion neither anonymous nor signed-in SSE replay, pull or delivery returns the text; the++ * sequence has no hole; only svc:chat can redact Chat's events.++ */++const assert = require('assert');++const fs = require('fs');++const net = require('net');++const path = require('path');++const nodeHttp = require('http');++const { validate, serviceAuth } = require('openvibe-contracts');++const { boot, suite } = require('./helpers');++++const t = suite('message-deleted');++const EVENTS_DIR = process.env.OPENVIBE_EVENTS_DIR || path.join(__dirname, '..', '..', 'OpenVibe.Events');++const haveEvents = ['server/index.js', 'server/redaction.js', 'node_modules/better-sqlite3'].every((f) => fs.existsSync(path.join(EVENTS_DIR, f)));++const SECRET = 'my number is 555-0199';++let h, events, eventsPort, admin, alice, streamer, streamId;++++/** Outbox chat.message.deleted envelopes after `afterSeq`. */++function deletions(afterSeq = 0) {++ return h.db.all("SELECT seq, event FROM events_outbox WHERE event_type = 'chat.message.deleted' AND seq > ? ORDER BY seq", [afterSeq])++ .map((r) => JSON.parse(r.event));++}++const outboxSeq = () => h.db.get('SELECT COALESCE(MAX(seq), 0) AS s FROM events_outbox').s;++function say(message, over = {}) {++ return Number(h.db.saveChatMessage({ stream_id: null, user_id: null, anon_id: 'anon7', username: 'anon7', message, message_type: 'chat', is_global: 1, ...over }).lastInsertRowid);++}++function assertDeletion(env, ids) {++ const v = validate('events.event-envelope@1', env);++ assert.ok(v.valid, JSON.stringify(v.errors));++ assert.strictEqual(env.event_type, 'chat.message.deleted');++ assert.strictEqual(env.source, 'chat');++ assert.strictEqual(env.visibility, 'public');++ assert.deepStrictEqual(env.actor, { type: 'service', id: 'chat' }, 'who deleted it is not published');++ assert.deepStrictEqual(env.subject, { type: 'chat_message', id: String(ids[0]) });++ assert.deepStrictEqual(env.payload, { message_ids: ids, redacts: { subject_type: 'chat_message', subject_ids: ids.map(String) } });++}++++async function freePort() {++ const s = net.createServer();++ await new Promise((r) => s.listen(0, '127.0.0.1', r));++ const { port } = s.address();++ await new Promise((r) => s.close(r));++ return port;++}++++t('boot (a real OpenVibe.Events on a reserved port when one is checked out next to this repo)', async () => {++ eventsPort = await freePort();++ h = await boot({ env: { EVENTS_URL: `http://127.0.0.1:${eventsPort}`, EVENTS_RELAY_INTERVAL_MS: '3600000' } });++ admin = h.addUser('admin', { role: 'admin' });++ alice = h.addUser('alice', { subject: 'usr_01J9AAAAAAAAAAAAAAAAAAAAAA' });++ streamer = h.addUser('streamer', { role: 'streamer' });++ streamId = h.addStream(streamer.id, h.addChannel(streamer.id));++ await h.ctx.sync();++ if (haveEvents) {++ const { load } = require(path.join(EVENTS_DIR, 'server', 'config'));++ const { start } = require(path.join(EVENTS_DIR, 'server', 'index'));++ const quiet = { log() {}, warn() {}, error() {} };++ events = await start({++ config: load({ NODE_ENV: 'test', PORT: String(eventsPort), EVENTS_DB_PATH: path.join(h.tmp, 'events.db'), OV_NETWORK_PUBLIC_KEY: h.keys.publicKey, EVENTS_WORKER: 'off' }),++ log: quiet,++ });++ } else {++ console.log(` (no OpenVibe.Events with redaction at ${EVENTS_DIR}: the end-to-end tests are skipped)`);++ }++});++++t('one message deleted (a moderator, or Live’s bridge): one deletion event, ids only', async () => {++ const id = say(SECRET);++ const from = outboxSeq();++ h.db.deleteChatMessage(id, admin.id);++ const [d] = deletions(from);++ assertDeletion(d, [id]);++ assert.ok(!JSON.stringify(d).includes('555-0199') && !JSON.stringify(d).includes('anon7'));++ // Through the bridge, the way Live's /api/mod forwards it.++ const id2 = say('via the bridge');++ const r = await h.http('POST', '/internal/live/calls', { token: h.serviceToken(['chat.live_bridge.write']), body: { boot: 'b1', ops: [{ seq: 1, op: 'db', args: ['deleteChatMessage', id2, admin.id] }] } });++ assert.strictEqual(r.status, 200, r.text);++ assertDeletion(deletions(from).at(-1), [id2]);++ const before = outboxSeq();++ h.db.deleteChatMessage(999999, admin.id);++ assert.strictEqual(outboxSeq(), before, 'nothing announced for a message that does not exist');++});++++t('a user’s, an anon’s and a relay user’s history (self-delete, /api/mod purge)', async () => {++ const mine = [say('a1', { user_id: alice.id, anon_id: null, username: 'alice' }), say('a2', { user_id: alice.id, anon_id: null, username: 'alice' })];++ const anon = [say('n1', { anon_id: 'anon99', username: 'anon99' })];++ const relay = [say('r1', { username: '[Twitch] zed', anon_id: null })];++ let from = outboxSeq();++ assert.deepStrictEqual(h.db.deleteUserChatMessages(alice.id, { deletedBy: alice.id }), mine, 'return value unchanged');++ assertDeletion(deletions(from)[0], mine);++ from = outboxSeq();++ assert.deepStrictEqual(h.db.deleteAnonChatMessages('anon99', { deletedBy: null }), anon);++ assertDeletion(deletions(from)[0], anon);++ from = outboxSeq();++ assert.deepStrictEqual(h.db.deleteRelayUserMessages('[Twitch] zed', { deletedBy: admin.id }), relay);++ assertDeletion(deletions(from)[0], relay);++ from = outboxSeq();++ assert.deepStrictEqual(h.db.deleteUserChatMessages(alice.id, { deletedBy: alice.id }), [], 'nothing left');++ assert.deepStrictEqual(deletions(from), [], 'no empty deletion events');++});++++t('self-delete over the socket announces what it deleted', async () => {++ const ws = await h.ws({ ip: '198.51.100.77', token: alice.token });++ ws.sendJson({ type: 'join', token: alice.token });++ await ws.next((m) => m.type === 'auth');++ const id = say('from alice', { user_id: alice.id, anon_id: null, username: 'alice' });++ const from = outboxSeq();++ ws.sendJson({ type: 'self-delete-history' });++ const res = await ws.next((m) => m.type === 'self-delete-result');++ assert.strictEqual(res.count, 1);++ assertDeletion(deletions(from)[0], [id]);++ ws.close();++});++++t('a time-range purge (REST) and the auto-delete sweep', async () => {++ const inRoom = [say('s1', { stream_id: streamId, is_global: 0 }), say('s2', { stream_id: streamId, is_global: 0 })];++ const global = say('g1');++ let from = outboxSeq();++ const r = await h.http('DELETE', '/api/chat/admin/purge', { token: admin.token, body: { streamId, from: h.sqliteNow(-60000), to: h.sqliteNow(60000) } });++ assert.strictEqual(r.status, 200, r.text);++ assert.strictEqual(r.body.deleted, 2);++ assertDeletion(deletions(from)[0], inRoom);++ assert.ok(!h.db.getChatMessageById(global).is_deleted, 'the other room is untouched');++ const expiring = say('ephemeral', { auto_delete_at: h.sqliteNow(-1000) });++ from = outboxSeq();++ const swept = h.db.deleteExpiredChatMessages(500);++ assert.deepStrictEqual(swept.map((x) => x.id), [expiring]);++ assertDeletion(deletions(from)[0], [expiring]);++});++++t('more than 500 ids: one event per 500, in order', async () => {++ const ids = [];++ h.db.transaction(() => { for (let i = 0; i < 1203; i++) ids.push(say(`bulk ${i}`, { anon_id: 'anon500', username: 'anon500' })); });++ const from = outboxSeq();++ h.db.deleteAnonChatMessages('anon500', { deletedBy: admin.id });++ const d = deletions(from);++ assert.deepStrictEqual(d.map((e) => e.payload.message_ids.length), [500, 500, 203]);++ assert.deepStrictEqual(d.flatMap((e) => e.payload.message_ids), ids);++ for (const e of d) assert.ok(Buffer.byteLength(JSON.stringify(e.payload)) < 64 * 1024, 'fits Events’ payload limit');++});++++t('a failed delete leaves no event (same transaction)', async () => {++ const id = say('stays');++ const from = outboxSeq();++ assert.throws(() => h.db.transaction(() => { h.db.deleteChatMessage(id, admin.id); throw new Error('boom'); }), /boom/);++ assert.deepStrictEqual(deletions(from), []);++ assert.ok(!h.db.getChatMessageById(id).is_deleted);++});++++// ── End to end with OpenVibe.Events ──────────────────────────────++++function sse(pathAndQuery, headers = {}) {++ return new Promise((resolve, reject) => {++ const req = nodeHttp.get({ host: '127.0.0.1', port: eventsPort, path: pathAndQuery, headers }, (res) => {++ let body = '';++ const c = {++ status: res.statusCode,++ body: () => body,++ // Complete blocks only; `event: gap` data has no envelope and is left out.++ events: () => body.split('\n\n').slice(0, -1).map((b) => b.split('\n').find((l) => l.startsWith('data: ')))++ .filter(Boolean).map((l) => JSON.parse(l.slice(6))).filter((m) => m.event),++ async waitFor(pred, ms = 3000) {++ const until = Date.now() + ms;++ while (!pred(c)) { if (Date.now() > until) throw new Error(`sse timeout: ${body}`); await h.sleep(20); }++ return c;++ },++ close: () => req.destroy(),++ };++ res.setEncoding('utf8');++ res.on('data', (d) => { body += d; });++ resolve(c);++ });++ req.on('error', reject);++ });++}++const userJwt = (subjectId) => serviceAuth.signServiceToken({++ sub: 57, subject_id: subjectId, role: 'user', iss: h.ISS, aud: ['openvibe.network'], iat: Math.floor(Date.now() / 1000), exp: Math.floor(Date.now() / 1000) + 600,++}, h.keys.privateKey);++++let firstSeq, msgId;++t('e2e: the relay publishes the message, then its deletion; replay serves only the tombstone', async () => {++ if (!haveEvents) return;++ h.db.run('UPDATE events_outbox SET sent_at = ? WHERE sent_at IS NULL', [new Date().toISOString()]); // earlier tests' rows++ msgId = say(SECRET, { user_id: alice.id, anon_id: null, username: 'alice' });++ assert.strictEqual((await h.eventsRelay.flush()).sent, 1);++ firstSeq = events.store.lastSeq();++ const before = await sse(`/realtime/stream?topics=chat.message.*&last_event_id=${firstSeq - 1}`);++ await before.waitFor((c) => c.events().length === 1);++ assert.ok(before.body().includes('555-0199'), 'before the deletion the text is there (the leak)');++ before.close();++++ h.db.deleteChatMessage(msgId, admin.id);++ assert.strictEqual((await h.eventsRelay.flush()).sent, 1);++ const views = [++ ['anonymous', {}],++ ['signed in (cookie)', { Cookie: `ov_token=${userJwt('usr_01J9AAAAAAAAAAAAAAAAAAAAAA')}` }],++ ['signed in (Bearer)', { Authorization: `Bearer ${userJwt('usr_01J9BBBBBBBBBBBBBBBBBBBBBB')}` }],++ ['service', { Authorization: `Bearer ${h.serviceToken(['events.event.read'], { aud: 'openvibe.events', sub: 'svc:search' })}` }],++ ];++ for (const [who, headers] of views) {++ const c = await sse(`/realtime/stream?topics=chat.message.*&last_event_id=${firstSeq - 1}`, headers);++ await c.waitFor((x) => x.events().length === 2);++ const [created, deleted] = c.events();++ assert.deepStrictEqual([created.seq, deleted.seq], [firstSeq, firstSeq + 1], `${who}: sequence intact`);++ assert.strictEqual(created.event.event_type, 'chat.message.created');++ assert.strictEqual(created.event.payload.redacted, true, who);++ assert.deepStrictEqual(created.event.subject, { type: 'chat_message', id: String(msgId) });++ assert.strictEqual(deleted.event.event_type, 'chat.message.deleted');++ assert.ok(!c.body().includes('555-0199') && !c.body().includes('usr_01J9AAAAAAAAAAAAAAAAAAAAAA'), `${who}: no text, no author`);++ c.close();++ }++ const pull = await fetch(`http://127.0.0.1:${eventsPort}/api/v1/events?topic=chat.message.*&after_seq=${firstSeq - 1}`, {++ headers: { Authorization: `Bearer ${h.serviceToken(['events.event.read'], { aud: 'openvibe.events', sub: 'svc:search' })}` },++ }).then((r) => r.json());++ assert.deepStrictEqual(pull.events.map((e) => e.seq), [firstSeq, firstSeq + 1]);++ assert.ok(!JSON.stringify(pull).includes('555-0199'));++});++++t('e2e: only svc:chat can redact Chat’s events', async () => {++ if (!haveEvents) return;++ const id = say('not yours to delete');++ await h.eventsRelay.flush();++ const created = events.store.scan(0, { patterns: ['chat.message.created'], limit: 1000 }).rows.find((r) => r.subject_id === String(id));++ const env = deletions(0).at(-1); // a well-formed deletion, re-sourced by another service++ const post = (sub, body) => fetch(`http://127.0.0.1:${eventsPort}/api/v1/events`, {++ method: 'POST', body: JSON.stringify(body),++ headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${h.serviceToken(['events.event.publish'], { aud: 'openvibe.events', sub })}` },++ });++ let r = await post('svc:live', { ...env, event_id: 'evt_01J9CCCCCCCCCCCCCCCCCCCCCC', source: 'live', event_type: 'live.chat.deleted', payload: { redacts: { event_ids: [created.id] } } });++ assert.strictEqual(r.status, 403);++ assert.strictEqual((await r.json()).code, 'events.redaction_not_allowed');++ r = await post('svc:live', { ...env, event_id: 'evt_01J9DDDDDDDDDDDDDDDDDDDDDD', payload: { redacts: { subject_type: 'chat_message', subject_ids: [String(id)] } } });++ assert.strictEqual(r.status, 403, 'svc:live cannot publish as chat');++ assert.ok(events.store.getEvent(created.id).payload.includes('not yours to delete'), 'still intact');++ h.db.deleteChatMessage(id, admin.id);++ await h.eventsRelay.flush();++ assert.ok(events.store.getEvent(created.id).redacted_at, 'Chat itself can');++});++++t.run(async () => { if (events) await events.close(); if (h) await h.close(); });
@@ -76,8 +76,16 @@ Stays in Live (decided with evidence, `docs/cutover.md`): `media_requests`, `med
@@ -76,8 +76,16 @@ Stays in Live (decided with evidence, `docs/cutover.md`): `media_requests`, `med
| --- | --- | --- |
| --- | --- | --- |
| `chat.message.created` | public | every stored message in a public room (global, channel, stream) |
| `chat.message.created` | public | every stored message in a public room (global, channel, stream) |
| `chat.dm.created` | subject | every DM; the payload names the participants, never the text |
| `chat.dm.created` | subject | every DM; the payload names the participants, never the text |
+| `chat.message.deleted` | public | every deletion of public-room messages: one message (moderation), a user's, an anon's or a relay user's history (self-delete, `/api/mod` purge), a time-range purge, and the auto-delete sweep. Only ids (`message_ids`, at most 500 per event), never the text, the author or who deleted it |
| `chat.moderation.action` | internal | every moderation log row (chat commands and Live's `/api/mod`) |
| `chat.moderation.action` | internal | every moderation log row (chat commands and Live's `/api/mod`) |
+`chat.message.deleted` also carries `payload.redacts: { subject_type: "chat_message", subject_ids }`,
+which makes OpenVibe.Events rewrite the stored `chat.message.created` of each id into a tombstone
+(no text, no anon id, no author) on every read path, public SSE replay included. Every delete runs
+in one transaction with its outbox row, and the relay publishes in outbox order, so the deletion
+always follows the message it removes. Messages deleted before this existed are redacted by
+OpenVibe.Events' `scripts/redact-backfill.js`.
+
## Running it
## Running it
```bash
```bash
@@ -137,8 +145,8 @@ Introduced here and registered in `openvibe-contracts` v0.13.0:
@@ -137,8 +145,8 @@ Introduced here and registered in `openvibe-contracts` v0.13.0:
`live.chat_effects.write`, `live.chat_mirror.write` (owner live). Planned families:
`live.chat_effects.write`, `live.chat_mirror.write` (owner live). Planned families:
`chat.room.*`, `chat.message.*`, `chat.dm.*`, `chat.moderation.*`, `chat.tts.*`, `chat.call.*`.
`chat.room.*`, `chat.message.*`, `chat.dm.*`, `chat.moderation.*`, `chat.tts.*`, `chat.call.*`.
-Events: `chat.message.created`, `chat.dm.created`, `chat.moderation.action` (produced);
-planned `chat.message.deleted`, `chat.room.updated`, `chat.call.*`, `chat.tts.queued|played|failed`.
+Events: `chat.message.created`, `chat.message.deleted`, `chat.dm.created`, `chat.moderation.action`
+(produced); planned `chat.room.updated`, `chat.call.*`, `chat.tts.queued|played|failed`.
## Depends on
## Depends on
@@ -11,7 +11,7 @@
@@ -11,7 +11,7 @@
"capabilities": ["chat.live_bridge.write", "chat.presence.read"],
"capabilities": ["chat.live_bridge.write", "chat.presence.read"],
"capabilitiesUsed": ["live.chat_context.read", "live.chat_effects.write", "live.chat_mirror.write", "events.event.publish"],
"capabilitiesUsed": ["live.chat_context.read", "live.chat_effects.write", "live.chat_mirror.write", "events.event.publish"],
"capabilitiesRegistered": true,
"capabilitiesRegistered": true,
- "events": ["chat.message.created", "chat.dm.created", "chat.moderation.action"],
+ "events": ["chat.message.created", "chat.message.deleted", "chat.dm.created", "chat.moderation.action"],
"liveAuthorityFlag": "CHAT_AUTHORITY=chat is set in production live.env since the cutover (2026-09-23 02:03 UTC); Live's chat tables are a read mirror Chat keeps current",
"liveAuthorityFlag": "CHAT_AUTHORITY=chat is set in production live.env since the cutover (2026-09-23 02:03 UTC); Live's chat tables are a read mirror Chat keeps current",
"deployed": true,
"deployed": true,
"deployment": "openvibe-ovh /opt/openvibe.chat, unit openvibe-chat, 127.0.0.1:4400 (loopback); public only through openvibe.live's nginx include for the chat paths; release 45f2b1f; EVENTS_URL set (chat.message.created reaches OpenVibe.Events); restore drill passed 2026-09-23",
"deployment": "openvibe-ovh /opt/openvibe.chat, unit openvibe-chat, 127.0.0.1:4400 (loopback); public only through openvibe.live's nginx include for the chat paths; release 45f2b1f; EVENTS_URL set (chat.message.created reaches OpenVibe.Events); restore drill passed 2026-09-23",
@@ -146,6 +146,31 @@ function subjectFor(userId) {
@@ -146,6 +146,31 @@ function subjectFor(userId) {
function _outbox() { return require('../events/outbox'); }
function _outbox() { return require('../events/outbox'); }
function _ctx() { return require('../live-context'); }
function _ctx() { return require('../live-context'); }
+// Ids per chat.message.deleted event (OpenVibe.Events takes up to 1000 per redaction directive).
+const DELETED_EVENT_IDS = 500;
+
+/**
+ * Announce deleted messages: one chat.message.deleted per DELETED_EVENT_IDS ids, in the caller's
+ * transaction. Public like chat.message.created, and it carries only ids (never the text, the
+ * author or who deleted it). payload.redacts asks OpenVibe.Events to turn the stored
+ * chat.message.created of each id into a tombstone, so the text stops being replayable there.
+ */
+function _announceDeleted(ids) {
+ const list = [...new Set((ids || []).map(Number).filter((n) => Number.isInteger(n) && n > 0))];
+ for (let i = 0; i < list.length; i += DELETED_EVENT_IDS) {
+ const part = list.slice(i, i + DELETED_EVENT_IDS);
+ _outbox().enqueue({
+ event_type: 'chat.message.deleted',
+ visibility: 'public',
+ subject: { type: 'chat_message', id: String(part[0]) },
+ payload: {
+ message_ids: part,
+ redacts: { subject_type: 'chat_message', subject_ids: part.map(String) },
+ },
+ });
+ }
+}
+
// ── Chat messages ─────────────────────────────────────────────
// ── Chat messages ─────────────────────────────────────────────
function saveChatMessage({ stream_id, channel_user_id, user_id, anon_id, username, message, message_type, is_global, reply_to_id, source_platform, auto_delete_at, metadata }) {
function saveChatMessage({ stream_id, channel_user_id, user_id, anon_id, username, message, message_type, is_global, reply_to_id, source_platform, auto_delete_at, metadata }) {
@@ -271,13 +296,18 @@ function mergeChatMessageMetadata(id, patch) {
@@ -271,13 +296,18 @@ function mergeChatMessageMetadata(id, patch) {
}
}
/**
/**
- * Soft-delete a chat message by ID. Sets is_deleted=1 and records who deleted it.
+ * Soft-delete a chat message by ID. Sets is_deleted=1 and records who deleted it. Every delete below
+ * also announces chat.message.deleted in the same transaction (_announceDeleted).
*/
*/
function deleteChatMessage(id, deletedBy = null) {
function deleteChatMessage(id, deletedBy = null) {
- return run(
- 'UPDATE chat_messages SET is_deleted = 1, deleted_by = ?, deleted_at = CURRENT_TIMESTAMP WHERE id = ?',
- [deletedBy, id]
- );
+ return transaction(() => {
+ const res = run(
+ 'UPDATE chat_messages SET is_deleted = 1, deleted_by = ?, deleted_at = CURRENT_TIMESTAMP WHERE id = ?',
+ [deletedBy, id]
+ );
+ if (res.changes) _announceDeleted([id]);
+ return res;
+ });
}
}
/**
/**
@@ -289,15 +319,12 @@ function deleteUserChatMessages(userId, { streamId = null, deletedBy = null } =
@@ -289,15 +319,12 @@ function deleteUserChatMessages(userId, { streamId = null, deletedBy = null } =
? 'user_id = ? AND stream_id = ? AND is_deleted = 0'
? 'user_id = ? AND stream_id = ? AND is_deleted = 0'
: 'user_id = ? AND is_deleted = 0';
: 'user_id = ? AND is_deleted = 0';
const params = streamId ? [userId, streamId] : [userId];
const params = streamId ? [userId, streamId] : [userId];
- const messages = all(`SELECT id FROM chat_messages WHERE ${condition}`, params);
- const ids = messages.map(m => m.id);
- if (ids.length === 0) return [];
- const placeholders = ids.map(() => '?').join(',');
- run(
- `UPDATE chat_messages SET is_deleted = 1, deleted_by = ?, deleted_at = CURRENT_TIMESTAMP WHERE id IN (${placeholders})`,
- [deletedBy, ...ids]
- );
- return ids;
+ return transaction(() => {
+ const ids = all(`SELECT id FROM chat_messages WHERE ${condition}`, params).map(m => m.id);
+ if (ids.length === 0) return [];
+ _softDeleteIds(ids, deletedBy);
+ return ids;
+ });
}
}
/**
/**
@@ -308,15 +335,12 @@ function deleteAnonChatMessages(anonId, { streamId = null, deletedBy = null } =
@@ -308,15 +335,12 @@ function deleteAnonChatMessages(anonId, { streamId = null, deletedBy = null } =
? 'anon_id = ? AND stream_id = ? AND is_deleted = 0'
? 'anon_id = ? AND stream_id = ? AND is_deleted = 0'
: 'anon_id = ? AND is_deleted = 0';
: 'anon_id = ? AND is_deleted = 0';
const params = streamId ? [anonId, streamId] : [anonId];
const params = streamId ? [anonId, streamId] : [anonId];
- const messages = all(`SELECT id FROM chat_messages WHERE ${condition}`, params);
- const ids = messages.map(m => m.id);
- if (ids.length === 0) return [];
- const placeholders = ids.map(() => '?').join(',');
- run(
- `UPDATE chat_messages SET is_deleted = 1, deleted_by = ?, deleted_at = CURRENT_TIMESTAMP WHERE id IN (${placeholders})`,
- [deletedBy, ...ids]
- );
- return ids;
+ return transaction(() => {
+ const ids = all(`SELECT id FROM chat_messages WHERE ${condition}`, params).map(m => m.id);
+ if (ids.length === 0) return [];
+ _softDeleteIds(ids, deletedBy);
+ return ids;
+ });
}
}
/**
/**
@@ -327,15 +351,24 @@ function deleteRelayUserMessages(username, { streamId = null, deletedBy = null }
@@ -327,15 +351,24 @@ function deleteRelayUserMessages(username, { streamId = null, deletedBy = null }
? 'username = ? AND stream_id = ? AND is_deleted = 0'
? 'username = ? AND stream_id = ? AND is_deleted = 0'
: 'username = ? AND is_deleted = 0';
: 'username = ? AND is_deleted = 0';
const params = streamId ? [username, streamId] : [username];
const params = streamId ? [username, streamId] : [username];
- const messages = all(`SELECT id FROM chat_messages WHERE ${condition}`, params);
- const ids = messages.map(m => m.id);
- if (ids.length === 0) return [];
- const placeholders = ids.map(() => '?').join(',');
- run(
- `UPDATE chat_messages SET is_deleted = 1, deleted_by = ?, deleted_at = CURRENT_TIMESTAMP WHERE id IN (${placeholders})`,
- [deletedBy, ...ids]
- );
- return ids;
+ return transaction(() => {
+ const ids = all(`SELECT id FROM chat_messages WHERE ${condition}`, params).map(m => m.id);
+ if (ids.length === 0) return [];
+ _softDeleteIds(ids, deletedBy);
+ return ids;
+ });
+}
+
+/** Mark `ids` deleted (in chunks, under SQLite's variable limit) and announce them. In a transaction. */
+function _softDeleteIds(ids, deletedBy) {
+ for (let i = 0; i < ids.length; i += DELETED_EVENT_IDS) {
+ const part = ids.slice(i, i + DELETED_EVENT_IDS);
+ run(
+ `UPDATE chat_messages SET is_deleted = 1, deleted_by = ?, deleted_at = CURRENT_TIMESTAMP WHERE id IN (${part.map(() => '?').join(',')})`,
+ [deletedBy, ...part]
+ );
+ }
+ _announceDeleted(ids);
}
}
function deleteExpiredChatMessages(limit = 500) {
function deleteExpiredChatMessages(limit = 500) {
@@ -357,29 +390,33 @@ function deleteExpiredChatMessages(limit = 500) {
@@ -357,29 +390,33 @@ function deleteExpiredChatMessages(limit = 500) {
const ids = rows.map(row => row.id);
const ids = rows.map(row => row.id);
const placeholders = ids.map(() => '?').join(',');
const placeholders = ids.map(() => '?').join(',');
- run(
- `UPDATE chat_messages
- SET is_deleted = 1, deleted_at = CURRENT_TIMESTAMP
- WHERE id IN (${placeholders})`,
- ids
- );
+ transaction(() => {
+ run(
+ `UPDATE chat_messages
+ SET is_deleted = 1, deleted_at = CURRENT_TIMESTAMP
+ WHERE id IN (${placeholders})`,
+ ids
+ );
+ _announceDeleted(ids);
+ });
return rows;
return rows;
}
}
function deleteChatMessagesByTimeRange(streamId, fromTime, toTime, deletedBy) {
function deleteChatMessagesByTimeRange(streamId, fromTime, toTime, deletedBy) {
- if (streamId) {
- return run(
- `UPDATE chat_messages SET is_deleted = 1, deleted_by = ?, deleted_at = CURRENT_TIMESTAMP
- WHERE stream_id = ? AND timestamp >= ? AND timestamp <= ? AND is_deleted = 0`,
- [deletedBy, streamId, fromTime, toTime]
+ // Global chat is is_global = 1. The ids are read first, in the same transaction, to announce them.
+ const where = streamId
+ ? 'stream_id = ? AND timestamp >= ? AND timestamp <= ? AND is_deleted = 0'
+ : 'is_global = 1 AND timestamp >= ? AND timestamp <= ? AND is_deleted = 0';
+ const params = streamId ? [streamId, fromTime, toTime] : [fromTime, toTime];
+ return transaction(() => {
+ const ids = all(`SELECT id FROM chat_messages WHERE ${where}`, params).map(m => m.id);
+ const res = run(
+ `UPDATE chat_messages SET is_deleted = 1, deleted_by = ?, deleted_at = CURRENT_TIMESTAMP WHERE ${where}`,
+ [deletedBy, ...params]
);
);
- }
- // Global chat (is_global = 1)
- return run(
- `UPDATE chat_messages SET is_deleted = 1, deleted_by = ?, deleted_at = CURRENT_TIMESTAMP
- WHERE is_global = 1 AND timestamp >= ? AND timestamp <= ? AND is_deleted = 0`,
- [deletedBy, fromTime, toTime]
- );
+ _announceDeleted(ids);
+ return res;
+ });
}
}
function countChatMessagesByTimeRange(streamId, fromTime, toTime) {
function countChatMessagesByTimeRange(streamId, fromTime, toTime) {
@@ -4,6 +4,9 @@
@@ -4,6 +4,9 @@
* enqueue() runs inside the same SQLite transaction as the change it announces (a chat message,
* enqueue() runs inside the same SQLite transaction as the change it announces (a chat message,
* a DM, a moderation action), so an event exists exactly when its effect does:
* a DM, a moderation action), so an event exists exactly when its effect does:
* chat.message.created visibility public — a message in a public room (global/channel/stream)
* chat.message.created visibility public — a message in a public room (global/channel/stream)
+ * chat.message.deleted visibility public — ids of deleted messages (moderation, the author,
+ * a purge, auto-delete); payload.redacts makes
+ * OpenVibe.Events tombstone their chat.message.created
* chat.dm.created visibility subject — a DM; the payload names participants, never the text
* chat.dm.created visibility subject — a DM; the payload names participants, never the text
* chat.moderation.action visibility internal — ban/timeout/clear/slow/deletes, from chat or Live
* chat.moderation.action visibility internal — ban/timeout/clear/slow/deletes, from chat or Live
*
*
new file mode 100644
new file mode 100644
@@ -0,0 +1,249 @@
@@ -0,0 +1,249 @@
+'use strict';
+/**
+ * chat.message.deleted: every way a public-room message is deleted (one message by a moderator or
+ * through Live's bridge, a user's / anon's / relay user's history, a time-range purge, the
+ * auto-delete sweep) adds one outbox envelope per 500 ids, in the same transaction, carrying only
+ * ids and payload.redacts. The relay publishes it after the message's chat.message.created.
+ *
+ * End to end (when a sibling OpenVibe.Events checkout with redaction is present, or
+ * OPENVIBE_EVENTS_DIR points at one): Chat's relay publishes to a real Events, and after the
+ * deletion neither anonymous nor signed-in SSE replay, pull or delivery returns the text; the
+ * sequence has no hole; only svc:chat can redact Chat's events.
+ */
+const assert = require('assert');
+const fs = require('fs');
+const net = require('net');
+const path = require('path');
+const nodeHttp = require('http');
+const { validate, serviceAuth } = require('openvibe-contracts');
+const { boot, suite } = require('./helpers');
+
+const t = suite('message-deleted');
+const EVENTS_DIR = process.env.OPENVIBE_EVENTS_DIR || path.join(__dirname, '..', '..', 'OpenVibe.Events');
+const haveEvents = ['server/index.js', 'server/redaction.js', 'node_modules/better-sqlite3'].every((f) => fs.existsSync(path.join(EVENTS_DIR, f)));
+const SECRET = 'my number is 555-0199';
+let h, events, eventsPort, admin, alice, streamer, streamId;
+
+/** Outbox chat.message.deleted envelopes after `afterSeq`. */
+function deletions(afterSeq = 0) {
+ return h.db.all("SELECT seq, event FROM events_outbox WHERE event_type = 'chat.message.deleted' AND seq > ? ORDER BY seq", [afterSeq])
+ .map((r) => JSON.parse(r.event));
+}
+const outboxSeq = () => h.db.get('SELECT COALESCE(MAX(seq), 0) AS s FROM events_outbox').s;
+function say(message, over = {}) {
+ return Number(h.db.saveChatMessage({ stream_id: null, user_id: null, anon_id: 'anon7', username: 'anon7', message, message_type: 'chat', is_global: 1, ...over }).lastInsertRowid);
+}
+function assertDeletion(env, ids) {
+ const v = validate('events.event-envelope@1', env);
+ assert.ok(v.valid, JSON.stringify(v.errors));
+ assert.strictEqual(env.event_type, 'chat.message.deleted');
+ assert.strictEqual(env.source, 'chat');
+ assert.strictEqual(env.visibility, 'public');
+ assert.deepStrictEqual(env.actor, { type: 'service', id: 'chat' }, 'who deleted it is not published');
+ assert.deepStrictEqual(env.subject, { type: 'chat_message', id: String(ids[0]) });
+ assert.deepStrictEqual(env.payload, { message_ids: ids, redacts: { subject_type: 'chat_message', subject_ids: ids.map(String) } });
+}
+
+async function freePort() {
+ const s = net.createServer();
+ await new Promise((r) => s.listen(0, '127.0.0.1', r));
+ const { port } = s.address();
+ await new Promise((r) => s.close(r));
+ return port;
+}
+
+t('boot (a real OpenVibe.Events on a reserved port when one is checked out next to this repo)', async () => {
+ eventsPort = await freePort();
+ h = await boot({ env: { EVENTS_URL: `http://127.0.0.1:${eventsPort}`, EVENTS_RELAY_INTERVAL_MS: '3600000' } });
+ admin = h.addUser('admin', { role: 'admin' });
+ alice = h.addUser('alice', { subject: 'usr_01J9AAAAAAAAAAAAAAAAAAAAAA' });
+ streamer = h.addUser('streamer', { role: 'streamer' });
+ streamId = h.addStream(streamer.id, h.addChannel(streamer.id));
+ await h.ctx.sync();
+ if (haveEvents) {
+ const { load } = require(path.join(EVENTS_DIR, 'server', 'config'));
+ const { start } = require(path.join(EVENTS_DIR, 'server', 'index'));
+ const quiet = { log() {}, warn() {}, error() {} };
+ events = await start({
+ config: load({ NODE_ENV: 'test', PORT: String(eventsPort), EVENTS_DB_PATH: path.join(h.tmp, 'events.db'), OV_NETWORK_PUBLIC_KEY: h.keys.publicKey, EVENTS_WORKER: 'off' }),
+ log: quiet,
+ });
+ } else {
+ console.log(` (no OpenVibe.Events with redaction at ${EVENTS_DIR}: the end-to-end tests are skipped)`);
+ }
+});
+
+t('one message deleted (a moderator, or Live’s bridge): one deletion event, ids only', async () => {
+ const id = say(SECRET);
+ const from = outboxSeq();
+ h.db.deleteChatMessage(id, admin.id);
+ const [d] = deletions(from);
+ assertDeletion(d, [id]);
+ assert.ok(!JSON.stringify(d).includes('555-0199') && !JSON.stringify(d).includes('anon7'));
+ // Through the bridge, the way Live's /api/mod forwards it.
+ const id2 = say('via the bridge');
+ const r = await h.http('POST', '/internal/live/calls', { token: h.serviceToken(['chat.live_bridge.write']), body: { boot: 'b1', ops: [{ seq: 1, op: 'db', args: ['deleteChatMessage', id2, admin.id] }] } });
+ assert.strictEqual(r.status, 200, r.text);
+ assertDeletion(deletions(from).at(-1), [id2]);
+ const before = outboxSeq();
+ h.db.deleteChatMessage(999999, admin.id);
+ assert.strictEqual(outboxSeq(), before, 'nothing announced for a message that does not exist');
+});
+
+t('a user’s, an anon’s and a relay user’s history (self-delete, /api/mod purge)', async () => {
+ const mine = [say('a1', { user_id: alice.id, anon_id: null, username: 'alice' }), say('a2', { user_id: alice.id, anon_id: null, username: 'alice' })];
+ const anon = [say('n1', { anon_id: 'anon99', username: 'anon99' })];
+ const relay = [say('r1', { username: '[Twitch] zed', anon_id: null })];
+ let from = outboxSeq();
+ assert.deepStrictEqual(h.db.deleteUserChatMessages(alice.id, { deletedBy: alice.id }), mine, 'return value unchanged');
+ assertDeletion(deletions(from)[0], mine);
+ from = outboxSeq();
+ assert.deepStrictEqual(h.db.deleteAnonChatMessages('anon99', { deletedBy: null }), anon);
+ assertDeletion(deletions(from)[0], anon);
+ from = outboxSeq();
+ assert.deepStrictEqual(h.db.deleteRelayUserMessages('[Twitch] zed', { deletedBy: admin.id }), relay);
+ assertDeletion(deletions(from)[0], relay);
+ from = outboxSeq();
+ assert.deepStrictEqual(h.db.deleteUserChatMessages(alice.id, { deletedBy: alice.id }), [], 'nothing left');
+ assert.deepStrictEqual(deletions(from), [], 'no empty deletion events');
+});
+
+t('self-delete over the socket announces what it deleted', async () => {
+ const ws = await h.ws({ ip: '198.51.100.77', token: alice.token });
+ ws.sendJson({ type: 'join', token: alice.token });
+ await ws.next((m) => m.type === 'auth');
+ const id = say('from alice', { user_id: alice.id, anon_id: null, username: 'alice' });
+ const from = outboxSeq();
+ ws.sendJson({ type: 'self-delete-history' });
+ const res = await ws.next((m) => m.type === 'self-delete-result');
+ assert.strictEqual(res.count, 1);
+ assertDeletion(deletions(from)[0], [id]);
+ ws.close();
+});
+
+t('a time-range purge (REST) and the auto-delete sweep', async () => {
+ const inRoom = [say('s1', { stream_id: streamId, is_global: 0 }), say('s2', { stream_id: streamId, is_global: 0 })];
+ const global = say('g1');
+ let from = outboxSeq();
+ const r = await h.http('DELETE', '/api/chat/admin/purge', { token: admin.token, body: { streamId, from: h.sqliteNow(-60000), to: h.sqliteNow(60000) } });
+ assert.strictEqual(r.status, 200, r.text);
+ assert.strictEqual(r.body.deleted, 2);
+ assertDeletion(deletions(from)[0], inRoom);
+ assert.ok(!h.db.getChatMessageById(global).is_deleted, 'the other room is untouched');
+ const expiring = say('ephemeral', { auto_delete_at: h.sqliteNow(-1000) });
+ from = outboxSeq();
+ const swept = h.db.deleteExpiredChatMessages(500);
+ assert.deepStrictEqual(swept.map((x) => x.id), [expiring]);
+ assertDeletion(deletions(from)[0], [expiring]);
+});
+
+t('more than 500 ids: one event per 500, in order', async () => {
+ const ids = [];
+ h.db.transaction(() => { for (let i = 0; i < 1203; i++) ids.push(say(`bulk ${i}`, { anon_id: 'anon500', username: 'anon500' })); });
+ const from = outboxSeq();
+ h.db.deleteAnonChatMessages('anon500', { deletedBy: admin.id });
+ const d = deletions(from);
+ assert.deepStrictEqual(d.map((e) => e.payload.message_ids.length), [500, 500, 203]);
+ assert.deepStrictEqual(d.flatMap((e) => e.payload.message_ids), ids);
+ for (const e of d) assert.ok(Buffer.byteLength(JSON.stringify(e.payload)) < 64 * 1024, 'fits Events’ payload limit');
+});
+
+t('a failed delete leaves no event (same transaction)', async () => {
+ const id = say('stays');
+ const from = outboxSeq();
+ assert.throws(() => h.db.transaction(() => { h.db.deleteChatMessage(id, admin.id); throw new Error('boom'); }), /boom/);
+ assert.deepStrictEqual(deletions(from), []);
+ assert.ok(!h.db.getChatMessageById(id).is_deleted);
+});
+
+// ── End to end with OpenVibe.Events ──────────────────────────────
+
+function sse(pathAndQuery, headers = {}) {
+ return new Promise((resolve, reject) => {
+ const req = nodeHttp.get({ host: '127.0.0.1', port: eventsPort, path: pathAndQuery, headers }, (res) => {
+ let body = '';
+ const c = {
+ status: res.statusCode,
+ body: () => body,
+ // Complete blocks only; `event: gap` data has no envelope and is left out.
+ events: () => body.split('\n\n').slice(0, -1).map((b) => b.split('\n').find((l) => l.startsWith('data: ')))
+ .filter(Boolean).map((l) => JSON.parse(l.slice(6))).filter((m) => m.event),
+ async waitFor(pred, ms = 3000) {
+ const until = Date.now() + ms;
+ while (!pred(c)) { if (Date.now() > until) throw new Error(`sse timeout: ${body}`); await h.sleep(20); }
+ return c;
+ },
+ close: () => req.destroy(),
+ };
+ res.setEncoding('utf8');
+ res.on('data', (d) => { body += d; });
+ resolve(c);
+ });
+ req.on('error', reject);
+ });
+}
+const userJwt = (subjectId) => serviceAuth.signServiceToken({
+ sub: 57, subject_id: subjectId, role: 'user', iss: h.ISS, aud: ['openvibe.network'], iat: Math.floor(Date.now() / 1000), exp: Math.floor(Date.now() / 1000) + 600,
+}, h.keys.privateKey);
+
+let firstSeq, msgId;
+t('e2e: the relay publishes the message, then its deletion; replay serves only the tombstone', async () => {
+ if (!haveEvents) return;
+ h.db.run('UPDATE events_outbox SET sent_at = ? WHERE sent_at IS NULL', [new Date().toISOString()]); // earlier tests' rows
+ msgId = say(SECRET, { user_id: alice.id, anon_id: null, username: 'alice' });
+ assert.strictEqual((await h.eventsRelay.flush()).sent, 1);
+ firstSeq = events.store.lastSeq();
+ const before = await sse(`/realtime/stream?topics=chat.message.*&last_event_id=${firstSeq - 1}`);
+ await before.waitFor((c) => c.events().length === 1);
+ assert.ok(before.body().includes('555-0199'), 'before the deletion the text is there (the leak)');
+ before.close();
+
+ h.db.deleteChatMessage(msgId, admin.id);
+ assert.strictEqual((await h.eventsRelay.flush()).sent, 1);
+ const views = [
+ ['anonymous', {}],
+ ['signed in (cookie)', { Cookie: `ov_token=${userJwt('usr_01J9AAAAAAAAAAAAAAAAAAAAAA')}` }],
+ ['signed in (Bearer)', { Authorization: `Bearer ${userJwt('usr_01J9BBBBBBBBBBBBBBBBBBBBBB')}` }],
+ ['service', { Authorization: `Bearer ${h.serviceToken(['events.event.read'], { aud: 'openvibe.events', sub: 'svc:search' })}` }],
+ ];
+ for (const [who, headers] of views) {
+ const c = await sse(`/realtime/stream?topics=chat.message.*&last_event_id=${firstSeq - 1}`, headers);
+ await c.waitFor((x) => x.events().length === 2);
+ const [created, deleted] = c.events();
+ assert.deepStrictEqual([created.seq, deleted.seq], [firstSeq, firstSeq + 1], `${who}: sequence intact`);
+ assert.strictEqual(created.event.event_type, 'chat.message.created');
+ assert.strictEqual(created.event.payload.redacted, true, who);
+ assert.deepStrictEqual(created.event.subject, { type: 'chat_message', id: String(msgId) });
+ assert.strictEqual(deleted.event.event_type, 'chat.message.deleted');
+ assert.ok(!c.body().includes('555-0199') && !c.body().includes('usr_01J9AAAAAAAAAAAAAAAAAAAAAA'), `${who}: no text, no author`);
+ c.close();
+ }
+ const pull = await fetch(`http://127.0.0.1:${eventsPort}/api/v1/events?topic=chat.message.*&after_seq=${firstSeq - 1}`, {
+ headers: { Authorization: `Bearer ${h.serviceToken(['events.event.read'], { aud: 'openvibe.events', sub: 'svc:search' })}` },
+ }).then((r) => r.json());
+ assert.deepStrictEqual(pull.events.map((e) => e.seq), [firstSeq, firstSeq + 1]);
+ assert.ok(!JSON.stringify(pull).includes('555-0199'));
+});
+
+t('e2e: only svc:chat can redact Chat’s events', async () => {
+ if (!haveEvents) return;
+ const id = say('not yours to delete');
+ await h.eventsRelay.flush();
+ const created = events.store.scan(0, { patterns: ['chat.message.created'], limit: 1000 }).rows.find((r) => r.subject_id === String(id));
+ const env = deletions(0).at(-1); // a well-formed deletion, re-sourced by another service
+ const post = (sub, body) => fetch(`http://127.0.0.1:${eventsPort}/api/v1/events`, {
+ method: 'POST', body: JSON.stringify(body),
+ headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${h.serviceToken(['events.event.publish'], { aud: 'openvibe.events', sub })}` },
+ });
+ let r = await post('svc:live', { ...env, event_id: 'evt_01J9CCCCCCCCCCCCCCCCCCCCCC', source: 'live', event_type: 'live.chat.deleted', payload: { redacts: { event_ids: [created.id] } } });
+ assert.strictEqual(r.status, 403);
+ assert.strictEqual((await r.json()).code, 'events.redaction_not_allowed');
+ r = await post('svc:live', { ...env, event_id: 'evt_01J9DDDDDDDDDDDDDDDDDDDDDD', payload: { redacts: { subject_type: 'chat_message', subject_ids: [String(id)] } } });
+ assert.strictEqual(r.status, 403, 'svc:live cannot publish as chat');
+ assert.ok(events.store.getEvent(created.id).payload.includes('not yours to delete'), 'still intact');
+ h.db.deleteChatMessage(id, admin.id);
+ await h.eventsRelay.flush();
+ assert.ok(events.store.getEvent(created.id).redacted_at, 'Chat itself can');
+});
+
+t.run(async () => { if (events) await events.close(); if (h) await h.close(); });