openvibers/openvibe.chat Commits

Bots on /ws/chat: a !word that is not a sound presses the channel robot's button of that name, a ?token= socket is told how to sign in, and a new Go Live moves sockets left on the ended stream (#194)

OP OpenVibers committed 17 days ago · c22ae60beb · 4 files, +151 −5

Changes

ddiff --git a/README.md b/README.md
iindex 9375fd4..a25d07f 100644
---- a/README.md
++++ b/README.md
@@@ -32,6 +32,13 @@ before changing behaviour.** Browser JavaScript does not change; nginx routes th
DMs (`dm.js`, `/api/dm/*`), chat history (`history-store.js`, `/api/chat/*`), TTS (`tts-engine.js`,
`/api/tts/*`), 101soundboards and channel sounds (`/api/sounds*`), moderation utils and the word
filter, the deploy-notice card. Changes are confined to where they touched Live's data.
++- **For bots on `/ws/chat`** (Maticus, 2026-10-26): a `!word` that is not a channel sound is sent to Live as
++ `button:<word>` for a signed-in viewer, so it presses the channel robot's button of that name, and says nothing when
++ there is none. A token in the socket's address (`?token=`) is never read; the socket gets a `system` frame with
++ `code: "token_in_url"` saying to send an `Authorization: Bearer` header or the join's `token`. When a channel goes
++ live again, sockets still on one of its ended streams (same slot) move to the new stream and get
++ `{ type: "stream_changed", stream_id, previous_stream_id, channel_user_id }`, so a bot stays in the viewer list,
++ TTS and sounds without re-joining (`live-context.onStreamLive`, from the active-streams sync).
- **The TTS and sound queue** (`audio-queue.js`, table `audio_requests`). TTS, channel `!sounds` and
101soundboards clips are persisted requests played one at a time per room, across every Chat
process (a claim holds the room's advisory lock; a partial unique index allows one `playing` row
ddiff --git a/server/chat/chat-server.js b/server/chat/chat-server.js
iindex 7ae4c8e..cb9bc5d 100644
---- a/server/chat/chat-server.js
++++ b/server/chat/chat-server.js
@@@ -105,6 +105,7 @@ class ChatServer {
/** @type {WeakMap<object, Set<string>|null>} frame → the subjects who blocked its author (_blockersOf) */
this._blockMemo = new WeakMap();
ctx.onChannelSettings(async (channelId, settings) => await this._channelSettingsSeen(channelId, settings));
++ ctx.onStreamLive(async (stream) => await this._followNewStream(stream));
this.heartbeatInterval = null;
// TTS and sound requests queue in Chat's database (./audio-queue.js), one room at a time.
/** @type {Map<string, number>} `${streamId}:${userKey}` → last soundboard trigger */
@@@ -296,6 +297,9 @@ class ChatServer {
// A bot's bearer; a browser's socket starts anonymous and signs in with its join.
const token = extractWsToken(req);
const streamId = parseInt(urlParams.get('stream')) || null;
++ // A token in the address is never read (addresses land in logs and proxies); a bot that put one there used to
++ // sit anonymous with no word why (Maticus, 2026-10-26). It is told how, once the socket is set up.
++ const tokenInUrl = !token && (urlParams.has('token') || urlParams.has('access_token'));
ws.isAlive = true;
ws.on('pong', () => {
@@@ -322,6 +326,9 @@ class ChatServer {
ws.off('error', onEarlyClose);
if (closedEarly || ws.readyState !== WebSocket.OPEN) return;
this._registerConnection(ws, req, { ip, streamId, user, early, tokenIat: user ? session.tokenIat(token) : null });
++ if (tokenInUrl) {
++ this.sendTo(ws, { type: 'system', code: 'token_in_url', message: 'A token in the address is ignored, so this socket is not signed in. Send it as an "Authorization: Bearer" header when connecting, or as "token" in your join message: https://openvibe.live/docs/api-tokens#websocket-chat' });
++ }
})().catch((err) => {
console.warn('[Chat] connection setup failed:', err.message);
try { ws.close(1011, 'setup failed'); } catch {}
@@@ -1555,7 +1562,10 @@ class ChatServer {
default:
// Unknown !command → try a per-channel viewer-uploaded sound clip.
// Forward trailing tokens (e.g. "500p", "0.5") as pitch/speed args.
-- await this.triggerChannelSound(ws, client, stream, cmd.slice(1), parts.slice(1));
++ // Not a sound: maybe a button on the channel's robot (!ir_light, !cycle_gear …), pressed quietly.
++ if ((await this.triggerChannelSound(ws, client, stream, cmd.slice(1), parts.slice(1))) === false) {
++ this._robotButton(stream, cmd.slice(1), client);
++ }
return;
}
} catch (err) {
@@@ -1563,6 +1573,40 @@ class ChatServer {
}
}
++ /**
++ * A channel went live with a new stream: sockets still on one of its ended streams (a bot that joined
++ * ?stream=<id> last time, a popout) move onto the new one and are told (`stream_changed`), so they are in its
++ * viewer list, its TTS and sounds, without checking and re-joining (Maticus, 2026-10-26: "every Go Live starts a
++ * new session id; bots drop off the guest list"). A socket on another stream that is still live (a second slot)
++ * stays, and so does one on a different slot of the channel.
++ */
++ async _followNewStream(stream) {
++ const to = Number(stream && stream.id);
++ const channel = Number(stream && stream.user_id);
++ if (!to || !channel) return;
++ const ended = new Map(); // stream id → may move (ended, and the same slot or no slot)
++ const moved = [];
++ for (const [ws, c] of this.clients) {
++ if (!c.streamId || Number(c.streamId) === to || Number(c.channelUserId) !== channel) continue;
++ if (!ended.has(c.streamId)) {
++ const old = await ctx.getStreamById(c.streamId).catch(() => null);
++ const sameSlot = !old || !old.managed_stream_id || !stream.managed_stream_id || Number(old.managed_stream_id) === Number(stream.managed_stream_id);
++ ended.set(c.streamId, !!old && !old.is_live && sameSlot);
++ }
++ if (ended.get(c.streamId)) moved.push([ws, c, c.streamId]);
++ }
++ if (!moved.length) return;
++ await ctx.warm({ user: null, streamId: to, channelUserId: channel }).catch(() => {});
++ for (const [ws, c, from] of moved) {
++ c.streamId = to;
++ c._modStream = null;
++ this.sendTo(ws, { type: 'stream_changed', stream_id: to, previous_stream_id: from, channel_user_id: channel });
++ }
++ audioQueue.roomJoined(audioQueue.roomKey({ streamId: to, channelUserId: null }));
++ for (const from of new Set(moved.map((m) => m[2]))) this.broadcastUserCount(from);
++ this.broadcastUserCount(to);
++ }
++
/**
* A robot command for the channel's OpenVibe.Bot robot, through Live (it forwards to Bot as this viewer, plan T15).
* The viewer is named by Network subject (none when not signed in: Bot only takes commands from people). Live's
@@@ -1579,17 +1623,31 @@ class ChatServer {
}
}
++ /**
++ * A `!word` that is not a channel sound, pressed as the button of that name on the channel's OpenVibe.Bot robot
++ * (Live: `button:<word>`; a robot made of buttons, Maticus 2026-10-26: "!ir_light does nothing"). Signed-in viewers
++ * only, a plain word only, and always quiet: most channels have no robot and most words are no button, so a
++ * refusal says nothing (the eight movement words and !say keep their own replies).
++ */
++ _robotButton(stream, word, client) {
++ if (!client.user || !stream?.user_id || !/^[a-z0-9][a-z0-9_.-]{0,39}$/.test(word)) return;
++ this._subjectOfUser(client.user)
++ .then((subject) => ctx.effects.hardwareCommand(stream.user_id, `button:${word}`, client.user.display_name || client.user.username || '', { subject, userId: client.user.id || null }))
++ .catch(() => { /* quiet by design */ });
++ }
++
/**
* Play a per-channel viewer-uploaded sound clip (triggered by !command).
* Silent no-op when the command is not a registered sound so unknown
-- * commands don't spam the chat.
++ * commands don't spam the chat. Resolves false only when the word is not a sound (the caller may try a robot
++ * button).
*/
async triggerChannelSound(ws, client, stream, command, args = [], relay = null) {
try {
-- if (!stream?.user_id || !command) return;
++ if (!stream?.user_id || !command) return false;
const cmd = String(command).toLowerCase();
const sound = await db.getChannelSoundByCommand(stream.user_id, cmd);
-- if (!sound) return; // not a sound command — stay silent
++ if (!sound) return false; // not a sound command — stay silent
const chatSettings = await this._getChannelChatSettings(client.streamId);
if (chatSettings.custom_sounds_enabled === 0) {
ddiff --git a/server/live-context.js b/server/live-context.js
iindex e7bd359..d9d1a45 100644
---- a/server/live-context.js
++++ b/server/live-context.js
@@@ -761,6 +761,13 @@ async function syncTable(kind, table, cols, { full = false } = {}) {
}
return total;
}
++// Everyone who wants to know a stream went live here (the chat server moves sockets left on the channel's ended
++// stream onto it). Called once per stream in this process, when Live's active list first has it live (the table
++// sync may have stored it live already, so the stored flag cannot tell).
++const _liveListeners = new Set();
++const _announcedLive = new Set();
++function onStreamLive(fn) { _liveListeners.add(fn); return () => _liveListeners.delete(fn); }
++
async function syncActiveStreams() {
const data = await read('/streams/active');
await upsertRows('ctx_streams', STREAM_COLS, data.rows || []);
@@@ -770,6 +777,14 @@ async function syncActiveStreams() {
for (const r of await db.all('SELECT id FROM ctx_streams WHERE is_live = 1')) {
if (!liveNow.has(Number(r.id))) await db.run('UPDATE ctx_streams SET is_live = 0 WHERE id = ?', [r.id]);
}
++ for (const id of _announcedLive) if (!liveNow.has(id)) _announcedLive.delete(id);
++ for (const r of data.rows || []) {
++ if (!r.is_live || _announcedLive.has(Number(r.id))) continue;
++ _announcedLive.add(Number(r.id));
++ for (const fn of _liveListeners) {
++ try { await fn(r); } catch (err) { console.warn('[LiveContext] stream live listener:', err.message); }
++ }
++ }
}
const SCHEDULE = [
@@@ -955,7 +970,7 @@ module.exports = {
emoteSources, setChannelEmoteSources,
// policy
getChannelModerationSettings, isChannelModerator, channelLanguage, getChannelAlertSoundsByUser, ensurePolicy, invalidateChannel,
-- reloadPolicy, onChannelSettings, defaultModerationSettings,
++ reloadPolicy, onChannelSettings, onStreamLive, defaultModerationSettings,
// subscriptions (sub-only chat)
isSubscriber, subscriberState, ensureSubscriber,
// bans, follows, approvals, decor
ddiff --git a/test/ws-protocol.test.js b/test/ws-protocol.test.js
iindex d98b3a4..7de37c8 100644
---- a/test/ws-protocol.test.js
++++ b/test/ws-protocol.test.js
@@@ -334,6 +334,45 @@ t('media commands go to Live’s queue with the same replies', async () => {
await aliceWs.next((x) => x.type === 'system' && x.message === `No robot here (asked as usr_01J9AAAAAAAAAAAAAAAAAAAAAA, user ${alice.id}).`);
});
++t('a !word that is not a channel sound is pressed as the robot button of that name, quietly', async () => {
++ await h.sleep(1100);
++ const before = h.live.effects.length;
++ const hardware = () => h.live.effects.slice(before).filter((e) => e.name === 'hardware');
++ const seen = aliceWs.all.length;
++ aliceWs.sendJson({ type: 'chat', message: '!ir_light' });
++ const deadline = Date.now() + 5000;
++ while (!hardware().length && Date.now() < deadline) await h.sleep(50);
++ assert.deepStrictEqual(hardware().map((e) => [e.body.command, e.body.from_subject, e.body.from_user_id]), [['button:ir_light', 'usr_01J9AAAAAAAAAAAAAAAAAAAAAA', alice.id]]);
++ // Live's "no robot" is not said: most channels have no robot and most words are no button.
++ await h.sleep(300);
++ assert.ok(!aliceWs.all.slice(seen).some((m) => m.type === 'system' && /No robot here/.test(m.message)));
++ // Not a plain word, or a viewer who is not signed in: nothing is asked.
++ await h.sleep(1100);
++ aliceWs.sendJson({ type: 'chat', message: '!Not/A_Button' });
++ const anon = await h.ws({ ip: '198.51.100.71', stream: streamId });
++ anon.sendJson({ type: 'join', streamId });
++ await anon.next((m) => m.type === 'auth');
++ anon.sendJson({ type: 'chat', message: '!cycle_gear' });
++ await h.sleep(600);
++ assert.strictEqual(hardware().length, 1);
++ anon.close();
++});
++
++t('a token in the socket address is not read, and the socket says how to send it', async () => {
++ const ws = await h.ws({ ip: '198.51.100.72', query: `?token=${alice.token}` });
++ const told = await ws.next((m) => m.type === 'system' && m.code === 'token_in_url');
++ assert.match(told.message, /Authorization: Bearer/);
++ assert.ok(!told.message.includes(alice.token), 'the token is never echoed');
++ ws.sendJson({ type: 'join' });
++ assert.strictEqual((await ws.next((m) => m.type === 'auth')).authenticated, false, 'still anonymous');
++ ws.close();
++ const fine = await h.ws({ ip: '198.51.100.73', token: alice.token });
++ fine.sendJson({ type: 'join' });
++ assert.strictEqual((await fine.next((m) => m.type === 'auth')).authenticated, true);
++ assert.ok(await fine.none((m) => m.code === 'token_in_url'), 'a header token gets no notice');
++ fine.close();
++});
++
t('upgrade guards: foreign origin refused, banned address refused (admins exempt)', async () => {
assert.strictEqual(await h.wsRefused({ origin: 'https://evil.example' }), true);
h.live.addBan({ ip_address: '192.0.2.0/24' });
@@@ -420,4 +459,31 @@ t('the same line again and again is refused after three in 30 s (punctuation cou
mws.close();
});
++t('a new Go Live moves sockets left on the channel\'s ended stream onto the new one, and says so', async () => {
++ const gus = h.addUser('gus', { role: 'streamer' });
++ const gusChannel = h.addChannel(gus.id);
++ const slot = { id: 81, slug: 'gus-main', title: 'Main' };
++ const first = h.addStream(gus.id, gusChannel, { managed: slot });
++ const other = h.addStream(gus.id, gusChannel, { managed: { id: 82, slug: 'gus-two', title: 'Second' } });
++ await h.ctx.sync();
++ const join = async (ip, stream) => {
++ const ws = await h.ws({ ip, token: bob.token, stream });
++ ws.sendJson({ type: 'join', streamId: stream, token: bob.token });
++ await ws.next((m) => m.type === 'auth');
++ return ws;
++ };
++ const bot = await join('198.51.100.74', first);
++ const secondSlot = await join('198.51.100.75', other);
++ // The first session ends; the same slot goes live again with a new stream id.
++ h.live.streams.get(first).is_live = 0;
++ const second = h.addStream(gus.id, gusChannel, { managed: slot });
++ await h.ctx.sync();
++ const told = await bot.next((m) => m.type === 'stream_changed');
++ assert.deepStrictEqual([told.stream_id, told.previous_stream_id, told.channel_user_id], [second, first, gus.id]);
++ const client = [...h.chatServer.clients.values()].find((c) => c.ip && c.ip.endsWith('100.74'));
++ assert.strictEqual(client.streamId, second, 'now in the new stream\'s room');
++ assert.ok(await secondSlot.none((m) => m.type === 'stream_changed'), 'a still-live second slot keeps its viewers');
++ bot.close(); secondSlot.close();
++});
++
t.run(async () => { for (const w of [aliceWs, bobWs, globalWs]) { try { w.close(); } catch { /* */ } } if (h) await h.close(); });