server for refapp and refbot and other stuff
feat: WebSocket task-progress channel for long-running jobs
+160
-14
+48
-11
src/engine/pools.ts
+48
-11
src/engine/pools.ts
···
1
+
import crypto from 'node:crypto';
1
2
import { GeneratedPools } from '../models/index.js';
2
3
import { HttpError } from '../errors.js';
3
4
import { getActiveEvent } from './event.js';
···
5
6
import { extractSongId, fetchAndCacheChart } from '../external/spinshare.js';
6
7
import { fetchOpenSets, type StartggSet } from '../external/startgg.js';
7
8
import { generatePools, type TierChart } from '../util/poolGenerator.js';
9
+
import { emitTaskProgress } from '../ws.js';
8
10
import { logger } from '../logger.js';
9
11
10
12
let generatedPools: {
···
218
220
}
219
221
220
222
// Fetches chart metadata from SpinShare in the background (non-blocking),
221
-
// mirroring the bot's fire-and-forget fetchAllChartData.
222
-
export function backgroundFetchCharts(tierPools: Record<string, { songId: number | null }[]>): void {
223
+
// mirroring the bot's fire-and-forget fetchAllChartData. Emits per-chart
224
+
// progress over the task WS channel when a taskId is supplied.
225
+
export function backgroundFetchCharts(
226
+
tierPools: Record<string, { songId: number | null }[]>,
227
+
taskId?: string,
228
+
): void {
223
229
const songIds = [...new Set(Object.values(tierPools).flat().map((c) => c.songId).filter((id): id is number => id != null))];
230
+
const total = songIds.length;
224
231
let done = 0;
232
+
if (taskId && total > 0) {
233
+
emitTaskProgress({ taskId, stage: 'charts', message: 'Fetching chart metadata', current: 0, total, done: false });
234
+
}
225
235
const next = () => {
226
-
if (done >= songIds.length) return;
236
+
if (done >= songIds.length) {
237
+
if (taskId) emitTaskProgress({ taskId, stage: 'complete', message: 'Done', current: total, total, done: true });
238
+
return;
239
+
}
227
240
const songId = songIds[done++]!;
228
241
fetchAndCacheChart(songId)
229
242
.catch((err) => logger.warn({ songId, err }, 'background chart fetch failed'))
230
-
.finally(() => setTimeout(next, 150));
243
+
.finally(() => {
244
+
if (taskId) emitTaskProgress({ taskId, stage: 'charts', message: 'Fetching chart metadata', current: done, total, done: false });
245
+
setTimeout(next, 150);
246
+
});
231
247
};
232
248
next();
233
249
}
234
250
235
251
export interface GenerateResult {
252
+
taskId: string;
236
253
bracketCount: number;
237
254
tierCounts: Record<string, number>;
238
255
chartCount: number;
239
256
}
240
257
241
258
// Full generate pipeline: fetch sheet CSV -> start.gg open sets -> generatePools -> save.
259
+
// Emits progress over the task WS channel via the returned taskId. The HTTP
260
+
// response returns as soon as pools are saved; the background chart fetch
261
+
// continues streaming progress events until complete.
242
262
export async function generatePoolsFromSheet(
243
263
sheetUrl: string,
244
264
tournamentSlug?: string | null,
245
265
): Promise<GenerateResult> {
266
+
const taskId = crypto.randomUUID();
246
267
const event = getActiveEvent();
247
-
if (!event) throw new HttpError(400, 'no_active_event', 'No active event! Activate an event first.');
268
+
if (!event) {
269
+
emitTaskProgress({ taskId, stage: 'error', error: 'No active event! Activate an event first.', done: true });
270
+
throw new HttpError(400, 'no_active_event', 'No active event! Activate an event first.');
271
+
}
248
272
const slug = tournamentSlug ?? event.tournamentSlug;
249
-
if (!slug) throw new HttpError(400, 'no_slug', 'Active event has no tournamentSlug set.');
273
+
if (!slug) {
274
+
emitTaskProgress({ taskId, stage: 'error', error: 'Active event has no tournamentSlug set.', done: true });
275
+
throw new HttpError(400, 'no_slug', 'Active event has no tournamentSlug set.');
276
+
}
277
+
278
+
emitTaskProgress({ taskId, stage: 'sheet', message: 'Fetching spreadsheet', current: 0, total: 4, done: false });
250
279
251
280
const csvUrl = toCsvUrl(sheetUrl);
252
281
let res: Response;
···
254
283
res = await fetch(csvUrl);
255
284
}
256
285
catch (err) {
286
+
emitTaskProgress({ taskId, stage: 'error', error: `Failed to fetch sheet: ${(err as Error).message}`, done: true });
257
287
throw new HttpError(400, 'sheet_fetch_failed', `Failed to fetch sheet: ${(err as Error).message}`);
258
288
}
259
-
if (!res.ok) throw new HttpError(400, 'sheet_fetch_failed', `Failed to fetch sheet: ${res.status} ${res.statusText}`);
289
+
if (!res.ok) {
290
+
emitTaskProgress({ taskId, stage: 'error', error: `Failed to fetch sheet: ${res.status} ${res.statusText}`, done: true });
291
+
throw new HttpError(400, 'sheet_fetch_failed', `Failed to fetch sheet: ${res.status} ${res.statusText}`);
292
+
}
260
293
const text = await res.text();
294
+
295
+
emitTaskProgress({ taskId, stage: 'sets', message: 'Fetching start.gg open sets', current: 1, total: 4, done: false });
296
+
const sets = await fetchOpenSets(slug);
261
297
262
298
const tierPools = parseSheetCsv(text);
263
299
const tierCounts: Record<string, number> = {};
···
267
303
chartCount += charts.length;
268
304
}
269
305
270
-
const sets = await fetchOpenSets(slug);
271
-
306
+
emitTaskProgress({ taskId, stage: 'pools', message: 'Generating brackets', current: 2, total: 4, done: false });
272
307
const indexedTierPools: Record<number, TierChart[]> = {};
273
308
for (const tier of Object.keys(tierPools)) {
274
309
indexedTierPools[Number(tier)] = tierPools[tier] as unknown as TierChart[];
···
276
311
277
312
const brackets = generatePools(indexedTierPools, bracketsConfig, sets as unknown as import('../util/poolGenerator.js').StartggSet[]);
278
313
314
+
emitTaskProgress({ taskId, stage: 'saving', message: 'Saving pools', current: 3, total: 4, done: false });
279
315
await setGeneratedPools(brackets, tierPools, String(event._id));
280
316
281
-
backgroundFetchCharts(tierPools);
317
+
// response returns now; background chart fetch streams the 4th stage.
318
+
backgroundFetchCharts(tierPools, taskId);
282
319
283
-
return { bracketCount: brackets.length, tierCounts, chartCount };
320
+
return { taskId, bracketCount: brackets.length, tierCounts, chartCount };
284
321
}
+52
-1
src/engine/wsBroadcast.test.ts
+52
-1
src/engine/wsBroadcast.test.ts
···
4
4
import { createServer, type Server } from 'node:http';
5
5
import mongoose from 'mongoose';
6
6
import { MongoMemoryServer } from 'mongodb-memory-server';
7
-
import { createWsServer, broadcast } from '../ws.js';
7
+
import { createWsServer, broadcast, setWsServer, emitTaskProgress } from '../ws.js';
8
8
import {
9
9
initMatchState,
10
10
banChart,
···
94
94
httpServer.close();
95
95
setBroadcast(() => {});
96
96
});
97
+
98
+
test('emitTaskProgress delivers to task:<id> subscribers only', async () => {
99
+
const httpServer: Server = createServer();
100
+
const wss = createWsServer(httpServer);
101
+
await new Promise<void>((resolve) => httpServer.listen(0, resolve));
102
+
setWsServer(wss);
103
+
104
+
const port = (httpServer.address() as { port: number }).port;
105
+
const client = new WebSocket(`ws://localhost:${port}`);
106
+
const received: any[] = [];
107
+
await new Promise<void>((resolve) => client.on('open', resolve));
108
+
client.on('message', (raw) => received.push(JSON.parse(raw.toString())));
109
+
110
+
// Wait for the server to confirm the subscription is live before emitting,
111
+
// otherwise the synchronous emits race ahead of the queued subscribe frame.
112
+
const subscribed = new Promise<void>((resolve, reject) => {
113
+
const timer = setTimeout(() => reject(new Error('timeout: no subscribe.ack')), 3000);
114
+
client.on('message', (raw) => {
115
+
const m = JSON.parse(raw.toString());
116
+
if (m.event === 'subscribe.ack' && m.data?.taskId === 'task-abc') {
117
+
clearTimeout(timer);
118
+
resolve();
119
+
}
120
+
});
121
+
});
122
+
123
+
// subscribe to task-abc, not task-xyz
124
+
client.send(JSON.stringify({ type: 'subscribe', taskId: 'task-abc' }));
125
+
126
+
await subscribed;
127
+
128
+
emitTaskProgress({ taskId: 'task-xyz', stage: 'charts', current: 1, total: 3, done: false });
129
+
emitTaskProgress({ taskId: 'task-abc', stage: 'charts', current: 1, total: 3, done: false });
130
+
131
+
const got1 = new Promise<void>((resolve) => client.on('message', (raw) => {
132
+
const m = JSON.parse(raw.toString());
133
+
if (m.event === 'task.progress' && m.data.taskId === 'task-abc') resolve();
134
+
}));
135
+
await got1;
136
+
137
+
const forAbc = received.filter((m) => m.event === 'task.progress' && m.data.taskId === 'task-abc');
138
+
const forXyz = received.filter((m) => m.event === 'task.progress' && m.data.taskId === 'task-xyz');
139
+
assert.equal(forAbc.length, 1);
140
+
assert.equal(forXyz.length, 0);
141
+
142
+
client.close();
143
+
await new Promise<void>((resolve) => wss.close(() => resolve()));
144
+
httpServer.close();
145
+
setWsServer(null as unknown as typeof wss);
146
+
setBroadcast(() => {});
147
+
});
+2
src/index.ts
+2
src/index.ts
···
7
7
import { setBroadcast } from './engine/match.js';
8
8
import { loadActiveEvent } from './engine/event.js';
9
9
import { loadGeneratedPoolsFromDB } from './engine/pools.js';
10
+
import { setWsServer } from './ws.js';
10
11
import { logger } from './logger.js';
11
12
12
13
async function main(): Promise<void> {
···
32
33
// Bridge the match engine's in-memory broadcast callback to the WebSocket
33
34
// server so connected overlay clients receive real-time match events.
34
35
setBroadcast((event, data) => broadcast(wsServer, event, data));
36
+
setWsServer(wsServer);
35
37
36
38
httpServer.listen(config.port, () => {
37
39
logger.info({ port: config.port }, 'http server listening');
+58
-2
src/ws.ts
+58
-2
src/ws.ts
···
4
4
5
5
interface WsClient {
6
6
ws: WebSocket;
7
-
subscriptions: Set<string>; // matchIds or eventIds
7
+
subscriptions: Set<string>; // matchIds, eventIds, or taskIds
8
8
}
9
9
10
10
const clients = new Set<WsClient>();
···
28
28
}
29
29
30
30
ws.on('message', (raw) => {
31
-
let msg: { type?: string; matchId?: string; eventId?: string };
31
+
let msg: { type?: string; matchId?: string; eventId?: string; taskId?: string };
32
32
try {
33
33
msg = JSON.parse(raw.toString());
34
34
}
···
60
60
logger.debug({ eventId: msg.eventId }, 'ws client subscribed to event');
61
61
}
62
62
63
+
if (msg.type === 'subscribe' && msg.taskId) {
64
+
client.subscriptions.add(`task:${msg.taskId}`);
65
+
logger.debug({ taskId: msg.taskId }, 'ws client subscribed to task');
66
+
ack({ type: 'subscribe', taskId: msg.taskId });
67
+
}
68
+
63
69
if (msg.type === 'unsubscribe' && msg.matchId) {
64
70
client.subscriptions.delete(`match:${msg.matchId}`);
65
71
logger.debug({ matchId: msg.matchId }, 'ws client unsubscribed from match');
···
68
74
if (msg.type === 'unsubscribe' && msg.eventId) {
69
75
client.subscriptions.delete(`event:${msg.eventId}`);
70
76
logger.debug({ eventId: msg.eventId }, 'ws client unsubscribed from event');
77
+
}
78
+
79
+
if (msg.type === 'unsubscribe' && msg.taskId) {
80
+
client.subscriptions.delete(`task:${msg.taskId}`);
81
+
logger.debug({ taskId: msg.taskId }, 'ws client unsubscribed from task');
71
82
}
72
83
});
73
84
85
+
const ack = (data: Record<string, unknown>) => {
86
+
try {
87
+
ws.send(JSON.stringify({
88
+
event: 'subscribe.ack',
89
+
data,
90
+
timestamp: new Date().toISOString(),
91
+
}));
92
+
}
93
+
catch (err) {
94
+
logger.warn({ err }, 'ws send failed');
95
+
}
96
+
};
97
+
74
98
ws.on('close', () => {
75
99
clients.delete(client);
76
100
});
···
88
112
return wss;
89
113
}
90
114
115
+
export interface TaskProgress {
116
+
taskId: string;
117
+
stage: string;
118
+
message?: string;
119
+
current?: number;
120
+
total?: number;
121
+
done?: boolean;
122
+
error?: string;
123
+
}
124
+
125
+
// Emits a task-progress event to any WS client subscribed to `task:<id>`.
126
+
// Reusable for any long-running server-side job (mappool import, bulk ops, etc).
127
+
export function emitTask(wss: WebSocketServer, progress: TaskProgress): void {
128
+
broadcast(wss, 'task.progress', progress);
129
+
}
130
+
131
+
// Module-level WS server reference so non-WS modules (engine, routes) can emit
132
+
// task progress without threading the server instance everywhere. Mirrors the
133
+
// setBroadcast pattern used by the match engine.
134
+
let wsServerRef: WebSocketServer | null = null;
135
+
136
+
export function setWsServer(wss: WebSocketServer): void {
137
+
wsServerRef = wss;
138
+
}
139
+
140
+
export function emitTaskProgress(progress: TaskProgress): void {
141
+
if (!wsServerRef) return;
142
+
emitTask(wsServerRef, progress);
143
+
}
144
+
91
145
// Broadcast to all connected clients (or filtered by subscriptions)
92
146
// The event string can be "match.snapshot" with data containing matchId,
93
147
// or "match.ban", "match.pick", etc.
···
113
167
const matchId = typeof dataObj?.matchId === 'string' ? dataObj.matchId : undefined;
114
168
const meta = typeof dataObj?.meta === 'object' && dataObj.meta !== null ? dataObj.meta as Record<string, unknown> : undefined;
115
169
const eventId = typeof meta?.eventId === 'string' ? meta.eventId : undefined;
170
+
const taskId = typeof dataObj?.taskId === 'string' ? dataObj.taskId : undefined;
116
171
117
172
const shouldSend =
118
173
(matchId && client.subscriptions.has(`match:${matchId}`))
119
174
|| (eventId && client.subscriptions.has(`event:${eventId}`))
175
+
|| (taskId && client.subscriptions.has(`task:${taskId}`))
120
176
|| client.subscriptions.has('event:*');
121
177
122
178
if (shouldSend) {