openvibers/openvibe.chat Commits
Serve Live's chat stats, message, timeline, queue and sound reads (#21)
Changes
@@@ -96,6 +96,10 @@ before changing behaviour.** Browser JavaScript does not change; nginx routes th`GET /internal/moderation/channels/:channelId/emote-count` — loopback + service token,capability `chat.moderation.read` (`chat.channel-moderation-result@1`,`chat.moderated-channels-result@1`, `chat.emote-count-result@1`). Live caches each answer 30 s.++ - Live's reads of Chat's message, moderation-queue and sound tables (plan T3 J4b):++ `POST /internal/chat/stats`, `GET /internal/chat/messages`, `/timeline`, `/moderation/*`,++ `/sounds` and `POST /internal/chat/sounds/asset` (`server/chat/internal-reads.js`; routes and++ capabilities in `docs/chat-ingress.md`).- Alert sounds: `playAlertSound(chatServer, streamerId, streamId, kind)` (`server/chat/alert-sounds.js`)resolves the channel's own settings row, reads the clip and broadcasts it to the channel room.- **The chat-AI summaries** (plan T3 step 2, decision 5) — Live's `server/ai/chat-ai.js` job moved@@@ -7,8 +7,8 @@"runtime": "node 22, express, ws, PostgreSQL through openvibe-sdk/db (DATABASE_URL via PgBouncer, DATABASE_DIRECT_URL for migrations/; embedded PGlite outside production), Valkey for the per-actor limits (VALKEY_URL); better-sqlite3 gone (plan T3)","port": 4400,"domain": "openvibe.chat (Chat's own site: global chat, messages, rooms, settings; the service manifest records it live since 2026-09-24) and openvibe.live (/ws/chat, /api/chat/, /api/dm/, /api/tts/, /api/sounds via an nginx include in openvibe.live's vhost)",-- "contracts": "openvibe-contracts v0.91.0",-- "capabilities": ["chat.message.send", "chat.event.publish", "chat.moderation.write", "chat.cache.invalidate", "chat.presence.read", "chat.live_bridge.write"],++ "contracts": "openvibe-contracts v0.96.0",++ "capabilities": ["chat.message.send", "chat.event.publish", "chat.moderation.write", "chat.cache.invalidate", "chat.presence.read", "chat.live_bridge.write", "chat.stats.read", "chat.messages.read", "chat.analysis.read", "chat.moderation.queue.read", "chat.sounds.read", "chat.sounds.write"],"capabilitiesUsed": ["live.chat_context.read","live.chat_effects.write",@@@ -14,12 +14,30 @@ a stream that does not exist is 404: do not retry a 4xx.| Route | Capability | Body and result || --- | --- | --- |--| `POST /internal/chat/messages` | `chat.message.send` | Chat message fields (`stream_id`, `channel_user_id`, `user_id`, `anon_id`, `username`, `message`, `message_type`, `is_global`, `reply_to_id`, `source_platform`, `metadata`), optional `mirror`, `tts:{voice,identity_key,key}`, `auto_delete_at`, and typed chat-frame fields in `frame` (`role`: `admin`/`global_mod`/`streamer`/`user`/`anon`/`external`, `profile_color`: hex, `avatar_url`, `is_ai`/`is_bot`/`filtered`: booleans, `core_username`, `display_name`); the broadcast carries `reply_to_id`, `is_global`, `auto_delete_at` and `metadata` from the body. Returns `{ok,id}`. With no TTS key, Chat uses `m<real message id>`. `dm:{to_user_id}` with `user_id` persists and delivers a direct message (the same `dm` frame as `POST /api/dm`, with the sender looked up by id), returning `{ok,id,conversation_id,to_user_id}`; a blank message is 400. Chat records the first chat in the channel (the welcome check) under `first_chat_key` when given (`user:<id>`, `anon:<anonId>` or `ext:<prefixed username>`), otherwise for a `chat` line only: `ext:<username>` with `source_platform`, else `user:<user_id>`, else `anon:<anon_id>`. |++| `POST /internal/chat/messages` | `chat.message.send` | Chat message fields (`stream_id`, `channel_user_id`, `user_id`, `anon_id`, `username`, `message`, `message_type`, `is_global`, `reply_to_id`, `source_platform`, `metadata`), optional `mirror`, `tts:{voice,identity_key,key}`, `auto_delete_at`, and typed chat-frame fields in `frame` (`role`: `admin`/`global_mod`/`streamer`/`user`/`anon`/`external`, `profile_color`: hex, `avatar_url`, `is_ai`/`is_bot`/`filtered`: booleans, `core_username`, `display_name`); the broadcast carries `reply_to_id`, `is_global`, `auto_delete_at` and `metadata` from the body. Returns `{ok,id,stream_id,channel_user_id,first_chat}`: `first_chat` (`chat.send-result` 1.2.0) is true when this line recorded the chatter's first chat in the channel, so Live can welcome them. With no TTS key, Chat uses `m<real message id>`. `dm:{to_user_id}` with `user_id` persists and delivers a direct message (the same `dm` frame as `POST /api/dm`, with the sender looked up by id), returning `{ok,id,conversation_id,to_user_id}`; a blank message is 400. Chat records the first chat in the channel (the welcome check) under `first_chat_key` when given (`user:<id>`, `anon:<anonId>` or `ext:<prefixed username>`), otherwise for a `chat` line only: `ext:<username>` with `source_platform`, else `user:<user_id>`, else `anon:<anon_id>`. || `POST /internal/chat/events` | `chat.event.publish` | `{key,target:{kind,id?},frame:{type,...}}`, returns `{ok:true}`. Target kinds: `stream`, `channel`, `owner-streams`, `user` (positive `id`), `global`, `all` (no `id`). Only the event frame types allowlisted in `internal-ingress.js` are accepted; `dm` and delete frames are refused, and `chat` only as the news card. || `POST /internal/chat/moderation` | `chat.moderation.write` | `{key,action,...}`; actions: `delete-message`, `delete-user-messages`, `delete-anon-messages`, `delete-relay-messages`, `delete-by-range`, `review-pending-ip`, `approve-ip-messages`, `deny-ip-messages`, `relay-hide`, `relay-unhide`, `relay-record`, `tts-voice-override`, `disconnect`, `log`. Deletes return `ids` and broadcast `delete-messages` to every affected surface. || `POST /internal/chat/invalidate` | `chat.cache.invalidate` | `{key,user?,user_data?,approvals?,bans?,channel?}`; a cache hint, returns `{ok:true}`. Include a typed `user_data` object when Live edited the account, so Chat updates sockets and stored message names. || `GET /internal/chat/presence` | `chat.presence.read` | Who is connected where: counts, per-stream viewers, slow modes, users/anons with their addresses (`server/chat/presence.js`). |++Live's reads of Chat's tables, in place of its mirrored copy (`server/chat/internal-reads.js`, plan++T3 J4b N1–N5, contracts in `openvibe-contracts` 0.95.0). Unknown or malformed parameters are 400;++deleted messages are never read; `since`/`until` are epoch ms (`until` exclusive). Every answer is++`{ok:true,...}` or `{ok:false,error}`.++++| Route | Capability | Request and result |++| --- | --- | --- |++| `POST /internal/chat/stats` | `chat.stats.read` | `{kind:'site'\|'user'\|'stream'\|'channel-top', user_id?, stream_id?, channel_user_id?, since?, until?, limit?}`. `site`/`user`(`user_id`)/`stream`(`stream_id`) → `{messages,chatters}` (+ `sounds`, soundboard plays, for `stream`); `channel-top` (by `stream_id`, `channel_user_id` or all chat; `limit` ≤ 50, default 10) → `{top_chatters:[{user_id,username,display_name,avatar_url,profile_color,count}]}`. |++| `GET /internal/chat/messages` | `chat.messages.read` | Exactly one of `channel_user_id`, `stream_id`, `user_id`, `anon_id`, `username`, `id`; `after_id` (oldest first), `before_id` (newest first; the default), `limit` ≤ 500 (default 100), `types=chat,donation,…`, `tail=1` (no rows) → `{messages,max_id}`; `max_id` is the newest matching id. |++| `GET /internal/chat/timeline` | `chat.analysis.read` | `channel_user_id` or `stream_id`, `since`, `until` (default now), `bucket_ms` ≥ 1000 (≤ 10 000 buckets) → `{buckets:[{t,count}],max_id}`; buckets start at `since`, oldest first, empty ones left out. |++| `GET /internal/chat/moderation/pending-ip?channel_id` | `chat.moderation.queue.read` | `{pending_ip}`: the channel's pending rows, oldest first (`limit` ≤ 500, default 50). |++| `GET /internal/chat/moderation/relay-users?channel_id` | `chat.moderation.queue.read` | `{relay_users}`: hidden or banned relay names on the channel or everywhere, newest first, with `created_by_username` (`limit` ≤ 500, default 100). |++| `GET /internal/chat/moderation/relay-users/:id` | `chat.moderation.queue.read` | `{relay_user}`, null when unknown. |++| `GET /internal/chat/moderation/tts-override?identity_key` | `chat.moderation.queue.read` | `{tts_override}` (key trimmed and lowercased), null when none. |++| `GET /internal/chat/sounds?channel_owner_id` (also `/sounds/count`) | `chat.sounds.read` | `{count}` of the channel's sounds. |++| `GET /internal/chat/sounds?pending_asset=1` | `chat.sounds.read` | `{sounds}` not yet on Media (`media_asset_id` null), by id; optional `channel_owner_id`, `after_id`, `limit` ≤ 500 (default 100). |++| `POST /internal/chat/sounds/asset` | `chat.sounds.write` | `{id,media_url,media_asset_id}` records Live's upload (`migrations/0004_channel_sound_media.sql`); repeating it is a 200 no-op; an unknown sound is 404. Until Live writes here, the read mirror leaves Live's own asset columns alone for a sound Chat has no asset for. |++`/internal/chat/events` is a narrow transient event ingress, not a generic ChatServer call.`live.release.deployed` continues through OpenVibe.Events; there is no deploy-notice endpoint.An `alert` targets its `streamerId` channel; a `channel-sound` targets its `streamId` stream.@@@ -41,7 +59,11 @@ Before Live repoints, the separate **OpenVibe.Contracts** repository must regist`live chat.event.publish openvibe.chat`, `live chat.moderation.write openvibe.chat`, and`live chat.cache.invalidate openvibe.chat`. Keep the Live grants for `chat.message.send`(`messages`), `chat.presence.read` (`/internal/chat/presence`) and `chat.live_bridge.write` (the call--hooks at `/internal/calls/stream-channel`): the bridge is gone, but each still guards a route. Chat's++hooks at `/internal/calls/stream-channel`): the bridge is gone, but each still guards a route. For++the read API above, Network's `DEFAULT_GRANTS` must add `live chat.stats.read openvibe.chat`,++`live chat.messages.read openvibe.chat`, `live chat.analysis.read openvibe.chat`,++`live chat.moderation.queue.read openvibe.chat`, `live chat.sounds.read openvibe.chat` and++`live chat.sounds.write openvibe.chat`. Chat'slocal proposals are in `docs/capabilities-proposal/`.Live J2 (done: Live #20/#21) replaced its bridge writer with a typed client: repoint message producers and persisted DMsnnew file mode 100644@@@ -0,0 +1,8 @@++-- phase: expand++-- Where Live's asset sync put each channel sound on OpenVibe.Media (plan T3 J4b, N5): Live reads the++-- sounds still to upload (GET /internal/chat/sounds?pending_asset=1) and writes the asset back++-- (POST /internal/chat/sounds/asset), instead of using its mirrored copy of channel_sounds.++-- Both stay NULL until the sound is uploaded.++ALTER TABLE channel_sounds ADD COLUMN IF NOT EXISTS media_url text COLLATE "C";++ALTER TABLE channel_sounds ADD COLUMN IF NOT EXISTS media_asset_id bigint;++CREATE INDEX IF NOT EXISTS idx_channel_sounds_pending_asset ON channel_sounds(id) WHERE media_asset_id IS NULL;@@@ -16,7 +16,7 @@"express-rate-limit": "^7.5.0","iovalkey": "^0.4.0","multer": "^2.0.2",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.91.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.96.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.5.0","pg": "^8.23.0",@@@ -1294,9 +1294,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.91.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.91.0",-- "integrity": "sha512-Sg5WUFRcnH+5TJGHciFU++dwWqXjhcKngscmICWHV8eGrt8kgTZfH85EeWtlG1LjdIRlWW9h2wK0Wtd5n4OGWw==",++ "version": "0.96.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.96.0",++ "integrity": "sha512-QEXfNYNkK6n49sXokgbP2FTKaIEawRJOS7Xij9g20fBFSqiVv14voZ60LqECPl1VMZedoSGuYLa8doVlnSxH0w==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -29,7 +29,7 @@"express-rate-limit": "^7.5.0","iovalkey": "^0.4.0","multer": "^2.0.2",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.91.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.96.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.5.0","pg": "^8.23.0",@@@ -102,6 +102,8 @@ function createApp({ chatServer, mirror, relay, events = null, callServer = null// ── Internal (loopback; never routed by nginx) ─────────────app.use('/internal/chat', express.json({ limit: '64kb' }), createInternalIngress({ chatServer }));++ // Live's reads of the chat tables Chat owns, in place of its mirrored copy (plan T3 J4b, server/chat/internal-reads.js).++ app.use('/internal/chat', express.json({ limit: '64kb' }), require('./chat/internal-reads'));// OpenVibe.Events deliveries (live.release.deployed, network.module.updated): signature v2 over// the raw body, so this router reads the body itself (server/events/consumer.js).if (events) app.use('/internal/events', events.router);@@@ -44,6 +44,13 @@ const ROW_SQL = {moderation_actions: 'SELECT * FROM moderation_actions WHERE id = ?',};++// Live's asset sync still writes media_url/media_asset_id on its own copy until it moves to++// POST /internal/chat/sounds/asset: a sound Chat has no asset for must not blank Live's.++function soundRow(row) {++ const { media_url, media_asset_id, ...rest } = row;++ return media_asset_id == null ? rest : row;++}++function createMirror({ config, fetchImpl = (...a) => globalThis.fetch(...a), log = console } = {}) {let timer = null;let busy = false;@@@ -63,7 +70,7 @@ function createMirror({ config, fetchImpl = (...a) => globalThis.fetch(...a), lotry { pk = JSON.parse(r.pk); } catch { continue; }if (r.op === 'delete') { changes.push({ table: r.tbl, op: 'delete', pk }); continue; }const row = await db.get(ROW_SQL[r.tbl], cols.map((c) => pk[c]));-- changes.push(row ? { table: r.tbl, op: 'upsert', row: r.tbl === 'moderation_actions' ? db.moderationRow(row) : r.tbl === 'chat_messages' ? db.chatMessageRow(row) : row } : { table: r.tbl, op: 'delete', pk });++ changes.push(row ? { table: r.tbl, op: 'upsert', row: r.tbl === 'moderation_actions' ? db.moderationRow(row) : r.tbl === 'chat_messages' ? db.chatMessageRow(row) : r.tbl === 'channel_sounds' ? soundRow(row) : row } : { table: r.tbl, op: 'delete', pk });}return changes;}@@@ -154,8 +154,9 @@ async function writeMessage(b) {const id = Number(saved.lastInsertRowid);const channelUserId = b.channel_user_id || (b.stream_id ? (await ctx.getStreamById(b.stream_id))?.user_id : null) || null;const chatterKey = firstChatKey(b);-- if (channelUserId && chatterKey) await db.recordFirstChat(chatterKey, channelUserId);-- return { id, stream_id: b.stream_id || null, channel_user_id: channelUserId };++ // first_chat (chat.send-result 1.2.0): this line used up the chatter's welcome in the channel.++ const first_chat = !!(channelUserId && chatterKey) && await db.recordFirstChat(chatterKey, channelUserId);++ return { id, stream_id: b.stream_id || null, channel_user_id: channelUserId, first_chat };}async function deliverMessage(chatServer, b, result) {nnew file mode 100644@@@ -0,0 +1,297 @@++/**++ * OpenVibe.Chat — the internal read API Live calls instead of reading its mirrored copy of Chat's++ * tables (plan T3 J4b, design §1d N1–N5). Loopback only, a service token for audience++ * openvibe.chat, one capability per route (openvibe-contracts 0.95.0):++ *++ * POST /internal/chat/stats chat.stats.read chat.stats-request@1 → chat.stats-result@1++ * GET /internal/chat/messages chat.messages.read → chat.messages-page@1++ * GET /internal/chat/timeline chat.analysis.read → chat.timeline-result@1++ * GET /internal/chat/moderation/pending-ip chat.moderation.queue.read → chat.moderation-queue-result@1 { pending_ip }++ * GET /internal/chat/moderation/relay-users chat.moderation.queue.read → { relay_users }++ * GET /internal/chat/moderation/relay-users/:id chat.moderation.queue.read → { relay_user }++ * GET /internal/chat/moderation/tts-override chat.moderation.queue.read → { tts_override }++ * GET /internal/chat/sounds(/count) chat.sounds.read → chat.sounds-result@1 { count } | { sounds }++ * POST /internal/chat/sounds/asset chat.sounds.write chat.sound-asset-request@1 → chat.ingress-ack@1++ *++ * Every parameter is validated (unknown ones are 400) and bound; deleted messages are never read.++ * Times in requests are epoch ms; chat_messages.timestamp is UTC 'YYYY-MM-DD HH:MM:SS' text, so a++ * window compares text and keeps the timestamp indexes.++ */++'use strict';++++const express = require('express');++const db = require('../db/database');++const serviceAuth = require('../net/service-auth');++++const MESSAGE_TYPES = ['chat', 'system', 'donation', 'command', 'tts', 'channel-sound', 'soundboard', 'clip'];++const MAX_MESSAGES = 500;++const MAX_ROWS = 500;++const MAX_BUCKETS = 10_000;++++const bad = (message) => { const e = new Error(message); e.status = 400; throw e; };++const id = (v) => Number.isSafeInteger(v) && v > 0;++const ms = (v) => Number.isSafeInteger(v) && v >= 0;++const sqlTime = (t) => new Date(t).toISOString().slice(0, 19).replace('T', ' ');++const num = (v) => (v == null ? null : Number(v));++const text = (v) => (v == null ? null : String(v));++++/** Query string → validated values; anything not in `spec` is 400. */++function query(req, spec) {++ const out = {};++ for (const [k, raw] of Object.entries(req.query)) {++ const kind = spec[k];++ if (!kind || typeof raw !== 'string') bad(`Invalid parameter ${k}`);++ if (kind === 'id' || kind === 'ms') {++ const v = /^\d{1,16}$/.test(raw) ? Number(raw) : NaN;++ if (!(kind === 'id' ? id(v) : ms(v))) bad(`Invalid ${k}`);++ out[k] = v;++ } else if (kind === 'flag') {++ if (raw !== '1' && raw !== '0') bad(`Invalid ${k}`);++ out[k] = raw === '1';++ } else if (kind === 'types') {++ const types = raw.split(',');++ if (!types.every((t) => MESSAGE_TYPES.includes(t))) bad(`Invalid ${k}`);++ out[k] = types;++ } else {++ if (!raw || raw.length > kind) bad(`Invalid ${k}`);++ out[k] = raw;++ }++ }++ return out;++}++++function limit(v, dflt, max) { return v == null ? dflt : Math.min(v, max); }++++function route(name, fn) {++ return async (req, res) => {++ try {++ res.json({ ok: true, ...await fn(req) });++ } catch (err) {++ if (err.status) return res.status(err.status).json({ ok: false, error: err.message });++ console.error(`[ChatReads] ${name}:`, err);++ res.status(503).json({ ok: false, error: 'Chat read unavailable' });++ }++ };++}++++// ── N1: stats ──++const CHATTER = `COALESCE('u:' || user_id, 'a:' || anon_id, source_platform || ':' || username)`;++const STATS_FIELDS = ['kind', 'user_id', 'stream_id', 'channel_user_id', 'since', 'until', 'limit'];++++function statsFilter(b) {++ const where = ['is_deleted = 0'];++ const params = [];++ for (const k of ['user_id', 'stream_id', 'channel_user_id']) if (b[k] != null) { where.push(`${k} = ?`); params.push(b[k]); }++ if (b.since != null) { where.push('timestamp >= ?'); params.push(sqlTime(b.since)); }++ if (b.until != null) { where.push('timestamp < ?'); params.push(sqlTime(b.until)); }++ return { where: where.join(' AND '), params };++}++++async function stats(req) {++ const b = req.body;++ if (!b || typeof b !== 'object' || Array.isArray(b) || Object.keys(b).some((k) => !STATS_FIELDS.includes(k))) bad('Invalid fields');++ if (!['site', 'user', 'stream', 'channel-top'].includes(b.kind)) bad('Invalid kind');++ for (const k of ['user_id', 'stream_id', 'channel_user_id']) if (b[k] != null && !id(b[k])) bad(`Invalid ${k}`);++ for (const k of ['since', 'until']) if (b[k] != null && !ms(b[k])) bad(`Invalid ${k}`);++ if (b.limit != null && !(Number.isSafeInteger(b.limit) && b.limit >= 1 && b.limit <= 50)) bad('Invalid limit');++ if (b.kind === 'user' && b.user_id == null) bad('kind user needs user_id');++ if (b.kind === 'stream' && b.stream_id == null) bad('kind stream needs stream_id');++ if (b.kind === 'channel-top' && b.user_id != null) bad('kind channel-top takes stream_id or channel_user_id');++ const { where, params } = statsFilter(b);++ if (b.kind !== 'channel-top') {++ const r = await db.get(++ `SELECT COUNT(*) AS messages, COUNT(DISTINCT ${CHATTER}) AS chatters,++ COUNT(*) FILTER (WHERE message_type = 'soundboard') AS sounds++ FROM chat_messages WHERE ${where}`, params);++ const out = { messages: Number(r.messages), chatters: Number(r.chatters) };++ if (b.kind === 'stream') out.sounds = Number(r.sounds);++ return out;++ }++ // Registered chatters group by user id; anonymous and relayed ones by name. System lines are no one's.++ const rows = await db.all(++ `SELECT t.user_id, COALESCE(u.username, t.username) AS username, u.display_name, u.avatar_url, u.profile_color, t.count++ FROM (SELECT user_id, MAX(username) AS username, COUNT(*) AS count, MAX(id) AS last_id++ FROM chat_messages++ WHERE ${where} AND message_type <> 'system' AND ${CHATTER} IS NOT NULL++ GROUP BY user_id, CASE WHEN user_id IS NULL THEN ${CHATTER} END) t++ LEFT JOIN ctx_users u ON u.id = t.user_id++ ORDER BY t.count DESC, t.last_id DESC LIMIT ?`, [...params, b.limit || 10]);++ return {++ top_chatters: rows.map((r) => ({++ user_id: num(r.user_id), username: String(r.username || ''), display_name: text(r.display_name),++ avatar_url: text(r.avatar_url), profile_color: text(r.profile_color), count: Number(r.count),++ })),++ };++}++++// ── N2: messages ──++const MESSAGE_FILTERS = { channel_user_id: 'id', stream_id: 'id', user_id: 'id', anon_id: 80, username: 120, id: 'id' };++const MESSAGE_COLUMNS = 'id, user_id, anon_id, username, message, message_type, is_global, stream_id, channel_user_id, source_platform, reply_to_id, timestamp';++++function messageRow(r) {++ return {++ id: Number(r.id), user_id: num(r.user_id), anon_id: text(r.anon_id), username: text(r.username),++ message: String(r.message), message_type: r.message_type || 'chat', is_global: Number(r.is_global) ? 1 : 0,++ stream_id: num(r.stream_id), channel_user_id: num(r.channel_user_id), source_platform: text(r.source_platform),++ reply_to_id: num(r.reply_to_id), timestamp: String(r.timestamp || ''),++ };++}++++async function messages(req) {++ const q = query(req, { ...MESSAGE_FILTERS, after_id: 'id', before_id: 'id', limit: 'id', tail: 'flag', types: 'types' });++ const filters = Object.keys(MESSAGE_FILTERS).filter((k) => q[k] != null);++ if (filters.length !== 1) bad('Give exactly one of channel_user_id, stream_id, user_id, anon_id, username or id');++ const where = [`${filters[0]} = ?`, 'is_deleted = 0'];++ const params = [q[filters[0]]];++ if (q.types) { where.push(`message_type IN (${q.types.map(() => '?').join(', ')})`); params.push(...q.types); }++ const max = await db.get(`SELECT MAX(id) AS max_id FROM chat_messages WHERE ${where.join(' AND ')}`, params);++ const max_id = num(max && max.max_id);++ if (q.tail) return { messages: [], max_id };++ const page = [...where];++ const pageParams = [...params];++ if (q.after_id != null) { page.push('id > ?'); pageParams.push(q.after_id); }++ if (q.before_id != null) { page.push('id < ?'); pageParams.push(q.before_id); }++ // after_id pages forward (oldest first); otherwise newest first, back from before_id.++ const order = q.after_id != null ? 'ASC' : 'DESC';++ const rows = await db.all(++ `SELECT ${MESSAGE_COLUMNS} FROM chat_messages WHERE ${page.join(' AND ')} ORDER BY id ${order} LIMIT ?`,++ [...pageParams, limit(q.limit, 100, MAX_MESSAGES)]);++ return { messages: rows.map(messageRow), max_id };++}++++// ── N3: timeline ──++async function timeline(req) {++ const q = query(req, { channel_user_id: 'id', stream_id: 'id', since: 'ms', until: 'ms', bucket_ms: 'id' });++ if ((q.channel_user_id == null) === (q.stream_id == null)) bad('Give channel_user_id or stream_id');++ if (q.since == null || q.bucket_ms == null) bad('since and bucket_ms are required');++ const until = q.until == null ? Date.now() : q.until;++ if (until <= q.since) bad('until must be after since');++ if (q.bucket_ms < 1000 || (until - q.since) / q.bucket_ms > MAX_BUCKETS) bad('Invalid bucket_ms');++ const col = q.channel_user_id != null ? 'channel_user_id' : 'stream_id';++ // Buckets start at since: bucket n covers [since + n·bucket_ms, since + (n+1)·bucket_ms).++ const rows = await db.all(++ `SELECT n, COUNT(*) AS count, MAX(id) AS max_id FROM (++ SELECT id, floor((extract(epoch FROM ov_ts(timestamp)) * 1000 - ?) / ?)::bigint AS n++ FROM chat_messages++ WHERE ${col} = ? AND is_deleted = 0 AND timestamp >= ? AND timestamp < ?) m++ WHERE n IS NOT NULL AND n >= 0 GROUP BY n ORDER BY n`,++ [q.since, q.bucket_ms, q[col], sqlTime(q.since), sqlTime(until)]);++ return {++ buckets: rows.map((r) => ({ t: q.since + Number(r.n) * q.bucket_ms, count: Number(r.count) })),++ max_id: rows.length ? Math.max(...rows.map((r) => Number(r.max_id))) : null,++ };++}++++// ── N4: moderation queues ──++function pendingIpRow(r) {++ return {++ id: Number(r.id), channel_id: Number(r.channel_id), stream_id: num(r.stream_id), ip_address: String(r.ip_address),++ user_id: num(r.user_id), anon_id: text(r.anon_id), username: text(r.username), message: String(r.message),++ status: r.status || 'pending', reviewed_by: num(r.reviewed_by), created_at: text(r.created_at),++ };++}++function relayUserRow(r) {++ return {++ id: Number(r.id), channel_id: num(r.channel_id), platform: String(r.platform), external_username: String(r.external_username),++ action: r.action || 'hide', reason: text(r.reason), created_by: num(r.created_by),++ created_by_username: text(r.created_by_username), created_at: text(r.created_at),++ };++}++const RELAY_SELECT = `SELECT h.*, u.username AS created_by_username++ FROM hidden_relay_users h LEFT JOIN ctx_users u ON u.id = h.created_by`;++++async function pendingIp(req) {++ const q = query(req, { channel_id: 'id', limit: 'id' });++ if (q.channel_id == null) bad('channel_id is required');++ const rows = await db.all(++ `SELECT * FROM pending_ip_messages WHERE channel_id = ? AND status = 'pending' ORDER BY created_at ASC, id ASC LIMIT ?`,++ [q.channel_id, limit(q.limit, 50, MAX_ROWS)]);++ return { pending_ip: rows.map(pendingIpRow) };++}++++async function relayUsers(req) {++ const q = query(req, { channel_id: 'id', limit: 'id' });++ if (q.channel_id == null) bad('channel_id is required');++ const rows = await db.all(++ `${RELAY_SELECT} WHERE h.channel_id = ? OR h.channel_id IS NULL ORDER BY h.created_at DESC, h.id DESC LIMIT ?`,++ [q.channel_id, limit(q.limit, 100, MAX_ROWS)]);++ return { relay_users: rows.map(relayUserRow) };++}++++async function relayUser(req) {++ query(req, {});++ const relayId = /^\d{1,16}$/.test(req.params.id) ? Number(req.params.id) : NaN;++ if (!id(relayId)) bad('Invalid id');++ const row = await db.get(`${RELAY_SELECT} WHERE h.id = ?`, [relayId]);++ return { relay_user: row ? relayUserRow(row) : null };++}++++async function ttsOverride(req) {++ const q = query(req, { identity_key: 120 });++ const key = String(q.identity_key || '').trim().toLowerCase();++ if (!key) bad('identity_key is required');++ const r = await db.get('SELECT * FROM tts_voice_overrides WHERE identity_key = ?', [key]);++ return {++ tts_override: r ? {++ identity_key: String(r.identity_key), voice: text(r.voice), pitch: num(r.pitch), speed: num(r.speed),++ gap: num(r.gap) || 0, set_by: num(r.set_by), updated_at: text(r.updated_at),++ } : null,++ };++}++++// ── N5: channel sounds ──++function soundRow(r) {++ return {++ id: Number(r.id), channel_owner_id: Number(r.channel_owner_id), command: String(r.command), url: String(r.url),++ mime: text(r.mime), duration_seconds: num(r.duration_seconds), created_by: num(r.created_by),++ created_by_name: text(r.created_by_name), media_url: text(r.media_url), media_asset_id: num(r.media_asset_id),++ };++}++++async function countSounds(ownerId) {++ if (ownerId == null) bad('channel_owner_id is required');++ const r = await db.get('SELECT COUNT(*) AS n FROM channel_sounds WHERE channel_owner_id = ?', [ownerId]);++ return { count: Number(r.n) };++}++const soundCount = (req) => countSounds(query(req, { channel_owner_id: 'id' }).channel_owner_id);++++/** ?channel_owner_id → { count }; ?pending_asset=1 → { sounds } not yet uploaded to Media (oldest first). */++async function sounds(req) {++ const q = query(req, { channel_owner_id: 'id', pending_asset: 'flag', after_id: 'id', limit: 'id' });++ if (!q.pending_asset) {++ if (q.after_id != null || q.limit != null) bad('after_id and limit page pending_asset=1');++ return countSounds(q.channel_owner_id);++ }++ const where = ['media_asset_id IS NULL'];++ const params = [];++ if (q.channel_owner_id != null) { where.push('channel_owner_id = ?'); params.push(q.channel_owner_id); }++ if (q.after_id != null) { where.push('id > ?'); params.push(q.after_id); }++ const rows = await db.all(`SELECT * FROM channel_sounds WHERE ${where.join(' AND ')} ORDER BY id LIMIT ?`,++ [...params, limit(q.limit, 100, MAX_ROWS)]);++ return { sounds: rows.map(soundRow) };++}++++/** Live's asset sync uploaded the file to Media: record where. Repeating the same write is a no-op 200. */++async function soundAsset(req) {++ const b = req.body;++ if (!b || typeof b !== 'object' || Array.isArray(b) || Object.keys(b).some((k) => !['id', 'media_url', 'media_asset_id'].includes(k))) bad('Invalid fields');++ if (!id(b.id) || !id(b.media_asset_id) || typeof b.media_url !== 'string' || !b.media_url || b.media_url.length > 2048) bad('Invalid sound asset');++ const r = await db.run('UPDATE channel_sounds SET media_url = ?, media_asset_id = ? WHERE id = ?', [b.media_url, b.media_asset_id, b.id]);++ if (!r.changes) { const e = new Error('Sound not found'); e.status = 404; throw e; }++ return {};++}++++const router = express.Router();++router.post('/stats', serviceAuth.guard('chat.stats.read'), route('stats', stats));++router.get('/messages', serviceAuth.guard('chat.messages.read'), route('messages', messages));++router.get('/timeline', serviceAuth.guard('chat.analysis.read'), route('timeline', timeline));++const queue = serviceAuth.guard('chat.moderation.queue.read');++router.get('/moderation/pending-ip', queue, route('pending-ip', pendingIp));++router.get('/moderation/relay-users', queue, route('relay-users', relayUsers));++router.get('/moderation/relay-users/:id', queue, route('relay-user', relayUser));++router.get('/moderation/tts-override', queue, route('tts-override', ttsOverride));++router.get('/sounds', serviceAuth.guard('chat.sounds.read'), route('sounds', sounds));++router.get('/sounds/count', serviceAuth.guard('chat.sounds.read'), route('sound-count', soundCount));++router.post('/sounds/asset', serviceAuth.guard('chat.sounds.write'), route('sound-asset', soundAsset));++++module.exports = router;@@@ -637,12 +637,14 @@ async function isFirstChatInChannel(chatterKey, channelUserId) {/*** Record that a chatter has chatted in a streamer's channel.++ * @returns {boolean} true when this was their first time (the row is new)*/async function recordFirstChat(chatterKey, channelUserId) {-- await run(++ const r = await run('INSERT INTO stream_first_chats (chatter_key, channel_user_id) VALUES (?, ?) ON CONFLICT DO NOTHING',[chatterKey, channelUserId]);++ return r.changes > 0;}// ── Moderation log ───────────────────────────────────────────nnew file mode 100644@@@ -0,0 +1,268 @@++'use strict';++/**++ * Live's internal read API over Chat's tables (server/chat/internal-reads.js, plan T3 J4b N1–N5):++ * one capability per route, each answer valid against its openvibe-contracts schema, and++ * first_chat on the message ingress's answer.++ */++++const assert = require('assert');++const { validate, catalog } = require('openvibe-contracts');++const { boot, suite } = require('./helpers');++++const t = suite('chat-internal-reads');++let h, streamer, viewer, other, host, channelId, streamId, otherStreamId, hostStreamId;++const T = Date.UTC(2026, 0, 1, 12, 0, 0);++const at = (ms) => new Date(T + ms).toISOString().slice(0, 19).replace('T', ' ');++const ids = {};++++// The read contracts arrive with openvibe-contracts 0.95.0. Until the pin reaches it, an older++// package cannot validate them: the run says so instead of failing (the shapes are asserted anyway).++const unknownContracts = new Set();++const sendResultHasFirstChat = (() => {++ const [major, minor] = catalog.find((c) => c.id === 'chat.send-result').version.split('.').map(Number);++ return major > 1 || minor >= 2;++})();++function check(ref, body) {++ let r;++ try { r = validate(ref, body); } catch (err) {++ if (!/unknown contract|is v\d/.test(err.message)) throw err;++ unknownContracts.add(ref);++ return body;++ }++ assert.ok(r.valid, `${ref} invalid: ${JSON.stringify(r.errors)} (${JSON.stringify(body)})`);++ return body;++}++const token = (cap) => h.serviceToken([cap]);++async function read(path, cap, ref, status = 200) {++ const r = await h.http('GET', `/internal/chat${path}`, { token: token(cap) });++ assert.strictEqual(r.status, status, `${path}: ${r.text}`);++ return status === 200 ? check(ref, r.body) : r.body;++}++async function stats(body, status = 200) {++ const r = await h.http('POST', '/internal/chat/stats', { body, token: token('chat.stats.read') });++ assert.strictEqual(r.status, status, `${JSON.stringify(body)}: ${r.text}`);++ return check('chat.stats-result@1', r.body);++}++async function line(name, { ms, user = null, anon = null, username, type = 'chat', platform = null, deleted = 0, stream = streamId, channel = streamer.id }) {++ const r = await h.db.run(++ `INSERT INTO chat_messages (stream_id, channel_user_id, user_id, anon_id, username, message, message_type, source_platform, is_deleted, timestamp)++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,++ [stream, channel, user, anon, username, `line ${name}`, type, platform, deleted, at(ms)]);++ ids[name] = Number(r.lastInsertRowid);++}++++t('boot', async () => {++ h = await boot();++ streamer = h.addUser('reads-streamer', { role: 'streamer' });++ viewer = h.addUser('reads-viewer', { display_name: 'Reads Viewer' });++ other = h.addUser('reads-other');++ host = h.addUser('reads-host', { role: 'streamer' });++ channelId = h.addChannel(streamer.id);++ streamId = h.addStream(streamer.id, channelId);++ otherStreamId = h.addStream(other.id, h.addChannel(other.id));++ hostStreamId = h.addStream(host.id, h.addChannel(host.id));++ await h.ctx.sync();++ await h.ctx.ensureUsers([streamer.id, viewer.id, other.id, host.id]);++++ await line('v1', { ms: 0, user: viewer.id, username: viewer.username });++ await line('gone', { ms: 10e3, user: viewer.id, username: viewer.username, deleted: 1 });++ await line('v2', { ms: 30e3, user: viewer.id, username: viewer.username });++ await line('o1', { ms: 65e3, user: other.id, username: other.username });++ await line('v3', { ms: 70e3, user: viewer.id, username: viewer.username });++ await line('a1', { ms: 130e3, anon: 'anon77', username: 'anon77' });++ await line('r1', { ms: 135e3, username: 'twitch:fan', platform: 'twitch' });++ await line('sb', { ms: 140e3, user: viewer.id, username: viewer.username, type: 'soundboard' });++ await line('sys', { ms: 150e3, username: 'System', type: 'system' });++ await line('elsewhere', { ms: 5e3, user: other.id, username: other.username, stream: otherStreamId, channel: other.id });++});++++t('each route needs its own capability, a service token for Chat and loopback', async () => {++ const routes = [++ ['POST', '/stats', 'chat.stats.read', { kind: 'site' }],++ ['GET', `/messages?stream_id=${streamId}`, 'chat.messages.read'],++ ['GET', `/timeline?stream_id=${streamId}&since=${T}&until=${T + 600e3}&bucket_ms=60000`, 'chat.analysis.read'],++ ['GET', `/moderation/pending-ip?channel_id=${channelId}`, 'chat.moderation.queue.read'],++ ['GET', `/moderation/relay-users?channel_id=${channelId}`, 'chat.moderation.queue.read'],++ ['GET', '/moderation/relay-users/1', 'chat.moderation.queue.read'],++ ['GET', '/moderation/tts-override?identity_key=x', 'chat.moderation.queue.read'],++ ['GET', `/sounds?channel_owner_id=${streamer.id}`, 'chat.sounds.read'],++ ['GET', `/sounds/count?channel_owner_id=${streamer.id}`, 'chat.sounds.read'],++ ['GET', '/sounds?pending_asset=1', 'chat.sounds.read'],++ ['POST', '/sounds/asset', 'chat.sounds.write', { id: 999999, media_url: 'https://media.test/x', media_asset_id: 1 }, 404],++ ];++ for (const [method, path, cap, body, ok = 200] of routes) {++ const url = `/internal/chat${path}`;++ const wrong = cap === 'chat.messages.read' ? 'chat.stats.read' : 'chat.messages.read';++ assert.strictEqual((await h.http(method, url, { body })).status, 401, `${path} without a token`);++ assert.strictEqual((await h.http(method, url, { body, token: token(wrong) })).status, 403, `${path} with ${wrong}`);++ assert.strictEqual((await h.http(method, url, { body, token: token('chat.moderation.read') })).status, 403, `${path} with chat.moderation.read`);++ assert.strictEqual((await h.http(method, url, { body, token: h.serviceToken([cap], { aud: 'openvibe.live' }) })).status, 401, `${path} for another audience`);++ assert.strictEqual((await h.http(method, url, { body, token: token(cap), headers: { 'X-Forwarded-For': '1.2.3.4' } })).status, 403, `${path} from outside`);++ assert.strictEqual((await h.http(method, url, { body, token: token(cap) })).status, ok, `${path} with ${cap}`);++ }++ // Reading sounds does not allow writing their asset.++ assert.strictEqual((await h.http('POST', '/internal/chat/sounds/asset', { body: {}, token: token('chat.sounds.read') })).status, 403);++});++++t('N1 stats: site, user, stream and channel-top over non-deleted lines in the window', async () => {++ const win = { since: T, until: T + 600e3 };++ assert.deepStrictEqual(await stats({ kind: 'site', ...win }), { ok: true, messages: 9, chatters: 4 });++ assert.ok((await stats({ kind: 'site' })).messages >= 9);++ assert.deepStrictEqual(await stats({ kind: 'user', user_id: viewer.id, ...win }), { ok: true, messages: 4, chatters: 1 });++ assert.deepStrictEqual(await stats({ kind: 'stream', stream_id: streamId, ...win }), { ok: true, messages: 8, chatters: 4, sounds: 1 });++ assert.deepStrictEqual(await stats({ kind: 'stream', stream_id: streamId, since: T + 60e3, until: T + 120e3 }), { ok: true, messages: 2, chatters: 2, sounds: 0 });++++ const top = (await stats({ kind: 'channel-top', stream_id: streamId, ...win })).top_chatters;++ assert.deepStrictEqual(top[0], { user_id: viewer.id, username: viewer.username, display_name: 'Reads Viewer', avatar_url: null, profile_color: '#8b5cf6', count: 4 });++ assert.deepStrictEqual(top.slice(1).map((c) => [c.user_id, c.username, c.count]).sort(),++ [[null, 'anon77', 1], [null, 'twitch:fan', 1], [other.id, other.username, 1]].sort());++ assert.deepStrictEqual((await stats({ kind: 'channel-top', channel_user_id: streamer.id, limit: 2, ...win })).top_chatters.map((c) => c.count), [4, 1]);++ const all = (await stats({ kind: 'channel-top', ...win })).top_chatters;++ assert.deepStrictEqual(all.slice(0, 2).map((c) => [c.user_id, c.count]), [[viewer.id, 4], [other.id, 2]]);++++ for (const body of [{}, { kind: 'everything' }, { kind: 'user' }, { kind: 'stream' }, { kind: 'site', extra: 1 },++ { kind: 'site', since: -1 }, { kind: 'site', since: '1' }, { kind: 'stream', stream_id: 1.5 }, { kind: 'channel-top', limit: 51 },++ { kind: 'channel-top', user_id: viewer.id }]) await stats(body, 400);++});++++t('N2 messages: one filter, a cursor on id, tail → max_id', async () => {++ const live = ['v1', 'v2', 'o1', 'v3', 'a1', 'r1', 'sb', 'sys'].map((k) => ids[k]);++ const page = (q, status) => read(`/messages?${q}`, 'chat.messages.read', 'chat.messages-page@1', status);++ const newest = await page(`stream_id=${streamId}`);++ assert.deepStrictEqual(newest.messages.map((m) => m.id), [...live].reverse());++ assert.strictEqual(newest.max_id, ids.sys);++ assert.deepStrictEqual(newest.messages.find((m) => m.id === ids.r1), {++ id: ids.r1, user_id: null, anon_id: null, username: 'twitch:fan', message: 'line r1', message_type: 'chat', is_global: 0,++ stream_id: streamId, channel_user_id: streamer.id, source_platform: 'twitch', reply_to_id: null, timestamp: at(135e3),++ });++ assert.deepStrictEqual((await page(`stream_id=${streamId}&after_id=${ids.v1}&limit=2`)).messages.map((m) => m.id), [ids.v2, ids.o1]);++ assert.deepStrictEqual((await page(`channel_user_id=${streamer.id}&before_id=${ids.o1}`)).messages.map((m) => m.id), [ids.v2, ids.v1]);++ assert.deepStrictEqual((await page(`user_id=${viewer.id}&after_id=${ids.v1}&before_id=${ids.sb}`)).messages.map((m) => m.id), [ids.v2, ids.v3]);++ assert.deepStrictEqual(await page(`channel_user_id=${streamer.id}&tail=1`), { ok: true, messages: [], max_id: ids.sys });++ assert.deepStrictEqual((await page(`stream_id=${streamId}&types=chat`)).messages.map((m) => m.id), live.slice(0, 6).reverse());++ assert.deepStrictEqual((await page('anon_id=anon77')).messages.map((m) => m.id), [ids.a1]);++ assert.deepStrictEqual((await page('username=twitch%3Afan')).messages.map((m) => m.id), [ids.r1]);++ assert.deepStrictEqual((await page(`id=${ids.o1}`)).messages.map((m) => m.message), ['line o1']);++ assert.deepStrictEqual(await page(`id=${ids.gone}`), { ok: true, messages: [], max_id: null });++ assert.strictEqual((await page(`stream_id=${streamId}&limit=100000`)).messages.length, live.length);++ for (const q of ['', `stream_id=${streamId}&user_id=${viewer.id}`, 'stream_id=abc', 'stream_id=0', `stream_id=${streamId}&limit=-1`,++ `stream_id=${streamId}&tail=yes`, `stream_id=${streamId}&types=chat,nope`, `stream_id=${streamId}&order=asc`, `stream_id=${streamId}&stream_id=1`])++ check('chat.messages-page@1', await page(q, 400));++});++++t('N3 timeline: buckets from since, oldest first, empty ones left out', async () => {++ const tl = (q, status) => read(`/timeline?${q}`, 'chat.analysis.read', 'chat.timeline-result@1', status);++ assert.deepStrictEqual(await tl(`stream_id=${streamId}&since=${T}&until=${T + 600e3}&bucket_ms=60000`), {++ ok: true, buckets: [{ t: T, count: 2 }, { t: T + 60e3, count: 2 }, { t: T + 120e3, count: 4 }], max_id: ids.sys,++ });++ assert.deepStrictEqual(await tl(`channel_user_id=${streamer.id}&since=${T + 30e3}&until=${T + 90e3}&bucket_ms=30000`), {++ ok: true, buckets: [{ t: T + 30e3, count: 1 }, { t: T + 60e3, count: 2 }], max_id: ids.v3,++ });++ assert.deepStrictEqual(await tl(`stream_id=${streamId}&since=${T + 900e3}&until=${T + 960e3}&bucket_ms=60000`), { ok: true, buckets: [], max_id: null });++ for (const q of [`since=${T}&bucket_ms=60000`, `stream_id=${streamId}&channel_user_id=${streamer.id}&since=${T}&bucket_ms=60000`,++ `stream_id=${streamId}&bucket_ms=60000`, `stream_id=${streamId}&since=${T}`, `stream_id=${streamId}&since=${T}&until=${T + 600e3}&bucket_ms=500`,++ `stream_id=${streamId}&since=${T}&until=${T}&bucket_ms=60000`, `stream_id=${streamId}&since=0&until=${T}&bucket_ms=1000`])++ check('chat.timeline-result@1', await tl(q, 400));++});++++t('N4 moderation queues: pending IP messages, hidden relay users, TTS overrides', async () => {++ const run = (sql, p) => h.db.run(sql, p);++ const pend = async (channel, status, ms) => Number((await run(++ `INSERT INTO pending_ip_messages (channel_id, stream_id, ip_address, anon_id, username, message, status, created_at) VALUES (?, ?, '203.0.113.9', 'anon5', 'anon5', 'held', ?, ?)`,++ [channel, streamId, status, at(ms)])).lastInsertRowid);++ const later = await pend(channelId, 'pending', 20e3);++ const first = await pend(channelId, 'pending', 10e3);++ await pend(channelId, 'approved', 5e3);++ await pend(channelId + 100, 'pending', 1e3);++ const q = (path, status) => read(`/moderation/${path}`, 'chat.moderation.queue.read', 'chat.moderation-queue-result@1', status);++ const pending = (await q(`pending-ip?channel_id=${channelId}`)).pending_ip;++ assert.deepStrictEqual(pending.map((r) => r.id), [first, later]);++ assert.strictEqual(pending[0].ip_address, '203.0.113.9');++ assert.strictEqual((await q(`pending-ip?channel_id=${channelId}&limit=1`)).pending_ip.length, 1);++++ const hide = async (channel, ms) => Number((await run(++ `INSERT INTO hidden_relay_users (channel_id, platform, external_username, action, reason, created_by, created_at) VALUES (?, 'twitch', 'spammer', 'hide', 'spam', ?, ?)`,++ [channel, streamer.id, at(ms)])).lastInsertRowid);++ const older = await hide(channelId, 1e3);++ const everywhere = await hide(null, 2e3);++ await hide(channelId + 100, 3e3);++ const relay = (await q(`relay-users?channel_id=${channelId}`)).relay_users;++ assert.deepStrictEqual(relay.map((r) => [r.id, r.channel_id]), [[everywhere, null], [older, channelId]]);++ assert.strictEqual(relay[0].created_by_username, streamer.username);++ assert.deepStrictEqual((await q(`relay-users/${older}`)).relay_user, { ...relay[1] });++ assert.deepStrictEqual(await q('relay-users/999999'), { ok: true, relay_user: null });++++ await run(`INSERT INTO tts_voice_overrides (identity_key, voice, pitch, speed, gap, set_by) VALUES ('user:42', 'brian', 2, -1, 3, ?)`, [streamer.id]);++ const override = (await q('tts-override?identity_key=%20USER:42%20')).tts_override;++ assert.deepStrictEqual({ ...override, updated_at: null }, { identity_key: 'user:42', voice: 'brian', pitch: 2, speed: -1, gap: 3, set_by: streamer.id, updated_at: null });++ assert.deepStrictEqual(await q('tts-override?identity_key=user:43'), { ok: true, tts_override: null });++++ for (const path of ['pending-ip', 'pending-ip?channel_id=x', 'relay-users', 'relay-users/abc', 'relay-users/0', 'tts-override', 'tts-override?identity_key=%20', `pending-ip?channel_id=${channelId}&status=denied`])++ check('chat.moderation-queue-result@1', await q(path, 400));++});++++t('N5 sounds: count, pending assets, and an idempotent asset write-back', async () => {++ const sound = async (owner, command, asset = null) => Number((await h.db.run(++ `INSERT INTO channel_sounds (channel_owner_id, command, url, duration_seconds, created_by, created_by_name, media_url, media_asset_id) VALUES (?, ?, ?, 1.5, ?, 'uploader', ?, ?)`,++ [owner, command, `/sounds/${command}.mp3`, owner, asset && `https://media.test/a/${asset}`, asset])).lastInsertRowid);++ const honk = await sound(streamer.id, 'honk');++ await sound(streamer.id, 'beep', 7);++ const boom = await sound(other.id, 'boom');++ const s = (path, status) => read(`/sounds${path}`, 'chat.sounds.read', 'chat.sounds-result@1', status);++ assert.deepStrictEqual(await s(`?channel_owner_id=${streamer.id}`), { ok: true, count: 2 });++ assert.deepStrictEqual(await s(`/count?channel_owner_id=${other.id}`), { ok: true, count: 1 });++ assert.deepStrictEqual(await s(`?channel_owner_id=${host.id}`), { ok: true, count: 0 });++ const pending = (await s('?pending_asset=1')).sounds;++ assert.deepStrictEqual(pending.map((r) => r.id), [honk, boom]);++ assert.deepStrictEqual(pending[0], {++ id: honk, channel_owner_id: streamer.id, command: 'honk', url: '/sounds/honk.mp3', mime: 'audio/mpeg', duration_seconds: 1.5,++ created_by: streamer.id, created_by_name: 'uploader', media_url: null, media_asset_id: null,++ });++ assert.deepStrictEqual((await s(`?pending_asset=1&channel_owner_id=${other.id}`)).sounds.map((r) => r.id), [boom]);++ assert.deepStrictEqual((await s(`?pending_asset=1&after_id=${honk}&limit=1`)).sounds.map((r) => r.id), [boom]);++++ const write = (body) => h.http('POST', '/internal/chat/sounds/asset', { body, token: token('chat.sounds.write') });++ const body = check('chat.sound-asset-request@1', { id: honk, media_url: 'https://media.test/a/11', media_asset_id: 11 });++ for (let i = 0; i < 2; i++) {++ const r = await write(body);++ assert.strictEqual(r.status, 200, r.text);++ assert.deepStrictEqual(check('chat.ingress-ack@1', r.body), { ok: true });++ }++ assert.deepStrictEqual((await s('?pending_asset=1')).sounds.map((r) => r.id), [boom]);++ const row = await h.db.get('SELECT media_url, media_asset_id FROM channel_sounds WHERE id = ?', [honk]);++ assert.deepStrictEqual([row.media_url, Number(row.media_asset_id)], ['https://media.test/a/11', 11]);++ assert.strictEqual((await write({ ...body, id: 999999 })).status, 404);++ for (const bad of [{}, { ...body, id: 0 }, { ...body, media_url: '' }, { ...body, media_asset_id: '11' }, { ...body, extra: 1 }, { ...body, media_url: 'x'.repeat(2049) }]) {++ const r = await write(bad);++ assert.strictEqual(r.status, 400, JSON.stringify(bad));++ check('chat.ingress-ack@1', r.body);++ }++ for (const path of ['', '?channel_owner_id=abc', '?pending_asset=2', `?channel_owner_id=${streamer.id}&limit=5`]) await s(path, 400);++});++++t('the message ingress answers first_chat: true once per chatter and channel', async () => {++ const send = async (key, extra) => {++ const r = await h.http('POST', '/internal/chat/messages', { body: { key, stream_id: hostStreamId, message: 'hello host', ...extra }, token: token('chat.message.send') });++ assert.strictEqual(r.status, 200, r.text);++ // first_chat is in chat.send-result from 1.2.0 (openvibe-contracts 0.95.0).++ const { first_chat, ...older } = r.body;++ check('chat.send-result@1', sendResultHasFirstChat ? r.body : older);++ return r.body;++ };++ const viewerLine = { user_id: viewer.id, username: viewer.username };++ assert.strictEqual((await send('reads:first:1', viewerLine)).first_chat, true);++ assert.strictEqual((await send('reads:first:2', viewerLine)).first_chat, false);++ assert.strictEqual((await send('reads:first:1', viewerLine)).first_chat, true, 'a retry answers what was saved');++ assert.strictEqual((await send('reads:first:3', { anon_id: 'anon88', username: 'anon88' })).first_chat, true);++ assert.strictEqual((await send('reads:first:4', { ...viewerLine, message_type: 'system' })).first_chat, false, 'only chat lines use up a welcome');++ const r = await h.http('POST', '/internal/chat/messages', { body: { key: 'reads:first:dm', user_id: viewer.id, username: viewer.username, message: 'dm', dm: { to_user_id: other.id } }, token: token('chat.message.send') });++ assert.strictEqual(r.status, 200, r.text);++ assert.ok(!('first_chat' in check('chat.send-result@1', r.body)));++});++++t.run(async () => {++ if (unknownContracts.size) {++ console.log(`chat-internal-reads: schemas not validated (openvibe-contracts ${require('openvibe-contracts/package.json').version} lacks ${[...unknownContracts].join(', ')}; pin v0.95.0)`);++ }++ if (h) await h.close();++});@@@ -104,6 +104,19 @@ t('mirror: every change to Chat’s tables reaches Live, newest state per row',assert.deepStrictEqual(h.live.mirror.at(-1), { table: 'dm_messages', op: 'delete', pk: { id: dmId } });});++t('mirror: a channel sound without a Media asset leaves Live’s asset columns alone', async () => {++ const soundId = Number((await h.db.run(`INSERT INTO channel_sounds (channel_owner_id, command, url) VALUES (?, 'mirrorhonk', '/sounds/m.mp3')`, [streamer.id])).lastInsertRowid);++ await h.mirrorRelay.flush();++ const sent = h.live.mirror.filter((c) => c.table === 'channel_sounds' && c.row && c.row.id === soundId);++ assert.strictEqual(sent.length, 1);++ assert.ok(!('media_url' in sent[0].row) && !('media_asset_id' in sent[0].row), JSON.stringify(sent[0].row));++ await h.db.run('UPDATE channel_sounds SET media_url = ?, media_asset_id = ? WHERE id = ?', ['https://media.test/s/9', 9, soundId]);++ await h.mirrorRelay.flush();++ const row = h.live.mirror.filter((c) => c.table === 'channel_sounds' && c.row && c.row.id === soundId).at(-1).row;++ assert.strictEqual(row.media_url, 'https://media.test/s/9');++ assert.strictEqual(Number(row.media_asset_id), 9);++});++t('mirror: Live refusing (not yet CHAT_AUTHORITY=chat) keeps the queue; accepted later', async () => {h.live.mirrorStatus = 409;await h.db.recordRelayUser('twitch', 'Zed');@@@ -42,8 +42,11 @@ t('the seed applies after 0001–0002 and before 0003', async () => {}assert.strictEqual((await db.prepare(`SELECT to_regclass('chat_ingress_applied') AS t`).get()).t, null, '0003 has not run yet');const second = await db.migrate({ dir: MIGRATIONS, log: { log() {} } });-- assert.strictEqual(second.held.length, 0, 'the branch migration is not held');-- assert.deepStrictEqual(second.applied.map((a) => a.id), ['0003'], 'only the branch\'s migration applies after the seed');++ const later = fs.readdirSync(MIGRATIONS).filter((f) => /^\d{4}_.*\.sql$/.test(f)).map((f) => f.slice(0, 4))++ .filter((id) => id > '0002').sort();++ assert.ok(later.includes('0003'), 'migrations/0003 exists');++ assert.strictEqual(second.held.length, 0, 'no later migration is held');++ assert.deepStrictEqual(second.applied.map((a) => a.id), later, 'every migration after 0002 applies on top of the seed');});t('cleanup', async () => { await db.close(); });
@@ -96,6 +96,10 @@ before changing behaviour.** Browser JavaScript does not change; nginx routes th
@@ -96,6 +96,10 @@ before changing behaviour.** Browser JavaScript does not change; nginx routes th
`GET /internal/moderation/channels/:channelId/emote-count` — loopback + service token,
`GET /internal/moderation/channels/:channelId/emote-count` — loopback + service token,
capability `chat.moderation.read` (`chat.channel-moderation-result@1`,
capability `chat.moderation.read` (`chat.channel-moderation-result@1`,
`chat.moderated-channels-result@1`, `chat.emote-count-result@1`). Live caches each answer 30 s.
`chat.moderated-channels-result@1`, `chat.emote-count-result@1`). Live caches each answer 30 s.
+ - Live's reads of Chat's message, moderation-queue and sound tables (plan T3 J4b):
+ `POST /internal/chat/stats`, `GET /internal/chat/messages`, `/timeline`, `/moderation/*`,
+ `/sounds` and `POST /internal/chat/sounds/asset` (`server/chat/internal-reads.js`; routes and
+ capabilities in `docs/chat-ingress.md`).
- Alert sounds: `playAlertSound(chatServer, streamerId, streamId, kind)` (`server/chat/alert-sounds.js`)
- Alert sounds: `playAlertSound(chatServer, streamerId, streamId, kind)` (`server/chat/alert-sounds.js`)
resolves the channel's own settings row, reads the clip and broadcasts it to the channel room.
resolves the channel's own settings row, reads the clip and broadcasts it to the channel room.
- **The chat-AI summaries** (plan T3 step 2, decision 5) — Live's `server/ai/chat-ai.js` job moved
- **The chat-AI summaries** (plan T3 step 2, decision 5) — Live's `server/ai/chat-ai.js` job moved
@@ -7,8 +7,8 @@
@@ -7,8 +7,8 @@
"runtime": "node 22, express, ws, PostgreSQL through openvibe-sdk/db (DATABASE_URL via PgBouncer, DATABASE_DIRECT_URL for migrations/; embedded PGlite outside production), Valkey for the per-actor limits (VALKEY_URL); better-sqlite3 gone (plan T3)",
"runtime": "node 22, express, ws, PostgreSQL through openvibe-sdk/db (DATABASE_URL via PgBouncer, DATABASE_DIRECT_URL for migrations/; embedded PGlite outside production), Valkey for the per-actor limits (VALKEY_URL); better-sqlite3 gone (plan T3)",
"port": 4400,
"port": 4400,
"domain": "openvibe.chat (Chat's own site: global chat, messages, rooms, settings; the service manifest records it live since 2026-09-24) and openvibe.live (/ws/chat, /api/chat/, /api/dm/, /api/tts/, /api/sounds via an nginx include in openvibe.live's vhost)",
"domain": "openvibe.chat (Chat's own site: global chat, messages, rooms, settings; the service manifest records it live since 2026-09-24) and openvibe.live (/ws/chat, /api/chat/, /api/dm/, /api/tts/, /api/sounds via an nginx include in openvibe.live's vhost)",
- "contracts": "openvibe-contracts v0.91.0",
- "capabilities": ["chat.message.send", "chat.event.publish", "chat.moderation.write", "chat.cache.invalidate", "chat.presence.read", "chat.live_bridge.write"],
+ "contracts": "openvibe-contracts v0.96.0",
+ "capabilities": ["chat.message.send", "chat.event.publish", "chat.moderation.write", "chat.cache.invalidate", "chat.presence.read", "chat.live_bridge.write", "chat.stats.read", "chat.messages.read", "chat.analysis.read", "chat.moderation.queue.read", "chat.sounds.read", "chat.sounds.write"],
"capabilitiesUsed": [
"capabilitiesUsed": [
"live.chat_context.read",
"live.chat_context.read",
"live.chat_effects.write",
"live.chat_effects.write",
@@ -14,12 +14,30 @@ a stream that does not exist is 404: do not retry a 4xx.
@@ -14,12 +14,30 @@ a stream that does not exist is 404: do not retry a 4xx.
| Route | Capability | Body and result |
| Route | Capability | Body and result |
| --- | --- | --- |
| --- | --- | --- |
-| `POST /internal/chat/messages` | `chat.message.send` | Chat message fields (`stream_id`, `channel_user_id`, `user_id`, `anon_id`, `username`, `message`, `message_type`, `is_global`, `reply_to_id`, `source_platform`, `metadata`), optional `mirror`, `tts:{voice,identity_key,key}`, `auto_delete_at`, and typed chat-frame fields in `frame` (`role`: `admin`/`global_mod`/`streamer`/`user`/`anon`/`external`, `profile_color`: hex, `avatar_url`, `is_ai`/`is_bot`/`filtered`: booleans, `core_username`, `display_name`); the broadcast carries `reply_to_id`, `is_global`, `auto_delete_at` and `metadata` from the body. Returns `{ok,id}`. With no TTS key, Chat uses `m<real message id>`. `dm:{to_user_id}` with `user_id` persists and delivers a direct message (the same `dm` frame as `POST /api/dm`, with the sender looked up by id), returning `{ok,id,conversation_id,to_user_id}`; a blank message is 400. Chat records the first chat in the channel (the welcome check) under `first_chat_key` when given (`user:<id>`, `anon:<anonId>` or `ext:<prefixed username>`), otherwise for a `chat` line only: `ext:<username>` with `source_platform`, else `user:<user_id>`, else `anon:<anon_id>`. |
+| `POST /internal/chat/messages` | `chat.message.send` | Chat message fields (`stream_id`, `channel_user_id`, `user_id`, `anon_id`, `username`, `message`, `message_type`, `is_global`, `reply_to_id`, `source_platform`, `metadata`), optional `mirror`, `tts:{voice,identity_key,key}`, `auto_delete_at`, and typed chat-frame fields in `frame` (`role`: `admin`/`global_mod`/`streamer`/`user`/`anon`/`external`, `profile_color`: hex, `avatar_url`, `is_ai`/`is_bot`/`filtered`: booleans, `core_username`, `display_name`); the broadcast carries `reply_to_id`, `is_global`, `auto_delete_at` and `metadata` from the body. Returns `{ok,id,stream_id,channel_user_id,first_chat}`: `first_chat` (`chat.send-result` 1.2.0) is true when this line recorded the chatter's first chat in the channel, so Live can welcome them. With no TTS key, Chat uses `m<real message id>`. `dm:{to_user_id}` with `user_id` persists and delivers a direct message (the same `dm` frame as `POST /api/dm`, with the sender looked up by id), returning `{ok,id,conversation_id,to_user_id}`; a blank message is 400. Chat records the first chat in the channel (the welcome check) under `first_chat_key` when given (`user:<id>`, `anon:<anonId>` or `ext:<prefixed username>`), otherwise for a `chat` line only: `ext:<username>` with `source_platform`, else `user:<user_id>`, else `anon:<anon_id>`. |
| `POST /internal/chat/events` | `chat.event.publish` | `{key,target:{kind,id?},frame:{type,...}}`, returns `{ok:true}`. Target kinds: `stream`, `channel`, `owner-streams`, `user` (positive `id`), `global`, `all` (no `id`). Only the event frame types allowlisted in `internal-ingress.js` are accepted; `dm` and delete frames are refused, and `chat` only as the news card. |
| `POST /internal/chat/events` | `chat.event.publish` | `{key,target:{kind,id?},frame:{type,...}}`, returns `{ok:true}`. Target kinds: `stream`, `channel`, `owner-streams`, `user` (positive `id`), `global`, `all` (no `id`). Only the event frame types allowlisted in `internal-ingress.js` are accepted; `dm` and delete frames are refused, and `chat` only as the news card. |
| `POST /internal/chat/moderation` | `chat.moderation.write` | `{key,action,...}`; actions: `delete-message`, `delete-user-messages`, `delete-anon-messages`, `delete-relay-messages`, `delete-by-range`, `review-pending-ip`, `approve-ip-messages`, `deny-ip-messages`, `relay-hide`, `relay-unhide`, `relay-record`, `tts-voice-override`, `disconnect`, `log`. Deletes return `ids` and broadcast `delete-messages` to every affected surface. |
| `POST /internal/chat/moderation` | `chat.moderation.write` | `{key,action,...}`; actions: `delete-message`, `delete-user-messages`, `delete-anon-messages`, `delete-relay-messages`, `delete-by-range`, `review-pending-ip`, `approve-ip-messages`, `deny-ip-messages`, `relay-hide`, `relay-unhide`, `relay-record`, `tts-voice-override`, `disconnect`, `log`. Deletes return `ids` and broadcast `delete-messages` to every affected surface. |
| `POST /internal/chat/invalidate` | `chat.cache.invalidate` | `{key,user?,user_data?,approvals?,bans?,channel?}`; a cache hint, returns `{ok:true}`. Include a typed `user_data` object when Live edited the account, so Chat updates sockets and stored message names. |
| `POST /internal/chat/invalidate` | `chat.cache.invalidate` | `{key,user?,user_data?,approvals?,bans?,channel?}`; a cache hint, returns `{ok:true}`. Include a typed `user_data` object when Live edited the account, so Chat updates sockets and stored message names. |
| `GET /internal/chat/presence` | `chat.presence.read` | Who is connected where: counts, per-stream viewers, slow modes, users/anons with their addresses (`server/chat/presence.js`). |
| `GET /internal/chat/presence` | `chat.presence.read` | Who is connected where: counts, per-stream viewers, slow modes, users/anons with their addresses (`server/chat/presence.js`). |
+Live's reads of Chat's tables, in place of its mirrored copy (`server/chat/internal-reads.js`, plan
+T3 J4b N1–N5, contracts in `openvibe-contracts` 0.95.0). Unknown or malformed parameters are 400;
+deleted messages are never read; `since`/`until` are epoch ms (`until` exclusive). Every answer is
+`{ok:true,...}` or `{ok:false,error}`.
+
+| Route | Capability | Request and result |
+| --- | --- | --- |
+| `POST /internal/chat/stats` | `chat.stats.read` | `{kind:'site'\|'user'\|'stream'\|'channel-top', user_id?, stream_id?, channel_user_id?, since?, until?, limit?}`. `site`/`user`(`user_id`)/`stream`(`stream_id`) → `{messages,chatters}` (+ `sounds`, soundboard plays, for `stream`); `channel-top` (by `stream_id`, `channel_user_id` or all chat; `limit` ≤ 50, default 10) → `{top_chatters:[{user_id,username,display_name,avatar_url,profile_color,count}]}`. |
+| `GET /internal/chat/messages` | `chat.messages.read` | Exactly one of `channel_user_id`, `stream_id`, `user_id`, `anon_id`, `username`, `id`; `after_id` (oldest first), `before_id` (newest first; the default), `limit` ≤ 500 (default 100), `types=chat,donation,…`, `tail=1` (no rows) → `{messages,max_id}`; `max_id` is the newest matching id. |
+| `GET /internal/chat/timeline` | `chat.analysis.read` | `channel_user_id` or `stream_id`, `since`, `until` (default now), `bucket_ms` ≥ 1000 (≤ 10 000 buckets) → `{buckets:[{t,count}],max_id}`; buckets start at `since`, oldest first, empty ones left out. |
+| `GET /internal/chat/moderation/pending-ip?channel_id` | `chat.moderation.queue.read` | `{pending_ip}`: the channel's pending rows, oldest first (`limit` ≤ 500, default 50). |
+| `GET /internal/chat/moderation/relay-users?channel_id` | `chat.moderation.queue.read` | `{relay_users}`: hidden or banned relay names on the channel or everywhere, newest first, with `created_by_username` (`limit` ≤ 500, default 100). |
+| `GET /internal/chat/moderation/relay-users/:id` | `chat.moderation.queue.read` | `{relay_user}`, null when unknown. |
+| `GET /internal/chat/moderation/tts-override?identity_key` | `chat.moderation.queue.read` | `{tts_override}` (key trimmed and lowercased), null when none. |
+| `GET /internal/chat/sounds?channel_owner_id` (also `/sounds/count`) | `chat.sounds.read` | `{count}` of the channel's sounds. |
+| `GET /internal/chat/sounds?pending_asset=1` | `chat.sounds.read` | `{sounds}` not yet on Media (`media_asset_id` null), by id; optional `channel_owner_id`, `after_id`, `limit` ≤ 500 (default 100). |
+| `POST /internal/chat/sounds/asset` | `chat.sounds.write` | `{id,media_url,media_asset_id}` records Live's upload (`migrations/0004_channel_sound_media.sql`); repeating it is a 200 no-op; an unknown sound is 404. Until Live writes here, the read mirror leaves Live's own asset columns alone for a sound Chat has no asset for. |
+
`/internal/chat/events` is a narrow transient event ingress, not a generic ChatServer call.
`/internal/chat/events` is a narrow transient event ingress, not a generic ChatServer call.
`live.release.deployed` continues through OpenVibe.Events; there is no deploy-notice endpoint.
`live.release.deployed` continues through OpenVibe.Events; there is no deploy-notice endpoint.
An `alert` targets its `streamerId` channel; a `channel-sound` targets its `streamId` stream.
An `alert` targets its `streamerId` channel; a `channel-sound` targets its `streamId` stream.
@@ -41,7 +59,11 @@ Before Live repoints, the separate **OpenVibe.Contracts** repository must regist
@@ -41,7 +59,11 @@ Before Live repoints, the separate **OpenVibe.Contracts** repository must regist
`live chat.event.publish openvibe.chat`, `live chat.moderation.write openvibe.chat`, and
`live chat.event.publish openvibe.chat`, `live chat.moderation.write openvibe.chat`, and
`live chat.cache.invalidate openvibe.chat`. Keep the Live grants for `chat.message.send`
`live chat.cache.invalidate openvibe.chat`. Keep the Live grants for `chat.message.send`
(`messages`), `chat.presence.read` (`/internal/chat/presence`) and `chat.live_bridge.write` (the call
(`messages`), `chat.presence.read` (`/internal/chat/presence`) and `chat.live_bridge.write` (the call
-hooks at `/internal/calls/stream-channel`): the bridge is gone, but each still guards a route. Chat's
+hooks at `/internal/calls/stream-channel`): the bridge is gone, but each still guards a route. For
+the read API above, Network's `DEFAULT_GRANTS` must add `live chat.stats.read openvibe.chat`,
+`live chat.messages.read openvibe.chat`, `live chat.analysis.read openvibe.chat`,
+`live chat.moderation.queue.read openvibe.chat`, `live chat.sounds.read openvibe.chat` and
+`live chat.sounds.write openvibe.chat`. Chat's
local proposals are in `docs/capabilities-proposal/`.
local proposals are in `docs/capabilities-proposal/`.
Live J2 (done: Live #20/#21) replaced its bridge writer with a typed client: repoint message producers and persisted DMs
Live J2 (done: Live #20/#21) replaced its bridge writer with a typed client: repoint message producers and persisted DMs
new file mode 100644
new file mode 100644
@@ -0,0 +1,8 @@
@@ -0,0 +1,8 @@
+-- phase: expand
+-- Where Live's asset sync put each channel sound on OpenVibe.Media (plan T3 J4b, N5): Live reads the
+-- sounds still to upload (GET /internal/chat/sounds?pending_asset=1) and writes the asset back
+-- (POST /internal/chat/sounds/asset), instead of using its mirrored copy of channel_sounds.
+-- Both stay NULL until the sound is uploaded.
+ALTER TABLE channel_sounds ADD COLUMN IF NOT EXISTS media_url text COLLATE "C";
+ALTER TABLE channel_sounds ADD COLUMN IF NOT EXISTS media_asset_id bigint;
+CREATE INDEX IF NOT EXISTS idx_channel_sounds_pending_asset ON channel_sounds(id) WHERE media_asset_id IS NULL;
@@ -16,7 +16,7 @@
@@ -16,7 +16,7 @@
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"multer": "^2.0.2",
"multer": "^2.0.2",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.91.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.96.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.5.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.5.0",
"pg": "^8.23.0",
"pg": "^8.23.0",
@@ -1294,9 +1294,9 @@
@@ -1294,9 +1294,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.91.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.91.0",
- "integrity": "sha512-Sg5WUFRcnH+5TJGHciFU++dwWqXjhcKngscmICWHV8eGrt8kgTZfH85EeWtlG1LjdIRlWW9h2wK0Wtd5n4OGWw==",
+ "version": "0.96.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.96.0",
+ "integrity": "sha512-QEXfNYNkK6n49sXokgbP2FTKaIEawRJOS7Xij9g20fBFSqiVv14voZ60LqECPl1VMZedoSGuYLa8doVlnSxH0w==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -29,7 +29,7 @@
@@ -29,7 +29,7 @@
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
"multer": "^2.0.2",
"multer": "^2.0.2",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.91.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.96.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.26.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.5.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v2.5.0",
"pg": "^8.23.0",
"pg": "^8.23.0",
@@ -102,6 +102,8 @@ function createApp({ chatServer, mirror, relay, events = null, callServer = null
@@ -102,6 +102,8 @@ function createApp({ chatServer, mirror, relay, events = null, callServer = null
// ── Internal (loopback; never routed by nginx) ─────────────
// ── Internal (loopback; never routed by nginx) ─────────────
app.use('/internal/chat', express.json({ limit: '64kb' }), createInternalIngress({ chatServer }));
app.use('/internal/chat', express.json({ limit: '64kb' }), createInternalIngress({ chatServer }));
+ // Live's reads of the chat tables Chat owns, in place of its mirrored copy (plan T3 J4b, server/chat/internal-reads.js).
+ app.use('/internal/chat', express.json({ limit: '64kb' }), require('./chat/internal-reads'));
// OpenVibe.Events deliveries (live.release.deployed, network.module.updated): signature v2 over
// OpenVibe.Events deliveries (live.release.deployed, network.module.updated): signature v2 over
// the raw body, so this router reads the body itself (server/events/consumer.js).
// the raw body, so this router reads the body itself (server/events/consumer.js).
if (events) app.use('/internal/events', events.router);
if (events) app.use('/internal/events', events.router);
@@ -44,6 +44,13 @@ const ROW_SQL = {
@@ -44,6 +44,13 @@ const ROW_SQL = {
moderation_actions: 'SELECT * FROM moderation_actions WHERE id = ?',
moderation_actions: 'SELECT * FROM moderation_actions WHERE id = ?',
};
};
+// Live's asset sync still writes media_url/media_asset_id on its own copy until it moves to
+// POST /internal/chat/sounds/asset: a sound Chat has no asset for must not blank Live's.
+function soundRow(row) {
+ const { media_url, media_asset_id, ...rest } = row;
+ return media_asset_id == null ? rest : row;
+}
+
function createMirror({ config, fetchImpl = (...a) => globalThis.fetch(...a), log = console } = {}) {
function createMirror({ config, fetchImpl = (...a) => globalThis.fetch(...a), log = console } = {}) {
let timer = null;
let timer = null;
let busy = false;
let busy = false;
@@ -63,7 +70,7 @@ function createMirror({ config, fetchImpl = (...a) => globalThis.fetch(...a), lo
@@ -63,7 +70,7 @@ function createMirror({ config, fetchImpl = (...a) => globalThis.fetch(...a), lo
try { pk = JSON.parse(r.pk); } catch { continue; }
try { pk = JSON.parse(r.pk); } catch { continue; }
if (r.op === 'delete') { changes.push({ table: r.tbl, op: 'delete', pk }); continue; }
if (r.op === 'delete') { changes.push({ table: r.tbl, op: 'delete', pk }); continue; }
const row = await db.get(ROW_SQL[r.tbl], cols.map((c) => pk[c]));
const row = await db.get(ROW_SQL[r.tbl], cols.map((c) => pk[c]));
- changes.push(row ? { table: r.tbl, op: 'upsert', row: r.tbl === 'moderation_actions' ? db.moderationRow(row) : r.tbl === 'chat_messages' ? db.chatMessageRow(row) : row } : { table: r.tbl, op: 'delete', pk });
+ changes.push(row ? { table: r.tbl, op: 'upsert', row: r.tbl === 'moderation_actions' ? db.moderationRow(row) : r.tbl === 'chat_messages' ? db.chatMessageRow(row) : r.tbl === 'channel_sounds' ? soundRow(row) : row } : { table: r.tbl, op: 'delete', pk });
}
}
return changes;
return changes;
}
}
@@ -154,8 +154,9 @@ async function writeMessage(b) {
@@ -154,8 +154,9 @@ async function writeMessage(b) {
const id = Number(saved.lastInsertRowid);
const id = Number(saved.lastInsertRowid);
const channelUserId = b.channel_user_id || (b.stream_id ? (await ctx.getStreamById(b.stream_id))?.user_id : null) || null;
const channelUserId = b.channel_user_id || (b.stream_id ? (await ctx.getStreamById(b.stream_id))?.user_id : null) || null;
const chatterKey = firstChatKey(b);
const chatterKey = firstChatKey(b);
- if (channelUserId && chatterKey) await db.recordFirstChat(chatterKey, channelUserId);
- return { id, stream_id: b.stream_id || null, channel_user_id: channelUserId };
+ // first_chat (chat.send-result 1.2.0): this line used up the chatter's welcome in the channel.
+ const first_chat = !!(channelUserId && chatterKey) && await db.recordFirstChat(chatterKey, channelUserId);
+ return { id, stream_id: b.stream_id || null, channel_user_id: channelUserId, first_chat };
}
}
async function deliverMessage(chatServer, b, result) {
async function deliverMessage(chatServer, b, result) {
new file mode 100644
new file mode 100644
@@ -0,0 +1,297 @@
@@ -0,0 +1,297 @@
+/**
+ * OpenVibe.Chat — the internal read API Live calls instead of reading its mirrored copy of Chat's
+ * tables (plan T3 J4b, design §1d N1–N5). Loopback only, a service token for audience
+ * openvibe.chat, one capability per route (openvibe-contracts 0.95.0):
+ *
+ * POST /internal/chat/stats chat.stats.read chat.stats-request@1 → chat.stats-result@1
+ * GET /internal/chat/messages chat.messages.read → chat.messages-page@1
+ * GET /internal/chat/timeline chat.analysis.read → chat.timeline-result@1
+ * GET /internal/chat/moderation/pending-ip chat.moderation.queue.read → chat.moderation-queue-result@1 { pending_ip }
+ * GET /internal/chat/moderation/relay-users chat.moderation.queue.read → { relay_users }
+ * GET /internal/chat/moderation/relay-users/:id chat.moderation.queue.read → { relay_user }
+ * GET /internal/chat/moderation/tts-override chat.moderation.queue.read → { tts_override }
+ * GET /internal/chat/sounds(/count) chat.sounds.read → chat.sounds-result@1 { count } | { sounds }
+ * POST /internal/chat/sounds/asset chat.sounds.write chat.sound-asset-request@1 → chat.ingress-ack@1
+ *
+ * Every parameter is validated (unknown ones are 400) and bound; deleted messages are never read.
+ * Times in requests are epoch ms; chat_messages.timestamp is UTC 'YYYY-MM-DD HH:MM:SS' text, so a
+ * window compares text and keeps the timestamp indexes.
+ */
+'use strict';
+
+const express = require('express');
+const db = require('../db/database');
+const serviceAuth = require('../net/service-auth');
+
+const MESSAGE_TYPES = ['chat', 'system', 'donation', 'command', 'tts', 'channel-sound', 'soundboard', 'clip'];
+const MAX_MESSAGES = 500;
+const MAX_ROWS = 500;
+const MAX_BUCKETS = 10_000;
+
+const bad = (message) => { const e = new Error(message); e.status = 400; throw e; };
+const id = (v) => Number.isSafeInteger(v) && v > 0;
+const ms = (v) => Number.isSafeInteger(v) && v >= 0;
+const sqlTime = (t) => new Date(t).toISOString().slice(0, 19).replace('T', ' ');
+const num = (v) => (v == null ? null : Number(v));
+const text = (v) => (v == null ? null : String(v));
+
+/** Query string → validated values; anything not in `spec` is 400. */
+function query(req, spec) {
+ const out = {};
+ for (const [k, raw] of Object.entries(req.query)) {
+ const kind = spec[k];
+ if (!kind || typeof raw !== 'string') bad(`Invalid parameter ${k}`);
+ if (kind === 'id' || kind === 'ms') {
+ const v = /^\d{1,16}$/.test(raw) ? Number(raw) : NaN;
+ if (!(kind === 'id' ? id(v) : ms(v))) bad(`Invalid ${k}`);
+ out[k] = v;
+ } else if (kind === 'flag') {
+ if (raw !== '1' && raw !== '0') bad(`Invalid ${k}`);
+ out[k] = raw === '1';
+ } else if (kind === 'types') {
+ const types = raw.split(',');
+ if (!types.every((t) => MESSAGE_TYPES.includes(t))) bad(`Invalid ${k}`);
+ out[k] = types;
+ } else {
+ if (!raw || raw.length > kind) bad(`Invalid ${k}`);
+ out[k] = raw;
+ }
+ }
+ return out;
+}
+
+function limit(v, dflt, max) { return v == null ? dflt : Math.min(v, max); }
+
+function route(name, fn) {
+ return async (req, res) => {
+ try {
+ res.json({ ok: true, ...await fn(req) });
+ } catch (err) {
+ if (err.status) return res.status(err.status).json({ ok: false, error: err.message });
+ console.error(`[ChatReads] ${name}:`, err);
+ res.status(503).json({ ok: false, error: 'Chat read unavailable' });
+ }
+ };
+}
+
+// ── N1: stats ──
+const CHATTER = `COALESCE('u:' || user_id, 'a:' || anon_id, source_platform || ':' || username)`;
+const STATS_FIELDS = ['kind', 'user_id', 'stream_id', 'channel_user_id', 'since', 'until', 'limit'];
+
+function statsFilter(b) {
+ const where = ['is_deleted = 0'];
+ const params = [];
+ for (const k of ['user_id', 'stream_id', 'channel_user_id']) if (b[k] != null) { where.push(`${k} = ?`); params.push(b[k]); }
+ if (b.since != null) { where.push('timestamp >= ?'); params.push(sqlTime(b.since)); }
+ if (b.until != null) { where.push('timestamp < ?'); params.push(sqlTime(b.until)); }
+ return { where: where.join(' AND '), params };
+}
+
+async function stats(req) {
+ const b = req.body;
+ if (!b || typeof b !== 'object' || Array.isArray(b) || Object.keys(b).some((k) => !STATS_FIELDS.includes(k))) bad('Invalid fields');
+ if (!['site', 'user', 'stream', 'channel-top'].includes(b.kind)) bad('Invalid kind');
+ for (const k of ['user_id', 'stream_id', 'channel_user_id']) if (b[k] != null && !id(b[k])) bad(`Invalid ${k}`);
+ for (const k of ['since', 'until']) if (b[k] != null && !ms(b[k])) bad(`Invalid ${k}`);
+ if (b.limit != null && !(Number.isSafeInteger(b.limit) && b.limit >= 1 && b.limit <= 50)) bad('Invalid limit');
+ if (b.kind === 'user' && b.user_id == null) bad('kind user needs user_id');
+ if (b.kind === 'stream' && b.stream_id == null) bad('kind stream needs stream_id');
+ if (b.kind === 'channel-top' && b.user_id != null) bad('kind channel-top takes stream_id or channel_user_id');
+ const { where, params } = statsFilter(b);
+ if (b.kind !== 'channel-top') {
+ const r = await db.get(
+ `SELECT COUNT(*) AS messages, COUNT(DISTINCT ${CHATTER}) AS chatters,
+ COUNT(*) FILTER (WHERE message_type = 'soundboard') AS sounds
+ FROM chat_messages WHERE ${where}`, params);
+ const out = { messages: Number(r.messages), chatters: Number(r.chatters) };
+ if (b.kind === 'stream') out.sounds = Number(r.sounds);
+ return out;
+ }
+ // Registered chatters group by user id; anonymous and relayed ones by name. System lines are no one's.
+ const rows = await db.all(
+ `SELECT t.user_id, COALESCE(u.username, t.username) AS username, u.display_name, u.avatar_url, u.profile_color, t.count
+ FROM (SELECT user_id, MAX(username) AS username, COUNT(*) AS count, MAX(id) AS last_id
+ FROM chat_messages
+ WHERE ${where} AND message_type <> 'system' AND ${CHATTER} IS NOT NULL
+ GROUP BY user_id, CASE WHEN user_id IS NULL THEN ${CHATTER} END) t
+ LEFT JOIN ctx_users u ON u.id = t.user_id
+ ORDER BY t.count DESC, t.last_id DESC LIMIT ?`, [...params, b.limit || 10]);
+ return {
+ top_chatters: rows.map((r) => ({
+ user_id: num(r.user_id), username: String(r.username || ''), display_name: text(r.display_name),
+ avatar_url: text(r.avatar_url), profile_color: text(r.profile_color), count: Number(r.count),
+ })),
+ };
+}
+
+// ── N2: messages ──
+const MESSAGE_FILTERS = { channel_user_id: 'id', stream_id: 'id', user_id: 'id', anon_id: 80, username: 120, id: 'id' };
+const MESSAGE_COLUMNS = 'id, user_id, anon_id, username, message, message_type, is_global, stream_id, channel_user_id, source_platform, reply_to_id, timestamp';
+
+function messageRow(r) {
+ return {
+ id: Number(r.id), user_id: num(r.user_id), anon_id: text(r.anon_id), username: text(r.username),
+ message: String(r.message), message_type: r.message_type || 'chat', is_global: Number(r.is_global) ? 1 : 0,
+ stream_id: num(r.stream_id), channel_user_id: num(r.channel_user_id), source_platform: text(r.source_platform),
+ reply_to_id: num(r.reply_to_id), timestamp: String(r.timestamp || ''),
+ };
+}
+
+async function messages(req) {
+ const q = query(req, { ...MESSAGE_FILTERS, after_id: 'id', before_id: 'id', limit: 'id', tail: 'flag', types: 'types' });
+ const filters = Object.keys(MESSAGE_FILTERS).filter((k) => q[k] != null);
+ if (filters.length !== 1) bad('Give exactly one of channel_user_id, stream_id, user_id, anon_id, username or id');
+ const where = [`${filters[0]} = ?`, 'is_deleted = 0'];
+ const params = [q[filters[0]]];
+ if (q.types) { where.push(`message_type IN (${q.types.map(() => '?').join(', ')})`); params.push(...q.types); }
+ const max = await db.get(`SELECT MAX(id) AS max_id FROM chat_messages WHERE ${where.join(' AND ')}`, params);
+ const max_id = num(max && max.max_id);
+ if (q.tail) return { messages: [], max_id };
+ const page = [...where];
+ const pageParams = [...params];
+ if (q.after_id != null) { page.push('id > ?'); pageParams.push(q.after_id); }
+ if (q.before_id != null) { page.push('id < ?'); pageParams.push(q.before_id); }
+ // after_id pages forward (oldest first); otherwise newest first, back from before_id.
+ const order = q.after_id != null ? 'ASC' : 'DESC';
+ const rows = await db.all(
+ `SELECT ${MESSAGE_COLUMNS} FROM chat_messages WHERE ${page.join(' AND ')} ORDER BY id ${order} LIMIT ?`,
+ [...pageParams, limit(q.limit, 100, MAX_MESSAGES)]);
+ return { messages: rows.map(messageRow), max_id };
+}
+
+// ── N3: timeline ──
+async function timeline(req) {
+ const q = query(req, { channel_user_id: 'id', stream_id: 'id', since: 'ms', until: 'ms', bucket_ms: 'id' });
+ if ((q.channel_user_id == null) === (q.stream_id == null)) bad('Give channel_user_id or stream_id');
+ if (q.since == null || q.bucket_ms == null) bad('since and bucket_ms are required');
+ const until = q.until == null ? Date.now() : q.until;
+ if (until <= q.since) bad('until must be after since');
+ if (q.bucket_ms < 1000 || (until - q.since) / q.bucket_ms > MAX_BUCKETS) bad('Invalid bucket_ms');
+ const col = q.channel_user_id != null ? 'channel_user_id' : 'stream_id';
+ // Buckets start at since: bucket n covers [since + n·bucket_ms, since + (n+1)·bucket_ms).
+ const rows = await db.all(
+ `SELECT n, COUNT(*) AS count, MAX(id) AS max_id FROM (
+ SELECT id, floor((extract(epoch FROM ov_ts(timestamp)) * 1000 - ?) / ?)::bigint AS n
+ FROM chat_messages
+ WHERE ${col} = ? AND is_deleted = 0 AND timestamp >= ? AND timestamp < ?) m
+ WHERE n IS NOT NULL AND n >= 0 GROUP BY n ORDER BY n`,
+ [q.since, q.bucket_ms, q[col], sqlTime(q.since), sqlTime(until)]);
+ return {
+ buckets: rows.map((r) => ({ t: q.since + Number(r.n) * q.bucket_ms, count: Number(r.count) })),
+ max_id: rows.length ? Math.max(...rows.map((r) => Number(r.max_id))) : null,
+ };
+}
+
+// ── N4: moderation queues ──
+function pendingIpRow(r) {
+ return {
+ id: Number(r.id), channel_id: Number(r.channel_id), stream_id: num(r.stream_id), ip_address: String(r.ip_address),
+ user_id: num(r.user_id), anon_id: text(r.anon_id), username: text(r.username), message: String(r.message),
+ status: r.status || 'pending', reviewed_by: num(r.reviewed_by), created_at: text(r.created_at),
+ };
+}
+function relayUserRow(r) {
+ return {
+ id: Number(r.id), channel_id: num(r.channel_id), platform: String(r.platform), external_username: String(r.external_username),
+ action: r.action || 'hide', reason: text(r.reason), created_by: num(r.created_by),
+ created_by_username: text(r.created_by_username), created_at: text(r.created_at),
+ };
+}
+const RELAY_SELECT = `SELECT h.*, u.username AS created_by_username
+ FROM hidden_relay_users h LEFT JOIN ctx_users u ON u.id = h.created_by`;
+
+async function pendingIp(req) {
+ const q = query(req, { channel_id: 'id', limit: 'id' });
+ if (q.channel_id == null) bad('channel_id is required');
+ const rows = await db.all(
+ `SELECT * FROM pending_ip_messages WHERE channel_id = ? AND status = 'pending' ORDER BY created_at ASC, id ASC LIMIT ?`,
+ [q.channel_id, limit(q.limit, 50, MAX_ROWS)]);
+ return { pending_ip: rows.map(pendingIpRow) };
+}
+
+async function relayUsers(req) {
+ const q = query(req, { channel_id: 'id', limit: 'id' });
+ if (q.channel_id == null) bad('channel_id is required');
+ const rows = await db.all(
+ `${RELAY_SELECT} WHERE h.channel_id = ? OR h.channel_id IS NULL ORDER BY h.created_at DESC, h.id DESC LIMIT ?`,
+ [q.channel_id, limit(q.limit, 100, MAX_ROWS)]);
+ return { relay_users: rows.map(relayUserRow) };
+}
+
+async function relayUser(req) {
+ query(req, {});
+ const relayId = /^\d{1,16}$/.test(req.params.id) ? Number(req.params.id) : NaN;
+ if (!id(relayId)) bad('Invalid id');
+ const row = await db.get(`${RELAY_SELECT} WHERE h.id = ?`, [relayId]);
+ return { relay_user: row ? relayUserRow(row) : null };
+}
+
+async function ttsOverride(req) {
+ const q = query(req, { identity_key: 120 });
+ const key = String(q.identity_key || '').trim().toLowerCase();
+ if (!key) bad('identity_key is required');
+ const r = await db.get('SELECT * FROM tts_voice_overrides WHERE identity_key = ?', [key]);
+ return {
+ tts_override: r ? {
+ identity_key: String(r.identity_key), voice: text(r.voice), pitch: num(r.pitch), speed: num(r.speed),
+ gap: num(r.gap) || 0, set_by: num(r.set_by), updated_at: text(r.updated_at),
+ } : null,
+ };
+}
+
+// ── N5: channel sounds ──
+function soundRow(r) {
+ return {
+ id: Number(r.id), channel_owner_id: Number(r.channel_owner_id), command: String(r.command), url: String(r.url),
+ mime: text(r.mime), duration_seconds: num(r.duration_seconds), created_by: num(r.created_by),
+ created_by_name: text(r.created_by_name), media_url: text(r.media_url), media_asset_id: num(r.media_asset_id),
+ };
+}
+
+async function countSounds(ownerId) {
+ if (ownerId == null) bad('channel_owner_id is required');
+ const r = await db.get('SELECT COUNT(*) AS n FROM channel_sounds WHERE channel_owner_id = ?', [ownerId]);
+ return { count: Number(r.n) };
+}
+const soundCount = (req) => countSounds(query(req, { channel_owner_id: 'id' }).channel_owner_id);
+
+/** ?channel_owner_id → { count }; ?pending_asset=1 → { sounds } not yet uploaded to Media (oldest first). */
+async function sounds(req) {
+ const q = query(req, { channel_owner_id: 'id', pending_asset: 'flag', after_id: 'id', limit: 'id' });
+ if (!q.pending_asset) {
+ if (q.after_id != null || q.limit != null) bad('after_id and limit page pending_asset=1');
+ return countSounds(q.channel_owner_id);
+ }
+ const where = ['media_asset_id IS NULL'];
+ const params = [];
+ if (q.channel_owner_id != null) { where.push('channel_owner_id = ?'); params.push(q.channel_owner_id); }
+ if (q.after_id != null) { where.push('id > ?'); params.push(q.after_id); }
+ const rows = await db.all(`SELECT * FROM channel_sounds WHERE ${where.join(' AND ')} ORDER BY id LIMIT ?`,
+ [...params, limit(q.limit, 100, MAX_ROWS)]);
+ return { sounds: rows.map(soundRow) };
+}
+
+/** Live's asset sync uploaded the file to Media: record where. Repeating the same write is a no-op 200. */
+async function soundAsset(req) {
+ const b = req.body;
+ if (!b || typeof b !== 'object' || Array.isArray(b) || Object.keys(b).some((k) => !['id', 'media_url', 'media_asset_id'].includes(k))) bad('Invalid fields');
+ if (!id(b.id) || !id(b.media_asset_id) || typeof b.media_url !== 'string' || !b.media_url || b.media_url.length > 2048) bad('Invalid sound asset');
+ const r = await db.run('UPDATE channel_sounds SET media_url = ?, media_asset_id = ? WHERE id = ?', [b.media_url, b.media_asset_id, b.id]);
+ if (!r.changes) { const e = new Error('Sound not found'); e.status = 404; throw e; }
+ return {};
+}
+
+const router = express.Router();
+router.post('/stats', serviceAuth.guard('chat.stats.read'), route('stats', stats));
+router.get('/messages', serviceAuth.guard('chat.messages.read'), route('messages', messages));
+router.get('/timeline', serviceAuth.guard('chat.analysis.read'), route('timeline', timeline));
+const queue = serviceAuth.guard('chat.moderation.queue.read');
+router.get('/moderation/pending-ip', queue, route('pending-ip', pendingIp));
+router.get('/moderation/relay-users', queue, route('relay-users', relayUsers));
+router.get('/moderation/relay-users/:id', queue, route('relay-user', relayUser));
+router.get('/moderation/tts-override', queue, route('tts-override', ttsOverride));
+router.get('/sounds', serviceAuth.guard('chat.sounds.read'), route('sounds', sounds));
+router.get('/sounds/count', serviceAuth.guard('chat.sounds.read'), route('sound-count', soundCount));
+router.post('/sounds/asset', serviceAuth.guard('chat.sounds.write'), route('sound-asset', soundAsset));
+
+module.exports = router;
@@ -637,12 +637,14 @@ async function isFirstChatInChannel(chatterKey, channelUserId) {
@@ -637,12 +637,14 @@ async function isFirstChatInChannel(chatterKey, channelUserId) {
/**
/**
* Record that a chatter has chatted in a streamer's channel.
* Record that a chatter has chatted in a streamer's channel.
+ * @returns {boolean} true when this was their first time (the row is new)
*/
*/
async function recordFirstChat(chatterKey, channelUserId) {
async function recordFirstChat(chatterKey, channelUserId) {
- await run(
+ const r = await run(
'INSERT INTO stream_first_chats (chatter_key, channel_user_id) VALUES (?, ?) ON CONFLICT DO NOTHING',
'INSERT INTO stream_first_chats (chatter_key, channel_user_id) VALUES (?, ?) ON CONFLICT DO NOTHING',
[chatterKey, channelUserId]
[chatterKey, channelUserId]
);
);
+ return r.changes > 0;
}
}
// ── Moderation log ───────────────────────────────────────────
// ── Moderation log ───────────────────────────────────────────
new file mode 100644
new file mode 100644
@@ -0,0 +1,268 @@
@@ -0,0 +1,268 @@
+'use strict';
+/**
+ * Live's internal read API over Chat's tables (server/chat/internal-reads.js, plan T3 J4b N1–N5):
+ * one capability per route, each answer valid against its openvibe-contracts schema, and
+ * first_chat on the message ingress's answer.
+ */
+
+const assert = require('assert');
+const { validate, catalog } = require('openvibe-contracts');
+const { boot, suite } = require('./helpers');
+
+const t = suite('chat-internal-reads');
+let h, streamer, viewer, other, host, channelId, streamId, otherStreamId, hostStreamId;
+const T = Date.UTC(2026, 0, 1, 12, 0, 0);
+const at = (ms) => new Date(T + ms).toISOString().slice(0, 19).replace('T', ' ');
+const ids = {};
+
+// The read contracts arrive with openvibe-contracts 0.95.0. Until the pin reaches it, an older
+// package cannot validate them: the run says so instead of failing (the shapes are asserted anyway).
+const unknownContracts = new Set();
+const sendResultHasFirstChat = (() => {
+ const [major, minor] = catalog.find((c) => c.id === 'chat.send-result').version.split('.').map(Number);
+ return major > 1 || minor >= 2;
+})();
+function check(ref, body) {
+ let r;
+ try { r = validate(ref, body); } catch (err) {
+ if (!/unknown contract|is v\d/.test(err.message)) throw err;
+ unknownContracts.add(ref);
+ return body;
+ }
+ assert.ok(r.valid, `${ref} invalid: ${JSON.stringify(r.errors)} (${JSON.stringify(body)})`);
+ return body;
+}
+const token = (cap) => h.serviceToken([cap]);
+async function read(path, cap, ref, status = 200) {
+ const r = await h.http('GET', `/internal/chat${path}`, { token: token(cap) });
+ assert.strictEqual(r.status, status, `${path}: ${r.text}`);
+ return status === 200 ? check(ref, r.body) : r.body;
+}
+async function stats(body, status = 200) {
+ const r = await h.http('POST', '/internal/chat/stats', { body, token: token('chat.stats.read') });
+ assert.strictEqual(r.status, status, `${JSON.stringify(body)}: ${r.text}`);
+ return check('chat.stats-result@1', r.body);
+}
+async function line(name, { ms, user = null, anon = null, username, type = 'chat', platform = null, deleted = 0, stream = streamId, channel = streamer.id }) {
+ const r = await h.db.run(
+ `INSERT INTO chat_messages (stream_id, channel_user_id, user_id, anon_id, username, message, message_type, source_platform, is_deleted, timestamp)
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+ [stream, channel, user, anon, username, `line ${name}`, type, platform, deleted, at(ms)]);
+ ids[name] = Number(r.lastInsertRowid);
+}
+
+t('boot', async () => {
+ h = await boot();
+ streamer = h.addUser('reads-streamer', { role: 'streamer' });
+ viewer = h.addUser('reads-viewer', { display_name: 'Reads Viewer' });
+ other = h.addUser('reads-other');
+ host = h.addUser('reads-host', { role: 'streamer' });
+ channelId = h.addChannel(streamer.id);
+ streamId = h.addStream(streamer.id, channelId);
+ otherStreamId = h.addStream(other.id, h.addChannel(other.id));
+ hostStreamId = h.addStream(host.id, h.addChannel(host.id));
+ await h.ctx.sync();
+ await h.ctx.ensureUsers([streamer.id, viewer.id, other.id, host.id]);
+
+ await line('v1', { ms: 0, user: viewer.id, username: viewer.username });
+ await line('gone', { ms: 10e3, user: viewer.id, username: viewer.username, deleted: 1 });
+ await line('v2', { ms: 30e3, user: viewer.id, username: viewer.username });
+ await line('o1', { ms: 65e3, user: other.id, username: other.username });
+ await line('v3', { ms: 70e3, user: viewer.id, username: viewer.username });
+ await line('a1', { ms: 130e3, anon: 'anon77', username: 'anon77' });
+ await line('r1', { ms: 135e3, username: 'twitch:fan', platform: 'twitch' });
+ await line('sb', { ms: 140e3, user: viewer.id, username: viewer.username, type: 'soundboard' });
+ await line('sys', { ms: 150e3, username: 'System', type: 'system' });
+ await line('elsewhere', { ms: 5e3, user: other.id, username: other.username, stream: otherStreamId, channel: other.id });
+});
+
+t('each route needs its own capability, a service token for Chat and loopback', async () => {
+ const routes = [
+ ['POST', '/stats', 'chat.stats.read', { kind: 'site' }],
+ ['GET', `/messages?stream_id=${streamId}`, 'chat.messages.read'],
+ ['GET', `/timeline?stream_id=${streamId}&since=${T}&until=${T + 600e3}&bucket_ms=60000`, 'chat.analysis.read'],
+ ['GET', `/moderation/pending-ip?channel_id=${channelId}`, 'chat.moderation.queue.read'],
+ ['GET', `/moderation/relay-users?channel_id=${channelId}`, 'chat.moderation.queue.read'],
+ ['GET', '/moderation/relay-users/1', 'chat.moderation.queue.read'],
+ ['GET', '/moderation/tts-override?identity_key=x', 'chat.moderation.queue.read'],
+ ['GET', `/sounds?channel_owner_id=${streamer.id}`, 'chat.sounds.read'],
+ ['GET', `/sounds/count?channel_owner_id=${streamer.id}`, 'chat.sounds.read'],
+ ['GET', '/sounds?pending_asset=1', 'chat.sounds.read'],
+ ['POST', '/sounds/asset', 'chat.sounds.write', { id: 999999, media_url: 'https://media.test/x', media_asset_id: 1 }, 404],
+ ];
+ for (const [method, path, cap, body, ok = 200] of routes) {
+ const url = `/internal/chat${path}`;
+ const wrong = cap === 'chat.messages.read' ? 'chat.stats.read' : 'chat.messages.read';
+ assert.strictEqual((await h.http(method, url, { body })).status, 401, `${path} without a token`);
+ assert.strictEqual((await h.http(method, url, { body, token: token(wrong) })).status, 403, `${path} with ${wrong}`);
+ assert.strictEqual((await h.http(method, url, { body, token: token('chat.moderation.read') })).status, 403, `${path} with chat.moderation.read`);
+ assert.strictEqual((await h.http(method, url, { body, token: h.serviceToken([cap], { aud: 'openvibe.live' }) })).status, 401, `${path} for another audience`);
+ assert.strictEqual((await h.http(method, url, { body, token: token(cap), headers: { 'X-Forwarded-For': '1.2.3.4' } })).status, 403, `${path} from outside`);
+ assert.strictEqual((await h.http(method, url, { body, token: token(cap) })).status, ok, `${path} with ${cap}`);
+ }
+ // Reading sounds does not allow writing their asset.
+ assert.strictEqual((await h.http('POST', '/internal/chat/sounds/asset', { body: {}, token: token('chat.sounds.read') })).status, 403);
+});
+
+t('N1 stats: site, user, stream and channel-top over non-deleted lines in the window', async () => {
+ const win = { since: T, until: T + 600e3 };
+ assert.deepStrictEqual(await stats({ kind: 'site', ...win }), { ok: true, messages: 9, chatters: 4 });
+ assert.ok((await stats({ kind: 'site' })).messages >= 9);
+ assert.deepStrictEqual(await stats({ kind: 'user', user_id: viewer.id, ...win }), { ok: true, messages: 4, chatters: 1 });
+ assert.deepStrictEqual(await stats({ kind: 'stream', stream_id: streamId, ...win }), { ok: true, messages: 8, chatters: 4, sounds: 1 });
+ assert.deepStrictEqual(await stats({ kind: 'stream', stream_id: streamId, since: T + 60e3, until: T + 120e3 }), { ok: true, messages: 2, chatters: 2, sounds: 0 });
+
+ const top = (await stats({ kind: 'channel-top', stream_id: streamId, ...win })).top_chatters;
+ assert.deepStrictEqual(top[0], { user_id: viewer.id, username: viewer.username, display_name: 'Reads Viewer', avatar_url: null, profile_color: '#8b5cf6', count: 4 });
+ assert.deepStrictEqual(top.slice(1).map((c) => [c.user_id, c.username, c.count]).sort(),
+ [[null, 'anon77', 1], [null, 'twitch:fan', 1], [other.id, other.username, 1]].sort());
+ assert.deepStrictEqual((await stats({ kind: 'channel-top', channel_user_id: streamer.id, limit: 2, ...win })).top_chatters.map((c) => c.count), [4, 1]);
+ const all = (await stats({ kind: 'channel-top', ...win })).top_chatters;
+ assert.deepStrictEqual(all.slice(0, 2).map((c) => [c.user_id, c.count]), [[viewer.id, 4], [other.id, 2]]);
+
+ for (const body of [{}, { kind: 'everything' }, { kind: 'user' }, { kind: 'stream' }, { kind: 'site', extra: 1 },
+ { kind: 'site', since: -1 }, { kind: 'site', since: '1' }, { kind: 'stream', stream_id: 1.5 }, { kind: 'channel-top', limit: 51 },
+ { kind: 'channel-top', user_id: viewer.id }]) await stats(body, 400);
+});
+
+t('N2 messages: one filter, a cursor on id, tail → max_id', async () => {
+ const live = ['v1', 'v2', 'o1', 'v3', 'a1', 'r1', 'sb', 'sys'].map((k) => ids[k]);
+ const page = (q, status) => read(`/messages?${q}`, 'chat.messages.read', 'chat.messages-page@1', status);
+ const newest = await page(`stream_id=${streamId}`);
+ assert.deepStrictEqual(newest.messages.map((m) => m.id), [...live].reverse());
+ assert.strictEqual(newest.max_id, ids.sys);
+ assert.deepStrictEqual(newest.messages.find((m) => m.id === ids.r1), {
+ id: ids.r1, user_id: null, anon_id: null, username: 'twitch:fan', message: 'line r1', message_type: 'chat', is_global: 0,
+ stream_id: streamId, channel_user_id: streamer.id, source_platform: 'twitch', reply_to_id: null, timestamp: at(135e3),
+ });
+ assert.deepStrictEqual((await page(`stream_id=${streamId}&after_id=${ids.v1}&limit=2`)).messages.map((m) => m.id), [ids.v2, ids.o1]);
+ assert.deepStrictEqual((await page(`channel_user_id=${streamer.id}&before_id=${ids.o1}`)).messages.map((m) => m.id), [ids.v2, ids.v1]);
+ assert.deepStrictEqual((await page(`user_id=${viewer.id}&after_id=${ids.v1}&before_id=${ids.sb}`)).messages.map((m) => m.id), [ids.v2, ids.v3]);
+ assert.deepStrictEqual(await page(`channel_user_id=${streamer.id}&tail=1`), { ok: true, messages: [], max_id: ids.sys });
+ assert.deepStrictEqual((await page(`stream_id=${streamId}&types=chat`)).messages.map((m) => m.id), live.slice(0, 6).reverse());
+ assert.deepStrictEqual((await page('anon_id=anon77')).messages.map((m) => m.id), [ids.a1]);
+ assert.deepStrictEqual((await page('username=twitch%3Afan')).messages.map((m) => m.id), [ids.r1]);
+ assert.deepStrictEqual((await page(`id=${ids.o1}`)).messages.map((m) => m.message), ['line o1']);
+ assert.deepStrictEqual(await page(`id=${ids.gone}`), { ok: true, messages: [], max_id: null });
+ assert.strictEqual((await page(`stream_id=${streamId}&limit=100000`)).messages.length, live.length);
+ for (const q of ['', `stream_id=${streamId}&user_id=${viewer.id}`, 'stream_id=abc', 'stream_id=0', `stream_id=${streamId}&limit=-1`,
+ `stream_id=${streamId}&tail=yes`, `stream_id=${streamId}&types=chat,nope`, `stream_id=${streamId}&order=asc`, `stream_id=${streamId}&stream_id=1`])
+ check('chat.messages-page@1', await page(q, 400));
+});
+
+t('N3 timeline: buckets from since, oldest first, empty ones left out', async () => {
+ const tl = (q, status) => read(`/timeline?${q}`, 'chat.analysis.read', 'chat.timeline-result@1', status);
+ assert.deepStrictEqual(await tl(`stream_id=${streamId}&since=${T}&until=${T + 600e3}&bucket_ms=60000`), {
+ ok: true, buckets: [{ t: T, count: 2 }, { t: T + 60e3, count: 2 }, { t: T + 120e3, count: 4 }], max_id: ids.sys,
+ });
+ assert.deepStrictEqual(await tl(`channel_user_id=${streamer.id}&since=${T + 30e3}&until=${T + 90e3}&bucket_ms=30000`), {
+ ok: true, buckets: [{ t: T + 30e3, count: 1 }, { t: T + 60e3, count: 2 }], max_id: ids.v3,
+ });
+ assert.deepStrictEqual(await tl(`stream_id=${streamId}&since=${T + 900e3}&until=${T + 960e3}&bucket_ms=60000`), { ok: true, buckets: [], max_id: null });
+ for (const q of [`since=${T}&bucket_ms=60000`, `stream_id=${streamId}&channel_user_id=${streamer.id}&since=${T}&bucket_ms=60000`,
+ `stream_id=${streamId}&bucket_ms=60000`, `stream_id=${streamId}&since=${T}`, `stream_id=${streamId}&since=${T}&until=${T + 600e3}&bucket_ms=500`,
+ `stream_id=${streamId}&since=${T}&until=${T}&bucket_ms=60000`, `stream_id=${streamId}&since=0&until=${T}&bucket_ms=1000`])
+ check('chat.timeline-result@1', await tl(q, 400));
+});
+
+t('N4 moderation queues: pending IP messages, hidden relay users, TTS overrides', async () => {
+ const run = (sql, p) => h.db.run(sql, p);
+ const pend = async (channel, status, ms) => Number((await run(
+ `INSERT INTO pending_ip_messages (channel_id, stream_id, ip_address, anon_id, username, message, status, created_at) VALUES (?, ?, '203.0.113.9', 'anon5', 'anon5', 'held', ?, ?)`,
+ [channel, streamId, status, at(ms)])).lastInsertRowid);
+ const later = await pend(channelId, 'pending', 20e3);
+ const first = await pend(channelId, 'pending', 10e3);
+ await pend(channelId, 'approved', 5e3);
+ await pend(channelId + 100, 'pending', 1e3);
+ const q = (path, status) => read(`/moderation/${path}`, 'chat.moderation.queue.read', 'chat.moderation-queue-result@1', status);
+ const pending = (await q(`pending-ip?channel_id=${channelId}`)).pending_ip;
+ assert.deepStrictEqual(pending.map((r) => r.id), [first, later]);
+ assert.strictEqual(pending[0].ip_address, '203.0.113.9');
+ assert.strictEqual((await q(`pending-ip?channel_id=${channelId}&limit=1`)).pending_ip.length, 1);
+
+ const hide = async (channel, ms) => Number((await run(
+ `INSERT INTO hidden_relay_users (channel_id, platform, external_username, action, reason, created_by, created_at) VALUES (?, 'twitch', 'spammer', 'hide', 'spam', ?, ?)`,
+ [channel, streamer.id, at(ms)])).lastInsertRowid);
+ const older = await hide(channelId, 1e3);
+ const everywhere = await hide(null, 2e3);
+ await hide(channelId + 100, 3e3);
+ const relay = (await q(`relay-users?channel_id=${channelId}`)).relay_users;
+ assert.deepStrictEqual(relay.map((r) => [r.id, r.channel_id]), [[everywhere, null], [older, channelId]]);
+ assert.strictEqual(relay[0].created_by_username, streamer.username);
+ assert.deepStrictEqual((await q(`relay-users/${older}`)).relay_user, { ...relay[1] });
+ assert.deepStrictEqual(await q('relay-users/999999'), { ok: true, relay_user: null });
+
+ await run(`INSERT INTO tts_voice_overrides (identity_key, voice, pitch, speed, gap, set_by) VALUES ('user:42', 'brian', 2, -1, 3, ?)`, [streamer.id]);
+ const override = (await q('tts-override?identity_key=%20USER:42%20')).tts_override;
+ assert.deepStrictEqual({ ...override, updated_at: null }, { identity_key: 'user:42', voice: 'brian', pitch: 2, speed: -1, gap: 3, set_by: streamer.id, updated_at: null });
+ assert.deepStrictEqual(await q('tts-override?identity_key=user:43'), { ok: true, tts_override: null });
+
+ for (const path of ['pending-ip', 'pending-ip?channel_id=x', 'relay-users', 'relay-users/abc', 'relay-users/0', 'tts-override', 'tts-override?identity_key=%20', `pending-ip?channel_id=${channelId}&status=denied`])
+ check('chat.moderation-queue-result@1', await q(path, 400));
+});
+
+t('N5 sounds: count, pending assets, and an idempotent asset write-back', async () => {
+ const sound = async (owner, command, asset = null) => Number((await h.db.run(
+ `INSERT INTO channel_sounds (channel_owner_id, command, url, duration_seconds, created_by, created_by_name, media_url, media_asset_id) VALUES (?, ?, ?, 1.5, ?, 'uploader', ?, ?)`,
+ [owner, command, `/sounds/${command}.mp3`, owner, asset && `https://media.test/a/${asset}`, asset])).lastInsertRowid);
+ const honk = await sound(streamer.id, 'honk');
+ await sound(streamer.id, 'beep', 7);
+ const boom = await sound(other.id, 'boom');
+ const s = (path, status) => read(`/sounds${path}`, 'chat.sounds.read', 'chat.sounds-result@1', status);
+ assert.deepStrictEqual(await s(`?channel_owner_id=${streamer.id}`), { ok: true, count: 2 });
+ assert.deepStrictEqual(await s(`/count?channel_owner_id=${other.id}`), { ok: true, count: 1 });
+ assert.deepStrictEqual(await s(`?channel_owner_id=${host.id}`), { ok: true, count: 0 });
+ const pending = (await s('?pending_asset=1')).sounds;
+ assert.deepStrictEqual(pending.map((r) => r.id), [honk, boom]);
+ assert.deepStrictEqual(pending[0], {
+ id: honk, channel_owner_id: streamer.id, command: 'honk', url: '/sounds/honk.mp3', mime: 'audio/mpeg', duration_seconds: 1.5,
+ created_by: streamer.id, created_by_name: 'uploader', media_url: null, media_asset_id: null,
+ });
+ assert.deepStrictEqual((await s(`?pending_asset=1&channel_owner_id=${other.id}`)).sounds.map((r) => r.id), [boom]);
+ assert.deepStrictEqual((await s(`?pending_asset=1&after_id=${honk}&limit=1`)).sounds.map((r) => r.id), [boom]);
+
+ const write = (body) => h.http('POST', '/internal/chat/sounds/asset', { body, token: token('chat.sounds.write') });
+ const body = check('chat.sound-asset-request@1', { id: honk, media_url: 'https://media.test/a/11', media_asset_id: 11 });
+ for (let i = 0; i < 2; i++) {
+ const r = await write(body);
+ assert.strictEqual(r.status, 200, r.text);
+ assert.deepStrictEqual(check('chat.ingress-ack@1', r.body), { ok: true });
+ }
+ assert.deepStrictEqual((await s('?pending_asset=1')).sounds.map((r) => r.id), [boom]);
+ const row = await h.db.get('SELECT media_url, media_asset_id FROM channel_sounds WHERE id = ?', [honk]);
+ assert.deepStrictEqual([row.media_url, Number(row.media_asset_id)], ['https://media.test/a/11', 11]);
+ assert.strictEqual((await write({ ...body, id: 999999 })).status, 404);
+ for (const bad of [{}, { ...body, id: 0 }, { ...body, media_url: '' }, { ...body, media_asset_id: '11' }, { ...body, extra: 1 }, { ...body, media_url: 'x'.repeat(2049) }]) {
+ const r = await write(bad);
+ assert.strictEqual(r.status, 400, JSON.stringify(bad));
+ check('chat.ingress-ack@1', r.body);
+ }
+ for (const path of ['', '?channel_owner_id=abc', '?pending_asset=2', `?channel_owner_id=${streamer.id}&limit=5`]) await s(path, 400);
+});
+
+t('the message ingress answers first_chat: true once per chatter and channel', async () => {
+ const send = async (key, extra) => {
+ const r = await h.http('POST', '/internal/chat/messages', { body: { key, stream_id: hostStreamId, message: 'hello host', ...extra }, token: token('chat.message.send') });
+ assert.strictEqual(r.status, 200, r.text);
+ // first_chat is in chat.send-result from 1.2.0 (openvibe-contracts 0.95.0).
+ const { first_chat, ...older } = r.body;
+ check('chat.send-result@1', sendResultHasFirstChat ? r.body : older);
+ return r.body;
+ };
+ const viewerLine = { user_id: viewer.id, username: viewer.username };
+ assert.strictEqual((await send('reads:first:1', viewerLine)).first_chat, true);
+ assert.strictEqual((await send('reads:first:2', viewerLine)).first_chat, false);
+ assert.strictEqual((await send('reads:first:1', viewerLine)).first_chat, true, 'a retry answers what was saved');
+ assert.strictEqual((await send('reads:first:3', { anon_id: 'anon88', username: 'anon88' })).first_chat, true);
+ assert.strictEqual((await send('reads:first:4', { ...viewerLine, message_type: 'system' })).first_chat, false, 'only chat lines use up a welcome');
+ const r = await h.http('POST', '/internal/chat/messages', { body: { key: 'reads:first:dm', user_id: viewer.id, username: viewer.username, message: 'dm', dm: { to_user_id: other.id } }, token: token('chat.message.send') });
+ assert.strictEqual(r.status, 200, r.text);
+ assert.ok(!('first_chat' in check('chat.send-result@1', r.body)));
+});
+
+t.run(async () => {
+ if (unknownContracts.size) {
+ console.log(`chat-internal-reads: schemas not validated (openvibe-contracts ${require('openvibe-contracts/package.json').version} lacks ${[...unknownContracts].join(', ')}; pin v0.95.0)`);
+ }
+ if (h) await h.close();
+});
@@ -104,6 +104,19 @@ t('mirror: every change to Chat’s tables reaches Live, newest state per row',
@@ -104,6 +104,19 @@ t('mirror: every change to Chat’s tables reaches Live, newest state per row',
assert.deepStrictEqual(h.live.mirror.at(-1), { table: 'dm_messages', op: 'delete', pk: { id: dmId } });
assert.deepStrictEqual(h.live.mirror.at(-1), { table: 'dm_messages', op: 'delete', pk: { id: dmId } });
});
});
+t('mirror: a channel sound without a Media asset leaves Live’s asset columns alone', async () => {
+ const soundId = Number((await h.db.run(`INSERT INTO channel_sounds (channel_owner_id, command, url) VALUES (?, 'mirrorhonk', '/sounds/m.mp3')`, [streamer.id])).lastInsertRowid);
+ await h.mirrorRelay.flush();
+ const sent = h.live.mirror.filter((c) => c.table === 'channel_sounds' && c.row && c.row.id === soundId);
+ assert.strictEqual(sent.length, 1);
+ assert.ok(!('media_url' in sent[0].row) && !('media_asset_id' in sent[0].row), JSON.stringify(sent[0].row));
+ await h.db.run('UPDATE channel_sounds SET media_url = ?, media_asset_id = ? WHERE id = ?', ['https://media.test/s/9', 9, soundId]);
+ await h.mirrorRelay.flush();
+ const row = h.live.mirror.filter((c) => c.table === 'channel_sounds' && c.row && c.row.id === soundId).at(-1).row;
+ assert.strictEqual(row.media_url, 'https://media.test/s/9');
+ assert.strictEqual(Number(row.media_asset_id), 9);
+});
+
t('mirror: Live refusing (not yet CHAT_AUTHORITY=chat) keeps the queue; accepted later', async () => {
t('mirror: Live refusing (not yet CHAT_AUTHORITY=chat) keeps the queue; accepted later', async () => {
h.live.mirrorStatus = 409;
h.live.mirrorStatus = 409;
await h.db.recordRelayUser('twitch', 'Zed');
await h.db.recordRelayUser('twitch', 'Zed');
@@ -42,8 +42,11 @@ t('the seed applies after 0001–0002 and before 0003', async () => {
@@ -42,8 +42,11 @@ t('the seed applies after 0001–0002 and before 0003', async () => {
}
}
assert.strictEqual((await db.prepare(`SELECT to_regclass('chat_ingress_applied') AS t`).get()).t, null, '0003 has not run yet');
assert.strictEqual((await db.prepare(`SELECT to_regclass('chat_ingress_applied') AS t`).get()).t, null, '0003 has not run yet');
const second = await db.migrate({ dir: MIGRATIONS, log: { log() {} } });
const second = await db.migrate({ dir: MIGRATIONS, log: { log() {} } });
- assert.strictEqual(second.held.length, 0, 'the branch migration is not held');
- assert.deepStrictEqual(second.applied.map((a) => a.id), ['0003'], 'only the branch\'s migration applies after the seed');
+ const later = fs.readdirSync(MIGRATIONS).filter((f) => /^\d{4}_.*\.sql$/.test(f)).map((f) => f.slice(0, 4))
+ .filter((id) => id > '0002').sort();
+ assert.ok(later.includes('0003'), 'migrations/0003 exists');
+ assert.strictEqual(second.held.length, 0, 'no later migration is held');
+ assert.deepStrictEqual(second.applied.map((a) => a.id), later, 'every migration after 0002 applies on top of the seed');
});
});
t('cleanup', async () => { await db.close(); });
t('cleanup', async () => { await db.close(); });