openvibers/openvibe.live Commits
Global chat widget: cursor-based history, refresh on open/reconnect/focus; chat history store
Changes
@@@ -2238,31 +2238,75 @@ function _renderHistoryPayload(p) {* be fetched exclusively through the main pane's hydrate, so the widget opened empty there and* stayed empty until someone typed. Fill it directly: cached copy first, then the network.*/--let _fcwHydratedAt = 0, _fcwHydrating = false;--async function hydrateWidgetOnly() {++let _fcwHydratedAt = 0, _fcwHydrating = false, _fcwRefreshedAt = 0, _fcwSendRetries = 0;++const _FCW_PAGE = '/chat/global/history?limit=500';++/**++ * Bring the widget up to date. Three shapes, picked by what the widget already holds:++ * empty → cached copy first (instant paint), then a delta from the cache's cursor;++ * rows + cursor → delta: only ids newer than the newest row we have (cheap PK scan);++ * gap too large → the server says `complete:false`: wipe and take a fresh page.++ * Runs on every open, on (re)connect, on tab focus and on coming back online — the old++ * "hydrated less than 10 minutes ago" guard is what left stale rows on screen after the SPA++ * router tore the socket down. A short throttle collapses the triggers that fire together.++ */++async function hydrateWidgetOnly({ reason = 'open' } = {}) {const container = document.getElementById('fcw-messages');if (!container || _fcwHydrating) return;-- if (container.querySelector('.chat-msg') && Date.now() - _fcwHydratedAt < 10 * 60 * 1000) return;++ const hasRows = !!container.querySelector('.chat-msg');++ if (hasRows && _fcwLastId && Date.now() - _fcwRefreshedAt < 1500) return;_fcwHydrating = true;-- const feed = (payload) => {-- const rows = (payload && payload.data && payload.data.messages) || [];-- if (!rows.length) return 0;++ const feed = (rows) => {++ if (!Array.isArray(rows) || !rows.length) return 0;const prev = _loadingHistory; _loadingHistory = true;try { rows.slice(-200).forEach((m) => _fcwAddMessage(m)); } finally { _loadingHistory = prev; }_fcwScrollToBottom();return rows.length;};++ const fresh = async () => {++ const page = await api(_FCW_PAGE);++ container.replaceChildren();++ _fcwIds.clear(); _fcwLastId = 0;++ feed(page && page.messages);++ try { _chatCacheWrite('global', { kind: 'global', data: page }); } catch { /* */ }++ };try {-- if (!container.querySelector('.chat-msg')) { const cached = _chatCacheRead('global'); if (cached) feed(cached); }-- try { if (typeof loadEmotes === 'function') await loadEmotes(); } catch { /* plain text is fine */ }-- const payload = await _fetchHistoryPayload(null);-- feed(payload);-- _fcwHydratedAt = Date.now();-- try { _chatCacheWrite('global', payload); } catch { /* */ }++ if (!hasRows) {++ const cached = _chatCacheRead('global');++ if (cached) feed(cached.data.messages);++ try { if (typeof loadEmotes === 'function') await loadEmotes(); } catch { /* plain text is fine */ }++ }++ if (_fcwLastId) {++ const d = await api(`/chat/global/history?after_id=${_fcwLastId}&limit=200`);++ if (!d || d.complete === false) await fresh();++ else if (d.messages && d.messages.length) { feed(d.messages); _fcwCacheAppend(d.messages); }++ } else {++ await fresh();++ }++ _fcwHydratedAt = _fcwRefreshedAt = Date.now();} catch (err) {-- console.warn('[chat] widget history failed:', err && err.message);++ console.warn(`[chat] widget history (${reason}) failed:`, err && err.message);} finally { _fcwHydrating = false; }}++/** Delta rows extend the cached global copy so the next cold open starts from them. */++function _fcwCacheAppend(rows) {++ try {++ const key = _chatCacheKey('global');++ const raw = localStorage.getItem(key);++ const entry = raw ? JSON.parse(raw) : { kind: 'global', data: { messages: [] } };++ const seen = new Set((entry.data.messages || []).map((m) => String(m.id)));++ for (const m of rows) if (m && m.id != null && !seen.has(String(m.id))) entry.data.messages.push(m);++ if (entry.data.messages.length > _CHAT_CACHE_ROWS) entry.data.messages.splice(0, entry.data.messages.length - _CHAT_CACHE_ROWS);++ entry.at = Date.now();++ localStorage.setItem(key, JSON.stringify(entry));++ } catch { /* best effort */ }++}++/** The widget refreshes itself whenever the page comes back: tab focus, network back, socket back. */++function _fcwRefreshIfRelevant(reason) {++ const container = document.getElementById('fcw-messages');++ if (!container) return;++ if (!_fcwOpen && !container.querySelector('.chat-msg')) return; // never opened: nothing to keep current++ hydrateWidgetOnly({ reason }).catch(() => {});++}async function hydrateActiveChatHistory(streamId, { clear = false } = {}) {const chatEl = getChatEl();@@@ -2566,6 +2610,9 @@ function initChat(streamId, channelUserId = null) {ws._authToken = token || null;_chatReconnectDelay = CHAT_RECONNECT_BASE; // reset backoff on successaddRichSystemMessage('<i class="fa-solid fa-plug-circle-check" style="margin-right:5px;opacity:0.8"></i> Connected to chat');++ // A fresh socket after the router tore the last one down: the widget may hold rows from++ // before the gap. Pull what it missed instead of waiting for someone to type.++ _fcwRefreshIfRelevant('connect');};// Load emotes + channel sounds for this stream context@@@ -2671,6 +2718,13 @@ window.addEventListener('online', () => {clearTimeout(_chatReconnectTimer); _chatReconnectTimer = null;_reconnectChatWs(chatStreamId ?? null);});++// Mobile browsers freeze the page (and often the socket) in the background; on return, the widget++// asks for what it missed rather than trusting rows that may be an hour old.++document.addEventListener('visibilitychange', () => {++ if (document.visibilityState !== 'visible' || !_fcwOpen) return;++ if (!chatWs || chatWs.readyState !== WebSocket.OPEN) { try { initChat(null); } catch { /* */ } }++ _fcwRefreshIfRelevant('focus');++});window.addEventListener('offline', () => { if (_chatActive && !_chatIntentionalClose) _chatConnState('offline'); });/**@@@ -2714,6 +2768,7 @@ function _reconnectChatWs(streamId) {// Messages sent while we were away are only in the history. Refill it — but not under a reader// who has scrolled back, since reloading history redraws the list.if (outage > 1500 && !_chatUserScrolledUp) hydrateActiveChatHistory(streamId).catch(() => {});++ _fcwRefreshIfRelevant('reconnect');};ws.onmessage = (e) => {@@@ -2780,6 +2835,8 @@ function destroyChat(forceClose = false) {}chatStreamId = null;}++ // The channel scope belongs to the page that set it; a later global fetch must not inherit it.++ chatChannelUserId = null;chatRenderTargetId = null;clearFullscreenChatMessages();_fullscreenChatRecent = [];@@@ -7079,8 +7136,9 @@ function fcwToggle() {if (!chatWs || chatWs.readyState !== WebSocket.OPEN) {initChat(null);}-- // The widget is its own chat surface on pages without a pane: give it the history.-- hydrateWidgetOnly();++ // The widget is its own chat surface on pages without a pane: give it the history —++ // a delta from the newest row it holds, or a fresh page when it has nothing.++ hydrateWidgetOnly({ reason: 'open' });// Scroll to bottomconst msgs = document.getElementById('fcw-messages');@@@ -7102,7 +7160,16 @@ function fcwSendChat() {const input = document.getElementById('fcw-chat-input');if (!input) return;const text = input.value.trim();-- if (!text || !chatWs || chatWs.readyState !== WebSocket.OPEN) return;++ if (!text) return;++ if (!chatWs || chatWs.readyState !== WebSocket.OPEN) {++ // The router closes the socket on navigation; bring it back and send once it is open.++ _fcwSendRetries = (_fcwSendRetries || 0) + 1;++ if (_fcwSendRetries > 4) { _fcwSendRetries = 0; return; }++ try { initChat(null); } catch { /* */ }++ setTimeout(() => { if (document.getElementById('fcw-chat-input')) fcwSendChat(); }, 900);++ return;++ }++ _fcwSendRetries = 0;// Save to shared message historyif (_chatMsgHistory[0] !== text) {@@@ -7143,11 +7210,36 @@ function _chatTrimWidget(container) {while (container.children.length > 200) container.removeChild(container.firstChild);}++// Ids the widget has shown, kept past the 200-row DOM trim so a trimmed row cannot come back++// through a delta, and the cursor the next delta read starts from.++const _fcwIds = new Set();++let _fcwLastId = 0;++function _fcwRemember(id) {++ _fcwIds.add(String(id));++ const n = Number(id);++ if (Number.isFinite(n) && n > _fcwLastId) _fcwLastId = n;++ if (_fcwIds.size > 1500) { const it = _fcwIds.values(); for (let i = 0; i < 500; i++) _fcwIds.delete(it.next().value); }++}++// Rows arrive out of order when a delta lands under live rows that beat it; keep the list in id order.++function _fcwInsertOrdered(container, el, id) {++ const n = Number(id);++ let ref = null;++ if (Number.isFinite(n)) {++ for (let node = container.lastElementChild; node; node = node.previousElementSibling) {++ const nid = Number(node.dataset && node.dataset.msgId);++ if (!Number.isFinite(nid) || nid <= n) break;++ ref = node;++ }++ }++ if (ref) container.insertBefore(el, ref); else container.appendChild(el);++}function _fcwAddMessage(msg) {if (_fcwHistoryBuffer) { _fcwHistoryBuffer.push(msg); return; }const container = document.getElementById('fcw-messages');if (!container) return;-- if (msg && msg.id != null && container.querySelector(`[data-msg-id="${CSS.escape(String(msg.id))}"]`)) return;++ const mid = msg && msg.id != null ? String(msg.id) : null;++ if (mid && (_fcwIds.has(mid) || container.querySelector(`[data-msg-id="${CSS.escape(mid)}"]`))) return;++ if (mid) _fcwRemember(mid);// Stream source badge — clickable stream TITLE (live → watch, offline → channel).const src = _fcwSourceLabel(msg);@@@ -7169,7 +7261,7 @@ function _fcwAddMessage(msg) {// user menu on tap — the widget's own renderer had no menu at all).const el = buildChatMessageEl(msg, { widget: true, prefixHtml: srcBadge });if (_friendlyHides(msg)) _friendlyAppend(container, el);-- else container.appendChild(el);++ else _fcwInsertOrdered(container, el, mid);_chatTrimWidget(container);// One scroll per frame, however many rows arrived in it (history mirrors hundreds at once).nnew file mode 100644@@@ -0,0 +1,149 @@++/**++ * history-store.js — the one place that answers "what happened in this chat room?".++ *++ * Rooms: 'global' | 'channel:<userId>' (the stream route keeps its own scope logic)++ * Reads:++ * 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.++ * 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.++ *++ * Backing store today is SQLite through `db`. `memo` collapses identical page reads inside a++ * short window (a burst of reconnects after a deploy used to run the five-table global join++ * once per socket). Swapping in a ring buffer or Redis later means replacing `readPage` /++ * `readDelta` and calling `invalidate(room)` from the write path — nothing in the routes changes.++ */++const db = require('../db/database');++++const PAGE_MEMO_MS = 2000;++const MAX_LIMIT = 500;++++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,++ s.is_live AS source_is_live,++ s.managed_stream_id AS source_managed_id,++ COALESCE(ms.title, s.title) AS source_stream_title,++ ms.slug AS source_slug++ FROM chat_messages cm++ LEFT JOIN users u ON cm.user_id = u.id++ LEFT JOIN streams s ON cm.stream_id = s.id++ LEFT JOIN users su ON s.user_id = su.id++ LEFT JOIN users cu ON cm.channel_user_id = cu.id++ LEFT JOIN 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')++ 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,++ u.username AS core_username,++ s.title AS source_stream_title, s.managed_stream_id AS source_managed_id,++ s.is_live AS source_is_live, ms.slug AS source_slug,++ COALESCE(bu.username, cu.username) AS source_channel++ FROM chat_messages cm++ LEFT JOIN users u ON cm.user_id = u.id++ LEFT JOIN streams s ON cm.stream_id = s.id++ LEFT JOIN managed_streams ms ON s.managed_stream_id = ms.id++ LEFT JOIN users bu ON s.user_id = bu.id++ LEFT JOIN users cu ON cm.channel_user_id = cu.id++ WHERE cm.channel_user_id = ? AND cm.is_deleted = 0++ AND (cm.auto_delete_at IS NULL OR datetime(cm.auto_delete_at) > CURRENT_TIMESTAMP)`;++++function clampLimit(v, dflt = 500) {++ const n = parseInt(v, 10);++ return Math.min(Number.isFinite(n) && n > 0 ? n : dflt, MAX_LIMIT);++}++++function parseRoom(room) {++ if (room === 'global') return { kind: 'global' };++ const m = /^channel:(\d+)$/.exec(String(room || ''));++ if (m) return { kind: 'channel', userId: parseInt(m[1], 10) };++ throw new Error(`history-store: unknown room "${room}"`);++}++++function baseSql(r, { channelUsername } = {}) {++ if (r.kind === 'global') {++ const params = [];++ let sql = GLOBAL_SELECT;++ if (channelUsername) { sql += ` AND LOWER(COALESCE(su.username, cu.username)) = LOWER(?)`; params.push(channelUsername); }++ return { sql, params };++ }++ return { sql: CHANNEL_SELECT, params: [r.userId] };++}++++function tipId() {++ try { const r = db.get('SELECT MAX(id) AS id FROM chat_messages'); return (r && r.id) || 0; } catch { return Date.now(); }++}++const latestIdOf = (rows) => rows.reduce((m, x) => (x && x.id > m ? x.id : m), 0);++++// ── reads ──────────────────────────────────────────────────────────────────────────────────────++const memo = new Map(); // key → { at, value }++function memoGet(key) {++ const hit = memo.get(key);++ if (hit && Date.now() - hit.at < PAGE_MEMO_MS) return hit.value;++ if (hit) memo.delete(key);++ return null;++}++function memoSet(key, value) {++ memo.set(key, { at: Date.now(), value });++ if (memo.size > 64) { const oldest = memo.keys().next().value; memo.delete(oldest); }++}++++function readPage(r, { limit, before, channelUsername }) {++ const { sql, params } = baseSql(r, { channelUsername });++ let q = sql;++ if (before) { q += ' AND cm.timestamp < ?'; params.push(before); }++ q += ' ORDER BY cm.timestamp DESC, cm.id DESC LIMIT ?'; params.push(limit);++ return db.all(q, params).reverse();++}++++function readDelta(r, { afterId, limit, channelUsername }) {++ const { sql, params } = baseSql(r, { channelUsername });++ // One more than asked: if it comes back, the gap is bigger than the client can splice in.++ const q = `${sql} AND cm.id > ? ORDER BY cm.id ASC LIMIT ?`;++ params.push(afterId, limit + 1);++ return db.all(q, params);++}++++/**++ * Newest rows of a room, oldest→newest.++ * @returns {{ messages: object[], latest_id: number }}++ */++function page(room, { limit, before, channelUsername, decorate } = {}) {++ const r = parseRoom(room);++ const lim = clampLimit(limit);++ // The newest row id is part of the memo key, so a memoised page can never miss a message:++ // one primary-key lookup instead of the five-table join for every repeat reader in the window.++ const key = !before ? `${room}|${lim}|${channelUsername || ''}|${tipId()}` : null;++ let rows = key ? memoGet(key) : null;++ if (!rows) {++ rows = readPage(r, { limit: lim, before, channelUsername });++ if (key) memoSet(key, rows);++ }++ const messages = decorate ? decorate(rows.map((x) => ({ ...x }))) : rows;++ return { messages, latest_id: latestIdOf(rows) };++}++++/**++ * Everything after a cursor, oldest→newest. `complete:false` = more than `limit` rows were++ * missing; take a fresh page instead of splicing.++ * @returns {{ messages: object[], latest_id: number, complete: boolean }}++ */++function delta(room, { afterId, limit, channelUsername, decorate } = {}) {++ const r = parseRoom(room);++ const lim = clampLimit(limit, 200);++ const after = Math.max(0, parseInt(afterId, 10) || 0);++ let rows = readDelta(r, { afterId: after, limit: lim, channelUsername });++ 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 };++}++++/** Drop memoised pages for a room (or all rooms). Call from write paths when a tier is added. */++function invalidate(room) {++ if (!room) { memo.clear(); return; }++ for (const k of memo.keys()) if (k.startsWith(`${room}|`)) memo.delete(k);++}++++module.exports = { page, delta, invalidate, clampLimit, PAGE_MEMO_MS };@@@ -12,6 +12,7 @@ const express = require('express');const db = require('../db/database');const { optionalAuth, requireAuth, requireAdmin } = require('../auth/auth');const permissions = require('../auth/permissions');++const historyStore = require('./history-store');const router = express.Router();@@@ -466,42 +467,17 @@ router.get('/filters/friendly', (req, res) => {router.get('/global/history', optionalAuth, (req, res) => {try {-- const limit = Math.min(parseInt(req.query.limit || '500'), 500);++ const limit = req.query.limit;const before = req.query.before;++ const afterId = req.query.after_id;const channelUsername = String(req.query.username || '').trim();---- let sql = `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,-- s.is_live AS source_is_live,-- s.managed_stream_id AS source_managed_id,-- COALESCE(ms.title, s.title) AS source_stream_title,-- ms.slug AS source_slug-- FROM chat_messages cm-- LEFT JOIN users u ON cm.user_id = u.id-- LEFT JOIN streams s ON cm.stream_id = s.id-- LEFT JOIN users su ON s.user_id = su.id-- LEFT JOIN users cu ON cm.channel_user_id = cu.id-- LEFT JOIN 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')-- AND (cm.auto_delete_at IS NULL OR datetime(cm.auto_delete_at) > CURRENT_TIMESTAMP)`;-- const params = [];---- if (channelUsername) {-- sql += ` AND LOWER(COALESCE(su.username, cu.username)) = LOWER(?)`;-- params.push(channelUsername);-- }---- if (before) {-- sql += ` AND cm.timestamp < ?`;-- params.push(before);-- }---- sql += ` ORDER BY cm.timestamp DESC LIMIT ?`;-- params.push(limit);---- const messages = enrichMessagesWithCosmetics(hydrateReplies(db.all(sql, params).reverse()));-- res.json({ messages });++ const decorate = (rows) => enrichMessagesWithCosmetics(hydrateReplies(rows));++ // after_id = cursor read: only what the caller has not seen (oldest→newest, PK range scan).++ // Anything else = a page: the newest rows, optionally before a timestamp.++ const out = afterId != null++ ? historyStore.delta('global', { afterId, limit, channelUsername, decorate })++ : historyStore.page('global', { limit, before, channelUsername, decorate });++ res.json(out);} catch (err) {res.status(500).json({ error: 'Failed to get global chat history' });}@@@ -578,15 +554,26 @@ router.get('/:streamId/history', optionalAuth, (req, res) => {params.push(req.params.streamId);}-- if (before) {-- sql += ` AND cm.timestamp < ?`;-- params.push(before);++ // after_id: cursor read (oldest→newest, primary-key range scan) for reopen/reconnect.++ // Otherwise a page: the newest rows, optionally before a timestamp.++ const afterId = req.query.after_id != null ? Math.max(0, parseInt(req.query.after_id, 10) || 0) : null;++ if (afterId != null) {++ sql += ` AND cm.id > ? ORDER BY cm.id ASC LIMIT ?`;++ params.push(afterId, limit + 1);++ } else {++ if (before) {++ sql += ` AND cm.timestamp < ?`;++ params.push(before);++ }++ sql += ` ORDER BY cm.timestamp DESC LIMIT ?`;++ params.push(limit);}-- sql += ` ORDER BY cm.timestamp DESC LIMIT ?`;-- params.push(limit);---- const messages = enrichMessagesWithCosmetics(hydrateReplies(db.all(sql, params).reverse()));++ let rows = db.all(sql, params);++ let complete = true;++ if (afterId != null) { complete = rows.length <= limit; rows = rows.slice(0, limit); } else rows.reverse();++ const latest_id = rows.reduce((m, x) => (x.id > m ? x.id : m), 0) || (afterId || 0);++ const messages = enrichMessagesWithCosmetics(hydrateReplies(rows));// Currently-live slots for this broadcaster — lets the client render a// "hop between live streams" affordance.@@@ -607,7 +594,7 @@ router.get('/:streamId/history', optionalAuth, (req, res) => {} catch { /* non-critical */ }}res.json({-- messages, liveSlots, channel,++ messages, latest_id, complete, liveSlots, channel,activeStreamId: parseInt(req.params.streamId) || null,activeManagedId: stream ? (stream.managed_stream_id || null) : null,});@@@ -623,26 +610,11 @@ router.get('/channel/:userId/history', optionalAuth, (req, res) => {try {const userId = parseInt(req.params.userId);if (!userId) return res.status(400).json({ error: 'Bad user id' });-- const limit = Math.min(parseInt(req.query.limit || '500'), 500);-- const before = req.query.before;-- const select = `SELECT cm.*, u.avatar_url, u.profile_color, u.role, u.display_name,-- u.username AS core_username,-- s.title AS source_stream_title, s.managed_stream_id AS source_managed_id,-- s.is_live AS source_is_live, ms.slug AS source_slug,-- COALESCE(bu.username, cu.username) AS source_channel`;-- let sql = `${select}-- FROM chat_messages cm-- LEFT JOIN users u ON cm.user_id = u.id-- LEFT JOIN streams s ON cm.stream_id = s.id-- LEFT JOIN managed_streams ms ON s.managed_stream_id = ms.id-- LEFT JOIN users bu ON s.user_id = bu.id-- LEFT JOIN users cu ON cm.channel_user_id = cu.id-- WHERE cm.channel_user_id = ? AND cm.is_deleted = 0-- AND (cm.auto_delete_at IS NULL OR datetime(cm.auto_delete_at) > CURRENT_TIMESTAMP)`;-- const params = [userId];-- if (before) { sql += ' AND cm.timestamp < ?'; params.push(before); }-- sql += ' ORDER BY cm.timestamp DESC LIMIT ?'; params.push(limit);-- const messages = enrichMessagesWithCosmetics(hydrateReplies(db.all(sql, params).reverse()));++ const decorate = (rows) => enrichMessagesWithCosmetics(hydrateReplies(rows));++ const room = `channel:${userId}`;++ const out = req.query.after_id != null++ ? historyStore.delta(room, { afterId: req.query.after_id, limit: req.query.limit, decorate })++ : historyStore.page(room, { limit: req.query.limit, before: req.query.before, decorate });let liveSlots = [], channel = null;try {const owner = db.getUserById(userId);@@@ -651,7 +623,7 @@ router.get('/channel/:userId/history', optionalAuth, (req, res) => {.filter(ms => ms.is_currently_live).map(ms => ({ managed_stream_id: ms.id, slug: ms.slug, title: ms.title, live_session_id: ms.live_session_id }));} catch { /* non-critical */ }-- res.json({ messages, liveSlots, channel, activeStreamId: null, activeManagedId: null });++ res.json({ ...out, liveSlots, channel, activeStreamId: null, activeManagedId: null });} catch (err) {res.status(500).json({ error: 'Failed to get channel chat history' });}nnew file mode 100644@@@ -0,0 +1,67 @@++'use strict';++/**++ * Chat history reads through one store: a page is the newest rows oldest→newest, a delta is only++ * what a client with a cursor has not seen, and a gap wider than the cap says so instead of++ * silently handing back a slice the client would splice in the wrong place.++ */++const assert = require('assert');++const os = require('os');++const path = require('path');++++process.env.DB_PATH = path.join(os.tmpdir(), `ov-chat-history-${Date.now()}.db`);++const db = require('../server/db/database');++db.initDb();++const store = require('../server/chat/history-store');++++const u = db.run(`INSERT INTO users (username, password_hash, email) VALUES ('alice', 'x', '[email protected]')`).lastInsertRowid;++const bob = db.run(`INSERT INTO users (username, password_hash, email) VALUES ('bob', 'x', '[email protected]')`).lastInsertRowid;++++const ids = [];++for (let i = 1; i <= 12; i++) {++ const r = db.saveChatMessage({ user_id: u, username: 'alice', message: `m${i}`, stream_id: null, is_global: 1, message_type: 'chat' });++ ids.push(Number(r.lastInsertRowid));++}++// One deleted and one auto-expired row must never come back through either read.++db.run(`UPDATE chat_messages SET is_deleted = 1 WHERE id = ?`, [ids[5]]);++db.run(`UPDATE chat_messages SET auto_delete_at = datetime('now', '-1 minute') WHERE id = ?`, [ids[6]]);++// The global feed is every public room with a source badge, so a channel row belongs in it too.++const chanId = Number(db.saveChatMessage({ user_id: bob, username: 'bob', message: 'channel only', stream_id: null, channel_user_id: bob, message_type: 'chat' }).lastInsertRowid);++++// ── page ──────────────────────────────────────────────────────────────────────────────────────++const page = store.page('global', { limit: 5 });++assert.equal(page.messages.length, 5, 'a page is capped by limit');++assert.deepEqual(page.messages.map((m) => m.message), ['m9', 'm10', 'm11', 'm12', 'channel only'], 'newest rows, oldest → newest, all rooms');++assert.equal(page.latest_id, chanId, 'latest_id is the cursor of the newest row');++++// ── delta ─────────────────────────────────────────────────────────────────────────────────────++const d = store.delta('global', { afterId: ids[3], limit: 200 });++assert.deepEqual(d.messages.map((m) => m.message), ['m5', 'm8', 'm9', 'm10', 'm11', 'm12', 'channel only'], 'only rows after the cursor; deleted and expired rows skipped');++assert.equal(d.complete, true);++assert.equal(d.latest_id, chanId);++++const none = store.delta('global', { afterId: chanId, limit: 200 });++assert.equal(none.messages.length, 0, 'nothing new → empty delta');++assert.equal(none.latest_id, chanId, 'the cursor is handed back unchanged when nothing is new');++++const wide = store.delta('global', { afterId: 0, limit: 3 });++assert.equal(wide.complete, false, 'a gap wider than the cap is reported, not silently truncated');++assert.equal(wide.messages.length, 3);++++// ── decorate gets copies, memo stays correct across writes ────────────────────────────────────++const seen = [];++store.page('global', { limit: 5, decorate: (rows) => { rows.forEach((r) => { r.touched = true; seen.push(r.id); }); return rows; } });++const again = store.page('global', { limit: 5 });++assert.ok(!again.messages.some((m) => m.touched), 'decorators work on copies; memoised rows are pristine');++db.saveChatMessage({ user_id: u, username: 'alice', message: 'm13', stream_id: null, is_global: 1, message_type: 'chat' });++const after = store.page('global', { limit: 5 });++assert.equal(after.messages[after.messages.length - 1].message, 'm13', 'a new row invalidates the memoised page at once');++++// ── channel room ──────────────────────────────────────────────────────────────────────────────++const ch = store.page(`channel:${bob}`, { limit: 10 });++assert.deepEqual(ch.messages.map((m) => m.message), ['channel only']);++const chDelta = store.delta(`channel:${bob}`, { afterId: ch.latest_id, limit: 50 });++assert.equal(chDelta.messages.length, 0);++++assert.throws(() => store.page('nope', {}), /unknown room/);++++console.log('chat-history-store: ok');
@@ -2238,31 +2238,75 @@ function _renderHistoryPayload(p) {
@@ -2238,31 +2238,75 @@ function _renderHistoryPayload(p) {
* be fetched exclusively through the main pane's hydrate, so the widget opened empty there and
* be fetched exclusively through the main pane's hydrate, so the widget opened empty there and
* stayed empty until someone typed. Fill it directly: cached copy first, then the network.
* stayed empty until someone typed. Fill it directly: cached copy first, then the network.
*/
*/
-let _fcwHydratedAt = 0, _fcwHydrating = false;
-async function hydrateWidgetOnly() {
+let _fcwHydratedAt = 0, _fcwHydrating = false, _fcwRefreshedAt = 0, _fcwSendRetries = 0;
+const _FCW_PAGE = '/chat/global/history?limit=500';
+/**
+ * Bring the widget up to date. Three shapes, picked by what the widget already holds:
+ * empty → cached copy first (instant paint), then a delta from the cache's cursor;
+ * rows + cursor → delta: only ids newer than the newest row we have (cheap PK scan);
+ * gap too large → the server says `complete:false`: wipe and take a fresh page.
+ * Runs on every open, on (re)connect, on tab focus and on coming back online — the old
+ * "hydrated less than 10 minutes ago" guard is what left stale rows on screen after the SPA
+ * router tore the socket down. A short throttle collapses the triggers that fire together.
+ */
+async function hydrateWidgetOnly({ reason = 'open' } = {}) {
const container = document.getElementById('fcw-messages');
const container = document.getElementById('fcw-messages');
if (!container || _fcwHydrating) return;
if (!container || _fcwHydrating) return;
- if (container.querySelector('.chat-msg') && Date.now() - _fcwHydratedAt < 10 * 60 * 1000) return;
+ const hasRows = !!container.querySelector('.chat-msg');
+ if (hasRows && _fcwLastId && Date.now() - _fcwRefreshedAt < 1500) return;
_fcwHydrating = true;
_fcwHydrating = true;
- const feed = (payload) => {
- const rows = (payload && payload.data && payload.data.messages) || [];
- if (!rows.length) return 0;
+ const feed = (rows) => {
+ if (!Array.isArray(rows) || !rows.length) return 0;
const prev = _loadingHistory; _loadingHistory = true;
const prev = _loadingHistory; _loadingHistory = true;
try { rows.slice(-200).forEach((m) => _fcwAddMessage(m)); } finally { _loadingHistory = prev; }
try { rows.slice(-200).forEach((m) => _fcwAddMessage(m)); } finally { _loadingHistory = prev; }
_fcwScrollToBottom();
_fcwScrollToBottom();
return rows.length;
return rows.length;
};
};
+ const fresh = async () => {
+ const page = await api(_FCW_PAGE);
+ container.replaceChildren();
+ _fcwIds.clear(); _fcwLastId = 0;
+ feed(page && page.messages);
+ try { _chatCacheWrite('global', { kind: 'global', data: page }); } catch { /* */ }
+ };
try {
try {
- if (!container.querySelector('.chat-msg')) { const cached = _chatCacheRead('global'); if (cached) feed(cached); }
- try { if (typeof loadEmotes === 'function') await loadEmotes(); } catch { /* plain text is fine */ }
- const payload = await _fetchHistoryPayload(null);
- feed(payload);
- _fcwHydratedAt = Date.now();
- try { _chatCacheWrite('global', payload); } catch { /* */ }
+ if (!hasRows) {
+ const cached = _chatCacheRead('global');
+ if (cached) feed(cached.data.messages);
+ try { if (typeof loadEmotes === 'function') await loadEmotes(); } catch { /* plain text is fine */ }
+ }
+ if (_fcwLastId) {
+ const d = await api(`/chat/global/history?after_id=${_fcwLastId}&limit=200`);
+ if (!d || d.complete === false) await fresh();
+ else if (d.messages && d.messages.length) { feed(d.messages); _fcwCacheAppend(d.messages); }
+ } else {
+ await fresh();
+ }
+ _fcwHydratedAt = _fcwRefreshedAt = Date.now();
} catch (err) {
} catch (err) {
- console.warn('[chat] widget history failed:', err && err.message);
+ console.warn(`[chat] widget history (${reason}) failed:`, err && err.message);
} finally { _fcwHydrating = false; }
} finally { _fcwHydrating = false; }
}
}
+/** Delta rows extend the cached global copy so the next cold open starts from them. */
+function _fcwCacheAppend(rows) {
+ try {
+ const key = _chatCacheKey('global');
+ const raw = localStorage.getItem(key);
+ const entry = raw ? JSON.parse(raw) : { kind: 'global', data: { messages: [] } };
+ const seen = new Set((entry.data.messages || []).map((m) => String(m.id)));
+ for (const m of rows) if (m && m.id != null && !seen.has(String(m.id))) entry.data.messages.push(m);
+ if (entry.data.messages.length > _CHAT_CACHE_ROWS) entry.data.messages.splice(0, entry.data.messages.length - _CHAT_CACHE_ROWS);
+ entry.at = Date.now();
+ localStorage.setItem(key, JSON.stringify(entry));
+ } catch { /* best effort */ }
+}
+/** The widget refreshes itself whenever the page comes back: tab focus, network back, socket back. */
+function _fcwRefreshIfRelevant(reason) {
+ const container = document.getElementById('fcw-messages');
+ if (!container) return;
+ if (!_fcwOpen && !container.querySelector('.chat-msg')) return; // never opened: nothing to keep current
+ hydrateWidgetOnly({ reason }).catch(() => {});
+}
async function hydrateActiveChatHistory(streamId, { clear = false } = {}) {
async function hydrateActiveChatHistory(streamId, { clear = false } = {}) {
const chatEl = getChatEl();
const chatEl = getChatEl();
@@ -2566,6 +2610,9 @@ function initChat(streamId, channelUserId = null) {
@@ -2566,6 +2610,9 @@ function initChat(streamId, channelUserId = null) {
ws._authToken = token || null;
ws._authToken = token || null;
_chatReconnectDelay = CHAT_RECONNECT_BASE; // reset backoff on success
_chatReconnectDelay = CHAT_RECONNECT_BASE; // reset backoff on success
addRichSystemMessage('<i class="fa-solid fa-plug-circle-check" style="margin-right:5px;opacity:0.8"></i> Connected to chat');
addRichSystemMessage('<i class="fa-solid fa-plug-circle-check" style="margin-right:5px;opacity:0.8"></i> Connected to chat');
+ // A fresh socket after the router tore the last one down: the widget may hold rows from
+ // before the gap. Pull what it missed instead of waiting for someone to type.
+ _fcwRefreshIfRelevant('connect');
};
};
// Load emotes + channel sounds for this stream context
// Load emotes + channel sounds for this stream context
@@ -2671,6 +2718,13 @@ window.addEventListener('online', () => {
@@ -2671,6 +2718,13 @@ window.addEventListener('online', () => {
clearTimeout(_chatReconnectTimer); _chatReconnectTimer = null;
clearTimeout(_chatReconnectTimer); _chatReconnectTimer = null;
_reconnectChatWs(chatStreamId ?? null);
_reconnectChatWs(chatStreamId ?? null);
});
});
+// Mobile browsers freeze the page (and often the socket) in the background; on return, the widget
+// asks for what it missed rather than trusting rows that may be an hour old.
+document.addEventListener('visibilitychange', () => {
+ if (document.visibilityState !== 'visible' || !_fcwOpen) return;
+ if (!chatWs || chatWs.readyState !== WebSocket.OPEN) { try { initChat(null); } catch { /* */ } }
+ _fcwRefreshIfRelevant('focus');
+});
window.addEventListener('offline', () => { if (_chatActive && !_chatIntentionalClose) _chatConnState('offline'); });
window.addEventListener('offline', () => { if (_chatActive && !_chatIntentionalClose) _chatConnState('offline'); });
/**
/**
@@ -2714,6 +2768,7 @@ function _reconnectChatWs(streamId) {
@@ -2714,6 +2768,7 @@ function _reconnectChatWs(streamId) {
// Messages sent while we were away are only in the history. Refill it — but not under a reader
// Messages sent while we were away are only in the history. Refill it — but not under a reader
// who has scrolled back, since reloading history redraws the list.
// who has scrolled back, since reloading history redraws the list.
if (outage > 1500 && !_chatUserScrolledUp) hydrateActiveChatHistory(streamId).catch(() => {});
if (outage > 1500 && !_chatUserScrolledUp) hydrateActiveChatHistory(streamId).catch(() => {});
+ _fcwRefreshIfRelevant('reconnect');
};
};
ws.onmessage = (e) => {
ws.onmessage = (e) => {
@@ -2780,6 +2835,8 @@ function destroyChat(forceClose = false) {
@@ -2780,6 +2835,8 @@ function destroyChat(forceClose = false) {
}
}
chatStreamId = null;
chatStreamId = null;
}
}
+ // The channel scope belongs to the page that set it; a later global fetch must not inherit it.
+ chatChannelUserId = null;
chatRenderTargetId = null;
chatRenderTargetId = null;
clearFullscreenChatMessages();
clearFullscreenChatMessages();
_fullscreenChatRecent = [];
_fullscreenChatRecent = [];
@@ -7079,8 +7136,9 @@ function fcwToggle() {
@@ -7079,8 +7136,9 @@ function fcwToggle() {
if (!chatWs || chatWs.readyState !== WebSocket.OPEN) {
if (!chatWs || chatWs.readyState !== WebSocket.OPEN) {
initChat(null);
initChat(null);
}
}
- // The widget is its own chat surface on pages without a pane: give it the history.
- hydrateWidgetOnly();
+ // The widget is its own chat surface on pages without a pane: give it the history —
+ // a delta from the newest row it holds, or a fresh page when it has nothing.
+ hydrateWidgetOnly({ reason: 'open' });
// Scroll to bottom
// Scroll to bottom
const msgs = document.getElementById('fcw-messages');
const msgs = document.getElementById('fcw-messages');
@@ -7102,7 +7160,16 @@ function fcwSendChat() {
@@ -7102,7 +7160,16 @@ function fcwSendChat() {
const input = document.getElementById('fcw-chat-input');
const input = document.getElementById('fcw-chat-input');
if (!input) return;
if (!input) return;
const text = input.value.trim();
const text = input.value.trim();
- if (!text || !chatWs || chatWs.readyState !== WebSocket.OPEN) return;
+ if (!text) return;
+ if (!chatWs || chatWs.readyState !== WebSocket.OPEN) {
+ // The router closes the socket on navigation; bring it back and send once it is open.
+ _fcwSendRetries = (_fcwSendRetries || 0) + 1;
+ if (_fcwSendRetries > 4) { _fcwSendRetries = 0; return; }
+ try { initChat(null); } catch { /* */ }
+ setTimeout(() => { if (document.getElementById('fcw-chat-input')) fcwSendChat(); }, 900);
+ return;
+ }
+ _fcwSendRetries = 0;
// Save to shared message history
// Save to shared message history
if (_chatMsgHistory[0] !== text) {
if (_chatMsgHistory[0] !== text) {
@@ -7143,11 +7210,36 @@ function _chatTrimWidget(container) {
@@ -7143,11 +7210,36 @@ function _chatTrimWidget(container) {
while (container.children.length > 200) container.removeChild(container.firstChild);
while (container.children.length > 200) container.removeChild(container.firstChild);
}
}
+// Ids the widget has shown, kept past the 200-row DOM trim so a trimmed row cannot come back
+// through a delta, and the cursor the next delta read starts from.
+const _fcwIds = new Set();
+let _fcwLastId = 0;
+function _fcwRemember(id) {
+ _fcwIds.add(String(id));
+ const n = Number(id);
+ if (Number.isFinite(n) && n > _fcwLastId) _fcwLastId = n;
+ if (_fcwIds.size > 1500) { const it = _fcwIds.values(); for (let i = 0; i < 500; i++) _fcwIds.delete(it.next().value); }
+}
+// Rows arrive out of order when a delta lands under live rows that beat it; keep the list in id order.
+function _fcwInsertOrdered(container, el, id) {
+ const n = Number(id);
+ let ref = null;
+ if (Number.isFinite(n)) {
+ for (let node = container.lastElementChild; node; node = node.previousElementSibling) {
+ const nid = Number(node.dataset && node.dataset.msgId);
+ if (!Number.isFinite(nid) || nid <= n) break;
+ ref = node;
+ }
+ }
+ if (ref) container.insertBefore(el, ref); else container.appendChild(el);
+}
function _fcwAddMessage(msg) {
function _fcwAddMessage(msg) {
if (_fcwHistoryBuffer) { _fcwHistoryBuffer.push(msg); return; }
if (_fcwHistoryBuffer) { _fcwHistoryBuffer.push(msg); return; }
const container = document.getElementById('fcw-messages');
const container = document.getElementById('fcw-messages');
if (!container) return;
if (!container) return;
- if (msg && msg.id != null && container.querySelector(`[data-msg-id="${CSS.escape(String(msg.id))}"]`)) return;
+ const mid = msg && msg.id != null ? String(msg.id) : null;
+ if (mid && (_fcwIds.has(mid) || container.querySelector(`[data-msg-id="${CSS.escape(mid)}"]`))) return;
+ if (mid) _fcwRemember(mid);
// Stream source badge — clickable stream TITLE (live → watch, offline → channel).
// Stream source badge — clickable stream TITLE (live → watch, offline → channel).
const src = _fcwSourceLabel(msg);
const src = _fcwSourceLabel(msg);
@@ -7169,7 +7261,7 @@ function _fcwAddMessage(msg) {
@@ -7169,7 +7261,7 @@ function _fcwAddMessage(msg) {
// user menu on tap — the widget's own renderer had no menu at all).
// user menu on tap — the widget's own renderer had no menu at all).
const el = buildChatMessageEl(msg, { widget: true, prefixHtml: srcBadge });
const el = buildChatMessageEl(msg, { widget: true, prefixHtml: srcBadge });
if (_friendlyHides(msg)) _friendlyAppend(container, el);
if (_friendlyHides(msg)) _friendlyAppend(container, el);
- else container.appendChild(el);
+ else _fcwInsertOrdered(container, el, mid);
_chatTrimWidget(container);
_chatTrimWidget(container);
// One scroll per frame, however many rows arrived in it (history mirrors hundreds at once).
// One scroll per frame, however many rows arrived in it (history mirrors hundreds at once).
new file mode 100644
new file mode 100644
@@ -0,0 +1,149 @@
@@ -0,0 +1,149 @@
+/**
+ * history-store.js — the one place that answers "what happened in this chat room?".
+ *
+ * Rooms: 'global' | 'channel:<userId>' (the stream route keeps its own scope logic)
+ * Reads:
+ * 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.
+ * 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.
+ *
+ * Backing store today is SQLite through `db`. `memo` collapses identical page reads inside a
+ * short window (a burst of reconnects after a deploy used to run the five-table global join
+ * once per socket). Swapping in a ring buffer or Redis later means replacing `readPage` /
+ * `readDelta` and calling `invalidate(room)` from the write path — nothing in the routes changes.
+ */
+const db = require('../db/database');
+
+const PAGE_MEMO_MS = 2000;
+const MAX_LIMIT = 500;
+
+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,
+ s.is_live AS source_is_live,
+ s.managed_stream_id AS source_managed_id,
+ COALESCE(ms.title, s.title) AS source_stream_title,
+ ms.slug AS source_slug
+ FROM chat_messages cm
+ LEFT JOIN users u ON cm.user_id = u.id
+ LEFT JOIN streams s ON cm.stream_id = s.id
+ LEFT JOIN users su ON s.user_id = su.id
+ LEFT JOIN users cu ON cm.channel_user_id = cu.id
+ LEFT JOIN 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')
+ 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,
+ u.username AS core_username,
+ s.title AS source_stream_title, s.managed_stream_id AS source_managed_id,
+ s.is_live AS source_is_live, ms.slug AS source_slug,
+ COALESCE(bu.username, cu.username) AS source_channel
+ FROM chat_messages cm
+ LEFT JOIN users u ON cm.user_id = u.id
+ LEFT JOIN streams s ON cm.stream_id = s.id
+ LEFT JOIN managed_streams ms ON s.managed_stream_id = ms.id
+ LEFT JOIN users bu ON s.user_id = bu.id
+ LEFT JOIN users cu ON cm.channel_user_id = cu.id
+ WHERE cm.channel_user_id = ? AND cm.is_deleted = 0
+ AND (cm.auto_delete_at IS NULL OR datetime(cm.auto_delete_at) > CURRENT_TIMESTAMP)`;
+
+function clampLimit(v, dflt = 500) {
+ const n = parseInt(v, 10);
+ return Math.min(Number.isFinite(n) && n > 0 ? n : dflt, MAX_LIMIT);
+}
+
+function parseRoom(room) {
+ if (room === 'global') return { kind: 'global' };
+ const m = /^channel:(\d+)$/.exec(String(room || ''));
+ if (m) return { kind: 'channel', userId: parseInt(m[1], 10) };
+ throw new Error(`history-store: unknown room "${room}"`);
+}
+
+function baseSql(r, { channelUsername } = {}) {
+ if (r.kind === 'global') {
+ const params = [];
+ let sql = GLOBAL_SELECT;
+ if (channelUsername) { sql += ` AND LOWER(COALESCE(su.username, cu.username)) = LOWER(?)`; params.push(channelUsername); }
+ return { sql, params };
+ }
+ return { sql: CHANNEL_SELECT, params: [r.userId] };
+}
+
+function tipId() {
+ try { const r = db.get('SELECT MAX(id) AS id FROM chat_messages'); return (r && r.id) || 0; } catch { return Date.now(); }
+}
+const latestIdOf = (rows) => rows.reduce((m, x) => (x && x.id > m ? x.id : m), 0);
+
+// ── reads ──────────────────────────────────────────────────────────────────────────────────────
+const memo = new Map(); // key → { at, value }
+function memoGet(key) {
+ const hit = memo.get(key);
+ if (hit && Date.now() - hit.at < PAGE_MEMO_MS) return hit.value;
+ if (hit) memo.delete(key);
+ return null;
+}
+function memoSet(key, value) {
+ memo.set(key, { at: Date.now(), value });
+ if (memo.size > 64) { const oldest = memo.keys().next().value; memo.delete(oldest); }
+}
+
+function readPage(r, { limit, before, channelUsername }) {
+ const { sql, params } = baseSql(r, { channelUsername });
+ let q = sql;
+ if (before) { q += ' AND cm.timestamp < ?'; params.push(before); }
+ q += ' ORDER BY cm.timestamp DESC, cm.id DESC LIMIT ?'; params.push(limit);
+ return db.all(q, params).reverse();
+}
+
+function readDelta(r, { afterId, limit, channelUsername }) {
+ const { sql, params } = baseSql(r, { channelUsername });
+ // One more than asked: if it comes back, the gap is bigger than the client can splice in.
+ const q = `${sql} AND cm.id > ? ORDER BY cm.id ASC LIMIT ?`;
+ params.push(afterId, limit + 1);
+ return db.all(q, params);
+}
+
+/**
+ * Newest rows of a room, oldest→newest.
+ * @returns {{ messages: object[], latest_id: number }}
+ */
+function page(room, { limit, before, channelUsername, decorate } = {}) {
+ const r = parseRoom(room);
+ const lim = clampLimit(limit);
+ // The newest row id is part of the memo key, so a memoised page can never miss a message:
+ // one primary-key lookup instead of the five-table join for every repeat reader in the window.
+ const key = !before ? `${room}|${lim}|${channelUsername || ''}|${tipId()}` : null;
+ let rows = key ? memoGet(key) : null;
+ if (!rows) {
+ rows = readPage(r, { limit: lim, before, channelUsername });
+ if (key) memoSet(key, rows);
+ }
+ const messages = decorate ? decorate(rows.map((x) => ({ ...x }))) : rows;
+ return { messages, latest_id: latestIdOf(rows) };
+}
+
+/**
+ * Everything after a cursor, oldest→newest. `complete:false` = more than `limit` rows were
+ * missing; take a fresh page instead of splicing.
+ * @returns {{ messages: object[], latest_id: number, complete: boolean }}
+ */
+function delta(room, { afterId, limit, channelUsername, decorate } = {}) {
+ const r = parseRoom(room);
+ const lim = clampLimit(limit, 200);
+ const after = Math.max(0, parseInt(afterId, 10) || 0);
+ let rows = readDelta(r, { afterId: after, limit: lim, channelUsername });
+ 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 };
+}
+
+/** Drop memoised pages for a room (or all rooms). Call from write paths when a tier is added. */
+function invalidate(room) {
+ if (!room) { memo.clear(); return; }
+ for (const k of memo.keys()) if (k.startsWith(`${room}|`)) memo.delete(k);
+}
+
+module.exports = { page, delta, invalidate, clampLimit, PAGE_MEMO_MS };
@@ -12,6 +12,7 @@ const express = require('express');
@@ -12,6 +12,7 @@ const express = require('express');
const db = require('../db/database');
const db = require('../db/database');
const { optionalAuth, requireAuth, requireAdmin } = require('../auth/auth');
const { optionalAuth, requireAuth, requireAdmin } = require('../auth/auth');
const permissions = require('../auth/permissions');
const permissions = require('../auth/permissions');
+const historyStore = require('./history-store');
const router = express.Router();
const router = express.Router();
@@ -466,42 +467,17 @@ router.get('/filters/friendly', (req, res) => {
@@ -466,42 +467,17 @@ router.get('/filters/friendly', (req, res) => {
router.get('/global/history', optionalAuth, (req, res) => {
router.get('/global/history', optionalAuth, (req, res) => {
try {
try {
- const limit = Math.min(parseInt(req.query.limit || '500'), 500);
+ const limit = req.query.limit;
const before = req.query.before;
const before = req.query.before;
+ const afterId = req.query.after_id;
const channelUsername = String(req.query.username || '').trim();
const channelUsername = String(req.query.username || '').trim();
-
- let sql = `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,
- s.is_live AS source_is_live,
- s.managed_stream_id AS source_managed_id,
- COALESCE(ms.title, s.title) AS source_stream_title,
- ms.slug AS source_slug
- FROM chat_messages cm
- LEFT JOIN users u ON cm.user_id = u.id
- LEFT JOIN streams s ON cm.stream_id = s.id
- LEFT JOIN users su ON s.user_id = su.id
- LEFT JOIN users cu ON cm.channel_user_id = cu.id
- LEFT JOIN 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')
- AND (cm.auto_delete_at IS NULL OR datetime(cm.auto_delete_at) > CURRENT_TIMESTAMP)`;
- const params = [];
-
- if (channelUsername) {
- sql += ` AND LOWER(COALESCE(su.username, cu.username)) = LOWER(?)`;
- params.push(channelUsername);
- }
-
- if (before) {
- sql += ` AND cm.timestamp < ?`;
- params.push(before);
- }
-
- sql += ` ORDER BY cm.timestamp DESC LIMIT ?`;
- params.push(limit);
-
- const messages = enrichMessagesWithCosmetics(hydrateReplies(db.all(sql, params).reverse()));
- res.json({ messages });
+ const decorate = (rows) => enrichMessagesWithCosmetics(hydrateReplies(rows));
+ // after_id = cursor read: only what the caller has not seen (oldest→newest, PK range scan).
+ // Anything else = a page: the newest rows, optionally before a timestamp.
+ const out = afterId != null
+ ? historyStore.delta('global', { afterId, limit, channelUsername, decorate })
+ : historyStore.page('global', { limit, before, channelUsername, decorate });
+ res.json(out);
} catch (err) {
} catch (err) {
res.status(500).json({ error: 'Failed to get global chat history' });
res.status(500).json({ error: 'Failed to get global chat history' });
}
}
@@ -578,15 +554,26 @@ router.get('/:streamId/history', optionalAuth, (req, res) => {
@@ -578,15 +554,26 @@ router.get('/:streamId/history', optionalAuth, (req, res) => {
params.push(req.params.streamId);
params.push(req.params.streamId);
}
}
- if (before) {
- sql += ` AND cm.timestamp < ?`;
- params.push(before);
+ // after_id: cursor read (oldest→newest, primary-key range scan) for reopen/reconnect.
+ // Otherwise a page: the newest rows, optionally before a timestamp.
+ const afterId = req.query.after_id != null ? Math.max(0, parseInt(req.query.after_id, 10) || 0) : null;
+ if (afterId != null) {
+ sql += ` AND cm.id > ? ORDER BY cm.id ASC LIMIT ?`;
+ params.push(afterId, limit + 1);
+ } else {
+ if (before) {
+ sql += ` AND cm.timestamp < ?`;
+ params.push(before);
+ }
+ sql += ` ORDER BY cm.timestamp DESC LIMIT ?`;
+ params.push(limit);
}
}
- sql += ` ORDER BY cm.timestamp DESC LIMIT ?`;
- params.push(limit);
-
- const messages = enrichMessagesWithCosmetics(hydrateReplies(db.all(sql, params).reverse()));
+ let rows = db.all(sql, params);
+ let complete = true;
+ if (afterId != null) { complete = rows.length <= limit; rows = rows.slice(0, limit); } else rows.reverse();
+ const latest_id = rows.reduce((m, x) => (x.id > m ? x.id : m), 0) || (afterId || 0);
+ const messages = enrichMessagesWithCosmetics(hydrateReplies(rows));
// Currently-live slots for this broadcaster — lets the client render a
// Currently-live slots for this broadcaster — lets the client render a
// "hop between live streams" affordance.
// "hop between live streams" affordance.
@@ -607,7 +594,7 @@ router.get('/:streamId/history', optionalAuth, (req, res) => {
@@ -607,7 +594,7 @@ router.get('/:streamId/history', optionalAuth, (req, res) => {
} catch { /* non-critical */ }
} catch { /* non-critical */ }
}
}
res.json({
res.json({
- messages, liveSlots, channel,
+ messages, latest_id, complete, 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,
});
});
@@ -623,26 +610,11 @@ router.get('/channel/:userId/history', optionalAuth, (req, res) => {
@@ -623,26 +610,11 @@ router.get('/channel/:userId/history', optionalAuth, (req, res) => {
try {
try {
const userId = parseInt(req.params.userId);
const userId = parseInt(req.params.userId);
if (!userId) return res.status(400).json({ error: 'Bad user id' });
if (!userId) return res.status(400).json({ error: 'Bad user id' });
- const limit = Math.min(parseInt(req.query.limit || '500'), 500);
- const before = req.query.before;
- const select = `SELECT cm.*, u.avatar_url, u.profile_color, u.role, u.display_name,
- u.username AS core_username,
- s.title AS source_stream_title, s.managed_stream_id AS source_managed_id,
- s.is_live AS source_is_live, ms.slug AS source_slug,
- COALESCE(bu.username, cu.username) AS source_channel`;
- let sql = `${select}
- FROM chat_messages cm
- LEFT JOIN users u ON cm.user_id = u.id
- LEFT JOIN streams s ON cm.stream_id = s.id
- LEFT JOIN managed_streams ms ON s.managed_stream_id = ms.id
- LEFT JOIN users bu ON s.user_id = bu.id
- LEFT JOIN users cu ON cm.channel_user_id = cu.id
- WHERE cm.channel_user_id = ? AND cm.is_deleted = 0
- AND (cm.auto_delete_at IS NULL OR datetime(cm.auto_delete_at) > CURRENT_TIMESTAMP)`;
- const params = [userId];
- if (before) { sql += ' AND cm.timestamp < ?'; params.push(before); }
- sql += ' ORDER BY cm.timestamp DESC LIMIT ?'; params.push(limit);
- const messages = enrichMessagesWithCosmetics(hydrateReplies(db.all(sql, params).reverse()));
+ const decorate = (rows) => enrichMessagesWithCosmetics(hydrateReplies(rows));
+ const room = `channel:${userId}`;
+ const out = req.query.after_id != null
+ ? historyStore.delta(room, { afterId: req.query.after_id, limit: req.query.limit, decorate })
+ : historyStore.page(room, { limit: req.query.limit, before: req.query.before, decorate });
let liveSlots = [], channel = null;
let liveSlots = [], channel = null;
try {
try {
const owner = db.getUserById(userId);
const owner = db.getUserById(userId);
@@ -651,7 +623,7 @@ router.get('/channel/:userId/history', optionalAuth, (req, res) => {
@@ -651,7 +623,7 @@ router.get('/channel/:userId/history', optionalAuth, (req, res) => {
.filter(ms => ms.is_currently_live)
.filter(ms => ms.is_currently_live)
.map(ms => ({ managed_stream_id: ms.id, slug: ms.slug, title: ms.title, live_session_id: ms.live_session_id }));
.map(ms => ({ managed_stream_id: ms.id, slug: ms.slug, title: ms.title, live_session_id: ms.live_session_id }));
} catch { /* non-critical */ }
} catch { /* non-critical */ }
- res.json({ messages, liveSlots, channel, activeStreamId: null, activeManagedId: null });
+ res.json({ ...out, liveSlots, channel, activeStreamId: null, activeManagedId: null });
} catch (err) {
} catch (err) {
res.status(500).json({ error: 'Failed to get channel chat history' });
res.status(500).json({ error: 'Failed to get channel chat history' });
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,67 @@
@@ -0,0 +1,67 @@
+'use strict';
+/**
+ * Chat history reads through one store: a page is the newest rows oldest→newest, a delta is only
+ * what a client with a cursor has not seen, and a gap wider than the cap says so instead of
+ * silently handing back a slice the client would splice in the wrong place.
+ */
+const assert = require('assert');
+const os = require('os');
+const path = require('path');
+
+process.env.DB_PATH = path.join(os.tmpdir(), `ov-chat-history-${Date.now()}.db`);
+const db = require('../server/db/database');
+db.initDb();
+const store = require('../server/chat/history-store');
+
+const u = db.run(`INSERT INTO users (username, password_hash, email) VALUES ('alice', 'x', '[email protected]')`).lastInsertRowid;
+const bob = db.run(`INSERT INTO users (username, password_hash, email) VALUES ('bob', 'x', '[email protected]')`).lastInsertRowid;
+
+const ids = [];
+for (let i = 1; i <= 12; i++) {
+ const r = db.saveChatMessage({ user_id: u, username: 'alice', message: `m${i}`, stream_id: null, is_global: 1, message_type: 'chat' });
+ ids.push(Number(r.lastInsertRowid));
+}
+// One deleted and one auto-expired row must never come back through either read.
+db.run(`UPDATE chat_messages SET is_deleted = 1 WHERE id = ?`, [ids[5]]);
+db.run(`UPDATE chat_messages SET auto_delete_at = datetime('now', '-1 minute') WHERE id = ?`, [ids[6]]);
+// The global feed is every public room with a source badge, so a channel row belongs in it too.
+const chanId = Number(db.saveChatMessage({ user_id: bob, username: 'bob', message: 'channel only', stream_id: null, channel_user_id: bob, message_type: 'chat' }).lastInsertRowid);
+
+// ── page ──────────────────────────────────────────────────────────────────────────────────────
+const page = store.page('global', { limit: 5 });
+assert.equal(page.messages.length, 5, 'a page is capped by limit');
+assert.deepEqual(page.messages.map((m) => m.message), ['m9', 'm10', 'm11', 'm12', 'channel only'], 'newest rows, oldest → newest, all rooms');
+assert.equal(page.latest_id, chanId, 'latest_id is the cursor of the newest row');
+
+// ── delta ─────────────────────────────────────────────────────────────────────────────────────
+const d = store.delta('global', { afterId: ids[3], limit: 200 });
+assert.deepEqual(d.messages.map((m) => m.message), ['m5', 'm8', 'm9', 'm10', 'm11', 'm12', 'channel only'], 'only rows after the cursor; deleted and expired rows skipped');
+assert.equal(d.complete, true);
+assert.equal(d.latest_id, chanId);
+
+const none = store.delta('global', { afterId: chanId, limit: 200 });
+assert.equal(none.messages.length, 0, 'nothing new → empty delta');
+assert.equal(none.latest_id, chanId, 'the cursor is handed back unchanged when nothing is new');
+
+const wide = store.delta('global', { afterId: 0, limit: 3 });
+assert.equal(wide.complete, false, 'a gap wider than the cap is reported, not silently truncated');
+assert.equal(wide.messages.length, 3);
+
+// ── decorate gets copies, memo stays correct across writes ────────────────────────────────────
+const seen = [];
+store.page('global', { limit: 5, decorate: (rows) => { rows.forEach((r) => { r.touched = true; seen.push(r.id); }); return rows; } });
+const again = store.page('global', { limit: 5 });
+assert.ok(!again.messages.some((m) => m.touched), 'decorators work on copies; memoised rows are pristine');
+db.saveChatMessage({ user_id: u, username: 'alice', message: 'm13', stream_id: null, is_global: 1, message_type: 'chat' });
+const after = store.page('global', { limit: 5 });
+assert.equal(after.messages[after.messages.length - 1].message, 'm13', 'a new row invalidates the memoised page at once');
+
+// ── channel room ──────────────────────────────────────────────────────────────────────────────
+const ch = store.page(`channel:${bob}`, { limit: 10 });
+assert.deepEqual(ch.messages.map((m) => m.message), ['channel only']);
+const chDelta = store.delta(`channel:${bob}`, { afterId: ch.latest_id, limit: 50 });
+assert.equal(chDelta.messages.length, 0);
+
+assert.throws(() => store.page('nope', {}), /unknown room/);
+
+console.log('chat-history-store: ok');