openvibers/openvibe.live Commits
Move Live's last three mirror readers to Chat: the home daily series, first chat in a channel (user:<id>), and sound by command; nothing in chat mode reads the mirror now (#31)
Changes
@@@ -46,20 +46,23 @@ path are deleted.`server/net/service-guard.js`). Effects re-check the acting moderator/owner/admin and are refusedunless `CHAT_AUTHORITY=chat`.- 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++ `POST /internal/chat/stats` (`site`, `user`, `stream`, `channel-top` and `site-daily`),++ `GET /internal/chat/messages`, `/timeline`, `/first-chat`, `/sounds`, `/sounds/by-command`,++ `/moderation/pending-ip`, `/moderation/relay-users`, `/moderation/tts-override` — 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.--- **What still reads the mirror.** Under `CHAT_AUTHORITY=chat` the readers above answer from Chat;-- three still read Live's mirror tables and are why the mirror stays until T3 J4c drops it: the two-- HOME_SERIES charts `messages` and `active` (Chat exposes no site-wide per-day message/chatter-- series), the AI context's "first time chatting here" flag (`stream_first_chats`; Chat exposes no-- first-chat read), and the RobotStreamer `!sound` lookup (`channel_sounds` by command; Chat's sounds-- read answers only a count and the pending-asset list). The other mirror readers are on the-- Live-runs-chat path `CHAT_AUTHORITY=chat` leaves unmounted — `dm_*` (no `/api/dm`), Live's chat-- server, the deploy notice, the admin chat log and the leftover dev/rollback fallbacks.++ `/api/mod` queues used to read. A Chat outage answers a cached value or Live's own table, never a 500.++- **What reads the mirror.** Nothing in chat mode any more: the two HOME_SERIES charts `messages`++ and `active` read Chat's site-daily series, the AI context's "first time chatting here" flag reads++ Chat's first-chat read, and the RobotStreamer `!sound` lookup reads Chat's sounds-by-command read —++ each through [server/chat/chat-reads.js](../server/chat/chat-reads.js), falling back to Live's own++ table only while Chat is unreachable. The mirror (`POST /internal/chat-effects/mirror`) therefore++ has no chat-mode reader left and is removed in the remaining step: Chat stops its `LIVE_MIRROR`++ writer, then Live deletes the receiver (`live-context-routes.js` `effectsRouter`'s mirror route and++ its `live.chat_mirror.write` capability) and the mirror-kept chat tables. The readers that still++ answer from Live's tables are on the Live-runs-chat path `CHAT_AUTHORITY=chat` leaves unmounted —++ `dm_*` (no `/api/dm`), Live's chat server, the deploy notice, the admin chat log and the leftover++ dev/rollback fallbacks.- **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 themonly through [server/chat/moderation-client.js](../server/chat/moderation-client.js) — Chat's internal@@@ -203,11 +203,11 @@ function personBlock(line, channelUserId, settings) {else if (db.isFollowing(line.userId, channelUserId)) flags.push('follower');}}-- const key = line.userId ? `user:${line.username}` : (line.anonId ? `anon:${line.anonId}` : `ext:${line.username}`);-- // stream_first_chats is Chat's (a read-mirror table) and Chat's internal read API has no-- // first-chat route, so this stays on Live's copy — which Chat no longer fills for registered-- // users, making the flag best-effort. Drop it when Chat offers the read.-- if (line.userId !== channelUserId && settings.greet_first_timers && db.isFirstChatInChannel && db.isFirstChatInChannel(key, channelUserId)) flags.push('first time chatting here');++ // stream_first_chats is Chat's; a registered identity is `user:<user_id>` (never the++ // username), a relay one `ext:<prefixed username>`. The welcome flag reads Chat's first-chat++ // read through a sync peek that falls back to Live's own mirror-kept table while Chat is down.++ const identity = line.userId ? `user:${line.userId}` : (line.anonId ? `anon:${line.anonId}` : `ext:${line.username}`);++ if (line.userId !== channelUserId && settings.greet_first_timers && chatReads.firstChatPeek(channelUserId, identity)) flags.push('first time chatting here');} catch { /* */ }const insight = ins ? clip([ins.overview_24h || ins.overview_alltime, ins.memory].filter(Boolean).join(' — '), 220) : '';if (!insight && !flags.length) return '';@@@ -14,11 +14,13 @@* 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++ * readFirstChat(q) GET /internal/chat/first-chat chat.analysis.read has this identity ever chatted here* 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++ * readSoundByCommand(q) GET /internal/chat/sounds/by-command chat.sounds.read the approved sound a !command plays** 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@@@ -167,11 +169,28 @@ async function readStats(body) {const readMessages = (params) => read('messages', `/internal/chat/messages${qs(params)}`);const readTimeline = (params) => read('timeline', `/internal/chat/timeline${qs(params)}`);++const readFirstChat = (params) => read('first-chat', `/internal/chat/first-chat${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)}`);++/**++ * GET /internal/chat/sounds/by-command — the approved sound a !command plays. Chat answers its own 404 body++ * `{ ok: false, error: 'Sound not found' }` for a definitive "no such sound", surfaced as `{ sound: null }`; any++ * other 404 (an older Chat without the route, a proxy page) and every other failure answer null, so the caller++ * falls back to Live's table rather than silencing every !sound.++ */++async function readSoundByCommand(params) {++ if (drill() || !principal.configured()) return null;++ try {++ const { status, data } = await request('GET', `/internal/chat/sounds/by-command${qs(params)}`);++ if (status === 200 && data && data.ok !== false && data.sound) return data;++ if (status === 404 && data && data.ok === false && data.error === 'Sound not found') return { sound: null };++ note(`sound-by-command: Chat answered ${status}${data && data.error ? ` (${data.error})` : ''}`);++ } catch (err) { note(`sound-by-command: ${err.message}`); }++ return null;++}/** 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;@@@ -193,6 +212,7 @@ module.exports = {moderation: (body) => send('moderation', body),invalidate: (body) => send('invalidate', body),presence,-- readStats, readMessages, readTimeline, readPendingIp, readRelayUsers, readRelayUser, readTtsOverride, readSounds, soundAsset,++ readStats, readMessages, readTimeline, readFirstChat, readPendingIp, readRelayUsers, readRelayUser, readTtsOverride,++ readSounds, readSoundByCommand, soundAsset,_setRetryMs(ms) { retryMs = ms || RETRY_MS; },};@@@ -65,7 +65,8 @@ function unsupported(message, status = 501) {return err;}--function fresh(key) { const h = cache.get(key); return h && !h.failed && Date.now() - h.at < CACHE_TTL_MS ? h : null; }++/** A fresh entry: within its own ttl when it set one (first-chat caches false longer than true), else the default. */++function fresh(key) { const h = cache.get(key); if (!h || h.failed) return null; const ttl = h.ttl != null ? h.ttl : CACHE_TTL_MS; return Date.now() - h.at < ttl ? 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. */@@@ -74,8 +75,10 @@ function evict() {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 });++function remember(key, value, ttl) {++ const h = { at: Date.now(), value };++ if (ttl != null) h.ttl = ttl;++ cache.set(key, h);evict();}@@@ -95,7 +98,7 @@ function rememberFailure(key) {* 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 } = {}) {++function load(key, remoteFn, localFn, { ttl = CACHE_TTL_MS, strict = false, ttlFor = null } = {}) {if (!remote()) {return Promise.resolve().then(async () => { try { return await localFn(); } catch (err) { note(key, err); return null; } });}@@@ -122,7 +125,7 @@ function load(key, remoteFn, localFn, { ttl = CACHE_TTL_MS, strict = false } = {return last ? last.value : null;}lastLog = null;-- if (ttl > 0) remember(key, value);++ if (ttl > 0) remember(key, value, ttlFor ? ttlFor(value) : undefined);return value;})().finally(() => inflight.delete(key));if (ttl > 0) inflight.set(key, p);@@@ -130,11 +133,11 @@ function load(key, remoteFn, localFn, { ttl = CACHE_TTL_MS, strict = false } = {}/** 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) {++function peek(key, remoteFn, localFn, opts) {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++ load(key, remoteFn, localFn, opts).catch(() => { /* logged in load() */ }); // floating-ok: refreshes for the next peekconst last = kept(key);return last ? last.value : null;}@@@ -166,6 +169,60 @@ function localSiteStatsMessages() {}function siteStatsPeek() { return peek('site', remoteSiteStats, localSiteStatsMessages); }++// ── Home daily series (the hero charts) ──────────────────────────────────────++// Live's home `messages`/`active` charts are Chat's site-wide per-day series. One peek serves the++// whole envelope getHomeStatSeries returns: the per-day counts come from Chat's site-daily read, and++// `before`/`prev_total` keep Live's window-wide semantics (a COUNT / COUNT(DISTINCT) over the whole++// range), so they are Chat's `site` stats over [epoch, window start) and over the window just before++// it. The peek key is coarse (per UTC day) so every compute within a day hits the same cached answer.++const DAY_MS = 86_400_000;++const utcDayStart = (t) => Math.floor(t / DAY_MS) * DAY_MS;++const clampSeriesDays = (days) => Math.max(1, Math.min(365, parseInt(days, 10) || 30));++++async function remoteHomeSeries(metric, days) {++ const until = utcDayStart(Date.now()) + DAY_MS; // exclusive: tomorrow 00:00 UTC++ const since = until - days * DAY_MS;++ const out = await client.readStats({ kind: 'site-daily', since, until });++ if (!out || !Array.isArray(out.days)) throw unavailable('home series');++ const field = metric === 'active' ? 'chatters' : 'messages';++ const byDay = new Map(out.days.map((d) => [String(d.day), Number(d[field]) || 0]));++ const points = [];++ for (let i = days - 1; i >= 0; i--) {++ const day = new Date(until - (i + 1) * DAY_MS).toISOString().slice(0, 10);++ points.push({ day, value: Number((byDay.get(day) || 0).toFixed(2)) });++ }++ const total = Number(points.reduce((a, p) => a + p.value, 0).toFixed(2));++ const [beforeOut, prevOut] = await Promise.all([++ client.readStats({ kind: 'site', until: since }),++ client.readStats({ kind: 'site', since: since - days * DAY_MS, until: since }),++ ]);++ if (!beforeOut || !prevOut) throw unavailable('home series totals');++ const pick = (o) => Number(metric === 'active' ? o.chatters : o.messages) || 0;++ return { metric, kind: 'count', days, points, total, peak: Math.max(0, ...points.map((p) => p.value)), before: pick(beforeOut), prev_total: pick(prevOut) };++}++++/** The daily series behind a hero stat, from Chat (remote) or Live's own tables (local), cached. */++function homeSeries(metric, days) {++ const d = clampSeriesDays(days);++ const key = `hs:${metric}:${d}:${utcDayStart(Date.now())}`;++ const localFn = () => local().homeSeriesLocal(metric, d);++ return load(key, () => remoteHomeSeries(metric, d), localFn).then((v) => v || safeLocal(localFn, null));++}++/**++ * The same for the synchronous home-stats series route: the last good answer (or, on a cold cache,++ * Live's mirror-filled tables) while Chat refreshes in the background — never zeros just because++ * Chat has not answered yet.++ */++function homeSeriesPeek(metric, days) {++ const d = clampSeriesDays(days);++ const key = `hs:${metric}:${d}:${utcDayStart(Date.now())}`;++ const localFn = () => local().homeSeriesLocal(metric, d);++ const v = peek(key, () => remoteHomeSeries(metric, d), localFn);++ if (v) return v;++ if (!remote()) return null; // the peek already answered from Live's own tables++ return safeLocal(localFn, null); // cold / Chat unreachable: Live's mirror-filled tables++}++/** 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}`;@@@ -487,6 +544,44 @@ function spikeOffsets(streamId, bucketSec = 30, topN = 8, sinceMs) {}, () => local().getChatSpikeOffsets(id, bucketSec, topN) || []);}++// ── First chat (the welcome check) ───────────────────────────────────────────++// Chat owns stream_first_chats; the AI context's "first time chatting here" flag asks Chat's++// first-chat read. A false answer (they have chatted here) is stable and cached longer; a true++// answer flips the moment they chat, so it is cached only briefly. The sync caller uses++// firstChatPeek; on a cold cache or a Chat outage it answers Live's own (mirror-kept) table.++const FIRST_CHAT_TTL_FALSE = 5 * 60_000; // has chatted here: stable++const FIRST_CHAT_TTL_TRUE = CACHE_TTL_MS; // first time: flips as soon as they chat++++function firstChatLocal(identity, channelUserId) {++ return safeLocal(() => local().isFirstChatInChannel(identity, channelUserId), false);++}++function firstChatRemote(channelUserId, identity) {++ return async () => {++ const out = await client.readFirstChat({ channel_id: channelUserId, identity });++ if (!out || typeof out.first !== 'boolean') throw unavailable('first-chat');++ return out.first;++ };++}++/** Has this identity (`user:<id>` | `anon:<anonId>` | `ext:<prefixed username>`) ever chatted here? */++function firstChat(channelUserId, identity) {++ const channelId = Number(channelUserId) || 0;++ const key = String(identity || '');++ if (!channelId || !key) return Promise.resolve(false);++ return load(`fc:${channelId}:${key}`, firstChatRemote(channelId, key), () => firstChatLocal(key, channelId), {++ ttl: FIRST_CHAT_TTL_TRUE, ttlFor: (v) => (v === false ? FIRST_CHAT_TTL_FALSE : FIRST_CHAT_TTL_TRUE),++ }).then((v) => (v == null ? firstChatLocal(key, channelId) : v === true));++}++/** The synchronous form (the AI persona prompt): the cached answer, else Live's own table. */++function firstChatPeek(channelUserId, identity) {++ const channelId = Number(channelUserId) || 0;++ const key = String(identity || '');++ if (!channelId || !key) return false;++ const v = peek(`fc:${channelId}:${key}`, firstChatRemote(channelId, key), () => firstChatLocal(key, channelId), {++ ttl: FIRST_CHAT_TTL_TRUE, ttlFor: (x) => (x === false ? FIRST_CHAT_TTL_FALSE : FIRST_CHAT_TTL_TRUE),++ });++ return (v === true || v === false) ? v : firstChatLocal(key, channelId);++}++// ── 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).@@@ -605,6 +700,37 @@ function soundCount(ownerId) {});}++const normalizeCommand = (command) => String(command || '').trim().toLowerCase().replace(/^!+/, '');++async function remoteSoundByCommand(ownerId, command) {++ const out = await client.readSoundByCommand({ channel_id: ownerId, command });++ if (out == null) throw unavailable('sound by command'); // unreachable: keep / fall back++ return out.sound || false; // the sound, or Chat's definitive none++}++/** The approved sound a !command plays (or null); Chat owns the row, Live's table is the fallback. */++function soundByCommand(ownerId, command) {++ const id = Number(ownerId) || 0;++ const cmd = normalizeCommand(command);++ if (!id || !cmd) return Promise.resolve(null);++ const localFn = () => local().getChannelSoundByCommand(id, cmd) || null;++ return load(`sbc:${id}:${cmd}`, () => remoteSoundByCommand(id, cmd), localFn).then((v) => {++ if (v === false) return null; // Chat answered: no such sound++ if (v) return v;++ return safeLocal(localFn, null); // cold / outage: Live's own table++ });++}++/** The synchronous form (the RobotStreamer !sound lookup): cached answer, else Live's own table. */++function soundByCommandPeek(ownerId, command) {++ const id = Number(ownerId) || 0;++ const cmd = normalizeCommand(command);++ if (!id || !cmd) return null;++ const localFn = () => local().getChannelSoundByCommand(id, cmd) || null;++ const v = peek(`sbc:${id}:${cmd}`, () => remoteSoundByCommand(id, cmd), localFn);++ if (v) return v;++ if (v === false) return null; // Chat answered: no such sound++ if (!remote()) return null; // the peek already answered from Live's own table++ return safeLocal(localFn, null); // cold / Chat unreachable: Live's mirror-filled table++}++/*** 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.@@@ -652,11 +778,11 @@ function _size() { return cache.size; }function _cacheMax(n) { if (n != null && Number(n) > 0) cacheMax = Number(n); return cacheMax; }module.exports = {-- siteStats, siteStatsPeek, topChatters,++ siteStats, siteStatsPeek, topChatters, homeSeries, homeSeriesPeek,windowStats, windowStatsPeek, streamStats, streamStatsPeek, channelMaxId, channelMaxIdPeek, userMessageCountPeek,messageById, channelMessages, channelMessagesPeek, userHistory, relayHistory, channelSamples, searchMessages,-- liveChatBuckets, recentChatText, spikeOffsets,++ liveChatBuckets, recentChatText, spikeOffsets, firstChat, firstChatPeek,pendingIp, relayUsers, relayUser, ttsOverride, ttsOverridePeek, isRelayUserHidden,-- soundCount, pendingSounds, recordSoundAsset, invalidate,++ soundCount, soundByCommand, soundByCommandPeek, pendingSounds, recordSoundAsset, invalidate,_reset, _age, _size, _cacheMax, CACHE_TTL_MS, CACHE_KEEP_MS,};@@@ -3805,9 +3805,10 @@ const HOME_SERIES = {anons: { table: 'anon_ip_mappings', ts: 'created_at', agg: 'COUNT(*)' },visitors: { table: 'anon_ip_mappings', ts: 'created_at', agg: 'COUNT(*)' },follows: { table: 'follows', ts: 'created_at', agg: 'COUNT(*)' },-- // OpenVibe.Chat exposes no site-wide per-day message/chatter series — its timeline read buckets-- // one channel or stream only, never the whole site — so these two charts still read Live's-- // mirror-filled chat_messages table. Every other series and the hero counters come from Chat.++ // Chat's site-wide per-day message/chatter series: in chat mode getHomeStatSeries answers them++ // from Chat's `site-daily` stats read (chat-reads.homeSeriesPeek); this local SQL stays the++ // dev / rollback / Chat-unreachable answer. `messages` maps to Chat's messages series, `active`++ // to its chatters series.messages: { table: 'chat_messages', ts: 'timestamp', agg: 'COUNT(*)', where: 'COALESCE(is_deleted, 0) = 0' },active: { table: 'chat_messages', ts: 'timestamp', agg: "COUNT(DISTINCT COALESCE('u' || user_id, 'a' || anon_id, 'r' || source_platform || '|' || username))", where: 'COALESCE(is_deleted, 0) = 0' },sessions: { table: 'streams', ts: 'created_at', agg: 'COUNT(*)' },@@@ -3827,7 +3828,11 @@ const HOME_SERIES = {};const HOME_SERIES_KEYS = [...Object.keys(HOME_SERIES), 'liveNow', 'viewersNow'];--function getHomeStatSeries(metric, days = 30) {++/**++ * The local (Live-table) daily series for one metric — the dev, rollback and Chat-unreachable++ * answer. getHomeStatSeries routes the two chat metrics through Chat first.++ */++function homeSeriesLocal(metric, days = 30) {const def = HOME_SERIES[metric];if (!def) return null;days = Math.max(1, Math.min(365, parseInt(days, 10) || 30));@@@ -3857,6 +3862,22 @@ function getHomeStatSeries(metric, days = 30) {before: Number(before.toFixed(2)), prev_total: Number(prevTotal.toFixed(2)) };}++/**++ * The daily series behind a hero stat. With CHAT_AUTHORITY=chat the two chat metrics (`messages`,++ * `active`) answer from Chat's site-wide per-day read (chat-reads.homeSeriesPeek — a cached++ * stale-while-revalidate peek that falls back to Live's own tables when Chat is unreachable);++ * every other metric is Live's own.++ */++function getHomeStatSeries(metric, days = 30) {++ if (metric === 'messages' || metric === 'active') {++ try {++ const s = require('../chat/chat-reads').homeSeriesPeek(metric, days);++ if (s) return s;++ } catch { /* fall through to Live's own tables */ }++ }++ return homeSeriesLocal(metric, days);++}++/*** The two live readings (streams live, people watching) over time, from the five-minute sampler.* Unlike counters they do not add up across a day, so each bucket carries the average reading@@@ -6995,7 +7016,7 @@ module.exports = {adoptOrphanedTimelineRows: () => _adoptOrphanedTimelineRows(getDb()),publicStream,getConcurrencyBaseline,-- getHomeStatSeries, HOME_SERIES_KEYS, vibesStatsSince, _computeHomeStats,++ getHomeStatSeries, homeSeriesLocal, HOME_SERIES_KEYS, vibesStatsSince, _computeHomeStats,getVodAiState, getClipAiState, forgetMediaItem,scheduleClipNotifyState, bumpClipNotifyNowState, markClipNotifiedState, getDueClipNotifies,getDb, initDb, run, get, all, close, announceModerationAction: _announceModerationAction,@@@ -677,9 +677,9 @@ class RobotStreamerService {if (trimmed.startsWith('!')) {const parts = trimmed.split(/\s+/);const scmd = parts[0].slice(1).toLowerCase();-- // Chat owns channel_sounds but its internal read API offers only a count and the-- // pending-asset list, no lookup by command, so this stays on Live's mirror table.-- if (scmd && db.getChannelSoundByCommand(stream.user_id, scmd)) {++ // Chat owns channel_sounds; ask its by-command read (a cached peek), which falls++ // back to Live's own table while Chat is unreachable.++ if (scmd && chatReads.soundByCommandPeek(stream.user_id, scmd)) {if (delivery.ingress()) {delivery.event({ kind: 'stream', id: stream.id }, {type: 'channel-sound', streamId: stream.id, command: scmd, args: parts.slice(1).map((a) => a.slice(0, 120)).slice(0, 20),@@@ -56,13 +56,14 @@ const network = http.createServer((req, res) => {// 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 };++const readState = { down: false, soundAssetDown: false, relayOnly: [], msgById: 0, firstChatCalls: 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';++ const isRead = path === '/internal/chat/stats' || path === '/internal/chat/messages' || path === '/internal/chat/first-chat'++ || path.startsWith('/internal/chat/moderation/') || path === '/internal/chat/sounds'++ || path === '/internal/chat/sounds/by-command' || 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; }@@@ -71,12 +72,39 @@ function readReply(req, res, raw) {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 || '{}');++ const sqlTime = (t) => new Date(t).toISOString().slice(0, 19).replace('T', ' ');++ const CHATTER = "COALESCE('u:' || user_id, 'a:' || anon_id, source_platform || ':' || username)";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 (b.kind === 'site-daily') {++ const DAY = 86400000;++ const dayStart = (t) => Math.floor(t / DAY) * DAY;++ const rows = d2.all(`SELECT substr(timestamp, 1, 10) AS day, COUNT(*) AS messages, COUNT(DISTINCT ${CHATTER}) AS chatters++ FROM chat_messages WHERE COALESCE(is_deleted, 0) = 0 AND timestamp >= ? AND timestamp < ? GROUP BY day`,++ [sqlTime(b.since), sqlTime(b.until)]);++ const by = new Map(rows.map((r) => [String(r.day), r]));++ const days = [];++ for (let t = dayStart(b.since), last = dayStart(b.until - 1); t <= last; t += DAY) {++ const day = new Date(t).toISOString().slice(0, 10);++ const r = by.get(day);++ days.push({ day, messages: r ? Number(r.messages) : 0, chatters: r ? Number(r.chatters) : 0 });++ }++ return ok({ days });++ }++ const where = ['is_deleted = 0'];++ const params = [];++ if (b.since != null) { where.push('timestamp >= ?'); params.push(sqlTime(b.since)); }++ if (b.until != null) { where.push('timestamp < ?'); params.push(sqlTime(b.until)); }++ if (b.kind === 'user' && b.user_id != null) { where.push('user_id = ?'); params.push(Number(b.user_id)); }++ const r = d2.get(`SELECT COUNT(*) AS messages, COUNT(DISTINCT ${CHATTER}) AS chatters FROM chat_messages WHERE ${where.join(' AND ')}`, params);++ return ok({ messages: Number(r.messages), chatters: Number(r.chatters) });++ }++ if (path === '/internal/chat/first-chat') {++ readState.firstChatCalls++;++ const r = d2.get('SELECT 1 AS present FROM stream_first_chats WHERE chatter_key = ? AND channel_user_id = ?', [String(q.get('identity')), Number(q.get('channel_id'))]);++ return ok({ first: !r });}if (path === '/internal/chat/messages') {if (q.get('id') != null) readState.msgById++;@@@ -114,6 +142,14 @@ function readReply(req, res, 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/by-command') {++ const cmd = String(q.get('command') || '').trim().toLowerCase().replace(/^!+/, '');++ // A 404 that is not Chat's own "no such sound" (an older Chat, a proxy page).++ if (cmd === 'bare404') { res.statusCode = 404; res.setHeader('content-type', 'text/html'); res.end('<html>Not Found</html>'); return true; }++ const r = d2.get('SELECT * FROM channel_sounds WHERE channel_owner_id = ? AND command = ? AND is_approved = 1 ORDER BY RANDOM() LIMIT 1', [Number(q.get('channel_id')), cmd]);++ if (!r) { res.statusCode = 404; res.end(JSON.stringify({ ok: false, error: 'Sound not found' })); return true; }++ return ok({ sound: r });++ }if (path === '/internal/chat/sounds') {if (q.get('pending_asset') === '1') {const where = ['media_asset_id IS NULL'];@@@ -134,7 +170,7 @@ const chat = http.createServer((req, res) => {req.on('data', (c) => { raw += c; });req.on('end', () => {res.setHeader('Content-Type', 'application/json');-- assert.ok(String(req.headers.authorization || '').startsWith('Bearer '), 'Live calls Chat with a service token');++ assert.ok(!String(req.url).startsWith('/internal/') || String(req.headers.authorization || '').startsWith('Bearer '), 'Live calls Chat with a service token');assert.ok(!String(req.url).startsWith('/internal/live/'), 'nothing goes over the old bridge');const im = String(req.url).match(/^\/internal\/chat\/(messages|events|moderation|invalidate)$/);if (im) {@@@ -756,6 +792,127 @@ const sleep = (ms) => new Promise((r) => setTimeout(r, ms));}}++ // 11b. The last three mirror readers read Chat now (roadmap T3 J4c): the home daily series,++ // the AI context's first-chat flag and the RobotStreamer !sound lookup.++ {++ const chatReads = require('../server/chat/chat-reads');++ const d2 = require('../server/db/database');++ const DAY = 86400000;++ const sqlNow = () => new Date().toISOString().slice(0, 19).replace('T', ' ');++++ // ── Home daily series: Chat's site-daily read, mapped per metric; a cold cache during a++ // Chat outage answers Live's own series (never zeros). ──++ d2.run('DELETE FROM chat_messages');++ chatReads._reset();++ const atDay = (back) => new Date(Date.now() - back * DAY).toISOString().slice(0, 10) + ' 12:00:00';++ d2.run("INSERT INTO chat_messages (user_id, username, message, message_type, timestamp) VALUES (?, 'viewer', 'a', 'chat', ?), (?, 'viewer', 'b', 'chat', ?), (?, 'moddy', 'c', 'chat', ?)",++ [viewer, atDay(2), viewer, atDay(2), mod, atDay(2)]);++ d2.run("INSERT INTO chat_messages (user_id, username, message, message_type, timestamp) VALUES (?, 'viewer', 'd', 'chat', ?), (?, 'moddy', 'e', 'chat', ?)",++ [viewer, atDay(1), mod, atDay(1)]);++ const series = chatReads.homeSeriesPeek('messages', 7);++ assert.ok(series && Array.isArray(series.points), 'the chat series peek answers in chat mode');++ assert.strictEqual(series.days, 7);++ assert.strictEqual(series.points.length, 7, 'one zero-filled point per day');++ assert.strictEqual(series.points[4].value, 3, 'two days ago: three messages (Chat)');++ assert.strictEqual(series.points[5].value, 2, 'yesterday: two messages (Chat)');++ assert.strictEqual(series.points[6].value, 0, 'today: none');++ assert.strictEqual(series.total, 5);++ assert.strictEqual(series.prev_total, 0, 'the window before the 7 days is empty');++ assert.strictEqual(series.before, 0, 'nothing before the window');++ const activeSeries = chatReads.homeSeriesPeek('active', 7);++ assert.strictEqual(activeSeries.points[4].value, 2, 'two days ago: two distinct chatters');++ assert.strictEqual(activeSeries.points[5].value, 2, 'yesterday: two distinct chatters');++ // getHomeStatSeries routes the two chat metrics through the chat series.++ const realSeriesPeek = chatReads.homeSeriesPeek;++ chatReads.homeSeriesPeek = () => ({ metric: 'messages', sentinel: true });++ try { assert.strictEqual(d2.getHomeStatSeries('messages', 7).sentinel, true, 'getHomeStatSeries reads the chat series in chat mode'); }++ finally { chatReads.homeSeriesPeek = realSeriesPeek; }++ // A cold cache during a Chat outage answers Live's own table, not zeros.++ chatReads._reset();++ readState.down = true;++ const fallbackSeries = chatReads.homeSeriesPeek('messages', 7);++ assert.ok(fallbackSeries && fallbackSeries.points.length === 7, 'a cold + down peek answers Live\'s own series');++ assert.strictEqual(fallbackSeries.total, 5, 'the fallback is Live\'s own series, not zeros');++ readState.down = false;++ chatReads._reset();++++ // ── First chat: `user:<user_id>` (never the username); a false answer is cached longer++ // than a true one; a cold/down peek falls to Live's own table. ──++ d2.run('DELETE FROM stream_first_chats');++ chatReads._reset();++ assert.strictEqual(await chatReads.firstChat(streamer, `user:${viewer}`), true, 'a new identity is first');++ d2.run('INSERT INTO stream_first_chats (chatter_key, channel_user_id) VALUES (?, ?)', [`user:${viewer}`, streamer]);++ chatReads._reset();++ assert.strictEqual(await chatReads.firstChat(streamer, `user:${viewer}`), false, 'after a recorded first chat, not first');++ chatReads._reset();++ readState.firstChatCalls = 0;++ await chatReads.firstChat(streamer, `user:${viewer}`);++ chatReads._age(chatReads.CACHE_TTL_MS + 1000, 'fc:');++ readState.firstChatCalls = 0;++ await chatReads.firstChat(streamer, `user:${viewer}`);++ assert.strictEqual(readState.firstChatCalls, 0, 'a false answer is not re-asked past the true TTL');++ // The same holds through the sync peek the AI context uses.++ chatReads._reset();++ readState.firstChatCalls = 0;++ assert.strictEqual(chatReads.firstChatPeek(streamer, `user:${viewer}`), false);++ await sleep(30);++ chatReads._age(chatReads.CACHE_TTL_MS + 1000, 'fc:');++ readState.firstChatCalls = 0;++ assert.strictEqual(chatReads.firstChatPeek(streamer, `user:${viewer}`), false, 'the peek keeps a false answer past the true TTL');++ assert.strictEqual(readState.firstChatCalls, 0, 'the peek does not re-ask Chat for the cached false answer');++ d2.run('DELETE FROM stream_first_chats WHERE chatter_key = ?', [`user:${viewer}`]);++ chatReads._reset();++ readState.firstChatCalls = 0;++ await chatReads.firstChat(streamer, `user:${viewer}`);++ chatReads._age(chatReads.CACHE_TTL_MS + 1000, 'fc:');++ readState.firstChatCalls = 0;++ await chatReads.firstChat(streamer, `user:${viewer}`);++ assert.strictEqual(readState.firstChatCalls, 1, 'a true answer is re-asked after the short TTL');++ chatReads._reset();++ d2.run('INSERT INTO stream_first_chats (chatter_key, channel_user_id) VALUES (?, ?)', [`user:${viewer}`, streamer]);++ readState.down = true;++ assert.strictEqual(chatReads.firstChatPeek(streamer, `user:${viewer}`), false, 'a cold + down peek falls to Live\'s own table');++ readState.down = false;++ chatReads._reset();++++ // The AI context passes `user:<user_id>`, never the username: a row keyed by the numeric id++ // suppresses the welcome flag, where the old `user:<username>` key would have missed it.++ d2.run("INSERT INTO chat_messages (user_id, username, message, message_type, channel_user_id, timestamp) VALUES (?, 'viewer', 'hi there', 'chat', ?, ?)", [viewer, streamer, sqlNow()]);++ await chatReads.channelMessages(streamer, 40);++ const context = require('../server/ai/context');++ const ctxStream = db.getStreamById(streamId);++ const greet = { remember_viewers: true, greet_first_timers: true, max_open_threads: 3, hear_enabled: false };++ chatReads.invalidate('fc:');++ let tail = context.volatileTail({ userId: streamer, stream: ctxStream, settings: greet, sinceChatId: 0, botNames: new Set() });++ assert.ok(!/first time chatting here/.test(tail.text), 'a user with a recorded first chat is not greeted again');++ d2.run('DELETE FROM stream_first_chats WHERE chatter_key = ?', [`user:${viewer}`]);++ chatReads.invalidate('fc:');++ tail = context.volatileTail({ userId: streamer, stream: ctxStream, settings: greet, sinceChatId: 0, botNames: new Set() });++ assert.ok(/first time chatting here/.test(tail.text), 'a new user is flagged first time (user:<id> identity)');++ chatReads._reset();++++ // ── Sounds by command: Chat owns the row; a 404 is authoritative; a cold/down peek falls++ // to Live's own table. ──++ d2.run('DELETE FROM channel_sounds WHERE channel_owner_id = ?', [streamer]);++ d2.run("INSERT INTO channel_sounds (channel_owner_id, command, url, created_by) VALUES (?, 'honk', '/sounds/h.mp3', ?)", [streamer, streamer]);++ chatReads._reset();++ const honk = chatReads.soundByCommandPeek(streamer, 'honk');++ assert.ok(honk && honk.command === 'honk', 'Chat answers the sound by command');++ assert.strictEqual(chatReads.soundByCommandPeek(streamer, '!HONK').command, 'honk', 'the command is normalized');++ assert.strictEqual(await chatReads.soundByCommand(streamer, 'nope'), null, 'Chat: no such sound');++ d2.run("INSERT INTO channel_sounds (channel_owner_id, command, url, created_by) VALUES (?, 'nope', '/sounds/n.mp3', ?)", [streamer, streamer]);++ assert.strictEqual(chatReads.soundByCommandPeek(streamer, 'nope'), null, 'a Chat 404 is cached: Live\'s own row is not used');++ chatReads._reset();++ readState.down = true;++ assert.ok(chatReads.soundByCommandPeek(streamer, 'honk'), 'a cold + down peek falls back to Live\'s own table');++ readState.down = false;++ chatReads._reset();++ d2.run("INSERT INTO channel_sounds (channel_owner_id, command, url, created_by) VALUES (?, 'bare404', '/sounds/b.mp3', ?)", [streamer, streamer]);++ const bare = await chatReads.soundByCommand(streamer, 'bare404');++ assert.ok(bare && bare.command === 'bare404', 'a 404 without Chat\'s own body is not "no such sound": Live\'s table answers');++ chatReads._reset();++ }++// 12. Robot commands from chat pass the control panel's gate: control mode, anonymous switch, whitelist, cooldown.{const sent = [];
@@ -46,20 +46,23 @@ path are deleted.
@@ -46,20 +46,23 @@ path are deleted.
`server/net/service-guard.js`). Effects re-check the acting moderator/owner/admin and are refused
`server/net/service-guard.js`). Effects re-check the acting moderator/owner/admin and are refused
unless `CHAT_AUTHORITY=chat`.
unless `CHAT_AUTHORITY=chat`.
- Live reads chat stats, queues and history from Chat's internal read API —
- 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
+ `POST /internal/chat/stats` (`site`, `user`, `stream`, `channel-top` and `site-daily`),
+ `GET /internal/chat/messages`, `/timeline`, `/first-chat`, `/sounds`, `/sounds/by-command`,
+ `/moderation/pending-ip`, `/moderation/relay-users`, `/moderation/tts-override` — through
[server/chat/chat-reads.js](../server/chat/chat-reads.js), in place of the read mirror
[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
(`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.
-- **What still reads the mirror.** Under `CHAT_AUTHORITY=chat` the readers above answer from Chat;
- three still read Live's mirror tables and are why the mirror stays until T3 J4c drops it: the two
- HOME_SERIES charts `messages` and `active` (Chat exposes no site-wide per-day message/chatter
- series), the AI context's "first time chatting here" flag (`stream_first_chats`; Chat exposes no
- first-chat read), and the RobotStreamer `!sound` lookup (`channel_sounds` by command; Chat's sounds
- read answers only a count and the pending-asset list). The other mirror readers are on the
- Live-runs-chat path `CHAT_AUTHORITY=chat` leaves unmounted — `dm_*` (no `/api/dm`), Live's chat
- server, the deploy notice, the admin chat log and the leftover dev/rollback fallbacks.
+ `/api/mod` queues used to read. A Chat outage answers a cached value or Live's own table, never a 500.
+- **What reads the mirror.** Nothing in chat mode any more: the two HOME_SERIES charts `messages`
+ and `active` read Chat's site-daily series, the AI context's "first time chatting here" flag reads
+ Chat's first-chat read, and the RobotStreamer `!sound` lookup reads Chat's sounds-by-command read —
+ each through [server/chat/chat-reads.js](../server/chat/chat-reads.js), falling back to Live's own
+ table only while Chat is unreachable. The mirror (`POST /internal/chat-effects/mirror`) therefore
+ has no chat-mode reader left and is removed in the remaining step: Chat stops its `LIVE_MIRROR`
+ writer, then Live deletes the receiver (`live-context-routes.js` `effectsRouter`'s mirror route and
+ its `live.chat_mirror.write` capability) and the mirror-kept chat tables. The readers that still
+ answer from Live's tables are on the Live-runs-chat path `CHAT_AUTHORITY=chat` leaves unmounted —
+ `dm_*` (no `/api/dm`), Live's chat server, the deploy notice, the admin chat log and the leftover
+ dev/rollback fallbacks.
- **The six chat tables are Chat's** (roadmap T3): `channel_moderators`, `channel_moderation_settings`,
- **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
`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
only through [server/chat/moderation-client.js](../server/chat/moderation-client.js) — Chat's internal
@@ -203,11 +203,11 @@ function personBlock(line, channelUserId, settings) {
@@ -203,11 +203,11 @@ function personBlock(line, channelUserId, settings) {
else if (db.isFollowing(line.userId, channelUserId)) flags.push('follower');
else if (db.isFollowing(line.userId, channelUserId)) flags.push('follower');
}
}
}
}
- const key = line.userId ? `user:${line.username}` : (line.anonId ? `anon:${line.anonId}` : `ext:${line.username}`);
- // stream_first_chats is Chat's (a read-mirror table) and Chat's internal read API has no
- // first-chat route, so this stays on Live's copy — which Chat no longer fills for registered
- // users, making the flag best-effort. Drop it when Chat offers the read.
- if (line.userId !== channelUserId && settings.greet_first_timers && db.isFirstChatInChannel && db.isFirstChatInChannel(key, channelUserId)) flags.push('first time chatting here');
+ // stream_first_chats is Chat's; a registered identity is `user:<user_id>` (never the
+ // username), a relay one `ext:<prefixed username>`. The welcome flag reads Chat's first-chat
+ // read through a sync peek that falls back to Live's own mirror-kept table while Chat is down.
+ const identity = line.userId ? `user:${line.userId}` : (line.anonId ? `anon:${line.anonId}` : `ext:${line.username}`);
+ if (line.userId !== channelUserId && settings.greet_first_timers && chatReads.firstChatPeek(channelUserId, identity)) flags.push('first time chatting here');
} catch { /* */ }
} catch { /* */ }
const insight = ins ? clip([ins.overview_24h || ins.overview_alltime, ins.memory].filter(Boolean).join(' — '), 220) : '';
const insight = ins ? clip([ins.overview_24h || ins.overview_alltime, ins.memory].filter(Boolean).join(' — '), 220) : '';
if (!insight && !flags.length) return '';
if (!insight && !flags.length) return '';
@@ -14,11 +14,13 @@
@@ -14,11 +14,13 @@
* readStats(body) POST /internal/chat/stats chat.stats.read counts, or top chatters
* 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
* 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
* readTimeline(q) GET /internal/chat/timeline chat.analysis.read time-bucketed message counts
+ * readFirstChat(q) GET /internal/chat/first-chat chat.analysis.read has this identity ever chatted here
* readPendingIp(q) GET /internal/chat/moderation/pending-ip chat.moderation.queue.read held IP-approval messages
* 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
* 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
* 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
* 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
* readSounds(q) GET /internal/chat/sounds chat.sounds.read a channel's sound count, or its pending assets
+ * readSoundByCommand(q) GET /internal/chat/sounds/by-command chat.sounds.read the approved sound a !command plays
*
*
* Auth: Live's Network service principal (client_credentials, audience openvibe.chat, server/net/network-principal.js).
* 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
* Every POST carries one idempotency `key` made when the operation is created and reused on every retry, so Chat
@@ -167,11 +169,28 @@ async function readStats(body) {
@@ -167,11 +169,28 @@ async function readStats(body) {
const readMessages = (params) => read('messages', `/internal/chat/messages${qs(params)}`);
const readMessages = (params) => read('messages', `/internal/chat/messages${qs(params)}`);
const readTimeline = (params) => read('timeline', `/internal/chat/timeline${qs(params)}`);
const readTimeline = (params) => read('timeline', `/internal/chat/timeline${qs(params)}`);
+const readFirstChat = (params) => read('first-chat', `/internal/chat/first-chat${qs(params)}`);
const readPendingIp = (params) => read('pending-ip', `/internal/chat/moderation/pending-ip${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 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 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 readTtsOverride = (params) => read('tts-override', `/internal/chat/moderation/tts-override${qs(params)}`);
const readSounds = (params) => read('sounds', `/internal/chat/sounds${qs(params)}`);
const readSounds = (params) => read('sounds', `/internal/chat/sounds${qs(params)}`);
+/**
+ * GET /internal/chat/sounds/by-command — the approved sound a !command plays. Chat answers its own 404 body
+ * `{ ok: false, error: 'Sound not found' }` for a definitive "no such sound", surfaced as `{ sound: null }`; any
+ * other 404 (an older Chat without the route, a proxy page) and every other failure answer null, so the caller
+ * falls back to Live's table rather than silencing every !sound.
+ */
+async function readSoundByCommand(params) {
+ if (drill() || !principal.configured()) return null;
+ try {
+ const { status, data } = await request('GET', `/internal/chat/sounds/by-command${qs(params)}`);
+ if (status === 200 && data && data.ok !== false && data.sound) return data;
+ if (status === 404 && data && data.ok === false && data.error === 'Sound not found') return { sound: null };
+ note(`sound-by-command: Chat answered ${status}${data && data.error ? ` (${data.error})` : ''}`);
+ } catch (err) { note(`sound-by-command: ${err.message}`); }
+ return null;
+}
/** Record where Live's asset sync uploaded a channel sound (idempotent; Chat applies the same write once). */
/** Record where Live's asset sync uploaded a channel sound (idempotent; Chat applies the same write once). */
async function soundAsset(body) {
async function soundAsset(body) {
if (drill() || !principal.configured()) return null;
if (drill() || !principal.configured()) return null;
@@ -193,6 +212,7 @@ module.exports = {
@@ -193,6 +212,7 @@ module.exports = {
moderation: (body) => send('moderation', body),
moderation: (body) => send('moderation', body),
invalidate: (body) => send('invalidate', body),
invalidate: (body) => send('invalidate', body),
presence,
presence,
- readStats, readMessages, readTimeline, readPendingIp, readRelayUsers, readRelayUser, readTtsOverride, readSounds, soundAsset,
+ readStats, readMessages, readTimeline, readFirstChat, readPendingIp, readRelayUsers, readRelayUser, readTtsOverride,
+ readSounds, readSoundByCommand, soundAsset,
_setRetryMs(ms) { retryMs = ms || RETRY_MS; },
_setRetryMs(ms) { retryMs = ms || RETRY_MS; },
};
};
@@ -65,7 +65,8 @@ function unsupported(message, status = 501) {
@@ -65,7 +65,8 @@ function unsupported(message, status = 501) {
return err;
return err;
}
}
-function fresh(key) { const h = cache.get(key); return h && !h.failed && Date.now() - h.at < CACHE_TTL_MS ? h : null; }
+/** A fresh entry: within its own ttl when it set one (first-chat caches false longer than true), else the default. */
+function fresh(key) { const h = cache.get(key); if (!h || h.failed) return null; const ttl = h.ttl != null ? h.ttl : CACHE_TTL_MS; return Date.now() - h.at < ttl ? h : null; }
function kept(key) { const h = cache.get(key); return h && !h.failed && Date.now() - h.at < CACHE_KEEP_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; }
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. */
/** Keep the cache bounded: drop entries past the keep window first, then the oldest. */
@@ -74,8 +75,10 @@ function evict() {
@@ -74,8 +75,10 @@ function evict() {
for (const [k, h] of cache) if (Date.now() - h.at >= CACHE_KEEP_MS) cache.delete(k);
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);
while (cache.size > cacheMax) cache.delete(cache.keys().next().value);
}
}
-function remember(key, value) {
- cache.set(key, { at: Date.now(), value });
+function remember(key, value, ttl) {
+ const h = { at: Date.now(), value };
+ if (ttl != null) h.ttl = ttl;
+ cache.set(key, h);
evict();
evict();
}
}
@@ -95,7 +98,7 @@ function rememberFailure(key) {
@@ -95,7 +98,7 @@ function rememberFailure(key) {
* console). `ttl: 0` always reads live and never caches. An error carrying a `status` (a
* 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.
* deliberate "not supported") is handed to the caller as it is.
*/
*/
-function load(key, remoteFn, localFn, { ttl = CACHE_TTL_MS, strict = false } = {}) {
+function load(key, remoteFn, localFn, { ttl = CACHE_TTL_MS, strict = false, ttlFor = null } = {}) {
if (!remote()) {
if (!remote()) {
return Promise.resolve().then(async () => { try { return await localFn(); } catch (err) { note(key, err); return null; } });
return Promise.resolve().then(async () => { try { return await localFn(); } catch (err) { note(key, err); return null; } });
}
}
@@ -122,7 +125,7 @@ function load(key, remoteFn, localFn, { ttl = CACHE_TTL_MS, strict = false } = {
@@ -122,7 +125,7 @@ function load(key, remoteFn, localFn, { ttl = CACHE_TTL_MS, strict = false } = {
return last ? last.value : null;
return last ? last.value : null;
}
}
lastLog = null;
lastLog = null;
- if (ttl > 0) remember(key, value);
+ if (ttl > 0) remember(key, value, ttlFor ? ttlFor(value) : undefined);
return value;
return value;
})().finally(() => inflight.delete(key));
})().finally(() => inflight.delete(key));
if (ttl > 0) inflight.set(key, p);
if (ttl > 0) inflight.set(key, p);
@@ -130,11 +133,11 @@ function load(key, remoteFn, localFn, { ttl = CACHE_TTL_MS, strict = false } = {
@@ -130,11 +133,11 @@ function load(key, remoteFn, localFn, { ttl = CACHE_TTL_MS, strict = false } = {
}
}
/** Sync answer for a caller that cannot await: fresh value, else the last good value + a background refresh, else null. */
/** 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) {
+function peek(key, remoteFn, localFn, opts) {
if (!remote()) { try { return localFn(); } catch (err) { note(key, err); return null; } }
if (!remote()) { try { return localFn(); } catch (err) { note(key, err); return null; } }
const hit = fresh(key);
const hit = fresh(key);
if (hit) return hit.value;
if (hit) return hit.value;
- load(key, remoteFn, localFn).catch(() => { /* logged in load() */ }); // floating-ok: refreshes for the next peek
+ load(key, remoteFn, localFn, opts).catch(() => { /* logged in load() */ }); // floating-ok: refreshes for the next peek
const last = kept(key);
const last = kept(key);
return last ? last.value : null;
return last ? last.value : null;
}
}
@@ -166,6 +169,60 @@ function localSiteStatsMessages() {
@@ -166,6 +169,60 @@ function localSiteStatsMessages() {
}
}
function siteStatsPeek() { return peek('site', remoteSiteStats, localSiteStatsMessages); }
function siteStatsPeek() { return peek('site', remoteSiteStats, localSiteStatsMessages); }
+// ── Home daily series (the hero charts) ──────────────────────────────────────
+// Live's home `messages`/`active` charts are Chat's site-wide per-day series. One peek serves the
+// whole envelope getHomeStatSeries returns: the per-day counts come from Chat's site-daily read, and
+// `before`/`prev_total` keep Live's window-wide semantics (a COUNT / COUNT(DISTINCT) over the whole
+// range), so they are Chat's `site` stats over [epoch, window start) and over the window just before
+// it. The peek key is coarse (per UTC day) so every compute within a day hits the same cached answer.
+const DAY_MS = 86_400_000;
+const utcDayStart = (t) => Math.floor(t / DAY_MS) * DAY_MS;
+const clampSeriesDays = (days) => Math.max(1, Math.min(365, parseInt(days, 10) || 30));
+
+async function remoteHomeSeries(metric, days) {
+ const until = utcDayStart(Date.now()) + DAY_MS; // exclusive: tomorrow 00:00 UTC
+ const since = until - days * DAY_MS;
+ const out = await client.readStats({ kind: 'site-daily', since, until });
+ if (!out || !Array.isArray(out.days)) throw unavailable('home series');
+ const field = metric === 'active' ? 'chatters' : 'messages';
+ const byDay = new Map(out.days.map((d) => [String(d.day), Number(d[field]) || 0]));
+ const points = [];
+ for (let i = days - 1; i >= 0; i--) {
+ const day = new Date(until - (i + 1) * DAY_MS).toISOString().slice(0, 10);
+ points.push({ day, value: Number((byDay.get(day) || 0).toFixed(2)) });
+ }
+ const total = Number(points.reduce((a, p) => a + p.value, 0).toFixed(2));
+ const [beforeOut, prevOut] = await Promise.all([
+ client.readStats({ kind: 'site', until: since }),
+ client.readStats({ kind: 'site', since: since - days * DAY_MS, until: since }),
+ ]);
+ if (!beforeOut || !prevOut) throw unavailable('home series totals');
+ const pick = (o) => Number(metric === 'active' ? o.chatters : o.messages) || 0;
+ return { metric, kind: 'count', days, points, total, peak: Math.max(0, ...points.map((p) => p.value)), before: pick(beforeOut), prev_total: pick(prevOut) };
+}
+
+/** The daily series behind a hero stat, from Chat (remote) or Live's own tables (local), cached. */
+function homeSeries(metric, days) {
+ const d = clampSeriesDays(days);
+ const key = `hs:${metric}:${d}:${utcDayStart(Date.now())}`;
+ const localFn = () => local().homeSeriesLocal(metric, d);
+ return load(key, () => remoteHomeSeries(metric, d), localFn).then((v) => v || safeLocal(localFn, null));
+}
+/**
+ * The same for the synchronous home-stats series route: the last good answer (or, on a cold cache,
+ * Live's mirror-filled tables) while Chat refreshes in the background — never zeros just because
+ * Chat has not answered yet.
+ */
+function homeSeriesPeek(metric, days) {
+ const d = clampSeriesDays(days);
+ const key = `hs:${metric}:${d}:${utcDayStart(Date.now())}`;
+ const localFn = () => local().homeSeriesLocal(metric, d);
+ const v = peek(key, () => remoteHomeSeries(metric, d), localFn);
+ if (v) return v;
+ if (!remote()) return null; // the peek already answered from Live's own tables
+ return safeLocal(localFn, null); // cold / Chat unreachable: Live's mirror-filled tables
+}
+
/** The busiest chatters of a room (or the site, when neither stream nor channel is given), newest window first. */
/** 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 } = {}) {
function topChatters({ since, streamId, channelUserId, limit = 10 } = {}) {
const key = `top:${since || 0}:${streamId || 0}:${channelUserId || 0}:${limit}`;
const key = `top:${since || 0}:${streamId || 0}:${channelUserId || 0}:${limit}`;
@@ -487,6 +544,44 @@ function spikeOffsets(streamId, bucketSec = 30, topN = 8, sinceMs) {
@@ -487,6 +544,44 @@ function spikeOffsets(streamId, bucketSec = 30, topN = 8, sinceMs) {
}, () => local().getChatSpikeOffsets(id, bucketSec, topN) || []);
}, () => local().getChatSpikeOffsets(id, bucketSec, topN) || []);
}
}
+// ── First chat (the welcome check) ───────────────────────────────────────────
+// Chat owns stream_first_chats; the AI context's "first time chatting here" flag asks Chat's
+// first-chat read. A false answer (they have chatted here) is stable and cached longer; a true
+// answer flips the moment they chat, so it is cached only briefly. The sync caller uses
+// firstChatPeek; on a cold cache or a Chat outage it answers Live's own (mirror-kept) table.
+const FIRST_CHAT_TTL_FALSE = 5 * 60_000; // has chatted here: stable
+const FIRST_CHAT_TTL_TRUE = CACHE_TTL_MS; // first time: flips as soon as they chat
+
+function firstChatLocal(identity, channelUserId) {
+ return safeLocal(() => local().isFirstChatInChannel(identity, channelUserId), false);
+}
+function firstChatRemote(channelUserId, identity) {
+ return async () => {
+ const out = await client.readFirstChat({ channel_id: channelUserId, identity });
+ if (!out || typeof out.first !== 'boolean') throw unavailable('first-chat');
+ return out.first;
+ };
+}
+/** Has this identity (`user:<id>` | `anon:<anonId>` | `ext:<prefixed username>`) ever chatted here? */
+function firstChat(channelUserId, identity) {
+ const channelId = Number(channelUserId) || 0;
+ const key = String(identity || '');
+ if (!channelId || !key) return Promise.resolve(false);
+ return load(`fc:${channelId}:${key}`, firstChatRemote(channelId, key), () => firstChatLocal(key, channelId), {
+ ttl: FIRST_CHAT_TTL_TRUE, ttlFor: (v) => (v === false ? FIRST_CHAT_TTL_FALSE : FIRST_CHAT_TTL_TRUE),
+ }).then((v) => (v == null ? firstChatLocal(key, channelId) : v === true));
+}
+/** The synchronous form (the AI persona prompt): the cached answer, else Live's own table. */
+function firstChatPeek(channelUserId, identity) {
+ const channelId = Number(channelUserId) || 0;
+ const key = String(identity || '');
+ if (!channelId || !key) return false;
+ const v = peek(`fc:${channelId}:${key}`, firstChatRemote(channelId, key), () => firstChatLocal(key, channelId), {
+ ttl: FIRST_CHAT_TTL_TRUE, ttlFor: (x) => (x === false ? FIRST_CHAT_TTL_FALSE : FIRST_CHAT_TTL_TRUE),
+ });
+ return (v === true || v === false) ? v : firstChatLocal(key, channelId);
+}
+
// ── Moderation queues ─────────────────────────────────────────────────────────
// ── Moderation queues ─────────────────────────────────────────────────────────
// These answer the console's own actions, so they are live (ttl: 0) and strict: a Chat failure is
// 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).
// a 503, never a stale or empty queue (it could hide a ban the moderator just set).
@@ -605,6 +700,37 @@ function soundCount(ownerId) {
@@ -605,6 +700,37 @@ function soundCount(ownerId) {
});
});
}
}
+const normalizeCommand = (command) => String(command || '').trim().toLowerCase().replace(/^!+/, '');
+async function remoteSoundByCommand(ownerId, command) {
+ const out = await client.readSoundByCommand({ channel_id: ownerId, command });
+ if (out == null) throw unavailable('sound by command'); // unreachable: keep / fall back
+ return out.sound || false; // the sound, or Chat's definitive none
+}
+/** The approved sound a !command plays (or null); Chat owns the row, Live's table is the fallback. */
+function soundByCommand(ownerId, command) {
+ const id = Number(ownerId) || 0;
+ const cmd = normalizeCommand(command);
+ if (!id || !cmd) return Promise.resolve(null);
+ const localFn = () => local().getChannelSoundByCommand(id, cmd) || null;
+ return load(`sbc:${id}:${cmd}`, () => remoteSoundByCommand(id, cmd), localFn).then((v) => {
+ if (v === false) return null; // Chat answered: no such sound
+ if (v) return v;
+ return safeLocal(localFn, null); // cold / outage: Live's own table
+ });
+}
+/** The synchronous form (the RobotStreamer !sound lookup): cached answer, else Live's own table. */
+function soundByCommandPeek(ownerId, command) {
+ const id = Number(ownerId) || 0;
+ const cmd = normalizeCommand(command);
+ if (!id || !cmd) return null;
+ const localFn = () => local().getChannelSoundByCommand(id, cmd) || null;
+ const v = peek(`sbc:${id}:${cmd}`, () => remoteSoundByCommand(id, cmd), localFn);
+ if (v) return v;
+ if (v === false) return null; // Chat answered: no such sound
+ if (!remote()) return null; // the peek already answered from Live's own table
+ return safeLocal(localFn, null); // cold / Chat unreachable: Live's mirror-filled table
+}
+
/**
/**
* Channel sounds not yet uploaded to Media (asset-sync's work list), oldest first, paged by id.
* 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.
* Never cached: the sync must see the upload it just recorded, or it uploads the same sound twice.
@@ -652,11 +778,11 @@ function _size() { return cache.size; }
@@ -652,11 +778,11 @@ function _size() { return cache.size; }
function _cacheMax(n) { if (n != null && Number(n) > 0) cacheMax = Number(n); return cacheMax; }
function _cacheMax(n) { if (n != null && Number(n) > 0) cacheMax = Number(n); return cacheMax; }
module.exports = {
module.exports = {
- siteStats, siteStatsPeek, topChatters,
+ siteStats, siteStatsPeek, topChatters, homeSeries, homeSeriesPeek,
windowStats, windowStatsPeek, streamStats, streamStatsPeek, channelMaxId, channelMaxIdPeek, userMessageCountPeek,
windowStats, windowStatsPeek, streamStats, streamStatsPeek, channelMaxId, channelMaxIdPeek, userMessageCountPeek,
messageById, channelMessages, channelMessagesPeek, userHistory, relayHistory, channelSamples, searchMessages,
messageById, channelMessages, channelMessagesPeek, userHistory, relayHistory, channelSamples, searchMessages,
- liveChatBuckets, recentChatText, spikeOffsets,
+ liveChatBuckets, recentChatText, spikeOffsets, firstChat, firstChatPeek,
pendingIp, relayUsers, relayUser, ttsOverride, ttsOverridePeek, isRelayUserHidden,
pendingIp, relayUsers, relayUser, ttsOverride, ttsOverridePeek, isRelayUserHidden,
- soundCount, pendingSounds, recordSoundAsset, invalidate,
+ soundCount, soundByCommand, soundByCommandPeek, pendingSounds, recordSoundAsset, invalidate,
_reset, _age, _size, _cacheMax, CACHE_TTL_MS, CACHE_KEEP_MS,
_reset, _age, _size, _cacheMax, CACHE_TTL_MS, CACHE_KEEP_MS,
};
};
@@ -3805,9 +3805,10 @@ const HOME_SERIES = {
@@ -3805,9 +3805,10 @@ const HOME_SERIES = {
anons: { table: 'anon_ip_mappings', ts: 'created_at', agg: 'COUNT(*)' },
anons: { table: 'anon_ip_mappings', ts: 'created_at', agg: 'COUNT(*)' },
visitors: { table: 'anon_ip_mappings', ts: 'created_at', agg: 'COUNT(*)' },
visitors: { table: 'anon_ip_mappings', ts: 'created_at', agg: 'COUNT(*)' },
follows: { table: 'follows', ts: 'created_at', agg: 'COUNT(*)' },
follows: { table: 'follows', ts: 'created_at', agg: 'COUNT(*)' },
- // OpenVibe.Chat exposes no site-wide per-day message/chatter series — its timeline read buckets
- // one channel or stream only, never the whole site — so these two charts still read Live's
- // mirror-filled chat_messages table. Every other series and the hero counters come from Chat.
+ // Chat's site-wide per-day message/chatter series: in chat mode getHomeStatSeries answers them
+ // from Chat's `site-daily` stats read (chat-reads.homeSeriesPeek); this local SQL stays the
+ // dev / rollback / Chat-unreachable answer. `messages` maps to Chat's messages series, `active`
+ // to its chatters series.
messages: { table: 'chat_messages', ts: 'timestamp', agg: 'COUNT(*)', where: 'COALESCE(is_deleted, 0) = 0' },
messages: { table: 'chat_messages', ts: 'timestamp', agg: 'COUNT(*)', where: 'COALESCE(is_deleted, 0) = 0' },
active: { table: 'chat_messages', ts: 'timestamp', agg: "COUNT(DISTINCT COALESCE('u' || user_id, 'a' || anon_id, 'r' || source_platform || '|' || username))", where: 'COALESCE(is_deleted, 0) = 0' },
active: { table: 'chat_messages', ts: 'timestamp', agg: "COUNT(DISTINCT COALESCE('u' || user_id, 'a' || anon_id, 'r' || source_platform || '|' || username))", where: 'COALESCE(is_deleted, 0) = 0' },
sessions: { table: 'streams', ts: 'created_at', agg: 'COUNT(*)' },
sessions: { table: 'streams', ts: 'created_at', agg: 'COUNT(*)' },
@@ -3827,7 +3828,11 @@ const HOME_SERIES = {
@@ -3827,7 +3828,11 @@ const HOME_SERIES = {
};
};
const HOME_SERIES_KEYS = [...Object.keys(HOME_SERIES), 'liveNow', 'viewersNow'];
const HOME_SERIES_KEYS = [...Object.keys(HOME_SERIES), 'liveNow', 'viewersNow'];
-function getHomeStatSeries(metric, days = 30) {
+/**
+ * The local (Live-table) daily series for one metric — the dev, rollback and Chat-unreachable
+ * answer. getHomeStatSeries routes the two chat metrics through Chat first.
+ */
+function homeSeriesLocal(metric, days = 30) {
const def = HOME_SERIES[metric];
const def = HOME_SERIES[metric];
if (!def) return null;
if (!def) return null;
days = Math.max(1, Math.min(365, parseInt(days, 10) || 30));
days = Math.max(1, Math.min(365, parseInt(days, 10) || 30));
@@ -3857,6 +3862,22 @@ function getHomeStatSeries(metric, days = 30) {
@@ -3857,6 +3862,22 @@ function getHomeStatSeries(metric, days = 30) {
before: Number(before.toFixed(2)), prev_total: Number(prevTotal.toFixed(2)) };
before: Number(before.toFixed(2)), prev_total: Number(prevTotal.toFixed(2)) };
}
}
+/**
+ * The daily series behind a hero stat. With CHAT_AUTHORITY=chat the two chat metrics (`messages`,
+ * `active`) answer from Chat's site-wide per-day read (chat-reads.homeSeriesPeek — a cached
+ * stale-while-revalidate peek that falls back to Live's own tables when Chat is unreachable);
+ * every other metric is Live's own.
+ */
+function getHomeStatSeries(metric, days = 30) {
+ if (metric === 'messages' || metric === 'active') {
+ try {
+ const s = require('../chat/chat-reads').homeSeriesPeek(metric, days);
+ if (s) return s;
+ } catch { /* fall through to Live's own tables */ }
+ }
+ return homeSeriesLocal(metric, days);
+}
+
/**
/**
* The two live readings (streams live, people watching) over time, from the five-minute sampler.
* The two live readings (streams live, people watching) over time, from the five-minute sampler.
* Unlike counters they do not add up across a day, so each bucket carries the average reading
* Unlike counters they do not add up across a day, so each bucket carries the average reading
@@ -6995,7 +7016,7 @@ module.exports = {
@@ -6995,7 +7016,7 @@ module.exports = {
adoptOrphanedTimelineRows: () => _adoptOrphanedTimelineRows(getDb()),
adoptOrphanedTimelineRows: () => _adoptOrphanedTimelineRows(getDb()),
publicStream,
publicStream,
getConcurrencyBaseline,
getConcurrencyBaseline,
- getHomeStatSeries, HOME_SERIES_KEYS, vibesStatsSince, _computeHomeStats,
+ getHomeStatSeries, homeSeriesLocal, HOME_SERIES_KEYS, vibesStatsSince, _computeHomeStats,
getVodAiState, getClipAiState, forgetMediaItem,
getVodAiState, getClipAiState, forgetMediaItem,
scheduleClipNotifyState, bumpClipNotifyNowState, markClipNotifiedState, getDueClipNotifies,
scheduleClipNotifyState, bumpClipNotifyNowState, markClipNotifiedState, getDueClipNotifies,
getDb, initDb, run, get, all, close, announceModerationAction: _announceModerationAction,
getDb, initDb, run, get, all, close, announceModerationAction: _announceModerationAction,
@@ -677,9 +677,9 @@ class RobotStreamerService {
@@ -677,9 +677,9 @@ class RobotStreamerService {
if (trimmed.startsWith('!')) {
if (trimmed.startsWith('!')) {
const parts = trimmed.split(/\s+/);
const parts = trimmed.split(/\s+/);
const scmd = parts[0].slice(1).toLowerCase();
const scmd = parts[0].slice(1).toLowerCase();
- // Chat owns channel_sounds but its internal read API offers only a count and the
- // pending-asset list, no lookup by command, so this stays on Live's mirror table.
- if (scmd && db.getChannelSoundByCommand(stream.user_id, scmd)) {
+ // Chat owns channel_sounds; ask its by-command read (a cached peek), which falls
+ // back to Live's own table while Chat is unreachable.
+ if (scmd && chatReads.soundByCommandPeek(stream.user_id, scmd)) {
if (delivery.ingress()) {
if (delivery.ingress()) {
delivery.event({ kind: 'stream', id: stream.id }, {
delivery.event({ kind: 'stream', id: stream.id }, {
type: 'channel-sound', streamId: stream.id, command: scmd, args: parts.slice(1).map((a) => a.slice(0, 120)).slice(0, 20),
type: 'channel-sound', streamId: stream.id, command: scmd, args: parts.slice(1).map((a) => a.slice(0, 120)).slice(0, 20),
@@ -56,13 +56,14 @@ const network = http.createServer((req, res) => {
@@ -56,13 +56,14 @@ const network = http.createServer((req, res) => {
// Chat's internal read API (roadmap T3 J4b): Live reads stats, queues, history and sounds through
// 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
// 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.
// `readState.down` fails every read the way a Chat outage does.
-const readState = { down: false, soundAssetDown: false, relayOnly: [], msgById: 0 };
+const readState = { down: false, soundAssetDown: false, relayOnly: [], msgById: 0, firstChatCalls: 0 };
function readReply(req, res, raw) {
function readReply(req, res, raw) {
const url = String(req.url);
const url = String(req.url);
const path = url.split('?')[0];
const path = url.split('?')[0];
const q = new URLSearchParams(url.split('?')[1] || '');
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';
+ const isRead = path === '/internal/chat/stats' || path === '/internal/chat/messages' || path === '/internal/chat/first-chat'
+ || path.startsWith('/internal/chat/moderation/') || path === '/internal/chat/sounds'
+ || path === '/internal/chat/sounds/by-command' || path === '/internal/chat/sounds/asset';
if (!isRead) return false;
if (!isRead) return false;
if (readState.down) { res.statusCode = 503; res.end(JSON.stringify({ ok: false, error: 'Chat read unavailable' })); return true; }
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; }
if (path === '/internal/chat/sounds/asset' && readState.soundAssetDown) { res.statusCode = 503; res.end(JSON.stringify({ ok: false, error: 'Chat sound write unavailable' })); return true; }
@@ -71,12 +72,39 @@ function readReply(req, res, raw) {
@@ -71,12 +72,39 @@ function readReply(req, res, raw) {
const param = (col, v) => (['channel_user_id', 'stream_id', 'user_id', 'id'].includes(col) ? Number(v) : String(v));
const param = (col, v) => (['channel_user_id', 'stream_id', 'user_id', 'id'].includes(col) ? Number(v) : String(v));
if (path === '/internal/chat/stats') {
if (path === '/internal/chat/stats') {
const b = JSON.parse(raw || '{}');
const b = JSON.parse(raw || '{}');
+ const sqlTime = (t) => new Date(t).toISOString().slice(0, 19).replace('T', ' ');
+ const CHATTER = "COALESCE('u:' || user_id, 'a:' || anon_id, source_platform || ':' || username)";
if (b.kind === 'channel-top') {
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]);
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) })) });
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 (b.kind === 'site-daily') {
+ const DAY = 86400000;
+ const dayStart = (t) => Math.floor(t / DAY) * DAY;
+ const rows = d2.all(`SELECT substr(timestamp, 1, 10) AS day, COUNT(*) AS messages, COUNT(DISTINCT ${CHATTER}) AS chatters
+ FROM chat_messages WHERE COALESCE(is_deleted, 0) = 0 AND timestamp >= ? AND timestamp < ? GROUP BY day`,
+ [sqlTime(b.since), sqlTime(b.until)]);
+ const by = new Map(rows.map((r) => [String(r.day), r]));
+ const days = [];
+ for (let t = dayStart(b.since), last = dayStart(b.until - 1); t <= last; t += DAY) {
+ const day = new Date(t).toISOString().slice(0, 10);
+ const r = by.get(day);
+ days.push({ day, messages: r ? Number(r.messages) : 0, chatters: r ? Number(r.chatters) : 0 });
+ }
+ return ok({ days });
+ }
+ const where = ['is_deleted = 0'];
+ const params = [];
+ if (b.since != null) { where.push('timestamp >= ?'); params.push(sqlTime(b.since)); }
+ if (b.until != null) { where.push('timestamp < ?'); params.push(sqlTime(b.until)); }
+ if (b.kind === 'user' && b.user_id != null) { where.push('user_id = ?'); params.push(Number(b.user_id)); }
+ const r = d2.get(`SELECT COUNT(*) AS messages, COUNT(DISTINCT ${CHATTER}) AS chatters FROM chat_messages WHERE ${where.join(' AND ')}`, params);
+ return ok({ messages: Number(r.messages), chatters: Number(r.chatters) });
+ }
+ if (path === '/internal/chat/first-chat') {
+ readState.firstChatCalls++;
+ const r = d2.get('SELECT 1 AS present FROM stream_first_chats WHERE chatter_key = ? AND channel_user_id = ?', [String(q.get('identity')), Number(q.get('channel_id'))]);
+ return ok({ first: !r });
}
}
if (path === '/internal/chat/messages') {
if (path === '/internal/chat/messages') {
if (q.get('id') != null) readState.msgById++;
if (q.get('id') != null) readState.msgById++;
@@ -114,6 +142,14 @@ function readReply(req, res, raw) {
@@ -114,6 +142,14 @@ function readReply(req, res, raw) {
d2.run('UPDATE channel_sounds SET media_url = ?, media_asset_id = ? WHERE id = ?', [b.media_url, b.media_asset_id, b.id]);
d2.run('UPDATE channel_sounds SET media_url = ?, media_asset_id = ? WHERE id = ?', [b.media_url, b.media_asset_id, b.id]);
return ok({});
return ok({});
}
}
+ if (path === '/internal/chat/sounds/by-command') {
+ const cmd = String(q.get('command') || '').trim().toLowerCase().replace(/^!+/, '');
+ // A 404 that is not Chat's own "no such sound" (an older Chat, a proxy page).
+ if (cmd === 'bare404') { res.statusCode = 404; res.setHeader('content-type', 'text/html'); res.end('<html>Not Found</html>'); return true; }
+ const r = d2.get('SELECT * FROM channel_sounds WHERE channel_owner_id = ? AND command = ? AND is_approved = 1 ORDER BY RANDOM() LIMIT 1', [Number(q.get('channel_id')), cmd]);
+ if (!r) { res.statusCode = 404; res.end(JSON.stringify({ ok: false, error: 'Sound not found' })); return true; }
+ return ok({ sound: r });
+ }
if (path === '/internal/chat/sounds') {
if (path === '/internal/chat/sounds') {
if (q.get('pending_asset') === '1') {
if (q.get('pending_asset') === '1') {
const where = ['media_asset_id IS NULL'];
const where = ['media_asset_id IS NULL'];
@@ -134,7 +170,7 @@ const chat = http.createServer((req, res) => {
@@ -134,7 +170,7 @@ const chat = http.createServer((req, res) => {
req.on('data', (c) => { raw += c; });
req.on('data', (c) => { raw += c; });
req.on('end', () => {
req.on('end', () => {
res.setHeader('Content-Type', 'application/json');
res.setHeader('Content-Type', 'application/json');
- assert.ok(String(req.headers.authorization || '').startsWith('Bearer '), 'Live calls Chat with a service token');
+ assert.ok(!String(req.url).startsWith('/internal/') || String(req.headers.authorization || '').startsWith('Bearer '), 'Live calls Chat with a service token');
assert.ok(!String(req.url).startsWith('/internal/live/'), 'nothing goes over the old bridge');
assert.ok(!String(req.url).startsWith('/internal/live/'), 'nothing goes over the old bridge');
const im = String(req.url).match(/^\/internal\/chat\/(messages|events|moderation|invalidate)$/);
const im = String(req.url).match(/^\/internal\/chat\/(messages|events|moderation|invalidate)$/);
if (im) {
if (im) {
@@ -756,6 +792,127 @@ const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
@@ -756,6 +792,127 @@ const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
}
}
}
}
+ // 11b. The last three mirror readers read Chat now (roadmap T3 J4c): the home daily series,
+ // the AI context's first-chat flag and the RobotStreamer !sound lookup.
+ {
+ const chatReads = require('../server/chat/chat-reads');
+ const d2 = require('../server/db/database');
+ const DAY = 86400000;
+ const sqlNow = () => new Date().toISOString().slice(0, 19).replace('T', ' ');
+
+ // ── Home daily series: Chat's site-daily read, mapped per metric; a cold cache during a
+ // Chat outage answers Live's own series (never zeros). ──
+ d2.run('DELETE FROM chat_messages');
+ chatReads._reset();
+ const atDay = (back) => new Date(Date.now() - back * DAY).toISOString().slice(0, 10) + ' 12:00:00';
+ d2.run("INSERT INTO chat_messages (user_id, username, message, message_type, timestamp) VALUES (?, 'viewer', 'a', 'chat', ?), (?, 'viewer', 'b', 'chat', ?), (?, 'moddy', 'c', 'chat', ?)",
+ [viewer, atDay(2), viewer, atDay(2), mod, atDay(2)]);
+ d2.run("INSERT INTO chat_messages (user_id, username, message, message_type, timestamp) VALUES (?, 'viewer', 'd', 'chat', ?), (?, 'moddy', 'e', 'chat', ?)",
+ [viewer, atDay(1), mod, atDay(1)]);
+ const series = chatReads.homeSeriesPeek('messages', 7);
+ assert.ok(series && Array.isArray(series.points), 'the chat series peek answers in chat mode');
+ assert.strictEqual(series.days, 7);
+ assert.strictEqual(series.points.length, 7, 'one zero-filled point per day');
+ assert.strictEqual(series.points[4].value, 3, 'two days ago: three messages (Chat)');
+ assert.strictEqual(series.points[5].value, 2, 'yesterday: two messages (Chat)');
+ assert.strictEqual(series.points[6].value, 0, 'today: none');
+ assert.strictEqual(series.total, 5);
+ assert.strictEqual(series.prev_total, 0, 'the window before the 7 days is empty');
+ assert.strictEqual(series.before, 0, 'nothing before the window');
+ const activeSeries = chatReads.homeSeriesPeek('active', 7);
+ assert.strictEqual(activeSeries.points[4].value, 2, 'two days ago: two distinct chatters');
+ assert.strictEqual(activeSeries.points[5].value, 2, 'yesterday: two distinct chatters');
+ // getHomeStatSeries routes the two chat metrics through the chat series.
+ const realSeriesPeek = chatReads.homeSeriesPeek;
+ chatReads.homeSeriesPeek = () => ({ metric: 'messages', sentinel: true });
+ try { assert.strictEqual(d2.getHomeStatSeries('messages', 7).sentinel, true, 'getHomeStatSeries reads the chat series in chat mode'); }
+ finally { chatReads.homeSeriesPeek = realSeriesPeek; }
+ // A cold cache during a Chat outage answers Live's own table, not zeros.
+ chatReads._reset();
+ readState.down = true;
+ const fallbackSeries = chatReads.homeSeriesPeek('messages', 7);
+ assert.ok(fallbackSeries && fallbackSeries.points.length === 7, 'a cold + down peek answers Live\'s own series');
+ assert.strictEqual(fallbackSeries.total, 5, 'the fallback is Live\'s own series, not zeros');
+ readState.down = false;
+ chatReads._reset();
+
+ // ── First chat: `user:<user_id>` (never the username); a false answer is cached longer
+ // than a true one; a cold/down peek falls to Live's own table. ──
+ d2.run('DELETE FROM stream_first_chats');
+ chatReads._reset();
+ assert.strictEqual(await chatReads.firstChat(streamer, `user:${viewer}`), true, 'a new identity is first');
+ d2.run('INSERT INTO stream_first_chats (chatter_key, channel_user_id) VALUES (?, ?)', [`user:${viewer}`, streamer]);
+ chatReads._reset();
+ assert.strictEqual(await chatReads.firstChat(streamer, `user:${viewer}`), false, 'after a recorded first chat, not first');
+ chatReads._reset();
+ readState.firstChatCalls = 0;
+ await chatReads.firstChat(streamer, `user:${viewer}`);
+ chatReads._age(chatReads.CACHE_TTL_MS + 1000, 'fc:');
+ readState.firstChatCalls = 0;
+ await chatReads.firstChat(streamer, `user:${viewer}`);
+ assert.strictEqual(readState.firstChatCalls, 0, 'a false answer is not re-asked past the true TTL');
+ // The same holds through the sync peek the AI context uses.
+ chatReads._reset();
+ readState.firstChatCalls = 0;
+ assert.strictEqual(chatReads.firstChatPeek(streamer, `user:${viewer}`), false);
+ await sleep(30);
+ chatReads._age(chatReads.CACHE_TTL_MS + 1000, 'fc:');
+ readState.firstChatCalls = 0;
+ assert.strictEqual(chatReads.firstChatPeek(streamer, `user:${viewer}`), false, 'the peek keeps a false answer past the true TTL');
+ assert.strictEqual(readState.firstChatCalls, 0, 'the peek does not re-ask Chat for the cached false answer');
+ d2.run('DELETE FROM stream_first_chats WHERE chatter_key = ?', [`user:${viewer}`]);
+ chatReads._reset();
+ readState.firstChatCalls = 0;
+ await chatReads.firstChat(streamer, `user:${viewer}`);
+ chatReads._age(chatReads.CACHE_TTL_MS + 1000, 'fc:');
+ readState.firstChatCalls = 0;
+ await chatReads.firstChat(streamer, `user:${viewer}`);
+ assert.strictEqual(readState.firstChatCalls, 1, 'a true answer is re-asked after the short TTL');
+ chatReads._reset();
+ d2.run('INSERT INTO stream_first_chats (chatter_key, channel_user_id) VALUES (?, ?)', [`user:${viewer}`, streamer]);
+ readState.down = true;
+ assert.strictEqual(chatReads.firstChatPeek(streamer, `user:${viewer}`), false, 'a cold + down peek falls to Live\'s own table');
+ readState.down = false;
+ chatReads._reset();
+
+ // The AI context passes `user:<user_id>`, never the username: a row keyed by the numeric id
+ // suppresses the welcome flag, where the old `user:<username>` key would have missed it.
+ d2.run("INSERT INTO chat_messages (user_id, username, message, message_type, channel_user_id, timestamp) VALUES (?, 'viewer', 'hi there', 'chat', ?, ?)", [viewer, streamer, sqlNow()]);
+ await chatReads.channelMessages(streamer, 40);
+ const context = require('../server/ai/context');
+ const ctxStream = db.getStreamById(streamId);
+ const greet = { remember_viewers: true, greet_first_timers: true, max_open_threads: 3, hear_enabled: false };
+ chatReads.invalidate('fc:');
+ let tail = context.volatileTail({ userId: streamer, stream: ctxStream, settings: greet, sinceChatId: 0, botNames: new Set() });
+ assert.ok(!/first time chatting here/.test(tail.text), 'a user with a recorded first chat is not greeted again');
+ d2.run('DELETE FROM stream_first_chats WHERE chatter_key = ?', [`user:${viewer}`]);
+ chatReads.invalidate('fc:');
+ tail = context.volatileTail({ userId: streamer, stream: ctxStream, settings: greet, sinceChatId: 0, botNames: new Set() });
+ assert.ok(/first time chatting here/.test(tail.text), 'a new user is flagged first time (user:<id> identity)');
+ chatReads._reset();
+
+ // ── Sounds by command: Chat owns the row; a 404 is authoritative; a cold/down peek falls
+ // to Live's own table. ──
+ d2.run('DELETE FROM channel_sounds WHERE channel_owner_id = ?', [streamer]);
+ d2.run("INSERT INTO channel_sounds (channel_owner_id, command, url, created_by) VALUES (?, 'honk', '/sounds/h.mp3', ?)", [streamer, streamer]);
+ chatReads._reset();
+ const honk = chatReads.soundByCommandPeek(streamer, 'honk');
+ assert.ok(honk && honk.command === 'honk', 'Chat answers the sound by command');
+ assert.strictEqual(chatReads.soundByCommandPeek(streamer, '!HONK').command, 'honk', 'the command is normalized');
+ assert.strictEqual(await chatReads.soundByCommand(streamer, 'nope'), null, 'Chat: no such sound');
+ d2.run("INSERT INTO channel_sounds (channel_owner_id, command, url, created_by) VALUES (?, 'nope', '/sounds/n.mp3', ?)", [streamer, streamer]);
+ assert.strictEqual(chatReads.soundByCommandPeek(streamer, 'nope'), null, 'a Chat 404 is cached: Live\'s own row is not used');
+ chatReads._reset();
+ readState.down = true;
+ assert.ok(chatReads.soundByCommandPeek(streamer, 'honk'), 'a cold + down peek falls back to Live\'s own table');
+ readState.down = false;
+ chatReads._reset();
+ d2.run("INSERT INTO channel_sounds (channel_owner_id, command, url, created_by) VALUES (?, 'bare404', '/sounds/b.mp3', ?)", [streamer, streamer]);
+ const bare = await chatReads.soundByCommand(streamer, 'bare404');
+ assert.ok(bare && bare.command === 'bare404', 'a 404 without Chat\'s own body is not "no such sound": Live\'s table answers');
+ chatReads._reset();
+ }
+
// 12. Robot commands from chat pass the control panel's gate: control mode, anonymous switch, whitelist, cooldown.
// 12. Robot commands from chat pass the control panel's gate: control mode, anonymous switch, whitelist, cooldown.
{
{
const sent = [];
const sent = [];