openvibers/openvibe.chat Commits
Presence: Network's online status in DMs and member lists, DND pings, DM peers (#378)
Changes
@@@ -30,6 +30,7 @@ deleted messages are never read; `since`/`until` are epoch ms (`until` exclusive| `POST /internal/chat/stats` | `chat.stats.read` | `{kind:'site'\|'user'\|'stream'\|'channel-top'\|'site-daily', 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}]}`; `site-daily` (`since` and `until` required, epoch ms, at most 400 UTC days) → `{days:[{day,messages,chatters}]}`, one row per UTC day in `[since,until)` with zero-filled gaps: every non-deleted message across all channels (Live's home `messages` series) and distinct chatters (Live's `active`: `COALESCE(user_id,anon_id,source_platform‖username)`). || `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/dm/block-state?a&b` | `chat.messages.read` | `{blocked}`: true when either user blocked the other, in Chat's `dm_blocks` or as a Network platform block (`server/chat/dm.js` `isBlockedEither`, the check the DM and call routes use). `a` and `b` are positive Live user ids and must differ; Live asks this before sending a call invite. |++| `POST /internal/chat/dm/peers` | `chat.dm.peers.read` | `{subject, candidates}` (subjects, at most 100) → `chat.dm-peers-result@1` `{subject, peers}`: the candidates that share a direct or group conversation with `subject`. Subjects only, never a message or a name. Network's presence privacy rule uses it ("people they follow or have a DM with"). This one route answers the contract shape, without the wrapper `ok`. || `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/first-chat?channel_id&identity` | `chat.analysis.read` | `{first}`: true when `identity` (`user:<user_id>`, `anon:<anonId>`, `ext:<prefixed username>`; ≤ 160) has never chatted in the channel — Live's `isFirstChatInChannel`, from `stream_first_chats`. `channel_id` is the channel owner's Live user id (`stream_first_chats.channel_user_id`), not `ctx_channels.id`. A registered identity is `user:<user_id>`, the numeric Live user id and **never** the username: Live's own `user:${line.username}` key would read `first:true` forever, so its welcome check must pass `user:${line.user_id}`. || `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). |@@@ -17,7 +17,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.315.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.319.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.94.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.129.0","pg": "^8.23.0",@@@ -1557,9 +1557,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.315.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.315.0",-- "integrity": "sha512-m1bWT4/sF6Hc7lia6FjAR+880r9SBRhO/v819Lr5ldxlSkFYiKMXTtt7MDbJ4FCgQ7Dkt+2OJaKanlvk2vNPlg==",++ "version": "0.319.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.319.0",++ "integrity": "sha512-v0HSuq5d4XZ3tKfXRDgNcmO+M6pQVMsl/pWOhAGRfzSIdjV/G1s9mnsoi7WXhpsJiZ3kEYPP1ie2viyJ+Lg/Zg==","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.315.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.319.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.94.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.129.0","pg": "^8.23.0",@@@ -34,6 +34,7 @@ const soundboard = require('./soundboard-service');const audioQueue = require('./audio-queue');const dm = require('./dm');const networkBlocks = require('./network-blocks');++const presence = require('../presence');const vipBadges = require('../vip/badges');// Cosmetics and tags are Live's (monetization/cosmetics, game/tags); read through live-context.const cosmetics = { getCosmeticProfile: (userId) => ctx.getCosmeticProfile(userId) };@@@ -660,7 +661,7 @@ class ChatServer {break;}case 'get-users':-- this.sendTo(ws, { type: 'users-list', users: await this.getUserList(client.streamId) });++ this.sendTo(ws, { type: 'users-list', users: await this.getUserList(client.streamId, client.user && client.user.subject_id) });break;default:break;@@@ -731,9 +732,10 @@ class ChatServer {* Build a deduplicated list of users in a given stream (or global if null).* Returns { logged: [{username, display_name, avatar_url, role}], anonCount: N }*/-- async getUserList(streamId) {++ async getUserList(streamId, viewerSubject = null) {const seen = new Set();const logged = [];++ const subjects = new Map();let anonCount = 0, hiddenCount = 0;for (const [, c] of this.clients) {if (c.streamId !== streamId) continue;@@@ -742,12 +744,14 @@ class ChatServer {seen.add(c.user.id);// chat.presence_prefs: someone who turned off "show my name in user lists" is counted, not named.if (await this._hiddenFromUserList(c.user)) { hiddenCount++; continue; }++ const subject = c.user.subject_id || await db.subjectFor(c.user.id);logged.push({username: c.user.username,display_name: c.user.display_name || c.user.username,avatar_url: c.user.avatar_url || null,role: c.user.role,});++ if (subject) subjects.set(c.user.username, subject);} else if (c.anonId) {if (seen.has(c.anonId)) continue;seen.add(c.anonId);@@@ -757,6 +761,10 @@ class ChatServer {// Sort: admins first, then mods, then alphabeticalconst rolePriority = { admin: 0, global_mod: 1, streamer: 2, user: 3 };logged.sort((a, b) => (rolePriority[a.role] ?? 9) - (rolePriority[b.role] ?? 9) || a.display_name.localeCompare(b.display_name));++ // The Network answers each viewer separately (a stranger's state is never disclosed), so a member++ // list is built for the person who asked; the read is cached 30 s per viewer and member set.++ const seenPresence = await presence.forViewer(viewerSubject, [...subjects.values()]);++ for (const u of logged) u.presence = seenPresence.get(subjects.get(u.username)) || null;return { logged, anonCount, hiddenCount };}@@@ -2911,12 +2919,14 @@ class ChatServer {if (!this._usersListTimers) this._usersListTimers = new Map();this._usersListTimers.set(key, setTimeout(async () => {this._usersListTimers.delete(key);-- const users = await this.getUserList(streamId);-- const data = JSON.stringify({ type: 'users-list', users });++ // Presence is per viewer (the Network decides what this person may see), so the list is built++ // once per signed-in viewer subject and shared by that viewer's tabs.++ const byViewer = new Map();for (const [ws, client] of this.clients) {-- if (client.streamId === streamId && ws.readyState === WebSocket.OPEN) {-- try { ws.send(data); } catch {}-- }++ if (client.streamId !== streamId || ws.readyState !== WebSocket.OPEN) continue;++ const viewer = (client.user && client.user.subject_id) || null;++ if (!byViewer.has(viewer)) byViewer.set(viewer, await this.getUserList(streamId, viewer));++ try { ws.send(JSON.stringify({ type: 'users-list', users: byViewer.get(viewer) })); } catch {}}}, 500));}@@@ -20,6 +20,7 @@ const { limits } = require('../net/actor-limits');const dm = require('./dm');const ctx = require('../live-context');const prefStores = require('../prefs/stores');++const presence = require('../presence');const { idParam } = require('./route-params');/**@@@ -37,6 +38,12 @@ function publicMessage(m) {return m;}++/** Attach the Network presence view to every person in these conversations (one read, cached 30 s). */++async function withPresence(viewerSubject, conversations) {++ await presence.attachTo(viewerSubject, conversations.flatMap((c) => (Array.isArray(c && c.participants) ? c.participants : [])));++ return conversations;++}++// All DM routes require authenticationrouter.use(requireAuth);@@@ -91,6 +98,7 @@ router.get('/conversations', async (req, res) => {for (const conv of conversations) {conv.participants = await dm.getParticipants(conv.id);}++ await withPresence(req.user.subject_id, conversations);res.json({ conversations });} catch (err) {console.error('[DM] List conversations error:', err.message);@@@ -155,6 +163,7 @@ router.post('/conversations', startLimit, async (req, res) => {const conversation = await dm.getConversation(conversationId);conversation.participants = await dm.getParticipants(conversationId);++ await withPresence(req.user.subject_id, [conversation]);res.json({ conversation });} catch (err) {console.error('[DM] Create conversation error:', err.message);@@@ -172,6 +181,7 @@ router.get('/conversations/:id', async (req, res) => {const conversation = await dm.getConversation(convId);if (!conversation) return res.status(404).json({ error: 'Not found' });conversation.participants = await dm.getParticipants(convId);++ await withPresence(req.user.subject_id, [conversation]);res.json({ conversation });} catch (err) {console.error('[DM] Get conversation error:', err.message);@@@ -235,27 +245,28 @@ router.post('/conversations/:id/messages', sendLimit, async (req, res) => {msg.avatar_url = sender?.avatar_url;msg.profile_color = sender?.profile_color;-- // Real-time delivery via chat WebSocket++ // Real-time delivery via chat WebSocket. A recipient in DND (Network presence) gets the message++ // with `silent: true` and no Network push: Chat's own ping respects their chosen state.try {const chatServer = require('./chat-server');const participants = await dm.getParticipants(convId);-- for (const p of participants) {-- if (p.id === req.user.id) continue; // skip sender++ const others = participants.filter(p => p.id !== req.user.id);++ // One Network read per recipient, in parallel; cached 30 s each. A group is at most 20 people.++ const own = await Promise.all(others.map((p) => (p.subject_id ? presence.self(p.subject_id) : null)));++ const dnd = new Set(others.filter((p, i) => own[i] && own[i].state === 'dnd').map((p) => p.id));++ for (const p of others) {await chatServer.sendDm(p.id, {type: 'dm',conversation_id: convId,message: msg,++ ...(dnd.has(p.id) ? { silent: true } : {}),});}-- } catch { /* non-critical */ }---- // Push notifications to other participants (for when they're offline) — Live's-- // utils/notify pushes them to the Network; the preview stays generic there.-- try {-- const participants = await dm.getParticipants(convId);-- const otherIds = participants.filter(p => p.id !== req.user.id).map(p => p.id);-- if (otherIds.length) {-- ctx.effects.notifyDm({ recipient_ids: otherIds, sender_id: req.user.id, conversation_id: convId });++ // Push notifications to other participants (for when they're offline) — Live's++ // utils/notify pushes them to the Network; the preview stays generic there.++ const notifyIds = others.filter(p => !dnd.has(p.id)).map(p => p.id);++ if (notifyIds.length) {++ ctx.effects.notifyDm({ recipient_ids: notifyIds, sender_id: req.user.id, conversation_id: convId });}} catch { /* non-critical */ }@@@ -142,7 +142,7 @@ async function getConversations(userId) {*/async function getParticipants(conversationId) {return await db.all(`-- SELECT u.id, u.username, u.display_name, u.avatar_url, u.profile_color++ SELECT u.id, COALESCE(p.subject_id, u.subject_id) AS subject_id, u.username, u.display_name, u.avatar_url, u.profile_colorFROM dm_participants pJOIN ctx_users u ON u.id = p.user_idWHERE p.conversation_id = ?@@@ -7,6 +7,8 @@* kind site-daily → chat.site-daily-result@1 { days }* GET /internal/chat/messages chat.messages.read → chat.messages-page@1* GET /internal/chat/dm/block-state chat.messages.read → { blocked }++ * POST /internal/chat/dm/peers chat.dm.peers.read chat.dm-peers-result@1++ * { subject, candidates } → { subject, peers }* GET /internal/chat/timeline chat.analysis.read → chat.timeline-result@1* GET /internal/chat/first-chat chat.analysis.read → chat.first-chat-result@1 { first }* GET /internal/chat/moderation/pending-ip chat.moderation.queue.read → chat.moderation-queue-result@1 { pending_ip }@@@ -209,6 +211,24 @@ async function dmBlockState(req) {return { blocked: await dm.isBlockedEither(q.a, q.b) };}++/** Network's presence privacy rule: which candidates share a conversation with `subject`. */++async function dmPeers(req) {++ const b = req.body;++ if (!b || typeof b !== 'object' || Array.isArray(b) || Object.keys(b).some((k) => !['subject', 'candidates'].includes(k))) bad('Invalid fields');++ if (typeof b.subject !== 'string' || !/^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(b.subject)) bad('Invalid subject');++ if (!Array.isArray(b.candidates) || b.candidates.length > 100) bad('candidates must be an array of at most 100 subjects');++ const candidates = [...new Set(b.candidates)];++ if (!candidates.every((c) => typeof c === 'string' && /^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(c))) bad('Invalid candidates');++ if (candidates.includes(b.subject)) bad('a person is not their own peer');++ if (!candidates.length) return { subject: b.subject, peers: [] };++ // A shared conversation is a DM relationship, direct or group (a group is at most 20 people).++ const rows = await db.all(`SELECT DISTINCT p2.subject_id AS peer++ FROM dm_participants p1++ JOIN dm_participants p2 ON p2.conversation_id = p1.conversation_id++ WHERE p1.subject_id = ? AND p2.subject_id <> ? AND p2.subject_id = ANY(?)`, [b.subject, b.subject, candidates]);++ return { subject: b.subject, peers: rows.map((r) => r.peer).filter(Boolean) };++}++// ── N3: timeline ──// People's chat only: Live's AI viewers (source_platform 'ai') are left out, so their chatter never makes a "spike" the// AI then clips (item 151, C07: AI clips are excluded from human signals).@@@ -373,6 +393,14 @@ 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('/dm/block-state', serviceAuth.guard('chat.messages.read'), route('dm-block-state', dmBlockState));++router.post('/dm/peers', serviceAuth.guard('chat.dm.peers.read'), async (req, res) => {++ try { res.json(await dmPeers(req)); }++ catch (err) {++ if (err.status) return res.status(err.status).json({ error: err.message });++ console.error('[ChatReads] dm-peers:', err);++ res.status(503).json({ error: 'Chat read unavailable' });++ }++});router.get('/timeline', serviceAuth.guard('chat.analysis.read'), route('timeline', timeline));router.get('/first-chat', serviceAuth.guard('chat.analysis.read'), route('first-chat', firstChat));const queue = serviceAuth.guard('chat.moderation.queue.read');nnew file mode 100644@@@ -0,0 +1,68 @@++'use strict';++/**++ * OpenVibe.Network presence for Chat (plan 2026-10-22 track X item 9; Contracts 0.318.0++ * network.presence@1).++ *++ * Network owns the person's chosen online status. Chat reads it through Network's internal API as a++ * first-party service (GET /internal/presence?for=usr_…&subjects=…, network.presence.read) and caches++ * the answer for 30 seconds. The answer already applies the privacy rule for `for`: themselves, people++ * they follow and people they share a direct conversation with, never a stranger, and never an++ * invisible person. Chat never widens it; an absent subject is simply drawn as offline.++ *++ * DND is read the same way for the person it belongs to (for = themselves): a DM message to a person++ * in DND arrives with `silent: true` and is left out of the Network push, so Chat does not ping them.++ *++ * When Network does not answer, nobody is shown a state (the UI draws offline) and no ping is muted:++ * failing closed on privacy, and never blocking a message on presence.++ */++const config = require('./config');++const auth = require('./net/service-auth');++++const TTL_MS = 30_000;++const FAIL_TTL_MS = 5_000;++const MAX_SUBJECTS = 100;++const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++++const cache = new Map();++++/** Network presence for `subjects` as `viewer` may see it → Map(subject → presence view). */++async function forViewer(viewer, subjects) {++ const list = [...new Set((subjects || []).filter((s) => SUBJECT_RE.test(String(s))))].slice(0, MAX_SUBJECTS);++ if (!viewer || !SUBJECT_RE.test(String(viewer)) || !list.length) return new Map();++ const key = `${viewer}|${list.sort().join(',')}`;++ const hit = cache.get(key);++ if (hit && Date.now() - hit.at < hit.ttl) return hit.value;++ let value = new Map();++ let ttl = FAIL_TTL_MS;++ try {++ const headers = await auth.headers('openvibe.network', ['network.presence.read']);++ const url = `${config.networkInternalUrl}/internal/presence?for=${encodeURIComponent(viewer)}&subjects=${list.map(encodeURIComponent).join(',')}`;++ const res = await fetch(url, { headers: { Accept: 'application/json', ...headers }, signal: AbortSignal.timeout(1500) });++ if (res.ok) {++ const body = await res.json().catch(() => ({}));++ for (const p of Array.isArray(body.presence) ? body.presence : []) if (SUBJECT_RE.test(String(p.subject))) value.set(p.subject, p);++ ttl = TTL_MS;++ }++ } catch { /* Network down: no state, no muted pings */ }++ if (cache.size > 1000) cache.clear();++ cache.set(key, { at: Date.now(), value, ttl });++ return value;++}++++/** Attach `presence` (the Network view or null) to each person in `people`, keyed by subject_id. */++async function attachTo(viewer, people) {++ const list = Array.isArray(people) ? people : [];++ const subjects = list.map((p) => p && p.subject_id).filter(Boolean);++ const seen = await forViewer(viewer, subjects);++ for (const p of list) if (p) p.presence = (p.subject_id && seen.get(p.subject_id)) || null;++ return list;++}++++/** The person's own presence (for DND). Null when Network cannot answer. */++async function self(subject) {++ return (await forViewer(subject, [subject])).get(subject) || null;++}++++function clearCache() { cache.clear(); }++++module.exports = { forViewer, attachTo, self, clearCache, TTL_MS, SUBJECT_RE, MAX_SUBJECTS };@@@ -96,9 +96,23 @@ async function boot({ env = {} } = {}) {netModules.set(ns, subject, data, `svc:${String(v.claims.sub).replace(/^svc:/, '')}`);return send(cur ? 200 : 201, moduleRecord(ns, subject, netModules.records.get(k)));}++ // Presence for the Network endpoint Chat reads (network.presence.read). Tests set rows here; the++ // visibility rule itself is Network's and is tested there, so the stub answers the subjects asked for.++ const netPresence = { rows: new Map(), calls: [] };const network = http.createServer(async (req, res) => {const raw = await new Promise((r) => { let s = ''; req.on('data', (c) => { s += c; }); req.on('end', () => r(s)); });res.setHeader('Content-Type', 'application/json');++ if (req.url.startsWith('/internal/presence')) {++ const url = new URL(req.url, ISS);++ const auth = String(req.headers.authorization || '');++ const v = auth.startsWith('Bearer ') ? serviceAuth.verifyServiceToken(auth.slice(7), { publicKey: keys.publicKey, issuer: ISS, audience: 'openvibe.network' }) : { ok: false, code: 'token.missing' };++ if (!v.ok) { res.statusCode = 401; return res.end(JSON.stringify({ error: v.code })); }++ if (!(v.claims.cap || []).includes('network.presence.read')) { res.statusCode = 403; return res.end(JSON.stringify({ error: 'capability.denied' })); }++ const forSubject = url.searchParams.get('for');++ const subjects = String(url.searchParams.get('subjects') || '').split(',').filter(Boolean);++ netPresence.calls.push({ for: forSubject, subjects });++ return res.end(JSON.stringify({ presence: subjects.map((s) => netPresence.rows.get(s)).filter(Boolean) }));++ }const mm = /^\/internal\/modules\/([^/?]+)\/([^/?]+)$/.exec(req.url);if (mm) return modulesRoute(req, res, raw, decodeURIComponent(mm[1]), decodeURIComponent(mm[2]));if (req.url === '/oauth/token') {@@@ -298,7 +312,7 @@ async function boot({ env = {} } = {}) {});const h = {-- tmp, keys, live, network, liveServer, tokenRequests, serviceToken, ISS, netModules,++ tmp, keys, live, network, liveServer, tokenRequests, serviceToken, ISS, netModules, netPresence,userSeq: 0,/** Add a Live user (projection row) with a browser token; returns { id, token, ...row }. */addUser(username, { role = 'user', subject = null, created_at = sqliteNow(-30 * 86400e3), apiScopes = null, is_owner = 0, display_name = null } = {}) {nnew file mode 100644@@@ -0,0 +1,117 @@++'use strict';++/**++ * Chat's Network presence (plan 2026-10-22 track X item 9; Contracts 0.318.0):++ * - server/presence.js reads GET /internal/presence for one viewer, caches it 30 s, and fails closed;++ * - DM lists and DM headers carry `presence` per participant;++ * - room member lists carry `presence` per person, built separately for each viewer;++ * - DND mutes Chat's own DM ping (`silent: true`) and the Network push;++ * - POST /internal/chat/dm/peers answers the shared-conversation set for Network++ * (chat.dm.peers.read, chat.dm-peers-result@1).++ */++const assert = require('assert');++const { validate } = require('openvibe-contracts');++const { boot, suite } = require('./helpers');++++const t = suite('presence');++const ANN = 'usr_01J9ANN0000000000000000AAA';++const BOB = 'usr_01J9B0B0000000000000000AAA';++const CAT = 'usr_01J9CAT0000000000000000AAA';++let h, ann, bob, cat, conv;++++const view = (subject, state, extra = {}) => ({++ subject, state, status_text: extra.status_text || null, emoji: extra.emoji || null,++ live: !!extra.live, updated_at: '2026-11-07T09:00:00.000Z',++});++++t('boot with a stub Network that knows ann, bob and cat', async () => {++ h = await boot();++ ann = h.addUser('ann', { subject: ANN });++ bob = h.addUser('bob', { subject: BOB });++ cat = h.addUser('cat', { subject: CAT });++ h.netModules.subjects.add(ANN).add(BOB).add(CAT);++ h.netPresence.rows.set(BOB, view(BOB, 'online', { status_text: 'Around' }));++ h.netPresence.rows.set(CAT, view(CAT, 'dnd'));++});++++t('a DM list and a DM header carry the presence Network returned for this viewer', async () => {++ let r = await h.http('POST', '/api/dm/conversations', { token: ann.token, body: { user_ids: [bob.id] } });++ assert.strictEqual(r.status, 200, r.text);++ conv = r.body.conversation.id;++ assert.deepStrictEqual(r.body.conversation.participants.find((p) => p.id === bob.id).presence.state, 'online');++ assert.strictEqual(h.netPresence.calls.at(-1).for, ANN, 'read as the person who asked');++++ r = await h.http('GET', '/api/dm/conversations', { token: ann.token });++ assert.strictEqual(r.status, 200, r.text);++ const listed = r.body.conversations.find((c) => c.id === conv);++ assert.deepStrictEqual(listed.participants.find((p) => p.id === bob.id).presence, view(BOB, 'online', { status_text: 'Around' }));++++ r = await h.http('GET', `/api/dm/conversations/${conv}`, { token: bob.token });++ assert.strictEqual(r.status, 200, r.text);++ assert.strictEqual(r.body.conversation.participants.find((p) => p.id === ann.id).presence, null, 'no row is a null presence, not an error');++++ // The read is cached 30 s: the same viewer and member set does not ask Network again.++ const calls = h.netPresence.calls.length;++ await h.http('GET', '/api/dm/conversations', { token: ann.token });++ assert.strictEqual(h.netPresence.calls.length, calls, 'cached');++});++++t('DND mutes the DM ping and the Network push, and only DND does', async () => {++ h.netPresence.rows.set(BOB, view(BOB, 'dnd'));++ require('../server/presence').clearCache();++ const sent = [];++ const orig = h.chatServer.sendDm;++ h.chatServer.sendDm = async (id, payload) => { sent.push({ id, payload }); };++ try {++ let r = await h.http('POST', `/api/dm/conversations/${conv}/messages`, { token: ann.token, body: { message: 'heads down?' } });++ assert.strictEqual(r.status, 200, r.text);++ assert.strictEqual(sent.length, 1);++ assert.strictEqual(sent[0].id, bob.id);++ assert.strictEqual(sent[0].payload.silent, true, 'Bob is in DND');++ assert.ok(!h.live.effects.some((e) => e.name === 'notify/dm'), 'no Network push for the DND recipient');++++ h.netPresence.rows.set(BOB, view(BOB, 'online'));++ require('../server/presence').clearCache();++ r = await h.http('POST', `/api/dm/conversations/${conv}/messages`, { token: ann.token, body: { message: 'lunch?' } });++ assert.strictEqual(r.status, 200, r.text);++ assert.strictEqual(sent[1].payload.silent, undefined, 'online gets the normal ping');++ assert.ok(h.live.effects.some((e) => e.name === 'notify/dm'), 'and the Network push');++ } finally { h.chatServer.sendDm = orig; }++});++++t('room member lists carry presence, built for the viewer', async () => {++ const streamer = h.addUser('streamer', { role: 'streamer' });++ const channelId = h.addChannel(streamer.id);++ const streamId = h.addStream(streamer.id, channelId);++ await h.ctx.sync();++ h.netPresence.rows.set(BOB, view(BOB, 'away', { live: true }));++ require('../server/presence').clearCache();++ const a = await h.ws({ ip: '198.51.100.41', token: ann.token, stream: streamId });++ const b = await h.ws({ ip: '198.51.100.42', token: bob.token, stream: streamId });++ try {++ a.sendJson({ type: 'join', streamId, token: ann.token });++ b.sendJson({ type: 'join', streamId, token: bob.token });++ await a.next((m) => m.type === 'auth' && m.user_id === ann.id);++ await b.next((m) => m.type === 'auth' && m.user_id === bob.id);++ a.sendJson({ type: 'get-users' });++ const list = await a.next((m) => m.type === 'users-list' && m.users.logged.some((u) => u.username === 'bob'));++ const listed = list.users.logged.find((u) => u.username === 'bob');++ assert.strictEqual(listed.presence.state, 'away');++ assert.strictEqual(listed.presence.live, true);++ assert.strictEqual(h.netPresence.calls.at(-1).for, ANN, 'the list was built for Ann');++ } finally { a.close(); b.close(); }++});++++t('the DM peer read answers shared conversations only, with subjects only', async () => {++ const token = h.serviceToken(['chat.dm.peers.read'], { aud: 'openvibe.chat' });++ let r = await h.http('POST', '/internal/chat/dm/peers', { token, body: { subject: ANN, candidates: [BOB, CAT] } });++ assert.strictEqual(r.status, 200, r.text);++ assert.ok(validate('chat.dm-peers-result@1', r.body).valid, JSON.stringify(validate('chat.dm-peers-result@1', r.body).errors));++ assert.deepStrictEqual([r.body.subject, r.body.peers], [ANN, [BOB]]);++ assert.ok(!r.text.includes('messages') && !r.text.includes('Around'), 'no message, status or name');++ r = await h.http('POST', '/internal/chat/dm/peers', { token: h.serviceToken(['chat.messages.read'], { aud: 'openvibe.chat' }), body: { subject: ANN, candidates: [BOB] } });++ assert.strictEqual(r.status, 403, 'another capability is refused');++ r = await h.http('POST', '/internal/chat/dm/peers', { token, body: { subject: ANN, candidates: ['bob'] } });++ assert.strictEqual(r.status, 400);++ r = await h.http('POST', '/internal/chat/dm/peers', { token, body: { subject: ANN, candidates: [BOB], extra: 1 } });++ assert.strictEqual(r.status, 400);++});@@@ -116,6 +116,7 @@ t('ratchet: every file that makes an outbound request itself is reviewed', () =>'server/net/service-auth.js': 'Network JWKS (configured)','server/chat/account-data.js': 'Network internal API (configured), fixed paths: export parts and deletion confirmations','server/prefs/network-modules.js': 'Network user modules (configured)',++ 'server/presence.js': 'Network internal presence (configured internal URL), fixed /internal/presence path, the subject list in the query, a service token, 1.5 s, held 30 s','server/chat/routes.js': 'GIF providers (fixed hosts)','server/chat/emotes-routes.js': 'FFZ / BTTV / 7TV emote providers (fixed hosts, hardcoded URLs)','server/chat/soundboard-service.js': '101soundboards (host allowlist; audio via safeLookup)',
@@ -30,6 +30,7 @@ deleted messages are never read; `since`/`until` are epoch ms (`until` exclusive
@@ -30,6 +30,7 @@ deleted messages are never read; `since`/`until` are epoch ms (`until` exclusive
| `POST /internal/chat/stats` | `chat.stats.read` | `{kind:'site'\|'user'\|'stream'\|'channel-top'\|'site-daily', 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}]}`; `site-daily` (`since` and `until` required, epoch ms, at most 400 UTC days) → `{days:[{day,messages,chatters}]}`, one row per UTC day in `[since,until)` with zero-filled gaps: every non-deleted message across all channels (Live's home `messages` series) and distinct chatters (Live's `active`: `COALESCE(user_id,anon_id,source_platform‖username)`). |
| `POST /internal/chat/stats` | `chat.stats.read` | `{kind:'site'\|'user'\|'stream'\|'channel-top'\|'site-daily', 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}]}`; `site-daily` (`since` and `until` required, epoch ms, at most 400 UTC days) → `{days:[{day,messages,chatters}]}`, one row per UTC day in `[since,until)` with zero-filled gaps: every non-deleted message across all channels (Live's home `messages` series) and distinct chatters (Live's `active`: `COALESCE(user_id,anon_id,source_platform‖username)`). |
| `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/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/dm/block-state?a&b` | `chat.messages.read` | `{blocked}`: true when either user blocked the other, in Chat's `dm_blocks` or as a Network platform block (`server/chat/dm.js` `isBlockedEither`, the check the DM and call routes use). `a` and `b` are positive Live user ids and must differ; Live asks this before sending a call invite. |
| `GET /internal/chat/dm/block-state?a&b` | `chat.messages.read` | `{blocked}`: true when either user blocked the other, in Chat's `dm_blocks` or as a Network platform block (`server/chat/dm.js` `isBlockedEither`, the check the DM and call routes use). `a` and `b` are positive Live user ids and must differ; Live asks this before sending a call invite. |
+| `POST /internal/chat/dm/peers` | `chat.dm.peers.read` | `{subject, candidates}` (subjects, at most 100) → `chat.dm-peers-result@1` `{subject, peers}`: the candidates that share a direct or group conversation with `subject`. Subjects only, never a message or a name. Network's presence privacy rule uses it ("people they follow or have a DM with"). This one route answers the contract shape, without the wrapper `ok`. |
| `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/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/first-chat?channel_id&identity` | `chat.analysis.read` | `{first}`: true when `identity` (`user:<user_id>`, `anon:<anonId>`, `ext:<prefixed username>`; ≤ 160) has never chatted in the channel — Live's `isFirstChatInChannel`, from `stream_first_chats`. `channel_id` is the channel owner's Live user id (`stream_first_chats.channel_user_id`), not `ctx_channels.id`. A registered identity is `user:<user_id>`, the numeric Live user id and **never** the username: Live's own `user:${line.username}` key would read `first:true` forever, so its welcome check must pass `user:${line.user_id}`. |
| `GET /internal/chat/first-chat?channel_id&identity` | `chat.analysis.read` | `{first}`: true when `identity` (`user:<user_id>`, `anon:<anonId>`, `ext:<prefixed username>`; ≤ 160) has never chatted in the channel — Live's `isFirstChatInChannel`, from `stream_first_chats`. `channel_id` is the channel owner's Live user id (`stream_first_chats.channel_user_id`), not `ctx_channels.id`. A registered identity is `user:<user_id>`, the numeric Live user id and **never** the username: Live's own `user:${line.username}` key would read `first:true` forever, so its welcome check must pass `user:${line.user_id}`. |
| `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/pending-ip?channel_id` | `chat.moderation.queue.read` | `{pending_ip}`: the channel's pending rows, oldest first (`limit` ≤ 500, default 50). |
@@ -17,7 +17,7 @@
@@ -17,7 +17,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.315.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.319.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.94.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.94.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.129.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.129.0",
"pg": "^8.23.0",
"pg": "^8.23.0",
@@ -1557,9 +1557,9 @@
@@ -1557,9 +1557,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.315.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.315.0",
- "integrity": "sha512-m1bWT4/sF6Hc7lia6FjAR+880r9SBRhO/v819Lr5ldxlSkFYiKMXTtt7MDbJ4FCgQ7Dkt+2OJaKanlvk2vNPlg==",
+ "version": "0.319.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.319.0",
+ "integrity": "sha512-v0HSuq5d4XZ3tKfXRDgNcmO+M6pQVMsl/pWOhAGRfzSIdjV/G1s9mnsoi7WXhpsJiZ3kEYPP1ie2viyJ+Lg/Zg==",
"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.315.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.319.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.94.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.94.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.129.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.129.0",
"pg": "^8.23.0",
"pg": "^8.23.0",
@@ -34,6 +34,7 @@ const soundboard = require('./soundboard-service');
@@ -34,6 +34,7 @@ const soundboard = require('./soundboard-service');
const audioQueue = require('./audio-queue');
const audioQueue = require('./audio-queue');
const dm = require('./dm');
const dm = require('./dm');
const networkBlocks = require('./network-blocks');
const networkBlocks = require('./network-blocks');
+const presence = require('../presence');
const vipBadges = require('../vip/badges');
const vipBadges = require('../vip/badges');
// Cosmetics and tags are Live's (monetization/cosmetics, game/tags); read through live-context.
// Cosmetics and tags are Live's (monetization/cosmetics, game/tags); read through live-context.
const cosmetics = { getCosmeticProfile: (userId) => ctx.getCosmeticProfile(userId) };
const cosmetics = { getCosmeticProfile: (userId) => ctx.getCosmeticProfile(userId) };
@@ -660,7 +661,7 @@ class ChatServer {
@@ -660,7 +661,7 @@ class ChatServer {
break;
break;
}
}
case 'get-users':
case 'get-users':
- this.sendTo(ws, { type: 'users-list', users: await this.getUserList(client.streamId) });
+ this.sendTo(ws, { type: 'users-list', users: await this.getUserList(client.streamId, client.user && client.user.subject_id) });
break;
break;
default:
default:
break;
break;
@@ -731,9 +732,10 @@ class ChatServer {
@@ -731,9 +732,10 @@ class ChatServer {
* Build a deduplicated list of users in a given stream (or global if null).
* Build a deduplicated list of users in a given stream (or global if null).
* Returns { logged: [{username, display_name, avatar_url, role}], anonCount: N }
* Returns { logged: [{username, display_name, avatar_url, role}], anonCount: N }
*/
*/
- async getUserList(streamId) {
+ async getUserList(streamId, viewerSubject = null) {
const seen = new Set();
const seen = new Set();
const logged = [];
const logged = [];
+ const subjects = new Map();
let anonCount = 0, hiddenCount = 0;
let anonCount = 0, hiddenCount = 0;
for (const [, c] of this.clients) {
for (const [, c] of this.clients) {
if (c.streamId !== streamId) continue;
if (c.streamId !== streamId) continue;
@@ -742,12 +744,14 @@ class ChatServer {
@@ -742,12 +744,14 @@ class ChatServer {
seen.add(c.user.id);
seen.add(c.user.id);
// chat.presence_prefs: someone who turned off "show my name in user lists" is counted, not named.
// chat.presence_prefs: someone who turned off "show my name in user lists" is counted, not named.
if (await this._hiddenFromUserList(c.user)) { hiddenCount++; continue; }
if (await this._hiddenFromUserList(c.user)) { hiddenCount++; continue; }
+ const subject = c.user.subject_id || await db.subjectFor(c.user.id);
logged.push({
logged.push({
username: c.user.username,
username: c.user.username,
display_name: c.user.display_name || c.user.username,
display_name: c.user.display_name || c.user.username,
avatar_url: c.user.avatar_url || null,
avatar_url: c.user.avatar_url || null,
role: c.user.role,
role: c.user.role,
});
});
+ if (subject) subjects.set(c.user.username, subject);
} else if (c.anonId) {
} else if (c.anonId) {
if (seen.has(c.anonId)) continue;
if (seen.has(c.anonId)) continue;
seen.add(c.anonId);
seen.add(c.anonId);
@@ -757,6 +761,10 @@ class ChatServer {
@@ -757,6 +761,10 @@ class ChatServer {
// Sort: admins first, then mods, then alphabetical
// Sort: admins first, then mods, then alphabetical
const rolePriority = { admin: 0, global_mod: 1, streamer: 2, user: 3 };
const rolePriority = { admin: 0, global_mod: 1, streamer: 2, user: 3 };
logged.sort((a, b) => (rolePriority[a.role] ?? 9) - (rolePriority[b.role] ?? 9) || a.display_name.localeCompare(b.display_name));
logged.sort((a, b) => (rolePriority[a.role] ?? 9) - (rolePriority[b.role] ?? 9) || a.display_name.localeCompare(b.display_name));
+ // The Network answers each viewer separately (a stranger's state is never disclosed), so a member
+ // list is built for the person who asked; the read is cached 30 s per viewer and member set.
+ const seenPresence = await presence.forViewer(viewerSubject, [...subjects.values()]);
+ for (const u of logged) u.presence = seenPresence.get(subjects.get(u.username)) || null;
return { logged, anonCount, hiddenCount };
return { logged, anonCount, hiddenCount };
}
}
@@ -2911,12 +2919,14 @@ class ChatServer {
@@ -2911,12 +2919,14 @@ class ChatServer {
if (!this._usersListTimers) this._usersListTimers = new Map();
if (!this._usersListTimers) this._usersListTimers = new Map();
this._usersListTimers.set(key, setTimeout(async () => {
this._usersListTimers.set(key, setTimeout(async () => {
this._usersListTimers.delete(key);
this._usersListTimers.delete(key);
- const users = await this.getUserList(streamId);
- const data = JSON.stringify({ type: 'users-list', users });
+ // Presence is per viewer (the Network decides what this person may see), so the list is built
+ // once per signed-in viewer subject and shared by that viewer's tabs.
+ const byViewer = new Map();
for (const [ws, client] of this.clients) {
for (const [ws, client] of this.clients) {
- if (client.streamId === streamId && ws.readyState === WebSocket.OPEN) {
- try { ws.send(data); } catch {}
- }
+ if (client.streamId !== streamId || ws.readyState !== WebSocket.OPEN) continue;
+ const viewer = (client.user && client.user.subject_id) || null;
+ if (!byViewer.has(viewer)) byViewer.set(viewer, await this.getUserList(streamId, viewer));
+ try { ws.send(JSON.stringify({ type: 'users-list', users: byViewer.get(viewer) })); } catch {}
}
}
}, 500));
}, 500));
}
}
@@ -20,6 +20,7 @@ const { limits } = require('../net/actor-limits');
@@ -20,6 +20,7 @@ const { limits } = require('../net/actor-limits');
const dm = require('./dm');
const dm = require('./dm');
const ctx = require('../live-context');
const ctx = require('../live-context');
const prefStores = require('../prefs/stores');
const prefStores = require('../prefs/stores');
+const presence = require('../presence');
const { idParam } = require('./route-params');
const { idParam } = require('./route-params');
/**
/**
@@ -37,6 +38,12 @@ function publicMessage(m) {
@@ -37,6 +38,12 @@ function publicMessage(m) {
return m;
return m;
}
}
+/** Attach the Network presence view to every person in these conversations (one read, cached 30 s). */
+async function withPresence(viewerSubject, conversations) {
+ await presence.attachTo(viewerSubject, conversations.flatMap((c) => (Array.isArray(c && c.participants) ? c.participants : [])));
+ return conversations;
+}
+
// All DM routes require authentication
// All DM routes require authentication
router.use(requireAuth);
router.use(requireAuth);
@@ -91,6 +98,7 @@ router.get('/conversations', async (req, res) => {
@@ -91,6 +98,7 @@ router.get('/conversations', async (req, res) => {
for (const conv of conversations) {
for (const conv of conversations) {
conv.participants = await dm.getParticipants(conv.id);
conv.participants = await dm.getParticipants(conv.id);
}
}
+ await withPresence(req.user.subject_id, conversations);
res.json({ conversations });
res.json({ conversations });
} catch (err) {
} catch (err) {
console.error('[DM] List conversations error:', err.message);
console.error('[DM] List conversations error:', err.message);
@@ -155,6 +163,7 @@ router.post('/conversations', startLimit, async (req, res) => {
@@ -155,6 +163,7 @@ router.post('/conversations', startLimit, async (req, res) => {
const conversation = await dm.getConversation(conversationId);
const conversation = await dm.getConversation(conversationId);
conversation.participants = await dm.getParticipants(conversationId);
conversation.participants = await dm.getParticipants(conversationId);
+ await withPresence(req.user.subject_id, [conversation]);
res.json({ conversation });
res.json({ conversation });
} catch (err) {
} catch (err) {
console.error('[DM] Create conversation error:', err.message);
console.error('[DM] Create conversation error:', err.message);
@@ -172,6 +181,7 @@ router.get('/conversations/:id', async (req, res) => {
@@ -172,6 +181,7 @@ router.get('/conversations/:id', async (req, res) => {
const conversation = await dm.getConversation(convId);
const conversation = await dm.getConversation(convId);
if (!conversation) return res.status(404).json({ error: 'Not found' });
if (!conversation) return res.status(404).json({ error: 'Not found' });
conversation.participants = await dm.getParticipants(convId);
conversation.participants = await dm.getParticipants(convId);
+ await withPresence(req.user.subject_id, [conversation]);
res.json({ conversation });
res.json({ conversation });
} catch (err) {
} catch (err) {
console.error('[DM] Get conversation error:', err.message);
console.error('[DM] Get conversation error:', err.message);
@@ -235,27 +245,28 @@ router.post('/conversations/:id/messages', sendLimit, async (req, res) => {
@@ -235,27 +245,28 @@ router.post('/conversations/:id/messages', sendLimit, async (req, res) => {
msg.avatar_url = sender?.avatar_url;
msg.avatar_url = sender?.avatar_url;
msg.profile_color = sender?.profile_color;
msg.profile_color = sender?.profile_color;
- // Real-time delivery via chat WebSocket
+ // Real-time delivery via chat WebSocket. A recipient in DND (Network presence) gets the message
+ // with `silent: true` and no Network push: Chat's own ping respects their chosen state.
try {
try {
const chatServer = require('./chat-server');
const chatServer = require('./chat-server');
const participants = await dm.getParticipants(convId);
const participants = await dm.getParticipants(convId);
- for (const p of participants) {
- if (p.id === req.user.id) continue; // skip sender
+ const others = participants.filter(p => p.id !== req.user.id);
+ // One Network read per recipient, in parallel; cached 30 s each. A group is at most 20 people.
+ const own = await Promise.all(others.map((p) => (p.subject_id ? presence.self(p.subject_id) : null)));
+ const dnd = new Set(others.filter((p, i) => own[i] && own[i].state === 'dnd').map((p) => p.id));
+ for (const p of others) {
await chatServer.sendDm(p.id, {
await chatServer.sendDm(p.id, {
type: 'dm',
type: 'dm',
conversation_id: convId,
conversation_id: convId,
message: msg,
message: msg,
+ ...(dnd.has(p.id) ? { silent: true } : {}),
});
});
}
}
- } catch { /* non-critical */ }
-
- // Push notifications to other participants (for when they're offline) — Live's
- // utils/notify pushes them to the Network; the preview stays generic there.
- try {
- const participants = await dm.getParticipants(convId);
- const otherIds = participants.filter(p => p.id !== req.user.id).map(p => p.id);
- if (otherIds.length) {
- ctx.effects.notifyDm({ recipient_ids: otherIds, sender_id: req.user.id, conversation_id: convId });
+ // Push notifications to other participants (for when they're offline) — Live's
+ // utils/notify pushes them to the Network; the preview stays generic there.
+ const notifyIds = others.filter(p => !dnd.has(p.id)).map(p => p.id);
+ if (notifyIds.length) {
+ ctx.effects.notifyDm({ recipient_ids: notifyIds, sender_id: req.user.id, conversation_id: convId });
}
}
} catch { /* non-critical */ }
} catch { /* non-critical */ }
@@ -142,7 +142,7 @@ async function getConversations(userId) {
@@ -142,7 +142,7 @@ async function getConversations(userId) {
*/
*/
async function getParticipants(conversationId) {
async function getParticipants(conversationId) {
return await db.all(`
return await db.all(`
- SELECT u.id, u.username, u.display_name, u.avatar_url, u.profile_color
+ SELECT u.id, COALESCE(p.subject_id, u.subject_id) AS subject_id, u.username, u.display_name, u.avatar_url, u.profile_color
FROM dm_participants p
FROM dm_participants p
JOIN ctx_users u ON u.id = p.user_id
JOIN ctx_users u ON u.id = p.user_id
WHERE p.conversation_id = ?
WHERE p.conversation_id = ?
@@ -7,6 +7,8 @@
@@ -7,6 +7,8 @@
* kind site-daily → chat.site-daily-result@1 { days }
* kind site-daily → chat.site-daily-result@1 { days }
* GET /internal/chat/messages chat.messages.read → chat.messages-page@1
* GET /internal/chat/messages chat.messages.read → chat.messages-page@1
* GET /internal/chat/dm/block-state chat.messages.read → { blocked }
* GET /internal/chat/dm/block-state chat.messages.read → { blocked }
+ * POST /internal/chat/dm/peers chat.dm.peers.read chat.dm-peers-result@1
+ * { subject, candidates } → { subject, peers }
* GET /internal/chat/timeline chat.analysis.read → chat.timeline-result@1
* GET /internal/chat/timeline chat.analysis.read → chat.timeline-result@1
* GET /internal/chat/first-chat chat.analysis.read → chat.first-chat-result@1 { first }
* GET /internal/chat/first-chat chat.analysis.read → chat.first-chat-result@1 { first }
* GET /internal/chat/moderation/pending-ip chat.moderation.queue.read → chat.moderation-queue-result@1 { pending_ip }
* GET /internal/chat/moderation/pending-ip chat.moderation.queue.read → chat.moderation-queue-result@1 { pending_ip }
@@ -209,6 +211,24 @@ async function dmBlockState(req) {
@@ -209,6 +211,24 @@ async function dmBlockState(req) {
return { blocked: await dm.isBlockedEither(q.a, q.b) };
return { blocked: await dm.isBlockedEither(q.a, q.b) };
}
}
+/** Network's presence privacy rule: which candidates share a conversation with `subject`. */
+async function dmPeers(req) {
+ const b = req.body;
+ if (!b || typeof b !== 'object' || Array.isArray(b) || Object.keys(b).some((k) => !['subject', 'candidates'].includes(k))) bad('Invalid fields');
+ if (typeof b.subject !== 'string' || !/^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(b.subject)) bad('Invalid subject');
+ if (!Array.isArray(b.candidates) || b.candidates.length > 100) bad('candidates must be an array of at most 100 subjects');
+ const candidates = [...new Set(b.candidates)];
+ if (!candidates.every((c) => typeof c === 'string' && /^usr_[0-9A-HJKMNP-TV-Z]{26}$/.test(c))) bad('Invalid candidates');
+ if (candidates.includes(b.subject)) bad('a person is not their own peer');
+ if (!candidates.length) return { subject: b.subject, peers: [] };
+ // A shared conversation is a DM relationship, direct or group (a group is at most 20 people).
+ const rows = await db.all(`SELECT DISTINCT p2.subject_id AS peer
+ FROM dm_participants p1
+ JOIN dm_participants p2 ON p2.conversation_id = p1.conversation_id
+ WHERE p1.subject_id = ? AND p2.subject_id <> ? AND p2.subject_id = ANY(?)`, [b.subject, b.subject, candidates]);
+ return { subject: b.subject, peers: rows.map((r) => r.peer).filter(Boolean) };
+}
+
// ── N3: timeline ──
// ── N3: timeline ──
// People's chat only: Live's AI viewers (source_platform 'ai') are left out, so their chatter never makes a "spike" the
// People's chat only: Live's AI viewers (source_platform 'ai') are left out, so their chatter never makes a "spike" the
// AI then clips (item 151, C07: AI clips are excluded from human signals).
// AI then clips (item 151, C07: AI clips are excluded from human signals).
@@ -373,6 +393,14 @@ const router = express.Router();
@@ -373,6 +393,14 @@ const router = express.Router();
router.post('/stats', serviceAuth.guard('chat.stats.read'), route('stats', stats));
router.post('/stats', serviceAuth.guard('chat.stats.read'), route('stats', stats));
router.get('/messages', serviceAuth.guard('chat.messages.read'), route('messages', messages));
router.get('/messages', serviceAuth.guard('chat.messages.read'), route('messages', messages));
router.get('/dm/block-state', serviceAuth.guard('chat.messages.read'), route('dm-block-state', dmBlockState));
router.get('/dm/block-state', serviceAuth.guard('chat.messages.read'), route('dm-block-state', dmBlockState));
+router.post('/dm/peers', serviceAuth.guard('chat.dm.peers.read'), async (req, res) => {
+ try { res.json(await dmPeers(req)); }
+ catch (err) {
+ if (err.status) return res.status(err.status).json({ error: err.message });
+ console.error('[ChatReads] dm-peers:', err);
+ res.status(503).json({ error: 'Chat read unavailable' });
+ }
+});
router.get('/timeline', serviceAuth.guard('chat.analysis.read'), route('timeline', timeline));
router.get('/timeline', serviceAuth.guard('chat.analysis.read'), route('timeline', timeline));
router.get('/first-chat', serviceAuth.guard('chat.analysis.read'), route('first-chat', firstChat));
router.get('/first-chat', serviceAuth.guard('chat.analysis.read'), route('first-chat', firstChat));
const queue = serviceAuth.guard('chat.moderation.queue.read');
const queue = serviceAuth.guard('chat.moderation.queue.read');
new file mode 100644
new file mode 100644
@@ -0,0 +1,68 @@
@@ -0,0 +1,68 @@
+'use strict';
+/**
+ * OpenVibe.Network presence for Chat (plan 2026-10-22 track X item 9; Contracts 0.318.0
+ * network.presence@1).
+ *
+ * Network owns the person's chosen online status. Chat reads it through Network's internal API as a
+ * first-party service (GET /internal/presence?for=usr_…&subjects=…, network.presence.read) and caches
+ * the answer for 30 seconds. The answer already applies the privacy rule for `for`: themselves, people
+ * they follow and people they share a direct conversation with, never a stranger, and never an
+ * invisible person. Chat never widens it; an absent subject is simply drawn as offline.
+ *
+ * DND is read the same way for the person it belongs to (for = themselves): a DM message to a person
+ * in DND arrives with `silent: true` and is left out of the Network push, so Chat does not ping them.
+ *
+ * When Network does not answer, nobody is shown a state (the UI draws offline) and no ping is muted:
+ * failing closed on privacy, and never blocking a message on presence.
+ */
+const config = require('./config');
+const auth = require('./net/service-auth');
+
+const TTL_MS = 30_000;
+const FAIL_TTL_MS = 5_000;
+const MAX_SUBJECTS = 100;
+const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+
+const cache = new Map();
+
+/** Network presence for `subjects` as `viewer` may see it → Map(subject → presence view). */
+async function forViewer(viewer, subjects) {
+ const list = [...new Set((subjects || []).filter((s) => SUBJECT_RE.test(String(s))))].slice(0, MAX_SUBJECTS);
+ if (!viewer || !SUBJECT_RE.test(String(viewer)) || !list.length) return new Map();
+ const key = `${viewer}|${list.sort().join(',')}`;
+ const hit = cache.get(key);
+ if (hit && Date.now() - hit.at < hit.ttl) return hit.value;
+ let value = new Map();
+ let ttl = FAIL_TTL_MS;
+ try {
+ const headers = await auth.headers('openvibe.network', ['network.presence.read']);
+ const url = `${config.networkInternalUrl}/internal/presence?for=${encodeURIComponent(viewer)}&subjects=${list.map(encodeURIComponent).join(',')}`;
+ const res = await fetch(url, { headers: { Accept: 'application/json', ...headers }, signal: AbortSignal.timeout(1500) });
+ if (res.ok) {
+ const body = await res.json().catch(() => ({}));
+ for (const p of Array.isArray(body.presence) ? body.presence : []) if (SUBJECT_RE.test(String(p.subject))) value.set(p.subject, p);
+ ttl = TTL_MS;
+ }
+ } catch { /* Network down: no state, no muted pings */ }
+ if (cache.size > 1000) cache.clear();
+ cache.set(key, { at: Date.now(), value, ttl });
+ return value;
+}
+
+/** Attach `presence` (the Network view or null) to each person in `people`, keyed by subject_id. */
+async function attachTo(viewer, people) {
+ const list = Array.isArray(people) ? people : [];
+ const subjects = list.map((p) => p && p.subject_id).filter(Boolean);
+ const seen = await forViewer(viewer, subjects);
+ for (const p of list) if (p) p.presence = (p.subject_id && seen.get(p.subject_id)) || null;
+ return list;
+}
+
+/** The person's own presence (for DND). Null when Network cannot answer. */
+async function self(subject) {
+ return (await forViewer(subject, [subject])).get(subject) || null;
+}
+
+function clearCache() { cache.clear(); }
+
+module.exports = { forViewer, attachTo, self, clearCache, TTL_MS, SUBJECT_RE, MAX_SUBJECTS };
@@ -96,9 +96,23 @@ async function boot({ env = {} } = {}) {
@@ -96,9 +96,23 @@ async function boot({ env = {} } = {}) {
netModules.set(ns, subject, data, `svc:${String(v.claims.sub).replace(/^svc:/, '')}`);
netModules.set(ns, subject, data, `svc:${String(v.claims.sub).replace(/^svc:/, '')}`);
return send(cur ? 200 : 201, moduleRecord(ns, subject, netModules.records.get(k)));
return send(cur ? 200 : 201, moduleRecord(ns, subject, netModules.records.get(k)));
}
}
+ // Presence for the Network endpoint Chat reads (network.presence.read). Tests set rows here; the
+ // visibility rule itself is Network's and is tested there, so the stub answers the subjects asked for.
+ const netPresence = { rows: new Map(), calls: [] };
const network = http.createServer(async (req, res) => {
const network = http.createServer(async (req, res) => {
const raw = await new Promise((r) => { let s = ''; req.on('data', (c) => { s += c; }); req.on('end', () => r(s)); });
const raw = await new Promise((r) => { let s = ''; req.on('data', (c) => { s += c; }); req.on('end', () => r(s)); });
res.setHeader('Content-Type', 'application/json');
res.setHeader('Content-Type', 'application/json');
+ if (req.url.startsWith('/internal/presence')) {
+ const url = new URL(req.url, ISS);
+ const auth = String(req.headers.authorization || '');
+ const v = auth.startsWith('Bearer ') ? serviceAuth.verifyServiceToken(auth.slice(7), { publicKey: keys.publicKey, issuer: ISS, audience: 'openvibe.network' }) : { ok: false, code: 'token.missing' };
+ if (!v.ok) { res.statusCode = 401; return res.end(JSON.stringify({ error: v.code })); }
+ if (!(v.claims.cap || []).includes('network.presence.read')) { res.statusCode = 403; return res.end(JSON.stringify({ error: 'capability.denied' })); }
+ const forSubject = url.searchParams.get('for');
+ const subjects = String(url.searchParams.get('subjects') || '').split(',').filter(Boolean);
+ netPresence.calls.push({ for: forSubject, subjects });
+ return res.end(JSON.stringify({ presence: subjects.map((s) => netPresence.rows.get(s)).filter(Boolean) }));
+ }
const mm = /^\/internal\/modules\/([^/?]+)\/([^/?]+)$/.exec(req.url);
const mm = /^\/internal\/modules\/([^/?]+)\/([^/?]+)$/.exec(req.url);
if (mm) return modulesRoute(req, res, raw, decodeURIComponent(mm[1]), decodeURIComponent(mm[2]));
if (mm) return modulesRoute(req, res, raw, decodeURIComponent(mm[1]), decodeURIComponent(mm[2]));
if (req.url === '/oauth/token') {
if (req.url === '/oauth/token') {
@@ -298,7 +312,7 @@ async function boot({ env = {} } = {}) {
@@ -298,7 +312,7 @@ async function boot({ env = {} } = {}) {
});
});
const h = {
const h = {
- tmp, keys, live, network, liveServer, tokenRequests, serviceToken, ISS, netModules,
+ tmp, keys, live, network, liveServer, tokenRequests, serviceToken, ISS, netModules, netPresence,
userSeq: 0,
userSeq: 0,
/** Add a Live user (projection row) with a browser token; returns { id, token, ...row }. */
/** Add a Live user (projection row) with a browser token; returns { id, token, ...row }. */
addUser(username, { role = 'user', subject = null, created_at = sqliteNow(-30 * 86400e3), apiScopes = null, is_owner = 0, display_name = null } = {}) {
addUser(username, { role = 'user', subject = null, created_at = sqliteNow(-30 * 86400e3), apiScopes = null, is_owner = 0, display_name = null } = {}) {
new file mode 100644
new file mode 100644
@@ -0,0 +1,117 @@
@@ -0,0 +1,117 @@
+'use strict';
+/**
+ * Chat's Network presence (plan 2026-10-22 track X item 9; Contracts 0.318.0):
+ * - server/presence.js reads GET /internal/presence for one viewer, caches it 30 s, and fails closed;
+ * - DM lists and DM headers carry `presence` per participant;
+ * - room member lists carry `presence` per person, built separately for each viewer;
+ * - DND mutes Chat's own DM ping (`silent: true`) and the Network push;
+ * - POST /internal/chat/dm/peers answers the shared-conversation set for Network
+ * (chat.dm.peers.read, chat.dm-peers-result@1).
+ */
+const assert = require('assert');
+const { validate } = require('openvibe-contracts');
+const { boot, suite } = require('./helpers');
+
+const t = suite('presence');
+const ANN = 'usr_01J9ANN0000000000000000AAA';
+const BOB = 'usr_01J9B0B0000000000000000AAA';
+const CAT = 'usr_01J9CAT0000000000000000AAA';
+let h, ann, bob, cat, conv;
+
+const view = (subject, state, extra = {}) => ({
+ subject, state, status_text: extra.status_text || null, emoji: extra.emoji || null,
+ live: !!extra.live, updated_at: '2026-11-07T09:00:00.000Z',
+});
+
+t('boot with a stub Network that knows ann, bob and cat', async () => {
+ h = await boot();
+ ann = h.addUser('ann', { subject: ANN });
+ bob = h.addUser('bob', { subject: BOB });
+ cat = h.addUser('cat', { subject: CAT });
+ h.netModules.subjects.add(ANN).add(BOB).add(CAT);
+ h.netPresence.rows.set(BOB, view(BOB, 'online', { status_text: 'Around' }));
+ h.netPresence.rows.set(CAT, view(CAT, 'dnd'));
+});
+
+t('a DM list and a DM header carry the presence Network returned for this viewer', async () => {
+ let r = await h.http('POST', '/api/dm/conversations', { token: ann.token, body: { user_ids: [bob.id] } });
+ assert.strictEqual(r.status, 200, r.text);
+ conv = r.body.conversation.id;
+ assert.deepStrictEqual(r.body.conversation.participants.find((p) => p.id === bob.id).presence.state, 'online');
+ assert.strictEqual(h.netPresence.calls.at(-1).for, ANN, 'read as the person who asked');
+
+ r = await h.http('GET', '/api/dm/conversations', { token: ann.token });
+ assert.strictEqual(r.status, 200, r.text);
+ const listed = r.body.conversations.find((c) => c.id === conv);
+ assert.deepStrictEqual(listed.participants.find((p) => p.id === bob.id).presence, view(BOB, 'online', { status_text: 'Around' }));
+
+ r = await h.http('GET', `/api/dm/conversations/${conv}`, { token: bob.token });
+ assert.strictEqual(r.status, 200, r.text);
+ assert.strictEqual(r.body.conversation.participants.find((p) => p.id === ann.id).presence, null, 'no row is a null presence, not an error');
+
+ // The read is cached 30 s: the same viewer and member set does not ask Network again.
+ const calls = h.netPresence.calls.length;
+ await h.http('GET', '/api/dm/conversations', { token: ann.token });
+ assert.strictEqual(h.netPresence.calls.length, calls, 'cached');
+});
+
+t('DND mutes the DM ping and the Network push, and only DND does', async () => {
+ h.netPresence.rows.set(BOB, view(BOB, 'dnd'));
+ require('../server/presence').clearCache();
+ const sent = [];
+ const orig = h.chatServer.sendDm;
+ h.chatServer.sendDm = async (id, payload) => { sent.push({ id, payload }); };
+ try {
+ let r = await h.http('POST', `/api/dm/conversations/${conv}/messages`, { token: ann.token, body: { message: 'heads down?' } });
+ assert.strictEqual(r.status, 200, r.text);
+ assert.strictEqual(sent.length, 1);
+ assert.strictEqual(sent[0].id, bob.id);
+ assert.strictEqual(sent[0].payload.silent, true, 'Bob is in DND');
+ assert.ok(!h.live.effects.some((e) => e.name === 'notify/dm'), 'no Network push for the DND recipient');
+
+ h.netPresence.rows.set(BOB, view(BOB, 'online'));
+ require('../server/presence').clearCache();
+ r = await h.http('POST', `/api/dm/conversations/${conv}/messages`, { token: ann.token, body: { message: 'lunch?' } });
+ assert.strictEqual(r.status, 200, r.text);
+ assert.strictEqual(sent[1].payload.silent, undefined, 'online gets the normal ping');
+ assert.ok(h.live.effects.some((e) => e.name === 'notify/dm'), 'and the Network push');
+ } finally { h.chatServer.sendDm = orig; }
+});
+
+t('room member lists carry presence, built for the viewer', async () => {
+ const streamer = h.addUser('streamer', { role: 'streamer' });
+ const channelId = h.addChannel(streamer.id);
+ const streamId = h.addStream(streamer.id, channelId);
+ await h.ctx.sync();
+ h.netPresence.rows.set(BOB, view(BOB, 'away', { live: true }));
+ require('../server/presence').clearCache();
+ const a = await h.ws({ ip: '198.51.100.41', token: ann.token, stream: streamId });
+ const b = await h.ws({ ip: '198.51.100.42', token: bob.token, stream: streamId });
+ try {
+ a.sendJson({ type: 'join', streamId, token: ann.token });
+ b.sendJson({ type: 'join', streamId, token: bob.token });
+ await a.next((m) => m.type === 'auth' && m.user_id === ann.id);
+ await b.next((m) => m.type === 'auth' && m.user_id === bob.id);
+ a.sendJson({ type: 'get-users' });
+ const list = await a.next((m) => m.type === 'users-list' && m.users.logged.some((u) => u.username === 'bob'));
+ const listed = list.users.logged.find((u) => u.username === 'bob');
+ assert.strictEqual(listed.presence.state, 'away');
+ assert.strictEqual(listed.presence.live, true);
+ assert.strictEqual(h.netPresence.calls.at(-1).for, ANN, 'the list was built for Ann');
+ } finally { a.close(); b.close(); }
+});
+
+t('the DM peer read answers shared conversations only, with subjects only', async () => {
+ const token = h.serviceToken(['chat.dm.peers.read'], { aud: 'openvibe.chat' });
+ let r = await h.http('POST', '/internal/chat/dm/peers', { token, body: { subject: ANN, candidates: [BOB, CAT] } });
+ assert.strictEqual(r.status, 200, r.text);
+ assert.ok(validate('chat.dm-peers-result@1', r.body).valid, JSON.stringify(validate('chat.dm-peers-result@1', r.body).errors));
+ assert.deepStrictEqual([r.body.subject, r.body.peers], [ANN, [BOB]]);
+ assert.ok(!r.text.includes('messages') && !r.text.includes('Around'), 'no message, status or name');
+ r = await h.http('POST', '/internal/chat/dm/peers', { token: h.serviceToken(['chat.messages.read'], { aud: 'openvibe.chat' }), body: { subject: ANN, candidates: [BOB] } });
+ assert.strictEqual(r.status, 403, 'another capability is refused');
+ r = await h.http('POST', '/internal/chat/dm/peers', { token, body: { subject: ANN, candidates: ['bob'] } });
+ assert.strictEqual(r.status, 400);
+ r = await h.http('POST', '/internal/chat/dm/peers', { token, body: { subject: ANN, candidates: [BOB], extra: 1 } });
+ assert.strictEqual(r.status, 400);
+});
@@ -116,6 +116,7 @@ t('ratchet: every file that makes an outbound request itself is reviewed', () =>
@@ -116,6 +116,7 @@ t('ratchet: every file that makes an outbound request itself is reviewed', () =>
'server/net/service-auth.js': 'Network JWKS (configured)',
'server/net/service-auth.js': 'Network JWKS (configured)',
'server/chat/account-data.js': 'Network internal API (configured), fixed paths: export parts and deletion confirmations',
'server/chat/account-data.js': 'Network internal API (configured), fixed paths: export parts and deletion confirmations',
'server/prefs/network-modules.js': 'Network user modules (configured)',
'server/prefs/network-modules.js': 'Network user modules (configured)',
+ 'server/presence.js': 'Network internal presence (configured internal URL), fixed /internal/presence path, the subject list in the query, a service token, 1.5 s, held 30 s',
'server/chat/routes.js': 'GIF providers (fixed hosts)',
'server/chat/routes.js': 'GIF providers (fixed hosts)',
'server/chat/emotes-routes.js': 'FFZ / BTTV / 7TV emote providers (fixed hosts, hardcoded URLs)',
'server/chat/emotes-routes.js': 'FFZ / BTTV / 7TV emote providers (fixed hosts, hardcoded URLs)',
'server/chat/soundboard-service.js': '101soundboards (host allowlist; audio via safeLookup)',
'server/chat/soundboard-service.js': '101soundboards (host allowlist; audio via safeLookup)',