openvibers/openvibe.live Commits
Live: the Kick chat relay asks Kick for its Pusher key (the old one failed with 4001); restream RTP ports are checked free before ffmpeg binds them (#141)
Changes
@@@ -21,6 +21,12 @@ const http = require('http');const db = require('../db/database');const chatReads = require('../chat/chat-reads');++// Kick's chat runs on Pusher with an app key Kick changes now and then (32cbd69e… stopped working in October 2026 with++// "App key … not in this cluster"). Kick's own web client asks web.kick.com which realtime provider and credentials to++// use; the relay asks the same, for Pusher only, and keeps the answer for six hours. The last key seen working is the++// fallback when that request fails.++const KICK_PUSHER_FALLBACK = { key: '34bf7a0ff419a2a775b9', cluster: 'us2' };++const KICK_PUSHER_TTL_MS = 6 * 3600 * 1000;const RECONNECT_BASE_MS = 5000;const RECONNECT_MAX_MS = 60000;@@@ -577,9 +583,10 @@ class ChatRelayService {if (bridge.stopped) return;-- // Step 2: Connect to Pusher-- const PUSHER_KEY = '32cbd69e4b950bf97679';-- const wsUrl = `wss://ws-us2.pusher.com/app/${PUSHER_KEY}?protocol=7&client=js&version=8.4.0-rc2&flash=false`;++ // Step 2: Connect to Pusher with the key and cluster Kick names today++ const pusher = await this._kickPusherConfig(bridge);++ if (bridge.stopped) return;++ const wsUrl = `wss://ws-${pusher.cluster}.pusher.com/app/${pusher.key}?protocol=7&client=js&version=8.4.0-rc2&flash=false`;const ws = new WebSocket(wsUrl);bridge.ws = ws;@@@ -628,7 +635,9 @@ class ChatRelayService {case 'pusher:error': {const errData = safeJsonParse(msg.data) || msg.data;-- console.warn(`[ChatRelay] Kick: Pusher error for ${bridge.channelName}:`, errData);++ // 4001: the app key is not in this cluster — Kick moved it; ask again before the reconnect.++ if (errData && errData.code === 4001) this._kickPusher = null;++ console.warn(`[ChatRelay] Kick: Pusher error for ${bridge.channelName}: ${errData && errData.code ? `${errData.code} ${errData.message || ''}` : JSON.stringify(errData)}`);break;}@@@ -704,6 +713,40 @@ class ChatRelayService {* @param {string} channelName - Kick channel slug* @returns {Promise<{chatroomId: number, kickChannelId: number|null}>}*/++ /**++ * Kick's current Pusher credentials ({ key, cluster }), as web.kick.com hands them to its own web client when the++ * client accepts only Pusher; cached six hours, the fallback when the request fails or answers something else.++ */++ async _kickPusherConfig(bridge) {++ const now = Date.now();++ if (this._kickPusher && now - this._kickPusher.at < KICK_PUSHER_TTL_MS) return this._kickPusher;++ const channelId = bridge && bridge.kickChannelId;++ const url = channelId ? `https://web.kick.com/api/v1/realtime/channels/${encodeURIComponent(channelId)}/chat/connection` : 'https://web.kick.com/api/v1/realtime/connection';++ try {++ const res = await fetch(url, {++ method: 'POST',++ headers: {++ 'Content-Type': 'application/json', Accept: 'application/json', 'x-app-platform': 'web',++ 'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/141.0.0.0 Safari/537.36',++ },++ body: JSON.stringify({ client: { id: require('crypto').randomUUID(), type: 'web' }, capabilities: { accepted_providers: [{ provider: 'pusher' }] } }),++ signal: AbortSignal.timeout(10000),++ });++ const body = await res.json().catch(() => null);++ const conn = res.ok && body && body.data && Array.isArray(body.data.connections) ? body.data.connections.find((c) => c && c.provider === 'pusher') : null;++ const c = conn && conn.credentials;++ if (c && /^[a-f0-9]{16,40}$/.test(String(c.app_key)) && /^[a-z0-9-]{2,20}$/.test(String(c.cluster))) {++ if (!this._kickPusher || this._kickPusher.key !== c.app_key) console.log(`[ChatRelay] Kick: Pusher key ${String(c.app_key).slice(0, 6)}… on ${c.cluster}`);++ this._kickPusher = { key: c.app_key, cluster: c.cluster, at: now };++ return this._kickPusher;++ }++ console.warn(`[ChatRelay] Kick: realtime connection answered ${res.status} without Pusher credentials; using the last known key`);++ } catch (err) {++ console.warn(`[ChatRelay] Kick: could not ask for Pusher credentials (${err.message}); using the last known key`);++ }++ return { ...(this._kickPusher || KICK_PUSHER_FALLBACK), at: now };++ }++async _getKickChannelInfo(channelName) {const slug = String(channelName || '').toLowerCase();try {@@@ -551,7 +551,7 @@ class RestreamManager extends EventEmitter {const audioProducer = webrtcSFU.findProducerByKind(roomId, 'audio');// Step 3: Allocate ports and create PlainRtpTransport consumers-- const videoRtpPort = this._allocatePort();++ const videoRtpPort = await this._allocateFreePort();const videoRtcpPort = videoRtpPort + 1;let audioRtpPort, audioRtcpPort;@@@ -563,7 +563,7 @@ class RestreamManager extends EventEmitter {console.log(`[Restream] WebRTC: Video consumer created — PT:${videoConsumer.payloadType} SSRC:${videoConsumer.ssrc} port:${videoRtpPort}`);if (audioProducer) {-- audioRtpPort = this._allocatePort();++ audioRtpPort = await this._allocateFreePort();audioRtcpPort = audioRtpPort + 1;audioConsumer = await webrtcSFU.createPlainConsumer(roomId, audioProducer.id, '127.0.0.1', audioRtpPort, audioRtcpPort@@@ -821,6 +821,26 @@ class RestreamManager extends EventEmitter {return port;}++ /**++ * The next RTP/RTCP pair that nothing on this host holds right now. The rotating counter wraps after 5000 pairs, and++ * a session that has run for days (or another program) may still hold the pair it comes back to: ffmpeg then fails++ * with "bind failed: Address already in use". Each candidate's two ports are bound for a moment on 127.0.0.1 and++ * released; a pair in use is skipped (64 tries, then the plain counter as before).++ */++ async _allocateFreePort() {++ const dgram = require('dgram');++ const free = (port) => new Promise((resolve) => {++ const sock = dgram.createSocket('udp4');++ sock.once('error', () => { try { sock.close(); } catch {} resolve(false); });++ sock.bind(port, '127.0.0.1', () => sock.close(() => resolve(true)));++ });++ for (let i = 0; i < 64; i++) {++ const port = this._allocatePort();++ if (await free(port) && await free(port + 1)) return port;++ }++ return this._allocatePort();++ }++/*** Spawn an FFmpeg child process and monitor its lifecycle.* Automatically uses the OpenSSL-based static FFmpeg for RTMPS destinationsnnew file mode 100644@@@ -0,0 +1,59 @@++/**++ * Kick's chat relay asks web.kick.com which Pusher key and cluster to use (Kick moved its key in October 2026 and the++ * hard-coded one failed with 4001 "not in this cluster" on every reconnect):++ * - the channel's realtime connection is asked for Pusher only, and its key and cluster are used and cached;++ * - a failed or odd answer falls back to the last key seen working;++ * - a 4001 error drops the cache, so the reconnect asks again.++ *++ * node test/kick-pusher-config.test.js++ */++'use strict';++const assert = require('assert');++process.env.NODE_ENV = 'test';++console.log = () => {};++const warn = console.warn;++console.warn = () => {};++++const relay = require('../server/integrations/chat-relay-service');++++(async () => {++ const realFetch = globalThis.fetch;++ const asked = [];++ const KEY = 'ab'.repeat(10); // a made-up key, built so no scanner mistakes it for a real one++ let answer = { status: 200, body: { data: { connections: [{ provider: 'pusher', credentials: { app_key: KEY, cluster: 'eu' } }], mode: 'websocket' } } };++ globalThis.fetch = async (url, init) => {++ asked.push({ url, init });++ if (answer instanceof Error) throw answer;++ return { ok: answer.status < 400, status: answer.status, json: async () => answer.body };++ };++ try {++ relay._kickPusher = null;++ const c = await relay._kickPusherConfig({ kickChannelId: 98564718 });++ assert.deepStrictEqual([c.key, c.cluster], [KEY, 'eu']);++ assert.strictEqual(asked[0].url, 'https://web.kick.com/api/v1/realtime/channels/98564718/chat/connection');++ const body = JSON.parse(asked[0].init.body);++ assert.deepStrictEqual(body.capabilities.accepted_providers, [{ provider: 'pusher' }], 'Pusher only');++ assert.strictEqual(asked[0].init.headers['x-app-platform'], 'web');++++ await relay._kickPusherConfig({ kickChannelId: 98564718 });++ assert.strictEqual(asked.length, 1, 'cached');++++ relay._kickPusher = null; // what a 4001 error does++ answer = { status: 200, body: { data: { connections: [{ provider: 'centrifugo', credentials: { url: 'wss://x' } }] } } };++ const fb = await relay._kickPusherConfig({});++ assert.strictEqual(asked[1].url, 'https://web.kick.com/api/v1/realtime/connection', 'no channel id: the general connection');++ assert.deepStrictEqual([fb.key, fb.cluster], ['34bf7a0ff419a2a775b9', 'us2'], 'no Pusher in the answer: the fallback');++++ answer = new Error('network down');++ const fb2 = await relay._kickPusherConfig({ kickChannelId: 1 });++ assert.deepStrictEqual([fb2.key, fb2.cluster], ['34bf7a0ff419a2a775b9', 'us2']);++++ answer = { status: 200, body: { data: { connections: [{ provider: 'pusher', credentials: { app_key: '"><script>', cluster: 'us2' } }] } } };++ relay._kickPusher = null;++ const bad = await relay._kickPusherConfig({ kickChannelId: 1 });++ assert.strictEqual(bad.key, '34bf7a0ff419a2a775b9', 'a malformed key is never put in the socket URL');++ } finally { globalThis.fetch = realFetch; }++ console.warn = warn;++ process.stdout.write('kick pusher config: all checks passed\n');++ process.exit(0);++})().catch((err) => { process.stderr.write(`${err.stack}\n`); process.exit(1); });@@@ -128,6 +128,14 @@ manager.sessions.delete('1:9');assert.ok(bad.nextRestartAt > Date.now(), 'dashboard can count down to the retry');bad.status = 'stopped'; clearTimeout(bad.restartTimer); manager.sessions.delete(bad.key);++ // RTP ports: a pair something already holds is skipped (the rotating counter wraps after 5000 pairs).++ const dgram = require('dgram');++ const holder = dgram.createSocket('udp4');++ await new Promise((r) => holder.bind(20001, '127.0.0.1', r));++ manager._nextPort = 20000;++ assert.strictEqual(await manager._allocateFreePort(), 20002, 'the pair 20000/20001 is in use, so the next pair is used');++ holder.close();++console.log = quiet;console.log('restream-manager lifecycle: all checks passed');process.exit(0);
@@ -21,6 +21,12 @@ const http = require('http');
@@ -21,6 +21,12 @@ const http = require('http');
const db = require('../db/database');
const db = require('../db/database');
const chatReads = require('../chat/chat-reads');
const chatReads = require('../chat/chat-reads');
+// Kick's chat runs on Pusher with an app key Kick changes now and then (32cbd69e… stopped working in October 2026 with
+// "App key … not in this cluster"). Kick's own web client asks web.kick.com which realtime provider and credentials to
+// use; the relay asks the same, for Pusher only, and keeps the answer for six hours. The last key seen working is the
+// fallback when that request fails.
+const KICK_PUSHER_FALLBACK = { key: '34bf7a0ff419a2a775b9', cluster: 'us2' };
+const KICK_PUSHER_TTL_MS = 6 * 3600 * 1000;
const RECONNECT_BASE_MS = 5000;
const RECONNECT_BASE_MS = 5000;
const RECONNECT_MAX_MS = 60000;
const RECONNECT_MAX_MS = 60000;
@@ -577,9 +583,10 @@ class ChatRelayService {
@@ -577,9 +583,10 @@ class ChatRelayService {
if (bridge.stopped) return;
if (bridge.stopped) return;
- // Step 2: Connect to Pusher
- const PUSHER_KEY = '32cbd69e4b950bf97679';
- const wsUrl = `wss://ws-us2.pusher.com/app/${PUSHER_KEY}?protocol=7&client=js&version=8.4.0-rc2&flash=false`;
+ // Step 2: Connect to Pusher with the key and cluster Kick names today
+ const pusher = await this._kickPusherConfig(bridge);
+ if (bridge.stopped) return;
+ const wsUrl = `wss://ws-${pusher.cluster}.pusher.com/app/${pusher.key}?protocol=7&client=js&version=8.4.0-rc2&flash=false`;
const ws = new WebSocket(wsUrl);
const ws = new WebSocket(wsUrl);
bridge.ws = ws;
bridge.ws = ws;
@@ -628,7 +635,9 @@ class ChatRelayService {
@@ -628,7 +635,9 @@ class ChatRelayService {
case 'pusher:error': {
case 'pusher:error': {
const errData = safeJsonParse(msg.data) || msg.data;
const errData = safeJsonParse(msg.data) || msg.data;
- console.warn(`[ChatRelay] Kick: Pusher error for ${bridge.channelName}:`, errData);
+ // 4001: the app key is not in this cluster — Kick moved it; ask again before the reconnect.
+ if (errData && errData.code === 4001) this._kickPusher = null;
+ console.warn(`[ChatRelay] Kick: Pusher error for ${bridge.channelName}: ${errData && errData.code ? `${errData.code} ${errData.message || ''}` : JSON.stringify(errData)}`);
break;
break;
}
}
@@ -704,6 +713,40 @@ class ChatRelayService {
@@ -704,6 +713,40 @@ class ChatRelayService {
* @param {string} channelName - Kick channel slug
* @param {string} channelName - Kick channel slug
* @returns {Promise<{chatroomId: number, kickChannelId: number|null}>}
* @returns {Promise<{chatroomId: number, kickChannelId: number|null}>}
*/
*/
+ /**
+ * Kick's current Pusher credentials ({ key, cluster }), as web.kick.com hands them to its own web client when the
+ * client accepts only Pusher; cached six hours, the fallback when the request fails or answers something else.
+ */
+ async _kickPusherConfig(bridge) {
+ const now = Date.now();
+ if (this._kickPusher && now - this._kickPusher.at < KICK_PUSHER_TTL_MS) return this._kickPusher;
+ const channelId = bridge && bridge.kickChannelId;
+ const url = channelId ? `https://web.kick.com/api/v1/realtime/channels/${encodeURIComponent(channelId)}/chat/connection` : 'https://web.kick.com/api/v1/realtime/connection';
+ try {
+ const res = await fetch(url, {
+ method: 'POST',
+ headers: {
+ 'Content-Type': 'application/json', Accept: 'application/json', 'x-app-platform': 'web',
+ 'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/141.0.0.0 Safari/537.36',
+ },
+ body: JSON.stringify({ client: { id: require('crypto').randomUUID(), type: 'web' }, capabilities: { accepted_providers: [{ provider: 'pusher' }] } }),
+ signal: AbortSignal.timeout(10000),
+ });
+ const body = await res.json().catch(() => null);
+ const conn = res.ok && body && body.data && Array.isArray(body.data.connections) ? body.data.connections.find((c) => c && c.provider === 'pusher') : null;
+ const c = conn && conn.credentials;
+ if (c && /^[a-f0-9]{16,40}$/.test(String(c.app_key)) && /^[a-z0-9-]{2,20}$/.test(String(c.cluster))) {
+ if (!this._kickPusher || this._kickPusher.key !== c.app_key) console.log(`[ChatRelay] Kick: Pusher key ${String(c.app_key).slice(0, 6)}… on ${c.cluster}`);
+ this._kickPusher = { key: c.app_key, cluster: c.cluster, at: now };
+ return this._kickPusher;
+ }
+ console.warn(`[ChatRelay] Kick: realtime connection answered ${res.status} without Pusher credentials; using the last known key`);
+ } catch (err) {
+ console.warn(`[ChatRelay] Kick: could not ask for Pusher credentials (${err.message}); using the last known key`);
+ }
+ return { ...(this._kickPusher || KICK_PUSHER_FALLBACK), at: now };
+ }
+
async _getKickChannelInfo(channelName) {
async _getKickChannelInfo(channelName) {
const slug = String(channelName || '').toLowerCase();
const slug = String(channelName || '').toLowerCase();
try {
try {
@@ -551,7 +551,7 @@ class RestreamManager extends EventEmitter {
@@ -551,7 +551,7 @@ class RestreamManager extends EventEmitter {
const audioProducer = webrtcSFU.findProducerByKind(roomId, 'audio');
const audioProducer = webrtcSFU.findProducerByKind(roomId, 'audio');
// Step 3: Allocate ports and create PlainRtpTransport consumers
// Step 3: Allocate ports and create PlainRtpTransport consumers
- const videoRtpPort = this._allocatePort();
+ const videoRtpPort = await this._allocateFreePort();
const videoRtcpPort = videoRtpPort + 1;
const videoRtcpPort = videoRtpPort + 1;
let audioRtpPort, audioRtcpPort;
let audioRtpPort, audioRtcpPort;
@@ -563,7 +563,7 @@ class RestreamManager extends EventEmitter {
@@ -563,7 +563,7 @@ class RestreamManager extends EventEmitter {
console.log(`[Restream] WebRTC: Video consumer created — PT:${videoConsumer.payloadType} SSRC:${videoConsumer.ssrc} port:${videoRtpPort}`);
console.log(`[Restream] WebRTC: Video consumer created — PT:${videoConsumer.payloadType} SSRC:${videoConsumer.ssrc} port:${videoRtpPort}`);
if (audioProducer) {
if (audioProducer) {
- audioRtpPort = this._allocatePort();
+ audioRtpPort = await this._allocateFreePort();
audioRtcpPort = audioRtpPort + 1;
audioRtcpPort = audioRtpPort + 1;
audioConsumer = await webrtcSFU.createPlainConsumer(
audioConsumer = await webrtcSFU.createPlainConsumer(
roomId, audioProducer.id, '127.0.0.1', audioRtpPort, audioRtcpPort
roomId, audioProducer.id, '127.0.0.1', audioRtpPort, audioRtcpPort
@@ -821,6 +821,26 @@ class RestreamManager extends EventEmitter {
@@ -821,6 +821,26 @@ class RestreamManager extends EventEmitter {
return port;
return port;
}
}
+ /**
+ * The next RTP/RTCP pair that nothing on this host holds right now. The rotating counter wraps after 5000 pairs, and
+ * a session that has run for days (or another program) may still hold the pair it comes back to: ffmpeg then fails
+ * with "bind failed: Address already in use". Each candidate's two ports are bound for a moment on 127.0.0.1 and
+ * released; a pair in use is skipped (64 tries, then the plain counter as before).
+ */
+ async _allocateFreePort() {
+ const dgram = require('dgram');
+ const free = (port) => new Promise((resolve) => {
+ const sock = dgram.createSocket('udp4');
+ sock.once('error', () => { try { sock.close(); } catch {} resolve(false); });
+ sock.bind(port, '127.0.0.1', () => sock.close(() => resolve(true)));
+ });
+ for (let i = 0; i < 64; i++) {
+ const port = this._allocatePort();
+ if (await free(port) && await free(port + 1)) return port;
+ }
+ return this._allocatePort();
+ }
+
/**
/**
* Spawn an FFmpeg child process and monitor its lifecycle.
* Spawn an FFmpeg child process and monitor its lifecycle.
* Automatically uses the OpenSSL-based static FFmpeg for RTMPS destinations
* Automatically uses the OpenSSL-based static FFmpeg for RTMPS destinations
new file mode 100644
new file mode 100644
@@ -0,0 +1,59 @@
@@ -0,0 +1,59 @@
+/**
+ * Kick's chat relay asks web.kick.com which Pusher key and cluster to use (Kick moved its key in October 2026 and the
+ * hard-coded one failed with 4001 "not in this cluster" on every reconnect):
+ * - the channel's realtime connection is asked for Pusher only, and its key and cluster are used and cached;
+ * - a failed or odd answer falls back to the last key seen working;
+ * - a 4001 error drops the cache, so the reconnect asks again.
+ *
+ * node test/kick-pusher-config.test.js
+ */
+'use strict';
+const assert = require('assert');
+process.env.NODE_ENV = 'test';
+console.log = () => {};
+const warn = console.warn;
+console.warn = () => {};
+
+const relay = require('../server/integrations/chat-relay-service');
+
+(async () => {
+ const realFetch = globalThis.fetch;
+ const asked = [];
+ const KEY = 'ab'.repeat(10); // a made-up key, built so no scanner mistakes it for a real one
+ let answer = { status: 200, body: { data: { connections: [{ provider: 'pusher', credentials: { app_key: KEY, cluster: 'eu' } }], mode: 'websocket' } } };
+ globalThis.fetch = async (url, init) => {
+ asked.push({ url, init });
+ if (answer instanceof Error) throw answer;
+ return { ok: answer.status < 400, status: answer.status, json: async () => answer.body };
+ };
+ try {
+ relay._kickPusher = null;
+ const c = await relay._kickPusherConfig({ kickChannelId: 98564718 });
+ assert.deepStrictEqual([c.key, c.cluster], [KEY, 'eu']);
+ assert.strictEqual(asked[0].url, 'https://web.kick.com/api/v1/realtime/channels/98564718/chat/connection');
+ const body = JSON.parse(asked[0].init.body);
+ assert.deepStrictEqual(body.capabilities.accepted_providers, [{ provider: 'pusher' }], 'Pusher only');
+ assert.strictEqual(asked[0].init.headers['x-app-platform'], 'web');
+
+ await relay._kickPusherConfig({ kickChannelId: 98564718 });
+ assert.strictEqual(asked.length, 1, 'cached');
+
+ relay._kickPusher = null; // what a 4001 error does
+ answer = { status: 200, body: { data: { connections: [{ provider: 'centrifugo', credentials: { url: 'wss://x' } }] } } };
+ const fb = await relay._kickPusherConfig({});
+ assert.strictEqual(asked[1].url, 'https://web.kick.com/api/v1/realtime/connection', 'no channel id: the general connection');
+ assert.deepStrictEqual([fb.key, fb.cluster], ['34bf7a0ff419a2a775b9', 'us2'], 'no Pusher in the answer: the fallback');
+
+ answer = new Error('network down');
+ const fb2 = await relay._kickPusherConfig({ kickChannelId: 1 });
+ assert.deepStrictEqual([fb2.key, fb2.cluster], ['34bf7a0ff419a2a775b9', 'us2']);
+
+ answer = { status: 200, body: { data: { connections: [{ provider: 'pusher', credentials: { app_key: '"><script>', cluster: 'us2' } }] } } };
+ relay._kickPusher = null;
+ const bad = await relay._kickPusherConfig({ kickChannelId: 1 });
+ assert.strictEqual(bad.key, '34bf7a0ff419a2a775b9', 'a malformed key is never put in the socket URL');
+ } finally { globalThis.fetch = realFetch; }
+ console.warn = warn;
+ process.stdout.write('kick pusher config: all checks passed\n');
+ process.exit(0);
+})().catch((err) => { process.stderr.write(`${err.stack}\n`); process.exit(1); });
@@ -128,6 +128,14 @@ manager.sessions.delete('1:9');
@@ -128,6 +128,14 @@ manager.sessions.delete('1:9');
assert.ok(bad.nextRestartAt > Date.now(), 'dashboard can count down to the retry');
assert.ok(bad.nextRestartAt > Date.now(), 'dashboard can count down to the retry');
bad.status = 'stopped'; clearTimeout(bad.restartTimer); manager.sessions.delete(bad.key);
bad.status = 'stopped'; clearTimeout(bad.restartTimer); manager.sessions.delete(bad.key);
+ // RTP ports: a pair something already holds is skipped (the rotating counter wraps after 5000 pairs).
+ const dgram = require('dgram');
+ const holder = dgram.createSocket('udp4');
+ await new Promise((r) => holder.bind(20001, '127.0.0.1', r));
+ manager._nextPort = 20000;
+ assert.strictEqual(await manager._allocateFreePort(), 20002, 'the pair 20000/20001 is in use, so the next pair is used');
+ holder.close();
+
console.log = quiet;
console.log = quiet;
console.log('restream-manager lifecycle: all checks passed');
console.log('restream-manager lifecycle: all checks passed');
process.exit(0);
process.exit(0);