openvibers/openvibe.chat Commits
Cursor reads name what was deleted under the cursor (reconnect convergence, roadmap D22 and WS-I task 6): a reader that reconnects and reads ?after_id=<cursor> got only the newer rows, so a line deleted while it was away stayed on its screen for good (Live's floating widget, openvibe.chat's global feed). Every cursor read (global, channel and stream history) now carries deleted_ids: the ids at or under the cursor, among the room's newest 1000 rows, that a fresh page would no longer show (deleted, or past their auto-delete time). Ids only; an id the reader never had is a no-op. Bounded: an index scan of at most 1000 rows. openvibe.chat's feed drops them on reconnect; Live's widget (chat.js hydrateWidgetOnly) should do the same. Pages are unchanged. Test: rest (global, stream, stream scope and channel cursor reads; above the cursor and other channels are not reported)
Changes
@@@ -66,11 +66,17 @@var t = el('span', 'oc-text'); textWithLinks(t, m.message); li.appendChild(t);return li;}++ function removeIds(ids) {++ if (!feed || !Array.isArray(ids)) return;++ ids.forEach(function (id) { var x = feed.querySelector('[data-id="' + Number(id) + '"]'); if (x) x.remove(); });++ }++ // After a reconnect: what was written meanwhile, and what was deleted meanwhile (deleted_ids).function catchUp() {return fetch('/api/chat/global/history?after_id=' + latest + '&limit=200', { credentials: 'same-origin' }).then(function (r) { return r.ok ? r.json() : null; }).then(function (d) {if (!d || !d.messages) return;++ removeIds(d.deleted_ids);d.messages.forEach(function (m) { if ((m.message_type || 'chat') === 'chat' && !m.is_deleted) append(globalItem(m), m.id); });if (d.latest_id) latest = Math.max(latest, d.latest_id);}).catch(function () {});@@@ -93,7 +99,7 @@if (m.id) latest = Math.max(latest, Number(m.id));append(globalItem(m), m.id);} else if (m.type === 'delete-messages' && Array.isArray(m.ids || m.message_ids)) {-- (m.ids || m.message_ids).forEach(function (id) { var x = feed.querySelector('[data-id="' + id + '"]'); if (x) x.remove(); });++ removeIds(m.ids || m.message_ids);} else if (m.type === 'error' && m.message) {showError(m.message);} else if (m.type === 'auth_revoked') {@@@ -6,6 +6,9 @@* page(room, { limit, before }) newest `limit` rows, oldest→newest — the first paint* delta(room, { afterId, limit }) rows with id > afterId, oldest→newest — reopen, reconnect,* tab focus. Cheap: a primary-key range scan, no reverse.++ * Plus `deleted_ids` (deletedIds below): the rows at or under++ * the cursor that are deleted now, so a reader that was away++ * drops them too (reconnect convergence, roadmap D22).* Every answer carries `latest_id`, the cursor the client stores; `complete:false` on a delta* means the gap was bigger than `limit` and the client should take a fresh page instead.*@@@ -20,6 +23,7 @@ const db = require('../db/database');const PAGE_MEMO_MS = 2000;const MAX_LIMIT = 500;++const GLOBAL_TYPES = "('chat', 'system', 'channel-sound', 'soundboard', 'donation')";const GLOBAL_SELECT = `SELECT cm.*, u.avatar_url, u.profile_color, u.role, u.display_name,u.username AS core_username,COALESCE(su.username, cu.username) AS stream_channel,@@@ -33,7 +37,7 @@ const GLOBAL_SELECT = `SELECT cm.*, u.avatar_url, u.profile_color, u.role, u.disLEFT JOIN ctx_users su ON s.user_id = su.idLEFT JOIN ctx_users cu ON cm.channel_user_id = cu.idLEFT JOIN ctx_managed_streams ms ON s.managed_stream_id = ms.id-- WHERE cm.is_deleted = 0 AND cm.message_type IN ('chat', 'system', 'channel-sound', 'soundboard', 'donation')++ WHERE cm.is_deleted = 0 AND cm.message_type IN ${GLOBAL_TYPES}AND (cm.auto_delete_at IS NULL OR datetime(cm.auto_delete_at) > CURRENT_TIMESTAMP)`;const CHANNEL_SELECT = `SELECT cm.*, u.avatar_url, u.profile_color, u.role, u.display_name,@@@ -138,7 +142,31 @@ function delta(room, { afterId, limit, channelUsername, decorate } = {}) {const complete = rows.length <= lim;if (!complete) rows = rows.slice(0, lim);const messages = decorate ? decorate(rows.map((x) => ({ ...x }))) : rows;-- return { messages, latest_id: latestIdOf(rows) || after, complete };++ return { messages, latest_id: latestIdOf(rows) || after, complete, deleted_ids: deletedIds(room, after) };++}++++/**++ * What a reader holding rows up to `afterId` may still show but a fresh page no longer would: the++ * ids at or under the cursor, among the room's newest DELETED_WINDOW rows (a page shows at most++ * MAX_LIMIT), that are deleted or past their auto-delete time. Ids only; one the reader never had++ * is a no-op for it. Rooms: 'global', 'channel:<userId>', 'stream:<streamId>'.++ * @returns {number[]} oldest→newest++ */++const DELETED_WINDOW = 2 * MAX_LIMIT;++function deletedIds(room, afterId) {++ const after = Math.max(0, parseInt(afterId, 10) || 0);++ if (!after) return [];++ const m = /^stream:(\d+)$/.exec(String(room || ''));++ const r = m ? { kind: 'stream', streamId: parseInt(m[1], 10) } : parseRoom(room);++ const scope = r.kind === 'global' ? { where: `message_type IN ${GLOBAL_TYPES}`, params: [] }++ : r.kind === 'channel' ? { where: 'channel_user_id = ?', params: [r.userId] }++ : { where: 'stream_id = ?', params: [r.streamId] };++ // Visible = what page()/delta() return; everything else in the window is reported.++ return db.all(`SELECT id FROM (++ SELECT id, is_deleted, auto_delete_at FROM chat_messages++ WHERE ${scope.where} AND id <= ? ORDER BY timestamp DESC, id DESC LIMIT ?)++ WHERE NOT (is_deleted = 0 AND (auto_delete_at IS NULL OR COALESCE(datetime(auto_delete_at) > CURRENT_TIMESTAMP, 0)))++ ORDER BY id`, [...scope.params, after, DELETED_WINDOW]).map((x) => x.id);}/** Drop memoised pages for a room (or all rooms). Call from write paths when a tier is added. */@@@ -147,4 +175,4 @@ function invalidate(room) {for (const k of memo.keys()) if (k.startsWith(`${room}|`)) memo.delete(k);}--module.exports = { page, delta, invalidate, clampLimit, PAGE_MEMO_MS };++module.exports = { page, delta, deletedIds, invalidate, clampLimit, PAGE_MEMO_MS, DELETED_WINDOW };@@@ -600,8 +600,10 @@ router.get('/:streamId/history', optionalAuth, async (req, res) => {}));} catch { /* non-critical */ }}++ // A cursor read also names the rows at or under the cursor deleted since (history-store.js).++ const deleted = afterId != null ? { deleted_ids: historyStore.deletedIds(spanStreamer ? `channel:${broadcasterId}` : `stream:${parseInt(req.params.streamId, 10) || 0}`, afterId) } : {};res.json({-- messages, latest_id, complete, liveSlots, channel,++ messages, latest_id, complete, ...deleted, liveSlots, channel,activeStreamId: parseInt(req.params.streamId) || null,activeManagedId: stream ? (stream.managed_stream_id || null) : null,});@@@ -123,6 +123,11 @@ t('chat history: global page and cursor delta, deleted and expired rows hidden',const delta = await h.http('GET', `/api/chat/global/history?after_id=${ids[3]}`);assert.deepStrictEqual(delta.body.messages.map((m) => m.message), ['g4']);assert.strictEqual(delta.body.complete, true);++ // Reconnect convergence: the rows at or under the cursor that a page would no longer show++ // (deleted, auto-deleted), so a reader who was away drops them too. Ids only.++ assert.deepStrictEqual(delta.body.deleted_ids.filter((id) => ids.includes(id)), [ids[1], ids[2]]);++ assert.deepStrictEqual((await h.http('GET', `/api/chat/global/history?after_id=${ids[1] - 1}`)).body.deleted_ids.filter((id) => ids.includes(id)), [], 'above the cursor is the delta’s business');++ assert.ok(!('deleted_ids' in page.body), 'a page has none');const small = await h.http('GET', `/api/chat/global/history?after_id=0&limit=1`);assert.strictEqual(small.body.complete, false, 'bigger gap than the limit → take a fresh page');// Channel filter by username (the owner of the source stream or the channel room).@@@ -130,6 +135,16 @@ t('chat history: global page and cursor delta, deleted and expired rows hidden',const byChan = await h.http('GET', '/api/chat/global/history?username=streamer');assert.deepStrictEqual(byChan.body.messages.map((m) => m.message), ['in the stream']);assert.strictEqual(byChan.body.messages[0].stream_channel, 'streamer');++ // The stream and channel cursor reads name deletions in their room too.++ const inRoom = [0, 1].map((i) => Number(h.db.saveChatMessage({ stream_id: streamId, channel_user_id: streamer.id, user_id: bob.id, username: 'Bob', message: `s${i}`, message_type: 'chat' }).lastInsertRowid));++ h.db.deleteChatMessage(inRoom[0], staff.id);++ for (const p of [`/api/chat/${streamId}/history`, `/api/chat/${streamId}/history?scope=stream&`, `/api/chat/channel/${streamer.id}/history`]) {++ const d = await h.http('GET', `${p}${p.endsWith('&') ? '' : '?'}after_id=${inRoom[1]}`);++ assert.strictEqual(d.status, 200, p);++ assert.deepStrictEqual(d.body.deleted_ids, [inRoom[0]], p);++ }++ assert.deepStrictEqual((await h.http('GET', `/api/chat/channel/${streamer.id + 999}/history?after_id=${inRoom[1]}`)).body.deleted_ids, [], 'another channel');++ h.db.deleteChatMessage(inRoom[1], staff.id); // the log counts below are about 'in the stream'});t('chat logs and search are gated like Live', async () => {
@@ -66,11 +66,17 @@
@@ -66,11 +66,17 @@
var t = el('span', 'oc-text'); textWithLinks(t, m.message); li.appendChild(t);
var t = el('span', 'oc-text'); textWithLinks(t, m.message); li.appendChild(t);
return li;
return li;
}
}
+ function removeIds(ids) {
+ if (!feed || !Array.isArray(ids)) return;
+ ids.forEach(function (id) { var x = feed.querySelector('[data-id="' + Number(id) + '"]'); if (x) x.remove(); });
+ }
+ // After a reconnect: what was written meanwhile, and what was deleted meanwhile (deleted_ids).
function catchUp() {
function catchUp() {
return fetch('/api/chat/global/history?after_id=' + latest + '&limit=200', { credentials: 'same-origin' })
return fetch('/api/chat/global/history?after_id=' + latest + '&limit=200', { credentials: 'same-origin' })
.then(function (r) { return r.ok ? r.json() : null; })
.then(function (r) { return r.ok ? r.json() : null; })
.then(function (d) {
.then(function (d) {
if (!d || !d.messages) return;
if (!d || !d.messages) return;
+ removeIds(d.deleted_ids);
d.messages.forEach(function (m) { if ((m.message_type || 'chat') === 'chat' && !m.is_deleted) append(globalItem(m), m.id); });
d.messages.forEach(function (m) { if ((m.message_type || 'chat') === 'chat' && !m.is_deleted) append(globalItem(m), m.id); });
if (d.latest_id) latest = Math.max(latest, d.latest_id);
if (d.latest_id) latest = Math.max(latest, d.latest_id);
}).catch(function () {});
}).catch(function () {});
@@ -93,7 +99,7 @@
@@ -93,7 +99,7 @@
if (m.id) latest = Math.max(latest, Number(m.id));
if (m.id) latest = Math.max(latest, Number(m.id));
append(globalItem(m), m.id);
append(globalItem(m), m.id);
} else if (m.type === 'delete-messages' && Array.isArray(m.ids || m.message_ids)) {
} else if (m.type === 'delete-messages' && Array.isArray(m.ids || m.message_ids)) {
- (m.ids || m.message_ids).forEach(function (id) { var x = feed.querySelector('[data-id="' + id + '"]'); if (x) x.remove(); });
+ removeIds(m.ids || m.message_ids);
} else if (m.type === 'error' && m.message) {
} else if (m.type === 'error' && m.message) {
showError(m.message);
showError(m.message);
} else if (m.type === 'auth_revoked') {
} else if (m.type === 'auth_revoked') {
@@ -6,6 +6,9 @@
@@ -6,6 +6,9 @@
* page(room, { limit, before }) newest `limit` rows, oldest→newest — the first paint
* page(room, { limit, before }) newest `limit` rows, oldest→newest — the first paint
* delta(room, { afterId, limit }) rows with id > afterId, oldest→newest — reopen, reconnect,
* delta(room, { afterId, limit }) rows with id > afterId, oldest→newest — reopen, reconnect,
* tab focus. Cheap: a primary-key range scan, no reverse.
* tab focus. Cheap: a primary-key range scan, no reverse.
+ * Plus `deleted_ids` (deletedIds below): the rows at or under
+ * the cursor that are deleted now, so a reader that was away
+ * drops them too (reconnect convergence, roadmap D22).
* Every answer carries `latest_id`, the cursor the client stores; `complete:false` on a delta
* Every answer carries `latest_id`, the cursor the client stores; `complete:false` on a delta
* means the gap was bigger than `limit` and the client should take a fresh page instead.
* means the gap was bigger than `limit` and the client should take a fresh page instead.
*
*
@@ -20,6 +23,7 @@ const db = require('../db/database');
@@ -20,6 +23,7 @@ const db = require('../db/database');
const PAGE_MEMO_MS = 2000;
const PAGE_MEMO_MS = 2000;
const MAX_LIMIT = 500;
const MAX_LIMIT = 500;
+const GLOBAL_TYPES = "('chat', 'system', 'channel-sound', 'soundboard', 'donation')";
const GLOBAL_SELECT = `SELECT cm.*, u.avatar_url, u.profile_color, u.role, u.display_name,
const GLOBAL_SELECT = `SELECT cm.*, u.avatar_url, u.profile_color, u.role, u.display_name,
u.username AS core_username,
u.username AS core_username,
COALESCE(su.username, cu.username) AS stream_channel,
COALESCE(su.username, cu.username) AS stream_channel,
@@ -33,7 +37,7 @@ const GLOBAL_SELECT = `SELECT cm.*, u.avatar_url, u.profile_color, u.role, u.dis
@@ -33,7 +37,7 @@ const GLOBAL_SELECT = `SELECT cm.*, u.avatar_url, u.profile_color, u.role, u.dis
LEFT JOIN ctx_users su ON s.user_id = su.id
LEFT JOIN ctx_users su ON s.user_id = su.id
LEFT JOIN ctx_users cu ON cm.channel_user_id = cu.id
LEFT JOIN ctx_users cu ON cm.channel_user_id = cu.id
LEFT JOIN ctx_managed_streams ms ON s.managed_stream_id = ms.id
LEFT JOIN ctx_managed_streams ms ON s.managed_stream_id = ms.id
- WHERE cm.is_deleted = 0 AND cm.message_type IN ('chat', 'system', 'channel-sound', 'soundboard', 'donation')
+ WHERE cm.is_deleted = 0 AND cm.message_type IN ${GLOBAL_TYPES}
AND (cm.auto_delete_at IS NULL OR datetime(cm.auto_delete_at) > CURRENT_TIMESTAMP)`;
AND (cm.auto_delete_at IS NULL OR datetime(cm.auto_delete_at) > CURRENT_TIMESTAMP)`;
const CHANNEL_SELECT = `SELECT cm.*, u.avatar_url, u.profile_color, u.role, u.display_name,
const CHANNEL_SELECT = `SELECT cm.*, u.avatar_url, u.profile_color, u.role, u.display_name,
@@ -138,7 +142,31 @@ function delta(room, { afterId, limit, channelUsername, decorate } = {}) {
@@ -138,7 +142,31 @@ function delta(room, { afterId, limit, channelUsername, decorate } = {}) {
const complete = rows.length <= lim;
const complete = rows.length <= lim;
if (!complete) rows = rows.slice(0, lim);
if (!complete) rows = rows.slice(0, lim);
const messages = decorate ? decorate(rows.map((x) => ({ ...x }))) : rows;
const messages = decorate ? decorate(rows.map((x) => ({ ...x }))) : rows;
- return { messages, latest_id: latestIdOf(rows) || after, complete };
+ return { messages, latest_id: latestIdOf(rows) || after, complete, deleted_ids: deletedIds(room, after) };
+}
+
+/**
+ * What a reader holding rows up to `afterId` may still show but a fresh page no longer would: the
+ * ids at or under the cursor, among the room's newest DELETED_WINDOW rows (a page shows at most
+ * MAX_LIMIT), that are deleted or past their auto-delete time. Ids only; one the reader never had
+ * is a no-op for it. Rooms: 'global', 'channel:<userId>', 'stream:<streamId>'.
+ * @returns {number[]} oldest→newest
+ */
+const DELETED_WINDOW = 2 * MAX_LIMIT;
+function deletedIds(room, afterId) {
+ const after = Math.max(0, parseInt(afterId, 10) || 0);
+ if (!after) return [];
+ const m = /^stream:(\d+)$/.exec(String(room || ''));
+ const r = m ? { kind: 'stream', streamId: parseInt(m[1], 10) } : parseRoom(room);
+ const scope = r.kind === 'global' ? { where: `message_type IN ${GLOBAL_TYPES}`, params: [] }
+ : r.kind === 'channel' ? { where: 'channel_user_id = ?', params: [r.userId] }
+ : { where: 'stream_id = ?', params: [r.streamId] };
+ // Visible = what page()/delta() return; everything else in the window is reported.
+ return db.all(`SELECT id FROM (
+ SELECT id, is_deleted, auto_delete_at FROM chat_messages
+ WHERE ${scope.where} AND id <= ? ORDER BY timestamp DESC, id DESC LIMIT ?)
+ WHERE NOT (is_deleted = 0 AND (auto_delete_at IS NULL OR COALESCE(datetime(auto_delete_at) > CURRENT_TIMESTAMP, 0)))
+ ORDER BY id`, [...scope.params, after, DELETED_WINDOW]).map((x) => x.id);
}
}
/** Drop memoised pages for a room (or all rooms). Call from write paths when a tier is added. */
/** Drop memoised pages for a room (or all rooms). Call from write paths when a tier is added. */
@@ -147,4 +175,4 @@ function invalidate(room) {
@@ -147,4 +175,4 @@ function invalidate(room) {
for (const k of memo.keys()) if (k.startsWith(`${room}|`)) memo.delete(k);
for (const k of memo.keys()) if (k.startsWith(`${room}|`)) memo.delete(k);
}
}
-module.exports = { page, delta, invalidate, clampLimit, PAGE_MEMO_MS };
+module.exports = { page, delta, deletedIds, invalidate, clampLimit, PAGE_MEMO_MS, DELETED_WINDOW };
@@ -600,8 +600,10 @@ router.get('/:streamId/history', optionalAuth, async (req, res) => {
@@ -600,8 +600,10 @@ router.get('/:streamId/history', optionalAuth, async (req, res) => {
}));
}));
} catch { /* non-critical */ }
} catch { /* non-critical */ }
}
}
+ // A cursor read also names the rows at or under the cursor deleted since (history-store.js).
+ const deleted = afterId != null ? { deleted_ids: historyStore.deletedIds(spanStreamer ? `channel:${broadcasterId}` : `stream:${parseInt(req.params.streamId, 10) || 0}`, afterId) } : {};
res.json({
res.json({
- messages, latest_id, complete, liveSlots, channel,
+ messages, latest_id, complete, ...deleted, liveSlots, channel,
activeStreamId: parseInt(req.params.streamId) || null,
activeStreamId: parseInt(req.params.streamId) || null,
activeManagedId: stream ? (stream.managed_stream_id || null) : null,
activeManagedId: stream ? (stream.managed_stream_id || null) : null,
});
});
@@ -123,6 +123,11 @@ t('chat history: global page and cursor delta, deleted and expired rows hidden',
@@ -123,6 +123,11 @@ t('chat history: global page and cursor delta, deleted and expired rows hidden',
const delta = await h.http('GET', `/api/chat/global/history?after_id=${ids[3]}`);
const delta = await h.http('GET', `/api/chat/global/history?after_id=${ids[3]}`);
assert.deepStrictEqual(delta.body.messages.map((m) => m.message), ['g4']);
assert.deepStrictEqual(delta.body.messages.map((m) => m.message), ['g4']);
assert.strictEqual(delta.body.complete, true);
assert.strictEqual(delta.body.complete, true);
+ // Reconnect convergence: the rows at or under the cursor that a page would no longer show
+ // (deleted, auto-deleted), so a reader who was away drops them too. Ids only.
+ assert.deepStrictEqual(delta.body.deleted_ids.filter((id) => ids.includes(id)), [ids[1], ids[2]]);
+ assert.deepStrictEqual((await h.http('GET', `/api/chat/global/history?after_id=${ids[1] - 1}`)).body.deleted_ids.filter((id) => ids.includes(id)), [], 'above the cursor is the delta’s business');
+ assert.ok(!('deleted_ids' in page.body), 'a page has none');
const small = await h.http('GET', `/api/chat/global/history?after_id=0&limit=1`);
const small = await h.http('GET', `/api/chat/global/history?after_id=0&limit=1`);
assert.strictEqual(small.body.complete, false, 'bigger gap than the limit → take a fresh page');
assert.strictEqual(small.body.complete, false, 'bigger gap than the limit → take a fresh page');
// Channel filter by username (the owner of the source stream or the channel room).
// Channel filter by username (the owner of the source stream or the channel room).
@@ -130,6 +135,16 @@ t('chat history: global page and cursor delta, deleted and expired rows hidden',
@@ -130,6 +135,16 @@ t('chat history: global page and cursor delta, deleted and expired rows hidden',
const byChan = await h.http('GET', '/api/chat/global/history?username=streamer');
const byChan = await h.http('GET', '/api/chat/global/history?username=streamer');
assert.deepStrictEqual(byChan.body.messages.map((m) => m.message), ['in the stream']);
assert.deepStrictEqual(byChan.body.messages.map((m) => m.message), ['in the stream']);
assert.strictEqual(byChan.body.messages[0].stream_channel, 'streamer');
assert.strictEqual(byChan.body.messages[0].stream_channel, 'streamer');
+ // The stream and channel cursor reads name deletions in their room too.
+ const inRoom = [0, 1].map((i) => Number(h.db.saveChatMessage({ stream_id: streamId, channel_user_id: streamer.id, user_id: bob.id, username: 'Bob', message: `s${i}`, message_type: 'chat' }).lastInsertRowid));
+ h.db.deleteChatMessage(inRoom[0], staff.id);
+ for (const p of [`/api/chat/${streamId}/history`, `/api/chat/${streamId}/history?scope=stream&`, `/api/chat/channel/${streamer.id}/history`]) {
+ const d = await h.http('GET', `${p}${p.endsWith('&') ? '' : '?'}after_id=${inRoom[1]}`);
+ assert.strictEqual(d.status, 200, p);
+ assert.deepStrictEqual(d.body.deleted_ids, [inRoom[0]], p);
+ }
+ assert.deepStrictEqual((await h.http('GET', `/api/chat/channel/${streamer.id + 999}/history?after_id=${inRoom[1]}`)).body.deleted_ids, [], 'another channel');
+ h.db.deleteChatMessage(inRoom[1], staff.id); // the log counts below are about 'in the stream'
});
});
t('chat logs and search are gated like Live', async () => {
t('chat logs and search are gated like Live', async () => {