import type { StatusPlayer, GameObservation } from '../../server-observation.js'; import { EventEmitter } from 'node:events'; import type { GameTelemetry } from '../../telemetry.js'; import type { RconEndpoint, RconFailure } from '../integrations/rcon.js'; import type { Health } from '../fleet/health.js'; // One in-process event bus, multiplexed by the SSE endpoints. Kept separate // from any specific subscriber so push/beta/sweep can emit without pulling // the whole worker module in. export interface ServerStatusEntry { server: string; containerState?: string | null; containerCheckedAt?: string | null; telemetryAfter?: string | null; rconStatus?: GameObservation; telemetry?: GameTelemetry; players: number | null; capacity?: number | null; capacityObservedAt?: string | null; map: string | null; gametype: string | null; roster: StatusPlayer[] | null; health: Health; lastSeenAt: string | null; lastError: string | null; endpoint: RconEndpoint | null; reason: RconFailure | null; replySnippet: string | null; activeSince: string | null; inactiveSince: string | null; gameStartedAt: string | null; gameEndedAt: string | null; } export type BetaStage = | 'queued' | 'observing' | 'inconclusive' | 'deploying' | 'awaiting-empty' | 'restarting' | 'awaiting-health' | 'awaiting-match' | 'ready' | 'failed'; export type BetaState = 'running' | 'ready-promote' | 'promoted' | 'rolled-back'; export type PipelineEvent = | { type: 'beta-changed'; betaId: number } | { type: 'fleet-invalidated' } | { type: 'server-telemetry'; entries: GameTelemetry[] } | { type: 'deployment-opened'; deploymentId: string; action: string; servers: string[] } | { type: 'server-changed'; deploymentId: string; server: string; files: number } | { type: 'server-state'; deploymentId: string; server: string; state: 'pending' | 'done' | 'failed' | 'held' | 'manual'; players: number | null; error: string | null; } | { type: 'sweep-tick'; frozen: boolean; reason?: string } | { type: 'server-status'; entries: ServerStatusEntry[]; delta?: boolean } | { type: 'beta-opened'; betaId: number; scriptIds: number[]; servers: string[] } | { type: 'beta-server-stage'; betaId: number; server: string; stage: BetaStage; error: string | null; } | { type: 'beta-state'; betaId: number; state: BetaState }; class Bus extends EventEmitter { constructor() { super(); // Deployments and sweep can produce a lot of events during a big push. this.setMaxListeners(64); } } const bus = new Bus(); const changedBetas = new Set(); export function invalidateBeta(betaId: number): void { const scheduled = changedBetas.size > 0; changedBetas.add(betaId); if (!scheduled) queueMicrotask(() => { const ids = [...changedBetas]; changedBetas.clear(); for (const betaId of ids) bus.emit('event', { type: 'beta-changed', betaId }); }); } let invalidationPending = false; export function invalidateFleet(): void { if (invalidationPending) return; invalidationPending = true; // Flush after nested database transactions have committed. queueMicrotask(() => { invalidationPending = false; bus.emit('event', { type: 'fleet-invalidated' }); }); } export function emit(event: PipelineEvent): void { bus.emit('event', event); if (event.type === 'beta-opened' || event.type === 'beta-state' || event.type === 'beta-server-stage') invalidateFleet(); if ('betaId' in event && event.type !== 'beta-changed') invalidateBeta(event.betaId); } export function subscribe(listener: (event: PipelineEvent) => void): () => void { bus.on('event', listener); return () => bus.off('event', listener); }