type SSEChannel = "gate" | "notification" | "pr" | "ci";
interface SSEClient {
id: string;
controller: ReadableStreamDefaultController;
userId?: string;
channels: Set<string>;
connectedAt: number;
}
const clients = new Map<string, SSEClient>();
let clientIdCounter = 0;
function nextClientId(): string {
return `sse_${++clientIdCounter}_${Date.now().toString(36)}`;
}
export function createSSEStream(
channels: string[],
userId?: string
): Response {
const clientId = nextClientId();
const stream = new ReadableStream({
start(controller) {
const client: SSEClient = {
id: clientId,
controller,
userId,
channels: new Set(channels),
connectedAt: Date.now(),
};
clients.set(clientId, client);
const data = `event: connected\ndata: ${JSON.stringify({ clientId, channels })}\n\n`;
controller.enqueue(new TextEncoder().encode(data));
},
cancel() {
clients.delete(clientId);
},
});
return new Response(stream, {
headers: {
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
Connection: "keep-alive",
"X-Accel-Buffering": "no",
},
});
}
export function broadcast(
channel: string,
event: string,
data: unknown
): number {
const encoded = new TextEncoder();
const payload = `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
const bytes = encoded.encode(payload);
let sent = 0;
for (const [id, client] of clients) {
if (client.channels.has(channel)) {
try {
client.controller.enqueue(bytes);
sent++;
} catch {
clients.delete(id);
}
}
}
return sent;
}
export function sendToUser(
userId: string,
event: string,
data: unknown
): number {
const encoded = new TextEncoder();
const payload = `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
const bytes = encoded.encode(payload);
let sent = 0;
for (const [id, client] of clients) {
if (client.userId === userId) {
try {
client.controller.enqueue(bytes);
sent++;
} catch {
clients.delete(id);
}
}
}
return sent;
}
export function getActiveConnections(): number {
return clients.size;
}
export function cleanupStaleConnections(maxAgeMs = 30 * 60 * 1000): number {
const cutoff = Date.now() - maxAgeMs;
let removed = 0;
for (const [id, client] of clients) {
if (client.connectedAt < cutoff) {
try {
client.controller.close();
} catch {}
clients.delete(id);
removed++;
}
}
return removed;
}
setInterval(() => cleanupStaleConnections(), 5 * 60 * 1000);
|