perf: SSE connection pooling and cleanup
Add per-IP SSE connection limit (max 5 concurrent streams). Track activeSSECount and ssePerIp maps with proper decrement in finally blocks. Export getActiveSSECount() for admin stats. Change heartbeat ping from 5s to 15s interval. Connections already clean up on client disconnect via Hono's streamSSE try/catch/finally pattern. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.6
parent
74939e995d
commit
38f94c2955
@@ -20,6 +20,15 @@ const spectatorCounts = new Map<string, number>()
|
||||
const VALID_REACTIONS = new Set(['fist', 'fire', 'skull', '100', 'clown'])
|
||||
const fightReactions = new Map<string, Record<string, number>>()
|
||||
|
||||
// Active SSE connection tracking
|
||||
let activeSSECount = 0
|
||||
const ssePerIp = new Map<string, number>()
|
||||
const MAX_SSE_PER_IP = 5
|
||||
|
||||
export function getActiveSSECount(): number {
|
||||
return activeSSECount
|
||||
}
|
||||
|
||||
export function getSpectatorCount(fightId: string): number {
|
||||
return spectatorCounts.get(fightId) || 0
|
||||
}
|
||||
@@ -332,8 +341,20 @@ fightsRouter.post('/:fightId/respond/:botId', async (c) => {
|
||||
// SSE stream for live fight events
|
||||
fightsRouter.get('/:id/stream', (c) => {
|
||||
const fightId = c.req.param('id')
|
||||
const xff = c.req.header('x-forwarded-for')
|
||||
const clientIp = xff ? xff.split(',')[0].trim() : c.req.header('x-real-ip') || 'unknown'
|
||||
|
||||
// Enforce per-IP SSE connection limit
|
||||
const ipCount = ssePerIp.get(clientIp) || 0
|
||||
if (ipCount >= MAX_SSE_PER_IP) {
|
||||
return c.json({ error: 'Too many SSE connections' }, 429)
|
||||
}
|
||||
|
||||
return streamSSE(c, async (stream) => {
|
||||
// Track connections
|
||||
activeSSECount++
|
||||
ssePerIp.set(clientIp, (ssePerIp.get(clientIp) || 0) + 1)
|
||||
|
||||
// Track spectator
|
||||
spectatorCounts.set(fightId, (spectatorCounts.get(fightId) || 0) + 1)
|
||||
const count = spectatorCounts.get(fightId)!
|
||||
@@ -366,7 +387,7 @@ fightsRouter.get('/:id/stream', (c) => {
|
||||
event: 'ping',
|
||||
data: JSON.stringify({ spectators: spectatorCounts.get(fightId) || 0 }),
|
||||
})
|
||||
await stream.sleep(5000)
|
||||
await stream.sleep(15000)
|
||||
const fight = await db.select({ status: schema.fights.status })
|
||||
.from(schema.fights)
|
||||
.where(eq(schema.fights.id, fightId))
|
||||
@@ -376,6 +397,12 @@ fightsRouter.get('/:id/stream', (c) => {
|
||||
} catch {
|
||||
// Client disconnected
|
||||
} finally {
|
||||
// Decrement connection counts
|
||||
activeSSECount = Math.max(0, activeSSECount - 1)
|
||||
const ipCurrent = ssePerIp.get(clientIp) || 1
|
||||
if (ipCurrent <= 1) ssePerIp.delete(clientIp)
|
||||
else ssePerIp.set(clientIp, ipCurrent - 1)
|
||||
|
||||
// Decrement spectator count
|
||||
const current = spectatorCounts.get(fightId) || 1
|
||||
if (current <= 1) {
|
||||
|
||||
Reference in New Issue
Block a user