This commit is contained in:
yohlo
2026-07-13 23:56:19 -07:00
parent 778f0f7994
commit 19f61ac454
10 changed files with 287 additions and 145 deletions
+1 -1
View File
@@ -6,7 +6,7 @@ metadata:
app: flxn app: flxn
component: app component: app
spec: spec:
replicas: 1 replicas: 1 # Must stay at 1 for SSE
selector: selector:
matchLabels: matchLabels:
app: flxn app: flxn
+25 -46
View File
@@ -1,9 +1,10 @@
import { createFileRoute } from "@tanstack/react-router"; 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 { logger } from "@/lib/logger";
import { superTokensRequestMiddleware } from "@/utils/supertokens"; import { superTokensRequestMiddleware } from "@/utils/supertokens";
let activeConnections = 0; let activeConnections = 0;
const encoder = new TextEncoder();
export const Route = createFileRoute("/api/events/$")({ export const Route = createFileRoute("/api/events/$")({
server: { server: {
@@ -13,63 +14,47 @@ export const Route = createFileRoute("/api/events/$")({
activeConnections++; activeConnections++;
const connectionId = `conn_${Date.now()}_${Math.random().toString(36).substring(2, 11)}`; const connectionId = `conn_${Date.now()}_${Math.random().toString(36).substring(2, 11)}`;
logger.info(`ServerEvents | New connection ${connectionId}. Active: ${activeConnections}`); logger.info(`ServerEvents | New connection ${connectionId}. Active: ${activeConnections}`);
let cleanedUp = false;
let cleanup = () => {};
const stream = new ReadableStream({ const stream = new ReadableStream({
start(controller) { start(controller) {
const connectMessage = `data: ${JSON.stringify({ type: "connected" })}\n\n`; const send = (payload: unknown) => {
controller.enqueue(new TextEncoder().encode(connectMessage));
const handleEvent = (event: ServerEvent) => {
logger.info("ServerEvents | Event received", event);
const message = `data: ${JSON.stringify(event)}\n\n`;
try { try {
if (!controller.desiredSize || controller.desiredSize <= 0) { controller.enqueue(encoder.encode(`data: ${JSON.stringify(payload)}\n\n`));
logger.warn("ServerEvents | Stream closed, skipping event");
return;
}
controller.enqueue(new TextEncoder().encode(message));
} catch (error) { } catch (error) {
logger.error("ServerEvents | Error sending SSE message", error); logger.error("ServerEvents | Error sending SSE message", error);
cleanup();
} }
}; };
serverEvents.on("test", handleEvent); const handleEvent = (event: ServerEvent) => send(event);
serverEvents.on("match", handleEvent); for (const type of EVENT_TYPES) {
serverEvents.on("reaction", handleEvent); serverEvents.on(type, handleEvent);
}
const pingInterval = setInterval(() => { const pingInterval = setInterval(() => {
try { send({ type: "ping", timestamp: Date.now() });
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);
}
}, 15000); }, 15000);
setTimeout(() => { cleanup = () => {
try { if (cleanedUp) return;
const heartbeatMessage = `data: ${JSON.stringify({ type: "heartbeat", timestamp: Date.now() })}\n\n`; cleanedUp = true;
controller.enqueue(new TextEncoder().encode(heartbeatMessage));
} catch (e) {
logger.error("ServerEvents | Heartbeat error", e);
}
}, 1000);
const cleanup = () => {
activeConnections--; activeConnections--;
serverEvents.off("test", handleEvent); for (const type of EVENT_TYPES) {
serverEvents.off("match", handleEvent); serverEvents.off(type, handleEvent);
serverEvents.off("reaction", handleEvent); }
clearInterval(pingInterval); clearInterval(pingInterval);
logger.info(`ServerEvents | Connection ${connectionId} cleanup completed. Active: ${activeConnections}`); logger.info(`ServerEvents | Connection ${connectionId} cleanup completed. Active: ${activeConnections}`);
}; };
request.signal?.addEventListener("abort", cleanup); request.signal?.addEventListener("abort", cleanup);
return cleanup;
send({ type: "connected" });
},
cancel() {
cleanup();
}, },
}); });
@@ -77,13 +62,7 @@ export const Route = createFileRoute("/api/events/$")({
headers: { headers: {
"Content-Type": "text/event-stream", "Content-Type": "text/event-stream",
"Cache-Control": "no-cache, no-store, must-revalidate", "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-Accel-Buffering": "no",
"X-Proxy-Buffering": "no",
"Proxy-Buffering": "off",
"Transfer-Encoding": "chunked",
}, },
}); });
}, },
+11 -2
View File
@@ -3,6 +3,7 @@ import { superTokensAdminFunctionMiddleware, superTokensFunctionMiddleware } fro
import { createServerFn } from "@tanstack/react-start"; import { createServerFn } from "@tanstack/react-start";
import { pbAdmin } from "@/lib/pocketbase/client"; import { pbAdmin } from "@/lib/pocketbase/client";
import { z } from "zod"; import { z } from "zod";
import { emitServerEvent } from "@/lib/events/emitter";
export const getPlayerBadges = createServerFn() export const getPlayerBadges = createServerFn()
.validator(z.string()) .validator(z.string())
@@ -14,7 +15,11 @@ export const getPlayerBadges = createServerFn()
export const migrateBadgeProgress = createServerFn() export const migrateBadgeProgress = createServerFn()
.middleware([superTokensAdminFunctionMiddleware]) .middleware([superTokensAdminFunctionMiddleware])
.handler(async () => .handler(async () =>
toServerResult(() => pbAdmin.migrateBadgeProgress()) toServerResult(async () => {
const result = await pbAdmin.migrateBadgeProgress();
emitServerEvent({ type: "badge" });
return result;
})
); );
export const getAllBadges = createServerFn() export const getAllBadges = createServerFn()
@@ -34,5 +39,9 @@ export const awardManualBadge = createServerFn()
})) }))
.middleware([superTokensAdminFunctionMiddleware]) .middleware([superTokensAdminFunctionMiddleware])
.handler(async ({ data }) => .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;
})
); );
+15 -6
View File
@@ -6,7 +6,7 @@ import { z } from "zod";
import { toServerResult } from "@/lib/tanstack-query/utils/to-server-result"; import { toServerResult } from "@/lib/tanstack-query/utils/to-server-result";
import brackets from "@/features/bracket/utils"; import brackets from "@/features/bracket/utils";
import { Match, MatchInput } from "@/features/matches/types"; 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 { superTokensFunctionMiddleware } from "@/utils/supertokens";
import { PlayerInfo } from "../players/types"; import { PlayerInfo } from "../players/types";
import { serverFnLoggingMiddleware } from "@/utils/activities"; import { serverFnLoggingMiddleware } from "@/utils/activities";
@@ -129,6 +129,8 @@ export const generateTournamentBracket = createServerFn()
matchCount: createdMatches.length, matchCount: createdMatches.length,
}); });
emitServerEvent({ type: "tournament", tournamentId });
return { return {
tournament, tournament,
matchCount: createdMatches.length, matchCount: createdMatches.length,
@@ -154,7 +156,7 @@ export const startMatch = createServerFn()
status: "started", status: "started",
}); });
serverEvents.emit("match", { emitServerEvent({
type: "match", type: "match",
matchId: match.id, matchId: match.id,
tournamentId: match.tournament.id tournamentId: match.tournament.id
@@ -180,7 +182,9 @@ export const populateKnockoutBracket = createServerFn()
throw new Error("Tournament must have group_config"); 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) { if (match.lid === -1) {
serverEvents.emit("match", { emitServerEvent({
type: "match", type: "match",
matchId: match.id, matchId: match.id,
tournamentId: match.tournament.id tournamentId: match.tournament.id
@@ -504,6 +508,11 @@ export const endMatch = createServerFn()
}); });
await pbAdmin.deleteMatch(winner.id); await pbAdmin.deleteMatch(winner.id);
emitServerEvent({
type: "match",
matchId: match.id,
tournamentId: match.tournament.id
});
return match; return match;
} }
} }
@@ -530,7 +539,7 @@ export const endMatch = createServerFn()
}); });
} }
serverEvents.emit("match", { emitServerEvent({
type: "match", type: "match",
matchId: match.id, matchId: match.id,
tournamentId: match.tournament.id tournamentId: match.tournament.id
@@ -590,7 +599,7 @@ export const toggleMatchReaction = createServerFn()
const reactions = Object.values(reactionsByEmoji); const reactions = Object.values(reactionsByEmoji);
serverEvents.emit("reaction", { emitServerEvent({
type: "reaction", type: "reaction",
matchId, matchId,
reactions, reactions,
+13
View File
@@ -68,6 +68,9 @@ export const updatePlayer = createServerFn()
await setUserMetadata({ data: { first_name: data.first_name, last_name: data.last_name } }); 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; 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() } }); await setUserMetadata({ data: { first_name: data.first_name, last_name: data.last_name, player_id: newPlayer?.id?.toString() } });
logger.info('Created player', newPlayer); logger.info('Created player', newPlayer);
if (newPlayer?.id) {
const { emitServerEvent } = await import("@/lib/events/emitter");
emitServerEvent({ type: "player", playerId: newPlayer.id });
}
return newPlayer; return newPlayer;
}) })
); );
@@ -123,6 +132,10 @@ export const associatePlayer = createServerFn()
const player = await pbAdmin.getPlayer(data); const player = await pbAdmin.getPlayer(data);
logger.info('Associated player', player); logger.info('Associated player', player);
const { emitServerEvent } = await import("@/lib/events/emitter");
emitServerEvent({ type: "player", playerId: data });
return player; return player;
}) })
); );
+7 -2
View File
@@ -7,6 +7,7 @@ import { teamInputSchema, teamUpdateSchema } from "./types";
import { logger } from "@/lib/logger"; import { logger } from "@/lib/logger";
import { Match } from "../matches/types"; import { Match } from "../matches/types";
import { serverFnLoggingMiddleware } from "@/utils/activities"; import { serverFnLoggingMiddleware } from "@/utils/activities";
import { emitServerEvent } from "@/lib/events/emitter";
export const listTeamInfos = createServerFn() export const listTeamInfos = createServerFn()
@@ -42,7 +43,9 @@ export const createTeam = createServerFn()
//} //}
logger.info("Creating team", { name: data.name, userId, isAdmin }); 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 }); 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;
}) })
); );
+28 -3
View File
@@ -6,6 +6,7 @@ import { logger } from ".";
import { z } from "zod"; import { z } from "zod";
import { toServerResult } from "@/lib/tanstack-query/utils/to-server-result"; import { toServerResult } from "@/lib/tanstack-query/utils/to-server-result";
import { serverFnLoggingMiddleware } from "@/utils/activities"; import { serverFnLoggingMiddleware } from "@/utils/activities";
import { emitServerEvent } from "@/lib/events/emitter";
import brackets from "@/features/bracket/utils"; import brackets from "@/features/bracket/utils";
import { MatchInput } from "@/features/matches/types"; import { MatchInput } from "@/features/matches/types";
import { generateSingleEliminationBracket } from "./utils/bracket-generator"; import { generateSingleEliminationBracket } from "./utils/bracket-generator";
@@ -20,7 +21,11 @@ export const createTournament = createServerFn()
.validator(tournamentInputSchema) .validator(tournamentInputSchema)
.middleware([superTokensAdminFunctionMiddleware, serverFnLoggingMiddleware]) .middleware([superTokensAdminFunctionMiddleware, serverFnLoggingMiddleware])
.handler(async ({ data }) => .handler(async ({ data }) =>
toServerResult(() => pbAdmin.createTournament(data)) toServerResult(async () => {
const tournament = await pbAdmin.createTournament(data);
emitServerEvent({ type: "tournament", tournamentId: tournament.id });
return tournament;
})
); );
export const updateTournament = createServerFn() export const updateTournament = createServerFn()
@@ -30,7 +35,11 @@ export const updateTournament = createServerFn()
})) }))
.middleware([superTokensAdminFunctionMiddleware, serverFnLoggingMiddleware]) .middleware([superTokensAdminFunctionMiddleware, serverFnLoggingMiddleware])
.handler(async ({ data }) => .handler(async ({ data }) =>
toServerResult(() => pbAdmin.updateTournament(data.id, data.updates)) toServerResult(async () => {
const tournament = await pbAdmin.updateTournament(data.id, data.updates);
emitServerEvent({ type: "tournament", tournamentId: data.id });
return tournament;
})
); );
export const getTournament = createServerFn() export const getTournament = createServerFn()
@@ -76,6 +85,7 @@ export const enrollTeam = createServerFn()
logger.info('Enrolling team in tournament', { tournamentId, teamId, userId }); logger.info('Enrolling team in tournament', { tournamentId, teamId, userId });
const tournament = await pbAdmin.enrollTeam(tournamentId, teamId); const tournament = await pbAdmin.enrollTeam(tournamentId, teamId);
emitServerEvent({ type: "tournament", tournamentId });
return tournament; return tournament;
}) })
); );
@@ -87,7 +97,11 @@ export const unenrollTeam = createServerFn()
})) }))
.middleware([superTokensFunctionMiddleware, serverFnLoggingMiddleware]) .middleware([superTokensFunctionMiddleware, serverFnLoggingMiddleware])
.handler(async ({ data: { tournamentId, teamId }, context }) => .handler(async ({ data: { tournamentId, teamId }, context }) =>
toServerResult(() => pbAdmin.unenrollTeam(tournamentId, teamId)) toServerResult(async () => {
const result = await pbAdmin.unenrollTeam(tournamentId, teamId);
emitServerEvent({ type: "tournament", tournamentId });
return result;
})
); );
export const getUnenrolledTeams = createServerFn() export const getUnenrolledTeams = createServerFn()
@@ -115,6 +129,7 @@ export const enrollFreeAgent = createServerFn()
await pbAdmin.enrollFreeAgent(player.id, data.phone, data.tournamentId); await pbAdmin.enrollFreeAgent(player.id, data.phone, data.tournamentId);
logger.info('Player enrolled as free agent', { playerId: player.id, phone: data.phone }); 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); await pbAdmin.unenrollFreeAgent(player.id, data.tournamentId);
logger.info('Player unenrolled as free agent', { playerId: player.id }); 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 newCount: createdTeams.length - reusedCount
}); });
emitServerEvent({ type: "team" });
emitServerEvent({ type: "tournament", tournamentId: data.tournamentId });
return { teams: createdTeams }; return { teams: createdTeams };
}) })
); );
@@ -702,6 +721,8 @@ export const generateKnockoutBracket = createServerFn()
qualifiedTeamCount: qualifiedTeams.length qualifiedTeamCount: qualifiedTeams.length
}); });
emitServerEvent({ type: "tournament", tournamentId: data.tournamentId });
return { return {
tournament, tournament,
matchCount: createdMatches.length, matchCount: createdMatches.length,
@@ -720,6 +741,7 @@ export const adminEnrollPlayer = createServerFn()
toServerResult(async () => { toServerResult(async () => {
await pbAdmin.enrollFreeAgent(data.playerId, "", data.tournamentId); await pbAdmin.enrollFreeAgent(data.playerId, "", data.tournamentId);
logger.info('Admin enrolled player', { playerId: data.playerId, tournamentId: 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 () => { toServerResult(async () => {
await pbAdmin.unenrollFreeAgent(data.playerId, data.tournamentId); await pbAdmin.unenrollFreeAgent(data.playerId, data.tournamentId);
logger.info('Admin unenrolled player', { playerId: data.playerId, tournamentId: 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 totalMatchCount: createdMatches.length
}); });
emitServerEvent({ type: "tournament", tournamentId: data.tournamentId });
return { return {
tournament, tournament,
groups: createdGroups, groups: createdGroups,
+113 -54
View File
@@ -1,9 +1,9 @@
import { useEffect, useRef } from "react"; import { useEffect } from "react";
import { useQueryClient } from "@tanstack/react-query"; import { useQueryClient } from "@tanstack/react-query";
import { Logger } from "@/lib/logger"; import { Logger } from "@/lib/logger";
import { useAuth } from "@/contexts/auth-context"; import { useAuth } from "@/contexts/auth-context";
import { tournamentQueries } from "@/features/tournaments/queries"; import { tournamentKeys } from "@/features/tournaments/queries";
import { reactionKeys, reactionQueries } from "@/features/reactions/queries"; import { reactionKeys } from "@/features/reactions/queries";
const logger = new Logger('ServerEvents'); const logger = new Logger('ServerEvents');
@@ -12,11 +12,14 @@ type SSEEvent = {
[key: string]: any; [key: string]: any;
}; };
type EventHandler = (event: SSEEvent, queryClient: ReturnType<typeof useQueryClient>, currentSessionId?: string) => void; type EventHandler = (event: SSEEvent, queryClient: ReturnType<typeof useQueryClient>) => void;
const INVALIDATE_DEBOUNCE_MS = 1500; const INVALIDATE_DEBOUNCE_MS = 1000;
const INVALIDATE_JITTER_MS = 1500;
const invalidateTimers = new Map<string, ReturnType<typeof setTimeout>>(); const invalidateTimers = new Map<string, ReturnType<typeof setTimeout>>();
const WATCHDOG_MS = 45_000;
function debouncedInvalidate( function debouncedInvalidate(
queryClient: ReturnType<typeof useQueryClient>, queryClient: ReturnType<typeof useQueryClient>,
filters: { queryKey: readonly unknown[] } filters: { queryKey: readonly unknown[] }
@@ -25,10 +28,11 @@ function debouncedInvalidate(
const existing = invalidateTimers.get(key); const existing = invalidateTimers.get(key);
if (existing) clearTimeout(existing); if (existing) clearTimeout(existing);
const delay = INVALIDATE_DEBOUNCE_MS + Math.random() * INVALIDATE_JITTER_MS;
invalidateTimers.set(key, setTimeout(() => { invalidateTimers.set(key, setTimeout(() => {
invalidateTimers.delete(key); invalidateTimers.delete(key);
queryClient.invalidateQueries(filters); queryClient.invalidateQueries(filters);
}, INVALIDATE_DEBOUNCE_MS)); }, delay));
} }
function clearPendingInvalidations() { function clearPendingInvalidations() {
@@ -39,53 +43,107 @@ function clearPendingInvalidations() {
} }
const eventHandlers: Record<string, EventHandler> = { const eventHandlers: Record<string, EventHandler> = {
"connected": () => {
logger.info("New Connection");
},
"ping": () => {}, "ping": () => {},
"heartbeat": () => {}, "test": (event) => {
logger.info("Test event", event);
},
"tournament": (event, queryClient) => {
debouncedInvalidate(queryClient, { queryKey: ['tournaments'] });
debouncedInvalidate(queryClient, { queryKey: ['players', 'unenrolled'] });
},
"match": (event, queryClient) => { "match": (event, queryClient) => {
debouncedInvalidate(queryClient, tournamentQueries.details(event.tournamentId)) debouncedInvalidate(queryClient, { queryKey: tournamentKeys.details(event.tournamentId) });
debouncedInvalidate(queryClient, tournamentQueries.current()) 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) => { "reaction": (event, queryClient) => {
queryClient.invalidateQueries(reactionQueries.match(event.matchId));
queryClient.setQueryData(reactionKeys.match(event.matchId), () => event.reactions); 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() { export function useServerEvents() {
const queryClient = useQueryClient(); const queryClient = useQueryClient();
const { user } = useAuth(); const { user } = useAuth();
const retryCountRef = useRef(0);
const shouldConnectRef = useRef(true);
const timeoutRef = useRef<NodeJS.Timeout | null>(null);
useEffect(() => { useEffect(() => {
if (typeof window === 'undefined') return; if (typeof window === 'undefined') return;
if (!user?.id) return; if (!user?.id) return;
shouldConnectRef.current = true; let disposed = false;
retryCountRef.current = 0; let eventSource: EventSource | null = null;
let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
let watchdogTimer: ReturnType<typeof setTimeout> | null = null;
let retryCount = 0;
let hasConnectedOnce = false;
const connectEventSource = () => { const disconnect = () => {
if (!shouldConnectRef.current) return; 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 = () => { eventSource.onopen = () => {
retryCountRef.current = 0; retryCount = 0;
}; };
eventSource.onmessage = (event) => { eventSource.onmessage = (event) => {
armWatchdog();
try { try {
const data: SSEEvent = JSON.parse(event.data); const data: SSEEvent = JSON.parse(event.data);
if (data.type !== "ping") {
logger.info("Event received", data); 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]; const handler = eventHandlers[data.type];
if (handler) { if (handler) {
handler(data, queryClient, user?.id); handler(data, queryClient);
} else { } else {
logger.warn(`Unhandled SSE event type: ${data.type}`); 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); logger.error("SSE connection error", error);
eventSource.close(); disconnect();
if (shouldConnectRef.current && retryCountRef.current < 10) { retryCount += 1;
retryCountRef.current += 1; const delay = Math.min(1000 * Math.pow(1.5, retryCount - 1), 15000);
const delay = Math.min( logger.info(`SSE reconnection attempt ${retryCount} in ${Math.round(delay)}ms`);
1000 * Math.pow(1.5, retryCountRef.current - 1),
15000
);
logger.info( try {
`SSE reconnection attempt ${retryCountRef.current}/10 in ${delay}ms` const { attemptRefreshingSession } = await import('supertokens-web-js/recipe/session');
); await attemptRefreshingSession();
} catch {
timeoutRef.current = setTimeout(() => {
if (shouldConnectRef.current) {
connectEventSource();
}
}, delay);
} else if (retryCountRef.current >= 10) {
logger.error("SSE max reconnection attempts reached");
} }
reconnect(delay);
};
}; };
return eventSource; const wake = () => {
if (disposed) return;
if (!eventSource || eventSource.readyState === EventSource.CLOSED) {
retryCount = 0;
reconnect(0);
}
};
const handleVisibility = () => {
if (document.visibilityState === 'visible') wake();
}; };
const eventSource = connectEventSource(); window.addEventListener('online', wake);
document.addEventListener('visibilitychange', handleVisibility);
connect();
return () => { return () => {
logger.info("Closing SSE connection"); logger.info("Closing SSE connection");
shouldConnectRef.current = false; disposed = true;
clearPendingInvalidations(); clearPendingInvalidations();
if (timeoutRef.current) { window.removeEventListener('online', wake);
clearTimeout(timeoutRef.current); document.removeEventListener('visibilitychange', handleVisibility);
timeoutRef.current = null;
}
if (eventSource) { disconnect();
eventSource.close();
}
}; };
}, [user?.id]); }, [user?.id, queryClient]);
} }
+66 -23
View File
@@ -1,38 +1,81 @@
import { EventEmitter } from "events"; 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 = { export type TestEvent = {
type: "test"; type: "test";
playerId: string; userId: string;
}; };
export type MatchEvent = { export type MatchEvent = {
type: "match"; type: "match";
matchId: string; matchId: string;
tournamentId: string; tournamentId: string;
} };
export type ReactionEvent = { export type ReactionEvent = {
type: "reaction"; type: "reaction";
matchId: string; 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);
}
+2 -2
View File
@@ -1,11 +1,11 @@
import { createServerFn } from "@tanstack/react-start"; import { createServerFn } from "@tanstack/react-start";
import { superTokensFunctionMiddleware } from "./supertokens"; import { superTokensFunctionMiddleware } from "./supertokens";
import { serverEvents } from "@/lib/events/emitter"; import { emitServerEvent } from "@/lib/events/emitter";
export const testEvent = createServerFn() export const testEvent = createServerFn()
.middleware([superTokensFunctionMiddleware]) .middleware([superTokensFunctionMiddleware])
.handler(async ({ context }) => { .handler(async ({ context }) => {
serverEvents.emit("test", { emitServerEvent({
type: "test", type: "test",
userId: context.userAuthId, userId: context.userAuthId,
}); });