openvibers/openvibe.live Commits

Route Live's chat producers through Chat's ingress behind a flag

OP OpenVibers committed 4 Oct 2026 · 14ef37813f · 29 files, +931 −193

Asked: T3 J2, first half (job 4 of the design /home/workstation/openvibe/agents/jobs/live-t3-j2-prerequisite-design-only-051405.out.md, sections (1)-(3)) — OpenVibe.Live: deliver every Chat-bound path through OpenVibe.Chat's own service-token ingress instead of the chat-remote.js bridge, behind a flag tha…
Harness job live-t3-j2-first-half-job-162702 (plan T3).

Changes

ddiff --git a/.env.example b/.env.example
iindex ffe65c51..03cb673e 100644
---- a/.env.example
++++ b/.env.example
@@@ -169,6 +169,13 @@ OV_COMMUNITY_INTERNAL_URL=http://127.0.0.1:4200
# chat calls to Chat and keeps its chat tables as Chat's read mirror. Unset = Live runs chat.
#CHAT_AUTHORITY=chat
OV_CHAT_INTERNAL_URL=http://127.0.0.1:4400
++# With CHAT_AUTHORITY=chat: "1" sends Live's chat producers (AI viewer lines, donations and alerts, relay
++# and RobotStreamer chat, cards, moderation, cache hints, presence) to Chat's typed ingress
++# /internal/chat/* (server/chat/chat-delivery.js) instead of the chat-remote.js bridge. A Chat 4xx/5xx is
++# logged and counted, never thrown and never handed to the bridge. Turn-on order: Network grants Live
++# chat.event.publish, chat.moderation.write and chat.cache.invalidate (audience openvibe.chat) → Chat's
++# ingress is deployed → set this → watch chat_bridge_outbox stop growing → then the bridge removal.
++#LIVE_CHAT_INGRESS=1
# Voice/video calls run in OpenVibe.Chat when this is "chat" (WS-I task 1;
# server/streaming/calls-authority.js): go-live, stream end, WHIP teardown and the admin force-end
# create/remove stream voice channels in Chat (POST/DELETE /internal/calls/stream-channel, the chat
ddiff --git a/README.md b/README.md
iindex 4ed0340b..60870880 100644
---- a/README.md
++++ b/README.md
@@@ -96,13 +96,21 @@ Live's own principal (`live`) holds grants to call:
- **Network**: `identity.subject.resolve`, `network.follows.read|write`, `network.modules.read|write`, `network.notifications.push`, `network.coins.credit|debit`, `network.analytics.creator.read`, `network.account.export.contribute`, `network.account.deletion.confirm`.
- **AI**: `ai.run.create|read` (namespaces `live.*`, `network.site_copy`, `media.analyze`), `ai.credential.manage`, `ai.quota.attribution.manage`.
--- **Chat**: `chat.live_bridge.write`, `chat.message.send`, `chat.presence.read`.
++- **Chat**: `chat.live_bridge.write`, `chat.message.send`, `chat.presence.read`; with `LIVE_CHAT_INGRESS=1` also `chat.event.publish`, `chat.moderation.write`, `chat.cache.invalidate`.
- **Community**: `community.paste.*`, `community.comment.*`, `community.pulse.write`.
- **Events**: `events.event.publish|read`, `events.subscription.manage`.
- **OpenRe**: `openre.stream.*`, `openre.key.rotate`, `openre.session.read`.
- **Billing**: `billing.*` (intents, transfers, subscriptions, cash-outs, balances, entitlements).
- **Elsewhere**: `tips.interaction.record`, `vip.entitlement.check`, `tools.tool.run`, `tools.job.read`.
++**Chat ingress (`LIVE_CHAT_INGRESS`, off by default).** With `CHAT_AUTHORITY=chat` and `LIVE_CHAT_INGRESS=1`, every chat producer in Live goes through one seam, `server/chat/chat-delivery.js`, to OpenVibe.Chat's typed service-token ingress (`server/chat/chat-client.js`: `POST /internal/chat/messages|events|moderation|invalidate`, `GET /internal/chat/presence`) instead of the `chat-remote.js` bridge. Each operation carries one idempotency key, reused on every retry. A Chat 4xx is logged, counted and dropped; a 5xx or timeout is retried with the same key past Chat's five-minute delivery lease, then logged, counted and dropped. Neither is thrown into the caller, and neither falls back to the bridge (that would deliver twice). Turn it on in this order:
++
++1. Network grants Live `chat.event.publish`, `chat.moderation.write` and `chat.cache.invalidate` (audience `openvibe.chat`).
++2. Chat's ingress is deployed.
++3. Set `LIVE_CHAT_INGRESS=1` and restart Live.
++4. Watch `chat_bridge_outbox` stop growing.
++5. Then the removal PR (the bridge writers and the outbox drop) and Chat's J3.
++
API writes are limited per person (`server/net/actor-limits.js`, `LIVE_LIMITS_MINUTE` / `_HOUR`).
## Acceptance
ddiff --git a/server/admin/mod-routes.js b/server/admin/mod-routes.js
iindex f74831a7..4cbffa96 100644
---- a/server/admin/mod-routes.js
++++ b/server/admin/mod-routes.js
@@@ -31,6 +31,7 @@ const db = require('../db/database');
const { requireAuth } = require('../auth/auth');
const permissions = require('../auth/permissions');
const chatServer = require('../chat/chat-server');
++const delivery = require('../chat/chat-delivery');
const ipUtils = require('./ip-utils');
const router = express.Router();
@@@ -121,7 +122,7 @@ function performGlobalBan(req, res, { userId, reason, durationHours, ipAddress }
[banReason + ` (alt of ${targetUser.username})`, alt.id]);
db.run(`INSERT INTO bans (user_id, reason, banned_by, expires_at) VALUES (?, ?, ?, ?)`,
[alt.id, banReason + ` (alt of ${targetUser.username})`, req.user.id, expires]);
-- chatServer.disconnectUser({ userId: alt.id });
++ delivery.disconnect({ userId: alt.id });
cascadeBanned++;
}
if (cascadeBanned || cascadeSkippedStaff) {
@@@ -130,9 +131,9 @@ function performGlobalBan(req, res, { userId, reason, durationHours, ipAddress }
}
// Immediately disconnect the user from chat
-- chatServer.disconnectUser({ userId: targetUser.id, ip: resolvedIp });
++ delivery.disconnect({ userId: targetUser.id, ip: resolvedIp });
-- db.logModerationAction({
++ delivery.logModeration({
scope_type: 'site',
actor_user_id: req.user.id,
target_user_id: targetUser.id,
@@@ -175,9 +176,9 @@ router.delete('/users/:id/ban', permissions.requireGlobalMod, (req, res) => {
db.run('UPDATE users SET is_banned = 0, ban_reason = NULL WHERE id = ?', [userId]);
db.run('DELETE FROM bans WHERE user_id = ? AND stream_id IS NULL', [userId]);
-- chatServer.disconnectUser({ userId });
++ delivery.disconnect({ userId });
-- db.logModerationAction({
++ delivery.logModeration({
scope_type: 'site',
actor_user_id: req.user.id,
target_user_id: userId,
@@@ -224,9 +225,9 @@ router.post('/stream-ban', async (req, res) => {
);
// Disconnect from this stream's chat
-- chatServer.disconnectUser({ userId: targetUser.id, ip: resolvedIp, streamId });
++ delivery.disconnect({ userId: targetUser.id, ip: resolvedIp, streamId });
-- db.logModerationAction({
++ delivery.logModeration({
scope_type: 'stream',
scope_id: streamId,
actor_user_id: req.user.id,
@@@ -247,9 +248,9 @@ router.post('/stream-ban', async (req, res) => {
);
// Disconnect the anon user
-- if (resolvedIp) chatServer.disconnectUser({ ip: resolvedIp, streamId });
++ if (resolvedIp) delivery.disconnect({ ip: resolvedIp, streamId });
-- db.logModerationAction({
++ delivery.logModeration({
scope_type: 'stream',
scope_id: streamId,
actor_user_id: req.user.id,
@@@ -292,7 +293,7 @@ router.delete('/ban/:id', async (req, res) => {
db.run('DELETE FROM bans WHERE id = ?', [req.params.id]);
-- db.logModerationAction({
++ delivery.logModeration({
scope_type: ban.stream_id ? 'stream' : 'site',
scope_id: ban.stream_id || undefined,
actor_user_id: req.user.id,
@@@ -385,13 +386,18 @@ router.post('/delete-message', async (req, res) => {
if (!allowed) return res.status(403).json({ error: 'You cannot moderate this stream' });
}
-- db.deleteChatMessage(parseInt(message_id), req.user.id);
++ // LIVE_CHAT_INGRESS: Chat deletes the row and broadcasts the delete to every surface it reached.
++ if (delivery.ingress()) {
++ delivery.mirror('deleteChatMessage', parseInt(message_id), req.user.id);
++ await delivery.moderate('delete-message', { id: parseInt(message_id), deleted_by: req.user.id }, { key: `delete:${message_id}` });
++ }
++ else db.deleteChatMessage(parseInt(message_id), req.user.id);
// Broadcast deletion to connected clients — reach the streamer's whole
// channel room (all slots + offline viewers), plus the global feed.
const delPayload = { type: 'delete-messages', ids: [parseInt(message_id)] };
const chanUid = message.channel_user_id || (message.stream_id ? (db.getStreamById(message.stream_id)?.user_id || null) : null);
-- if (chanUid || message.stream_id) {
++ if (delivery.ingress()) { /* broadcast by Chat */ } else if (chanUid || message.stream_id) {
chatServer.broadcastToChannelRoom(chanUid, message.stream_id, delPayload);
if (message.stream_id) chatServer.forwardToGlobal(message.stream_id, delPayload);
else chatServer.broadcastGlobal(delPayload);
@@@ -399,7 +405,7 @@ router.post('/delete-message', async (req, res) => {
chatServer.broadcastGlobal(delPayload);
}
-- db.logModerationAction({
++ delivery.logModeration({
scope_type: message.stream_id ? 'stream' : 'site',
scope_id: message.stream_id || undefined,
actor_user_id: req.user.id,
@@@ -435,7 +441,17 @@ router.post('/delete-user-messages', async (req, res) => {
}
let ids = [];
-- if (user_id) {
++ if (delivery.ingress()) {
++ // Chat deletes the rows and broadcasts the deletes (also the site-wide one) itself.
++ const [action, subject] = user_id ? ['delete-user-messages', { user_id: parseInt(user_id) }]
++ : anon_id ? ['delete-anon-messages', { anon_id: String(anon_id) }] : ['delete-relay-messages', { username: String(relay_username) }];
++ const opts = { streamId: scopedStreamId, deletedBy: req.user.id };
++ if (user_id) delivery.mirror('deleteUserChatMessages', parseInt(user_id), opts);
++ else if (anon_id) delivery.mirror('deleteAnonChatMessages', anon_id, opts);
++ else delivery.mirror('deleteRelayUserMessages', relay_username, opts);
++ const r = await delivery.moderate(action, { ...subject, stream_id: scopedStreamId ? parseInt(scopedStreamId) : undefined, deleted_by: req.user.id });
++ ids = (r && Array.isArray(r.ids)) ? r.ids : [];
++ } else if (user_id) {
ids = db.deleteUserChatMessages(parseInt(user_id), { streamId: scopedStreamId, deletedBy: req.user.id });
} else if (anon_id) {
ids = db.deleteAnonChatMessages(anon_id, { streamId: scopedStreamId, deletedBy: req.user.id });
@@@ -444,7 +460,7 @@ router.post('/delete-user-messages', async (req, res) => {
}
// Broadcast deletions to connected clients
-- if (ids.length > 0) {
++ if (ids.length > 0 && !delivery.ingress()) {
const deleteMsg = { type: 'delete-messages', ids };
if (scopedStreamId) {
const chanUid = db.getStreamById(scopedStreamId)?.user_id || null;
@@@ -460,7 +476,7 @@ router.post('/delete-user-messages', async (req, res) => {
}
const target = user_id ? `user ${user_id}` : anon_id ? `anon ${anon_id}` : `relay ${relay_username}`;
-- db.logModerationAction({
++ delivery.logModeration({
scope_type: scopedStreamId ? 'stream' : 'site',
scope_id: scopedStreamId || undefined,
actor_user_id: req.user.id,
@@@ -503,8 +519,12 @@ router.post('/relay-user/hide', async (req, res) => {
// Unban / unhide — remove the ban for this exact relay identity only.
if (action === 'unban' || action === 'unhide') {
-- db.unhideRelayUserByIdentity(channel_id || null, platform, external_username);
-- db.logModerationAction({
++ if (delivery.ingress()) {
++ delivery.mirror('unhideRelayUserByIdentity', channel_id || null, platform, external_username);
++ await delivery.moderate('relay-unhide', { channel_id: channel_id || undefined, platform, external_username });
++ }
++ else db.unhideRelayUserByIdentity(channel_id || null, platform, external_username);
++ delivery.logModeration({
scope_type: channel_id ? 'channel' : 'site',
scope_id: channel_id || undefined,
actor_user_id: req.user.id,
@@@ -514,7 +534,9 @@ router.post('/relay-user/hide', async (req, res) => {
return res.json({ message: `[${platform}] ${external_username} unbanned` });
}
-- db.hideRelayUser({
++ if (delivery.ingress()) {
++ await delivery.moderate('relay-hide', { channel_id: channel_id || undefined, platform, external_username, mode: action || 'hide', reason: reason || undefined, created_by: req.user.id });
++ } else db.hideRelayUser({
channelId: channel_id || null,
platform,
externalUsername: external_username,
@@@ -523,7 +545,7 @@ router.post('/relay-user/hide', async (req, res) => {
createdBy: req.user.id,
});
-- db.logModerationAction({
++ delivery.logModeration({
scope_type: channel_id ? 'channel' : 'site',
scope_id: channel_id || undefined,
actor_user_id: req.user.id,
@@@ -547,9 +569,10 @@ router.delete('/relay-user/:id', async (req, res) => {
const allowed = permissions.isGlobalModOrAbove(req.user)
|| (row.channel_id != null && await permissions.canModerateChannel(req.user, row.channel_id));
if (!allowed) return res.status(403).json({ error: 'You cannot moderate this channel' });
-- db.unhideRelayUser(row.id);
++ if (delivery.ingress()) { delivery.mirror('unhideRelayUser', row.id); await delivery.moderate('relay-unhide', { id: row.id }); }
++ else db.unhideRelayUser(row.id);
-- db.logModerationAction({
++ delivery.logModeration({
scope_type: 'site',
actor_user_id: req.user.id,
action_type: 'relay_user_unhide',
@@@ -608,7 +631,7 @@ router.get('/ip-approval/:channelId/pending', async (req, res) => {
});
// ── Approve all messages from an IP ──────────────────────────
--router.post('/ip-approval/:channelId/approve', (req, res) => {
++router.post('/ip-approval/:channelId/approve', async (req, res) => {
try {
const channelId = parseInt(req.params.channelId);
const { ip } = req.body;
@@@ -626,10 +649,15 @@ router.post('/ip-approval/:channelId/approve', (req, res) => {
[channelId, ip]
);
-- db.approveAllFromIp(channelId, ip, req.user.id);
++ // LIVE_CHAT_INGRESS: Chat approves its held rows; re-showing them live is Chat's (it holds them).
++ if (delivery.ingress()) {
++ delivery.mirror('approveAllFromIp', channelId, ip, req.user.id);
++ await delivery.moderate('approve-ip-messages', { channel_id: parseInt(channelId), ip, reviewed_by: req.user.id });
++ }
++ else db.approveAllFromIp(channelId, ip, req.user.id);
// Broadcast previously-held messages as real chat messages
-- for (const msg of pendingMsgs) {
++ for (const msg of delivery.ingress() ? [] : pendingMsgs) {
if (msg.stream_id) {
const chatMsg = {
type: 'chat',
@@@ -649,7 +677,7 @@ router.post('/ip-approval/:channelId/approve', (req, res) => {
}
}
-- db.logModerationAction({
++ delivery.logModeration({
scope_type: 'channel',
scope_id: channelId,
actor_user_id: req.user.id,
@@@ -665,7 +693,7 @@ router.post('/ip-approval/:channelId/approve', (req, res) => {
});
// ── Deny all messages from an IP ─────────────────────────────
--router.post('/ip-approval/:channelId/deny', (req, res) => {
++router.post('/ip-approval/:channelId/deny', async (req, res) => {
try {
const channelId = parseInt(req.params.channelId);
const { ip } = req.body;
@@@ -677,9 +705,13 @@ router.post('/ip-approval/:channelId/deny', (req, res) => {
return res.status(403).json({ error: 'Access denied' });
}
-- db.denyAllFromIp(channelId, ip, req.user.id);
++ if (delivery.ingress()) {
++ delivery.mirror('denyAllFromIp', channelId, ip, req.user.id);
++ await delivery.moderate('deny-ip-messages', { channel_id: parseInt(channelId), ip, reviewed_by: req.user.id });
++ }
++ else db.denyAllFromIp(channelId, ip, req.user.id);
-- db.logModerationAction({
++ delivery.logModeration({
scope_type: 'channel',
scope_id: channelId,
actor_user_id: req.user.id,
@@@ -695,7 +727,7 @@ router.post('/ip-approval/:channelId/deny', (req, res) => {
});
// ── Review a single pending message ──────────────────────────
--router.post('/ip-approval/:channelId/review', (req, res) => {
++router.post('/ip-approval/:channelId/review', async (req, res) => {
try {
const channelId = parseInt(req.params.channelId);
const { message_id, status } = req.body;
@@@ -709,7 +741,10 @@ router.post('/ip-approval/:channelId/review', (req, res) => {
return res.status(403).json({ error: 'Access denied' });
}
-- db.reviewPendingIpMessage(parseInt(message_id), {
++ if (delivery.ingress()) {
++ delivery.mirror('reviewPendingIpMessage', parseInt(message_id), { status, reviewedBy: req.user.id, channelId });
++ await delivery.moderate('review-pending-ip', { id: parseInt(message_id), status, reviewed_by: req.user.id, channel_id: parseInt(channelId) });
++ } else db.reviewPendingIpMessage(parseInt(message_id), {
status,
reviewedBy: req.user.id,
channelId,
@@@ -827,10 +862,10 @@ router.post('/ip/ban-all', permissions.requireGlobalMod, (req, res) => {
});
// Disconnect all clients on this IP
-- chatServer.disconnectUser({ ip });
++ delivery.disconnect({ ip });
// Log the action
-- db.logModerationAction({
++ delivery.logModeration({
scope_type: 'site',
actor_user_id: req.user.id,
action_type: 'ip_ban_all',
@@@ -911,7 +946,8 @@ router.put('/tts-voice/:kind/:id', permissions.requireGlobalMod, (req, res) => {
try {
const identityKey = _ttsIdentityKey(req.params.kind, req.params.id);
if (req.body?.reset) {
-- db.deleteTtsVoiceOverride(identityKey);
++ if (delivery.ingress()) { delivery.mirror('deleteTtsVoiceOverride', identityKey); delivery.moderate('tts-voice-override', { identity_key: identityKey }); }
++ else db.deleteTtsVoiceOverride(identityKey);
return res.json({ isOverride: false, params: ttsEngine.autoUserVoiceParams(identityKey) });
}
let params;
@@@ -928,7 +964,8 @@ router.put('/tts-voice/:kind/:id', permissions.requireGlobalMod, (req, res) => {
params = { voice: req.body?.voice, pitch: req.body?.pitch, speed: req.body?.speed, gap: req.body?.gap };
}
const clamped = ttsEngine.clampVoiceParams(params);
-- db.setTtsVoiceOverride(identityKey, clamped, req.user.id);
++ if (delivery.ingress()) { delivery.mirror('setTtsVoiceOverride', identityKey, clamped, req.user.id); delivery.moderate('tts-voice-override', { identity_key: identityKey, params: clamped, set_by: req.user.id }); }
++ else db.setTtsVoiceOverride(identityKey, clamped, req.user.id);
res.json({ isOverride: true, params: clamped });
} catch (err) {
console.error('[Mod] tts-voice set error:', err.message);
ddiff --git a/server/admin/routes.js b/server/admin/routes.js
iindex 8ac780ae..4d79e4de 100644
---- a/server/admin/routes.js
++++ b/server/admin/routes.js
@@@ -43,6 +43,7 @@ const { execSync } = require('child_process');
const db = require('../db/database');
const { requireAuth } = require('../auth/auth');
const chatServer = require('../chat/chat-server');
++const delivery = require('../chat/chat-delivery');
const permissions = require('../auth/permissions');
const money = require('../monetization/money-authority');
@@@ -244,7 +245,12 @@ router.put('/users/:id', (req, res) => {
// Push real-time update to the affected user's chat connections
if (updates.length > 0) {
-- chatServer.sendUserUpdate(parseInt(req.params.id), safeUser);
++ const id = parseInt(req.params.id);
++ // LIVE_CHAT_INGRESS: a typed account hint; Chat updates the sockets and stored message names.
++ delivery.invalidate({
++ user: id,
++ user_data: { id, username: safeUser.username, display_name: safeUser.display_name || null, role: safeUser.role || null, avatar_url: safeUser.avatar_url || null, profile_color: safeUser.profile_color || null },
++ }, () => chatServer.sendUserUpdate(id, safeUser));
}
res.json({ user: safeUser });
@@@ -277,7 +283,7 @@ router.post('/users/:id/ban', (req, res) => {
[req.params.id, reason || 'Banned by admin', req.user.id, expires]
);
-- db.logModerationAction({
++ delivery.logModeration({
scope_type: 'site',
actor_user_id: req.user.id,
target_user_id: Number(req.params.id),
@@@ -297,7 +303,7 @@ router.delete('/users/:id/ban', (req, res) => {
db.run('UPDATE users SET is_banned = 0, ban_reason = NULL WHERE id = ?', [req.params.id]);
db.run('DELETE FROM bans WHERE user_id = ?', [req.params.id]);
-- db.logModerationAction({
++ delivery.logModeration({
scope_type: 'site',
actor_user_id: req.user.id,
target_user_id: Number(req.params.id),
@@@ -358,7 +364,7 @@ router.delete('/streams/:id', (req, res) => {
// Accountability: force-ending someone's live stream is a moderation action
// and MUST be logged (previously it left no trace at all).
try {
-- db.logModerationAction({
++ delivery.logModeration({
scope_type: 'stream',
scope_id: stream.id,
actor_user_id: req.user.id,
@@@ -804,7 +810,7 @@ router.post('/moderators', (req, res) => {
db.run("UPDATE users SET role = 'global_mod' WHERE id = ?", [user.id]);
-- db.logModerationAction({
++ delivery.logModeration({
scope_type: 'site',
actor_user_id: req.user.id,
target_user_id: user.id,
@@@ -827,7 +833,7 @@ router.delete('/moderators/:id', (req, res) => {
db.run("UPDATE users SET role = 'user' WHERE id = ?", [user.id]);
-- db.logModerationAction({
++ delivery.logModeration({
scope_type: 'site',
actor_user_id: req.user.id,
target_user_id: user.id,
@@@ -863,7 +869,7 @@ router.post('/admins', permissions.requireOwner, (req, res) => {
if (!user) return res.status(404).json({ error: 'User not found' });
if (user.role === 'admin') return res.status(400).json({ error: 'User is already an admin' });
db.run("UPDATE users SET role = 'admin' WHERE id = ?", [user.id]);
-- db.logModerationAction({
++ delivery.logModeration({
scope_type: 'site', actor_user_id: req.user.id, target_user_id: user.id,
action_type: 'admin_promote', details: { username: user.username },
});
@@@ -881,7 +887,7 @@ router.delete('/admins/:id', permissions.requireOwner, (req, res) => {
if (user.role !== 'admin') return res.status(400).json({ error: 'User is not an admin' });
if (user.is_owner) return res.status(400).json({ error: 'Cannot demote an owner' });
db.run("UPDATE users SET role = 'user' WHERE id = ?", [user.id]);
-- db.logModerationAction({
++ delivery.logModeration({
scope_type: 'site', actor_user_id: req.user.id, target_user_id: user.id,
action_type: 'admin_demote', details: { username: user.username },
});
ddiff --git a/server/ai/viewers/poster.js b/server/ai/viewers/poster.js
iindex 80cd1eec..c7f7fba5 100644
---- a/server/ai/viewers/poster.js
++++ b/server/ai/viewers/poster.js
@@@ -43,6 +43,20 @@ function post(worker, bot, message, { threadId = null, replyToId = null } = {})
const streamId = worker.streamId;
const persona = botPersona(bot);
const color = botColor(bot);
++ const ttsOn = worker.settings.tts_enabled !== false && persona.tts !== false;
++ const delivery = require('../../chat/chat-delivery');
++ if (delivery.ingress()) {
++ // Chat persists, broadcasts, mirrors to global and reads it aloud in one call; the real id arrives later.
++ delivery.after(delivery.message({
++ stream_id: streamId, channel_user_id: worker.userId || undefined, username: bot.username, message,
++ message_type: 'chat', is_global: false, reply_to_id: replyToId || undefined, source_platform: 'ai',
++ metadata: { bot: 1, bot_id: bot.id, source: bot.source, thread_id: threadId || undefined },
++ mirror: true, frame: { role: 'user', profile_color: color, is_ai: true, filtered: false },
++ tts: ttsOn ? { identity_key: `aibot:${bot.username.toLowerCase()}` } : undefined,
++ }), (id) => forwardToPowerChat(worker, bot, message, id));
++ try { db.touchChannelAiBot(bot.id); } catch { /* */ }
++ return null;
++ }
let id = null;
try {
const res = db.saveChatMessage({
@@@ -74,10 +88,16 @@ function post(worker, bot, message, { threadId = null, replyToId = null } = {})
const chatServer = require('../../chat/chat-server');
chatServer.broadcastToStream(streamId, chatMsg);
chatServer.forwardToGlobal(streamId, chatMsg);
-- const ttsOn = worker.settings.tts_enabled !== false && persona.tts !== false;
if (ttsOn) chatServer.synthesizeAndBroadcastTTS(streamId, bot.username, message, null, null, `aibot:${bot.username.toLowerCase()}`, null, id ? `m${id}` : null);
} catch (e) { console.warn('[AI-Viewers] broadcast failed:', e.message); }
++ forwardToPowerChat(worker, bot, message, id);
++ try { db.touchChannelAiBot(bot.id); } catch { /* */ }
++ return id;
++}
++
++function forwardToPowerChat(worker, bot, message, id) {
++ const streamId = worker.streamId;
if (worker.settings.powerchat_forward !== false) {
try {
const pc = require('../../integrations/powerchat-platform');
@@@ -92,8 +112,6 @@ function post(worker, bot, message, { threadId = null, replyToId = null } = {})
}
} catch { /* non-critical */ }
}
-- try { db.touchChannelAiBot(bot.id); } catch { /* */ }
-- return id;
}
/** Append a row to ai_viewer_log (best-effort). */
ddiff --git a/server/auth/network-events.js b/server/auth/network-events.js
iindex 73ac27cb..327288c6 100644
---- a/server/auth/network-events.js
++++ b/server/auth/network-events.js
@@@ -23,7 +23,13 @@ const revocations = require('./revocations');
const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;
const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const stats = { received: 0, revoked: 0, unchanged: 0, profiles: 0, ignored: 0, refused: 0 };
--const notifyChat = (userId) => { try { const cs = require('../chat/chat-server'); if (cs.remote) cs.userChanged(userId); } catch { /* non-critical */ } };
++const notifyChat = (userId) => {
++ try {
++ const delivery = require('../chat/chat-delivery');
++ if (delivery.ingress()) { delivery.invalidate({ user: Number(userId) }); return; }
++ const cs = require('../chat/chat-server'); if (cs.remote) cs.userChanged(userId);
++ } catch { /* non-critical */ }
++};
function secret() { return process.env.LIVE_EVENTS_SECRET || process.env.MEDIA_EVENTS_SECRET || ''; }
ddiff --git a/server/chat/chat-client.js b/server/chat/chat-client.js
nnew file mode 100644
iindex 00000000..539a8109
---- /dev/null
++++ b/server/chat/chat-client.js
@@@ -0,0 +1,127 @@
++'use strict';
++/**
++ * OpenVibe.Chat's typed service-token ingress (T3 J2), the replacement for the chat-remote.js bridge.
++ *
++ * message(body) POST /internal/chat/messages chat.message.send persist + broadcast a line (or a DM)
++ * event(body) POST /internal/chat/events chat.event.publish a transient card, alert, sound, notice
++ * moderation(body) POST /internal/chat/moderation chat.moderation.write deletes, IP reviews, relay hides, log, disconnect
++ * invalidate(body) POST /internal/chat/invalidate chat.cache.invalidate a cache hint (user, approvals, bans)
++ * presence() GET /internal/chat/presence chat.presence.read the connection snapshot
++ *
++ * Auth: Live's Network service principal (client_credentials, audience openvibe.chat, server/net/network-principal.js).
++ * Every POST carries one idempotency `key` made when the operation is created and reused on every retry, so Chat
++ * applies it once (Chat answers the first result again; see OpenVibe.Chat docs/chat-ingress.md).
++ *
++ * Errors are logged and counted (`stats`), never thrown into the caller: every POST resolves to Chat's answer or
++ * null. A 4xx is final (Chat refused the body; retrying cannot help). A 5xx, a timeout or Chat being unreachable
++ * is retried with the same key on a back-off that outlasts Chat's five-minute delivery lease, then dropped. There
++ * is deliberately no fallback to the bridge: once Chat owns delivery, a second path would deliver twice.
++ * Retries live in memory only (a Live restart drops the ones still waiting). Off under LIVE_DRILL.
++ */
++const crypto = require('crypto');
++const principal = require('../net/network-principal');
++const { CHAT_URL } = require('./chat-authority');
++
++const AUDIENCE = 'openvibe.chat';
++const TIMEOUT_MS = 5000;
++const RETRY_MS = [1000, 5000, 30000, 120000, 360000];
++const MAX_WAITING = 2000;
++const KEY_RE = /[^A-Za-z0-9:._-]/g;
++
++const stats = { sent: 0, refused: 0, failed: 0, retried: 0, dropped: 0, byFamily: {}, lastError: null };
++let waiting = 0;
++let lastLog = null;
++let retryMs = RETRY_MS;
++
++function drill() { try { return require('../drill').enabled; } catch { return false; } }
++
++function count(family, result) {
++ const f = stats.byFamily[family] || (stats.byFamily[family] = { sent: 0, refused: 0, failed: 0, dropped: 0 });
++ f[result]++;
++ stats[result]++;
++}
++
++function note(message) {
++ stats.lastError = message;
++ if (message === lastLog) return; // log once until it changes
++ lastLog = message;
++ console.warn(`[ChatIngress] ${message}`);
++}
++
++/** A stable idempotency key: `live:<family>:<natural id or a fresh uuid>`, safe for Chat's key pattern. */
++function key(family, natural) {
++ const tail = natural == null || natural === '' ? crypto.randomUUID() : String(natural);
++ return `live:${family}:${tail}`.replace(KEY_RE, '_').slice(0, 160);
++}
++
++async function request(method, path, body, retried = false) {
++ const res = await fetch(`${CHAT_URL}${path}`, {
++ method,
++ headers: { Accept: 'application/json', ...(body ? { 'Content-Type': 'application/json' } : {}), ...(await principal.serviceHeaders(AUDIENCE)) },
++ body: body ? JSON.stringify(body) : undefined,
++ signal: AbortSignal.timeout(TIMEOUT_MS),
++ });
++ if (res.status === 401 && !retried) { principal.invalidate(AUDIENCE); return request(method, path, body, true); }
++ let data = null;
++ try { data = await res.json(); } catch { /* not JSON */ }
++ return { status: res.status, data };
++}
++
++const sleep = (ms) => new Promise((resolve) => { const t = setTimeout(resolve, ms); if (t.unref) t.unref(); });
++
++/** POST one operation to /internal/chat/<family>, retrying 5xx/unreachable with the same key. Never rejects. */
++async function send(family, body) {
++ if (drill()) return null;
++ if (!principal.configured()) { count(family, 'dropped'); note(`${family}: not configured (OV_OAUTH_CLIENT_SECRET unset); dropped`); return null; }
++ const payload = { ...body, key: body.key || key(family) };
++ if (waiting >= MAX_WAITING) { count(family, 'dropped'); note(`${family}: ${waiting} operations already waiting for Chat; dropped`); return null; }
++ waiting++;
++ try {
++ for (let attempt = 0; ; attempt++) {
++ let status = 0;
++ let data = null;
++ let why;
++ try {
++ ({ status, data } = await request('POST', `/internal/chat/${family}`, payload));
++ why = `answered ${status}${data && data.error ? ` (${data.error})` : ''}`;
++ } catch (err) { why = err.message; }
++ if (status >= 200 && status < 300) { count(family, 'sent'); lastLog = null; return data || { ok: true }; }
++ if (status >= 400 && status < 500) {
++ count(family, 'refused');
++ note(`${family} ${payload.key}: Chat ${why}; not retried`);
++ return null;
++ }
++ if (attempt >= retryMs.length) {
++ count(family, 'failed');
++ note(`${family} ${payload.key}: Chat ${why}; gave up after ${attempt + 1} attempts`);
++ return null;
++ }
++ stats.retried++;
++ note(`${family}: Chat ${why}; retrying with the same key`);
++ await sleep(retryMs[attempt]);
++ }
++ } finally { waiting--; }
++}
++
++/** Chat's presence snapshot ({ total, streams, slow_mode, users, anons }), or null. */
++async function presence() {
++ if (drill() || !principal.configured()) return null;
++ try {
++ const { status, data } = await request('GET', '/internal/chat/presence');
++ if (status === 200 && data) return data;
++ note(`presence: Chat answered ${status}`);
++ } catch (err) { note(`presence: ${err.message}`); }
++ return null;
++}
++
++module.exports = {
++ AUDIENCE,
++ key,
++ stats,
++ message: (body) => send('messages', body),
++ event: (body) => send('events', body),
++ moderation: (body) => send('moderation', body),
++ invalidate: (body) => send('invalidate', body),
++ presence,
++ _setRetryMs(ms) { retryMs = ms || RETRY_MS; },
++};
ddiff --git a/server/chat/chat-delivery.js b/server/chat/chat-delivery.js
nnew file mode 100644
iindex 00000000..55c4b1ad
---- /dev/null
++++ b/server/chat/chat-delivery.js
@@@ -0,0 +1,101 @@
++'use strict';
++/**
++ * The one seam Live's chat producers call (T3 J2, first half).
++ *
++ * LIVE_CHAT_INGRESS=1 (with CHAT_AUTHORITY=chat): OpenVibe.Chat's typed ingress (chat-client.js) — Chat persists,
++ * broadcasts, mirrors, speaks and moderates in one call per operation, and nothing goes through the bridge.
++ * Otherwise (the default): the existing path — require('./chat-server'), which is the chat-remote.js bridge under
++ * CHAT_AUTHORITY=chat and Live's own ChatServer without it.
++ *
++ * With the flag on, a Chat 4xx/5xx is logged and counted by chat-client.js and never thrown into the caller; the
++ * operation is not handed to the bridge (it would deliver twice once Chat owns delivery). Every function here
++ * returns a promise in ingress mode and the legacy call's own value otherwise.
++ *
++ * ingress() which path is active
++ * message(body) ingress only: a chat line (Chat's /messages body) → Promise<real id | null>
++ * event(target, frame, opts) a transient frame; target { kind: stream|channel|global|all|user, id, stream }
++ * moderate(action, fields) ingress only: Chat's /moderation actions → Promise<Chat's answer | null>
++ * logModeration(entry) a moderation-log row (db.logModerationAction shape)
++ * disconnect({ userId, ip, streamId })
++ * invalidate(hint, legacy) a cache hint ({ user, user_data, approvals, bans, channel })
++ * mirror(fn, ...args) ingress only: keep Live's own copy of a bridge 'local' write (pending IP rows,
++ * hidden relay users, first chats, TTS overrides) that Live still reads
++ * after(value, fn) fn(value) now for a plain value, after it resolves for a promise
++ */
++const chatAuthority = require('./chat-authority');
++const client = require('./chat-client');
++
++function ingress() { return process.env.LIVE_CHAT_INGRESS === '1' && chatAuthority.isRemote(); }
++// Lazy: chat-server pulls in most of Live, and several producers are required by it.
++function chatServer() { return require('./chat-server'); }
++
++function defined(o) {
++ const out = {};
++ for (const [k, v] of Object.entries(o || {})) if (v !== undefined) out[k] = v;
++ return out;
++}
++
++// Chat's ingress takes these as positive integers; route params and bodies hand Live strings.
++const INT_FIELDS = ['id', 'user_id', 'stream_id', 'channel_id', 'channel_user_id', 'reply_to_id', 'reviewed_by', 'actor_user_id', 'target_user_id', 'created_by', 'set_by'];
++function ints(o) {
++ for (const k of INT_FIELDS) if (typeof o[k] === 'string' && /^\d+$/.test(o[k])) o[k] = Number(o[k]);
++ return o;
++}
++
++function message(body) {
++ const b = ints(defined(body));
++ if (b.key) b.key = client.key('messages', b.key);
++ if (b.is_global != null) b.is_global = !!b.is_global;
++ if (b.frame) b.frame = defined(b.frame);
++ if (b.tts) b.tts = defined(b.tts);
++ return client.message(b).then((r) => (r && r.id != null ? Number(r.id) : null));
++}
++
++function event(target, frame, { key } = {}) {
++ if (ingress()) {
++ const t = { kind: target.kind };
++ if (target.id != null) t.id = Number(target.id);
++ return client.event({ key: key ? client.key('events', key) : undefined, target: t, frame: defined(frame) });
++ }
++ const cs = chatServer();
++ if (target.kind === 'stream') return cs.broadcastToStream(target.id, frame);
++ if (target.kind === 'channel') return cs.broadcastToChannelRoom(target.id, target.stream || null, frame);
++ if (target.kind === 'global') return cs.broadcastGlobal(frame);
++ if (target.kind === 'all') return cs.broadcastAll(frame);
++ if (target.kind === 'user') return cs.sendDm(target.id, frame);
++ throw new Error(`unknown chat target ${target.kind}`);
++}
++
++function moderate(action, fields, { key } = {}) {
++ return client.moderation(ints({ ...defined(fields), action, key: key ? client.key('moderation', key) : undefined }));
++}
++
++const LOG_FIELDS = ['scope_type', 'scope_id', 'actor_user_id', 'target_user_id', 'action_type', 'details'];
++function logModeration(entry) {
++ if (!ingress()) return require('../db/database').logModerationAction(entry);
++ const fields = {};
++ for (const k of LOG_FIELDS) if (entry[k] != null) fields[k] = entry[k];
++ if (typeof fields.scope_id === 'string' && fields.scope_type !== 'room' && /^\d+$/.test(fields.scope_id)) fields.scope_id = Number(fields.scope_id);
++ return moderate('log', fields);
++}
++
++function disconnect({ userId, ip, streamId } = {}) {
++ if (!ingress()) return chatServer().disconnectUser(defined({ userId, ip, streamId }));
++ return moderate('disconnect', { user_id: userId || undefined, ip: ip || undefined, stream_id: streamId || undefined });
++}
++
++/** legacy: what to run without the flag (cache hints had no single legacy call). */
++function invalidate(hint, legacy) {
++ if (ingress()) return client.invalidate(defined(hint));
++ return legacy ? legacy() : undefined;
++}
++
++function mirror(fn, ...args) {
++ try { return require('./chat-remote').localWrite(fn, ...args); } catch (err) { console.warn(`[ChatIngress] local ${fn}: ${err.message}`); return undefined; }
++}
++
++function after(value, fn) {
++ return value && typeof value.then === 'function' ? value.then(fn) : fn(value);
++}
++
++module.exports = { ingress, message, event, moderate, logModeration, disconnect, invalidate, mirror, after, client };
ddiff --git a/server/chat/chat-remote.js b/server/chat/chat-remote.js
iindex b34f0810..d5a07bfa 100644
---- a/server/chat/chat-remote.js
++++ b/server/chat/chat-remote.js
@@@ -28,6 +28,9 @@ const crypto = require('crypto');
const db = require('../db/database');
const principal = require('../net/network-principal');
const { CHAT_URL } = require('./chat-authority');
++// LIVE_CHAT_INGRESS=1: Live's producers call Chat's typed ingress through chat-delivery.js; this proxy then reads
++// presence from /internal/chat/presence and sends its cache hints there too. Queued outbox rows still drain here.
++const delivery = require('./chat-delivery');
const AUDIENCE = 'openvibe.chat';
const BOOT = crypto.randomUUID();
@@@ -75,6 +78,15 @@ const OBSERVED_DB = {
forgiveBan: () => ['bans', null],
};
++// db functions before forwarding wrapped them (module-wide: installed once per process).
++const originals = {};
++
++/** Apply a 'local'-mode write to Live's own mirror only, without the bridge (LIVE_CHAT_INGRESS sends it to Chat). */
++function localWrite(fn, ...args) {
++ if (FORWARDED_DB[fn] !== 'local') throw new Error(`${fn} is not a local mirror write`);
++ return (originals[fn] || db[fn])(...args);
++}
++
function create(ChatServer) {
// A local ChatServer that never listens: its address and anon-number helpers are Live's own
// (call-server uses them for voice calls, and Chat asks Live for anon numbers).
@@@ -198,7 +210,8 @@ function create(ChatServer) {
async function pollPresence() {
try {
-- const p = await post('/internal/live/presence');
++ const p = delivery.ingress() ? await delivery.client.presence() : await post('/internal/live/presence');
++ if (!p) return;
presence = p;
slowModeByStream.clear();
for (const [k, v] of Object.entries(p.slow_mode || {})) slowModeByStream.set(Number(k), Number(v) || 0);
@@@ -206,7 +219,6 @@ function create(ChatServer) {
}
// ── Forwarded database writes ───────────────────────────────
-- const originals = {};
function installDbForwarding() {
for (const [fn, mode] of Object.entries(FORWARDED_DB)) {
if (typeof db[fn] !== 'function' || originals[fn]) continue;
@@@ -231,7 +243,11 @@ function create(ChatServer) {
originals[fn] = db[fn];
db[fn] = function observedLiveWrite(...args) {
const result = originals[fn](...args);
-- try { const [kind, id] = target(args); enqueue('invalidate', [kind, id]); } catch { /* */ }
++ try {
++ const [kind, id] = target(args);
++ if (delivery.ingress()) delivery.invalidate(kind === 'bans' ? { bans: true } : { [kind]: Number(id) || undefined });
++ else enqueue('invalidate', [kind, id]);
++ } catch { /* */ }
return result;
};
}
@@@ -376,4 +392,4 @@ function drainToLocal(log = console) {
} catch (err) { log.warn('[ChatRemote] drain:', err.message); return 0; }
}
--module.exports = { create, drainToLocal, FORWARDED_DB, OBSERVED_DB, REF_BASE, AUDIENCE };
++module.exports = { create, drainToLocal, localWrite, FORWARDED_DB, OBSERVED_DB, REF_BASE, AUDIENCE };
ddiff --git a/server/chat/deploy-notice.js b/server/chat/deploy-notice.js
iindex 38e6f56b..c6a898c5 100644
---- a/server/chat/deploy-notice.js
++++ b/server/chat/deploy-notice.js
@@@ -112,7 +112,12 @@ async function announce({ db, chatServer, log = console }) {
// message and shows it (its own copy of this module). It learns the commits from the
// live.release.deployed event; only with Events publishing off are they handed over the bridge.
if (chatServer && chatServer.remote) {
-- if (!outbox) chatServer.deployNotice(commits);
++ if (!outbox) {
++ // Chat's ingress has no deploy endpoint: without Events the commits stay unannounced and the
++ // next boot tries again (LIVE_CHAT_INGRESS, docs: OpenVibe.Chat docs/chat-ingress.md).
++ if (require('./chat-delivery').ingress()) { log.warn('[Deploy notice] Events outbox unavailable; left unannounced for the next boot'); return { announced: 0 }; }
++ chatServer.deployNotice(commits);
++ }
try { inTransaction(recordDeploy); } catch (err) { log.warn('[Deploy notice] not recorded:', err.message); return { announced: commits.length, event_id: null }; }
if (outbox) outbox.kick();
return { announced: commits.length, event_id: eventId };
ddiff --git a/server/chat/live-context-routes.js b/server/chat/live-context-routes.js
iindex 2b80ae67..5d765d6b 100644
---- a/server/chat/live-context-routes.js
++++ b/server/chat/live-context-routes.js
@@@ -456,14 +456,20 @@ effectsRouter.post('/ai/mod-command', (req, res) => {
// !hype / !beef / !arena: arena-chat answers the sender through Chat (sendToConn).
effectsRouter.post('/arena-command', (req, res) => {
const client = req.body?.client || {};
-- const cs = chatServer();
-- const shim = {
-- sendTo: (_ws, payload) => cs.sendToConn(client.conn_id, payload),
-- broadcastToStream: (streamId, payload) => cs.broadcastToStream(streamId, payload),
++ const delivery = require('./chat-delivery');
++ // LIVE_CHAT_INGRESS: room lines go to Chat's typed events; the sender-only replies come back in this
++ // response as `replies` (Chat has no per-connection ingress; it reads them from here at J4b).
++ const replies = [];
++ const shim = delivery.ingress() ? {
++ sendTo: (_ws, payload) => { replies.push(payload); },
++ broadcastToStream: (streamId, payload) => delivery.event({ kind: 'stream', id: streamId }, payload),
++ } : {
++ sendTo: (_ws, payload) => chatServer().sendToConn(client.conn_id, payload),
++ broadcastToStream: (streamId, payload) => chatServer().broadcastToStream(streamId, payload),
};
let handled = false;
try { handled = require('../arena/arena-chat').handle(shim, null, client, String(req.body.cmd || ''), Array.isArray(req.body.parts) ? req.body.parts : []); } catch (e) { console.warn('[Arena] chat command:', e.message); }
-- res.json({ handled: !!handled });
++ res.json(replies.length ? { handled: !!handled, replies } : { handled: !!handled });
});
// !sr / !queue / !np / !skip — the media queue is Live's.
ddiff --git a/server/index.js b/server/index.js
iindex 34c2a003..f634c33e 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -844,7 +844,7 @@ app.post('/api/admin/broadcast', requireAuth, permissions.requireAdmin, (req, re
try {
const { type = 'system', message, summary, url } = req.body;
if (!message && !summary) return res.status(400).json({ error: 'message or summary required' });
-- chatServer.broadcastAll({
++ require('./chat/chat-delivery').event({ kind: 'all' }, {
type,
message: message || summary,
summary,
@@@ -990,7 +990,7 @@ app.post('/banned/continue', (req, res) => {
let lifted = [];
try {
lifted = db.forgiveBan(subject.id);
-- db.logModerationAction({ scope_type: 'site', target_user_id: subject.id, action_type: 'unban', details: { via: 'ban-page-continue', ip: req.ip, lifted: lifted.map(r => r.ip_address || 'account') } });
++ require('./chat/chat-delivery').logModeration({ scope_type: 'site', target_user_id: subject.id, action_type: 'unban', details: { via: 'ban-page-continue', ip: req.ip, lifted: lifted.map(r => r.ip_address || 'account') } });
console.log(`[Ban] ${subject.username} (id ${subject.id}) pressed Continue on the ban page from ${req.ip} — ${lifted.length} ban row(s) lifted`);
} catch (e) {
console.error('[Ban] continue failed:', e.message);
@@@ -1609,7 +1609,7 @@ function shutdown() {
// Notify all chat clients before closing connections
try {
-- chatServer.broadcastAll({
++ require('./chat/chat-delivery').event({ kind: 'all' }, {
type: 'server_restart',
message: '⚙️ Chat server restarting — you will be reconnected automatically.',
timestamp: new Date().toISOString(),
ddiff --git a/server/integrations/chat-relay-service.js b/server/integrations/chat-relay-service.js
iindex 73ffeec2..fe18caa7 100644
---- a/server/integrations/chat-relay-service.js
++++ b/server/integrations/chat-relay-service.js
@@@ -395,6 +395,9 @@ class ChatRelayService {
}
} catch { /* non-critical — allow message through on error */ }
++ const delivery = require('../chat/chat-delivery');
++ if (delivery.ingress()) return this._deliverToChat(delivery, bridge, username, prefixedUsername, color, extras, String(message || ''));
++
// Record this relay user's activity; their first message becomes a join date.
try { db.recordRelayUser(bridge.platform, username); } catch { /* non-critical */ }
@@@ -428,13 +431,7 @@ class ChatRelayService {
} catch {}
chatServer.broadcastToStream(bridge.streamId, chatMsg);
-- // AI viewers see relayed chat too (they used to only see native chat).
-- try {
-- require('./ai-chatbot-service').onRealChatMessage(bridge.streamId, {
-- username: prefixedUsername, message: chatMsg.message, userId: null, anonId: null,
-- platform: bridge.platform, relayUsername: username, isStreamer: false, isMod: false, msgId: chatMsg.id || null,
-- });
-- } catch { /* non-critical */ }
++ this._relayFollowUps(bridge, username, prefixedUsername, chatMsg.message, extras, chatMsg.id || null, 'ai');
// Also surface on the global / username-only overlay (tags stream_channel)
try { chatServer.forwardToGlobal(bridge.streamId, chatMsg); } catch { /* non-critical */ }
// And to viewers of the streamer's other live slots (cross-slot chat)
@@@ -445,6 +442,37 @@ class ChatRelayService {
chatServer.synthesizeAndBroadcastTTS(bridge.streamId, prefixedUsername, chatMsg.message, null, bridge.platform, `${bridge.platform}:${prefixedUsername}`, null, chatMsg.id ? `m${chatMsg.id}` : null);
} catch { /* non-critical */ }
++ this._relayFollowUps(bridge, username, prefixedUsername, chatMsg.message, extras, chatMsg.id || null, 'powerchat');
++
++ // Welcome first-time external chatters in this streamer's channel
++ try {
++ const stream = db.getStreamById(bridge.streamId);
++ if (stream?.user_id) {
++ const chatterKey = `ext:${prefixedUsername}`;
++ if (db.isFirstChatInChannel(chatterKey, stream.user_id)) {
++ db.recordFirstChat(chatterKey, stream.user_id);
++ chatServer.broadcastToStream(bridge.streamId, {
++ type: 'system',
++ message: `Welcome ${username} from ${PLATFORM_LABELS[bridge.platform] || bridge.platform}! 👋`,
++ timestamp: new Date().toISOString(),
++ });
++ }
++ }
++ } catch { /* non-critical */ }
++ }
++
++ /** What follows a relayed line once it has an id: AI viewers hear it, PowerChat's overlay shows it. */
++ _relayFollowUps(bridge, username, prefixedUsername, message, extras, id, which = 'all') {
++ if (which !== 'powerchat') {
++ // AI viewers see relayed chat too (they used to only see native chat).
++ try {
++ require('./ai-chatbot-service').onRealChatMessage(bridge.streamId, {
++ username: prefixedUsername, message, userId: null, anonId: null,
++ platform: bridge.platform, relayUsername: username, isStreamer: false, isMod: false, msgId: id,
++ });
++ } catch { /* non-critical */ }
++ }
++ if (which === 'ai') return;
// Merge relayed chat into the streamer's PowerChat unified overlay too. Only
// native OpenVibe messages were forwarded before, so a streamer whose audience
// mostly chats on Twitch/Kick saw an empty overlay.
@@@ -455,8 +483,8 @@ class ChatRelayService {
pc.forwardChat(stream.user_id, {
chatterName: prefixedUsername,
externalChatterId: `${bridge.platform}:${username}`,
-- message: chatMsg.message,
-- messageId: chatMsg.id ? `ov-${chatMsg.id}` : undefined,
++ message: message,
++ messageId: id ? `ov-${id}` : undefined,
avatarUrl: extras.avatar_url || undefined,
// Placeholder letter from the REAL name — "[Twitch] name" would
// otherwise render "[" for every relayed chatter.
@@@ -464,22 +492,32 @@ class ChatRelayService {
});
}
} catch { /* non-critical */ }
++ }
-- // Welcome first-time external chatters in this streamer's channel
++ /** LIVE_CHAT_INGRESS: Chat persists, broadcasts, mirrors, reads aloud and records the first chat in one call. */
++ _deliverToChat(delivery, bridge, username, prefixedUsername, color, extras, message) {
++ let first = false;
try {
const stream = db.getStreamById(bridge.streamId);
-- if (stream?.user_id) {
-- const chatterKey = `ext:${prefixedUsername}`;
-- if (db.isFirstChatInChannel(chatterKey, stream.user_id)) {
-- db.recordFirstChat(chatterKey, stream.user_id);
-- chatServer.broadcastToStream(bridge.streamId, {
-- type: 'system',
-- message: `Welcome ${username} from ${PLATFORM_LABELS[bridge.platform] || bridge.platform}! 👋`,
-- timestamp: new Date().toISOString(),
-- });
-- }
-- }
++ first = !!(stream?.user_id && db.isFirstChatInChannel(`ext:${prefixedUsername}`, stream.user_id));
} catch { /* non-critical */ }
++ delivery.mirror('recordRelayUser', bridge.platform, username);
++ delivery.moderate('relay-record', { platform: bridge.platform, username });
++ delivery.after(delivery.message({
++ stream_id: bridge.streamId, username: prefixedUsername, message, message_type: 'chat', is_global: false,
++ source_platform: bridge.platform, mirror: true,
++ frame: { role: 'external', profile_color: color, avatar_url: extras.avatar_url || undefined },
++ tts: { identity_key: `${bridge.platform}:${prefixedUsername}` },
++ }), (id) => this._relayFollowUps(bridge, username, prefixedUsername, message, extras, id));
++ if (first) {
++ // Chat records the first chat in its copy; Live's copy decides the next welcome.
++ try { delivery.mirror('recordFirstChat', `ext:${prefixedUsername}`, db.getStreamById(bridge.streamId).user_id); } catch { /* */ }
++ delivery.event({ kind: 'stream', id: bridge.streamId }, {
++ type: 'system',
++ message: `Welcome ${username} from ${PLATFORM_LABELS[bridge.platform] || bridge.platform}! 👋`,
++ timestamp: new Date().toISOString(),
++ });
++ }
}
// ── Twitch IRC ────────────────────────────────────────────
ddiff --git a/server/integrations/powerchat-webhook.js b/server/integrations/powerchat-webhook.js
iindex 92e716d4..2a3566c1 100644
---- a/server/integrations/powerchat-webhook.js
++++ b/server/integrations/powerchat-webhook.js
@@@ -15,6 +15,7 @@
const crypto = require('crypto');
const db = require('../db/database');
++const delivery = require('../chat/chat-delivery');
const powerchatOAuth = require('./powerchat-oauth');
const MAX_SKEW_MS = 15 * 60 * 1000;
@@@ -90,6 +91,15 @@ function _testFulfillmentAllowed() {
}
// ── Donation handling — mirrors POST /api/funds/donate ───────────────────────
++// A persisted chat line plus its live card: one Chat ingress call under LIVE_CHAT_INGRESS (Chat persists and
++// shows it; `mirror` also puts it in global chat), else the legacy card broadcast(s) and a local save.
++function _chatLine(chatServer, { frame, line, mirror = false, key }) {
++ if (delivery.ingress()) { delivery.message({ ...line, mirror, key }); return; }
++ chatServer.broadcastToChannelRoom(line.channel_user_id, line.stream_id, frame);
++ if (mirror) { try { chatServer.broadcastGlobal({ ...frame, global: true, channel_user_id: line.channel_user_id }); } catch { /* */ } }
++ db.saveChatMessage(line);
++}
++
function _handleDonation(userId, data) {
const chatServer = require('../chat/chat-server');
const alerts = require('../monetization/alerts');
@@@ -117,14 +127,16 @@ function _handleDonation(userId, data) {
type: 'donation', username: donor, user_id: null, avatar_url: null,
amount, message, source: 'powerchat', timestamp: ts,
};
-- chatServer.broadcastToChannelRoom(userId, streamId, donationEvent);
-- try { chatServer.broadcastGlobal({ ...donationEvent, global: true, channel_user_id: userId }); } catch { /* */ }
++ const eventKey = data.eventId ? `powerchat:${data.eventId}` : undefined;
try {
-- db.saveChatMessage({
-- stream_id: streamId, channel_user_id: userId, user_id: null, username: donor,
-- message: `${donor} tipped ${amount.toLocaleString()} Vibes${message ? ': ' + message : ''} (PowerChat)`,
-- message_type: 'donation',
-- metadata: { kind: 'donation', amount, message, username: donor, source: 'powerchat' },
++ _chatLine(chatServer, {
++ frame: donationEvent, mirror: true, key: eventKey,
++ line: {
++ stream_id: streamId, channel_user_id: userId, user_id: null, username: donor,
++ message: `${donor} tipped ${amount.toLocaleString()} Vibes${message ? ': ' + message : ''} (PowerChat)`,
++ message_type: 'donation',
++ metadata: { kind: 'donation', amount, message, username: donor, source: 'powerchat' },
++ },
});
} catch { /* */ }
@@@ -133,17 +145,20 @@ function _handleDonation(userId, data) {
// 3) Goal progress + 4) goal reached.
if (goalResult && goalResult.goal) {
-- chatServer.broadcastToChannelRoom(userId, streamId, { type: 'goal-update', goal: publicGoal(goalResult.goal) });
++ delivery.event({ kind: 'channel', id: userId, stream: streamId }, { type: 'goal-update', goal: publicGoal(goalResult.goal) });
}
if (goalResult && goalResult.reached) {
const g = goalResult.goal;
-- chatServer.broadcastToChannelRoom(userId, streamId, { type: 'goal-reached', goal: publicGoal(g), by: donor, timestamp: ts });
try {
-- db.saveChatMessage({
-- stream_id: streamId, channel_user_id: userId, user_id: null, username: 'Donation Goal',
-- message: `🎉 Goal reached: ${g.title} (${Number(g.target_amount || 0).toLocaleString()} HB)`,
-- message_type: 'donation',
-- metadata: { kind: 'goal-reached', goal_id: g.id, title: g.title, target: g.target_amount, image: g.image_url || null, media_type: g.media_type || null, by: donor },
++ _chatLine(chatServer, {
++ frame: { type: 'goal-reached', goal: publicGoal(g), by: donor, timestamp: ts },
++ key: eventKey && `${eventKey}:goal`,
++ line: {
++ stream_id: streamId, channel_user_id: userId, user_id: null, username: 'Donation Goal',
++ message: `🎉 Goal reached: ${g.title} (${Number(g.target_amount || 0).toLocaleString()} HB)`,
++ message_type: 'donation',
++ metadata: { kind: 'goal-reached', goal_id: g.id, title: g.title, target: g.target_amount, image: g.image_url || null, media_type: g.media_type || null, by: donor },
++ },
});
} catch { /* */ }
try { alerts.playAlertSound(chatServer, userId, streamId, 'goal'); } catch { /* */ }
@@@ -157,12 +172,15 @@ function _handleSubscription(userId, data) {
const chatServer = require('../chat/chat-server');
const name = String(data.subscriberName || data.donorName || 'Someone').slice(0, 80);
const ts = new Date().toISOString();
-- chatServer.broadcastToChannelRoom(userId, null, { type: 'donation', username: name, amount: 0, message: 'subscribed via PowerChat', source: 'powerchat-sub', timestamp: ts });
try {
-- db.saveChatMessage({
-- stream_id: null, channel_user_id: userId, user_id: null, username: name,
-- message: `${name} subscribed (PowerChat)`, message_type: 'donation',
-- metadata: { kind: 'donation', amount: 0, username: name, source: 'powerchat-sub' },
++ _chatLine(chatServer, {
++ frame: { type: 'donation', username: name, amount: 0, message: 'subscribed via PowerChat', source: 'powerchat-sub', timestamp: ts },
++ key: data.eventId ? `powerchat:${data.eventId}` : undefined,
++ line: {
++ stream_id: null, channel_user_id: userId, user_id: null, username: name,
++ message: `${name} subscribed (PowerChat)`, message_type: 'donation',
++ metadata: { kind: 'donation', amount: 0, username: name, source: 'powerchat-sub' },
++ },
});
} catch { /* */ }
}
@@@ -172,10 +190,12 @@ function _handleNotice(userId, message, kind) {
try {
const chatServer = require('../chat/chat-server');
const ts = new Date().toISOString();
-- chatServer.broadcastToChannelRoom(userId, null, { type: 'system', message, source: 'powerchat', kind, timestamp: ts });
-- db.saveChatMessage({
-- stream_id: null, channel_user_id: userId, user_id: null, username: 'PowerChat',
-- message, message_type: 'system', metadata: { kind: kind || 'powerchat', source: 'powerchat' },
++ _chatLine(chatServer, {
++ frame: { type: 'system', message, source: 'powerchat', kind, timestamp: ts },
++ line: {
++ stream_id: null, channel_user_id: userId, user_id: null, username: 'PowerChat',
++ message, message_type: 'system', metadata: { kind: kind || 'powerchat', source: 'powerchat' },
++ },
});
} catch { /* */ }
}
@@@ -279,18 +299,20 @@ function simulateDonation(userId, { amountUsd = 5, donor = 'Test Tipper', messag
type: 'donation', username: donor, user_id: null, avatar_url: null,
amount, message, source: 'powerchat-test', timestamp: ts,
};
-- chatServer.broadcastToChannelRoom(userId, streamId, testEvent);
// Not to global chat: a test tip is the streamer's own preview, and any signed-in user could
// otherwise announce a fake 99,900-Vibe donation site-wide.
// Persist it, exactly like a real tip. A test that vanishes on reload does not
// actually test what the streamer is checking — that the alert lands in chat AND
// survives a refresh. This was the only donation path that broadcast without saving.
try {
-- db.saveChatMessage({
-- stream_id: streamId, channel_user_id: userId, user_id: null, username: donor,
-- message: `${donor} donated ${amount.toLocaleString()} Vibes${message ? ': ' + message : ''}`,
-- message_type: 'donation',
-- metadata: { kind: 'donation', amount, message, username: donor, source: 'powerchat-test' },
++ _chatLine(chatServer, {
++ frame: testEvent,
++ line: {
++ stream_id: streamId, channel_user_id: userId, user_id: null, username: donor,
++ message: `${donor} donated ${amount.toLocaleString()} Vibes${message ? ': ' + message : ''}`,
++ message_type: 'donation',
++ metadata: { kind: 'donation', amount, message, username: donor, source: 'powerchat-test' },
++ },
});
} catch (e) { console.warn('[PowerChat] test tip not saved to history:', e.message); }
try { alerts.playAlertSound(chatServer, userId, streamId, 'donation'); } catch { /* */ }
@@@ -301,7 +323,7 @@ function simulateDonation(userId, { amountUsd = 5, donor = 'Test Tipper', messag
if (active.length === 1) {
const g = active[0];
const preview = { ...g, current_amount: Math.min((g.current_amount || 0) + amount, g.target_amount) };
-- chatServer.broadcastToChannelRoom(userId, streamId, { type: 'goal-update', goal: publicGoal(preview), preview: true });
++ delivery.event({ kind: 'channel', id: userId, stream: streamId }, { type: 'goal-update', goal: publicGoal(preview), preview: true });
}
} catch { /* */ }
return { amount, live: !!streamId };
ddiff --git a/server/integrations/robotstreamer-service.js b/server/integrations/robotstreamer-service.js
iindex 0751748e..18e8c05a 100644
---- a/server/integrations/robotstreamer-service.js
++++ b/server/integrations/robotstreamer-service.js
@@@ -4,12 +4,42 @@ const WebSocket = require('ws');
const db = require('../db/database');
const chatServer = require('../chat/chat-server');
++const delivery = require('../chat/chat-delivery');
const { authenticateWs } = require('../auth/auth');
const API_HOST = 'api.robotstreamer.com';
const API_PORT = 443;
const RS_ORIGIN = 'https://robotstreamer.com';
++/** What follows a mirrored RobotStreamer line once it has an id: AI viewers hear it, PowerChat's overlay shows it. */
++function rsFollowUps(stream, username, rawUsername, message, avatar, id, which = 'all') {
++ if (which !== 'powerchat') {
++ try {
++ require('./ai-chatbot-service').onRealChatMessage(stream.id, {
++ username, message, userId: null, anonId: null,
++ platform: 'rs', relayUsername: username, isStreamer: false, isMod: false, msgId: id,
++ });
++ } catch { /* non-critical */ }
++ }
++ if (which === 'ai') return;
++ // ...and into the streamer's PowerChat overlay (relayed chat was never forwarded).
++ try {
++ const pc = require('./powerchat-platform');
++ if (stream.user_id && pc.slotRelayEnabled(stream.id)) {
++ pc.forwardChat(stream.user_id, {
++ chatterName: username,
++ externalChatterId: `rs:${username}`,
++ message,
++ messageId: id ? `ov-${id}` : undefined,
++ avatarUrl: avatar || undefined,
++ // Placeholder letter from the REAL name — "[RS] name"
++ // would otherwise render "[" for every RS chatter.
++ avatarFallback: [...String(rawUsername || '')][0] || undefined,
++ });
++ }
++ } catch { /* non-critical */ }
++}
++
function safeJsonParse(value, fallback = null) {
try {
return JSON.parse(value);
@@@ -588,7 +618,7 @@ class RobotStreamerService {
robot_id: bridge.robotId,
owner_id: bridge.ownerId,
}));
-- chatServer.broadcastToStream(stream.id, {
++ delivery.event({ kind: 'stream', id: stream.id }, {
type: 'system',
message: 'RobotStreamer chat mirror connected',
timestamp: new Date().toISOString(),
@@@ -602,7 +632,7 @@ class RobotStreamerService {
if (data.type === 'history' || data.type === 'privileges') return;
if (data.username === '[RS BOT]') {
-- chatServer.broadcastToStream(stream.id, {
++ delivery.event({ kind: 'stream', id: stream.id }, {
type: 'system',
message: `[RS BOT] ${data.message}`,
timestamp: new Date().toISOString(),
@@@ -636,7 +666,8 @@ class RobotStreamerService {
// Record this relay user (first message = join date) so RobotStreamer
// chatters get the same chat logs + AI insight as other relay users.
-- try { db.recordRelayUser('rs', rawUsername); } catch { /* non-critical */ }
++ if (delivery.ingress()) { delivery.mirror('recordRelayUser', 'rs', rawUsername); delivery.moderate('relay-record', { platform: 'rs', username: rawUsername }); }
++ else { try { db.recordRelayUser('rs', rawUsername); } catch { /* non-critical */ } }
// Let RobotStreamer viewers trigger channel !sound commands too. If the
// message is a registered !sound, play it (attributed to the RS user) and
@@@ -646,6 +677,13 @@ class RobotStreamerService {
const parts = trimmed.split(/\s+/);
const scmd = parts[0].slice(1).toLowerCase();
if (scmd && db.getChannelSoundByCommand(stream.user_id, scmd)) {
++ if (delivery.ingress()) {
++ delivery.event({ kind: 'stream', id: stream.id }, {
++ type: 'channel-sound', streamId: stream.id, command: scmd, args: parts.slice(1).map((a) => a.slice(0, 120)).slice(0, 20),
++ relay: { username, role: 'external', profile_color: '#7dd3fc', avatar_url: data.avatar || null, sourcePlatform: 'rs' },
++ });
++ return;
++ }
try {
chatServer.triggerChannelSound(
null,
@@@ -673,6 +711,16 @@ class RobotStreamerService {
source_platform: 'rs',
};
++ if (delivery.ingress()) {
++ // Chat persists, broadcasts, mirrors and reads it aloud in one call.
++ delivery.after(delivery.message({
++ stream_id: stream.id, username, message: msgText, message_type: 'chat', is_global: false,
++ source_platform: 'rs', mirror: true,
++ frame: { role: 'external', profile_color: '#7dd3fc', avatar_url: data.avatar || undefined },
++ tts: { identity_key: `rs:${username}` },
++ }), (id) => rsFollowUps(stream, username, rawUsername, msgText, data.avatar, id));
++ return;
++ }
try {
const result = db.saveChatMessage({
stream_id: stream.id,
@@@ -688,12 +736,7 @@ class RobotStreamerService {
} catch {}
chatServer.broadcastToStream(stream.id, mirrored);
-- try {
-- require('./ai-chatbot-service').onRealChatMessage(stream.id, {
-- username, message: mirrored.message, userId: null, anonId: null,
-- platform: 'rs', relayUsername: username, isStreamer: false, isMod: false, msgId: mirrored.id || null,
-- });
-- } catch { /* non-critical */ }
++ rsFollowUps(stream, username, rawUsername, mirrored.message, data.avatar, mirrored.id || null, 'ai');
// Also surface on the global / username-only overlay (tags stream_channel)
try { chatServer.forwardToGlobal(stream.id, mirrored); } catch { /* non-critical */ }
// And to viewers of the streamer's other live slots (cross-slot chat)
@@@ -704,27 +747,12 @@ class RobotStreamerService {
chatServer.synthesizeAndBroadcastTTS(stream.id, username, mirrored.message, null, 'rs', `rs:${username}`, null, mirrored.id ? `m${mirrored.id}` : null);
} catch { /* non-critical */ }
-- // ...and into the streamer's PowerChat overlay (relayed chat was never forwarded).
-- try {
-- const pc = require('./powerchat-platform');
-- if (stream.user_id && pc.slotRelayEnabled(stream.id)) {
-- pc.forwardChat(stream.user_id, {
-- chatterName: username,
-- externalChatterId: `rs:${username}`,
-- message: mirrored.message,
-- messageId: mirrored.id ? `ov-${mirrored.id}` : undefined,
-- avatarUrl: data.avatar || undefined,
-- // Placeholder letter from the REAL name — "[RS] name"
-- // would otherwise render "[" for every RS chatter.
-- avatarFallback: [...String(rawUsername || '')][0] || undefined,
-- });
-- }
-- } catch { /* non-critical */ }
++ rsFollowUps(stream, username, rawUsername, mirrored.message, data.avatar, mirrored.id || null, 'powerchat');
});
bridge.ws.on('close', () => {
if (bridge.stopped) return;
-- chatServer.broadcastToStream(stream.id, {
++ delivery.event({ kind: 'stream', id: stream.id }, {
type: 'system',
message: 'RobotStreamer chat mirror disconnected — retrying',
timestamp: new Date().toISOString(),
ddiff --git a/server/internal/routes.js b/server/internal/routes.js
iindex ecd2dafe..73487242 100644
---- a/server/internal/routes.js
++++ b/server/internal/routes.js
@@@ -12,7 +12,11 @@ const { guard } = require('../net/service-guard');
// CHAT_AUTHORITY=chat: OpenVibe.Chat caches users; a role or avatar pushed here reaches it at once.
function notifyChat(userId) {
-- try { const cs = require('../chat/chat-server'); if (cs.remote) cs.userChanged(userId); } catch { /* non-critical */ }
++ try {
++ const delivery = require('../chat/chat-delivery');
++ if (delivery.ingress()) { delivery.invalidate({ user: Number(userId) }); return; }
++ const cs = require('../chat/chat-server'); if (cs.remote) cs.userChanged(userId);
++ } catch { /* non-critical */ }
}
// Footer site copy is written by OpenVibe.AI for the Network directly (network.site_copy); the
ddiff --git a/server/media-proxy/clip-notify.js b/server/media-proxy/clip-notify.js
iindex 0f2febb4..beed6759 100644
---- a/server/media-proxy/clip-notify.js
++++ b/server/media-proxy/clip-notify.js
@@@ -67,16 +67,24 @@ async function _sendOne(clipId) {
clip: meta,
timestamp: new Date().toISOString(),
};
++ const line = {
++ stream_id: streamId,
++ channel_user_id: ownerId,
++ user_id: null,
++ username: creatorName,
++ message: `clipped: ${title}`,
++ message_type: 'clip',
++ metadata: meta,
++ };
try {
-- const saved = db.saveChatMessage({
-- stream_id: streamId,
-- channel_user_id: ownerId,
-- user_id: null,
-- username: creatorName,
-- message: `clipped: ${title}`,
-- message_type: 'clip',
-- metadata: meta,
-- });
++ const delivery = require('../chat/chat-delivery');
++ if (delivery.ingress()) {
++ // Chat persists and shows the line (one key per clip, so a re-announce cannot post twice).
++ await delivery.message({ ...line, key: `clip:${clipId}` });
++ db.markClipNotifiedState(clipId);
++ return;
++ }
++ const saved = db.saveChatMessage(line);
payload.id = saved && saved.lastInsertRowid;
require('../chat/chat-server').broadcastToChannelRoom(ownerId, streamId, payload);
} catch (e) {
ddiff --git a/server/media/media-queue.js b/server/media/media-queue.js
iindex ea8079bb..a394e412 100644
---- a/server/media/media-queue.js
++++ b/server/media/media-queue.js
@@@ -893,6 +893,9 @@ class MediaQueue {
broadcast(streamerId, payload) {
try {
++ const delivery = require('../chat/chat-delivery');
++ // Chat's ingress fans a media frame out to the owner's live streams itself.
++ if (delivery.ingress()) { delivery.event({ kind: 'owner-streams', id: streamerId }, payload); return; }
const chatServer = require('../chat/chat-server');
const streams = db.getLiveStreamsByUserId(streamerId) || [];
for (const stream of streams) {
ddiff --git a/server/monetization/alerts.js b/server/monetization/alerts.js
iindex fe294301..0c8273de 100644
---- a/server/monetization/alerts.js
++++ b/server/monetization/alerts.js
@@@ -18,6 +18,12 @@
*/
function playAlertSound(chatServer, streamerId, streamId, kind) {
try {
++ const delivery = require('../chat/chat-delivery');
++ if (delivery.ingress()) {
++ // LIVE_CHAT_INGRESS: the same request as an `alert` event on Chat's typed ingress.
++ delivery.event({ kind: 'channel', id: streamerId }, { type: 'alert', streamerId: Number(streamerId), streamId: streamId ? Number(streamId) : null, kind: kind === 'goal' ? 'goal' : 'donation' });
++ return;
++ }
if (!chatServer || typeof chatServer.playAlertSound !== 'function') return;
chatServer.playAlertSound(streamerId, streamId, kind === 'goal' ? 'goal' : 'donation');
} catch { /* non-critical */ }
ddiff --git a/server/monetization/coins-routes.js b/server/monetization/coins-routes.js
iindex b39bd969..081fca33 100644
---- a/server/monetization/coins-routes.js
++++ b/server/monetization/coins-routes.js
@@@ -220,9 +220,8 @@ router.post('/redeem', requireAuth, (req, res) => {
// Broadcast redemption to chat so streamer sees it
try {
-- const chatServer = require('../chat/chat-server');
const reward = result.redemption.reward;
-- chatServer.broadcastToStream(streamId, {
++ require('../chat/chat-delivery').event({ kind: 'stream', id: streamId }, {
type: 'redemption',
username: req.user.display_name || req.user.username,
reward_title: reward.title,
@@@ -231,7 +230,7 @@ router.post('/redeem', requireAuth, (req, res) => {
cost: reward.cost,
user_input: userInput || '',
timestamp: new Date().toISOString(),
-- });
++ }, { key: `redemption:${result.redemption.id}` });
} catch { /* chat broadcast optional */ }
// Feed the redemption into PowerChat as a virtual-currency event (alerts + leaderboard).
ddiff --git a/server/monetization/routes.js b/server/monetization/routes.js
iindex b332be23..6278e055 100644
---- a/server/monetization/routes.js
++++ b/server/monetization/routes.js
@@@ -15,6 +15,7 @@ const { requireAuth, requireAdmin } = require('../auth/auth');
const { requireOwner } = require('../auth/permissions');
const openvibeBucks = require('./vibes');
const db = require('../db/database');
++const delivery = require('../chat/chat-delivery');
// BILLING_AUTHORITY=billing: every money action below goes to OpenVibe.Billing instead of Live's
// columns (billing-actions.js); money_writes_frozen refuses them in both modes.
const money = require('./money-authority');
@@@ -100,19 +101,23 @@ router.post('/donate', requireAuth, money.guardWrite, async (req, res) => {
avatar_url: donorUser?.avatar_url || null,
amount: result.amount, message: message || '', timestamp: ts,
};
-- chatServer.broadcastToChannelRoom(streamer_id, stream_id || null, donationEvent);
-- // Tips are a site-wide event worth celebrating, so mirror them into global chat
-- // instead of confining them to the channel that received them.
-- try { chatServer.broadcastGlobal({ ...donationEvent, global: true, channel_user_id: streamer_id }); } catch { /* */ }
-- try {
-- db.saveChatMessage({
-- stream_id: stream_id || null, channel_user_id: streamer_id, user_id: req.user.id,
-- username: donor,
-- message: `${donor} donated ${result.amount.toLocaleString()} Vibes${message ? ': ' + message : ''}`,
-- message_type: 'donation',
-- metadata: { kind: 'donation', amount: result.amount, message: message || '', username: donor, user_id: req.user.id, avatar_url: donorUser?.avatar_url || null },
-- });
-- } catch { /* non-critical */ }
++ const donationLine = {
++ stream_id: stream_id || null, channel_user_id: streamer_id, user_id: req.user.id,
++ username: donor,
++ message: `${donor} donated ${result.amount.toLocaleString()} Vibes${message ? ': ' + message : ''}`,
++ message_type: 'donation',
++ metadata: { kind: 'donation', amount: result.amount, message: message || '', username: donor, user_id: req.user.id, avatar_url: donorUser?.avatar_url || null },
++ };
++ if (delivery.ingress()) {
++ // Chat persists the line and shows it in the channel and (mirror) global chat in one call.
++ delivery.message({ ...donationLine, mirror: true, key: result.transactionId ? `donation:${result.transactionId}` : undefined });
++ } else {
++ chatServer.broadcastToChannelRoom(streamer_id, stream_id || null, donationEvent);
++ // Tips are a site-wide event worth celebrating, so mirror them into global chat
++ // instead of confining them to the channel that received them.
++ try { chatServer.broadcastGlobal({ ...donationEvent, global: true, channel_user_id: streamer_id }); } catch { /* */ }
++ try { db.saveChatMessage(donationLine); } catch { /* non-critical */ }
++ }
// 2) Donation sound (streamer-configured).
alerts.playAlertSound(chatServer, streamer_id, stream_id, 'donation');
@@@ -132,24 +137,26 @@ router.post('/donate', requireAuth, money.guardWrite, async (req, res) => {
// 3) Live goal progress → widget.
if (result.goal) {
-- chatServer.broadcastToChannelRoom(streamer_id, stream_id || null, { type: 'goal-update', goal: publicGoal(result.goal) });
++ delivery.event({ kind: 'channel', id: streamer_id, stream: stream_id || null }, { type: 'goal-update', goal: publicGoal(result.goal) });
}
// 4) Goal reached → flashy animated chat event (persisted) + goal sound.
if (result.goalReached) {
const g = result.goalReached;
-- chatServer.broadcastToChannelRoom(streamer_id, stream_id || null, {
-- type: 'goal-reached', goal: publicGoal(g), by: donor, timestamp: ts,
-- });
-- try {
-- db.saveChatMessage({
-- stream_id: stream_id || null, channel_user_id: streamer_id, user_id: null,
-- username: 'Donation Goal',
-- message: `🎉 Goal reached: ${g.title} (${(g.target_amount || 0).toLocaleString()} Vibes)`,
-- message_type: 'donation',
-- metadata: { kind: 'goal-reached', goal_id: g.id, title: g.title, target: g.target_amount, image: g.image_url || null, media_type: g.media_type || null, by: donor },
++ const goalLine = {
++ stream_id: stream_id || null, channel_user_id: streamer_id, user_id: null,
++ username: 'Donation Goal',
++ message: `🎉 Goal reached: ${g.title} (${(g.target_amount || 0).toLocaleString()} Vibes)`,
++ message_type: 'donation',
++ metadata: { kind: 'goal-reached', goal_id: g.id, title: g.title, target: g.target_amount, image: g.image_url || null, media_type: g.media_type || null, by: donor },
++ };
++ if (delivery.ingress()) delivery.message({ ...goalLine, key: result.transactionId ? `goal:${result.transactionId}` : undefined });
++ else {
++ chatServer.broadcastToChannelRoom(streamer_id, stream_id || null, {
++ type: 'goal-reached', goal: publicGoal(g), by: donor, timestamp: ts,
});
-- } catch { /* non-critical */ }
++ try { db.saveChatMessage(goalLine); } catch { /* non-critical */ }
++ }
alerts.playAlertSound(chatServer, streamer_id, stream_id, 'goal');
}
@@@ -262,7 +269,7 @@ router.put('/goals/:id', requireAuth, (req, res) => {
// same as a donation does (a plain goal-update — no celebration).
if (req.body.current_amount !== undefined && g) {
try {
-- require('../chat/chat-server').broadcastToChannelRoom(req.user.id, null, { type: 'goal-update', goal: publicGoal(g) });
++ delivery.event({ kind: 'channel', id: req.user.id }, { type: 'goal-update', goal: publicGoal(g) });
} catch { /* live update is best-effort */ }
}
res.json({ goals: openvibeBucks.getManageGoals(req.user.id).map(publicGoal) });
ddiff --git a/server/news/news-service.js b/server/news/news-service.js
iindex c4015b60..c3859e57 100644
---- a/server/news/news-service.js
++++ b/server/news/news-service.js
@@@ -14,6 +14,7 @@
'use strict';
const db = require('../db/database');
++const delivery = require('../chat/chat-delivery');
const NewsApiSource = require('./sources/newsapi-source');
const RedditSource = require('./sources/reddit-source');
const RssSource = require('./sources/rss-source');
@@@ -244,7 +245,7 @@ class NewsService {
if (now - lastInject < MIN_INJECT_INTERVAL_MS) continue;
if (!this.isEnabledForStream(streamId)) continue;
-- this._chatServer.broadcastToStream(streamId, {
++ delivery.event({ kind: 'stream', id: streamId }, {
type: 'chat',
message_type: 'news',
username: '📰 Breaking News',
@@@ -254,7 +255,7 @@ class NewsService {
timestamp: new Date().toISOString(),
source_platform: 'news',
system: true,
-- });
++ }, { key: `news:${streamId}:${now}` });
this._lastInjectPerStream.set(streamId, now);
}
ddiff --git a/server/recap/recap.js b/server/recap/recap.js
iindex 72eb4ca4..d2448557 100644
---- a/server/recap/recap.js
++++ b/server/recap/recap.js
@@@ -199,11 +199,16 @@ function pending() {
function announce(recap) {
try {
-- const chat = require('../chat/chat-server');
-- const server = chat.chatServer || chat.default || chat;
-- if (!server || typeof server.broadcastToChannelRoom !== 'function') return false;
++ const delivery = require('../chat/chat-delivery');
const g = recap.write.grade;
-- server.broadcastToChannelRoom(recap.streamer.id, recap.stream.id, { type: 'system', message: `📋 After-show report for "${recap.stream.title}" is in — grade ${g}: ${recap.write.headline}. Read it: /recap/${recap.stream.id}` });
++ const frame = { type: 'system', message: `📋 After-show report for "${recap.stream.title}" is in — grade ${g}: ${recap.write.headline}. Read it: /recap/${recap.stream.id}` };
++ if (delivery.ingress()) delivery.event({ kind: 'channel', id: recap.streamer.id }, frame, { key: `recap:${recap.stream.id}` });
++ else {
++ const chat = require('../chat/chat-server');
++ const server = chat.chatServer || chat.default || chat;
++ if (!server || typeof server.broadcastToChannelRoom !== 'function') return false;
++ server.broadcastToChannelRoom(recap.streamer.id, recap.stream.id, frame);
++ }
db.run('UPDATE stream_recaps SET announced_at = CURRENT_TIMESTAMP WHERE stream_id = ?', [recap.stream.id]);
return true;
} catch (e) { console.warn('[Recap] announce:', e.message); return false; }
ddiff --git a/server/streaming/call-server.js b/server/streaming/call-server.js
iindex f5729f21..b7ef25ef 100644
---- a/server/streaming/call-server.js
++++ b/server/streaming/call-server.js
@@@ -138,7 +138,7 @@ class CallServer {
if (this._notifyTimer) return;
this._notifyTimer = setTimeout(() => {
this._notifyTimer = null;
-- try { chatServer.broadcastAll({ type: 'voice-channels', channels: this.listChannels(null) }); } catch { /* */ }
++ try { require('../chat/chat-delivery').event({ kind: 'all' }, { type: 'voice-channels', channels: this.listChannels(null) }); } catch { /* */ }
}, 400);
}
ddiff --git a/server/streaming/routes.js b/server/streaming/routes.js
iindex 355c2182..53fa4841 100644
---- a/server/streaming/routes.js
++++ b/server/streaming/routes.js
@@@ -1406,7 +1406,7 @@ router.post('/voice-channels/call-user', requireAuth, (req, res) => {
};
// Real-time invite for online users via existing chat WS connections.
-- chatServer.sendDm(targetUser.id, payload);
++ require('../chat/chat-delivery').event({ kind: 'user', id: targetUser.id }, payload);
// Persistent cross-site notification for offline users / later join.
pushNotification({
@@@ -1450,7 +1450,7 @@ router.post('/voice-channels/call-user/respond', requireAuth, (req, res) => {
if (!ch || ch.createdBy !== callerUserId || !callServer.hasInvite(channelId, req.user.id)) return res.status(403).json({ error: 'No such invite' });
const fromDisplayName = req.user.display_name || req.user.username || 'Someone';
-- chatServer.sendDm(callerUserId, {
++ require('../chat/chat-delivery').event({ kind: 'user', id: callerUserId }, {
type: 'vc-call-response',
status,
channelId,
ddiff --git a/server/tips/delivery-routes.js b/server/tips/delivery-routes.js
iindex cbd78c90..0b23bd52 100644
---- a/server/tips/delivery-routes.js
++++ b/server/tips/delivery-routes.js
@@@ -22,6 +22,7 @@
*/
const express = require('express');
const db = require('../db/database');
++const delivery = require('../chat/chat-delivery');
const { guard } = require('../net/service-guard');
const router = express.Router();
@@@ -125,20 +126,36 @@ router.post('/deliveries', guard('live.tips_delivery.write'), express.json({ lim
type: 'donation', username: name, user_id: null, avatar_url: null, amount, message, timestamp: new Date().toISOString(),
source: 'tips', paid_message: b.effect === 'paid_message', highlight_seconds: Number(b.highlight_seconds) || 0,
};
-- chatServer.broadcastToChannelRoom(userId, streamId, event);
-- try { chatServer.broadcastGlobal({ ...event, global: true, channel_user_id: userId }); } catch { /* non-critical */ }
-- const saved = db.saveChatMessage({
++ const line = {
stream_id: streamId, channel_user_id: userId, user_id: null, username: name,
message: String(b.text || `${name} ${amount == null ? 'sent a tip' : `tipped ${amount.toLocaleString()} Vibes`}${message ? ': ' + message : ''}`).slice(0, 1000),
message_type: 'donation',
metadata: { kind: 'donation', amount, message, username: name, source: 'tips', interaction_id: i.id || null, paid_message: event.paid_message, highlight_seconds: event.highlight_seconds, test: !!b.test },
-- });
++ };
++ let chatMessageId = null;
++ if (delivery.ingress()) {
++ // Chat persists the line and shows it in the channel and global chat; Tips' key is the operation's key.
++ chatMessageId = await delivery.message({ ...line, mirror: true, key: `tips:${key}` });
++ } else {
++ chatServer.broadcastToChannelRoom(userId, streamId, event);
++ try { chatServer.broadcastGlobal({ ...event, global: true, channel_user_id: userId }); } catch { /* non-critical */ }
++ const saved = db.saveChatMessage(line);
++ chatMessageId = saved && saved.lastInsertRowid != null ? Number(saved.lastInsertRowid) : null;
++ }
require('../monetization/alerts').playAlertSound(chatServer, userId, streamId, 'donation');
-- return res.json(remember(key, { ok: true, ref: { chat_message_id: saved && saved.lastInsertRowid != null ? Number(saved.lastInsertRowid) : null } }));
++ return res.json(remember(key, { ok: true, ref: { chat_message_id: chatMessageId } }));
}
if (b.effect === 'tts') {
const text = String((b.tts && b.tts.text) || '').slice(0, 1200);
if (!text) return res.status(422).json({ error: 'no text' });
++ if (delivery.ingress()) {
++ // Chat's ingress speaks a persisted line: a `tts` line in the channel, read aloud under the same keys.
++ const id = await delivery.message({
++ stream_id: streamId, channel_user_id: userId, username: name, message: text, message_type: 'tts', source_platform: 'tips',
++ tts: { identity_key: `tips:${i.id}`, key: `tips-${i.id}` }, key: `tips:${key}`,
++ });
++ return res.json(remember(key, { ok: true, ref: { stream_id: streamId, chat_message_id: id } }));
++ }
await chatServer.synthesizeAndBroadcastTTS(streamId, name, text, null, 'tips', `tips:${i.id}`, userId, `tips-${i.id}`);
return res.json(remember(key, { ok: true, ref: { stream_id: streamId } }));
}
ddiff --git a/server/vibe-coding/publish-server.js b/server/vibe-coding/publish-server.js
iindex 1ea0e7bf..c6f18b9a 100644
---- a/server/vibe-coding/publish-server.js
++++ b/server/vibe-coding/publish-server.js
@@@ -171,13 +171,13 @@ class VibeCodingPublishServer {
const settings = vibeService.getManagedStreamVibeCodingSettings(client.managedStreamId);
const projected = vibeService.projectViewerEvent(event, settings);
if (projected && liveStream?.id) {
-- this.chatServer.broadcastToStream(liveStream.id, {
++ require('../chat/chat-delivery').event({ kind: 'stream', id: liveStream.id }, {
type: 'vibe-coding',
managed_stream_id: client.managedStreamId,
slot_slug: client.slotSlug,
delay_ms: settings.delay_ms,
event: projected,
-- });
++ }, { key: event.eventId ? `vibe:${client.managedStreamId}:${event.eventId}` : undefined });
}
this.sendTo(ws, {
ddiff --git a/test/chat-ingress.test.js b/test/chat-ingress.test.js
nnew file mode 100644
iindex 00000000..d2492616
---- /dev/null
++++ b/test/chat-ingress.test.js
@@@ -0,0 +1,262 @@
++'use strict';
++
++// LIVE_CHAT_INGRESS=1 (T3 J2, first half): Live's chat producers call OpenVibe.Chat's typed
++// service-token ingress (/internal/chat/messages|events|moderation|invalidate, presence) through
++// server/chat/chat-delivery.js instead of the chat-remote.js bridge. Each producer family reaches
++// its endpoint with a bearer token and a stable idempotency key, nothing lands in
++// chat_bridge_outbox, and a Chat 5xx is logged and counted, never thrown and never bridged.
++// Against stub Network and Chat servers. The flag-off bridge path is test/chat-context.test.js.
++
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const http = require('http');
++const crypto = require('crypto');
++const jwt = require('jsonwebtoken');
++const { WebSocketServer } = require('ws');
++const { serviceAuth } = require('openvibe-contracts');
++
++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-chatingress-'));
++const keys = crypto.generateKeyPairSync('rsa', { modulusLength: 2048, publicKeyEncoding: { type: 'spki', format: 'pem' }, privateKeyEncoding: { type: 'pkcs8', format: 'pem' } });
++fs.writeFileSync(path.join(tmp, 'network.pem'), keys.publicKey);
++process.env.DB_PATH = path.join(tmp, 'live.db');
++process.env.OV_NETWORK_PUBLIC_KEY = path.join(tmp, 'network.pem');
++process.env.OV_NETWORK_URL = 'https://openvibe.network';
++process.env.OV_OAUTH_CLIENT_ID = 'live';
++process.env.OV_OAUTH_CLIENT_SECRET = 'live-secret';
++process.env.CHAT_AUTHORITY = 'chat';
++process.env.LIVE_CHAT_INGRESS = '1';
++const quiet = console.log;
++console.log = () => {};
++console.warn = () => {};
++
++const ISS = 'https://openvibe.network';
++const now = () => Math.floor(Date.now() / 1000);
++const CAPS = ['chat.message.send', 'chat.event.publish', 'chat.moderation.write', 'chat.cache.invalidate', 'chat.presence.read'];
++
++const network = http.createServer((req, res) => {
++ let raw = '';
++ req.on('data', (c) => { raw += c; });
++ req.on('end', () => {
++ res.setHeader('Content-Type', 'application/json');
++ if (req.url !== '/oauth/token') { res.statusCode = 404; return res.end('{}'); }
++ const f = new URLSearchParams(raw);
++ const token = serviceAuth.signServiceToken({ iss: ISS, sub: 'svc:live', actor_type: 'service', aud: [f.get('audience')], cap: CAPS, iat: now(), exp: now() + 300, jti: `tok_${crypto.randomUUID()}` }, keys.privateKey);
++ res.end(JSON.stringify({ access_token: token, expires_in: 300 }));
++ });
++});
++
++// Stub Chat: records every ingress call; `fail` answers chosen families with a status.
++const calls = [];
++const bridgeCalls = [];
++const fail = {};
++let nextId = 5000;
++const chat = http.createServer((req, res) => {
++ let raw = '';
++ req.on('data', (c) => { raw += c; });
++ req.on('end', () => {
++ res.setHeader('Content-Type', 'application/json');
++ const auth = String(req.headers.authorization || '');
++ if (req.url.startsWith('/internal/live/')) {
++ if (req.url === '/internal/live/calls') bridgeCalls.push(JSON.parse(raw));
++ return res.end(JSON.stringify({ ok: true, results: [] }));
++ }
++ const m = req.url.match(/^\/internal\/chat\/(messages|events|moderation|invalidate|presence)$/);
++ if (!m) { res.statusCode = 404; return res.end('{}'); }
++ const family = m[1];
++ if (family === 'presence') return res.end(JSON.stringify({ total: 4, streams: { 1: 2 }, slow_mode: {}, users: [{ user_id: 99, ip: '198.51.100.9', stream_id: 1 }], anons: [] }));
++ const body = JSON.parse(raw || '{}');
++ calls.push({ family, auth, body });
++ const f = fail[family];
++ if (f && f.times > 0) { f.times--; res.statusCode = f.status; return res.end(JSON.stringify({ error: 'down' })); }
++ if (family === 'messages') return res.end(JSON.stringify({ ok: true, id: ++nextId }));
++ if (family === 'moderation' && /^delete-/.test(body.action)) return res.end(JSON.stringify({ ok: true, ids: [Number(body.id) || 1] }));
++ res.end(JSON.stringify({ ok: true }));
++ });
++});
++
++const listen = (s) => new Promise((r) => s.listen(0, '127.0.0.1', () => r(s.address().port)));
++const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
++async function waitFor(pred, what, ms = 3000) {
++ const end = Date.now() + ms;
++ while (Date.now() < end) { const v = pred(); if (v) return v; await sleep(10); }
++ throw new Error(`timed out waiting for ${what}`);
++}
++const KEY_RE = /^live:[a-z]+:[A-Za-z0-9:._-]+$/;
++const find = (family, pred) => calls.find((c) => c.family === family && pred(c.body));
++function assertSigned(c, what) {
++ assert.ok(c, `${what} reached Chat`);
++ assert.ok(/^Bearer \S+$/.test(c.auth), `${what} carries a bearer token`);
++ assert.ok(KEY_RE.test(c.body.key) && c.body.key.length <= 160, `${what} carries an idempotency key (${c.body.key})`);
++}
++
++(async () => {
++ process.env.OV_NETWORK_INTERNAL_URL = `http://127.0.0.1:${await listen(network)}`;
++ process.env.OV_CHAT_INTERNAL_URL = `http://127.0.0.1:${await listen(chat)}`;
++
++ const db = require('../server/db/database');
++ db.initDb();
++ const d = db.getDb();
++ const mkUser = (username, role = 'user') => {
++ const id = Number(db.createUser({ username, email: `${username}@example.test`, password_hash: '!x', display_name: username.toUpperCase(), stream_key: `key-${username}` }).lastInsertRowid);
++ d.prepare('UPDATE users SET role = ? WHERE id = ?').run(role, id);
++ return id;
++ };
++ const admin = mkUser('admin2', 'admin');
++ const streamer = mkUser('streamer', 'streamer');
++ const viewer = mkUser('viewer');
++ db.createChannel({ user_id: streamer, title: 'Streamer TV' });
++ const channel = db.getChannelByUserId(streamer);
++ const streamId = Number(db.createStream({ user_id: streamer, channel_id: channel.id, title: 'Live now' }).lastInsertRowid);
++ const outbox = () => d.prepare('SELECT COUNT(*) AS n FROM chat_bridge_outbox').get().n;
++
++ const delivery = require('../server/chat/chat-delivery');
++ const chatServer = require('../server/chat/chat-server');
++ assert.strictEqual(chatServer.remote, true);
++ assert.strictEqual(delivery.ingress(), true, 'LIVE_CHAT_INGRESS=1 with CHAT_AUTHORITY=chat');
++ chatServer.init();
++ delivery.client._setRetryMs([20, 20]);
++
++ let exit = 0;
++ let rs = null;
++ try {
++ // 1. AI viewer line: one /messages call (persist + mirror + TTS); the caller gets no placeholder id.
++ const poster = require('../server/ai/viewers/poster');
++ const ret = poster.post({ streamId, userId: streamer, settings: { powerchat_forward: false } }, { id: 1, username: 'Botty', persona_json: '{}' }, 'hello from a bot');
++ assert.strictEqual(ret, null, 'no placeholder id with the flag on');
++ const ai = await waitFor(() => find('messages', (b) => b.source_platform === 'ai'), 'AI line');
++ assertSigned(ai, 'AI line');
++ assert.deepStrictEqual([ai.body.stream_id, ai.body.channel_user_id, ai.body.username, ai.body.mirror, ai.body.is_global, ai.body.frame.is_ai, ai.body.tts.identity_key],
++ [streamId, streamer, 'Botty', true, false, true, 'aibot:botty']);
++
++ // 2. Donation line + alert sound (PowerChat test tip) and a direct alert.
++ require('../server/integrations/powerchat-webhook').simulateDonation(streamer, { amountUsd: 2, donor: 'Tipper', message: 'gg' });
++ const tip = await waitFor(() => find('messages', (b) => b.message_type === 'donation' && b.username === 'Tipper'), 'donation line');
++ assertSigned(tip, 'donation line');
++ assert.strictEqual(tip.body.metadata.amount, 200);
++ const alert = await waitFor(() => find('events', (b) => b.frame.type === 'alert'), 'alert');
++ assertSigned(alert, 'alert');
++ assert.deepStrictEqual(alert.body.target, { kind: 'channel', id: streamer }, 'an alert targets its channel owner');
++ assert.deepStrictEqual([alert.body.frame.streamerId, alert.body.frame.streamId, alert.body.frame.kind], [streamer, streamId, 'donation']);
++
++ // 3. Relayed chat (Twitch): the line, the relay-user record, the welcome card; Live's own first-chat copy is kept.
++ const relay = require('../server/integrations/chat-relay-service');
++ relay._broadcastMessage({ platform: 'twitch', streamId }, 'alice', 'hi from twitch', {});
++ const rl = await waitFor(() => find('messages', (b) => b.source_platform === 'twitch'), 'relay line');
++ assertSigned(rl, 'relay line');
++ assert.deepStrictEqual([rl.body.username, rl.body.frame.role, rl.body.mirror, rl.body.tts.identity_key], ['[Twitch] alice', 'external', true, 'twitch:[Twitch] alice']);
++ assertSigned(await waitFor(() => find('moderation', (b) => b.action === 'relay-record' && b.username === 'alice'), 'relay record'), 'relay record');
++ assertSigned(await waitFor(() => find('events', (b) => b.frame.type === 'system' && /Welcome alice/.test(b.frame.message)), 'welcome'), 'welcome');
++ assert.ok(d.prepare("SELECT 1 FROM stream_first_chats WHERE chatter_key = '[Twitch] alice' OR chatter_key = 'ext:[Twitch] alice'").get(), 'Live keeps its first-chat copy');
++
++ // 4. RobotStreamer mirror: a stub RS chat socket sends one line.
++ const wss = new WebSocketServer({ port: 0, host: '127.0.0.1' });
++ await new Promise((r) => wss.on('listening', r));
++ rs = require('../server/integrations/robotstreamer-service');
++ rs.getIntegrationForStream = () => ({ enabled: 1, token: 'rs-token', robot_id: '42', owner_id: '7', mirror_chat: 1, chat_url: `ws://127.0.0.1:${wss.address().port}` });
++ wss.on('connection', (ws) => ws.on('message', () => ws.send(JSON.stringify({ robot_id: '42', username: 'Goose', message: 'beep boop' }))));
++ await rs.startForStream({ id: streamId, user_id: streamer, channel_id: channel.id, protocol: 'rtmp' });
++ const robot = await waitFor(() => find('messages', (b) => b.source_platform === 'rs'), 'RobotStreamer line');
++ assertSigned(robot, 'RobotStreamer line');
++ assert.deepStrictEqual([robot.body.username, robot.body.message, robot.body.tts.identity_key], ['[RS] Goose', 'beep boop', 'rs:[RS] Goose']);
++ rs.stopChatBridge(streamId);
++ wss.close();
++
++ // 5. Moderation through /api/mod: a message delete and a site ban (disconnect + log), as an admin.
++ const express = require('express');
++ const app = express();
++ app.use(express.json());
++ app.use('/api/mod', require('../server/admin/mod-routes'));
++ const port = await listen(http.createServer(app));
++ const session = jwt.sign({ sub: '601', username: 'admin2', role: 'admin' }, keys.privateKey, { algorithm: 'RS256', issuer: ISS, expiresIn: 300 });
++ const mod = async (p, body) => {
++ const r = await fetch(`http://127.0.0.1:${port}/api/mod${p}`, { method: 'POST', headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${session}` }, body: JSON.stringify(body) });
++ return { status: r.status, body: await r.json().catch(() => null) };
++ };
++ d.prepare("INSERT INTO chat_messages (id, stream_id, channel_user_id, user_id, username, message) VALUES (900010, ?, ?, ?, 'VIEWER', 'bad')").run(streamId, streamer, viewer);
++ assert.strictEqual((await mod('/delete-message', { message_id: 900010 })).status, 200);
++ const del = find('moderation', (b) => b.action === 'delete-message');
++ assertSigned(del, 'message delete');
++ assert.deepStrictEqual([del.body.id, del.body.deleted_by, del.body.key], [900010, admin, 'live:moderation:delete:900010']);
++ assert.strictEqual(db.getChatMessageById(900010).is_deleted, 1, "Live's copy is deleted too");
++ assertSigned(await waitFor(() => find('moderation', (b) => b.action === 'log' && b.action_type === 'message_delete'), 'delete log'), 'delete log');
++ const ban = await mod(`/users/${viewer}/ban`, { reason: 'spam' });
++ assert.ok(ban.status < 300, `ban answered ${ban.status}`);
++ const disc = await waitFor(() => find('moderation', (b) => b.action === 'disconnect' && b.user_id === viewer), 'ban disconnect');
++ assertSigned(disc, 'ban disconnect');
++ assertSigned(await waitFor(() => find('moderation', (b) => b.action === 'log' && b.target_user_id === viewer && b.action_type !== 'message_delete'), 'ban log'), 'ban log');
++
++ // 5b. A news headline: one keyed /events card to the stream.
++ const news = require('../server/news/news-service');
++ news._chatServer = chatServer;
++ news._getActiveStreamIds = () => [streamId];
++ news.isEnabledForStream = () => true;
++ news._pendingQueue.push({ headline: 'Big news', sourceId: 'test' });
++ news._processQueue();
++ const nw = await waitFor(() => find('events', (b) => b.frame.message_type === 'news'), 'news card');
++ assertSigned(nw, 'news card');
++ assert.deepStrictEqual(nw.body.target, { kind: 'stream', id: streamId });
++ assert.ok(/^live:events:news:\d+:\d+$/.test(nw.body.key), `news key ${nw.body.key}`);
++
++ // 6. Deploy notice: no ingress endpoint, Events off → left unannounced (next boot retries), never bridged.
++ const before = db.getSetting('deploy_last_announced');
++ const dn = await require('../server/chat/deploy-notice').announce({ db, chatServer, log: { warn() {}, log() {}, info() {} } });
++ assert.strictEqual(dn.announced, 0);
++ assert.strictEqual(db.getSetting('deploy_last_announced'), before, 'not recorded as announced');
++
++ // 7. Cache hints: an IP approval Live wrote, and an account change pushed by Network.
++ db.approveIp(channel.id, '203.0.113.5', admin, 'manual');
++ assertSigned(await waitFor(() => find('invalidate', (b) => b.approvals === channel.id), 'approvals hint'), 'approvals hint');
++ delivery.invalidate({ user: viewer });
++ assertSigned(await waitFor(() => find('invalidate', (b) => b.user === viewer), 'user hint'), 'user hint');
++
++ // 8. Presence comes from /internal/chat/presence.
++ await waitFor(() => chatServer.getTotalConnections() === 4, 'presence snapshot');
++ assert.strictEqual(chatServer.getConnectedUserIp(99), '198.51.100.9');
++
++ // 9. A Chat 5xx: retried with the same key, then given up; never thrown, never bridged.
++ fail.events = { status: 503, times: 1 };
++ const keyed = await delivery.event({ kind: 'stream', id: streamId }, { type: 'system', message: 'retry me' }, { key: 'retry-test' });
++ assert.deepStrictEqual(keyed, { ok: true }, 'the retry after a 503 delivered it');
++ const tries = calls.filter((c) => c.family === 'events' && c.body.frame.message === 'retry me');
++ assert.deepStrictEqual(tries.map((c) => c.body.key), ['live:events:retry-test', 'live:events:retry-test'], 'a retry reuses the key');
++ fail.messages = { status: 500, times: 99 };
++ const failedBefore = delivery.client.stats.failed;
++ let threw = null;
++ try { require('../server/integrations/powerchat-webhook').simulateDonation(streamer, { donor: 'Unlucky' }); } catch (err) { threw = err; }
++ assert.strictEqual(threw, null, 'a Chat 5xx is not thrown into the caller');
++ await waitFor(() => delivery.client.stats.failed > failedBefore, 'the failure to be counted');
++ const lost = calls.filter((c) => c.family === 'messages' && c.body.username === 'Unlucky');
++ assert.strictEqual(lost.length, 3, 'tried three times');
++ assert.ok(lost.every((c) => c.body.key === lost[0].body.key), 'with one key');
++ fail.messages = null;
++ fail.moderation = { status: 400, times: 1 };
++ assert.strictEqual(await delivery.moderate('log', { action_type: 'x' }), null, 'a 4xx resolves null');
++ assert.strictEqual(calls.filter((c) => c.family === 'moderation' && c.body.action_type === 'x').length, 1, 'a 4xx is not retried');
++
++ // 10. Nothing went over the bridge.
++ await chatServer.flush();
++ assert.strictEqual(outbox(), 0, 'nothing written to chat_bridge_outbox');
++ assert.deepStrictEqual(bridgeCalls.flatMap((c) => c.ops || []).map((o) => o.op), [], 'no bridge calls');
++
++ // 11. Flag off: the same seam is the legacy path again.
++ process.env.LIVE_CHAT_INGRESS = '';
++ assert.strictEqual(delivery.ingress(), false);
++ chatServer.broadcastToStream(streamId, { type: 'system', message: 'x' });
++ delivery.event({ kind: 'stream', id: streamId }, { type: 'system', message: 'legacy' });
++ await chatServer.flush();
++ assert.deepStrictEqual(bridgeCalls.flatMap((c) => c.ops || []).map((o) => o.op), ['broadcastToStream', 'broadcastToStream']);
++ chatServer.close();
++
++ quiet('chat ingress (LIVE_CHAT_INGRESS=1): all checks passed');
++ } catch (err) {
++ console.error(err);
++ exit = 1;
++ } finally {
++ try { if (rs) rs.stopChatBridge(streamId); } catch { /* */ }
++ try { db.close(); } catch { /* */ }
++ fs.rmSync(tmp, { recursive: true, force: true });
++ process.exit(exit);
++ }
++})();
ddiff --git a/test/security-ssrf.test.js b/test/security-ssrf.test.js
iindex 660bb256..95915b5c 100644
---- a/test/security-ssrf.test.js
++++ b/test/security-ssrf.test.js
@@@ -349,6 +349,7 @@ async function check(name, fn) {
'server/chat/chat-remote.js': 'OpenVibe.Chat (configured)',
'server/chat/moderation-client.js': 'OpenVibe.Chat internal read API (configured)',
'server/chat/insight-client.js': 'OpenVibe.Chat public chat-AI reads (configured OV_CHAT_INTERNAL_URL, fixed paths, ids validated)',
++ 'server/chat/chat-client.js': 'OpenVibe.Chat typed ingress (configured OV_CHAT_INTERNAL_URL, fixed /internal/chat/* paths)',
'server/chat/chat-server.js': 'OpenVibe.Tools anon resolve (configured)',
'server/streaming/calls-authority.js': 'OpenVibe.Chat (configured)',
'server/streaming/golive-notify.js': 'OpenVibe.Network (configured)',