openvibers/openvibe.live Commits

Read chat stats, history, queues and sounds from OpenVibe.Chat in chat mode (the read mirror stays until every reader is repointed) (#28)

OP OpenVibers committed 5 Oct 2026 · cd8128337f · 21 files, +1131 −262

* Read chat stats, history, queues and sounds from OpenVibe.Chat in chat mode; delete the read mirror and its receiver (T3)

## Where the task's line map differed
- **`database.js:3259/3271/3286` and `4567/4605/4612/4630/4712/4716`** are synchronous `db` helpers (`getLiveChatBuckets`, `getRecentChatText`, `getChatSpikeOffsets`, `searchChatMessages`, `getUserChatHistory`, `getChatSamplesInChannel`, `getRelayUserChatHistory`). I could not make them read Chat in place without turning them async (which `authorization.test.js` and local mode call synchronously), so they stay as the **local / rollback implementation** and their **callers** now read Chat. `database.js:3958` (home-stats `chatMessages`) and `1474` (relay-users backfill) were rewritten/deleted in place.
- **`ai/context/index.js:224`** was `viewerCountOf` (presence, not a mirror read); the real mirror reader is `chatDelta` at **:171** (`db.getChannelChatSince`) — repointed.
- **`deploy-notice.js:71`** already runs only when `CHAT_AUTHORITY≠chat` (`announce()` returns before `persist()` in Chat mode), so no repoint was needed.
- Beyond the enumerated list I also repointed the reads the new read-methods clearly serve: `getChatMessageById` (delete-message), `getTtsVoiceOverride`, `getHiddenRelayUsers`, pending-IP, and channel-sounds. Chat's `/messages` has no free-text filter, so admin text search degrades to an empty page rather than lie (noted in code).

## Changes (file:line)
- `server/chat/chat-reads.js` — **new**: mode-aware read layer (Chat when `CHAT_AUTHORITY=chat`, Live's own tables otherwise; 15 s cache, sync `peek*`, `null`/cached degrade, off under `LIVE_DRILL`).
- `server/chat/chat-client.js:126-196` — added `readStats/readMessages/readTimeline/readPendingIp/readRelayUsers/readRelayUser/readTtsOverride/readSounds` beside `presence()` + `soundAsset()` write (no idempotency key — Chat's `/sounds/asset` rejects extra fields).
- `server/chat/live-context-routes.js:260,638` — deleted `POST /mirror` receiver, guard special-case, `MIRROR_TABLES`/`applyMirror`/`liveColumns`, their exports.
- `server/chat/chat-delivery.js:94` — deleted `MIRROR_WRITES`/`mirror()` and their export.
- `server/chat/history-store.js` — deleted (dead).
- `server/db/database.js:1454,3944` — dropped the `relay_users` backfill that read `chat_messages`; home-stats `chatMessages` now `siteStatsPeek()`.
- `server/admin/routes.js:81` — admin stats `chat.totalMessages` from `chatReads.siteStats()`.
- `server/admin/mod-routes.js:340,354,371,562,590,611,646,921` — search/history/message-by-id/relay-user/hidden-relays/pending-IP/tts reads through `chatReads`; all nine `delivery.mirror(...)` calls removed.
- `server/ai/auto-clip-job.js:72,110` — `recentChatText`/`liveChatBuckets`.
- `server/ai/ai-moments-job.js:140` — `spikeOffsets` (offset base = `streams.started_at`).
- `server/ai/context/index.js:171` — `channelMessagesPeek`.
- `server/ai/slogan-job.js:84` — `topChatters` (site-wide 14-day, banned filtered locally).
- `server/ai/viewers/routes.js:267,268,282,283` — `userHistory`/`relayHistory`/`channelSamples`.
- `server/chat/tts-engine.js:214` — `ttsOverridePeek`.
- `server/media-proxy/asset-sync.js:70,80` — `pendingSounds` + `recordSoundAsset` (`/sounds/asset`).
- `server/streaming/routes.js:1635` — setup-progress sound count via `chatReads.soundCount`.
- `server/integrations/chat-relay-service.js:506,514`, `server/integrations/robotstreamer-service.js:669` — `mirror()` calls removed (relay-record still moderated; the `ext:` welcome stays a direct local `recordFirstChat` since `stream_first_chats` is kept until J4c).
- `server/cha

* Keep the read mirror for now; chat reads serve the last good value while refreshing, never cache failures, answer moderators 503 when Chat is down, never cache moderation queues, and fix relay hiding, relay history, totals and sound paging (review)

Everything is green and the tree matches the intent. Final summary:

**Merge:** `git merge origin/main` — already up to date (branch was 1 commit ahead). Working tree unchanged beyond the fixes below; nothing committed or pushed.

**Mirror restored (per scope change):**
- `POST /internal/chat-effects/mirror` with its `live.chat_mirror.write` guard, `MIRROR_TABLES`/`applyMirror`/`liveColumns` and exports restored verbatim — `server/chat/live-context-routes.js:261,648-710`; `server/chat/chat-delivery.js:92-111` is byte-identical to origin/main; all 14 `delivery.mirror(...)` call sites restored in `server/admin/mod-routes.js:402,461-463,536,585,671,727,763,968,986`, `chat-relay-service.js:507,517`, `robotstreamer-service.js:671`; the `relay_users` backfill restored at `server/db/database.js:1461-1478`. README/STATUS keep `live.chat_mirror.write` (untouched). `history-store.js` stays deleted — on origin/main only its own test used it. `docs/` left as-is.

**Findings:**
1. **Fixed** — new sync `chatReads.isRelayUserHidden()` (`server/chat/chat-reads.js:376`), called from both relay paths (`chat-relay-service.js:396`, `robotstreamer-service.js:664`). **Chosen semantics: fail OPEN.** Live's own table (mirror-fresh) answers first and is decisive, so a hide takes effect on the next message; in chat mode Chat's hidden list (a ≤15 s-cached, background-refreshed peek of `relay-users`) is also consulted; Chat unreachable or not yet cached falls back to Live's own table — a Chat outage never drops un-hides or hangs the relay path on HTTP.
2. **Fixed** — cache now separates freshness (`CACHE_TTL_MS` 15 s) from retention (`CACHE_KEEP_MS` 15 min): `peek` returns the last good value and refreshes in the background (`chat-reads.js:62,116`), covering `siteStatsPeek`, `ttsOverridePeek` and `channelMessagesPeek`.
3. **Fixed** — a failure never stores null over a good value (`rememberFailure`, `chat-reads.js:73`); it answers the last good value, else null, and sets a 2 s no-retry backoff (`FAIL_TTL_MS`, `chat-reads.js:31,63`) so a hot reader can't retry per call.
4. **Fixed** — the moderation reads throw `Chat unavailable` (`.unavailable`) instead of degrading, and all eight console routes answer `503 {error:'Chat unavailable'}` via `chatReadError` (`server/admin/mod-routes.js:41`, catches at 350/364/433/612/625/657/713/975); `|| []` sentinels removed.
5. **Fixed** — `ttl: 0` (never cached) for `pendingIp`, `relayUsers`, `relayUser`, `ttsOverride` and `messageById` (`chat-reads.js:203,339,350,364,402`), so approve/deny/review/hide/unhide/TTS writes show on the next read.
6. **Fixed** — a text-only search throws `status 501` with a clear message, surfaced by the route; username search is documented (and now verified) as EXACT (`chat-reads.js:166-193`, message names the alternatives).
7. **Fixed** — relay lookups use the stored prefixed username via `RELAY_LABELS`/`relayUsername` (`chat-reads.js:244-250`) in `relayHistory` and the relay branch of `channelSamples` (`chat-reads.js:251-263`).
8. **Fixed** — `userHistory` total is now a page lower bound (`rows.length + offset`), not `max_id` (`chat-reads.js:238`).
10. **Fixed** — `pendingSounds` is never cached (`chat-reads.js:448`) and asset-sync pages by `afterId` until a short page (`server/media-proxy/asset-sync.js:69-93`), so fileless sounds can no longer starve later ones and a just-uploaded sound is not re-uploaded.
- *Deferred as instructed:* finding 9's remaining readers, finding 11 (Chat's sender), findin

* Chat reads: moderator reads always reach Chat, ttl:0 never joins an older read, failures stay bounded, TTS override falls back to Live's table, hide/unhide invalidate at once, a cheap local stats fallback, failed sound records retried, case-insensitive q (second review)

All eight findings fixed; the read mirror is untouched. Changes are uncommitted, as instructed.

1. **Backoff blocked a moderator retry** — `load()` now applies the cooling short-circuit only when `ttl > 0 && !strict`, so a live/strict read always reaches Chat. `server/chat/chat-reads.js:102`.
2. **`ttl:0` joined a pre-write in-flight read** — `inflight` is now only checked and registered when `ttl > 0`. `server/chat/chat-reads.js:108,124`.
3. **Cache cap bypassed on failure** — extracted `evict()` and routed `rememberFailure()` through it, so failures evict like good values. `server/chat/chat-reads.js:68,79`.
4. **TTS override regressed in chat mode** — `ttsOverridePeek()` falls back to `local().getTtsVoiceOverride(k)` on a cache miss/Chat failure in remote mode. `server/chat/chat-reads.js:437`.
5. **Rollback path slower** — `siteStatsPeek()`'s local fallback is a new messages-only `COUNT(*)`; `siteStats()` keeps the distinct-chatter count. `server/chat/chat-reads.js:158,163`.
6. **Failed sound record counted as synced** — a null `recordSoundAsset` answer now increments `failed`, not `synced`. `server/media-proxy/asset-sync.js:89`.
7. **Relay hide/unhide lagged up to 15 s** — new exported `invalidate(prefix)`; `invalidateRelayHidden()` clears `hru:<channel>` (or all `hru:*` for a site-wide row) after relay-hide/relay-unhide on both the identity and `:id` routes. `server/chat/chat-reads.js:500`, `server/admin/mod-routes.js:52,549,570,598`.
8. **Case-sensitive `q` post-filter** — lowered both sides, and documented the newest-page limitation in the `searchMessages` JSDoc. `server/chat/chat-reads.js:217`.

Tests added to `test/chat-context.test.js` (one per finding: blip retry, concurrent ttl:0 count, bounded failures via `_cacheMax`/`_size`, outage peek fallback, local messages-only peek, failed-record logging, cached-list invalidation on hide/unhide/site-wide, case-insensitive `q`), plus the pre-existing cold-peek assertion updated because it now correctly answers Live's table.

Results: `ov test --no-cache test/chat-context.test.js test/chat-ingress.test.js test/drill-mode.test.js test/home-stats-series.test.js` → **4 passed in 7 s** (chat-context 1.9 s, chat-ingress 1.8 s, drill-mode 4.0 s, home-stats-series 0.8 s).

Changes

ddiff --git a/docs/chat-system.md b/docs/chat-system.md
iindex 0ccf3a92..c753faf0 100644
---- a/docs/chat-system.md
++++ b/docs/chat-system.md
@@@ -20,7 +20,7 @@ with the same WebSocket protocol and REST paths. `CHAT_AUTHORITY=chat` switches
cards, `/api/mod` deletes and moderation, cache hints) go through one seam,
[server/chat/chat-delivery.js](../server/chat/chat-delivery.js), to Chat's typed service-token ingress
([server/chat/chat-client.js](../server/chat/chat-client.js): `POST /internal/chat/messages|events|moderation|invalidate`,
-- `GET /internal/chat/presence`). Chat persists, broadcasts, mirrors and speaks (`tts`) in one call per
++ `GET /internal/chat/presence`). Chat persists, broadcasts and speaks (`tts`) in one call per
operation; each carries one idempotency key reused on every retry. A Chat 4xx is logged and dropped; a 5xx
or timeout is retried past Chat's five-minute delivery lease, then logged and dropped; neither is thrown.
- `require('./chat/chat-server')` returns a `RemoteChatServer` that never listens: synchronous reads
@@@ -39,11 +39,16 @@ with the same WebSocket protocol and REST paths. `CHAT_AUTHORITY=chat` switches
refuses while any `op = 'db'` row remains.
- Chat reads Live data and asks for side effects on `/internal/chat-context/*` and
`/internal/chat-effects/*` (`server/chat/live-context-routes.js`), with Network service tokens
-- (`live.chat_context.read`, `live.chat_effects.write`, `live.chat_mirror.write`,
++ (`live.chat_context.read`, `live.chat_effects.write`,
`server/net/service-guard.js`). Effects re-check the acting moderator/owner/admin and are refused
unless `CHAT_AUTHORITY=chat`.
--- Live's chat tables become Chat's read mirror (`POST /internal/chat-effects/mirror`): same ids, so home
-- stats, recaps, AI context, VOD chat replay and the `/api/mod` queues keep reading them in place.
++- Live reads chat stats, queues and history from Chat's internal read API —
++ `POST /internal/chat/stats`, `GET /internal/chat/messages`, `/timeline`,
++ `/moderation/pending-ip`, `/moderation/relay-users`, `/moderation/tts-override`, `/sounds` — through
++ [server/chat/chat-reads.js](../server/chat/chat-reads.js), in place of the read mirror
++ (`POST /internal/chat-effects/mirror`) home stats, recaps, AI context, VOD chat replay and the
++ `/api/mod` queues used to read. A Chat outage answers a cached or empty value, never a 500. Live's
++ chat tables stay until T3 J4c drops them.
- **The six chat tables are Chat's** (roadmap T3): `channel_moderators`, `channel_moderation_settings`,
`emotes`, `user_tags`, `chat_ai_summaries` and `chat_timeline_events`. Live keeps no copy and reads them
only through [server/chat/moderation-client.js](../server/chat/moderation-client.js) — Chat's internal
ddiff --git a/server/admin/mod-routes.js b/server/admin/mod-routes.js
iindex 3e22bdf2..17dd687c 100644
---- a/server/admin/mod-routes.js
++++ b/server/admin/mod-routes.js
@@@ -32,10 +32,27 @@ const { requireAuth } = require('../auth/auth');
const permissions = require('../auth/permissions');
const chatServer = require('../chat/chat-server');
const delivery = require('../chat/chat-delivery');
++const chatReads = require('../chat/chat-reads');
const ipUtils = require('./ip-utils');
const router = express.Router();
++/** A read the console acts on failed against Chat: answer 503 (or the read's own status), never empty data. */
++function chatReadError(res, err) {
++ if (err && err.status) { res.status(err.status).json({ error: err.message }); return true; }
++ if (err && err.unavailable) { res.status(503).json({ error: 'Chat unavailable' }); return true; }
++ return false;
++}
++
++/**
++ * A relay hide or unhide just changed Chat's hidden-user list: drop the cached per-channel list so
++ * the next relay line sees it at once instead of waiting out the peek's TTL. A site-wide row
++ * (channel_id null) changes every channel's list, so all `hru:` answers are dropped.
++ */
++function invalidateRelayHidden(channelId) {
++ chatReads.invalidate(channelId ? `hru:${Number(channelId)}` : 'hru:');
++}
++
// All mod routes require auth (individual routes check specific permissions)
router.use(requireAuth);
@@@ -313,7 +330,7 @@ router.delete('/ban/:id', async (req, res) => {
// Unscoped: searches every channel's chat, by user or by text. Every other route in this file
// checks a permission in its body; these two had none, so requireAuth alone let any signed-in
// account read anyone's chat history site-wide. Their only caller is the staff console.
--router.get('/chat/search', permissions.requireGlobalMod, (req, res) => {
++router.get('/chat/search', permissions.requireGlobalMod, async (req, res) => {
try {
const limit = Math.min(parseInt(req.query.limit || '50'), 200);
const offset = parseInt(req.query.offset || '0');
@@@ -336,23 +353,25 @@ router.get('/chat/search', permissions.requireGlobalMod, (req, res) => {
}
}
-- const result = db.searchChatMessages({ query, userId, anonId, username, streamId, limit, offset });
-- res.json(result);
++ const result = await chatReads.searchMessages({ query, userId, anonId, username, streamId, limit, offset });
++ res.json(result || { messages: [], total: 0 });
} catch (err) {
++ if (chatReadError(res, err)) return;
res.status(500).json({ error: 'Search failed' });
}
});
// ── View a user's chat history ───────────────────────────────
--router.get('/chat/user/:userId', permissions.requireGlobalMod, (req, res) => {
++router.get('/chat/user/:userId', permissions.requireGlobalMod, async (req, res) => {
try {
const userId = parseInt(req.params.userId);
const limit = Math.min(parseInt(req.query.limit || '50'), 200);
const offset = parseInt(req.query.offset || '0');
-- const result = db.getUserChatHistory(userId, limit, offset);
-- res.json(result);
++ const result = await chatReads.userHistory(userId, { limit, offset });
++ res.json(result || { messages: [], total: 0 });
} catch (err) {
++ if (chatReadError(res, err)) return;
res.status(500).json({ error: 'Failed to get chat history' });
}
});
@@@ -367,7 +386,7 @@ router.post('/delete-message', async (req, res) => {
const { message_id, stream_id } = req.body;
if (!message_id) return res.status(400).json({ error: 'message_id required' });
-- const message = db.getChatMessageById(parseInt(message_id));
++ const message = await chatReads.messageById(parseInt(message_id));
if (!message) return res.status(404).json({ error: 'Message not found' });
// Permission: stream mod can delete messages in their stream, global mod can delete anything.
@@@ -386,7 +405,8 @@ router.post('/delete-message', async (req, res) => {
if (!allowed) return res.status(403).json({ error: 'You cannot moderate this stream' });
}
-- // Chat ingress: Chat deletes the row and broadcasts the delete to every surface it reached.
++ // Chat ingress: Chat deletes the row and broadcasts the delete to every surface it reached;
++ // the mirror keeps Live's own copy in step for its remaining readers.
if (delivery.ingress()) {
delivery.mirror('deleteChatMessage', parseInt(message_id), req.user.id);
await delivery.moderate('delete-message', { id: parseInt(message_id), deleted_by: req.user.id }, { key: `delete:${message_id}` });
@@@ -416,6 +436,7 @@ router.post('/delete-message', async (req, res) => {
res.json({ message: 'Message deleted', ids: [parseInt(message_id)] });
} catch (err) {
++ if (chatReadError(res, err)) return;
console.error('[Mod] Delete message error:', err.message);
res.status(500).json({ error: 'Failed to delete message' });
}
@@@ -449,7 +470,8 @@ router.post('/delete-user-messages', async (req, res) => {
if (user_id) delivery.mirror('deleteUserChatMessages', parseInt(user_id), opts);
else if (anon_id) delivery.mirror('deleteAnonChatMessages', anon_id, opts);
else delivery.mirror('deleteRelayUserMessages', relay_username, opts);
-- const r = await delivery.moderate(action, { ...subject, stream_id: scopedStreamId ? parseInt(scopedStreamId) : undefined, deleted_by: req.user.id });
++ const scoped = { stream_id: scopedStreamId ? parseInt(scopedStreamId) : undefined, deleted_by: req.user.id };
++ const r = await delivery.moderate(action, { ...subject, ...scoped });
ids = (r && Array.isArray(r.ids)) ? r.ids : [];
} else if (user_id) {
ids = db.deleteUserChatMessages(parseInt(user_id), { streamId: scopedStreamId, deletedBy: req.user.id });
@@@ -524,6 +546,7 @@ router.post('/relay-user/hide', async (req, res) => {
await delivery.moderate('relay-unhide', { channel_id: channel_id || undefined, platform, external_username });
}
else db.unhideRelayUserByIdentity(channel_id || null, platform, external_username);
++ invalidateRelayHidden(channel_id || null);
delivery.logModeration({
scope_type: channel_id ? 'channel' : 'site',
scope_id: channel_id || undefined,
@@@ -544,6 +567,7 @@ router.post('/relay-user/hide', async (req, res) => {
reason,
createdBy: req.user.id,
});
++ invalidateRelayHidden(channel_id || null);
delivery.logModeration({
scope_type: channel_id ? 'channel' : 'site',
@@@ -563,7 +587,7 @@ router.post('/relay-user/hide', async (req, res) => {
// ── Unhide a relayed user ────────────────────────────────────
router.delete('/relay-user/:id', async (req, res) => {
try {
-- const row = db.get('SELECT id, channel_id FROM hidden_relay_users WHERE id = ?', [parseInt(req.params.id)]);
++ const row = await chatReads.relayUser(parseInt(req.params.id));
if (!row) return res.status(404).json({ error: 'Not found' });
// Site rows are staff's; a channel row belongs to that channel's moderators. This had no check.
const allowed = permissions.isGlobalModOrAbove(req.user)
@@@ -571,6 +595,7 @@ router.delete('/relay-user/:id', async (req, res) => {
if (!allowed) return res.status(403).json({ error: 'You cannot moderate this channel' });
if (delivery.ingress()) { delivery.mirror('unhideRelayUser', row.id); await delivery.moderate('relay-unhide', { id: row.id }); }
else db.unhideRelayUser(row.id);
++ invalidateRelayHidden(row.channel_id);
delivery.logModeration({
scope_type: 'site',
@@@ -581,6 +606,7 @@ router.delete('/relay-user/:id', async (req, res) => {
res.json({ message: 'Relay user unhidden' });
} catch (err) {
++ if (chatReadError(res, err)) return;
console.error('[Mod] Relay user unhide error:', err.message);
res.status(500).json({ error: 'Failed to unhide relay user' });
}
@@@ -591,9 +617,10 @@ router.get('/relay-users/hidden/:channelId', async (req, res) => {
try {
const channelId = parseInt(req.params.channelId);
if (!(await permissions.canModerateChannel(req.user, channelId))) return res.status(403).json({ error: 'Access denied' });
-- const hidden = db.getHiddenRelayUsers(channelId);
++ const hidden = await chatReads.relayUsers(channelId);
res.json({ hidden });
} catch (err) {
++ if (chatReadError(res, err)) return;
console.error('[Mod] List hidden relay users error:', err.message);
res.status(500).json({ error: 'Failed to list hidden relay users' });
}
@@@ -612,7 +639,7 @@ router.get('/ip-approval/:channelId/pending', async (req, res) => {
if (!(await permissions.canModerateChannel(req.user, channelId))) {
return res.status(403).json({ error: 'Access denied' });
}
-- const pending = db.getPendingIpMessages(channelId);
++ const pending = await chatReads.pendingIp(channelId);
// Add geo data
const enriched = pending.map(msg => {
@@@ -625,6 +652,7 @@ router.get('/ip-approval/:channelId/pending', async (req, res) => {
res.json({ pending: enriched });
} catch (err) {
++ if (chatReadError(res, err)) return;
console.error('[Mod] IP approval pending error:', err.message);
res.status(500).json({ error: 'Failed to get pending messages' });
}
@@@ -643,11 +671,12 @@ router.post('/ip-approval/:channelId/approve', async (req, res) => {
return res.status(403).json({ error: 'Access denied' });
}
-- // Get pending messages that will be approved (to broadcast them as real chat)
-- const pendingMsgs = db.all(
-- "SELECT * FROM pending_ip_messages WHERE channel_id = ? AND ip_address = ? AND status = 'pending'",
-- [channelId, ip]
-- );
++ // The held rows this IP is about — from Chat when Chat is the authority (Live holds no
++ // copy), else Live's own queue. In Chat mode they are not re-broadcast here: Chat releases
++ // them and broadcasts; Live only needs the count for its answer and log.
++ const pendingMsgs = delivery.ingress()
++ ? (await chatReads.pendingIp(channelId, { limit: 500 })).filter((m) => m.ip_address === ip)
++ : db.all("SELECT * FROM pending_ip_messages WHERE channel_id = ? AND ip_address = ? AND status = 'pending'", [channelId, ip]);
// Chat ingress: Chat approves its held rows; re-showing them live is Chat's (it holds them).
if (delivery.ingress()) {
@@@ -687,6 +716,7 @@ router.post('/ip-approval/:channelId/approve', async (req, res) => {
res.json({ message: `IP ${ip} approved — ${pendingMsgs.length} held message(s) released`, count: pendingMsgs.length });
} catch (err) {
++ if (chatReadError(res, err)) return;
console.error('[Mod] IP approve error:', err.message);
res.status(500).json({ error: 'Failed to approve IP' });
}
@@@ -921,10 +951,10 @@ function _ttsIdentityKey(kind, id) {
}
// GET current voice (override or auto-assigned) + editing metadata.
--router.get('/tts-voice/:kind/:id', permissions.requireGlobalMod, (req, res) => {
++router.get('/tts-voice/:kind/:id', permissions.requireGlobalMod, async (req, res) => {
try {
const identityKey = _ttsIdentityKey(req.params.kind, req.params.id);
-- const override = db.getTtsVoiceOverride(identityKey);
++ const override = await chatReads.ttsOverride(identityKey);
const auto = ttsEngine.autoUserVoiceParams(identityKey);
res.json({
identityKey,
@@@ -936,6 +966,7 @@ router.get('/tts-voice/:kind/:id', permissions.requireGlobalMod, (req, res) => {
presets: TTS_VOICE_PRESETS,
});
} catch (err) {
++ if (chatReadError(res, err)) return;
console.error('[Mod] tts-voice get error:', err.message);
res.status(500).json({ error: 'Failed to load voice' });
}
ddiff --git a/server/admin/routes.js b/server/admin/routes.js
iindex f500f72d..5e1df6d7 100644
---- a/server/admin/routes.js
++++ b/server/admin/routes.js
@@@ -44,6 +44,7 @@ const db = require('../db/database');
const { requireAuth } = require('../auth/auth');
const chatServer = require('../chat/chat-server');
const delivery = require('../chat/chat-delivery');
++const chatReads = require('../chat/chat-reads');
const permissions = require('../auth/permissions');
const money = require('../monetization/money-authority');
@@@ -76,6 +77,8 @@ router.get('/stats', async (req, res) => {
try {
// VODs live in OpenVibe.Media (null when it does not answer); Live's vods table is frozen.
const vodCounts = await require('../media-proxy/lookups').vodCounts();
++ // Chat's message total comes from OpenVibe.Chat (Live holds no chat copy in Chat mode).
++ const chatStats = await chatReads.siteStats();
const stats = {
users: {
total: db.get('SELECT COUNT(*) as c FROM users').c,
@@@ -102,7 +105,7 @@ router.get('/stats', async (req, res) => {
public: vodCounts.public,
},
chat: {
-- totalMessages: db.get('SELECT COUNT(*) as c FROM chat_messages').c,
++ totalMessages: chatStats ? chatStats.messages : null,
},
};
ddiff --git a/server/ai/ai-moments-job.js b/server/ai/ai-moments-job.js
iindex 8cb133f5..ae054c4f 100644
---- a/server/ai/ai-moments-job.js
++++ b/server/ai/ai-moments-job.js
@@@ -16,6 +16,7 @@ const path = require('node:path');
const os = require('node:os');
const fs = require('node:fs');
const db = require('../db/database');
++const chatReads = require('../chat/chat-reads');
const ai = require('./ai-analysis');
const aiService = require('./ai-service');
const media = require('../media-client');
@@@ -134,7 +135,9 @@ async function _momentContext(streamId, vodId) {
const transcript = _sample(db.getStreamTranscriptSegments(streamId) || [], 60);
// Where viewers clipped: the clips live in OpenVibe.Media.
const clipTimes = await require('../media-proxy/lookups').clipStartTimes(streamId, vodId);
-- const spikes = db.getChatSpikeOffsets(streamId, 30, 8) || [];
++ // Chat's own buckets since the stream began (offsets are relative to started_at, as before).
++ const started = (() => { try { const r = db.get('SELECT started_at FROM streams WHERE id = ?', [streamId]); return r && r.started_at ? Date.parse(String(r.started_at).replace(' ', 'T') + 'Z') : 0; } catch { return 0; } })();
++ const spikes = started > 0 ? (await chatReads.spikeOffsets(streamId, 30, 8, started) || []) : (db.getChatSpikeOffsets(streamId, 30, 8) || []);
// Non-speech sounds are strong moment candidates — an explosion or a burst of
// laughter marks a highlight as reliably as anything said out loud.
let sounds = [];
ddiff --git a/server/ai/auto-clip-job.js b/server/ai/auto-clip-job.js
iindex 2f2a0c96..1f08522c 100644
---- a/server/ai/auto-clip-job.js
++++ b/server/ai/auto-clip-job.js
@@@ -16,6 +16,7 @@
* off (a much stronger chat spike is then required).
*/
const db = require('../db/database');
++const chatReads = require('../chat/chat-reads');
const aiService = require('./ai-service');
const ai = require('./ai-analysis');
const media = require('../media-client');
@@@ -68,7 +69,7 @@ function _countClipsSince(streamId, minutes) {
async function _confirmLiveMoment(stream) {
const memories = (db.getStreamMemories(stream.id) || []).filter(m => m.description).slice(-5);
const transcript = (db.getStreamTranscriptSegments(stream.id) || []).slice(-14);
-- const chat = db.getRecentChatText(stream.id, WINDOW_SEC, 40) || [];
++ const chat = await chatReads.recentChatText(stream.id, WINDOW_SEC, 40) || [];
if (!_aiOn()) return { clip: null }; // caller decides via the stricter no-AI threshold
// Sound events are a strong clip signal on their own — a burst of gunfire, an explosion or laughter is exactly
// the kind of thing viewers clip, and it is often the reason chat spiked in the first place.
@@@ -106,7 +107,7 @@ async function _checkLiveStream(stream) {
if (!rec || !rec.vodId) return;
// Detect a clear chat spike in the recent chunk.
-- const buckets = db.getLiveChatBuckets(streamId, WINDOW_SEC, BUCKET_SEC);
++ const buckets = (await chatReads.liveChatBuckets(streamId, WINDOW_SEC, BUCKET_SEC)) || [];
if (buckets.length < 3) return;
const counts = buckets.map(b => b.count);
const avg = counts.reduce((a, b) => a + b, 0) / counts.length;
ddiff --git a/server/ai/context/index.js b/server/ai/context/index.js
iindex 4543ae65..6577c9ed 100644
---- a/server/ai/context/index.js
++++ b/server/ai/context/index.js
@@@ -22,6 +22,7 @@
'use strict';
const crypto = require('crypto');
const db = require('../../db/database');
++const chatReads = require('../../chat/chat-reads');
const insightClient = require('../../chat/insight-client');
function clip(str, n) { return (str == null ? '' : String(str)).replace(/\s+/g, ' ').trim().slice(0, n); }
@@@ -165,7 +166,9 @@ function seenBlock(stream) {
function chatDelta(channelUserId, sinceId, { limit = 40, botNames = new Set() } = {}) {
let rows = [];
-- try { rows = db.getChannelChatSince(channelUserId, sinceId || 0, limit) || []; } catch { rows = []; }
++ // The last `limit` lines of the channel come from Chat (or Live's own table when Live runs
++ // chat); a synchronous peek answers the cached page, else empty, and warms for the next tick.
++ try { rows = (chatReads.channelMessagesPeek(channelUserId, limit) || []).filter((r) => Number(r.id) > (sinceId || 0)); } catch { rows = []; }
const lines = rows.map(r => {
const isBot = r.source_platform === 'ai' || botNames.has(String(r.username || '').toLowerCase());
const tag = isBot ? ' (AI viewer)' : (r.source_platform && r.source_platform !== 'ai' ? '' : '');
ddiff --git a/server/ai/slogan-job.js b/server/ai/slogan-job.js
iindex 163abc46..f1ed5637 100644
---- a/server/ai/slogan-job.js
++++ b/server/ai/slogan-job.js
@@@ -11,6 +11,7 @@
*/
'use strict';
const db = require('../db/database');
++const chatReads = require('../chat/chat-reads');
const aiService = require('./ai-service');
const ai = require('./ai-analysis');
@@@ -76,13 +77,17 @@ async function tick() {
} catch { /* */ }
// ── Active chatters (ids) → per-USER chat analysis (running jokes / personalities) ──
++ // Chat's top-chatters stat (site-wide, the same 14-day window), then Live drops banned
++ // accounts from its own user table before asking Chat for each one's insight.
let activeRows = [];
try {
-- activeRows = db.all(`
-- SELECT u.id, u.username FROM chat_messages c JOIN users u ON c.user_id = u.id
-- WHERE c.timestamp >= datetime('now','-14 days') AND COALESCE(u.is_banned,0)=0 AND COALESCE(c.is_deleted,0)=0
-- GROUP BY u.id ORDER BY COUNT(*) DESC LIMIT 12
-- `) || [];
++ const top = await chatReads.topChatters({ since: Date.now() - 14 * 86400e3, limit: 12 });
++ activeRows = (top || [])
++ .map((r) => ({ id: Number(r.user_id) || 0, username: r.username }))
++ .filter((r) => {
++ if (!r.id) return false;
++ try { const u = db.getUserById(r.id); return !!(u && !u.is_banned); } catch { return false; }
++ });
} catch { /* */ }
const usernames = activeRows.map(r => r.username).filter(Boolean);
// The prompt is OpenVibe.AI's versioned template live.hero.slogans (WS-O task 2); Live sends the
ddiff --git a/server/ai/viewers/routes.js b/server/ai/viewers/routes.js
iindex df25ac6e..dd8823b7 100644
---- a/server/ai/viewers/routes.js
++++ b/server/ai/viewers/routes.js
@@@ -13,6 +13,7 @@
'use strict';
const express = require('express');
const db = require('../../db/database');
++const chatReads = require('../../chat/chat-reads');
const { requireAuth } = require('../../auth/auth');
const { isStaff } = require('../../auth/permissions');
const budget = require('./budget');
@@@ -263,8 +264,8 @@ router.post('/clone', requireAuth, async (req, res) => {
kind: 'user', ref: String(userId),
displayName: u.display_name || u.username,
insight: wide ? await require('../../chat/insight-client').getUser(userId) : null,
-- samples: wide ? (db.getUserChatHistory(userId, 30).messages || [])
-- : db.getChatSamplesInChannel(req.user.id, { userId, limit: 30 }),
++ samples: wide ? ((await chatReads.userHistory(userId, { limit: 30 }))?.messages || [])
++ : (await chatReads.channelSamples(req.user.id, { userId, limit: 30 })) || [],
};
} else if (kind === 'relay') {
// ref = "platform:username"
@@@ -278,8 +279,8 @@ router.post('/clone', requireAuth, async (req, res) => {
kind: 'relay', ref: `${platform}:${username}`,
displayName: username,
insight: wide ? await require('../../chat/insight-client').getRelay(platform, username) : null,
-- samples: wide ? (db.getRelayUserChatHistory(platform, username, { limit: 30 }).messages || [])
-- : db.getChatSamplesInChannel(req.user.id, { relay: { platform, rawUsername: username }, limit: 30 }),
++ samples: wide ? ((await chatReads.relayHistory(platform, username, { limit: 30 }))?.messages || [])
++ : (await chatReads.channelSamples(req.user.id, { relay: { platform, rawUsername: username }, limit: 30 })) || [],
};
} else {
return res.status(400).json({ error: 'kind must be user or relay' });
ddiff --git a/server/chat/chat-authority.js b/server/chat/chat-authority.js
iindex c1d75e7e..3627f8cc 100644
---- a/server/chat/chat-authority.js
++++ b/server/chat/chat-authority.js
@@@ -8,9 +8,10 @@
* - delivers every chat call its other modules make through Chat's typed ingress
* (server/chat/chat-delivery.js → chat-client.js); require('./chat/chat-server') returns a
* RemoteChatServer that reads Chat's presence and never listens;
-- * - keeps its chat tables as a read mirror that Chat writes (POST /internal/chat-effects/mirror)
-- * so its own readers (home stats, recaps, AI context, /api/mod queues) keep working and a
-- * rollback loses nothing;
++ * - reads chat stats, queues and history from Chat's internal read API
++ * (server/chat/chat-reads.js → chat-client.js), and still keeps its chat tables as a read
++ * mirror Chat writes (POST /internal/chat-effects/mirror) so the readers that have not moved
++ * yet keep working and a rollback loses nothing;
* - answers Chat's reads and side effects on /internal/chat-context/* and /internal/chat-effects/*.
* Anything else (default): Live runs chat itself, exactly as before.
*/
ddiff --git a/server/chat/chat-client.js b/server/chat/chat-client.js
iindex 662c453f..759dfed2 100644
---- a/server/chat/chat-client.js
++++ b/server/chat/chat-client.js
@@@ -7,6 +7,18 @@
* moderation(body) POST /internal/chat/moderation chat.moderation.write deletes, IP reviews, relay hides, log, disconnect
* invalidate(body) POST /internal/chat/invalidate chat.cache.invalidate a cache hint (user, approvals, bans)
* presence() GET /internal/chat/presence chat.presence.read the connection snapshot
++ * soundAsset(body) POST /internal/chat/sounds/asset chat.sounds.write record where Live's asset sync put a channel sound
++ *
++ * Reads (Read Live's chat stats, queues and history from Chat — server/chat/chat-reads.js is the mode-aware
++ * layer the callers use; these are the raw asks):
++ * readStats(body) POST /internal/chat/stats chat.stats.read counts, or top chatters
++ * readMessages(q) GET /internal/chat/messages chat.messages.read a page of messages
++ * readTimeline(q) GET /internal/chat/timeline chat.analysis.read time-bucketed message counts
++ * readPendingIp(q) GET /internal/chat/moderation/pending-ip chat.moderation.queue.read held IP-approval messages
++ * readRelayUsers(q) GET /internal/chat/moderation/relay-users chat.moderation.queue.read hidden relay users
++ * readRelayUser(id) GET /internal/chat/moderation/relay-users/:id chat.moderation.queue.read one hidden relay user
++ * readTtsOverride(q) GET /internal/chat/moderation/tts-override chat.moderation.queue.read a voice override
++ * readSounds(q) GET /internal/chat/sounds chat.sounds.read a channel's sound count, or its pending assets
*
* Auth: Live's Network service principal (client_credentials, audience openvibe.chat, server/net/network-principal.js).
* Every POST carries one idempotency `key` made when the operation is created and reused on every retry, so Chat
@@@ -70,7 +82,7 @@ async function request(method, path, body, retried = false) {
const sleep = (ms) => new Promise((resolve) => { const t = setTimeout(resolve, ms); if (t.unref) t.unref(); });
/** POST one operation to /internal/chat/<family>, retrying 5xx/unreachable with the same key. Never rejects. */
--async function send(family, body) {
++async function send(family, body, path = `/internal/chat/${family}`) {
if (drill()) return null;
if (!principal.configured()) { count(family, 'dropped'); note(`${family}: not configured (OV_OAUTH_CLIENT_SECRET unset); dropped`); return null; }
const payload = { ...body, key: body.key || key(family) };
@@@ -82,7 +94,7 @@ async function send(family, body) {
let data = null;
let why;
try {
-- ({ status, data } = await request('POST', `/internal/chat/${family}`, payload));
++ ({ status, data } = await request('POST', path, payload));
why = `answered ${status}${data && data.error ? ` (${data.error})` : ''}`;
} catch (err) { why = err.message; }
if (status >= 200 && status < 300) { count(family, 'sent'); lastLog = null; return data || { ok: true }; }
@@@ -114,6 +126,64 @@ async function presence() {
return null;
}
++// ── Reads (Chat's internal read API; see the header) ─────────────────────────
++// All of these degrade like presence(): Chat's answer, or null plus one logged warning. A caller
++// that can only show the data answers with its own cached or empty value — never a 500.
++
++/** Query string for a GET read; arrays become Chat's comma lists, booleans its 1/0 flags. */
++function qs(params) {
++ const u = new URLSearchParams();
++ for (const [k, v] of Object.entries(params || {})) {
++ if (v == null || v === '') continue;
++ if (Array.isArray(v)) u.set(k, v.join(','));
++ else if (typeof v === 'boolean') u.set(k, v ? '1' : '0');
++ else u.set(k, String(v));
++ }
++ const s = u.toString();
++ return s ? `?${s}` : '';
++}
++
++/** GET one Chat read; its JSON body or null. Ordinary failures are logged and answered null. */
++async function read(name, path) {
++ if (drill() || !principal.configured()) return null;
++ try {
++ const { status, data } = await request('GET', path);
++ if (status === 200 && data && data.ok !== false) return data;
++ note(`${name}: Chat answered ${status}${data && data.error ? ` (${data.error})` : ''}`);
++ } catch (err) { note(`${name}: ${err.message}`); }
++ return null;
++}
++
++/** POST /internal/chat/stats — message/chatter counts, or the top chatters of a room. */
++async function readStats(body) {
++ if (drill() || !principal.configured()) return null;
++ try {
++ const { status, data } = await request('POST', '/internal/chat/stats', body || {});
++ if (status === 200 && data && data.ok !== false) return data;
++ note(`stats: Chat answered ${status}${data && data.error ? ` (${data.error})` : ''}`);
++ } catch (err) { note(`stats: ${err.message}`); }
++ return null;
++}
++
++const readMessages = (params) => read('messages', `/internal/chat/messages${qs(params)}`);
++const readTimeline = (params) => read('timeline', `/internal/chat/timeline${qs(params)}`);
++const readPendingIp = (params) => read('pending-ip', `/internal/chat/moderation/pending-ip${qs(params)}`);
++const readRelayUsers = (params) => read('relay-users', `/internal/chat/moderation/relay-users${qs(params)}`);
++const readRelayUser = (id) => read('relay-user', `/internal/chat/moderation/relay-users/${Number(id)}`);
++const readTtsOverride = (params) => read('tts-override', `/internal/chat/moderation/tts-override${qs(params)}`);
++const readSounds = (params) => read('sounds', `/internal/chat/sounds${qs(params)}`);
++/** Record where Live's asset sync uploaded a channel sound (idempotent; Chat applies the same write once). */
++async function soundAsset(body) {
++ if (drill() || !principal.configured()) return null;
++ try {
++ // Chat's /sounds/asset takes exactly { id, media_url, media_asset_id } — no idempotency key.
++ const { status, data } = await request('POST', '/internal/chat/sounds/asset', body);
++ if (status === 200 && data && data.ok !== false) return data;
++ note(`sounds-asset: Chat answered ${status}${data && data.error ? ` (${data.error})` : ''}`);
++ } catch (err) { note(`sounds-asset: ${err.message}`); }
++ return null;
++}
++
module.exports = {
AUDIENCE,
key,
@@@ -123,5 +193,6 @@ module.exports = {
moderation: (body) => send('moderation', body),
invalidate: (body) => send('invalidate', body),
presence,
++ readStats, readMessages, readTimeline, readPendingIp, readRelayUsers, readRelayUser, readTtsOverride, readSounds, soundAsset,
_setRetryMs(ms) { retryMs = ms || RETRY_MS; },
};
ddiff --git a/server/chat/chat-reads.js b/server/chat/chat-reads.js
nnew file mode 100644
iindex 00000000..930af805
---- /dev/null
++++ b/server/chat/chat-reads.js
@@@ -0,0 +1,519 @@
++'use strict';
++/**
++ * The chat reads Live's own features still make, answered from OpenVibe.Chat (plan T3 J4b:
++ * "Read Live's chat stats, queues and history from Chat"). Chat owns chat_messages, channel_sounds,
++ * pending_ip_messages, hidden_relay_users and tts_voice_overrides, so Live's home stats, recaps,
++ * AI context, admin console and mod queues read them through Chat's internal read API
++ * (server/chat/chat-client.js). Live's own tables remain as Chat's read mirror until a later step
++ * drops them, so the rollback path and the readers that have not moved yet keep working.
++ *
++ * Mode-aware: when CHAT_AUTHORITY=chat, every answer comes from Chat; otherwise Live runs chat
++ * itself and the same call answers from Live's own tables (the rollback / dev path) — the caller
++ * never has to know which.
++ *
++ * Caching: an answer is fresh for CACHE_TTL_MS. An async read past that asks Chat again; a peek
++ * (for a caller that cannot await) answers the last good value while that refresh runs, so a peek
++ * never drops a known answer just because its TTL passed. A good value is kept for CACHE_KEEP_MS,
++ * so a Chat failure answers the last good value, else null — a failure never overwrites a good
++ * answer, and it sets a FAIL_TTL_MS backoff so a Chat outage cannot turn a hot reader (a relay
++ * line, a peek per tick) into a retry per call. That backoff is only for cached reads: a read the
++ * moderation console acts on (queues, one message, a voice override) is never cached (ttl: 0), is
++ * never short-circuited by a cooling failure and never shares another read's in-flight request, and
++ * throws a `Chat unavailable` error (`.unavailable`) on a Chat failure, so those routes answer 503
++ * instead of stale or empty queue data. Off under LIVE_DRILL.
++ *
++ * Callers that can await use the async function; the few that cannot (the AI persona prompt, the
++ * home-stats snapshot, the TTS engine) use the matching peek*().
++ */
++const { isRemote } = require('./chat-authority');
++const client = require('./chat-client');
++
++const CACHE_TTL_MS = 15_000; // how long an answer is used without asking Chat again
++const CACHE_KEEP_MS = 15 * 60_000; // how long a last-good value answers peeks and failures
++const FAIL_TTL_MS = 2_000; // how long a failure is not retried (it never replaces the value)
++
++const cache = new Map(); // key -> { at, value, failed?, failedAt? }
++const inflight = new Map(); // key -> Promise
++let cacheMax = 2000; // test hook; the cap remember()/rememberFailure() evict down to
++let lastLog = null;
++
++function drill() { try { return require('../drill').enabled; } catch { return false; } }
++function remote() { return isRemote() && !drill(); }
++function local() { return require('../db/database'); }
++
++function note(what, err) {
++ const m = `Chat read ${what}: ${(err && err.message) || err}`;
++ if (m === lastLog) return; // log once until it changes or a read succeeds
++ lastLog = m;
++ console.warn(`[ChatRead] ${m}`);
++}
++
++/** A Chat failure on a read the moderation console acts on: its route answers 503, never stale data. */
++function unavailable(what) {
++ const err = new Error(`Chat unavailable (${what})`);
++ err.unavailable = true;
++ return err;
++}
++/** A read Chat's API does not offer: the route answers with err.status (501) and this message. */
++function unsupported(message, status = 501) {
++ const err = new Error(message);
++ err.status = status;
++ return err;
++}
++
++function fresh(key) { const h = cache.get(key); return h && !h.failed && Date.now() - h.at < CACHE_TTL_MS ? h : null; }
++function kept(key) { const h = cache.get(key); return h && !h.failed && Date.now() - h.at < CACHE_KEEP_MS ? h : null; }
++function cooling(key) { const h = cache.get(key); const at = h && (h.failedAt || (h.failed && h.at)); return !!at && Date.now() - at < FAIL_TTL_MS; }
++/** Keep the cache bounded: drop entries past the keep window first, then the oldest. */
++function evict() {
++ if (cache.size <= cacheMax) return;
++ for (const [k, h] of cache) if (Date.now() - h.at >= CACHE_KEEP_MS) cache.delete(k);
++ while (cache.size > cacheMax) cache.delete(cache.keys().next().value);
++}
++function remember(key, value) {
++ cache.set(key, { at: Date.now(), value });
++ evict();
++}
++
++/** Record that this key just failed without overwriting its last good value. */
++function rememberFailure(key) {
++ const h = cache.get(key);
++ if (h && !h.failed) { h.failedAt = Date.now(); return; }
++ // Same bounded insert as a good value: a Chat outage fails per-chatter, per-message and
++ // per-query keys, and must not grow the cache past the cap.
++ cache.set(key, { at: Date.now(), value: null, failed: true });
++ evict();
++}
++
++/**
++ * Async answer: Chat (remote) or Live's own tables (local), cached. A failure answers the last
++ * good value, else null; with `strict` it throws `Chat unavailable` instead (the moderation
++ * console). `ttl: 0` always reads live and never caches. An error carrying a `status` (a
++ * deliberate "not supported") is handed to the caller as it is.
++ */
++function load(key, remoteFn, localFn, { ttl = CACHE_TTL_MS, strict = false } = {}) {
++ if (!remote()) {
++ return Promise.resolve().then(async () => { try { return await localFn(); } catch (err) { note(key, err); return null; } });
++ }
++ if (ttl > 0) { const hit = fresh(key); if (hit) return Promise.resolve(hit.value); }
++ // The backoff only short-circuits a cached, non-strict read. A read the console acts on
++ // (ttl 0 / strict) always reaches Chat, so one blip cannot answer the moderator's retry with an
++ // instant 503 from the failure that is still cooling.
++ if (ttl > 0 && !strict && cooling(key)) {
++ const last = kept(key);
++ return Promise.resolve(last ? last.value : null);
++ }
++ // Only cached reads share an in-flight request. A live (ttl 0) read must not join a read that
++ // started before the write it follows (an approve, a delete, an unhide).
++ if (ttl > 0 && inflight.has(key)) return inflight.get(key);
++ const p = (async () => {
++ let value;
++ try { value = await remoteFn(); }
++ catch (err) {
++ note(key, err);
++ if (err && err.status) throw err;
++ rememberFailure(key);
++ if (strict) throw unavailable(key);
++ const last = kept(key);
++ return last ? last.value : null;
++ }
++ lastLog = null;
++ if (ttl > 0) remember(key, value);
++ return value;
++ })().finally(() => inflight.delete(key));
++ if (ttl > 0) inflight.set(key, p);
++ return p;
++}
++
++/** Sync answer for a caller that cannot await: fresh value, else the last good value + a background refresh, else null. */
++function peek(key, remoteFn, localFn) {
++ if (!remote()) { try { return localFn(); } catch (err) { note(key, err); return null; } }
++ const hit = fresh(key);
++ if (hit) return hit.value;
++ load(key, remoteFn, localFn).catch(() => { /* logged in load() */ }); // floating-ok: refreshes for the next peek
++ const last = kept(key);
++ return last ? last.value : null;
++}
++
++// ── Stats ─────────────────────────────────────────────────────────────────────
++function localSiteStats() {
++ const db = local();
++ const r = db.get(`SELECT COUNT(*) AS messages,
++ COUNT(DISTINCT COALESCE('u:' || user_id, 'a:' || anon_id, source_platform || ':' || username)) AS chatters
++ FROM chat_messages WHERE is_deleted = 0`) || {};
++ return { messages: Number(r.messages) || 0, chatters: Number(r.chatters) || 0 };
++}
++async function remoteSiteStats() {
++ const out = await client.readStats({ kind: 'site' });
++ if (!out) throw unavailable('site stats');
++ return { messages: Number(out.messages) || 0, chatters: Number(out.chatters) || 0 };
++}
++/** { messages, chatters } across the whole site. */
++function siteStats() { return load('site', remoteSiteStats, localSiteStats); }
++/**
++ * The home-stats snapshot answers with `messages` only, so its local (rollback / dev) answer is a
++ * plain COUNT(*): the distinct-chatter scan `siteStats()` runs is a full scan this synchronous peek
++ * would otherwise force on every home-stats compute. In chat mode it answers Chat's cached value.
++ */
++function localSiteStatsMessages() {
++ const db = local();
++ const r = db.get('SELECT COUNT(*) AS messages FROM chat_messages WHERE is_deleted = 0') || {};
++ return { messages: Number(r.messages) || 0 };
++}
++function siteStatsPeek() { return peek('site', remoteSiteStats, localSiteStatsMessages); }
++
++/** The busiest chatters of a room (or the site, when neither stream nor channel is given), newest window first. */
++function topChatters({ since, streamId, channelUserId, limit = 10 } = {}) {
++ const key = `top:${since || 0}:${streamId || 0}:${channelUserId || 0}:${limit}`;
++ return load(key, async () => {
++ const out = await client.readStats({ kind: 'channel-top', since, stream_id: streamId, channel_user_id: channelUserId, limit });
++ if (!out || !Array.isArray(out.top_chatters)) throw unavailable('top chatters');
++ return out.top_chatters;
++ }, () => {
++ const db = local();
++ const where = ['c.is_deleted = 0', "c.message_type <> 'system'", 'COALESCE(u.is_banned, 0) = 0'];
++ const params = [];
++ if (since) { where.push('c.timestamp >= ?'); params.push(new Date(since).toISOString().slice(0, 19).replace('T', ' ')); }
++ if (streamId) { where.push('c.stream_id = ?'); params.push(streamId); }
++ if (channelUserId) { where.push('c.channel_user_id = ?'); params.push(channelUserId); }
++ params.push(limit);
++ return db.all(`SELECT c.user_id, MAX(c.username) AS username, u.display_name, u.avatar_url, u.profile_color, COUNT(*) AS count
++ FROM chat_messages c LEFT JOIN users u ON u.id = c.user_id
++ WHERE ${where.join(' AND ')}
++ GROUP BY c.user_id, CASE WHEN c.user_id IS NULL THEN c.username END
++ ORDER BY count DESC LIMIT ?`, params) || [];
++ });
++}
++
++// ── Messages / history ────────────────────────────────────────────────────────
++/**
++ * The admin console's chat search: { messages, total }. Chat's message read takes exactly one
++ * exact filter, so the first of user/anon/stream/username that is given is the one asked for and
++ * the rest are applied as a local post-filter. A username is matched EXACTLY (Chat's filter; it
++ * used to be a substring) — searching "bob" finds bob, not bob2. Chat offers no free-text search,
++ * so a search with only `query` answers 501 with a clear message rather than an empty page.
++ *
++ * A `query` combined with a filter is a local post-filter (case-insensitive `includes`) over the
++ * newest `limit + offset` rows Chat returned, not a scan of the whole table: a match older than
++ * that window is missed. (Full text search lands when Chat offers it.)
++ */
++function searchMessages({ query = '', userId = null, anonId = null, username = null, streamId = null, limit = 50, offset = 0 } = {}) {
++ const key = `search:${userId || ''}:${anonId || ''}:${username || ''}:${streamId || ''}:${query}:${limit}:${offset}`;
++ return load(key, async () => {
++ if (query && !userId && !anonId && !username && !streamId) {
++ throw unsupported('Text search is not available while OpenVibe.Chat serves chat — search by user id, anon id, stream id, or an exact username.');
++ }
++ const filter = userId ? { user_id: Number(userId) }
++ : anonId ? { anon_id: String(anonId) }
++ : streamId ? { stream_id: Number(streamId) }
++ : { username: String(username) };
++ const out = await client.readMessages({ ...filter, limit: Math.min(500, limit + offset) });
++ if (!out || !Array.isArray(out.messages)) throw unavailable('message search');
++ let rows = out.messages;
++ if (userId) rows = rows.filter((m) => Number(m.user_id) === Number(userId));
++ if (streamId) rows = rows.filter((m) => Number(m.stream_id) === Number(streamId));
++ if (anonId) rows = rows.filter((m) => String(m.anon_id) === String(anonId));
++ if (username) rows = rows.filter((m) => String(m.username || '').toLowerCase() === String(username).toLowerCase());
++ if (query) { const q = String(query).toLowerCase(); rows = rows.filter((m) => String(m.message || '').toLowerCase().includes(q)); }
++ const page = offset ? rows.slice(offset, offset + limit) : rows.slice(0, limit);
++ return { messages: page, total: rows.length + offset };
++ }, () => local().searchChatMessages({ query, userId, anonId, username, streamId, limit, offset }), { strict: true });
++}
++
++/** One message by id (the moderation delete path checks the row it is about). Live, never cached. */
++function messageById(id) {
++ if (!(Number(id) > 0)) return Promise.resolve(null);
++ return load(`m:${id}`, async () => {
++ const out = await client.readMessages({ id: Number(id) });
++ if (!out || !Array.isArray(out.messages)) throw unavailable('message');
++ return out.messages[0] || null;
++ }, () => local().getChatMessageById(Number(id)) || null, { ttl: 0, strict: true });
++}
++
++/** Newest `limit` messages of a channel, oldest→newest (the AI persona's chat delta reads the cache). */
++function channelMessages(channelUserId, limit = 40) {
++ const id = Number(channelUserId) || 0;
++ if (!id) return Promise.resolve([]);
++ const key = `ch:${id}:${limit}`;
++ return load(key, async () => {
++ const out = await client.readMessages({ channel_user_id: id, limit, types: ['chat', 'donation'] });
++ if (!out || !Array.isArray(out.messages)) throw unavailable('channel messages');
++ return out.messages.slice().reverse(); // Chat answers newest-first; the caller wants oldest-first
++ }, () => local().getChannelChatSince(id, 0, limit) || []);
++}
++function channelMessagesPeek(channelUserId, limit = 40) {
++ const id = Number(channelUserId) || 0;
++ if (!id) return [];
++ return peek(`ch:${id}:${limit}`, async () => {
++ const out = await client.readMessages({ channel_user_id: id, limit, types: ['chat', 'donation'] });
++ if (!out || !Array.isArray(out.messages)) throw unavailable('channel messages');
++ return out.messages.slice().reverse();
++ }, () => local().getChannelChatSince(id, 0, limit) || []) || [];
++}
++
++/** A user's chat history page (admin console): { messages, total }. */
++function userHistory(userId, { limit = 50, offset = 0 } = {}) {
++ const id = Number(userId) || 0;
++ if (!id) return Promise.resolve({ messages: [], total: 0 });
++ const key = `uh:${id}:${limit}:${offset}`;
++ return load(key, async () => {
++ const want = Math.min(500, limit + offset);
++ const out = await client.readMessages({ user_id: id, limit: want });
++ if (!out || !Array.isArray(out.messages)) throw unavailable('user history');
++ const rows = offset ? out.messages.slice(offset, offset + limit) : out.messages.slice(0, limit);
++ // Chat's read API answers a page, not a count; a lower bound keeps "load more" honest.
++ return { messages: rows, total: rows.length + offset };
++ }, () => local().getUserChatHistory(id, limit, offset), { strict: true });
++}
++
++// A relay row's username is stored prefixed: "[Label] name". The labels are the ones the relay
++// services write (PLATFORM_LABELS in chat-relay-service.js, "[RS] " in robotstreamer-service.js).
++const RELAY_LABELS = { twitch: 'Twitch', kick: 'Kick', youtube: 'YT', rs: 'RS' };
++function relayUsername(platform, username) {
++ const label = RELAY_LABELS[String(platform || '').toLowerCase()] || platform;
++ return `[${label}] ${username}`;
++}
++
++/** A relayed (external-platform) chatter's history page: { messages, total }. */
++function relayHistory(platform, username, { limit = 50, offset = 0, query = '' } = {}) {
++ if (!platform || !username) return Promise.resolve({ messages: [], total: 0 });
++ const key = `rh:${platform}:${username}:${limit}:${offset}:${query}`;
++ return load(key, async () => {
++ if (query) throw unsupported('Text search over relay history is not available while OpenVibe.Chat serves chat.');
++ const out = await client.readMessages({ username: relayUsername(platform, username), limit: Math.min(500, limit + offset) });
++ if (!out || !Array.isArray(out.messages)) throw unavailable('relay history');
++ const rows = out.messages.filter((m) => (m.source_platform || '') === String(platform).toLowerCase());
++ const page = offset ? rows.slice(offset, offset + limit) : rows.slice(0, limit);
++ return { messages: page, total: rows.length + offset };
++ }, () => local().getRelayUserChatHistory(platform, username, { limit, offset, query }));
++}
++
++/** A chatter's recent messages in ONE channel (an AI-viewer clone source); relay matches by handle. */
++function channelSamples(channelUserId, { userId = null, relay = null, limit = 30 } = {}) {
++ const key = `cs:${channelUserId}:${userId || ''}:${relay ? `${relay.platform}:${relay.rawUsername}` : ''}:${limit}`;
++ return load(key, async () => {
++ // Chat's message read takes exactly one filter. Chat owns the relay flag (source_platform),
++ // so a relay sample is by the exact stored username and re-filtered to the platform here.
++ if (relay) {
++ const out = await client.readMessages({ username: relayUsername(relay.platform, relay.rawUsername), limit: Math.min(500, limit * 4) });
++ if (!out || !Array.isArray(out.messages)) throw unavailable('channel samples');
++ return out.messages.filter((m) => (m.source_platform || '') === String(relay.platform).toLowerCase() && Number(m.channel_user_id) === Number(channelUserId)).slice(0, limit);
++ }
++ const out = await client.readMessages({ user_id: Number(userId), limit: Math.min(500, limit * 4) });
++ if (!out || !Array.isArray(out.messages)) throw unavailable('channel samples');
++ return out.messages.filter((m) => Number(m.channel_user_id) === Number(channelUserId)).slice(0, limit);
++ }, () => local().getChatSamplesInChannel(channelUserId, { userId, relay, limit }));
++}
++
++/** A channel's chat volume over the last `windowSec`, in `bucketSec` buckets: [{ count, tsEpoch }]. */
++function liveChatBuckets(streamId, windowSec = 150, bucketSec = 15) {
++ const id = Number(streamId) || 0;
++ if (!id) return Promise.resolve([]);
++ const key = `lcb:${id}:${windowSec}:${bucketSec}`;
++ return load(key, async () => {
++ const since = Date.now() - windowSec * 1000;
++ const out = await client.readTimeline({ stream_id: id, since, bucket_ms: bucketSec * 1000 });
++ if (!out || !Array.isArray(out.buckets)) throw unavailable('timeline');
++ return out.buckets.map((b) => ({ count: Number(b.count) || 0, tsEpoch: Math.round(Number(b.t) / 1000) }));
++ }, () => local().getLiveChatBuckets(id, windowSec, bucketSec) || []);
++}
++
++/** The message texts of a stream in the last `sinceSec` seconds (AI clip confirmation), oldest→newest. */
++function recentChatText(streamId, sinceSec = 120, limit = 40) {
++ const id = Number(streamId) || 0;
++ if (!id) return Promise.resolve([]);
++ const key = `rct:${id}:${sinceSec}:${limit}`;
++ return load(key, async () => {
++ const out = await client.readMessages({ stream_id: id, limit });
++ if (!out || !Array.isArray(out.messages)) throw unavailable('recent messages');
++ const floor = Date.now() - Math.max(1, sinceSec) * 1000;
++ return out.messages
++ .filter((m) => m.message && Date.parse(String(m.timestamp).replace(' ', 'T') + 'Z') >= floor)
++ .map((m) => String(m.message))
++ .reverse();
++ }, () => local().getRecentChatText(id, sinceSec, limit) || []);
++}
++
++/** The busiest `bucketSec` time-buckets of a stream, as offsets from `sinceMs`: [{ offset, count }]. */
++function spikeOffsets(streamId, bucketSec = 30, topN = 8, sinceMs) {
++ const id = Number(streamId) || 0;
++ if (!id) return Promise.resolve([]);
++ const key = `sp:${id}:${bucketSec}:${topN}:${sinceMs || 0}`;
++ return load(key, async () => {
++ const out = await client.readTimeline({ stream_id: id, since: sinceMs || 0, bucket_ms: bucketSec * 1000 });
++ if (!out || !Array.isArray(out.buckets)) throw unavailable('timeline');
++ const base = sinceMs || 0;
++ return out.buckets
++ .map((b) => ({ offset: Math.max(0, Math.round((Number(b.t) - base) / 1000 / bucketSec) * bucketSec), count: Number(b.count) || 0 }))
++ .sort((a, b) => b.count - a.count || a.offset - b.offset)
++ .slice(0, topN);
++ }, () => local().getChatSpikeOffsets(id, bucketSec, topN) || []);
++}
++
++// ── Moderation queues ─────────────────────────────────────────────────────────
++// These answer the console's own actions, so they are live (ttl: 0) and strict: a Chat failure is
++// a 503, never a stale or empty queue (it could hide a ban the moderator just set).
++
++/** Held IP-approval messages for a channel: [{ … }]. */
++function pendingIp(channelId, { limit = 50 } = {}) {
++ const id = Number(channelId) || 0;
++ if (!id) return Promise.resolve([]);
++ const key = `pip:${id}:${limit}`;
++ return load(key, async () => {
++ const out = await client.readPendingIp({ channel_id: id, limit });
++ if (!out || !Array.isArray(out.pending_ip)) throw unavailable('pending IP queue');
++ return out.pending_ip;
++ }, () => local().getPendingIpMessages(id, { limit }) || [], { ttl: 0, strict: true });
++}
++
++/** The hidden relay users of a channel (plus the site-wide rows): [{ … }]. */
++function relayUsers(channelId, { limit = 100 } = {}) {
++ const id = Number(channelId) || 0;
++ const key = `ru:${id}:${limit}`;
++ return load(key, async () => {
++ const out = await client.readRelayUsers({ channel_id: id || undefined, limit });
++ if (!out || !Array.isArray(out.relay_users)) throw unavailable('hidden relay users');
++ return out.relay_users;
++ }, () => local().getHiddenRelayUsers(id, { limit }) || [], { ttl: 0, strict: true });
++}
++
++/** One hidden relay user by id: { id, channel_id, … } | null. */
++function relayUser(id) {
++ const rid = Number(id) || 0;
++ if (!rid) return Promise.resolve(null);
++ return load(`r1:${rid}`, async () => {
++ const out = await client.readRelayUser(rid);
++ if (!out) throw unavailable('hidden relay user');
++ return out.relay_user || null;
++ }, () => {
++ const r = local().get('SELECT id, channel_id FROM hidden_relay_users WHERE id = ?', [rid]);
++ return r || null;
++ }, { ttl: 0, strict: true });
++}
++
++/**
++ * Is this relay identity hidden (banned) in this channel or site-wide? Synchronous: the relay
++ * path can carry a message per second and must not wait on an HTTP call.
++ *
++ * Live's own table answers first — Chat's mirror keeps it current, so a hide takes effect on the
++ * very next message. In chat mode Chat's queue (a cached peek, at most CACHE_TTL_MS old) is also
++ * consulted. Chat unreachable, or not cached yet, answers from Live's own table: fail OPEN, never
++ * drop a relayed line on a Chat outage.
++ */
++function isRelayUserHidden(channelId, platform, username) {
++ let localHidden = false;
++ try { localHidden = !!local().isRelayUserHidden(channelId, platform, username); } catch { /* not hidden */ }
++ if (localHidden) return true;
++ if (!remote()) return false;
++ const key = `hru:${Number(channelId) || 0}`;
++ const hit = fresh(key); // only a fresh answer is used: an unhide must not leave an old list in force
++ if (hit && Array.isArray(hit.value)) {
++ return hit.value.some((r) => String(r.platform) === String(platform) && String(r.external_username) === String(username));
++ }
++ load(key, async () => {
++ const out = await client.readRelayUsers({ channel_id: Number(channelId) || undefined, limit: 500 });
++ if (!out || !Array.isArray(out.relay_users)) throw unavailable('hidden relay users');
++ return out.relay_users;
++ }, () => []).catch(() => { /* warms the cache for the next check */ });
++ return false;
++}
++
++/** A user's TTS voice override: { voice, pitch, speed, gap } | null. Live, never cached. */
++function ttsOverride(identityKey) {
++ const k = String(identityKey || '').trim().toLowerCase();
++ if (!k) return Promise.resolve(null);
++ return load(`tts:${k}`, async () => {
++ const out = await client.readTtsOverride({ identity_key: k });
++ if (!out) throw unavailable('tts override');
++ return out.tts_override || null;
++ }, () => local().getTtsVoiceOverride(k), { ttl: 0, strict: true });
++}
++/**
++ * The TTS engine's synchronous read: the last good override while Chat refreshes in the background.
++ * On a cache miss or a Chat failure it answers Live's own mirror table (which Chat keeps current),
++ * so the first line after a restart, an eviction or during a Chat outage is still spoken with the
++ * override rather than the auto voice — the same fallback `isRelayUserHidden` uses.
++ */
++function ttsOverridePeek(identityKey) {
++ const k = String(identityKey || '').trim().toLowerCase();
++ if (!k) return null;
++ const localOverride = () => local().getTtsVoiceOverride(k);
++ const got = peek(`ttsp:${k}`, async () => {
++ const out = await client.readTtsOverride({ identity_key: k });
++ if (!out) throw unavailable('tts override');
++ return out.tts_override || null;
++ }, localOverride);
++ if (got) return got;
++ if (!remote()) return null; // peek already answered from Live's own table
++ try { return localOverride(); } catch { return null; }
++}
++
++// ── Channel sounds ────────────────────────────────────────────────────────────
++/** A channel's soundboard count. */
++function soundCount(ownerId) {
++ const id = Number(ownerId) || 0;
++ if (!id) return Promise.resolve(0);
++ return load(`sc:${id}`, async () => {
++ const out = await client.readSounds({ channel_owner_id: id });
++ if (!out || out.count == null) throw unavailable('sound count');
++ return Number(out.count);
++ }, () => {
++ const r = local().get('SELECT COUNT(*) AS count FROM channel_sounds WHERE channel_owner_id = ?', [id]);
++ return Number(r && r.count) || 0;
++ });
++}
++
++/**
++ * Channel sounds not yet uploaded to Media (asset-sync's work list), oldest first, paged by id.
++ * Never cached: the sync must see the upload it just recorded, or it uploads the same sound twice.
++ */
++function pendingSounds({ channelOwnerId, afterId, limit = 100 } = {}) {
++ const key = `ps:${channelOwnerId || 0}:${afterId || 0}:${limit}`;
++ return load(key, async () => {
++ const out = await client.readSounds({ pending_asset: true, channel_owner_id: channelOwnerId || undefined, after_id: afterId || undefined, limit });
++ if (!out || !Array.isArray(out.sounds)) throw unavailable('pending sounds');
++ return out.sounds;
++ }, () => {
++ const db = local();
++ const params = [];
++ let sql = 'SELECT * FROM channel_sounds WHERE media_asset_id IS NULL';
++ if (channelOwnerId) { sql += ' AND channel_owner_id = ?'; params.push(Number(channelOwnerId)); }
++ if (afterId) { sql += ' AND id > ?'; params.push(Number(afterId)); }
++ params.push(limit);
++ return db.all(`${sql} ORDER BY id LIMIT ?`, params) || [];
++ }, { ttl: 0 });
++}
++
++/** Record where the asset sync put a channel sound (Chat owns the row; a repeat is a no-op). */
++async function recordSoundAsset(id, mediaUrl, mediaAssetId) {
++ if (!(Number(id) > 0) || !(Number(mediaAssetId) > 0) || !mediaUrl) return null;
++ if (!remote()) {
++ try { local().run('UPDATE channel_sounds SET media_url = ?, media_asset_id = ? WHERE id = ?', [mediaUrl, mediaAssetId, id]); return {}; } catch { return null; }
++ }
++ return client.soundAsset({ id: Number(id), media_url: String(mediaUrl), media_asset_id: Number(mediaAssetId) });
++}
++
++/**
++ * Drop every cached answer whose key starts with `prefix`. A moderation write (a relay hide or
++ * unhide) calls this so the next read re-asks Chat instead of serving the list from before it.
++ */
++function invalidate(prefix) {
++ for (const k of cache.keys()) if (k.startsWith(prefix)) cache.delete(k);
++}
++
++/** Drop every cached answer (tests and env flips). */
++function _reset() { cache.clear(); inflight.clear(); lastLog = null; }
++/** Test hook: age every cached answer by `ms` (fresh → stale, or stale → dropped when it passes the keep window). */
++function _age(ms, prefix = '') { for (const [k, h] of cache) if (k.startsWith(prefix)) h.at -= ms; }
++/** Test hooks: the current entry count, and the eviction cap (set to a small value to exercise it). */
++function _size() { return cache.size; }
++function _cacheMax(n) { if (n != null && Number(n) > 0) cacheMax = Number(n); return cacheMax; }
++
++module.exports = {
++ siteStats, siteStatsPeek, topChatters,
++ messageById, channelMessages, channelMessagesPeek, userHistory, relayHistory, channelSamples, searchMessages,
++ liveChatBuckets, recentChatText, spikeOffsets,
++ pendingIp, relayUsers, relayUser, ttsOverride, ttsOverridePeek, isRelayUserHidden,
++ soundCount, pendingSounds, recordSoundAsset, invalidate,
++ _reset, _age, _size, _cacheMax, CACHE_TTL_MS, CACHE_KEEP_MS,
++};
ddiff --git a/server/chat/history-store.js b/server/chat/history-store.js
ddeleted file mode 100644
iindex c9e4240d..00000000
---- a/server/chat/history-store.js
++++ /dev/null
@@@ -1,149 +0,0 @@
--/**
-- * 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 };
ddiff --git a/server/chat/tts-engine.js b/server/chat/tts-engine.js
iindex e359f536..23a8781e 100644
---- a/server/chat/tts-engine.js
++++ b/server/chat/tts-engine.js
@@@ -16,6 +16,7 @@ const path = require('path');
const fs = require('fs');
const os = require('os');
const db = require('../db/database');
++const chatReads = require('./chat-reads');
// ── Voice Catalog ─────────────────────────────────────────────
// Every voice available on the platform — combines RS-Companion's
@@@ -208,8 +209,9 @@ function autoUserVoiceParams(identityKey) {
}
function deriveUserVoiceParams(identityKey) {
const key = String(identityKey || 'anon').trim().toLowerCase() || 'anon';
-- // An admin-set override wins over the auto-assigned voice.
-- try { const ov = db.getTtsVoiceOverride(key); if (ov && ov.voice) return _clampVoiceParams(ov); } catch { /* fall through to auto */ }
++ // An admin-set override (Chat's, or Live's own when Live runs chat) wins over the auto-assigned
++ // voice. A synchronous read: the cache, else the auto voice, then Chat warms for the next call.
++ try { const ov = chatReads.ttsOverridePeek(key); if (ov && ov.voice) return _clampVoiceParams(ov); } catch { /* fall through to auto */ }
return autoUserVoiceParams(key);
}
ddiff --git a/server/db/database.js b/server/db/database.js
iindex b6298b20..7a4bc74f 100644
---- a/server/db/database.js
++++ b/server/db/database.js
@@@ -3955,7 +3955,10 @@ function _computeHomeStats() {
clips: null,
liveSessions: c(`SELECT COUNT(*) AS count FROM streams`),
streamers: c(`SELECT COUNT(DISTINCT user_id) AS count FROM streams WHERE user_id IS NOT NULL`),
-- chatMessages: c(`SELECT COUNT(*) AS count FROM chat_messages`),
++ // OpenVibe.Chat owns chat_messages (roadmap T3); the total comes from Chat's read API when
++ // it is the authority, from Live's own table when Live runs chat itself. A Chat outage
++ // answers the cache, else null — never a 500 for the home page.
++ chatMessages: (() => { try { const s = require('../chat/chat-reads').siteStatsPeek(); return s && s.messages != null ? s.messages : null; } catch { return null; } })(),
users: c(`SELECT COUNT(*) AS count FROM users WHERE COALESCE(is_banned, 0) = 0`),
anons: c(`SELECT COUNT(*) AS count FROM anon_ip_mappings`),
follows: c(`SELECT COUNT(*) AS count FROM follows`),
ddiff --git a/server/integrations/chat-relay-service.js b/server/integrations/chat-relay-service.js
iindex d56194a2..b98ac8dc 100644
---- a/server/integrations/chat-relay-service.js
++++ b/server/integrations/chat-relay-service.js
@@@ -20,6 +20,7 @@ const http = require('http');
const db = require('../db/database');
const chatServer = require('../chat/chat-server');
++const chatReads = require('../chat/chat-reads');
const RECONNECT_BASE_MS = 5000;
const RECONNECT_MAX_MS = 60000;
@@@ -386,11 +387,13 @@ class ChatRelayService {
const color = PLATFORM_COLORS[bridge.platform] || '#888';
const prefixedUsername = `[${label}] ${username}`;
-- // Check if this relay user is hidden/banned
++ // Check if this relay user is hidden/banned. Chat's queue answers in chat mode (a cached
++ // peek — this path must not wait on an HTTP call), Live's own mirror-fed table otherwise;
++ // Chat unreachable falls back to that table, so a Chat outage never un-hides anyone.
try {
const stream = db.getStreamById(bridge.streamId);
const channel = stream?.channel_id ? db.getChannelById(stream.channel_id) : (stream ? db.getChannelByUserId(stream.user_id) : null);
-- if (channel && db.isRelayUserHidden(channel.id, bridge.platform, username)) {
++ if (channel && chatReads.isRelayUserHidden(channel.id, bridge.platform, username)) {
return; // Silently drop messages from hidden relay users
}
} catch { /* non-critical — allow message through on error */ }
ddiff --git a/server/integrations/robotstreamer-service.js b/server/integrations/robotstreamer-service.js
iindex 18e8c05a..04517659 100644
---- a/server/integrations/robotstreamer-service.js
++++ b/server/integrations/robotstreamer-service.js
@@@ -4,6 +4,7 @@ const WebSocket = require('ws');
const db = require('../db/database');
const chatServer = require('../chat/chat-server');
++const chatReads = require('../chat/chat-reads');
const delivery = require('../chat/chat-delivery');
const { authenticateWs } = require('../auth/auth');
@@@ -654,12 +655,13 @@ class RobotStreamerService {
const rawUsername = String(data.username || 'anon');
const username = `[RS] ${rawUsername}`;
-- // Check if this relay user is hidden/banned
++ // Check if this relay user is hidden/banned: Chat's queue in chat mode (a cached
++ // peek), Live's own mirror-fed table otherwise; Chat unreachable falls back to it.
try {
const channel = stream.channel_id
? db.getChannelById(stream.channel_id)
: db.getChannelByUserId(stream.user_id);
-- if (channel && db.isRelayUserHidden(channel.id, 'rs', rawUsername)) {
++ if (channel && chatReads.isRelayUserHidden(channel.id, 'rs', rawUsername)) {
return; // Silently drop messages from hidden/banned relay users
}
} catch { /* non-critical — allow message through on error */ }
ddiff --git a/server/media-proxy/asset-sync.js b/server/media-proxy/asset-sync.js
iindex 9178bb6e..190ab107 100644
---- a/server/media-proxy/asset-sync.js
++++ b/server/media-proxy/asset-sync.js
@@@ -16,6 +16,7 @@ const fs = require('fs');
const path = require('path');
const db = require('../db/database');
const media = require('../media-client');
++const chatReads = require('../chat/chat-reads');
const paths = require('../paths');
const SOUND_DIR = paths.dir('SOUNDS_PATH', 'sounds');
@@@ -65,18 +66,34 @@ async function syncAll() {
};
let synced = 0, failed = 0;
-- for (const s of db.all('SELECT * FROM channel_sounds WHERE media_asset_id IS NULL')) {
-- const f = _localFile(SOUND_DIR, s.url);
-- if (!f) continue;
-- try {
-- const asset = await _upload({
-- kind: 'sound', name: s.command, filePath: f,
-- user_id: s.created_by, username: s.created_by_name || uname(s.created_by),
-- channel_username: uname(s.channel_owner_id),
-- duration_seconds: s.duration_seconds || 0,
-- });
-- if (asset) { db.run('UPDATE channel_sounds SET media_url = ?, media_asset_id = ? WHERE id = ?', [asset.url, asset.id, s.id]); synced++; }
-- } catch (err) { failed++; if (failed <= 3) console.warn('[AssetSync] sound', s.command, err.message); }
++ // The work list is Chat's (channel_sounds is Chat's table); Live's own rows when Live runs
++ // chat. Page by id until a short page: sounds with no local file stay pending and would
++ // otherwise fill the first window forever, starving every later sound.
++ const PAGE = 100;
++ let afterId = 0;
++ for (;;) {
++ const batch = (await chatReads.pendingSounds({ afterId, limit: PAGE })) || [];
++ for (const s of batch) {
++ const f = _localFile(SOUND_DIR, s.url);
++ if (!f) continue;
++ try {
++ const asset = await _upload({
++ kind: 'sound', name: s.command, filePath: f,
++ user_id: s.created_by, username: s.created_by_name || uname(s.created_by),
++ channel_username: uname(s.channel_owner_id),
++ duration_seconds: s.duration_seconds || 0,
++ });
++ if (asset) {
++ // Chat owns the row: a null answer means the record did not land, so the
++ // sound stays pending and must be counted failed, not synced.
++ const recorded = await chatReads.recordSoundAsset(s.id, asset.url, asset.id);
++ if (recorded) synced++; else failed++;
++ }
++ } catch (err) { failed++; if (failed <= 3) console.warn('[AssetSync] sound', s.command, err.message); }
++ }
++ if (batch.length < PAGE) break;
++ afterId = Number(batch[batch.length - 1].id) || 0;
++ if (!afterId) break;
}
if (synced || failed) console.log(`[AssetSync] Synced ${synced} chat assets to Media${failed ? ` (${failed} failed — will retry next pass)` : ''}`);
ddiff --git a/server/streaming/routes.js b/server/streaming/routes.js
iindex 643782db..8ea073f6 100644
---- a/server/streaming/routes.js
++++ b/server/streaming/routes.js
@@@ -37,6 +37,7 @@ const chatRelayService = require('../integrations/chat-relay-service');
const chatServer = require('../chat/chat-server');
// The six chat tables are OpenVibe.Chat's (roadmap T3); the few reads here go through its client.
const moderation = require('../chat/moderation-client');
++const chatReads = require('../chat/chat-reads');
// Stream voice channels: Live's call server, or OpenVibe.Chat's with CALLS_AUTHORITY=chat.
const callsAuthority = require('./calls-authority');
const { sanitizeOfflineHtml, sanitizeOfflineCss } = require('./offline-html-sanitize');
@@@ -1631,7 +1632,7 @@ router.get('/setup-progress', requireAuth, async (req, res) => {
const user = safe(() => db.getUserById(uid), {}) || {};
const channel = safe(() => db.getChannelByUserId(uid), {}) || {};
const emotes = await moderation.getEmoteCount(uid).catch(() => 0);
-- const sounds = safe(() => db.get('SELECT COUNT(*) AS n FROM channel_sounds WHERE channel_owner_id = ?', [uid]).n, 0);
++ const sounds = (await chatReads.soundCount(uid).catch(() => 0)) || 0;
const goals = safe(() => db.get('SELECT COUNT(*) AS n FROM donation_goals WHERE user_id = ? AND is_active = 1', [uid]).n, 0);
const powerchat = safe(() => !!db.get('SELECT 1 FROM powerchat_connections WHERE user_id = ? LIMIT 1', [uid]), false);
const followers = await require('../social/network-follows').followerCount(uid).catch(() => 0);
ddiff --git a/test/chat-context.test.js b/test/chat-context.test.js
iindex a124d3c4..6df8bfba 100644
---- a/test/chat-context.test.js
++++ b/test/chat-context.test.js
@@@ -1,10 +1,9 @@
'use strict';
// Chat moved to OpenVibe.Chat (roadmap Wave 6). What Live answers it on /internal/chat-context/*
--// and /internal/chat-effects/* (service tokens, capabilities, re-checked moderators, the read
--// mirror), and the chat server Live's own modules get with CHAT_AUTHORITY=chat (presence reads,
--// cache hints and pushes over Chat's typed ingress; no bridge, no outbox). Against stub Network and
--// Chat servers.
++// and /internal/chat-effects/* (service tokens, capabilities, re-checked moderators), and the chat
++// server Live's own modules get with CHAT_AUTHORITY=chat (presence reads, cache hints and pushes
++// over Chat's typed ingress; no bridge, no outbox). Against stub Network and Chat servers.
const assert = require('assert');
const fs = require('fs');
@@@ -54,6 +53,82 @@ const network = http.createServer((req, res) => {
res.statusCode = 404; res.end('{}');
});
});
++// Chat's internal read API (roadmap T3 J4b): Live reads stats, queues, history and sounds through
++// these. The stub answers from Live's own tables (emulating Chat's copy) so the seeds apply, and
++// `readState.down` fails every read the way a Chat outage does.
++const readState = { down: false, soundAssetDown: false, relayOnly: [], msgById: 0 };
++function readReply(req, res, raw) {
++ const url = String(req.url);
++ const path = url.split('?')[0];
++ const q = new URLSearchParams(url.split('?')[1] || '');
++ const isRead = path === '/internal/chat/stats' || path === '/internal/chat/messages'
++ || path.startsWith('/internal/chat/moderation/') || path === '/internal/chat/sounds' || path === '/internal/chat/sounds/asset';
++ if (!isRead) return false;
++ if (readState.down) { res.statusCode = 503; res.end(JSON.stringify({ ok: false, error: 'Chat read unavailable' })); return true; }
++ if (path === '/internal/chat/sounds/asset' && readState.soundAssetDown) { res.statusCode = 503; res.end(JSON.stringify({ ok: false, error: 'Chat sound write unavailable' })); return true; }
++ const d2 = require('../server/db/database');
++ const ok = (o) => { res.end(JSON.stringify({ ok: true, ...o })); return true; };
++ const param = (col, v) => (['channel_user_id', 'stream_id', 'user_id', 'id'].includes(col) ? Number(v) : String(v));
++ if (path === '/internal/chat/stats') {
++ const b = JSON.parse(raw || '{}');
++ if (b.kind === 'channel-top') {
++ const t = d2.all(`SELECT user_id, MAX(username) AS username, COUNT(*) AS count FROM chat_messages WHERE is_deleted = 0 AND message_type <> 'system' GROUP BY user_id ORDER BY count DESC LIMIT ?`, [b.limit || 10]);
++ return ok({ top_chatters: t.map((r) => ({ user_id: r.user_id, username: r.username, count: Number(r.count) })) });
++ }
++ const r = d2.get('SELECT COUNT(*) AS messages FROM chat_messages WHERE is_deleted = 0');
++ return ok({ messages: Number(r.messages), chatters: 2 });
++ }
++ if (path === '/internal/chat/messages') {
++ if (q.get('id') != null) readState.msgById++;
++ const filters = ['channel_user_id', 'stream_id', 'user_id', 'anon_id', 'username', 'id'].filter((k) => q.get(k) != null);
++ if (filters.length !== 1) { res.statusCode = 400; res.end(JSON.stringify({ ok: false, error: 'Give exactly one filter' })); return true; }
++ const col = filters[0];
++ const where = [`${col} = ?`, 'is_deleted = 0'];
++ const params = [param(col, q.get(col))];
++ if (q.get('types')) { const types = q.get('types').split(','); where.push(`message_type IN (${types.map(() => '?').join(',')})`); params.push(...types); }
++ const rows = d2.all(`SELECT id, user_id, anon_id, username, message, message_type, is_global, stream_id, channel_user_id, source_platform, timestamp
++ FROM chat_messages WHERE ${where.join(' AND ')} ORDER BY id DESC LIMIT ?`, [...params, Number(q.get('limit') || 100)]);
++ return ok({ messages: rows, max_id: rows.length ? Number(rows[0].id) : null });
++ }
++ if (path === '/internal/chat/moderation/pending-ip') {
++ const rows = d2.all("SELECT * FROM pending_ip_messages WHERE channel_id = ? AND status = 'pending' ORDER BY id LIMIT ?", [Number(q.get('channel_id')), Number(q.get('limit') || 50)]);
++ return ok({ pending_ip: rows });
++ }
++ if (path === '/internal/chat/moderation/relay-users') {
++ const rows = d2.all(`SELECT h.*, u.username AS created_by_username FROM hidden_relay_users h LEFT JOIN users u ON u.id = h.created_by
++ WHERE h.channel_id = ? OR h.channel_id IS NULL ORDER BY h.id DESC LIMIT ?`, [Number(q.get('channel_id')), Number(q.get('limit') || 100)]);
++ return ok({ relay_users: rows.concat(readState.relayOnly) });
++ }
++ if (/^\/internal\/chat\/moderation\/relay-users\/\d+$/.test(path)) {
++ const rid = Number(path.split('/').pop());
++ const r = readState.relayOnly.find((x) => x.id === rid)
++ || d2.get('SELECT h.*, u.username AS created_by_username FROM hidden_relay_users h LEFT JOIN users u ON u.id = h.created_by WHERE h.id = ?', [rid]) || null;
++ return ok({ relay_user: r });
++ }
++ if (path === '/internal/chat/moderation/tts-override') {
++ const r = d2.get('SELECT * FROM tts_voice_overrides WHERE identity_key = ?', [String(q.get('identity_key'))]) || null;
++ return ok({ tts_override: r });
++ }
++ if (path === '/internal/chat/sounds/asset') {
++ const b = JSON.parse(raw || '{}');
++ d2.run('UPDATE channel_sounds SET media_url = ?, media_asset_id = ? WHERE id = ?', [b.media_url, b.media_asset_id, b.id]);
++ return ok({});
++ }
++ if (path === '/internal/chat/sounds') {
++ if (q.get('pending_asset') === '1') {
++ const where = ['media_asset_id IS NULL'];
++ const params = [];
++ if (q.get('channel_owner_id') != null) { where.push('channel_owner_id = ?'); params.push(Number(q.get('channel_owner_id'))); }
++ if (q.get('after_id') != null) { where.push('id > ?'); params.push(Number(q.get('after_id'))); }
++ const rows = d2.all(`SELECT * FROM channel_sounds WHERE ${where.join(' AND ')} ORDER BY id LIMIT ?`, [...params, Number(q.get('limit') || 100)]);
++ return ok({ sounds: rows });
++ }
++ const r = d2.get('SELECT COUNT(*) AS n FROM channel_sounds WHERE channel_owner_id = ?', [Number(q.get('channel_owner_id'))]);
++ return ok({ count: Number(r.n) });
++ }
++ return false;
++}
++
const chat = http.createServer((req, res) => {
let raw = '';
req.on('data', (c) => { raw += c; });
@@@ -69,6 +144,7 @@ const chat = http.createServer((req, res) => {
if (req.url === '/internal/chat/presence') {
return res.end(JSON.stringify({ total: 7, streams: { 1: 3 }, slow_mode: { 1: 5000 }, users: [{ user_id: 3, ip: '198.51.100.3', stream_id: 1 }], anons: [{ anon_id: 'anon9', ip: '203.0.113.9', stream_id: 1 }] }));
}
++ if (readReply(req, res, raw)) return;
// Chat's internal read API (roadmap T3): Live reads the six staged tables through it now.
const mm = String(req.url).match(/^\/internal\/moderation\/channels\/(\d+)(\/emote-count)?$/);
if (mm) {
@@@ -155,6 +231,7 @@ const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
app.use(express.json());
app.use('/internal/chat-context', routes.contextRouter);
app.use('/internal/chat-effects', routes.effectsRouter);
++ app.use('/api/mod', require('../server/admin/mod-routes'));
const port = await listen(http.createServer(app));
const call = async (method, p, { token, body, headers = {} } = {}) => {
const res = await fetch(`http://127.0.0.1:${port}${p}`, {
@@@ -287,7 +364,9 @@ const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
[viewer, 'VC_CALL_INVITE', 'STREAMER is calling you', 'Join voice channel: STREAMER\'s call', streamer, 'streamer']);
assert.ok(pushed[0].url.endsWith(`/?vcInvite=${encodeURIComponent(`user-${streamer}-x`)}`));
-- // 9. The read mirror: same ids, Chat-only columns ignored, Live-only columns kept.
++ // 9. The read mirror: same ids, Chat-only columns ignored, Live-only columns kept. Chat's
++ // LIVE_MIRROR is on in production, so the tables it keeps current stay Live's until the
++ // later step that removes both the mirror and its remaining readers.
const mirror = (changes) => call('POST', '/internal/chat-effects/mirror', { token: MIRROR, body: { changes } });
assert.strictEqual((await call('POST', '/internal/chat-effects/mirror', { token: WRITE, body: { changes: [] } })).status, 403, 'mirror has its own capability');
let m = await mirror([{ table: 'chat_messages', op: 'upsert', row: { id: 900001, stream_id: streamId, channel_user_id: streamer, user_id: viewer, username: 'VIEWER', message: 'hi from chat', message_type: 'chat', is_global: 0, is_deleted: 0, timestamp: '2026-09-22 10:00:00', subject_id: 'usr_01J9ZZZZZZZZZZZZZZZZZZZZZZ' } }]);
@@@ -337,7 +416,262 @@ const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
assert.match(written, /^HTTP\/1\.1 503/);
chatServer.close();
-- // 11. Robot commands from chat pass the control panel's gate: control mode, anonymous switch, whitelist, cooldown.
++ // 11. Live's chat reads (roadmap T3 J4b): stats, queues and history come from Chat's read
++ // API through server/chat/chat-reads.js, with the caching and failure rules the moderation
++ // console depends on.
++ {
++ const chatReads = require('../server/chat/chat-reads');
++ const d2 = require('../server/db/database');
++ chatReads._reset();
++ for (const c of ['media_url TEXT', 'media_asset_id INTEGER']) { try { d.exec(`ALTER TABLE channel_sounds ADD COLUMN ${c}`); } catch { /* present */ } }
++
++ // Chat's copy (the stub reads Live's tables): two lines by the same user, a held IP
++ // message, a hidden relay user, a TTS override, two pending sounds and a relayed line.
++ const m1 = Number(d2.saveChatMessage({ user_id: viewer, username: 'viewer', message: 'hello', channel_user_id: streamer, message_type: 'chat' }).lastInsertRowid);
++ d2.saveChatMessage({ user_id: viewer, username: 'viewer', message: 'again', channel_user_id: streamer, message_type: 'chat' });
++ d2.run("INSERT INTO pending_ip_messages (channel_id, ip_address, user_id, username, message, status) VALUES (?, '203.0.113.9', ?, 'viewer', 'held', 'pending')", [channel.id, viewer]);
++ d2.run("INSERT INTO hidden_relay_users (channel_id, platform, external_username, action, created_by) VALUES (?, 'twitch', 'alice', 'hide', ?)", [channel.id, streamer]);
++ d2.run("INSERT INTO tts_voice_overrides (identity_key, voice, pitch, speed, gap, set_by) VALUES ('user:viewer', 'en+f3', 99, 200, 0, ?)", [admin]);
++ d2.run("INSERT INTO channel_sounds (channel_owner_id, command, url, created_by) VALUES (?, 'honk', '/sounds/a.mp3', ?), (?, 'beep', '/sounds/b.mp3', ?)", [streamer, streamer, streamer, streamer]);
++ d2.saveChatMessage({ username: '[Twitch] alice', message: 'hi from twitch', source_platform: 'twitch', channel_user_id: streamer, message_type: 'chat' });
++
++ // (2)(3) A peek answers the last good value after its TTL passed, and a failure answers
++ // that value too — a failure never overwrites it with null.
++ assert.ok((await chatReads.siteStats()).messages >= 3);
++ chatReads._age(chatReads.CACHE_TTL_MS + 1000, 'site');
++ readState.down = true;
++ assert.ok(chatReads.siteStatsPeek(), 'a peek keeps the last good value past its TTL');
++ assert.ok(await chatReads.siteStats(), 'a failed read answers the last good value, never null');
++ chatReads._reset();
++ assert.strictEqual(chatReads.siteStatsPeek(), null, 'a cold peek is null and warms in the background');
++ readState.down = false;
++ for (let i = 0; i < 300 && !chatReads.siteStatsPeek(); i++) await sleep(10);
++ assert.ok(chatReads.siteStatsPeek(), 'the background warm-up answered');
++
++ // (2)(4) A TTS override survives a TTL and a Chat outage (the engine reads it
++ // synchronously); a cold peek answers Live's own mirror table, not the auto voice.
++ assert.strictEqual(chatReads.ttsOverridePeek('user:viewer').voice, 'en+f3', 'a cold peek falls back to Live\'s own table');
++ for (let i = 0; i < 300 && !chatReads.ttsOverridePeek('user:viewer'); i++) await sleep(10);
++ assert.strictEqual(chatReads.ttsOverridePeek('user:viewer').voice, 'en+f3');
++ chatReads._age(chatReads.CACHE_TTL_MS + 1000, 'ttsp:');
++ readState.down = true;
++ assert.strictEqual(chatReads.ttsOverridePeek('user:viewer').voice, 'en+f3', 'an override never reverts on a TTL + outage');
++ readState.down = false;
++
++ // (2) The AI persona's chat delta keeps its last page over a TTL + outage, not an empty one.
++ for (let i = 0; i < 300 && !chatReads.channelMessagesPeek(streamer, 10).length; i++) await sleep(10);
++ assert.ok(chatReads.channelMessagesPeek(streamer, 10).length >= 2, 'the channel page warmed');
++ chatReads._age(chatReads.CACHE_TTL_MS + 1000, 'ch:');
++ readState.down = true;
++ assert.ok(chatReads.channelMessagesPeek(streamer, 10).length >= 2, 'the delta answers its last page, never empty');
++ readState.down = false;
++
++ // (5) Queue reads are live: a change in Chat shows on the very next read.
++ const pending = await chatReads.pendingIp(channel.id, { limit: 5 });
++ assert.strictEqual(pending.length, 1);
++ assert.strictEqual(pending[0].message, 'held');
++ d2.run("UPDATE pending_ip_messages SET status = 'approved' WHERE channel_id = ?", [channel.id]);
++ assert.deepStrictEqual(await chatReads.pendingIp(channel.id, { limit: 5 }), [], 'the queue is never served stale');
++
++ // (3)(4) Moderation reads fail closed: Chat unavailable throws, never empty data or a 404.
++ readState.down = true;
++ chatReads._reset();
++ await assert.rejects(chatReads.pendingIp(channel.id, { limit: 5 }), (e) => e.unavailable === true);
++ await assert.rejects(chatReads.messageById(m1), (e) => e.unavailable === true);
++ readState.down = false;
++ chatReads._reset();
++
++ // (8) A user's history total is a page lower bound, not the highest message id.
++ const hist = await chatReads.userHistory(viewer, { limit: 1, offset: 0 });
++ assert.strictEqual(hist.messages.length, 1);
++ assert.strictEqual(hist.total, 1, 'total counts the page, not a message id');
++ const hist2 = await chatReads.userHistory(viewer, { limit: 1, offset: 1 });
++ assert.strictEqual(hist2.messages.length, 1, 'the second page comes from Chat');
++ assert.strictEqual(hist2.total, 2);
++
++ // (7) Relay history searches the stored "[Label] name", not the raw handle.
++ const rh = await chatReads.relayHistory('twitch', 'alice', { limit: 10 });
++ assert.strictEqual(rh.messages.length, 1, 'relay history finds [Twitch] alice');
++ assert.strictEqual(rh.messages[0].message, 'hi from twitch');
++ const cs = await chatReads.channelSamples(streamer, { relay: { platform: 'twitch', rawUsername: 'alice' }, limit: 10 });
++ assert.strictEqual(cs.length, 1, 'a relay clone sample finds the prefixed row');
++
++ // (1) Hidden relay users: Live's mirror row hides at once; Chat's queue hides even when
++ // the mirror has not seen it; a Chat outage fails open to Live's own table.
++ assert.strictEqual(chatReads.isRelayUserHidden(channel.id, 'twitch', 'alice'), true, 'the mirror row hides at once');
++ readState.relayOnly = [{ id: 999999, channel_id: channel.id, platform: 'youtube', external_username: 'ghost' }];
++ assert.strictEqual(chatReads.isRelayUserHidden(channel.id, 'youtube', 'ghost'), false, 'the first check warms Chat in the background');
++ for (let i = 0; i < 300 && !chatReads.isRelayUserHidden(channel.id, 'youtube', 'ghost'); i++) await sleep(10);
++ assert.strictEqual(chatReads.isRelayUserHidden(channel.id, 'youtube', 'ghost'), true, "Chat's queue hides a user Live's mirror has not seen");
++ readState.down = true;
++ chatReads._age(chatReads.CACHE_TTL_MS + 1000, 'hru:');
++ assert.strictEqual(chatReads.isRelayUserHidden(channel.id, 'youtube', 'ghost'), false, "Chat unreachable fails open to Live's own table");
++ assert.strictEqual(chatReads.isRelayUserHidden(channel.id, 'twitch', 'alice'), true, "Live's own hidden row still applies");
++ readState.down = false;
++ readState.relayOnly = [];
++ chatReads._reset();
++
++ // (10) The asset-sync work list pages by id and is never a cached list of what was just uploaded.
++ const first = await chatReads.pendingSounds({ limit: 1 });
++ assert.strictEqual(first[0].command, 'honk');
++ const second = await chatReads.pendingSounds({ afterId: first[0].id, limit: 1 });
++ assert.strictEqual(second[0].command, 'beep', 'afterId pages past the first window');
++ await chatReads.recordSoundAsset(first[0].id, 'https://media.example/a/1', 4242);
++ const after = await chatReads.pendingSounds({ afterId: 0, limit: 10 });
++ assert.deepStrictEqual(after.map((s) => s.command), ['beep'], 'the work list is read live, not served from a cache');
++
++ // The sync itself walks the whole list: 119 fileless sounds first, then the one with a
++ // local file — a single 100-wide window would never see it.
++ fs.writeFileSync(path.join(tmp, 'sounds', 's120.mp3'), 'not really audio');
++ const values = [];
++ const soundParams = [];
++ for (let i = 1; i <= 120; i++) { values.push('(?, ?, ?, ?)'); soundParams.push(streamer, `s${i}`, `/sounds/s${i}.mp3`, streamer); }
++ d2.run(`INSERT INTO channel_sounds (channel_owner_id, command, url, created_by) VALUES ${values.join(', ')}`, soundParams);
++ const media = require('../server/media-client');
++ const realRequest = media.request;
++ const uploaded = [];
++ media.request = async (method, p) => {
++ if (method === 'POST' && p === '/assets') { uploaded.push(1); return { asset: { id: 9000 + uploaded.length, url: `https://media.example/a/${9000 + uploaded.length}` } }; }
++ return { ok: true };
++ };
++ try { await require('../server/media-proxy/asset-sync').syncAll(); } finally { media.request = realRequest; }
++ assert.strictEqual(uploaded.length, 1, 'the sync pages past the first window to the sound with a local file');
++ const s120 = d2.get("SELECT media_asset_id FROM channel_sounds WHERE command = 's120'");
++ assert.ok(s120.media_asset_id, 'the paged sound was recorded as uploaded');
++
++ // (4)(6) The console routes: a text-only search says 501 and usernames match exactly; a
++ // Chat outage answers 503 from every queue route, never an empty page or a 404.
++ const adminToken = jwt.sign({ sub: String(admin), username: 'admin2', role: 'admin' }, keys.privateKey, { algorithm: 'RS256', issuer: ISS, expiresIn: 300 });
++ const modGet = (path) => call('GET', `/api/mod${path}`, { token: adminToken });
++ const textOnly = await modGet('/chat/search?q=hello');
++ assert.strictEqual(textOnly.status, 501);
++ assert.match(textOnly.body.error, /Text search/);
++ const byUser = await modGet(`/chat/search?user_id=${viewer}`);
++ assert.strictEqual(byUser.status, 200);
++ assert.strictEqual(byUser.body.messages.length, 2);
++ const byName = await modGet('/chat/search?user_id=viewer');
++ assert.strictEqual(byName.body.messages.length, 2, 'an exact username finds the rows');
++ assert.strictEqual((await modGet('/chat/search?user_id=view')).body.messages.length, 0, 'username search is exact');
++ readState.down = true;
++ chatReads._reset();
++ assert.strictEqual((await modGet(`/chat/search?user_id=${viewer}`)).status, 503);
++ assert.strictEqual((await modGet(`/chat/user/${viewer}`)).status, 503);
++ assert.strictEqual((await modGet(`/ip-approval/${channel.id}/pending`)).status, 503);
++ assert.strictEqual((await modGet(`/relay-users/hidden/${channel.id}`)).status, 503);
++ assert.strictEqual((await modGet('/tts-voice/user/viewer')).status, 503);
++ assert.strictEqual((await call('POST', '/api/mod/delete-message', { token: adminToken, body: { message_id: m1 } })).status, 503);
++ assert.strictEqual((await call('DELETE', '/api/mod/relay-user/999999', { token: adminToken })).status, 503);
++ readState.down = false;
++ chatReads._reset();
++
++ // ── Second review round ──────────────────────────────────────────
++
++ // (1) A Chat blip must not block the moderator's very next live read with the 2 s
++ // backoff: a ttl:0 / strict read always reaches Chat.
++ readState.down = true;
++ chatReads._reset();
++ await assert.rejects(chatReads.messageById(m1), (e) => e.unavailable === true, 'a live read 503s while Chat is down');
++ readState.down = false;
++ assert.ok(await chatReads.messageById(m1), 'the next live read reaches Chat, not a cooling 503');
++
++ // (2) A live (ttl 0) read never shares another read's in-flight request.
++ readState.msgById = 0;
++ await Promise.all([chatReads.messageById(m1), chatReads.messageById(m1)]);
++ assert.strictEqual(readState.msgById, 2, 'two concurrent ttl:0 reads both reach Chat');
++
++ // (3) A Chat failure is remembered through the same bounded setter as a good value.
++ chatReads._reset();
++ const realCap = chatReads._cacheMax();
++ chatReads._cacheMax(5);
++ readState.down = true;
++ try {
++ for (let i = 0; i < 12; i++) await chatReads.userHistory(viewer, { limit: i + 1 }).catch(() => {});
++ assert.ok(chatReads._size() <= 5, 'failed reads are evicted, never grow the cache past the cap');
++ } finally {
++ readState.down = false;
++ chatReads._cacheMax(realCap);
++ chatReads._reset();
++ }
++
++ // (4) A cold or outage peek answers Live's own override table, not the auto voice.
++ readState.down = true;
++ assert.strictEqual(chatReads.ttsOverridePeek('user:viewer').voice, 'en+f3', 'a peek falls back to Live\'s own table');
++ readState.down = false;
++ chatReads._reset();
++
++ // (5) The home-stats peek's local (rollback) answer is a messages-only count.
++ process.env.CHAT_AUTHORITY = '';
++ try {
++ chatReads._reset();
++ const peeked = chatReads.siteStatsPeek();
++ assert.strictEqual(peeked.chatters, undefined, 'the peek does not run the distinct-chatter scan');
++ assert.ok(peeked.messages >= 2, 'the peek still counts the messages');
++ const full = await chatReads.siteStats();
++ assert.ok(full.chatters >= 1, 'the full stats read still counts chatters');
++ } finally {
++ process.env.CHAT_AUTHORITY = 'chat';
++ chatReads._reset();
++ }
++
++ // (6) A sound whose Chat record fails is counted failed, not synced, and stays pending.
++ fs.writeFileSync(path.join(tmp, 'sounds', 's201.mp3'), 'audio');
++ d2.run("INSERT INTO channel_sounds (channel_owner_id, command, url, created_by) VALUES (?, 's201', '/sounds/s201.mp3', ?)", [streamer, streamer]);
++ readState.soundAssetDown = true;
++ {
++ const logs = [];
++ const origLog = console.log;
++ const mediaSync = require('../server/media-client');
++ const realReq = mediaSync.request;
++ mediaSync.request = async (method, p2) => (method === 'POST' && p2 === '/assets' ? { asset: { id: 9100, url: 'https://media.example/a/9100' } } : { ok: true });
++ console.log = (...a) => logs.push(a.join(' '));
++ try { await require('../server/media-proxy/asset-sync').syncAll(); }
++ finally { mediaSync.request = realReq; console.log = origLog; readState.soundAssetDown = false; }
++ assert.ok(logs.some((l) => /\(1 failed/.test(l)), 'a failed Chat record is counted failed, not synced');
++ assert.strictEqual(d2.get("SELECT media_asset_id FROM channel_sounds WHERE command = 's201'").media_asset_id, null, 'the sound stays pending for the next pass');
++ }
++
++ // (7) A relay hide or unhide through the console drops the cached hidden list at once.
++ chatReads._reset();
++ readState.relayOnly = [{ id: 999998, channel_id: channel.id, platform: 'twitch', external_username: 'marker1' }];
++ for (let i = 0; i < 300 && !chatReads.isRelayUserHidden(channel.id, 'twitch', 'marker1'); i++) await sleep(10);
++ assert.strictEqual(chatReads.isRelayUserHidden(channel.id, 'twitch', 'marker1'), true, 'warm: the marker is in the cached list');
++ assert.strictEqual(chatReads.isRelayUserHidden(channel.id, 'youtube', 'ghost3'), false, 'warm: ghost3 is not hidden');
++ readState.relayOnly = [
++ { id: 999998, channel_id: channel.id, platform: 'twitch', external_username: 'marker1' },
++ { id: 999997, channel_id: channel.id, platform: 'youtube', external_username: 'ghost3' },
++ ];
++ const hideRes = await call('POST', '/api/mod/relay-user/hide', { token: adminToken, body: { channel_id: channel.id, platform: 'youtube', external_username: 'ghost3' } });
++ assert.strictEqual(hideRes.status, 200, 'the hide route ran');
++ for (let i = 0; i < 300 && !chatReads.isRelayUserHidden(channel.id, 'youtube', 'ghost3'); i++) await sleep(10);
++ assert.strictEqual(chatReads.isRelayUserHidden(channel.id, 'youtube', 'ghost3'), true, 'the hide dropped the cached list at once');
++ readState.relayOnly = [];
++ const unhideRes = await call('POST', '/api/mod/relay-user/hide', { token: adminToken, body: { channel_id: channel.id, platform: 'youtube', external_username: 'ghost3', action: 'unhide' } });
++ assert.strictEqual(unhideRes.status, 200, 'the unhide route ran');
++ for (let i = 0; i < 300 && chatReads.isRelayUserHidden(channel.id, 'youtube', 'ghost3'); i++) await sleep(10);
++ assert.strictEqual(chatReads.isRelayUserHidden(channel.id, 'youtube', 'ghost3'), false, 'the unhide dropped the cached list at once');
++ // A site-wide row (channel_id null) invalidates every channel's cached list.
++ chatReads._reset();
++ readState.relayOnly = [{ id: 999996, channel_id: null, platform: 'twitch', external_username: 'marker2' }];
++ for (let i = 0; i < 300 && !chatReads.isRelayUserHidden(424242, 'twitch', 'marker2'); i++) await sleep(10);
++ assert.strictEqual(chatReads.isRelayUserHidden(424242, 'twitch', 'marker2'), true, 'warm: another channel sees the site-wide marker');
++ readState.relayOnly = [
++ { id: 999996, channel_id: null, platform: 'twitch', external_username: 'marker2' },
++ { id: 999995, channel_id: null, platform: 'youtube', external_username: 'siteghost' },
++ ];
++ const siteHide = await call('POST', '/api/mod/relay-user/hide', { token: adminToken, body: { platform: 'youtube', external_username: 'siteghost' } });
++ assert.strictEqual(siteHide.status, 200, 'the site-wide hide ran');
++ for (let i = 0; i < 300 && !chatReads.isRelayUserHidden(424242, 'youtube', 'siteghost'); i++) await sleep(10);
++ assert.strictEqual(chatReads.isRelayUserHidden(424242, 'youtube', 'siteghost'), true, 'a site-wide hide clears every cached list');
++ readState.relayOnly = [];
++ chatReads._reset();
++
++ // (8) A `q` combined with a filter matches case-insensitively.
++ const qUpper = await chatReads.searchMessages({ userId: viewer, query: 'HELLO', limit: 10 });
++ assert.strictEqual(qUpper.messages.length, 1, 'q matches regardless of case');
++ assert.strictEqual(qUpper.messages[0].message, 'hello');
++ }
++
++ // 12. Robot commands from chat pass the control panel's gate: control mode, anonymous switch, whitelist, cooldown.
{
const sent = [];
require('../server/controls/control-server').hardwareClients.set('key-streamer', { readyState: 1, send: (m) => sent.push(JSON.parse(m)) });
ddiff --git a/test/chat-history-store.test.js b/test/chat-history-store.test.js
iindex de6e020b..45ff824b 100644
---- a/test/chat-history-store.test.js
++++ b/test/chat-history-store.test.js
@@@ -1,17 +1,25 @@
'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.
++ * Live's chat history reads now come from OpenVibe.Chat (roadmap T3 J4b: Read Live's chat stats,
++ * queues and history from Chat). Live's own history store — the page/delta reads over its copy of
++ * chat_messages — is gone with the read mirror, so nothing keeps a second answer to "what happened
++ * in this room?". These checks guard that the dead module stays deleted, and that the replacement
++ * (server/chat/chat-reads.js) answers history from Live's own tables while Live runs chat itself
++ * (the rollback / dev path) with the same rules: deleted and auto-expired rows never come back.
*/
const assert = require('assert');
++const fs = require('fs');
const os = require('os');
const path = require('path');
++const ROOT = path.join(__dirname, '..');
++assert.ok(!fs.existsSync(path.join(ROOT, 'server', 'chat', 'history-store.js')), 'history-store.js is deleted — history is Chat\'s now');
++
process.env.DB_PATH = path.join(os.tmpdir(), `ov-chat-history-${Date.now()}.db`);
++delete process.env.CHAT_AUTHORITY; // Live runs chat itself: chat-reads answers from Live's tables
const db = require('../server/db/database');
db.initDb();
--const store = require('../server/chat/history-store');
++const reads = require('../server/chat/chat-reads');
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;
@@@ -24,44 +32,29 @@ for (let i = 1; i <= 12; i++) {
// 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.
++// A channel room row (channel_user_id set) belongs to its channel's history.
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');
++(async () => {
++ // A user's site-wide history (deleted and auto-expired rows never come back). The rows share a
++ // timestamp here, so compare as a set.
++ const hist = await reads.userHistory(u, { limit: 50 });
++ assert.deepStrictEqual(hist.messages.map((m) => m.message).sort(), ['m1', 'm10', 'm11', 'm12', 'm2', 'm3', 'm4', 'm5', 'm8', 'm9'], 'user history, no deleted/expired rows');
++ assert.ok(hist.total >= hist.messages.length, 'a total is reported for paging');
++
++ // A channel's recent lines, oldest→newest — what an AI viewer's prompt sees.
++ const ch = await reads.channelMessages(bob, 10);
++ assert.deepEqual(ch.map((m) => m.message), ['channel only'], 'only the channel room, oldest → newest');
++ assert.strictEqual(ch[ch.length - 1].id, chanId);
++
++ // A chat search by user id (the admin console), same rows.
++ const search = await reads.searchMessages({ userId: u, limit: 50 });
++ assert.strictEqual(search.messages.length, hist.messages.length, 'search by user id matches the history');
++
++ // One message by id (the moderation delete path).
++ const one = await reads.messageById(ids[0]);
++ assert.strictEqual(one.message, 'm1');
++ assert.strictEqual(await reads.messageById(10 ** 9), null, 'an unknown id is null');
++
++ console.log('chat-history-store: ok — Live\'s store is deleted; chat-reads answers history in local mode');
++})().catch((e) => { console.error(e); process.exit(1); });
ddiff --git a/test/chat-ingress.test.js b/test/chat-ingress.test.js
iindex 076b7b00..521f6154 100644
---- a/test/chat-ingress.test.js
++++ b/test/chat-ingress.test.js
@@@ -60,10 +60,18 @@ const chat = http.createServer((req, res) => {
if (req.url === '/internal/live/calls') bridgeCalls.push(JSON.parse(raw));
return res.end(JSON.stringify({ ok: true, results: [] }));
}
-- const m = req.url.match(/^\/internal\/chat\/(messages|events|moderation|invalidate|presence)$/);
++ const m = req.url.match(/^\/internal\/chat\/(messages|events|moderation|invalidate|presence)(?:\?|$)/);
if (!m) { res.statusCode = 404; return res.end('{}'); }
const family = m[1];
if (family === 'presence') return res.end(JSON.stringify({ total: 4, streams: { 1: 2 }, slow_mode: {}, users: [{ user_id: 99, ip: '198.51.100.9', stream_id: 1 }], anons: [] }));
++ // Chat's message READ (Chat holds the rows now; Live reads them through it): answer the
++ // row Live's own copy carries, so the moderation path's permission check sees it.
++ if (family === 'messages' && req.method === 'GET') {
++ const d3 = require('../server/db/database');
++ const id = Number(new URL(req.url, 'http://x').searchParams.get('id'));
++ const row = d3.getChatMessageById(id);
++ return res.end(JSON.stringify({ ok: true, messages: row ? [row] : [], max_id: row ? id : null }));
++ }
const body = JSON.parse(raw || '{}');
calls.push({ family, auth, body });
const f = fail[family];
@@@ -147,6 +155,18 @@ function assertSigned(c, what) {
assertSigned(await waitFor(() => find('events', (b) => b.frame.type === 'system' && /Welcome alice/.test(b.frame.message)), 'welcome'), 'welcome');
assert.ok(d.prepare("SELECT 1 FROM stream_first_chats WHERE chatter_key = '[Twitch] alice' OR chatter_key = 'ext:[Twitch] alice'").get(), 'Live keeps its first-chat copy');
++ // 3b. A hidden relay user is dropped before any call to Chat; unhiding resumes relaying.
++ const relayLines = (message) => calls.filter((c) => c.family === 'messages' && c.body.username === '[Twitch] alice' && (!message || c.body.message === message)).length;
++ d.prepare("INSERT INTO hidden_relay_users (channel_id, platform, external_username, action, created_by) VALUES (?, 'twitch', 'alice', 'hide', ?)").run(channel.id, admin);
++ const beforeHidden = relayLines();
++ relay._broadcastMessage({ platform: 'twitch', streamId }, 'alice', 'should not relay', {});
++ await sleep(80);
++ assert.strictEqual(relayLines(), beforeHidden, 'a hidden relay user is dropped before Chat');
++ assert.strictEqual(relayLines('should not relay'), 0);
++ d.prepare("DELETE FROM hidden_relay_users WHERE platform = 'twitch' AND external_username = 'alice'").run();
++ relay._broadcastMessage({ platform: 'twitch', streamId }, 'alice', 'relayed again', {});
++ assertSigned(await waitFor(() => find('messages', (b) => b.source_platform === 'twitch' && b.message === 'relayed again'), 'the unhidden relay line'), 'relay line after unhide');
++
// 4. RobotStreamer mirror: a stub RS chat socket sends one line.
const wss = new WebSocketServer({ port: 0, host: '127.0.0.1' });
await new Promise((r) => wss.on('listening', r));
@@@ -176,7 +196,7 @@ function assertSigned(c, what) {
const del = find('moderation', (b) => b.action === 'delete-message');
assertSigned(del, 'message delete');
assert.deepStrictEqual([del.body.id, del.body.deleted_by, del.body.key], [900010, admin, 'live:moderation:delete:900010']);
-- assert.strictEqual(db.getChatMessageById(900010).is_deleted, 1, "Live's copy is deleted too");
++ assert.strictEqual(db.getChatMessageById(900010).is_deleted, 1, "Live's copy is deleted too (the read mirror is on in production)");
assertSigned(await waitFor(() => find('moderation', (b) => b.action === 'log' && b.action_type === 'message_delete'), 'delete log'), 'delete log');
const ban = await mod(`/users/${viewer}/ban`, { reason: 'spam' });
assert.ok(ban.status < 300, `ban answered ${ban.status}`);