Something went wrong. Try again.
A streaming bot for Streamplace forked from bot.timtinkers.online/streamplace-bot
Something went wrong. Try again.
4.5 kB · 163 lines
TypeScript
at flake
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164import { type JetstreamEvent } from "@atcute/jetstream";import { WebSocket as WsWebSocket } from "ws";import { JETSTREAM_URL } from "../../env.ts";import { EventHandler } from "./eventHandler.ts";
// Collections we want to filter in Jetstream forconst wantedCollections = [ "app.bsky.actor.profile", "app.bsky.graph.follow", "online.timtinkers.bot.command", "online.timtinkers.bot.recurring", "online.timtinkers.bot.shoutout", "online.timtinkers.bot.variable", "place.stream.chat.message", "place.stream.chat.profile", "place.stream.livestream", "place.stream.live.teleport",];
// Client subscription messageinterface SubscriptionMessage { type: "subscribe" | "unsubscribe"; streamer: string; // DID}
// WebSocket service classclass BotWebSocketService { private jetstreamWs: WebSocket | null = null; // native/global WebSocket — outgoing connection private clients = new Map<WsWebSocket, Set<string>>(); // client -> subscribed streamers private streamerClients = new Map<string, Set<WsWebSocket>>(); // streamer -> clients private _eventHandler = new EventHandler(this.streamerClients);
constructor() {}
start() { console.log("Starting WebSocket service"); this.connectToJetstream(); }
initClient(client: WsWebSocket) { this.clients.set(client, new Set()); }
// Handle new connection with pre-defined streamer DID handleNewConnection(client: WsWebSocket, streamerDid: Did) { this.initClient(client); // Auto-subscribe to the streamer from the route parameter this.handleClientMessage(client, { type: "subscribe", streamer: streamerDid, }); }
private connectToJetstream() { const baseUrl = JETSTREAM_URL; const jetstreamUrl = `${baseUrl}?${ wantedCollections.map((c) => `wantedCollections=${c}`).join("&") }`; this.jetstreamWs = new WebSocket(jetstreamUrl);
let pingInterval: ReturnType<typeof setInterval> | null = null;
this.jetstreamWs.onopen = () => { console.log("Connected to Jetstream");
// Send a ping every 15 seconds to keep the connection alive pingInterval = setInterval(() => { if (this.jetstreamWs?.readyState === WebSocket.OPEN) { this.jetstreamWs.send(JSON.stringify({ type: "ping" })); } }, 15_000); };
this.jetstreamWs.onmessage = (event) => { try { const jetstreamEvent: JetstreamEvent = JSON.parse(event.data); this._eventHandler.handleEvent(jetstreamEvent); } catch (error) { console.error("Error processing jetstream message:", error); } };
this.jetstreamWs.onclose = () => { console.log( "Jetstream connection closed, attempting to reconnect...", ); if (pingInterval !== null) clearInterval(pingInterval); setTimeout(() => this.connectToJetstream(), 5000); };
this.jetstreamWs.onerror = (error) => { console.error("Jetstream WebSocket error:", error); }; }
handleClientMessage(client: WsWebSocket, message: SubscriptionMessage) { const clientStreamers = this.clients.get(client); if (!clientStreamers) return;
switch (message.type) { case "subscribe": { clientStreamers.add(message.streamer);
if (!this.streamerClients.has(message.streamer)) { this.streamerClients.set(message.streamer, new Set()); } this.streamerClients.get(message.streamer)!.add(client);
client.send(JSON.stringify({ type: "subscribed", streamer: message.streamer, })); break; }
case "unsubscribe": { clientStreamers.delete(message.streamer);
const streamerClientSet = this.streamerClients.get( message.streamer, ); if (streamerClientSet) { streamerClientSet.delete(client); if (streamerClientSet.size === 0) { this.streamerClients.delete(message.streamer); } }
console.log( `Client unsubscribed from streamer: ${message.streamer}`, ); client.send(JSON.stringify({ type: "unsubscribed", streamer: message.streamer, })); break; } } }
removeClient(client: WsWebSocket) { if (!this.clients.has(client)) return; const clientStreamers = this.clients.get(client); if (clientStreamers) { for (const streamer of clientStreamers) { const streamerClientSet = this.streamerClients.get(streamer); if (streamerClientSet) { streamerClientSet.delete(client); if (streamerClientSet.size === 0) { this.streamerClients.delete(streamer); } } } } this.clients.delete(client); }
get eventHandler(): EventHandler { return this._eventHandler; }}
export const botWS = new BotWebSocketService();