Lumi/src/services/twitch-eventsub.js

248 lines
11 KiB
JavaScript

const WebSocket = require("ws");
const { getSetting, setSetting } = require("./settings");
const { emitLumiEvent } = require("./lumi-events");
const { createLogger } = require("./logger");
const EVENTSUB_URL = "wss://eventsub.wss.twitch.tv/ws?keepalive_timeout_seconds=30";
const eventLog = createLogger("platform:twitch:eventsub", { category: "integration" });
function configuredChannels() {
const channels = String(getSetting("twitch_channels", ""))
.split(/[\s,]+/)
.map((value) => value.replace(/^#/, "").trim().toLowerCase())
.filter(Boolean);
if (!channels.length) {
const authorizedLogin = String(getSetting("twitch_event_login", "")).trim().toLowerCase();
if (authorizedLogin) channels.push(authorizedLogin);
}
return Array.from(new Set(channels));
}
async function twitchRequest(url, options = {}) {
const response = await fetch(url, options);
if (!response.ok) {
const body = await response.text().catch(() => "");
throw new Error(`Twitch request failed (${response.status})${body ? `: ${body.slice(0, 300)}` : ""}`);
}
return response.json();
}
async function refreshAccessToken() {
const refreshToken = getSetting("twitch_event_refresh_token", "");
const clientId = getSetting("twitch_client_id", "");
const clientSecret = getSetting("twitch_client_secret", "");
if (!refreshToken || !clientId || !clientSecret) return null;
const params = new URLSearchParams({
grant_type: "refresh_token",
refresh_token: refreshToken,
client_id: clientId,
client_secret: clientSecret
});
const token = await twitchRequest(`https://id.twitch.tv/oauth2/token?${params.toString()}`, { method: "POST" });
setSetting("twitch_event_oauth", token.access_token);
if (token.refresh_token) setSetting("twitch_event_refresh_token", token.refresh_token);
setSetting("twitch_event_expires_at", Date.now() + Number(token.expires_in || 0) * 1000);
return token.access_token;
}
async function validCredentials() {
let accessToken = getSetting("twitch_event_oauth", "");
if (!accessToken) return null;
let validation;
try {
validation = await twitchRequest("https://id.twitch.tv/oauth2/validate", {
headers: { Authorization: `OAuth ${accessToken}` }
});
} catch {
accessToken = await refreshAccessToken();
if (!accessToken) return null;
validation = await twitchRequest("https://id.twitch.tv/oauth2/validate", {
headers: { Authorization: `OAuth ${accessToken}` }
});
}
return { accessToken, validation, clientId: getSetting("twitch_client_id", "") };
}
class TwitchEventSubManager {
constructor() {
this.socket = null;
this.running = false;
this.reconnectTimer = null;
this.reconnectAttempt = 0;
this.generation = 0;
this.status = { state: "disconnected", detail: "Not connected", subscriptions: 0 };
}
getStatus() {
return { ...this.status };
}
async start() {
if (this.running) return this.getStatus();
this.running = true;
if (!getSetting("twitch_event_oauth", "")) {
this.status = { state: "needs_authorization", detail: "Connect Twitch events to receive alerts.", subscriptions: 0 };
return this.getStatus();
}
await this.connect(EVENTSUB_URL, false);
return this.getStatus();
}
async restart() {
await this.stop();
return this.start();
}
async stop() {
this.running = false;
this.generation += 1;
clearTimeout(this.reconnectTimer);
this.reconnectTimer = null;
const socket = this.socket;
this.socket = null;
if (socket && socket.readyState < WebSocket.CLOSING) socket.close(1000, "Lumi shutdown");
this.status = { state: "disconnected", detail: "Not connected", subscriptions: 0 };
}
async connect(url, preservingSubscriptions) {
if (!this.running) return;
const credentials = await validCredentials().catch((error) => {
eventLog.error("Twitch EventSub credentials could not be validated", error, { event: "eventsub_auth_failed" });
return null;
});
if (!credentials?.accessToken || !credentials.clientId) {
this.status = { state: "needs_authorization", detail: "Twitch event authorization is missing or expired.", subscriptions: 0 };
return;
}
const generation = ++this.generation;
this.status = { state: "connecting", detail: "Connecting to Twitch events…", subscriptions: 0 };
const socket = new WebSocket(url);
this.socket = socket;
socket.on("message", (raw) => {
this.handleMessage(raw, credentials, preservingSubscriptions, generation).catch((error) => {
eventLog.error("Twitch EventSub message could not be handled", error, { event: "eventsub_message_failed" });
});
});
socket.on("error", (error) => eventLog.error("Twitch EventSub socket error", error, { event: "eventsub_socket_error" }));
socket.on("close", () => {
if (!this.running || generation !== this.generation) return;
this.status = { state: "reconnecting", detail: "Twitch events disconnected; retrying automatically.", subscriptions: 0 };
this.scheduleReconnect();
});
}
scheduleReconnect() {
clearTimeout(this.reconnectTimer);
const delay = Math.min(60000, 1000 * 2 ** Math.min(this.reconnectAttempt++, 6));
this.reconnectTimer = setTimeout(() => {
this.connect(EVENTSUB_URL, false).catch((error) => {
eventLog.error("Twitch EventSub reconnect failed", error, { event: "eventsub_reconnect_failed" });
if (this.running) this.scheduleReconnect();
});
}, delay);
}
async handleMessage(raw, credentials, preservingSubscriptions, generation) {
let message;
try {
message = JSON.parse(raw.toString());
} catch {
return;
}
const type = message.metadata?.message_type;
if (type === "session_welcome") {
this.reconnectAttempt = 0;
this.status = { state: "connected", detail: "Receiving Twitch events.", subscriptions: preservingSubscriptions ? this.status.subscriptions : 0 };
if (!preservingSubscriptions) {
await this.subscribe(message.payload.session.id, credentials, generation);
}
return;
}
if (type === "session_reconnect") {
const reconnectUrl = message.payload?.session?.reconnect_url;
if (reconnectUrl) {
this.connect(reconnectUrl, true).catch((error) => {
eventLog.error("Twitch requested reconnect failed", error, { event: "eventsub_reconnect_failed" });
if (this.running) this.scheduleReconnect();
});
}
return;
}
if (type === "notification") this.emitNotification(message);
if (type === "revocation") {
this.status = { ...this.status, detail: "A Twitch event permission was revoked. Reconnect Twitch events if alerts stop." };
const subscription = message.payload?.subscription || {};
eventLog.warn("Twitch revoked an EventSub subscription", {
subscription_id: subscription.id || null,
subscription_type: subscription.type || null,
status: subscription.status || null,
reason: message.payload?.status || null
}, { event: "eventsub_revoked" });
}
}
async subscribe(sessionId, credentials, generation) {
const headers = {
"Client-Id": credentials.clientId,
Authorization: `Bearer ${credentials.accessToken}`,
"Content-Type": "application/json"
};
const scopes = new Set(credentials.validation.scopes || []);
const query = configuredChannels();
const users = query.length
? await twitchRequest(`https://api.twitch.tv/helix/users?${query.map((login) => `login=${encodeURIComponent(login)}`).join("&")}`, { headers })
: { data: [] };
const definitions = [];
for (const broadcaster of users.data || []) {
definitions.push({ type: "channel.raid", version: "1", condition: { to_broadcaster_user_id: broadcaster.id } });
if (scopes.has("moderator:read:followers")) {
definitions.push({ type: "channel.follow", version: "2", condition: { broadcaster_user_id: broadcaster.id, moderator_user_id: credentials.validation.user_id } });
}
if (scopes.has("channel:read:subscriptions") && broadcaster.id === credentials.validation.user_id) {
definitions.push({ type: "channel.subscribe", version: "1", condition: { broadcaster_user_id: broadcaster.id } });
definitions.push({ type: "channel.subscription.gift", version: "1", condition: { broadcaster_user_id: broadcaster.id } });
}
}
let active = 0;
for (const definition of definitions) {
try {
await twitchRequest("https://api.twitch.tv/helix/eventsub/subscriptions", {
method: "POST",
headers,
body: JSON.stringify({ ...definition, transport: { method: "websocket", session_id: sessionId } })
});
active += 1;
} catch (error) {
eventLog.error(`Twitch EventSub subscription failed for ${definition.type}`, error, { event: "eventsub_subscribe_failed" });
}
}
if (generation !== this.generation) return;
this.status = {
state: "connected",
detail: active ? `Receiving ${active} Twitch event feed${active === 1 ? "" : "s"}.` : "Connected, but no configured channel events could be subscribed.",
subscriptions: active
};
eventLog.info("Twitch EventSub connected", { channels: users.data?.length || 0, subscriptions: active }, { event: "eventsub_ready" });
}
emitNotification(message) {
const subscription = message.payload?.subscription || {};
const event = message.payload?.event || {};
const common = {
broadcaster_id: event.broadcaster_user_id || event.to_broadcaster_user_id || null,
broadcaster_name: event.broadcaster_user_name || event.to_broadcaster_user_name || null,
user_id: event.user_id || event.from_broadcaster_user_id || null,
user_name: event.user_name || event.from_broadcaster_user_name || null
};
const metadata = { id: message.metadata?.message_id, source: "twitch-eventsub", occurredAt: message.metadata?.message_timestamp };
if (subscription.type === "channel.follow") emitLumiEvent("twitch.follow", { ...common, followed_at: event.followed_at }, metadata);
if (subscription.type === "channel.raid") emitLumiEvent("twitch.raid", { ...common, viewers: Number(event.viewers || 0) }, metadata);
if (subscription.type === "channel.subscribe") emitLumiEvent("twitch.subscribe", { ...common, tier: event.tier, gifted: Boolean(event.is_gift) }, metadata);
if (subscription.type === "channel.subscription.gift") emitLumiEvent("twitch.subscription_gift", { ...common, tier: event.tier, total: Number(event.total || 0), anonymous: Boolean(event.is_anonymous) }, metadata);
}
}
const twitchEventSubManager = new TwitchEventSubManager();
module.exports = { TwitchEventSubManager, twitchEventSubManager, configuredChannels };