openvibers/openvibe.live Commits
Repoint Live's remaining chat readers to OpenVibe.Chat (recap, home stats, profiles, stream analytics, star, VOD chat stats, AI viewers); the three without a Chat read stay on Live's tables, named (#29)
Changes
@@@ -49,6 +49,14 @@ with the same WebSocket protocol and REST paths. `CHAT_AUTHORITY=chat` switches(`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'schat 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.- **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@@@ -462,7 +462,9 @@</div>`;}const coins = p.openvibe_coins_balance || 0;-- const msgs = p.messageCount || 0;++ // Chat answers the message count; a cold cache leaves it null and the card omits the stat++ // rather than showing a stale mirror number as real.++ const msgs = p.messageCount == null ? null : Number(p.messageCount);let modBtns = '';if (canMod || isGlobalMod) {if (d.msgId) modBtns += `<button class="ctx-btn ctx-btn-warn" onclick="ctxDeleteMessage('${esc(String(d.msgId))}')"><i class="fa-solid fa-trash"></i> Delete Message</button>`;@@@ -478,7 +480,7 @@</div><div class="ctx-stats"><div class="ctx-stat"><i class="fa-solid fa-coins"></i> ${numf(coins)}</div>-- <div class="ctx-stat"><i class="fa-solid fa-message"></i> ${numf(msgs)}</div>++ ${msgs == null ? '' : `<div class="ctx-stat"><i class="fa-solid fa-message"></i> ${numf(msgs)}</div>`}<div class="ctx-stat"><i class="fa-solid fa-heart"></i> ${numf(p.followerCount || 0)}</div></div>${gameHtml}@@@ -204,6 +204,9 @@ function personBlock(line, channelUserId, settings) {}}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');} catch { /* */ }const insight = ins ? clip([ins.overview_24h || ins.overview_alltime, ins.memory].filter(Boolean).join(' — '), 220) : '';@@@ -10,6 +10,7 @@*/'use strict';const db = require('../../db/database');++const chatReads = require('../../chat/chat-reads');const settingsMod = require('./settings');const roster = require('./roster');const budget = require('./budget');@@@ -56,7 +57,7 @@ class AiViewersEngineV3 {const worker = {streamId: stream.id, userId: stream.user_id, stream: full, cfg, settings, bots,stable: null, // cached stable prefix {text,hash,at}-- lastChatId: db.getMaxChatMessageIdForChannel(stream.user_id),++ lastChatId: chatReads.channelMaxIdPeek(stream.user_id), // Chat's cursor, so the first tick does not replay old lineslastTickAt: 0, nextTickAt: 0, tickTimer: null, foldTimer: null, ticking: false,lastRealInputAt: 0, lastStreamerReplyAt: 0, lastSkipLogAt: 0, lastVolatileHash: null,intents: [], // [{ kind:'streamer'|'mention'|'sound'|'scene', text, bots:[], at }]@@@ -73,6 +74,9 @@ class AiViewersEngineV3 {dropped: (line, why) => poster.log(worker, { event: 'skip', bot_username: line.bot, text: line.text, reason: why }),});this.workers.set(stream.id, worker);++ // The peek above may have answered Live's mirror while Chat's cache was cold; refresh the++ // cursor from Chat before the first tick (4 s away) so it cannot replay lines from earlier.++ chatReads.channelMaxId(stream.user_id).then((id) => { if (!worker.stopped) worker.lastChatId = Math.max(worker.lastChatId || 0, id); }).catch(() => {});poster.log(worker, { event: 'info', reason: `engine v3 started — ${bots.length} viewer(s), ${settings.activity} activity, director every ${settings.director_interval_sec}s` });console.log(`[AI-Viewers v3] started for stream ${stream.id} (user ${stream.user_id}, ${bots.length} bots)`);this._scheduleTick(worker, 4000);@@@ -413,8 +417,11 @@ class AiViewersEngineV3 {if (!live.length) return { error: 'Go live first — the preview runs on your real stream context.' };const cfg = db.getChannelAiConfig(userId); const settings = settingsMod.getSettings(userId, cfg);const bots = roster.ensureRoster(userId, settings.roster_size || 3);-- w = { streamId: live[0].id, userId, stream: db.getStreamById(live[0].id) || live[0], cfg, settings, bots, lastChatId: Math.max(0, db.getMaxChatMessageIdForChannel(userId) - 40), intents: [], scheduler: { botShare: () => 0 } };++ w = { streamId: live[0].id, userId, stream: db.getStreamById(live[0].id) || live[0], cfg, settings, bots, lastChatId: Math.max(0, chatReads.channelMaxIdPeek(userId) - 40), intents: [], scheduler: { botShare: () => 0 } };temp = true;++ // The peek above may answer Live's mirror on a cold cache; await Chat's cursor so the++ // preview never replays lines from before the window it means to show.++ try { w.lastChatId = Math.max(0, (await chatReads.channelMaxId(userId)) - 40); } catch { /* keep the peek */ }}const s = w.settings;const stable = context.stablePrefix({ userId: w.userId, stream: w.stream, bots: w.bots, settings: s, cacheHolder: temp ? null : w });@@@ -273,7 +273,9 @@ router.post('/clone', requireAuth, async (req, res) => {const platform = idx > 0 ? ref.slice(0, idx) : '';const username = idx > 0 ? ref.slice(idx + 1) : '';if (!platform || !username) return res.status(400).json({ error: 'Bad relay ref' });-- const relay = db.getRelayUser(platform, username);++ // Chat owns relay_users and exposes no relay-identity read (only the hidden-relay++ // moderation list and message history, both read below), so there is nothing to look up++ // here; the clone uses the handle.const wide = isStaff(req.user);src = {kind: 'relay', ref: `${platform}:${username}`,@@@ -40,6 +40,10 @@ let lastLog = null;function drill() { try { return require('../drill').enabled; } catch { return false; } }function remote() { return isRemote() && !drill(); }function local() { return require('../db/database'); }++/** Run a local (Live-table) read for a synchronous caller, answering `dflt` if it throws. */++function safeLocal(fn, dflt) { try { const v = fn(); return v == null ? dflt : v; } catch { return dflt; } }++/** Epoch ms → the UTC 'YYYY-MM-DD HH:MM:SS' text chat_messages.timestamp stores. */++const sqlTime = (ms) => new Date(ms).toISOString().slice(0, 19).replace('T', ' ');function note(what, err) {const m = `Chat read ${what}: ${(err && err.message) || err}`;@@@ -185,6 +189,139 @@ function topChatters({ since, streamId, channelUserId, limit = 10 } = {}) {});}++// ── Stats over a window / one room ───────────────────────────────────────────++// Chat's stats read takes since/until plus at most one of stream_id / channel_user_id and answers++// { messages, chatters } for that window (kind 'stream' also answers the soundboard count). The++// home-stats snapshot and the digest ask site-wide, the recap and stream analytics ask per stream,++// the star picker per channel. The local (rollback / dev) answer is the same SQL Live's own chat++// runs; a peek answers the last good value while Chat refreshes in the background.++function localWindowStats({ since, until, streamId, channelUserId } = {}) {++ const db = local();++ const where = ['is_deleted = 0'];++ const params = [];++ if (since != null) { where.push('timestamp >= ?'); params.push(sqlTime(since)); }++ if (until != null) { where.push('timestamp < ?'); params.push(sqlTime(until)); }++ if (streamId != null) { where.push('stream_id = ?'); params.push(Number(streamId)); }++ if (channelUserId != null) { where.push('channel_user_id = ?'); params.push(Number(channelUserId)); }++ 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 ${where.join(' AND ')}`, params) || {};++ return { messages: Number(r.messages) || 0, chatters: Number(r.chatters) || 0 };++}++async function remoteWindowStats({ since, until, streamId, channelUserId } = {}) {++ const out = await client.readStats({ kind: 'site', since, until, stream_id: streamId, channel_user_id: channelUserId });++ if (!out) throw unavailable('window stats');++ return { messages: Number(out.messages) || 0, chatters: Number(out.chatters) || 0 };++}++const MIN_BUCKET_MS = 60_000; // sub-day windows snap to the minute++const HOUR_BUCKET_MS = 60 * 60_000; // a 7/14/30-day window snaps to the hour++/**++ * Snap a window's since/until to a stable bucket BEFORE both the cache key and the request use them.++ * Callers build these from Date.now(), which changes on every compute, so an exact key is cold on++ * every call: Chat is never actually consulted (the peek answers a mirror scan instead), each call++ * mints a new cache entry, and the request it sends is thrown away. A minute — or an hour once the++ * window spans a day or more — is far finer than these coarse counts need and keeps the key stable++ * well past CACHE_TTL_MS. The request is sent with the bucketed values so it matches its own key.++ */++function bucketWindow({ since, until, streamId, channelUserId } = {}) {++ const span = (until != null ? Number(until) : Date.now()) - (since != null ? Number(since) : 0);++ const g = span >= 24 * 60 * 60_000 ? HOUR_BUCKET_MS : MIN_BUCKET_MS;++ const snap = (t) => (t == null ? undefined : Math.floor(Number(t) / g) * g);++ return { since: snap(since), until: snap(until), streamId, channelUserId };++}++const windowKey = (o) => `win:${o.since || 0}:${o.until || 0}:${o.streamId || 0}:${o.channelUserId || 0}`;++/** Message/chatter counts over [since, until), site-wide or scoped to one stream/channel (async). */++function windowStats(o = {}) {++ const b = bucketWindow(o);++ return load(windowKey(b), () => remoteWindowStats(b), () => localWindowStats(b))++ .then((v) => v || safeLocal(() => localWindowStats(b), { messages: 0, chatters: 0 }));++}++/**++ * The same, for a caller that cannot await (the home-stats snapshot, the digest). Answers the last++ * good value, else null — never a synchronous mirror scan, so the hero never presents Live's stale++ * count as Chat's; the caller decides what an unknown window means.++ */++function windowStatsPeek(o = {}) {++ const b = bucketWindow(o);++ return peek(windowKey(b), () => remoteWindowStats(b), () => localWindowStats(b));++}++++function localStreamStats(streamId) {++ const db = local();++ const r = db.get(`SELECT COUNT(*) AS messages,++ COUNT(DISTINCT COALESCE(user_id, anon_id, username)) AS chatters,++ SUM(CASE WHEN message_type = 'soundboard' THEN 1 ELSE 0 END) AS sounds++ FROM chat_messages WHERE stream_id = ? AND COALESCE(is_deleted, 0) = 0`, [Number(streamId)]) || {};++ return { messages: Number(r.messages) || 0, chatters: Number(r.chatters) || 0, sounds: Number(r.sounds) || 0 };++}++async function remoteStreamStats(streamId) {++ const out = await client.readStats({ kind: 'stream', stream_id: Number(streamId) });++ if (!out) throw unavailable('stream stats');++ return { messages: Number(out.messages) || 0, chatters: Number(out.chatters) || 0, sounds: Number(out.sounds) || 0 };++}++/** One stream's chat totals: { messages, chatters, sounds } (the recap, stream analytics). */++function streamStats(streamId) {++ const id = Number(streamId) || 0;++ if (!id) return Promise.resolve({ messages: 0, chatters: 0, sounds: 0 });++ return load(`st:${id}`, () => remoteStreamStats(id), () => localStreamStats(id))++ .then((v) => v || safeLocal(() => localStreamStats(id), { messages: 0, chatters: 0, sounds: 0 }));++}++/** The same, for a caller that cannot await. */++function streamStatsPeek(streamId) {++ const id = Number(streamId) || 0;++ if (!id) return { messages: 0, chatters: 0, sounds: 0 };++ const v = peek(`st:${id}`, () => remoteStreamStats(id), () => localStreamStats(id));++ if (v) return v;++ return safeLocal(() => localStreamStats(id), { messages: 0, chatters: 0, sounds: 0 });++}++++async function remoteChannelMaxId(channelUserId) {++ const out = await client.readMessages({ channel_user_id: channelUserId, tail: true });++ if (!out) throw unavailable('channel max id');++ return Number(out.max_id) || 0;++}++/**++ * The newest chat id in a channel — Chat's id, so the AI context delta (channelMessagesPeek reads++ * Chat's ids) stays aligned. `channelMaxId` awaits Chat (and warms the peek); `channelMaxIdPeek` is++ * for the caller that cannot await and answers the last good id, else Live's own table.++ */++function channelMaxId(channelUserId) {++ const id = Number(channelUserId) || 0;++ if (!id) return Promise.resolve(0);++ return load(`cmid:${id}`, () => remoteChannelMaxId(id), () => local().getMaxChatMessageIdForChannel(id))++ .then((v) => (v != null ? Number(v) || 0 : safeLocal(() => local().getMaxChatMessageIdForChannel(id), 0)));++}++function channelMaxIdPeek(channelUserId) {++ const id = Number(channelUserId) || 0;++ if (!id) return 0;++ const v = peek(`cmid:${id}`, () => remoteChannelMaxId(id), () => local().getMaxChatMessageIdForChannel(id));++ if (v != null) return Number(v) || 0;++ return safeLocal(() => local().getMaxChatMessageIdForChannel(id), 0);++}++++function localUserMessageCount(userId) {++ const r = local().get(`SELECT COUNT(*) AS c FROM chat_messages++ WHERE user_id = ? AND is_deleted = 0++ AND (auto_delete_at IS NULL OR datetime(auto_delete_at) > CURRENT_TIMESTAMP)`, [Number(userId)]);++ return Number(r && r.c) || 0;++}++async function remoteUserMessageCount(userId) {++ const out = await client.readStats({ kind: 'user', user_id: Number(userId) });++ if (!out) throw unavailable('user message count');++ return Number(out.messages) || 0;++}++/**++ * A user's non-deleted chat total, for a caller that cannot await (the profile card). Answers the++ * last good value, else null when neither Chat nor (in dev / rollback) Live's table answered — the++ * card omits the count then rather than presenting a cold mirror number as real.++ */++function userMessageCountPeek(userId) {++ const id = Number(userId) || 0;++ if (!id) return null;++ const v = peek(`um:${id}`, () => remoteUserMessageCount(id), () => localUserMessageCount(id));++ return v == null ? null : Number(v) || 0;++}++// ── Messages / history ────────────────────────────────────────────────────────/*** The admin console's chat search: { messages, total }. Chat's message read takes exactly one@@@ -395,27 +532,32 @@ function relayUser(id) {* 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.++ * Live's own table OR Chat's list (the #28 rule): a hide the mirror applied wins at once even when a++ * cached Chat list fetched just before it lacks the row, and a Chat row hides even before the mirror++ * catches up. A Live-route hide/unhide invalidates the cached list; a cold cache or a Chat outage++ * falls to Live's own table and fails OPEN, never dropping a relayed line.*/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 localHidden = () => { try { return !!local().isRelayUserHidden(channelId, platform, username); } catch { return false; } };++ if (!remote()) return localHidden();++ // Live's own table OR Chat's fresh list — the rule agreed with #28. A fresh Chat list alone++ // would hide a local hide the mirror applied from any other writer (another Live instance, or a++ // write straight in Chat) for up to CACHE_TTL_MS. A Live-route unhide stays immediate: mod-routes++ // drops the `hru:` entry on every hide/unhide, so the next check re-asks Chat. Until Chat has++ // answered (cold cache) or while it is unreachable, Live's own table answers and fails OPEN —++ // never drop a relayed line on a Chat outage.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++ const hit = fresh(key);if (hit && Array.isArray(hit.value)) {-- return hit.value.some((r) => String(r.platform) === String(platform) && String(r.external_username) === String(username));++ if (hit.value.some((r) => String(r.platform) === String(platform) && String(r.external_username) === String(username))) return true;++ return localHidden(); // no fresh Chat match: a local hide still applies}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;++ return localHidden();}/** A user's TTS voice override: { voice, pitch, speed, gap } | null. Live, never cached. */@@@ -511,6 +653,7 @@ function _cacheMax(n) { if (n != null && Number(n) > 0) cacheMax = Number(n); remodule.exports = {siteStats, siteStatsPeek, topChatters,++ windowStats, windowStatsPeek, streamStats, streamStatsPeek, channelMaxId, channelMaxIdPeek, userMessageCountPeek,messageById, channelMessages, channelMessagesPeek, userHistory, relayHistory, channelSamples, searchMessages,liveChatBuckets, recentChatText, spikeOffsets,pendingIp, relayUsers, relayUser, ttsOverride, ttsOverridePeek, isRelayUserHidden,@@@ -3805,6 +3805,9 @@ 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.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(*)' },@@@ -3903,6 +3906,7 @@ function getHomeStats() {function _computeHomeStats() {// Each stat is isolated so a missing table / column can never blank the whole hero.const c = (sql, p = []) => { try { return get(sql, p)?.count || 0; } catch { return 0; } };++ const nowMs = Date.now(); // the home-stats snapshot's windows are relative to this instant// Vibes tipped before this instant were test money (site setting `stats_vibes_reset_at`,// ISO timestamp). Everything Vibes-related on the hero starts counting from it.const vibesSince = vibesStatsSince();@@@ -3974,32 +3978,17 @@ function _computeHomeStats() {// Total hours of video the platform has archived (OpenVibe.Media's figure).streamHours: null,// Active chatters this week across EVERYONE — registered users, anons, and relay chatters.-- weeklyActive: c(`SELECT COUNT(*) AS count FROM (-- SELECT DISTINCT 'u' || user_id AS id FROM chat_messages-- WHERE user_id IS NOT NULL AND COALESCE(is_deleted, 0) = 0 AND timestamp >= datetime('now', '-7 days')-- UNION-- SELECT DISTINCT 'a' || anon_id FROM chat_messages-- WHERE anon_id IS NOT NULL AND COALESCE(is_deleted, 0) = 0 AND timestamp >= datetime('now', '-7 days')-- UNION-- SELECT DISTINCT 'r' || source_platform || '|' || username FROM chat_messages-- WHERE source_platform IS NOT NULL AND source_platform <> '' AND COALESCE(is_deleted, 0) = 0 AND timestamp >= datetime('now', '-7 days')-- )`),++ // OpenVibe.Chat counts them (stats kind 'site' over the window); Live's own tables when Live++ // runs chat. The peek answers the last good count or null (never a mirror scan or a 500);++ // the hero treats null as unknown.++ weeklyActive: (() => { try { const s = require('../chat/chat-reads').windowStatsPeek({ since: nowMs - 7 * 86400000 }); return s && s.chatters != null ? s.chatters : null; } catch { return null; } })(),// New unique visitors this week (first-seen anon fingerprints) — a proxy for people who// showed up, not just those who chatted.weeklyVisitors: c(`SELECT COUNT(*) AS count FROM anon_ip_mappings WHERE created_at >= datetime('now', '-7 days')`),// The same two windows again, shifted back a week, so the hero can say whether this week// beat last week rather than just how big it was.prevWeeklyVisitors: c(`SELECT COUNT(*) AS count FROM anon_ip_mappings WHERE created_at >= datetime('now', '-14 days') AND created_at < datetime('now', '-7 days')`),-- prevWeeklyActive: c(`SELECT COUNT(*) AS count FROM (-- SELECT DISTINCT 'u' || user_id AS id FROM chat_messages-- WHERE user_id IS NOT NULL AND COALESCE(is_deleted, 0) = 0 AND timestamp >= datetime('now', '-14 days') AND timestamp < datetime('now', '-7 days')-- UNION-- SELECT DISTINCT 'a' || anon_id FROM chat_messages-- WHERE anon_id IS NOT NULL AND COALESCE(is_deleted, 0) = 0 AND timestamp >= datetime('now', '-14 days') AND timestamp < datetime('now', '-7 days')-- UNION-- SELECT DISTINCT 'r' || source_platform || '|' || username FROM chat_messages-- WHERE source_platform IS NOT NULL AND source_platform <> '' AND COALESCE(is_deleted, 0) = 0 AND timestamp >= datetime('now', '-14 days') AND timestamp < datetime('now', '-7 days')-- )`),++ prevWeeklyActive: (() => { try { const s = require('../chat/chat-reads').windowStatsPeek({ since: nowMs - 14 * 86400000, until: nowMs - 7 * 86400000 }); return s && s.chatters != null ? s.chatters : null; } catch { return null; } })(),liveNow: c(`SELECT COUNT(*) AS count FROM streams WHERE is_live = 1`),// Rolling last-day / week / month deltas ({ d, w, m }) for the hero stat tooltips + subs.recent: {@@@ -4009,7 +3998,16 @@ function _computeHomeStats() {vods: null, // OpenVibe.Media (withArchiveStats)clips: null,aiMoments: winCount('stream_memories', 'created_at'),-- messages: winCount('chat_messages', 'timestamp', 'COALESCE(is_deleted, 0) = 0'),++ // OpenVibe.Chat's message counts over each window (Live's own tables when Live runs chat).++ messages: (() => {++ const w = (o) => { try { const s = require('../chat/chat-reads').windowStatsPeek(o); return s && s.messages != null ? s.messages : null; } catch { return null; } };++ return {++ d: w({ since: nowMs - 86400000 }),++ w: w({ since: nowMs - 7 * 86400000 }),++ m: w({ since: nowMs - 30 * 86400000 }),++ pw: w({ since: nowMs - 14 * 86400000, until: nowMs - 7 * 86400000 }),++ };++ })(),hours: null, // OpenVibe.Mediastreamers: streamersWin(),// Live's `emotes` copy was unread since N+2 and is now dropped, so the@@@ -4688,12 +4686,6 @@ function recordRelayUser(platform, username) {[String(platform).toLowerCase(), key, String(username)]);} catch { /* non-critical */ }}--function getRelayUser(platform, username) {-- if (!platform || !username) return null;-- // rowid is a stable integer id for a relay user (no dedicated id column).-- return get('SELECT rowid AS id, * FROM relay_users WHERE platform = ? AND username = ?',-- [String(platform).toLowerCase(), String(username).toLowerCase()]) || null;--}function getRelayUserByRowid(id) {return get('SELECT rowid AS id, * FROM relay_users WHERE rowid = ?', [id]) || null;}@@@ -4776,12 +4768,10 @@ function getUserProfile(userId) {openvibe_bucks_balance, openvibe_coins_balance, created_at, last_seenFROM users WHERE id = ?`, [userId]);if (!user) return null;-- user.messageCount = get(-- `SELECT COUNT(*) as c FROM chat_messages-- WHERE user_id = ? AND is_deleted = 0-- AND (auto_delete_at IS NULL OR datetime(auto_delete_at) > CURRENT_TIMESTAMP)`,-- [userId]-- )?.c || 0;++ // The user's chat total is OpenVibe.Chat's (Live's own tables when Live runs chat); a synchronous++ // peek answers the last good count, else null while Chat refreshes in the background. The profile++ // card omits the count when it is null rather than showing a cold mirror number as real.++ user.messageCount = (() => { try { return require('../chat/chat-reads').userMessageCountPeek(userId); } catch { return null; } })();user.followerCount = get('SELECT COUNT(*) as c FROM follows WHERE streamer_id = ?', [userId])?.c || 0;user.followingCount = get('SELECT COUNT(*) as c FROM follows WHERE follower_id = ?', [userId])?.c || 0;return user;@@@ -6650,21 +6640,22 @@ function computeAndCacheStreamAnalytics(streamId) {);const avgViewers = avgRow?.avg_vc || 0;-- // Unique chatters-- const chattersRow = get(-- `SELECT COUNT(DISTINCT COALESCE(user_id, anon_id)) as cnt-- FROM chat_messages WHERE stream_id = ? AND is_deleted = 0 AND is_global = 0`,-- [streamId]-- );-- const uniqueChatters = chattersRow?.cnt || 0;---- // Total messages-- const msgsRow = get(-- `SELECT COUNT(*) as cnt FROM chat_messages-- WHERE stream_id = ? AND is_deleted = 0 AND is_global = 0 AND message_type = 'chat'`,-- [streamId]-- );-- const totalMessages = msgsRow?.cnt || 0;++ // Unique chatters + total messages come from OpenVibe.Chat (Live's own tables when Live runs++ // chat). Chat's stream stats count every message type — Live used to keep only message_type='chat'++ // and is_global=0 — and its numbers are the authority now. This runs synchronously at stream end,++ // when `st:<id>` is still cold, so a peek would answer the mirror (or 0 on a throw): it keeps the++ // totals already stored and asks Chat right after (setStreamAnalyticsChatTotals writes them back),++ // the same pattern as the clip count below.++ const prior = get('SELECT unique_chatters, total_messages FROM stream_analytics WHERE stream_id = ?', [streamId]) || {};++ const uniqueChatters = Number(prior.unique_chatters) || 0;++ const totalMessages = Number(prior.total_messages) || 0;++ setImmediate(() => {++ try {++ require('../chat/chat-reads').streamStats(streamId)++ .then((t) => { if (t) setStreamAnalyticsChatTotals(streamId, t.chatters, t.messages); })++ .catch(() => {});++ } catch { /* */ }++ });// Total watch minutesconst watchRow = get(@@@ -6737,6 +6728,13 @@ function setStreamAnalyticsClipCount(streamId, count) {return run('UPDATE stream_analytics SET clips_created = ? WHERE stream_id = ?', [Math.max(0, Math.floor(Number(count) || 0)), streamId]);}++// The chat totals OpenVibe.Chat reported for a stream (the setImmediate refresh in++// computeAndCacheStreamAnalytics).++function setStreamAnalyticsChatTotals(streamId, chatters, messages) {++ return run('UPDATE stream_analytics SET unique_chatters = ?, total_messages = ? WHERE stream_id = ?',++ [Math.max(0, Math.floor(Number(chatters) || 0)), Math.max(0, Math.floor(Number(messages) || 0)), streamId]);++}++function getChannelAnalyticsSummary(userId, days) {const cutoff = new Date(Date.now() - days * 86400000).toISOString();@@@ -7129,7 +7127,7 @@ module.exports = {holdMessageForApproval, getPendingIpMessages, reviewPendingIpMessage,approveAllFromIp, denyAllFromIp,// Hidden Relay Users-- hideRelayUser, isRelayUserHidden, unhideRelayUser, unhideRelayUserByIdentity, getHiddenRelayUsers, recordRelayUser, getRelayUser,++ hideRelayUser, isRelayUserHidden, unhideRelayUser, unhideRelayUserByIdentity, getHiddenRelayUsers, recordRelayUser,getRelayUserByRowid, getRelayUserChatHistory,anonSubjectId, getAnonMeta, getAnonChatHistory,// IP Tracking@@@ -7137,7 +7135,7 @@ module.exports = {getLatestIpForUser, getLatestIpForAnon, getIpLog, banAllAccountsOnIp,// Stream AnalyticsinsertViewerSnapshot, getViewerSnapshots, computeAndCacheStreamAnalytics,-- getStreamAnalytics, setStreamAnalyticsClipCount, getChannelAnalyticsSummary, getRecentChatActivity,++ getStreamAnalytics, setStreamAnalyticsClipCount, setStreamAnalyticsChatTotals, getChannelAnalyticsSummary, getRecentChatActivity,// User PreferencesgetUserPreferences, saveUserPreferences,// Chat Log Management@@@ -468,7 +468,9 @@ router.get('/digest', optionalAuth, (req, res) => {const stats = {streams: Number((one('SELECT COUNT(*) AS n FROM streams WHERE started_at >= ?', [sinceSql]) || {}).n || 0),hours: Number((one(`SELECT ROUND(SUM(COALESCE(duration_seconds, CASE WHEN ended_at IS NOT NULL THEN (julianday(ended_at) - julianday(started_at)) * 86400 ELSE (julianday('now') - julianday(started_at)) * 86400 END)) / 3600.0, 1) AS h FROM streams WHERE started_at >= ?`, [sinceSql]) || {}).h || 0),-- chat_lines: Number((one("SELECT COUNT(*) AS n FROM chat_messages WHERE timestamp >= ? AND COALESCE(is_deleted, 0) = 0", [sinceSql]) || {}).n || 0),++ // OpenVibe.Chat's message count since the window opened (Live's own tables in dev /++ // rollback); a synchronous peek answers the last good count while Chat refreshes.++ chat_lines: (() => { try { const s = require('../chat/chat-reads').windowStatsPeek({ since: sinceMs }); return s ? s.messages : 0; } catch { return 0; } })(),new_follows: Number((one('SELECT COUNT(*) AS n FROM follows WHERE created_at >= ?', [sinceSql]) || {}).n || 0),new_members: Number((one('SELECT COUNT(*) AS n FROM users WHERE created_at >= ?', [sinceSql]) || {}).n || 0),mic_moments: Number((one('SELECT COUNT(*) AS n FROM arena_mic_moments WHERE said_at >= ?', [sinceSql]) || {}).n || 0),@@@ -13,6 +13,7 @@*/'use strict';const db = require('../db/database');++const chatReads = require('../chat/chat-reads');const ROTATE_MS = 24 * 60 * 60 * 1000; // a new star every dayconst CHECK_MS = 30 * 60 * 1000; // how often the job looks at the clock@@@ -29,7 +30,8 @@ function loadPick() {function pinned() { try { return String(db.getSetting('star_streamer_pinned') || '').trim(); } catch { return ''; } }/** Everyone who streamed recently, with the numbers the picker looks at. */--function candidates() {++async function candidates() {++ const sinceMs = Date.now() - CANDIDATE_DAYS * 86400000;const rows = db.all(`SELECT u.id, u.username, u.display_name, u.bio,COUNT(s.id) AS sessions,@@@ -37,7 +39,6 @@ function candidates() {MAX(COALESCE(s.peak_viewers, 0)) AS peak_viewers,ROUND(AVG(COALESCE(s.peak_viewers, 0)), 1) AS avg_peak,MAX(s.started_at) AS last_live_at,-- (SELECT COUNT(*) FROM chat_messages c WHERE c.stream_id IN (SELECT id FROM streams x WHERE x.user_id = u.id AND x.started_at >= datetime('now', ?)) AND COALESCE(c.is_deleted, 0) = 0) AS chat_lines,(SELECT COUNT(*) FROM follows f WHERE f.streamer_id = u.id AND f.created_at >= datetime('now', ?)) AS new_followers,(SELECT COUNT(*) FROM follows f WHERE f.streamer_id = u.id) AS followers,(SELECT COUNT(*) FROM arena_mic_moments m WHERE m.user_id = u.id AND m.said_at >= datetime('now', ?)) AS mic_moments,@@@ -46,7 +47,13 @@ function candidates() {FROM users u JOIN streams s ON s.user_id = u.idWHERE s.started_at >= datetime('now', ?) AND COALESCE(u.is_banned, 0) = 0GROUP BY u.id-- ORDER BY hours DESC LIMIT 40`, [`-${CANDIDATE_DAYS} days`, `-${CANDIDATE_DAYS} days`, `-${CANDIDATE_DAYS} days`, `-${CANDIDATE_DAYS} days`]) || [];++ ORDER BY hours DESC LIMIT 40`, [`-${CANDIDATE_DAYS} days`, `-${CANDIDATE_DAYS} days`, `-${CANDIDATE_DAYS} days`]) || [];++ // Chat lines per streamer over the window, from OpenVibe.Chat (Live's own tables when Live runs++ // chat). One read per candidate; the star pick is a daily background job, so the fan-out is cheap.++ await Promise.all(rows.map(async (r) => {++ try { const s = await chatReads.windowStats({ channelUserId: r.id, since: sinceMs }); r.chat_lines = (s && s.messages) || 0; }++ catch { r.chat_lines = 0; }++ }));let i18n = null; try { i18n = require('../i18n/translate'); } catch { i18n = null; }for (const r of rows) {r.language = i18n ? (() => { try { return i18n.channelLanguage(r.id); } catch { return 'en'; } })() : 'en';@@@ -91,7 +98,7 @@ async function rotate({ force = false } = {}) {if (!force && cur && cur.next_at && now < Number(cur.next_at)) return { current: cur.username, next_at: cur.next_at };_busy = true;try {-- const cands = candidates();++ const cands = await candidates();if (!cands.length) return { none: true };// Recent stars (and whatever the setting / env currently holds) are ineligible.const history = Array.isArray(cur && cur.history) ? cur.history.slice() : [];@@@ -678,6 +678,8 @@ 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)) {if (delivery.ingress()) {delivery.event({ kind: 'stream', id: stream.id }, {@@@ -11,6 +11,7 @@const express = require('express');const multer = require('multer');const db = require('../db/database');++const chatReads = require('../chat/chat-reads');const media = require('../media-client');const recorder = require('../streaming/recorder');const { requireAuth, optionalAuth } = require('../auth/auth');@@@ -430,16 +431,22 @@ router.get('/:id/context', optionalAuth, async (req, res) => {let stats = null;if (stream) {const start = String(stream.started_at || ''), end = String(stream.ended_at || new Date().toISOString().replace('T', ' ').slice(0, 19));-- const chat = safe(() => db.get('SELECT COUNT(*) AS n, COUNT(DISTINCT COALESCE(user_id, anon_id, username)) AS c FROM chat_messages WHERE stream_id = ? AND COALESCE(is_deleted, 0) = 0', [stream.id]), { n: 0, c: 0 });++ // Chat totals, the sound count and the top chatters come from OpenVibe.Chat (Live's own++ // tables when Live runs chat); a Chat outage answers a cached or zero value, never a 500.++ const chatStats = (await chatReads.streamStats(stream.id)) || { messages: 0, chatters: 0, sounds: 0 };++ const topChatters = ((await chatReads.topChatters({ streamId: stream.id, limit: 3 })) || [])++ .filter(r => r.username && (r.user_id == null || Number(r.user_id) !== Number(stream.user_id)))++ .slice(0, 3)++ .map(r => ({ username: r.username, display_name: r.display_name || r.username, avatar_url: r.avatar_url || null, profile_color: r.profile_color || null, n: Number(r.count) }));const avg = safe(() => db.get('SELECT ROUND(AVG(viewer_count), 1) AS a, COUNT(*) AS k FROM viewer_snapshots WHERE stream_id = ?', [stream.id]), { a: null, k: 0 });stats = {-- chat_messages: Number(chat.n) || 0, chatters: Number(chat.c) || 0,++ chat_messages: Number(chatStats.messages) || 0, chatters: Number(chatStats.chatters) || 0,peak_viewers: Number(stream.peak_viewers) || 0, avg_viewers: avg.k ? Number(avg.a) : null,-- sound_commands: safe(() => db.get("SELECT COUNT(*) AS n FROM chat_messages WHERE stream_id = ? AND message_type = 'soundboard'", [stream.id]).n, 0),++ sound_commands: Number(chatStats.sounds) || 0,mic_moments: safe(() => db.get('SELECT COUNT(*) AS n FROM arena_mic_moments WHERE stream_id = ?', [stream.id]).n, 0),follows_gained: safe(() => db.get('SELECT COUNT(*) AS n FROM follows WHERE streamer_id = ? AND created_at BETWEEN ? AND ?', [stream.user_id, start, end]).n, 0),tips: safe(() => db.get("SELECT COUNT(*) AS n, COALESCE(SUM(amount), 0) AS t FROM transactions WHERE type = 'donation' AND (stream_id = ? OR (to_user_id = ? AND created_at BETWEEN ? AND ?))", [stream.id, stream.user_id, start, end]), { n: 0, t: 0 }),-- top_chatters: safe(() => db.all('SELECT c.username, u.display_name, u.avatar_url, u.profile_color, COUNT(*) AS n FROM chat_messages c LEFT JOIN users u ON u.id = c.user_id WHERE c.stream_id = ? AND COALESCE(c.is_deleted, 0) = 0 AND c.username IS NOT NULL AND (c.user_id IS NULL OR c.user_id != ?) GROUP BY COALESCE(c.user_id, c.username) ORDER BY n DESC LIMIT 3', [stream.id, stream.user_id]), []),++ top_chatters: topChatters,};}let clips = [];@@@ -14,6 +14,7 @@'use strict';const db = require('../db/database');++const chatReads = require('../chat/chat-reads');const MIN_DURATION_SEC = 8 * 60; // shorter streams don't get a reportconst LOOKBACK_HOURS = 12; // job only looks at streams that ended this recently@@@ -58,14 +59,12 @@ async function gather(streamId) {const viewersAvg = curve.length ? Math.round(curve.reduce((n, p) => n + Number(p.v || 0), 0) / curve.length * 10) / 10 : null;const peakAt = curve.length ? curve.reduce((best, p) => (Number(p.v) > Number(best.v) ? p : best), curve[0]) : null;-- // Chat.-- const chat = safe(() => db.get(`SELECT COUNT(*) AS n, COUNT(DISTINCT COALESCE(user_id, anon_id, username)) AS chatters FROM chat_messages WHERE stream_id = ? AND COALESCE(is_deleted, 0) = 0`, [streamId]), { n: 0, chatters: 0 });-- const topChatters = safe(() => db.all(`SELECT c.username, u.display_name, u.avatar_url, u.profile_color, COUNT(*) AS n-- FROM chat_messages c LEFT JOIN users u ON u.id = c.user_id-- WHERE c.stream_id = ? AND COALESCE(c.is_deleted, 0) = 0 AND c.username IS NOT NULL AND (c.user_id IS NULL OR c.user_id != ?)-- GROUP BY COALESCE(c.user_id, c.username) ORDER BY n DESC LIMIT 5`, [streamId, stream.user_id]), []);++ // Chat: totals and top chatters from OpenVibe.Chat (Live's own tables when Live runs chat).++ const chat = (await chatReads.streamStats(streamId)) || { messages: 0, chatters: 0, sounds: 0 };++ const topChatters = ((await chatReads.topChatters({ streamId, limit: 5 })) || [])++ .filter(r => r.username && (r.user_id == null || Number(r.user_id) !== Number(stream.user_id)))++ .slice(0, 5);const busiest = curve.length ? curve.reduce((best, p) => (Number(p.c) > Number(best.c) ? p : best), curve[0]) : null;-- const sounds = safe(() => db.get(`SELECT COUNT(*) AS n FROM chat_messages WHERE stream_id = ? AND message_type = 'soundboard'`, [streamId]).n, 0);// Arena mic lines said on this stream.const micLines = safe(() => db.all(`SELECT text, quality, kind, aimed_at, sec, vod_id FROM arena_mic_moments WHERE stream_id = ? ORDER BY quality DESC, said_at ASC LIMIT 4`, [streamId]), []);@@@ -104,7 +103,7 @@ async function gather(streamId) {stream: { id: stream.id, title: stream.title || 'Untitled stream', category: stream.ai_category || stream.category || null, protocol: stream.protocol || null, started_at: stream.started_at, ended_at: stream.ended_at || end, duration_seconds: durationSec, peak_viewers: Math.max(Number(stream.peak_viewers) || 0, peakAt ? Number(peakAt.v) || 0 : 0), ai_overview: stream.ai_overview_short || stream.ai_overview || null },streamer: { id: user.id, username: user.username, display_name: user.display_name || user.username, avatar_url: user.avatar_url || null, profile_color: user.profile_color || null },viewers: { avg: viewersAvg, peak_at: peakAt ? peakAt.t : null, curve: curve.map(p => [p.t, Number(p.v) || 0, Number(p.c) || 0]) },-- chat: { messages: Number(chat.n) || 0, chatters: Number(chat.chatters) || 0, top: topChatters.map(r => ({ username: r.username, display_name: r.display_name || r.username, avatar_url: r.avatar_url || null, profile_color: r.profile_color || null, n: Number(r.n) })), busiest_at: busiest && Number(busiest.c) > 0 ? busiest.t : null, busiest_n: busiest ? Number(busiest.c) : 0, sounds },++ chat: { messages: Number(chat.messages) || 0, chatters: Number(chat.chatters) || 0, top: topChatters.map(r => ({ username: r.username, display_name: r.display_name || r.username, avatar_url: r.avatar_url || null, profile_color: r.profile_color || null, n: Number(r.count) })), busiest_at: busiest && Number(busiest.c) > 0 ? busiest.t : null, busiest_n: busiest ? Number(busiest.c) : 0, sounds: Number(chat.sounds) || 0 },mic: micLines.map(m => ({ text: m.text, quality: m.quality, kind: m.kind, aimed_at: m.aimed_at || null, sec: m.sec, vod_id: m.vod_id || null })),love: { tips: Number(tips.n) || 0, tips_total: Number(tips.total) || 0, top_tipper: topTipper ? { username: topTipper.username, display_name: topTipper.display_name || topTipper.username, total: Number(topTipper.total) } : null, follows, followers_now: followersNow },speech: { lines: speech.length, sample: speechSample },@@@ -13,6 +13,7 @@const express = require('express');const db = require('../db/database');++const chatReads = require('../chat/chat-reads');const { can } = require('../auth/permissions');const { optionalAuth } = require('../auth/auth');@@@ -176,18 +177,13 @@ router.get('/channel/:username/dashboard', optionalAuth, async (req, res) => {new_followers: s.new_followers,}));-- // Top chatters for the period++ // Top chatters for the period, from OpenVibe.Chat (Live's own tables when Live runs chat).++ // Chat groups by user/anon/relay; the dashboard wants registered chatters, hence the filter.const cutoff = new Date(Date.now() - days * 86400000).toISOString();-- const topChatters = db.all(`-- SELECT cm.user_id, cm.username, COUNT(*) as message_count-- FROM chat_messages cm-- JOIN streams s ON cm.stream_id = s.id-- WHERE s.user_id = ? AND cm.timestamp >= ? AND cm.is_deleted = 0-- AND cm.is_global = 0 AND cm.message_type = 'chat' AND cm.user_id IS NOT NULL-- GROUP BY cm.user_id-- ORDER BY message_count DESC-- LIMIT 20-- `, [channel.user_id, cutoff]);++ const topChatters = ((await chatReads.topChatters({ channelUserId: channel.user_id, since: Date.parse(cutoff), limit: 20 })) || [])++ .filter(r => r.user_id != null)++ .slice(0, 20)++ .map(r => ({ user_id: r.user_id, username: r.username, message_count: Number(r.count) }));// Top watchers by watch timeconst topWatchers = db.all(`@@@ -669,6 +669,93 @@ const sleep = (ms) => new Promise((r) => setTimeout(r, ms));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');++++ // ── Third review round ───────────────────────────────────────────++++ // (1) A window's since/until is bucketed before the cache key, so a home-stats compute++ // whose clock moved a few seconds still hits the warm entry, and a cold hero peek answers++ // null rather than a synchronous mirror scan.++ {++ const hour = 3600000;++ const w7 = Math.floor(Date.now() / hour) * hour - 7 * 86400000; // hour-aligned, 7 days back++ chatReads._reset();++ const warm = await chatReads.windowStats({ since: w7 });++ chatReads._age(chatReads.CACHE_TTL_MS + 1000, 'win:');++ readState.down = true;++ try {++ const peeked = chatReads.windowStatsPeek({ since: w7 + 3000 });++ assert.ok(peeked && peeked.messages === warm.messages, 'a since a few seconds later buckets to the same warm key');++ } finally { readState.down = false; }++ chatReads._reset();++ readState.down = true;++ try {++ assert.strictEqual(chatReads.windowStatsPeek({ since: Date.now() - 7 * 86400000 }), null, 'a cold hero peek is null, never a mirror scan');++ } finally { readState.down = false; chatReads._reset(); }++ }++++ // (2) The stream-end analytics write keeps the totals it has and re-upserts Chat's right++ // after (the clip-count pattern) instead of persisting a cold mirror (or zero).++ {++ chatReads._reset();++ d2.run('DELETE FROM stream_analytics WHERE stream_id = ?', [streamId]);++ const computed = d2.computeAndCacheStreamAnalytics(streamId);++ assert.strictEqual(computed.total_messages, 0, 'the synchronous compute writes its prior totals, not a cold scan');++ for (let i = 0; i < 300; i++) { await sleep(10); const a = d2.getStreamAnalytics(streamId); if (a && a.total_messages > 0) break; }++ const sa = d2.getStreamAnalytics(streamId);++ assert.ok(sa.total_messages >= 3, "the setImmediate refresh wrote Chat's totals");++ assert.strictEqual(sa.unique_chatters, 2, "and Chat's chatter count");++ chatReads._reset();++ }++++ // (3) Live's own hidden row OR Chat's fresh list — a local hide the mirror applied still++ // hides even when a fresh Chat list lacks it (the #28 rule, restored).++ {++ const chatClient = require('../server/chat/chat-client');++ const realReadRelayUsers = chatClient.readRelayUsers;++ const relOwner = mkUser('relch');++ db.createChannel({ user_id: relOwner, title: 'Relay channel' });++ const freshCh = db.getChannelByUserId(relOwner).id; // no hidden rows yet++ chatClient.readRelayUsers = async () => ({ relay_users: [] }); // Chat's list holds nothing++ chatReads._reset();++ try {++ d2.run('DELETE FROM hidden_relay_users WHERE channel_id = ?', [freshCh]);++ chatReads.isRelayUserHidden(freshCh, 'twitch', 'nobody'); // warm Chat's fresh (empty) list++ for (let i = 0; i < 100; i++) await sleep(10);++ d2.run("INSERT INTO hidden_relay_users (channel_id, platform, external_username, action, created_by) VALUES (?, 'twitch', 'localonly', 'hide', ?)", [freshCh, streamer]);++ assert.strictEqual(chatReads.isRelayUserHidden(freshCh, 'twitch', 'localonly'), true, "Live's own hidden row wins over a fresh Chat list that lacks it");++ } finally {++ chatClient.readRelayUsers = realReadRelayUsers;++ d2.run('DELETE FROM hidden_relay_users WHERE channel_id = ?', [freshCh]);++ chatReads._reset();++ }++ }++++ // (4) A cold profile count is null (the card omits it), never a mirror number passed off++ // as Chat's.++ {++ chatReads._reset();++ readState.down = true;++ try {++ assert.strictEqual(chatReads.userMessageCountPeek(viewer), null, 'a cold count is null, not a mirror number');++ assert.strictEqual(d2.getUserProfile(viewer).messageCount, null, 'the profile carries the null through');++ } finally { readState.down = false; chatReads._reset(); }++ }++++ // (5) channelMaxId — the preview path's awaited read — warms the peek's key, so the++ // cursor is Chat's id rather than the mirror once it has answered.++ {++ const chatClient = require('../server/chat/chat-client');++ const realReadMessages = chatClient.readMessages;++ chatClient.readMessages = async (o) => (o && o.tail ? { max_id: 777777, messages: [] } : realReadMessages(o));++ chatReads._reset();++ try {++ await chatReads.channelMaxId(streamer);++ assert.strictEqual(chatReads.channelMaxIdPeek(streamer), 777777, "the peek answers Chat's id after the awaited read");++ } finally {++ chatClient.readMessages = realReadMessages;++ chatReads._reset();++ }++ }}// 12. Robot commands from chat pass the control panel's gate: control mode, anonymous switch, whitelist, cooldown.
@@ -49,6 +49,14 @@ with the same WebSocket protocol and REST paths. `CHAT_AUTHORITY=chat` switches
@@ -49,6 +49,14 @@ with the same WebSocket protocol and REST paths. `CHAT_AUTHORITY=chat` switches
(`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
`/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.
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.
- **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
@@ -462,7 +462,9 @@
@@ -462,7 +462,9 @@
</div>`;
</div>`;
}
}
const coins = p.openvibe_coins_balance || 0;
const coins = p.openvibe_coins_balance || 0;
- const msgs = p.messageCount || 0;
+ // Chat answers the message count; a cold cache leaves it null and the card omits the stat
+ // rather than showing a stale mirror number as real.
+ const msgs = p.messageCount == null ? null : Number(p.messageCount);
let modBtns = '';
let modBtns = '';
if (canMod || isGlobalMod) {
if (canMod || isGlobalMod) {
if (d.msgId) modBtns += `<button class="ctx-btn ctx-btn-warn" onclick="ctxDeleteMessage('${esc(String(d.msgId))}')"><i class="fa-solid fa-trash"></i> Delete Message</button>`;
if (d.msgId) modBtns += `<button class="ctx-btn ctx-btn-warn" onclick="ctxDeleteMessage('${esc(String(d.msgId))}')"><i class="fa-solid fa-trash"></i> Delete Message</button>`;
@@ -478,7 +480,7 @@
@@ -478,7 +480,7 @@
</div>
</div>
<div class="ctx-stats">
<div class="ctx-stats">
<div class="ctx-stat"><i class="fa-solid fa-coins"></i> ${numf(coins)}</div>
<div class="ctx-stat"><i class="fa-solid fa-coins"></i> ${numf(coins)}</div>
- <div class="ctx-stat"><i class="fa-solid fa-message"></i> ${numf(msgs)}</div>
+ ${msgs == null ? '' : `<div class="ctx-stat"><i class="fa-solid fa-message"></i> ${numf(msgs)}</div>`}
<div class="ctx-stat"><i class="fa-solid fa-heart"></i> ${numf(p.followerCount || 0)}</div>
<div class="ctx-stat"><i class="fa-solid fa-heart"></i> ${numf(p.followerCount || 0)}</div>
</div>
</div>
${gameHtml}
${gameHtml}
@@ -204,6 +204,9 @@ function personBlock(line, channelUserId, settings) {
@@ -204,6 +204,9 @@ function personBlock(line, channelUserId, settings) {
}
}
}
}
const key = line.userId ? `user:${line.username}` : (line.anonId ? `anon:${line.anonId}` : `ext:${line.username}`);
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');
if (line.userId !== channelUserId && settings.greet_first_timers && db.isFirstChatInChannel && db.isFirstChatInChannel(key, channelUserId)) 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) : '';
@@ -10,6 +10,7 @@
@@ -10,6 +10,7 @@
*/
*/
'use strict';
'use strict';
const db = require('../../db/database');
const db = require('../../db/database');
+const chatReads = require('../../chat/chat-reads');
const settingsMod = require('./settings');
const settingsMod = require('./settings');
const roster = require('./roster');
const roster = require('./roster');
const budget = require('./budget');
const budget = require('./budget');
@@ -56,7 +57,7 @@ class AiViewersEngineV3 {
@@ -56,7 +57,7 @@ class AiViewersEngineV3 {
const worker = {
const worker = {
streamId: stream.id, userId: stream.user_id, stream: full, cfg, settings, bots,
streamId: stream.id, userId: stream.user_id, stream: full, cfg, settings, bots,
stable: null, // cached stable prefix {text,hash,at}
stable: null, // cached stable prefix {text,hash,at}
- lastChatId: db.getMaxChatMessageIdForChannel(stream.user_id),
+ lastChatId: chatReads.channelMaxIdPeek(stream.user_id), // Chat's cursor, so the first tick does not replay old lines
lastTickAt: 0, nextTickAt: 0, tickTimer: null, foldTimer: null, ticking: false,
lastTickAt: 0, nextTickAt: 0, tickTimer: null, foldTimer: null, ticking: false,
lastRealInputAt: 0, lastStreamerReplyAt: 0, lastSkipLogAt: 0, lastVolatileHash: null,
lastRealInputAt: 0, lastStreamerReplyAt: 0, lastSkipLogAt: 0, lastVolatileHash: null,
intents: [], // [{ kind:'streamer'|'mention'|'sound'|'scene', text, bots:[], at }]
intents: [], // [{ kind:'streamer'|'mention'|'sound'|'scene', text, bots:[], at }]
@@ -73,6 +74,9 @@ class AiViewersEngineV3 {
@@ -73,6 +74,9 @@ class AiViewersEngineV3 {
dropped: (line, why) => poster.log(worker, { event: 'skip', bot_username: line.bot, text: line.text, reason: why }),
dropped: (line, why) => poster.log(worker, { event: 'skip', bot_username: line.bot, text: line.text, reason: why }),
});
});
this.workers.set(stream.id, worker);
this.workers.set(stream.id, worker);
+ // The peek above may have answered Live's mirror while Chat's cache was cold; refresh the
+ // cursor from Chat before the first tick (4 s away) so it cannot replay lines from earlier.
+ chatReads.channelMaxId(stream.user_id).then((id) => { if (!worker.stopped) worker.lastChatId = Math.max(worker.lastChatId || 0, id); }).catch(() => {});
poster.log(worker, { event: 'info', reason: `engine v3 started — ${bots.length} viewer(s), ${settings.activity} activity, director every ${settings.director_interval_sec}s` });
poster.log(worker, { event: 'info', reason: `engine v3 started — ${bots.length} viewer(s), ${settings.activity} activity, director every ${settings.director_interval_sec}s` });
console.log(`[AI-Viewers v3] started for stream ${stream.id} (user ${stream.user_id}, ${bots.length} bots)`);
console.log(`[AI-Viewers v3] started for stream ${stream.id} (user ${stream.user_id}, ${bots.length} bots)`);
this._scheduleTick(worker, 4000);
this._scheduleTick(worker, 4000);
@@ -413,8 +417,11 @@ class AiViewersEngineV3 {
@@ -413,8 +417,11 @@ class AiViewersEngineV3 {
if (!live.length) return { error: 'Go live first — the preview runs on your real stream context.' };
if (!live.length) return { error: 'Go live first — the preview runs on your real stream context.' };
const cfg = db.getChannelAiConfig(userId); const settings = settingsMod.getSettings(userId, cfg);
const cfg = db.getChannelAiConfig(userId); const settings = settingsMod.getSettings(userId, cfg);
const bots = roster.ensureRoster(userId, settings.roster_size || 3);
const bots = roster.ensureRoster(userId, settings.roster_size || 3);
- w = { streamId: live[0].id, userId, stream: db.getStreamById(live[0].id) || live[0], cfg, settings, bots, lastChatId: Math.max(0, db.getMaxChatMessageIdForChannel(userId) - 40), intents: [], scheduler: { botShare: () => 0 } };
+ w = { streamId: live[0].id, userId, stream: db.getStreamById(live[0].id) || live[0], cfg, settings, bots, lastChatId: Math.max(0, chatReads.channelMaxIdPeek(userId) - 40), intents: [], scheduler: { botShare: () => 0 } };
temp = true;
temp = true;
+ // The peek above may answer Live's mirror on a cold cache; await Chat's cursor so the
+ // preview never replays lines from before the window it means to show.
+ try { w.lastChatId = Math.max(0, (await chatReads.channelMaxId(userId)) - 40); } catch { /* keep the peek */ }
}
}
const s = w.settings;
const s = w.settings;
const stable = context.stablePrefix({ userId: w.userId, stream: w.stream, bots: w.bots, settings: s, cacheHolder: temp ? null : w });
const stable = context.stablePrefix({ userId: w.userId, stream: w.stream, bots: w.bots, settings: s, cacheHolder: temp ? null : w });
@@ -273,7 +273,9 @@ router.post('/clone', requireAuth, async (req, res) => {
@@ -273,7 +273,9 @@ router.post('/clone', requireAuth, async (req, res) => {
const platform = idx > 0 ? ref.slice(0, idx) : '';
const platform = idx > 0 ? ref.slice(0, idx) : '';
const username = idx > 0 ? ref.slice(idx + 1) : '';
const username = idx > 0 ? ref.slice(idx + 1) : '';
if (!platform || !username) return res.status(400).json({ error: 'Bad relay ref' });
if (!platform || !username) return res.status(400).json({ error: 'Bad relay ref' });
- const relay = db.getRelayUser(platform, username);
+ // Chat owns relay_users and exposes no relay-identity read (only the hidden-relay
+ // moderation list and message history, both read below), so there is nothing to look up
+ // here; the clone uses the handle.
const wide = isStaff(req.user);
const wide = isStaff(req.user);
src = {
src = {
kind: 'relay', ref: `${platform}:${username}`,
kind: 'relay', ref: `${platform}:${username}`,
@@ -40,6 +40,10 @@ let lastLog = null;
@@ -40,6 +40,10 @@ let lastLog = null;
function drill() { try { return require('../drill').enabled; } catch { return false; } }
function drill() { try { return require('../drill').enabled; } catch { return false; } }
function remote() { return isRemote() && !drill(); }
function remote() { return isRemote() && !drill(); }
function local() { return require('../db/database'); }
function local() { return require('../db/database'); }
+/** Run a local (Live-table) read for a synchronous caller, answering `dflt` if it throws. */
+function safeLocal(fn, dflt) { try { const v = fn(); return v == null ? dflt : v; } catch { return dflt; } }
+/** Epoch ms → the UTC 'YYYY-MM-DD HH:MM:SS' text chat_messages.timestamp stores. */
+const sqlTime = (ms) => new Date(ms).toISOString().slice(0, 19).replace('T', ' ');
function note(what, err) {
function note(what, err) {
const m = `Chat read ${what}: ${(err && err.message) || err}`;
const m = `Chat read ${what}: ${(err && err.message) || err}`;
@@ -185,6 +189,139 @@ function topChatters({ since, streamId, channelUserId, limit = 10 } = {}) {
@@ -185,6 +189,139 @@ function topChatters({ since, streamId, channelUserId, limit = 10 } = {}) {
});
});
}
}
+// ── Stats over a window / one room ───────────────────────────────────────────
+// Chat's stats read takes since/until plus at most one of stream_id / channel_user_id and answers
+// { messages, chatters } for that window (kind 'stream' also answers the soundboard count). The
+// home-stats snapshot and the digest ask site-wide, the recap and stream analytics ask per stream,
+// the star picker per channel. The local (rollback / dev) answer is the same SQL Live's own chat
+// runs; a peek answers the last good value while Chat refreshes in the background.
+function localWindowStats({ since, until, streamId, channelUserId } = {}) {
+ const db = local();
+ const where = ['is_deleted = 0'];
+ const params = [];
+ if (since != null) { where.push('timestamp >= ?'); params.push(sqlTime(since)); }
+ if (until != null) { where.push('timestamp < ?'); params.push(sqlTime(until)); }
+ if (streamId != null) { where.push('stream_id = ?'); params.push(Number(streamId)); }
+ if (channelUserId != null) { where.push('channel_user_id = ?'); params.push(Number(channelUserId)); }
+ 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 ${where.join(' AND ')}`, params) || {};
+ return { messages: Number(r.messages) || 0, chatters: Number(r.chatters) || 0 };
+}
+async function remoteWindowStats({ since, until, streamId, channelUserId } = {}) {
+ const out = await client.readStats({ kind: 'site', since, until, stream_id: streamId, channel_user_id: channelUserId });
+ if (!out) throw unavailable('window stats');
+ return { messages: Number(out.messages) || 0, chatters: Number(out.chatters) || 0 };
+}
+const MIN_BUCKET_MS = 60_000; // sub-day windows snap to the minute
+const HOUR_BUCKET_MS = 60 * 60_000; // a 7/14/30-day window snaps to the hour
+/**
+ * Snap a window's since/until to a stable bucket BEFORE both the cache key and the request use them.
+ * Callers build these from Date.now(), which changes on every compute, so an exact key is cold on
+ * every call: Chat is never actually consulted (the peek answers a mirror scan instead), each call
+ * mints a new cache entry, and the request it sends is thrown away. A minute — or an hour once the
+ * window spans a day or more — is far finer than these coarse counts need and keeps the key stable
+ * well past CACHE_TTL_MS. The request is sent with the bucketed values so it matches its own key.
+ */
+function bucketWindow({ since, until, streamId, channelUserId } = {}) {
+ const span = (until != null ? Number(until) : Date.now()) - (since != null ? Number(since) : 0);
+ const g = span >= 24 * 60 * 60_000 ? HOUR_BUCKET_MS : MIN_BUCKET_MS;
+ const snap = (t) => (t == null ? undefined : Math.floor(Number(t) / g) * g);
+ return { since: snap(since), until: snap(until), streamId, channelUserId };
+}
+const windowKey = (o) => `win:${o.since || 0}:${o.until || 0}:${o.streamId || 0}:${o.channelUserId || 0}`;
+/** Message/chatter counts over [since, until), site-wide or scoped to one stream/channel (async). */
+function windowStats(o = {}) {
+ const b = bucketWindow(o);
+ return load(windowKey(b), () => remoteWindowStats(b), () => localWindowStats(b))
+ .then((v) => v || safeLocal(() => localWindowStats(b), { messages: 0, chatters: 0 }));
+}
+/**
+ * The same, for a caller that cannot await (the home-stats snapshot, the digest). Answers the last
+ * good value, else null — never a synchronous mirror scan, so the hero never presents Live's stale
+ * count as Chat's; the caller decides what an unknown window means.
+ */
+function windowStatsPeek(o = {}) {
+ const b = bucketWindow(o);
+ return peek(windowKey(b), () => remoteWindowStats(b), () => localWindowStats(b));
+}
+
+function localStreamStats(streamId) {
+ const db = local();
+ const r = db.get(`SELECT COUNT(*) AS messages,
+ COUNT(DISTINCT COALESCE(user_id, anon_id, username)) AS chatters,
+ SUM(CASE WHEN message_type = 'soundboard' THEN 1 ELSE 0 END) AS sounds
+ FROM chat_messages WHERE stream_id = ? AND COALESCE(is_deleted, 0) = 0`, [Number(streamId)]) || {};
+ return { messages: Number(r.messages) || 0, chatters: Number(r.chatters) || 0, sounds: Number(r.sounds) || 0 };
+}
+async function remoteStreamStats(streamId) {
+ const out = await client.readStats({ kind: 'stream', stream_id: Number(streamId) });
+ if (!out) throw unavailable('stream stats');
+ return { messages: Number(out.messages) || 0, chatters: Number(out.chatters) || 0, sounds: Number(out.sounds) || 0 };
+}
+/** One stream's chat totals: { messages, chatters, sounds } (the recap, stream analytics). */
+function streamStats(streamId) {
+ const id = Number(streamId) || 0;
+ if (!id) return Promise.resolve({ messages: 0, chatters: 0, sounds: 0 });
+ return load(`st:${id}`, () => remoteStreamStats(id), () => localStreamStats(id))
+ .then((v) => v || safeLocal(() => localStreamStats(id), { messages: 0, chatters: 0, sounds: 0 }));
+}
+/** The same, for a caller that cannot await. */
+function streamStatsPeek(streamId) {
+ const id = Number(streamId) || 0;
+ if (!id) return { messages: 0, chatters: 0, sounds: 0 };
+ const v = peek(`st:${id}`, () => remoteStreamStats(id), () => localStreamStats(id));
+ if (v) return v;
+ return safeLocal(() => localStreamStats(id), { messages: 0, chatters: 0, sounds: 0 });
+}
+
+async function remoteChannelMaxId(channelUserId) {
+ const out = await client.readMessages({ channel_user_id: channelUserId, tail: true });
+ if (!out) throw unavailable('channel max id');
+ return Number(out.max_id) || 0;
+}
+/**
+ * The newest chat id in a channel — Chat's id, so the AI context delta (channelMessagesPeek reads
+ * Chat's ids) stays aligned. `channelMaxId` awaits Chat (and warms the peek); `channelMaxIdPeek` is
+ * for the caller that cannot await and answers the last good id, else Live's own table.
+ */
+function channelMaxId(channelUserId) {
+ const id = Number(channelUserId) || 0;
+ if (!id) return Promise.resolve(0);
+ return load(`cmid:${id}`, () => remoteChannelMaxId(id), () => local().getMaxChatMessageIdForChannel(id))
+ .then((v) => (v != null ? Number(v) || 0 : safeLocal(() => local().getMaxChatMessageIdForChannel(id), 0)));
+}
+function channelMaxIdPeek(channelUserId) {
+ const id = Number(channelUserId) || 0;
+ if (!id) return 0;
+ const v = peek(`cmid:${id}`, () => remoteChannelMaxId(id), () => local().getMaxChatMessageIdForChannel(id));
+ if (v != null) return Number(v) || 0;
+ return safeLocal(() => local().getMaxChatMessageIdForChannel(id), 0);
+}
+
+function localUserMessageCount(userId) {
+ const r = local().get(`SELECT COUNT(*) AS c FROM chat_messages
+ WHERE user_id = ? AND is_deleted = 0
+ AND (auto_delete_at IS NULL OR datetime(auto_delete_at) > CURRENT_TIMESTAMP)`, [Number(userId)]);
+ return Number(r && r.c) || 0;
+}
+async function remoteUserMessageCount(userId) {
+ const out = await client.readStats({ kind: 'user', user_id: Number(userId) });
+ if (!out) throw unavailable('user message count');
+ return Number(out.messages) || 0;
+}
+/**
+ * A user's non-deleted chat total, for a caller that cannot await (the profile card). Answers the
+ * last good value, else null when neither Chat nor (in dev / rollback) Live's table answered — the
+ * card omits the count then rather than presenting a cold mirror number as real.
+ */
+function userMessageCountPeek(userId) {
+ const id = Number(userId) || 0;
+ if (!id) return null;
+ const v = peek(`um:${id}`, () => remoteUserMessageCount(id), () => localUserMessageCount(id));
+ return v == null ? null : Number(v) || 0;
+}
+
// ── Messages / history ────────────────────────────────────────────────────────
// ── Messages / history ────────────────────────────────────────────────────────
/**
/**
* The admin console's chat search: { messages, total }. Chat's message read takes exactly one
* The admin console's chat search: { messages, total }. Chat's message read takes exactly one
@@ -395,27 +532,32 @@ function relayUser(id) {
@@ -395,27 +532,32 @@ function relayUser(id) {
* Is this relay identity hidden (banned) in this channel or site-wide? Synchronous: the relay
* 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.
* 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.
+ * Live's own table OR Chat's list (the #28 rule): a hide the mirror applied wins at once even when a
+ * cached Chat list fetched just before it lacks the row, and a Chat row hides even before the mirror
+ * catches up. A Live-route hide/unhide invalidates the cached list; a cold cache or a Chat outage
+ * falls to Live's own table and fails OPEN, never dropping a relayed line.
*/
*/
function isRelayUserHidden(channelId, platform, username) {
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 localHidden = () => { try { return !!local().isRelayUserHidden(channelId, platform, username); } catch { return false; } };
+ if (!remote()) return localHidden();
+ // Live's own table OR Chat's fresh list — the rule agreed with #28. A fresh Chat list alone
+ // would hide a local hide the mirror applied from any other writer (another Live instance, or a
+ // write straight in Chat) for up to CACHE_TTL_MS. A Live-route unhide stays immediate: mod-routes
+ // drops the `hru:` entry on every hide/unhide, so the next check re-asks Chat. Until Chat has
+ // answered (cold cache) or while it is unreachable, Live's own table answers and fails OPEN —
+ // never drop a relayed line on a Chat outage.
const key = `hru:${Number(channelId) || 0}`;
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
+ const hit = fresh(key);
if (hit && Array.isArray(hit.value)) {
if (hit && Array.isArray(hit.value)) {
- return hit.value.some((r) => String(r.platform) === String(platform) && String(r.external_username) === String(username));
+ if (hit.value.some((r) => String(r.platform) === String(platform) && String(r.external_username) === String(username))) return true;
+ return localHidden(); // no fresh Chat match: a local hide still applies
}
}
load(key, async () => {
load(key, async () => {
const out = await client.readRelayUsers({ channel_id: Number(channelId) || undefined, limit: 500 });
const out = await client.readRelayUsers({ channel_id: Number(channelId) || undefined, limit: 500 });
if (!out || !Array.isArray(out.relay_users)) throw unavailable('hidden relay users');
if (!out || !Array.isArray(out.relay_users)) throw unavailable('hidden relay users');
return out.relay_users;
return out.relay_users;
}, () => []).catch(() => { /* warms the cache for the next check */ });
}, () => []).catch(() => { /* warms the cache for the next check */ });
- return false;
+ return localHidden();
}
}
/** A user's TTS voice override: { voice, pitch, speed, gap } | null. Live, never cached. */
/** A user's TTS voice override: { voice, pitch, speed, gap } | null. Live, never cached. */
@@ -511,6 +653,7 @@ function _cacheMax(n) { if (n != null && Number(n) > 0) cacheMax = Number(n); re
@@ -511,6 +653,7 @@ function _cacheMax(n) { if (n != null && Number(n) > 0) cacheMax = Number(n); re
module.exports = {
module.exports = {
siteStats, siteStatsPeek, topChatters,
siteStats, siteStatsPeek, topChatters,
+ 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,
pendingIp, relayUsers, relayUser, ttsOverride, ttsOverridePeek, isRelayUserHidden,
pendingIp, relayUsers, relayUser, ttsOverride, ttsOverridePeek, isRelayUserHidden,
@@ -3805,6 +3805,9 @@ const HOME_SERIES = {
@@ -3805,6 +3805,9 @@ 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.
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(*)' },
@@ -3903,6 +3906,7 @@ function getHomeStats() {
@@ -3903,6 +3906,7 @@ function getHomeStats() {
function _computeHomeStats() {
function _computeHomeStats() {
// Each stat is isolated so a missing table / column can never blank the whole hero.
// Each stat is isolated so a missing table / column can never blank the whole hero.
const c = (sql, p = []) => { try { return get(sql, p)?.count || 0; } catch { return 0; } };
const c = (sql, p = []) => { try { return get(sql, p)?.count || 0; } catch { return 0; } };
+ const nowMs = Date.now(); // the home-stats snapshot's windows are relative to this instant
// Vibes tipped before this instant were test money (site setting `stats_vibes_reset_at`,
// Vibes tipped before this instant were test money (site setting `stats_vibes_reset_at`,
// ISO timestamp). Everything Vibes-related on the hero starts counting from it.
// ISO timestamp). Everything Vibes-related on the hero starts counting from it.
const vibesSince = vibesStatsSince();
const vibesSince = vibesStatsSince();
@@ -3974,32 +3978,17 @@ function _computeHomeStats() {
@@ -3974,32 +3978,17 @@ function _computeHomeStats() {
// Total hours of video the platform has archived (OpenVibe.Media's figure).
// Total hours of video the platform has archived (OpenVibe.Media's figure).
streamHours: null,
streamHours: null,
// Active chatters this week across EVERYONE — registered users, anons, and relay chatters.
// Active chatters this week across EVERYONE — registered users, anons, and relay chatters.
- weeklyActive: c(`SELECT COUNT(*) AS count FROM (
- SELECT DISTINCT 'u' || user_id AS id FROM chat_messages
- WHERE user_id IS NOT NULL AND COALESCE(is_deleted, 0) = 0 AND timestamp >= datetime('now', '-7 days')
- UNION
- SELECT DISTINCT 'a' || anon_id FROM chat_messages
- WHERE anon_id IS NOT NULL AND COALESCE(is_deleted, 0) = 0 AND timestamp >= datetime('now', '-7 days')
- UNION
- SELECT DISTINCT 'r' || source_platform || '|' || username FROM chat_messages
- WHERE source_platform IS NOT NULL AND source_platform <> '' AND COALESCE(is_deleted, 0) = 0 AND timestamp >= datetime('now', '-7 days')
- )`),
+ // OpenVibe.Chat counts them (stats kind 'site' over the window); Live's own tables when Live
+ // runs chat. The peek answers the last good count or null (never a mirror scan or a 500);
+ // the hero treats null as unknown.
+ weeklyActive: (() => { try { const s = require('../chat/chat-reads').windowStatsPeek({ since: nowMs - 7 * 86400000 }); return s && s.chatters != null ? s.chatters : null; } catch { return null; } })(),
// New unique visitors this week (first-seen anon fingerprints) — a proxy for people who
// New unique visitors this week (first-seen anon fingerprints) — a proxy for people who
// showed up, not just those who chatted.
// showed up, not just those who chatted.
weeklyVisitors: c(`SELECT COUNT(*) AS count FROM anon_ip_mappings WHERE created_at >= datetime('now', '-7 days')`),
weeklyVisitors: c(`SELECT COUNT(*) AS count FROM anon_ip_mappings WHERE created_at >= datetime('now', '-7 days')`),
// The same two windows again, shifted back a week, so the hero can say whether this week
// The same two windows again, shifted back a week, so the hero can say whether this week
// beat last week rather than just how big it was.
// beat last week rather than just how big it was.
prevWeeklyVisitors: c(`SELECT COUNT(*) AS count FROM anon_ip_mappings WHERE created_at >= datetime('now', '-14 days') AND created_at < datetime('now', '-7 days')`),
prevWeeklyVisitors: c(`SELECT COUNT(*) AS count FROM anon_ip_mappings WHERE created_at >= datetime('now', '-14 days') AND created_at < datetime('now', '-7 days')`),
- prevWeeklyActive: c(`SELECT COUNT(*) AS count FROM (
- SELECT DISTINCT 'u' || user_id AS id FROM chat_messages
- WHERE user_id IS NOT NULL AND COALESCE(is_deleted, 0) = 0 AND timestamp >= datetime('now', '-14 days') AND timestamp < datetime('now', '-7 days')
- UNION
- SELECT DISTINCT 'a' || anon_id FROM chat_messages
- WHERE anon_id IS NOT NULL AND COALESCE(is_deleted, 0) = 0 AND timestamp >= datetime('now', '-14 days') AND timestamp < datetime('now', '-7 days')
- UNION
- SELECT DISTINCT 'r' || source_platform || '|' || username FROM chat_messages
- WHERE source_platform IS NOT NULL AND source_platform <> '' AND COALESCE(is_deleted, 0) = 0 AND timestamp >= datetime('now', '-14 days') AND timestamp < datetime('now', '-7 days')
- )`),
+ prevWeeklyActive: (() => { try { const s = require('../chat/chat-reads').windowStatsPeek({ since: nowMs - 14 * 86400000, until: nowMs - 7 * 86400000 }); return s && s.chatters != null ? s.chatters : null; } catch { return null; } })(),
liveNow: c(`SELECT COUNT(*) AS count FROM streams WHERE is_live = 1`),
liveNow: c(`SELECT COUNT(*) AS count FROM streams WHERE is_live = 1`),
// Rolling last-day / week / month deltas ({ d, w, m }) for the hero stat tooltips + subs.
// Rolling last-day / week / month deltas ({ d, w, m }) for the hero stat tooltips + subs.
recent: {
recent: {
@@ -4009,7 +3998,16 @@ function _computeHomeStats() {
@@ -4009,7 +3998,16 @@ function _computeHomeStats() {
vods: null, // OpenVibe.Media (withArchiveStats)
vods: null, // OpenVibe.Media (withArchiveStats)
clips: null,
clips: null,
aiMoments: winCount('stream_memories', 'created_at'),
aiMoments: winCount('stream_memories', 'created_at'),
- messages: winCount('chat_messages', 'timestamp', 'COALESCE(is_deleted, 0) = 0'),
+ // OpenVibe.Chat's message counts over each window (Live's own tables when Live runs chat).
+ messages: (() => {
+ const w = (o) => { try { const s = require('../chat/chat-reads').windowStatsPeek(o); return s && s.messages != null ? s.messages : null; } catch { return null; } };
+ return {
+ d: w({ since: nowMs - 86400000 }),
+ w: w({ since: nowMs - 7 * 86400000 }),
+ m: w({ since: nowMs - 30 * 86400000 }),
+ pw: w({ since: nowMs - 14 * 86400000, until: nowMs - 7 * 86400000 }),
+ };
+ })(),
hours: null, // OpenVibe.Media
hours: null, // OpenVibe.Media
streamers: streamersWin(),
streamers: streamersWin(),
// Live's `emotes` copy was unread since N+2 and is now dropped, so the
// Live's `emotes` copy was unread since N+2 and is now dropped, so the
@@ -4688,12 +4686,6 @@ function recordRelayUser(platform, username) {
@@ -4688,12 +4686,6 @@ function recordRelayUser(platform, username) {
[String(platform).toLowerCase(), key, String(username)]);
[String(platform).toLowerCase(), key, String(username)]);
} catch { /* non-critical */ }
} catch { /* non-critical */ }
}
}
-function getRelayUser(platform, username) {
- if (!platform || !username) return null;
- // rowid is a stable integer id for a relay user (no dedicated id column).
- return get('SELECT rowid AS id, * FROM relay_users WHERE platform = ? AND username = ?',
- [String(platform).toLowerCase(), String(username).toLowerCase()]) || null;
-}
function getRelayUserByRowid(id) {
function getRelayUserByRowid(id) {
return get('SELECT rowid AS id, * FROM relay_users WHERE rowid = ?', [id]) || null;
return get('SELECT rowid AS id, * FROM relay_users WHERE rowid = ?', [id]) || null;
}
}
@@ -4776,12 +4768,10 @@ function getUserProfile(userId) {
@@ -4776,12 +4768,10 @@ function getUserProfile(userId) {
openvibe_bucks_balance, openvibe_coins_balance, created_at, last_seen
openvibe_bucks_balance, openvibe_coins_balance, created_at, last_seen
FROM users WHERE id = ?`, [userId]);
FROM users WHERE id = ?`, [userId]);
if (!user) return null;
if (!user) return null;
- user.messageCount = get(
- `SELECT COUNT(*) as c FROM chat_messages
- WHERE user_id = ? AND is_deleted = 0
- AND (auto_delete_at IS NULL OR datetime(auto_delete_at) > CURRENT_TIMESTAMP)`,
- [userId]
- )?.c || 0;
+ // The user's chat total is OpenVibe.Chat's (Live's own tables when Live runs chat); a synchronous
+ // peek answers the last good count, else null while Chat refreshes in the background. The profile
+ // card omits the count when it is null rather than showing a cold mirror number as real.
+ user.messageCount = (() => { try { return require('../chat/chat-reads').userMessageCountPeek(userId); } catch { return null; } })();
user.followerCount = get('SELECT COUNT(*) as c FROM follows WHERE streamer_id = ?', [userId])?.c || 0;
user.followerCount = get('SELECT COUNT(*) as c FROM follows WHERE streamer_id = ?', [userId])?.c || 0;
user.followingCount = get('SELECT COUNT(*) as c FROM follows WHERE follower_id = ?', [userId])?.c || 0;
user.followingCount = get('SELECT COUNT(*) as c FROM follows WHERE follower_id = ?', [userId])?.c || 0;
return user;
return user;
@@ -6650,21 +6640,22 @@ function computeAndCacheStreamAnalytics(streamId) {
@@ -6650,21 +6640,22 @@ function computeAndCacheStreamAnalytics(streamId) {
);
);
const avgViewers = avgRow?.avg_vc || 0;
const avgViewers = avgRow?.avg_vc || 0;
- // Unique chatters
- const chattersRow = get(
- `SELECT COUNT(DISTINCT COALESCE(user_id, anon_id)) as cnt
- FROM chat_messages WHERE stream_id = ? AND is_deleted = 0 AND is_global = 0`,
- [streamId]
- );
- const uniqueChatters = chattersRow?.cnt || 0;
-
- // Total messages
- const msgsRow = get(
- `SELECT COUNT(*) as cnt FROM chat_messages
- WHERE stream_id = ? AND is_deleted = 0 AND is_global = 0 AND message_type = 'chat'`,
- [streamId]
- );
- const totalMessages = msgsRow?.cnt || 0;
+ // Unique chatters + total messages come from OpenVibe.Chat (Live's own tables when Live runs
+ // chat). Chat's stream stats count every message type — Live used to keep only message_type='chat'
+ // and is_global=0 — and its numbers are the authority now. This runs synchronously at stream end,
+ // when `st:<id>` is still cold, so a peek would answer the mirror (or 0 on a throw): it keeps the
+ // totals already stored and asks Chat right after (setStreamAnalyticsChatTotals writes them back),
+ // the same pattern as the clip count below.
+ const prior = get('SELECT unique_chatters, total_messages FROM stream_analytics WHERE stream_id = ?', [streamId]) || {};
+ const uniqueChatters = Number(prior.unique_chatters) || 0;
+ const totalMessages = Number(prior.total_messages) || 0;
+ setImmediate(() => {
+ try {
+ require('../chat/chat-reads').streamStats(streamId)
+ .then((t) => { if (t) setStreamAnalyticsChatTotals(streamId, t.chatters, t.messages); })
+ .catch(() => {});
+ } catch { /* */ }
+ });
// Total watch minutes
// Total watch minutes
const watchRow = get(
const watchRow = get(
@@ -6737,6 +6728,13 @@ function setStreamAnalyticsClipCount(streamId, count) {
@@ -6737,6 +6728,13 @@ function setStreamAnalyticsClipCount(streamId, count) {
return run('UPDATE stream_analytics SET clips_created = ? WHERE stream_id = ?', [Math.max(0, Math.floor(Number(count) || 0)), streamId]);
return run('UPDATE stream_analytics SET clips_created = ? WHERE stream_id = ?', [Math.max(0, Math.floor(Number(count) || 0)), streamId]);
}
}
+// The chat totals OpenVibe.Chat reported for a stream (the setImmediate refresh in
+// computeAndCacheStreamAnalytics).
+function setStreamAnalyticsChatTotals(streamId, chatters, messages) {
+ return run('UPDATE stream_analytics SET unique_chatters = ?, total_messages = ? WHERE stream_id = ?',
+ [Math.max(0, Math.floor(Number(chatters) || 0)), Math.max(0, Math.floor(Number(messages) || 0)), streamId]);
+}
+
function getChannelAnalyticsSummary(userId, days) {
function getChannelAnalyticsSummary(userId, days) {
const cutoff = new Date(Date.now() - days * 86400000).toISOString();
const cutoff = new Date(Date.now() - days * 86400000).toISOString();
@@ -7129,7 +7127,7 @@ module.exports = {
@@ -7129,7 +7127,7 @@ module.exports = {
holdMessageForApproval, getPendingIpMessages, reviewPendingIpMessage,
holdMessageForApproval, getPendingIpMessages, reviewPendingIpMessage,
approveAllFromIp, denyAllFromIp,
approveAllFromIp, denyAllFromIp,
// Hidden Relay Users
// Hidden Relay Users
- hideRelayUser, isRelayUserHidden, unhideRelayUser, unhideRelayUserByIdentity, getHiddenRelayUsers, recordRelayUser, getRelayUser,
+ hideRelayUser, isRelayUserHidden, unhideRelayUser, unhideRelayUserByIdentity, getHiddenRelayUsers, recordRelayUser,
getRelayUserByRowid, getRelayUserChatHistory,
getRelayUserByRowid, getRelayUserChatHistory,
anonSubjectId, getAnonMeta, getAnonChatHistory,
anonSubjectId, getAnonMeta, getAnonChatHistory,
// IP Tracking
// IP Tracking
@@ -7137,7 +7135,7 @@ module.exports = {
@@ -7137,7 +7135,7 @@ module.exports = {
getLatestIpForUser, getLatestIpForAnon, getIpLog, banAllAccountsOnIp,
getLatestIpForUser, getLatestIpForAnon, getIpLog, banAllAccountsOnIp,
// Stream Analytics
// Stream Analytics
insertViewerSnapshot, getViewerSnapshots, computeAndCacheStreamAnalytics,
insertViewerSnapshot, getViewerSnapshots, computeAndCacheStreamAnalytics,
- getStreamAnalytics, setStreamAnalyticsClipCount, getChannelAnalyticsSummary, getRecentChatActivity,
+ getStreamAnalytics, setStreamAnalyticsClipCount, setStreamAnalyticsChatTotals, getChannelAnalyticsSummary, getRecentChatActivity,
// User Preferences
// User Preferences
getUserPreferences, saveUserPreferences,
getUserPreferences, saveUserPreferences,
// Chat Log Management
// Chat Log Management
@@ -468,7 +468,9 @@ router.get('/digest', optionalAuth, (req, res) => {
@@ -468,7 +468,9 @@ router.get('/digest', optionalAuth, (req, res) => {
const stats = {
const stats = {
streams: Number((one('SELECT COUNT(*) AS n FROM streams WHERE started_at >= ?', [sinceSql]) || {}).n || 0),
streams: Number((one('SELECT COUNT(*) AS n FROM streams WHERE started_at >= ?', [sinceSql]) || {}).n || 0),
hours: Number((one(`SELECT ROUND(SUM(COALESCE(duration_seconds, CASE WHEN ended_at IS NOT NULL THEN (julianday(ended_at) - julianday(started_at)) * 86400 ELSE (julianday('now') - julianday(started_at)) * 86400 END)) / 3600.0, 1) AS h FROM streams WHERE started_at >= ?`, [sinceSql]) || {}).h || 0),
hours: Number((one(`SELECT ROUND(SUM(COALESCE(duration_seconds, CASE WHEN ended_at IS NOT NULL THEN (julianday(ended_at) - julianday(started_at)) * 86400 ELSE (julianday('now') - julianday(started_at)) * 86400 END)) / 3600.0, 1) AS h FROM streams WHERE started_at >= ?`, [sinceSql]) || {}).h || 0),
- chat_lines: Number((one("SELECT COUNT(*) AS n FROM chat_messages WHERE timestamp >= ? AND COALESCE(is_deleted, 0) = 0", [sinceSql]) || {}).n || 0),
+ // OpenVibe.Chat's message count since the window opened (Live's own tables in dev /
+ // rollback); a synchronous peek answers the last good count while Chat refreshes.
+ chat_lines: (() => { try { const s = require('../chat/chat-reads').windowStatsPeek({ since: sinceMs }); return s ? s.messages : 0; } catch { return 0; } })(),
new_follows: Number((one('SELECT COUNT(*) AS n FROM follows WHERE created_at >= ?', [sinceSql]) || {}).n || 0),
new_follows: Number((one('SELECT COUNT(*) AS n FROM follows WHERE created_at >= ?', [sinceSql]) || {}).n || 0),
new_members: Number((one('SELECT COUNT(*) AS n FROM users WHERE created_at >= ?', [sinceSql]) || {}).n || 0),
new_members: Number((one('SELECT COUNT(*) AS n FROM users WHERE created_at >= ?', [sinceSql]) || {}).n || 0),
mic_moments: Number((one('SELECT COUNT(*) AS n FROM arena_mic_moments WHERE said_at >= ?', [sinceSql]) || {}).n || 0),
mic_moments: Number((one('SELECT COUNT(*) AS n FROM arena_mic_moments WHERE said_at >= ?', [sinceSql]) || {}).n || 0),
@@ -13,6 +13,7 @@
@@ -13,6 +13,7 @@
*/
*/
'use strict';
'use strict';
const db = require('../db/database');
const db = require('../db/database');
+const chatReads = require('../chat/chat-reads');
const ROTATE_MS = 24 * 60 * 60 * 1000; // a new star every day
const ROTATE_MS = 24 * 60 * 60 * 1000; // a new star every day
const CHECK_MS = 30 * 60 * 1000; // how often the job looks at the clock
const CHECK_MS = 30 * 60 * 1000; // how often the job looks at the clock
@@ -29,7 +30,8 @@ function loadPick() {
@@ -29,7 +30,8 @@ function loadPick() {
function pinned() { try { return String(db.getSetting('star_streamer_pinned') || '').trim(); } catch { return ''; } }
function pinned() { try { return String(db.getSetting('star_streamer_pinned') || '').trim(); } catch { return ''; } }
/** Everyone who streamed recently, with the numbers the picker looks at. */
/** Everyone who streamed recently, with the numbers the picker looks at. */
-function candidates() {
+async function candidates() {
+ const sinceMs = Date.now() - CANDIDATE_DAYS * 86400000;
const rows = db.all(`
const rows = db.all(`
SELECT u.id, u.username, u.display_name, u.bio,
SELECT u.id, u.username, u.display_name, u.bio,
COUNT(s.id) AS sessions,
COUNT(s.id) AS sessions,
@@ -37,7 +39,6 @@ function candidates() {
@@ -37,7 +39,6 @@ function candidates() {
MAX(COALESCE(s.peak_viewers, 0)) AS peak_viewers,
MAX(COALESCE(s.peak_viewers, 0)) AS peak_viewers,
ROUND(AVG(COALESCE(s.peak_viewers, 0)), 1) AS avg_peak,
ROUND(AVG(COALESCE(s.peak_viewers, 0)), 1) AS avg_peak,
MAX(s.started_at) AS last_live_at,
MAX(s.started_at) AS last_live_at,
- (SELECT COUNT(*) FROM chat_messages c WHERE c.stream_id IN (SELECT id FROM streams x WHERE x.user_id = u.id AND x.started_at >= datetime('now', ?)) AND COALESCE(c.is_deleted, 0) = 0) AS chat_lines,
(SELECT COUNT(*) FROM follows f WHERE f.streamer_id = u.id AND f.created_at >= datetime('now', ?)) AS new_followers,
(SELECT COUNT(*) FROM follows f WHERE f.streamer_id = u.id AND f.created_at >= datetime('now', ?)) AS new_followers,
(SELECT COUNT(*) FROM follows f WHERE f.streamer_id = u.id) AS followers,
(SELECT COUNT(*) FROM follows f WHERE f.streamer_id = u.id) AS followers,
(SELECT COUNT(*) FROM arena_mic_moments m WHERE m.user_id = u.id AND m.said_at >= datetime('now', ?)) AS mic_moments,
(SELECT COUNT(*) FROM arena_mic_moments m WHERE m.user_id = u.id AND m.said_at >= datetime('now', ?)) AS mic_moments,
@@ -46,7 +47,13 @@ function candidates() {
@@ -46,7 +47,13 @@ function candidates() {
FROM users u JOIN streams s ON s.user_id = u.id
FROM users u JOIN streams s ON s.user_id = u.id
WHERE s.started_at >= datetime('now', ?) AND COALESCE(u.is_banned, 0) = 0
WHERE s.started_at >= datetime('now', ?) AND COALESCE(u.is_banned, 0) = 0
GROUP BY u.id
GROUP BY u.id
- ORDER BY hours DESC LIMIT 40`, [`-${CANDIDATE_DAYS} days`, `-${CANDIDATE_DAYS} days`, `-${CANDIDATE_DAYS} days`, `-${CANDIDATE_DAYS} days`]) || [];
+ ORDER BY hours DESC LIMIT 40`, [`-${CANDIDATE_DAYS} days`, `-${CANDIDATE_DAYS} days`, `-${CANDIDATE_DAYS} days`]) || [];
+ // Chat lines per streamer over the window, from OpenVibe.Chat (Live's own tables when Live runs
+ // chat). One read per candidate; the star pick is a daily background job, so the fan-out is cheap.
+ await Promise.all(rows.map(async (r) => {
+ try { const s = await chatReads.windowStats({ channelUserId: r.id, since: sinceMs }); r.chat_lines = (s && s.messages) || 0; }
+ catch { r.chat_lines = 0; }
+ }));
let i18n = null; try { i18n = require('../i18n/translate'); } catch { i18n = null; }
let i18n = null; try { i18n = require('../i18n/translate'); } catch { i18n = null; }
for (const r of rows) {
for (const r of rows) {
r.language = i18n ? (() => { try { return i18n.channelLanguage(r.id); } catch { return 'en'; } })() : 'en';
r.language = i18n ? (() => { try { return i18n.channelLanguage(r.id); } catch { return 'en'; } })() : 'en';
@@ -91,7 +98,7 @@ async function rotate({ force = false } = {}) {
@@ -91,7 +98,7 @@ async function rotate({ force = false } = {}) {
if (!force && cur && cur.next_at && now < Number(cur.next_at)) return { current: cur.username, next_at: cur.next_at };
if (!force && cur && cur.next_at && now < Number(cur.next_at)) return { current: cur.username, next_at: cur.next_at };
_busy = true;
_busy = true;
try {
try {
- const cands = candidates();
+ const cands = await candidates();
if (!cands.length) return { none: true };
if (!cands.length) return { none: true };
// Recent stars (and whatever the setting / env currently holds) are ineligible.
// Recent stars (and whatever the setting / env currently holds) are ineligible.
const history = Array.isArray(cur && cur.history) ? cur.history.slice() : [];
const history = Array.isArray(cur && cur.history) ? cur.history.slice() : [];
@@ -678,6 +678,8 @@ class RobotStreamerService {
@@ -678,6 +678,8 @@ 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)) {
if (scmd && db.getChannelSoundByCommand(stream.user_id, scmd)) {
if (delivery.ingress()) {
if (delivery.ingress()) {
delivery.event({ kind: 'stream', id: stream.id }, {
delivery.event({ kind: 'stream', id: stream.id }, {
@@ -11,6 +11,7 @@
@@ -11,6 +11,7 @@
const express = require('express');
const express = require('express');
const multer = require('multer');
const multer = require('multer');
const db = require('../db/database');
const db = require('../db/database');
+const chatReads = require('../chat/chat-reads');
const media = require('../media-client');
const media = require('../media-client');
const recorder = require('../streaming/recorder');
const recorder = require('../streaming/recorder');
const { requireAuth, optionalAuth } = require('../auth/auth');
const { requireAuth, optionalAuth } = require('../auth/auth');
@@ -430,16 +431,22 @@ router.get('/:id/context', optionalAuth, async (req, res) => {
@@ -430,16 +431,22 @@ router.get('/:id/context', optionalAuth, async (req, res) => {
let stats = null;
let stats = null;
if (stream) {
if (stream) {
const start = String(stream.started_at || ''), end = String(stream.ended_at || new Date().toISOString().replace('T', ' ').slice(0, 19));
const start = String(stream.started_at || ''), end = String(stream.ended_at || new Date().toISOString().replace('T', ' ').slice(0, 19));
- const chat = safe(() => db.get('SELECT COUNT(*) AS n, COUNT(DISTINCT COALESCE(user_id, anon_id, username)) AS c FROM chat_messages WHERE stream_id = ? AND COALESCE(is_deleted, 0) = 0', [stream.id]), { n: 0, c: 0 });
+ // Chat totals, the sound count and the top chatters come from OpenVibe.Chat (Live's own
+ // tables when Live runs chat); a Chat outage answers a cached or zero value, never a 500.
+ const chatStats = (await chatReads.streamStats(stream.id)) || { messages: 0, chatters: 0, sounds: 0 };
+ const topChatters = ((await chatReads.topChatters({ streamId: stream.id, limit: 3 })) || [])
+ .filter(r => r.username && (r.user_id == null || Number(r.user_id) !== Number(stream.user_id)))
+ .slice(0, 3)
+ .map(r => ({ username: r.username, display_name: r.display_name || r.username, avatar_url: r.avatar_url || null, profile_color: r.profile_color || null, n: Number(r.count) }));
const avg = safe(() => db.get('SELECT ROUND(AVG(viewer_count), 1) AS a, COUNT(*) AS k FROM viewer_snapshots WHERE stream_id = ?', [stream.id]), { a: null, k: 0 });
const avg = safe(() => db.get('SELECT ROUND(AVG(viewer_count), 1) AS a, COUNT(*) AS k FROM viewer_snapshots WHERE stream_id = ?', [stream.id]), { a: null, k: 0 });
stats = {
stats = {
- chat_messages: Number(chat.n) || 0, chatters: Number(chat.c) || 0,
+ chat_messages: Number(chatStats.messages) || 0, chatters: Number(chatStats.chatters) || 0,
peak_viewers: Number(stream.peak_viewers) || 0, avg_viewers: avg.k ? Number(avg.a) : null,
peak_viewers: Number(stream.peak_viewers) || 0, avg_viewers: avg.k ? Number(avg.a) : null,
- sound_commands: safe(() => db.get("SELECT COUNT(*) AS n FROM chat_messages WHERE stream_id = ? AND message_type = 'soundboard'", [stream.id]).n, 0),
+ sound_commands: Number(chatStats.sounds) || 0,
mic_moments: safe(() => db.get('SELECT COUNT(*) AS n FROM arena_mic_moments WHERE stream_id = ?', [stream.id]).n, 0),
mic_moments: safe(() => db.get('SELECT COUNT(*) AS n FROM arena_mic_moments WHERE stream_id = ?', [stream.id]).n, 0),
follows_gained: safe(() => db.get('SELECT COUNT(*) AS n FROM follows WHERE streamer_id = ? AND created_at BETWEEN ? AND ?', [stream.user_id, start, end]).n, 0),
follows_gained: safe(() => db.get('SELECT COUNT(*) AS n FROM follows WHERE streamer_id = ? AND created_at BETWEEN ? AND ?', [stream.user_id, start, end]).n, 0),
tips: safe(() => db.get("SELECT COUNT(*) AS n, COALESCE(SUM(amount), 0) AS t FROM transactions WHERE type = 'donation' AND (stream_id = ? OR (to_user_id = ? AND created_at BETWEEN ? AND ?))", [stream.id, stream.user_id, start, end]), { n: 0, t: 0 }),
tips: safe(() => db.get("SELECT COUNT(*) AS n, COALESCE(SUM(amount), 0) AS t FROM transactions WHERE type = 'donation' AND (stream_id = ? OR (to_user_id = ? AND created_at BETWEEN ? AND ?))", [stream.id, stream.user_id, start, end]), { n: 0, t: 0 }),
- top_chatters: safe(() => db.all('SELECT c.username, u.display_name, u.avatar_url, u.profile_color, COUNT(*) AS n FROM chat_messages c LEFT JOIN users u ON u.id = c.user_id WHERE c.stream_id = ? AND COALESCE(c.is_deleted, 0) = 0 AND c.username IS NOT NULL AND (c.user_id IS NULL OR c.user_id != ?) GROUP BY COALESCE(c.user_id, c.username) ORDER BY n DESC LIMIT 3', [stream.id, stream.user_id]), []),
+ top_chatters: topChatters,
};
};
}
}
let clips = [];
let clips = [];
@@ -14,6 +14,7 @@
@@ -14,6 +14,7 @@
'use strict';
'use strict';
const db = require('../db/database');
const db = require('../db/database');
+const chatReads = require('../chat/chat-reads');
const MIN_DURATION_SEC = 8 * 60; // shorter streams don't get a report
const MIN_DURATION_SEC = 8 * 60; // shorter streams don't get a report
const LOOKBACK_HOURS = 12; // job only looks at streams that ended this recently
const LOOKBACK_HOURS = 12; // job only looks at streams that ended this recently
@@ -58,14 +59,12 @@ async function gather(streamId) {
@@ -58,14 +59,12 @@ async function gather(streamId) {
const viewersAvg = curve.length ? Math.round(curve.reduce((n, p) => n + Number(p.v || 0), 0) / curve.length * 10) / 10 : null;
const viewersAvg = curve.length ? Math.round(curve.reduce((n, p) => n + Number(p.v || 0), 0) / curve.length * 10) / 10 : null;
const peakAt = curve.length ? curve.reduce((best, p) => (Number(p.v) > Number(best.v) ? p : best), curve[0]) : null;
const peakAt = curve.length ? curve.reduce((best, p) => (Number(p.v) > Number(best.v) ? p : best), curve[0]) : null;
- // Chat.
- const chat = safe(() => db.get(`SELECT COUNT(*) AS n, COUNT(DISTINCT COALESCE(user_id, anon_id, username)) AS chatters FROM chat_messages WHERE stream_id = ? AND COALESCE(is_deleted, 0) = 0`, [streamId]), { n: 0, chatters: 0 });
- const topChatters = safe(() => db.all(`SELECT c.username, u.display_name, u.avatar_url, u.profile_color, COUNT(*) AS n
- FROM chat_messages c LEFT JOIN users u ON u.id = c.user_id
- WHERE c.stream_id = ? AND COALESCE(c.is_deleted, 0) = 0 AND c.username IS NOT NULL AND (c.user_id IS NULL OR c.user_id != ?)
- GROUP BY COALESCE(c.user_id, c.username) ORDER BY n DESC LIMIT 5`, [streamId, stream.user_id]), []);
+ // Chat: totals and top chatters from OpenVibe.Chat (Live's own tables when Live runs chat).
+ const chat = (await chatReads.streamStats(streamId)) || { messages: 0, chatters: 0, sounds: 0 };
+ const topChatters = ((await chatReads.topChatters({ streamId, limit: 5 })) || [])
+ .filter(r => r.username && (r.user_id == null || Number(r.user_id) !== Number(stream.user_id)))
+ .slice(0, 5);
const busiest = curve.length ? curve.reduce((best, p) => (Number(p.c) > Number(best.c) ? p : best), curve[0]) : null;
const busiest = curve.length ? curve.reduce((best, p) => (Number(p.c) > Number(best.c) ? p : best), curve[0]) : null;
- const sounds = safe(() => db.get(`SELECT COUNT(*) AS n FROM chat_messages WHERE stream_id = ? AND message_type = 'soundboard'`, [streamId]).n, 0);
// Arena mic lines said on this stream.
// Arena mic lines said on this stream.
const micLines = safe(() => db.all(`SELECT text, quality, kind, aimed_at, sec, vod_id FROM arena_mic_moments WHERE stream_id = ? ORDER BY quality DESC, said_at ASC LIMIT 4`, [streamId]), []);
const micLines = safe(() => db.all(`SELECT text, quality, kind, aimed_at, sec, vod_id FROM arena_mic_moments WHERE stream_id = ? ORDER BY quality DESC, said_at ASC LIMIT 4`, [streamId]), []);
@@ -104,7 +103,7 @@ async function gather(streamId) {
@@ -104,7 +103,7 @@ async function gather(streamId) {
stream: { id: stream.id, title: stream.title || 'Untitled stream', category: stream.ai_category || stream.category || null, protocol: stream.protocol || null, started_at: stream.started_at, ended_at: stream.ended_at || end, duration_seconds: durationSec, peak_viewers: Math.max(Number(stream.peak_viewers) || 0, peakAt ? Number(peakAt.v) || 0 : 0), ai_overview: stream.ai_overview_short || stream.ai_overview || null },
stream: { id: stream.id, title: stream.title || 'Untitled stream', category: stream.ai_category || stream.category || null, protocol: stream.protocol || null, started_at: stream.started_at, ended_at: stream.ended_at || end, duration_seconds: durationSec, peak_viewers: Math.max(Number(stream.peak_viewers) || 0, peakAt ? Number(peakAt.v) || 0 : 0), ai_overview: stream.ai_overview_short || stream.ai_overview || null },
streamer: { id: user.id, username: user.username, display_name: user.display_name || user.username, avatar_url: user.avatar_url || null, profile_color: user.profile_color || null },
streamer: { id: user.id, username: user.username, display_name: user.display_name || user.username, avatar_url: user.avatar_url || null, profile_color: user.profile_color || null },
viewers: { avg: viewersAvg, peak_at: peakAt ? peakAt.t : null, curve: curve.map(p => [p.t, Number(p.v) || 0, Number(p.c) || 0]) },
viewers: { avg: viewersAvg, peak_at: peakAt ? peakAt.t : null, curve: curve.map(p => [p.t, Number(p.v) || 0, Number(p.c) || 0]) },
- chat: { messages: Number(chat.n) || 0, chatters: Number(chat.chatters) || 0, top: topChatters.map(r => ({ username: r.username, display_name: r.display_name || r.username, avatar_url: r.avatar_url || null, profile_color: r.profile_color || null, n: Number(r.n) })), busiest_at: busiest && Number(busiest.c) > 0 ? busiest.t : null, busiest_n: busiest ? Number(busiest.c) : 0, sounds },
+ chat: { messages: Number(chat.messages) || 0, chatters: Number(chat.chatters) || 0, top: topChatters.map(r => ({ username: r.username, display_name: r.display_name || r.username, avatar_url: r.avatar_url || null, profile_color: r.profile_color || null, n: Number(r.count) })), busiest_at: busiest && Number(busiest.c) > 0 ? busiest.t : null, busiest_n: busiest ? Number(busiest.c) : 0, sounds: Number(chat.sounds) || 0 },
mic: micLines.map(m => ({ text: m.text, quality: m.quality, kind: m.kind, aimed_at: m.aimed_at || null, sec: m.sec, vod_id: m.vod_id || null })),
mic: micLines.map(m => ({ text: m.text, quality: m.quality, kind: m.kind, aimed_at: m.aimed_at || null, sec: m.sec, vod_id: m.vod_id || null })),
love: { tips: Number(tips.n) || 0, tips_total: Number(tips.total) || 0, top_tipper: topTipper ? { username: topTipper.username, display_name: topTipper.display_name || topTipper.username, total: Number(topTipper.total) } : null, follows, followers_now: followersNow },
love: { tips: Number(tips.n) || 0, tips_total: Number(tips.total) || 0, top_tipper: topTipper ? { username: topTipper.username, display_name: topTipper.display_name || topTipper.username, total: Number(topTipper.total) } : null, follows, followers_now: followersNow },
speech: { lines: speech.length, sample: speechSample },
speech: { lines: speech.length, sample: speechSample },
@@ -13,6 +13,7 @@
@@ -13,6 +13,7 @@
const express = require('express');
const express = require('express');
const db = require('../db/database');
const db = require('../db/database');
+const chatReads = require('../chat/chat-reads');
const { can } = require('../auth/permissions');
const { can } = require('../auth/permissions');
const { optionalAuth } = require('../auth/auth');
const { optionalAuth } = require('../auth/auth');
@@ -176,18 +177,13 @@ router.get('/channel/:username/dashboard', optionalAuth, async (req, res) => {
@@ -176,18 +177,13 @@ router.get('/channel/:username/dashboard', optionalAuth, async (req, res) => {
new_followers: s.new_followers,
new_followers: s.new_followers,
}));
}));
- // Top chatters for the period
+ // Top chatters for the period, from OpenVibe.Chat (Live's own tables when Live runs chat).
+ // Chat groups by user/anon/relay; the dashboard wants registered chatters, hence the filter.
const cutoff = new Date(Date.now() - days * 86400000).toISOString();
const cutoff = new Date(Date.now() - days * 86400000).toISOString();
- const topChatters = db.all(`
- SELECT cm.user_id, cm.username, COUNT(*) as message_count
- FROM chat_messages cm
- JOIN streams s ON cm.stream_id = s.id
- WHERE s.user_id = ? AND cm.timestamp >= ? AND cm.is_deleted = 0
- AND cm.is_global = 0 AND cm.message_type = 'chat' AND cm.user_id IS NOT NULL
- GROUP BY cm.user_id
- ORDER BY message_count DESC
- LIMIT 20
- `, [channel.user_id, cutoff]);
+ const topChatters = ((await chatReads.topChatters({ channelUserId: channel.user_id, since: Date.parse(cutoff), limit: 20 })) || [])
+ .filter(r => r.user_id != null)
+ .slice(0, 20)
+ .map(r => ({ user_id: r.user_id, username: r.username, message_count: Number(r.count) }));
// Top watchers by watch time
// Top watchers by watch time
const topWatchers = db.all(`
const topWatchers = db.all(`
@@ -669,6 +669,93 @@ const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
@@ -669,6 +669,93 @@ const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const qUpper = await chatReads.searchMessages({ userId: viewer, query: 'HELLO', limit: 10 });
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.length, 1, 'q matches regardless of case');
assert.strictEqual(qUpper.messages[0].message, 'hello');
assert.strictEqual(qUpper.messages[0].message, 'hello');
+
+ // ── Third review round ───────────────────────────────────────────
+
+ // (1) A window's since/until is bucketed before the cache key, so a home-stats compute
+ // whose clock moved a few seconds still hits the warm entry, and a cold hero peek answers
+ // null rather than a synchronous mirror scan.
+ {
+ const hour = 3600000;
+ const w7 = Math.floor(Date.now() / hour) * hour - 7 * 86400000; // hour-aligned, 7 days back
+ chatReads._reset();
+ const warm = await chatReads.windowStats({ since: w7 });
+ chatReads._age(chatReads.CACHE_TTL_MS + 1000, 'win:');
+ readState.down = true;
+ try {
+ const peeked = chatReads.windowStatsPeek({ since: w7 + 3000 });
+ assert.ok(peeked && peeked.messages === warm.messages, 'a since a few seconds later buckets to the same warm key');
+ } finally { readState.down = false; }
+ chatReads._reset();
+ readState.down = true;
+ try {
+ assert.strictEqual(chatReads.windowStatsPeek({ since: Date.now() - 7 * 86400000 }), null, 'a cold hero peek is null, never a mirror scan');
+ } finally { readState.down = false; chatReads._reset(); }
+ }
+
+ // (2) The stream-end analytics write keeps the totals it has and re-upserts Chat's right
+ // after (the clip-count pattern) instead of persisting a cold mirror (or zero).
+ {
+ chatReads._reset();
+ d2.run('DELETE FROM stream_analytics WHERE stream_id = ?', [streamId]);
+ const computed = d2.computeAndCacheStreamAnalytics(streamId);
+ assert.strictEqual(computed.total_messages, 0, 'the synchronous compute writes its prior totals, not a cold scan');
+ for (let i = 0; i < 300; i++) { await sleep(10); const a = d2.getStreamAnalytics(streamId); if (a && a.total_messages > 0) break; }
+ const sa = d2.getStreamAnalytics(streamId);
+ assert.ok(sa.total_messages >= 3, "the setImmediate refresh wrote Chat's totals");
+ assert.strictEqual(sa.unique_chatters, 2, "and Chat's chatter count");
+ chatReads._reset();
+ }
+
+ // (3) Live's own hidden row OR Chat's fresh list — a local hide the mirror applied still
+ // hides even when a fresh Chat list lacks it (the #28 rule, restored).
+ {
+ const chatClient = require('../server/chat/chat-client');
+ const realReadRelayUsers = chatClient.readRelayUsers;
+ const relOwner = mkUser('relch');
+ db.createChannel({ user_id: relOwner, title: 'Relay channel' });
+ const freshCh = db.getChannelByUserId(relOwner).id; // no hidden rows yet
+ chatClient.readRelayUsers = async () => ({ relay_users: [] }); // Chat's list holds nothing
+ chatReads._reset();
+ try {
+ d2.run('DELETE FROM hidden_relay_users WHERE channel_id = ?', [freshCh]);
+ chatReads.isRelayUserHidden(freshCh, 'twitch', 'nobody'); // warm Chat's fresh (empty) list
+ for (let i = 0; i < 100; i++) await sleep(10);
+ d2.run("INSERT INTO hidden_relay_users (channel_id, platform, external_username, action, created_by) VALUES (?, 'twitch', 'localonly', 'hide', ?)", [freshCh, streamer]);
+ assert.strictEqual(chatReads.isRelayUserHidden(freshCh, 'twitch', 'localonly'), true, "Live's own hidden row wins over a fresh Chat list that lacks it");
+ } finally {
+ chatClient.readRelayUsers = realReadRelayUsers;
+ d2.run('DELETE FROM hidden_relay_users WHERE channel_id = ?', [freshCh]);
+ chatReads._reset();
+ }
+ }
+
+ // (4) A cold profile count is null (the card omits it), never a mirror number passed off
+ // as Chat's.
+ {
+ chatReads._reset();
+ readState.down = true;
+ try {
+ assert.strictEqual(chatReads.userMessageCountPeek(viewer), null, 'a cold count is null, not a mirror number');
+ assert.strictEqual(d2.getUserProfile(viewer).messageCount, null, 'the profile carries the null through');
+ } finally { readState.down = false; chatReads._reset(); }
+ }
+
+ // (5) channelMaxId — the preview path's awaited read — warms the peek's key, so the
+ // cursor is Chat's id rather than the mirror once it has answered.
+ {
+ const chatClient = require('../server/chat/chat-client');
+ const realReadMessages = chatClient.readMessages;
+ chatClient.readMessages = async (o) => (o && o.tail ? { max_id: 777777, messages: [] } : realReadMessages(o));
+ chatReads._reset();
+ try {
+ await chatReads.channelMaxId(streamer);
+ assert.strictEqual(chatReads.channelMaxIdPeek(streamer), 777777, "the peek answers Chat's id after the awaited read");
+ } finally {
+ chatClient.readMessages = realReadMessages;
+ 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.