From 19f61ac454a77efb59cb73d84eb15607c781635b Mon Sep 17 00:00:00 2001 From: yohlo Date: Mon, 13 Jul 2026 23:56:19 -0700 Subject: [PATCH] SSE --- k8s/base/app-deployment.yaml | 2 +- src/app/routes/api/events.$.ts | 71 +++++------- src/features/badges/server.ts | 13 ++- src/features/matches/server.ts | 21 +++- src/features/players/server.ts | 13 +++ src/features/teams/server.ts | 9 +- src/features/tournaments/server.ts | 37 +++++- src/hooks/use-server-events.ts | 173 +++++++++++++++++++---------- src/lib/events/emitter.ts | 89 +++++++++++---- src/utils/test-event.ts | 4 +- 10 files changed, 287 insertions(+), 145 deletions(-) diff --git a/k8s/base/app-deployment.yaml b/k8s/base/app-deployment.yaml index f3fec21..3a79c08 100644 --- a/k8s/base/app-deployment.yaml +++ b/k8s/base/app-deployment.yaml @@ -6,7 +6,7 @@ metadata: app: flxn component: app spec: - replicas: 1 + replicas: 1 # Must stay at 1 for SSE selector: matchLabels: app: flxn diff --git a/src/app/routes/api/events.$.ts b/src/app/routes/api/events.$.ts index c5160b7..ff488fa 100644 --- a/src/app/routes/api/events.$.ts +++ b/src/app/routes/api/events.$.ts @@ -1,9 +1,10 @@ import { createFileRoute } from "@tanstack/react-router"; -import { serverEvents, type ServerEvent } from "@/lib/events/emitter"; +import { serverEvents, EVENT_TYPES, type ServerEvent } from "@/lib/events/emitter"; import { logger } from "@/lib/logger"; import { superTokensRequestMiddleware } from "@/utils/supertokens"; let activeConnections = 0; +const encoder = new TextEncoder(); export const Route = createFileRoute("/api/events/$")({ server: { @@ -13,63 +14,47 @@ export const Route = createFileRoute("/api/events/$")({ activeConnections++; const connectionId = `conn_${Date.now()}_${Math.random().toString(36).substring(2, 11)}`; logger.info(`ServerEvents | New connection ${connectionId}. Active: ${activeConnections}`); + + let cleanedUp = false; + let cleanup = () => {}; + const stream = new ReadableStream({ start(controller) { - const connectMessage = `data: ${JSON.stringify({ type: "connected" })}\n\n`; - controller.enqueue(new TextEncoder().encode(connectMessage)); - - const handleEvent = (event: ServerEvent) => { - logger.info("ServerEvents | Event received", event); - const message = `data: ${JSON.stringify(event)}\n\n`; + const send = (payload: unknown) => { try { - if (!controller.desiredSize || controller.desiredSize <= 0) { - logger.warn("ServerEvents | Stream closed, skipping event"); - return; - } - controller.enqueue(new TextEncoder().encode(message)); + controller.enqueue(encoder.encode(`data: ${JSON.stringify(payload)}\n\n`)); } catch (error) { logger.error("ServerEvents | Error sending SSE message", error); + cleanup(); } }; - serverEvents.on("test", handleEvent); - serverEvents.on("match", handleEvent); - serverEvents.on("reaction", handleEvent); + const handleEvent = (event: ServerEvent) => send(event); + for (const type of EVENT_TYPES) { + serverEvents.on(type, handleEvent); + } const pingInterval = setInterval(() => { - try { - if (!controller.desiredSize || controller.desiredSize <= 0) { - clearInterval(pingInterval); - return; - } - const pingMessage = `data: ${JSON.stringify({ type: "ping", timestamp: Date.now() })}\n\n`; - controller.enqueue(new TextEncoder().encode(pingMessage)); - } catch (e) { - logger.error("ServerEvents | Ping interval error", e); - clearInterval(pingInterval); - } + send({ type: "ping", timestamp: Date.now() }); }, 15000); - setTimeout(() => { - try { - const heartbeatMessage = `data: ${JSON.stringify({ type: "heartbeat", timestamp: Date.now() })}\n\n`; - controller.enqueue(new TextEncoder().encode(heartbeatMessage)); - } catch (e) { - logger.error("ServerEvents | Heartbeat error", e); - } - }, 1000); - - const cleanup = () => { + cleanup = () => { + if (cleanedUp) return; + cleanedUp = true; activeConnections--; - serverEvents.off("test", handleEvent); - serverEvents.off("match", handleEvent); - serverEvents.off("reaction", handleEvent); + for (const type of EVENT_TYPES) { + serverEvents.off(type, handleEvent); + } clearInterval(pingInterval); logger.info(`ServerEvents | Connection ${connectionId} cleanup completed. Active: ${activeConnections}`); }; request.signal?.addEventListener("abort", cleanup); - return cleanup; + + send({ type: "connected" }); + }, + cancel() { + cleanup(); }, }); @@ -77,13 +62,7 @@ export const Route = createFileRoute("/api/events/$")({ headers: { "Content-Type": "text/event-stream", "Cache-Control": "no-cache, no-store, must-revalidate", - "Connection": "keep-alive", - "Access-Control-Allow-Origin": "*", - "Access-Control-Allow-Headers": "Cache-Control", "X-Accel-Buffering": "no", - "X-Proxy-Buffering": "no", - "Proxy-Buffering": "off", - "Transfer-Encoding": "chunked", }, }); }, diff --git a/src/features/badges/server.ts b/src/features/badges/server.ts index bcc454f..1b9ad54 100644 --- a/src/features/badges/server.ts +++ b/src/features/badges/server.ts @@ -3,6 +3,7 @@ import { superTokensAdminFunctionMiddleware, superTokensFunctionMiddleware } fro import { createServerFn } from "@tanstack/react-start"; import { pbAdmin } from "@/lib/pocketbase/client"; import { z } from "zod"; +import { emitServerEvent } from "@/lib/events/emitter"; export const getPlayerBadges = createServerFn() .validator(z.string()) @@ -14,7 +15,11 @@ export const getPlayerBadges = createServerFn() export const migrateBadgeProgress = createServerFn() .middleware([superTokensAdminFunctionMiddleware]) .handler(async () => - toServerResult(() => pbAdmin.migrateBadgeProgress()) + toServerResult(async () => { + const result = await pbAdmin.migrateBadgeProgress(); + emitServerEvent({ type: "badge" }); + return result; + }) ); export const getAllBadges = createServerFn() @@ -34,5 +39,9 @@ export const awardManualBadge = createServerFn() })) .middleware([superTokensAdminFunctionMiddleware]) .handler(async ({ data }) => - toServerResult(() => pbAdmin.awardManualBadge(data.playerId, data.badgeId)) + toServerResult(async () => { + const result = await pbAdmin.awardManualBadge(data.playerId, data.badgeId); + emitServerEvent({ type: "badge", playerId: data.playerId }); + return result; + }) ); diff --git a/src/features/matches/server.ts b/src/features/matches/server.ts index 8ea403b..3052219 100644 --- a/src/features/matches/server.ts +++ b/src/features/matches/server.ts @@ -6,7 +6,7 @@ import { z } from "zod"; import { toServerResult } from "@/lib/tanstack-query/utils/to-server-result"; import brackets from "@/features/bracket/utils"; import { Match, MatchInput } from "@/features/matches/types"; -import { serverEvents } from "@/lib/events/emitter"; +import { emitServerEvent } from "@/lib/events/emitter"; import { superTokensFunctionMiddleware } from "@/utils/supertokens"; import { PlayerInfo } from "../players/types"; import { serverFnLoggingMiddleware } from "@/utils/activities"; @@ -129,6 +129,8 @@ export const generateTournamentBracket = createServerFn() matchCount: createdMatches.length, }); + emitServerEvent({ type: "tournament", tournamentId }); + return { tournament, matchCount: createdMatches.length, @@ -154,7 +156,7 @@ export const startMatch = createServerFn() status: "started", }); - serverEvents.emit("match", { + emitServerEvent({ type: "match", matchId: match.id, tournamentId: match.tournament.id @@ -180,7 +182,9 @@ export const populateKnockoutBracket = createServerFn() throw new Error("Tournament must have group_config"); } - return await populateKnockoutBracketInternal(tournamentId, tournament.group_config); + const result = await populateKnockoutBracketInternal(tournamentId, tournament.group_config); + emitServerEvent({ type: "tournament", tournamentId }); + return result; }) ); @@ -479,7 +483,7 @@ export const endMatch = createServerFn() }); if (match.lid === -1) { - serverEvents.emit("match", { + emitServerEvent({ type: "match", matchId: match.id, tournamentId: match.tournament.id @@ -504,6 +508,11 @@ export const endMatch = createServerFn() }); await pbAdmin.deleteMatch(winner.id); + emitServerEvent({ + type: "match", + matchId: match.id, + tournamentId: match.tournament.id + }); return match; } } @@ -530,7 +539,7 @@ export const endMatch = createServerFn() }); } - serverEvents.emit("match", { + emitServerEvent({ type: "match", matchId: match.id, tournamentId: match.tournament.id @@ -590,7 +599,7 @@ export const toggleMatchReaction = createServerFn() const reactions = Object.values(reactionsByEmoji); - serverEvents.emit("reaction", { + emitServerEvent({ type: "reaction", matchId, reactions, diff --git a/src/features/players/server.ts b/src/features/players/server.ts index e827240..2fcf8bb 100644 --- a/src/features/players/server.ts +++ b/src/features/players/server.ts @@ -68,6 +68,9 @@ export const updatePlayer = createServerFn() await setUserMetadata({ data: { first_name: data.first_name, last_name: data.last_name } }); + const { emitServerEvent } = await import("@/lib/events/emitter"); + emitServerEvent({ type: "player", playerId: existing.id! }); + return updatedPlayer; }) ); @@ -95,6 +98,12 @@ export const createPlayer = createServerFn() await setUserMetadata({ data: { first_name: data.first_name, last_name: data.last_name, player_id: newPlayer?.id?.toString() } }); logger.info('Created player', newPlayer); + + if (newPlayer?.id) { + const { emitServerEvent } = await import("@/lib/events/emitter"); + emitServerEvent({ type: "player", playerId: newPlayer.id }); + } + return newPlayer; }) ); @@ -123,6 +132,10 @@ export const associatePlayer = createServerFn() const player = await pbAdmin.getPlayer(data); logger.info('Associated player', player); + + const { emitServerEvent } = await import("@/lib/events/emitter"); + emitServerEvent({ type: "player", playerId: data }); + return player; }) ); diff --git a/src/features/teams/server.ts b/src/features/teams/server.ts index ae1dd18..9a3547e 100644 --- a/src/features/teams/server.ts +++ b/src/features/teams/server.ts @@ -7,6 +7,7 @@ import { teamInputSchema, teamUpdateSchema } from "./types"; import { logger } from "@/lib/logger"; import { Match } from "../matches/types"; import { serverFnLoggingMiddleware } from "@/utils/activities"; +import { emitServerEvent } from "@/lib/events/emitter"; export const listTeamInfos = createServerFn() @@ -42,7 +43,9 @@ export const createTeam = createServerFn() //} logger.info("Creating team", { name: data.name, userId, isAdmin }); - return pbAdmin.createTeam(data); + const team = await pbAdmin.createTeam(data); + emitServerEvent({ type: "team", teamId: team?.id }); + return team; }) ); @@ -68,7 +71,9 @@ export const updateTeam = createServerFn() // } logger.info("Updating team", { teamId: id, userId, isAdmin }); - return pbAdmin.updateTeam(id, updates); + const updated = await pbAdmin.updateTeam(id, updates); + emitServerEvent({ type: "team", teamId: id }); + return updated; }) ); diff --git a/src/features/tournaments/server.ts b/src/features/tournaments/server.ts index 520097a..ab9f523 100644 --- a/src/features/tournaments/server.ts +++ b/src/features/tournaments/server.ts @@ -6,6 +6,7 @@ import { logger } from "."; import { z } from "zod"; import { toServerResult } from "@/lib/tanstack-query/utils/to-server-result"; import { serverFnLoggingMiddleware } from "@/utils/activities"; +import { emitServerEvent } from "@/lib/events/emitter"; import brackets from "@/features/bracket/utils"; import { MatchInput } from "@/features/matches/types"; import { generateSingleEliminationBracket } from "./utils/bracket-generator"; @@ -19,8 +20,12 @@ export const listTournaments = createServerFn() export const createTournament = createServerFn() .validator(tournamentInputSchema) .middleware([superTokensAdminFunctionMiddleware, serverFnLoggingMiddleware]) - .handler(async ({ data }) => - toServerResult(() => pbAdmin.createTournament(data)) + .handler(async ({ data }) => + toServerResult(async () => { + const tournament = await pbAdmin.createTournament(data); + emitServerEvent({ type: "tournament", tournamentId: tournament.id }); + return tournament; + }) ); export const updateTournament = createServerFn() @@ -29,8 +34,12 @@ export const updateTournament = createServerFn() updates: tournamentInputSchema.partial() })) .middleware([superTokensAdminFunctionMiddleware, serverFnLoggingMiddleware]) - .handler(async ({ data }) => - toServerResult(() => pbAdmin.updateTournament(data.id, data.updates)) + .handler(async ({ data }) => + toServerResult(async () => { + const tournament = await pbAdmin.updateTournament(data.id, data.updates); + emitServerEvent({ type: "tournament", tournamentId: data.id }); + return tournament; + }) ); export const getTournament = createServerFn() @@ -76,6 +85,7 @@ export const enrollTeam = createServerFn() logger.info('Enrolling team in tournament', { tournamentId, teamId, userId }); const tournament = await pbAdmin.enrollTeam(tournamentId, teamId); + emitServerEvent({ type: "tournament", tournamentId }); return tournament; }) ); @@ -86,8 +96,12 @@ export const unenrollTeam = createServerFn() teamId: z.string() })) .middleware([superTokensFunctionMiddleware, serverFnLoggingMiddleware]) - .handler(async ({ data: { tournamentId, teamId }, context }) => - toServerResult(() => pbAdmin.unenrollTeam(tournamentId, teamId)) + .handler(async ({ data: { tournamentId, teamId }, context }) => + toServerResult(async () => { + const result = await pbAdmin.unenrollTeam(tournamentId, teamId); + emitServerEvent({ type: "tournament", tournamentId }); + return result; + }) ); export const getUnenrolledTeams = createServerFn() @@ -115,6 +129,7 @@ export const enrollFreeAgent = createServerFn() await pbAdmin.enrollFreeAgent(player.id, data.phone, data.tournamentId); logger.info('Player enrolled as free agent', { playerId: player.id, phone: data.phone }); + emitServerEvent({ type: "tournament", tournamentId: data.tournamentId }); }) ); @@ -129,6 +144,7 @@ export const unenrollFreeAgent = createServerFn() await pbAdmin.unenrollFreeAgent(player.id, data.tournamentId); logger.info('Player unenrolled as free agent', { playerId: player.id }); + emitServerEvent({ type: "tournament", tournamentId: data.tournamentId }); }) ); @@ -382,6 +398,9 @@ export const confirmTeamAssignments = createServerFn() newCount: createdTeams.length - reusedCount }); + emitServerEvent({ type: "team" }); + emitServerEvent({ type: "tournament", tournamentId: data.tournamentId }); + return { teams: createdTeams }; }) ); @@ -702,6 +721,8 @@ export const generateKnockoutBracket = createServerFn() qualifiedTeamCount: qualifiedTeams.length }); + emitServerEvent({ type: "tournament", tournamentId: data.tournamentId }); + return { tournament, matchCount: createdMatches.length, @@ -720,6 +741,7 @@ export const adminEnrollPlayer = createServerFn() toServerResult(async () => { await pbAdmin.enrollFreeAgent(data.playerId, "", data.tournamentId); logger.info('Admin enrolled player', { playerId: data.playerId, tournamentId: data.tournamentId }); + emitServerEvent({ type: "tournament", tournamentId: data.tournamentId }); }) ); @@ -733,6 +755,7 @@ export const adminUnenrollPlayer = createServerFn() toServerResult(async () => { await pbAdmin.unenrollFreeAgent(data.playerId, data.tournamentId); logger.info('Admin unenrolled player', { playerId: data.playerId, tournamentId: data.tournamentId }); + emitServerEvent({ type: "tournament", tournamentId: data.tournamentId }); }) ); @@ -881,6 +904,8 @@ export const generateGroupStage = createServerFn() totalMatchCount: createdMatches.length }); + emitServerEvent({ type: "tournament", tournamentId: data.tournamentId }); + return { tournament, groups: createdGroups, diff --git a/src/hooks/use-server-events.ts b/src/hooks/use-server-events.ts index 2fe15ba..40a4f26 100644 --- a/src/hooks/use-server-events.ts +++ b/src/hooks/use-server-events.ts @@ -1,9 +1,9 @@ -import { useEffect, useRef } from "react"; +import { useEffect } from "react"; import { useQueryClient } from "@tanstack/react-query"; import { Logger } from "@/lib/logger"; import { useAuth } from "@/contexts/auth-context"; -import { tournamentQueries } from "@/features/tournaments/queries"; -import { reactionKeys, reactionQueries } from "@/features/reactions/queries"; +import { tournamentKeys } from "@/features/tournaments/queries"; +import { reactionKeys } from "@/features/reactions/queries"; const logger = new Logger('ServerEvents'); @@ -12,11 +12,14 @@ type SSEEvent = { [key: string]: any; }; -type EventHandler = (event: SSEEvent, queryClient: ReturnType, currentSessionId?: string) => void; +type EventHandler = (event: SSEEvent, queryClient: ReturnType) => void; -const INVALIDATE_DEBOUNCE_MS = 1500; +const INVALIDATE_DEBOUNCE_MS = 1000; +const INVALIDATE_JITTER_MS = 1500; const invalidateTimers = new Map>(); +const WATCHDOG_MS = 45_000; + function debouncedInvalidate( queryClient: ReturnType, filters: { queryKey: readonly unknown[] } @@ -25,10 +28,11 @@ function debouncedInvalidate( const existing = invalidateTimers.get(key); if (existing) clearTimeout(existing); + const delay = INVALIDATE_DEBOUNCE_MS + Math.random() * INVALIDATE_JITTER_MS; invalidateTimers.set(key, setTimeout(() => { invalidateTimers.delete(key); queryClient.invalidateQueries(filters); - }, INVALIDATE_DEBOUNCE_MS)); + }, delay)); } function clearPendingInvalidations() { @@ -39,53 +43,107 @@ function clearPendingInvalidations() { } const eventHandlers: Record = { - "connected": () => { - logger.info("New Connection"); - }, "ping": () => {}, - "heartbeat": () => {}, + "test": (event) => { + logger.info("Test event", event); + }, + "tournament": (event, queryClient) => { + debouncedInvalidate(queryClient, { queryKey: ['tournaments'] }); + debouncedInvalidate(queryClient, { queryKey: ['players', 'unenrolled'] }); + }, "match": (event, queryClient) => { - debouncedInvalidate(queryClient, tournamentQueries.details(event.tournamentId)) - debouncedInvalidate(queryClient, tournamentQueries.current()) + debouncedInvalidate(queryClient, { queryKey: tournamentKeys.details(event.tournamentId) }); + debouncedInvalidate(queryClient, { queryKey: tournamentKeys.current }); + debouncedInvalidate(queryClient, { queryKey: ['players', 'stats'] }); + debouncedInvalidate(queryClient, { queryKey: ['players', 'matches'] }); + debouncedInvalidate(queryClient, { queryKey: ['players', 'activity'] }); + debouncedInvalidate(queryClient, { queryKey: ['teams', 'stats'] }); + debouncedInvalidate(queryClient, { queryKey: ['teams', 'matches'] }); + debouncedInvalidate(queryClient, { queryKey: ['matches'] }); }, "reaction": (event, queryClient) => { - queryClient.invalidateQueries(reactionQueries.match(event.matchId)); queryClient.setQueryData(reactionKeys.match(event.matchId), () => event.reactions); - } + }, + "team": (event, queryClient) => { + debouncedInvalidate(queryClient, { queryKey: ['teams'] }); + debouncedInvalidate(queryClient, { queryKey: ['tournaments'] }); + }, + "player": (event, queryClient) => { + debouncedInvalidate(queryClient, { queryKey: ['players'] }); + }, + "badge": (event, queryClient) => { + debouncedInvalidate(queryClient, { queryKey: ['badges'] }); + }, }; export function useServerEvents() { const queryClient = useQueryClient(); const { user } = useAuth(); - const retryCountRef = useRef(0); - const shouldConnectRef = useRef(true); - const timeoutRef = useRef(null); useEffect(() => { if (typeof window === 'undefined') return; if (!user?.id) return; - shouldConnectRef.current = true; - retryCountRef.current = 0; + let disposed = false; + let eventSource: EventSource | null = null; + let reconnectTimer: ReturnType | null = null; + let watchdogTimer: ReturnType | null = null; + let retryCount = 0; + let hasConnectedOnce = false; - const connectEventSource = () => { - if (!shouldConnectRef.current) return; + const disconnect = () => { + if (watchdogTimer) { clearTimeout(watchdogTimer); watchdogTimer = null; } + if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; } + eventSource?.close(); + eventSource = null; + }; + const reconnect = (delay: number) => { + if (disposed) return; + disconnect(); + reconnectTimer = setTimeout(connect, delay); + }; - const eventSource = new EventSource(`/api/events/$`); + const armWatchdog = () => { + if (watchdogTimer) clearTimeout(watchdogTimer); + watchdogTimer = setTimeout(() => { + logger.warn(`SSE watchdog: no messages in ${WATCHDOG_MS}ms, reconnecting`); + retryCount = 0; + reconnect(0); + }, WATCHDOG_MS); + }; + + const connect = () => { + if (disposed || eventSource) return; + + eventSource = new EventSource(`/api/events/$`); + armWatchdog(); eventSource.onopen = () => { - retryCountRef.current = 0; + retryCount = 0; }; eventSource.onmessage = (event) => { + armWatchdog(); try { const data: SSEEvent = JSON.parse(event.data); - logger.info("Event received", data); + if (data.type !== "ping") { + logger.info("Event received", data); + } + + if (data.type === "connected") { + if (hasConnectedOnce) { + setTimeout(() => { + if (!disposed) queryClient.invalidateQueries(); + }, Math.random() * 2000); + } + hasConnectedOnce = true; + return; + } const handler = eventHandlers[data.type]; if (handler) { - handler(data, queryClient, user?.id); + handler(data, queryClient); } else { logger.warn(`Unhandled SSE event type: ${data.type}`); } @@ -94,50 +152,51 @@ export function useServerEvents() { } }; - eventSource.onerror = (error) => { + eventSource.onerror = async (error) => { + if (disposed) return; logger.error("SSE connection error", error); - eventSource.close(); + disconnect(); - if (shouldConnectRef.current && retryCountRef.current < 10) { - retryCountRef.current += 1; - const delay = Math.min( - 1000 * Math.pow(1.5, retryCountRef.current - 1), - 15000 - ); + retryCount += 1; + const delay = Math.min(1000 * Math.pow(1.5, retryCount - 1), 15000); + logger.info(`SSE reconnection attempt ${retryCount} in ${Math.round(delay)}ms`); - logger.info( - `SSE reconnection attempt ${retryCountRef.current}/10 in ${delay}ms` - ); - - timeoutRef.current = setTimeout(() => { - if (shouldConnectRef.current) { - connectEventSource(); - } - }, delay); - } else if (retryCountRef.current >= 10) { - logger.error("SSE max reconnection attempts reached"); + try { + const { attemptRefreshingSession } = await import('supertokens-web-js/recipe/session'); + await attemptRefreshingSession(); + } catch { } - }; - return eventSource; + reconnect(delay); + }; }; - const eventSource = connectEventSource(); + const wake = () => { + if (disposed) return; + if (!eventSource || eventSource.readyState === EventSource.CLOSED) { + retryCount = 0; + reconnect(0); + } + }; + const handleVisibility = () => { + if (document.visibilityState === 'visible') wake(); + }; + + window.addEventListener('online', wake); + document.addEventListener('visibilitychange', handleVisibility); + + connect(); return () => { logger.info("Closing SSE connection"); - shouldConnectRef.current = false; + disposed = true; clearPendingInvalidations(); - if (timeoutRef.current) { - clearTimeout(timeoutRef.current); - timeoutRef.current = null; - } + window.removeEventListener('online', wake); + document.removeEventListener('visibilitychange', handleVisibility); - if (eventSource) { - eventSource.close(); - } + disconnect(); }; - }, [user?.id]); -} \ No newline at end of file + }, [user?.id, queryClient]); +} diff --git a/src/lib/events/emitter.ts b/src/lib/events/emitter.ts index 944153e..06ba7ae 100644 --- a/src/lib/events/emitter.ts +++ b/src/lib/events/emitter.ts @@ -1,38 +1,81 @@ import { EventEmitter } from "events"; -export const serverEvents = new EventEmitter(); - -serverEvents.setMaxListeners(50); - -// Debug logging for listener count -if (process.env.NODE_ENV === 'development') { - setInterval(() => { - const listenerCounts = { - test: serverEvents.listenerCount('test'), - match: serverEvents.listenerCount('match'), - reaction: serverEvents.listenerCount('reaction'), - }; - - if (listenerCounts.test > 0 || listenerCounts.match > 0 || listenerCounts.reaction > 0) { - console.log('ServerEvents listener count:', listenerCounts); - } - }, 30000); // Log every 30 seconds in development -} - export type TestEvent = { type: "test"; - playerId: string; + userId: string; }; export type MatchEvent = { type: "match"; matchId: string; tournamentId: string; -} +}; export type ReactionEvent = { type: "reaction"; matchId: string; -} + reactions: Array<{ + emoji: string; + count: number; + players: Array<{ id?: string; first_name?: string; last_name?: string }>; + }>; +}; -export type ServerEvent = TestEvent | MatchEvent | ReactionEvent; +export type TournamentEvent = { + type: "tournament"; + tournamentId: string; +}; + +export type TeamEvent = { + type: "team"; + teamId?: string; +}; + +export type PlayerEvent = { + type: "player"; + playerId: string; +}; + +export type BadgeEvent = { + type: "badge"; + playerId?: string; +}; + +export type ServerEvent = + | TestEvent + | MatchEvent + | ReactionEvent + | TournamentEvent + | TeamEvent + | PlayerEvent + | BadgeEvent; + +export const EVENT_TYPES = [ + "test", + "match", + "reaction", + "tournament", + "team", + "player", + "badge", +] as const satisfies readonly ServerEvent["type"][]; + +export const serverEvents = new EventEmitter(); + +serverEvents.setMaxListeners(200); + +export const emitServerEvent = (event: ServerEvent) => { + serverEvents.emit(event.type, event); +}; + +if (process.env.NODE_ENV === 'development') { + setInterval(() => { + const listenerCounts = Object.fromEntries( + EVENT_TYPES.map((type) => [type, serverEvents.listenerCount(type)]) + ); + + if (Object.values(listenerCounts).some((count) => count > 0)) { + console.log('ServerEvents listener count:', listenerCounts); + } + }, 30000); +} diff --git a/src/utils/test-event.ts b/src/utils/test-event.ts index 283ee24..81b8b28 100644 --- a/src/utils/test-event.ts +++ b/src/utils/test-event.ts @@ -1,11 +1,11 @@ import { createServerFn } from "@tanstack/react-start"; import { superTokensFunctionMiddleware } from "./supertokens"; -import { serverEvents } from "@/lib/events/emitter"; +import { emitServerEvent } from "@/lib/events/emitter"; export const testEvent = createServerFn() .middleware([superTokensFunctionMiddleware]) .handler(async ({ context }) => { - serverEvents.emit("test", { + emitServerEvent({ type: "test", userId: context.userAuthId, });