416 lines
16 KiB
TypeScript
416 lines
16 KiB
TypeScript
import { createHash } from 'node:crypto';
|
|
import type {
|
|
DiscordGame,
|
|
DiscordMessage,
|
|
DiscordPublicServer,
|
|
DiscordSettings
|
|
} from '../../discord.js';
|
|
import { requireDiscordCredentials } from './config.js';
|
|
import { DEFAULT_DISCORD_LOBBY_CAPACITY, discordText, publicFleet } from './projection.js';
|
|
import { renderBoard } from './render.js';
|
|
import { DiscordStore, discordStore, type DiscordJob, type DiscordSubscription } from './store.js';
|
|
import { DiscordDeliveryError, DiscordRest, type DiscordTransport } from './transport.js';
|
|
|
|
const TICK_MS = 30_000;
|
|
const VERIFY_MS = 15 * 60_000;
|
|
const FRESH_MS = 90_000;
|
|
let client: { token: string; rest: DiscordRest } | undefined;
|
|
|
|
export function discordRest(): DiscordRest {
|
|
const { botToken } = requireDiscordCredentials();
|
|
if (!client || client.token !== botToken)
|
|
client = { token: botToken, rest: new DiscordRest(botToken) };
|
|
return client.rest;
|
|
}
|
|
|
|
interface WorkerDependencies {
|
|
store: DiscordStore;
|
|
transport: () => DiscordTransport;
|
|
fleet: (settings: DiscordSettings, now: number) => DiscordPublicServer[];
|
|
render: (game: DiscordGame, servers: DiscordPublicServer[]) => DiscordMessage[];
|
|
now?: () => number;
|
|
}
|
|
|
|
/** A single independent reconciliation loop; it never issues game-server commands. */
|
|
export class DiscordWorker {
|
|
private running: Promise<void> | null = null;
|
|
private readonly now: () => number;
|
|
constructor(private readonly dependencies: WorkerDependencies) {
|
|
this.now = dependencies.now ?? Date.now;
|
|
}
|
|
|
|
tick(): Promise<void> {
|
|
if (this.running) return this.running;
|
|
this.running = this.run()
|
|
.catch((error: unknown) => {
|
|
// No raw fetch/config/SQL exceptions are persisted in operator-facing status.
|
|
try {
|
|
this.dependencies.store.error(
|
|
error instanceof DiscordDeliveryError
|
|
? error.message
|
|
: 'Discord delivery could not complete. Check the integration configuration.'
|
|
);
|
|
} catch {
|
|
console.error('[discord] Could not record delivery health.');
|
|
}
|
|
})
|
|
.finally(() => {
|
|
this.running = null;
|
|
});
|
|
return this.running;
|
|
}
|
|
|
|
private async run(): Promise<void> {
|
|
const { store, fleet, render } = this.dependencies;
|
|
store.expireSubscriptions();
|
|
const settings = store.settings();
|
|
const servers = settings.enabled ? fleet(settings, this.now()) : [];
|
|
store.reconcileBoards(
|
|
settings.enabled
|
|
? (['t5', 't6'] as const)
|
|
.filter((game) => settings.channels[game])
|
|
.map((game) => ({
|
|
game,
|
|
guildId: settings.guildId,
|
|
channelId: settings.channels[game],
|
|
messages: render(game, servers)
|
|
}))
|
|
: []
|
|
);
|
|
if (settings.enabled) {
|
|
for (const watch of store.subscriptions()) {
|
|
if (watch.state !== 'active') continue;
|
|
if (watch.guildId !== settings.guildId) {
|
|
store.cancel(watch.id, watch.guildId, watch.userId);
|
|
continue;
|
|
}
|
|
const match = matchingServer(watch, servers, this.now());
|
|
if (!match) {
|
|
store.candidate(watch.id, null);
|
|
continue;
|
|
}
|
|
// Persist the server identity, not each telemetry timestamp: two fresh
|
|
// observations spanning a tick stabilize brief capacity fluctuations.
|
|
if (watch.candidate?.split('|')[0] !== match.name) {
|
|
store.candidate(watch.id, `${match.name}|${match.observedAt ?? ''}`);
|
|
continue;
|
|
}
|
|
if (
|
|
watch.candidateAt !== null &&
|
|
this.now() - watch.candidateAt >= 15_000 &&
|
|
watch.candidate !== `${match.name}|${match.observedAt ?? ''}`
|
|
)
|
|
store.queueAlert(watch, alertMessage(watch, match));
|
|
}
|
|
}
|
|
const jobs = store.jobs(10);
|
|
const toVerify = store
|
|
.boards()
|
|
.find(
|
|
(board) =>
|
|
board.messageId &&
|
|
board.verifiedAt < this.now() - VERIFY_MS &&
|
|
!jobs.some((job) => job.targetId === board.id)
|
|
);
|
|
if (!jobs.length && (!settings.enabled || !toVerify)) return;
|
|
const rest = this.dependencies.transport();
|
|
const channelGuilds = new Map<string, string>();
|
|
const ensureChannel = async (channelId: string, guildId: string) => {
|
|
if (!channelGuilds.has(channelId)) {
|
|
const channel = await rest.request<{ guild_id?: string; type?: number }>(
|
|
'GET',
|
|
`/channels/${channelId}`
|
|
);
|
|
if (![0, 5].includes(channel.type ?? -1))
|
|
throw new DiscordDeliveryError(
|
|
'Choose a Discord text or announcement channel for the board.',
|
|
403
|
|
);
|
|
channelGuilds.set(channelId, channel.guild_id ?? '');
|
|
}
|
|
if (channelGuilds.get(channelId) !== guildId)
|
|
throw new DiscordDeliveryError(
|
|
'The board channel does not belong to the configured Discord server.',
|
|
403
|
|
);
|
|
};
|
|
for (const job of jobs) {
|
|
store.attempt();
|
|
try {
|
|
await this.deliver(job, rest, ensureChannel);
|
|
} catch (error) {
|
|
const failure =
|
|
error instanceof DiscordDeliveryError
|
|
? error
|
|
: new DiscordDeliveryError(
|
|
'Discord delivery could not complete.',
|
|
0,
|
|
this.now() + 5_000
|
|
);
|
|
const canRetry =
|
|
failure.retryAt !== null && (failure.status === 429 || job.attempts < 7);
|
|
const delay = Math.min(15 * 60_000, 5_000 * 2 ** Math.min(job.attempts, 8));
|
|
store.failed(
|
|
job,
|
|
failure.message,
|
|
canRetry
|
|
? Math.max(
|
|
failure.retryAt!,
|
|
this.now() + (failure.status === 429 ? 0 : delay)
|
|
)
|
|
: null
|
|
);
|
|
if (failure.status === 401 || failure.status === 429) break;
|
|
}
|
|
}
|
|
if (settings.enabled && toVerify) {
|
|
const current = store.settings();
|
|
if (
|
|
!current.enabled ||
|
|
current.guildId !== toVerify.guildId ||
|
|
current.channels[toVerify.game] !== toVerify.channelId
|
|
)
|
|
return;
|
|
store.attempt();
|
|
try {
|
|
await ensureChannel(toVerify.channelId, current.guildId);
|
|
await rest.request(
|
|
'GET',
|
|
`/channels/${toVerify.channelId}/messages/${toVerify.messageId}`
|
|
);
|
|
store.verified(toVerify.id);
|
|
} catch (error) {
|
|
if (
|
|
error instanceof DiscordDeliveryError &&
|
|
error.status === 404 &&
|
|
error.code === 10008
|
|
)
|
|
store.missingMessage(toVerify.id);
|
|
else {
|
|
store.verified(toVerify.id); // Check permissions again later, not every tick.
|
|
store.error(
|
|
error instanceof DiscordDeliveryError
|
|
? error.message
|
|
: 'Discord board verification failed.'
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private async deliver(
|
|
job: DiscordJob,
|
|
rest: DiscordTransport,
|
|
ensureChannel: (channelId: string, guildId: string) => Promise<void>
|
|
): Promise<void> {
|
|
const { store, fleet, render } = this.dependencies;
|
|
const settings = store.settings();
|
|
if (job.kind === 'alert') {
|
|
const watch = store.subscription(job.targetId);
|
|
if (
|
|
!watch ||
|
|
watch.state !== 'queued' ||
|
|
watch.expiresAt <= this.now() ||
|
|
!settings.enabled ||
|
|
settings.guildId !== watch.guildId
|
|
) {
|
|
store.dropJob(job.id);
|
|
return;
|
|
}
|
|
const match = matchingServer(watch, fleet(settings, this.now()), this.now());
|
|
if (!match) {
|
|
store.resumeAlert(watch.id);
|
|
return;
|
|
}
|
|
let channelId = job.channelId;
|
|
if (!channelId) {
|
|
const channel = await rest.request<{ id: string }>('POST', '/users/@me/channels', {
|
|
recipient_id: watch.userId
|
|
});
|
|
channelId = channel.id;
|
|
if (!/^\d+$/.test(channelId))
|
|
throw new DiscordDeliveryError(
|
|
'Discord returned an invalid channel.',
|
|
502,
|
|
this.now() + 5_000
|
|
);
|
|
store.cacheDm(job.id, channelId);
|
|
}
|
|
// Recheck opt-in/config after the asynchronous channel lookup.
|
|
const current = store.settings();
|
|
const stillWatching = store.subscription(watch.id);
|
|
if (
|
|
!current.enabled ||
|
|
current.guildId !== watch.guildId ||
|
|
stillWatching?.state !== 'queued' ||
|
|
stillWatching.expiresAt <= this.now()
|
|
) {
|
|
store.dropJob(job.id);
|
|
return;
|
|
}
|
|
const fresh = matchingServer(watch, fleet(current, this.now()), this.now());
|
|
if (!fresh) {
|
|
store.resumeAlert(watch.id);
|
|
return;
|
|
}
|
|
await rest.request('POST', `/channels/${channelId}/messages`, {
|
|
...alertMessage(watch, fresh),
|
|
nonce: nonce(job.id),
|
|
enforce_nonce: true
|
|
});
|
|
store.delivered(job);
|
|
return;
|
|
}
|
|
const board = store.board(job.targetId);
|
|
if (!board) {
|
|
store.dropJob(job.id);
|
|
return;
|
|
}
|
|
if (job.kind === 'delete') {
|
|
if (board.messageId) {
|
|
try {
|
|
await rest.request(
|
|
'DELETE',
|
|
`/channels/${board.channelId}/messages/${board.messageId}`
|
|
);
|
|
} catch (error) {
|
|
if (!(error instanceof DiscordDeliveryError && error.status === 404))
|
|
throw error;
|
|
}
|
|
}
|
|
store.delivered(job);
|
|
return;
|
|
}
|
|
if (
|
|
!settings.enabled ||
|
|
settings.guildId !== board.guildId ||
|
|
settings.channels[board.game] !== board.channelId
|
|
) {
|
|
store.dropJob(job.id);
|
|
return;
|
|
}
|
|
await ensureChannel(board.channelId, settings.guildId);
|
|
// Rebuild after awaits so pending jobs cannot publish withdrawn fields or servers.
|
|
const current = store.settings();
|
|
if (
|
|
!current.enabled ||
|
|
current.guildId !== settings.guildId ||
|
|
current.channels[board.game] !== board.channelId
|
|
) {
|
|
store.dropJob(job.id);
|
|
return;
|
|
}
|
|
const message = render(board.game, fleet(current, this.now()))[board.page];
|
|
if (!message) {
|
|
store.dropJob(job.id);
|
|
return;
|
|
}
|
|
const updated = {
|
|
...job,
|
|
payload: message,
|
|
contentHash: createHash('sha256').update(JSON.stringify(message)).digest('hex')
|
|
};
|
|
if (board.messageId) {
|
|
try {
|
|
await rest.request(
|
|
'PATCH',
|
|
`/channels/${board.channelId}/messages/${board.messageId}`,
|
|
message
|
|
);
|
|
store.delivered(updated, board.messageId);
|
|
return;
|
|
} catch (error) {
|
|
if (!(
|
|
error instanceof DiscordDeliveryError &&
|
|
error.status === 404 &&
|
|
error.code === 10008
|
|
))
|
|
throw error;
|
|
store.missingMessage(board.id);
|
|
}
|
|
}
|
|
const result = await rest.request<{ id: string }>(
|
|
'POST',
|
|
`/channels/${board.channelId}/messages`,
|
|
{
|
|
...message,
|
|
nonce: nonce(`${board.id}:${store.board(board.id)?.generation ?? 0}`),
|
|
enforce_nonce: true
|
|
}
|
|
);
|
|
if (!/^\d+$/.test(result.id))
|
|
throw new DiscordDeliveryError(
|
|
'Discord returned an invalid message.',
|
|
502,
|
|
this.now() + 5_000
|
|
);
|
|
store.delivered(updated, result.id);
|
|
}
|
|
}
|
|
|
|
function matchingServer(
|
|
watch: DiscordSubscription,
|
|
servers: DiscordPublicServer[],
|
|
now: number
|
|
): DiscordPublicServer | undefined {
|
|
return servers.find(
|
|
(server) =>
|
|
server.game === watch.game &&
|
|
server.mapKey === watch.mapKey &&
|
|
server.observedAt !== null &&
|
|
now - Date.parse(server.observedAt) >= 0 &&
|
|
now - Date.parse(server.observedAt) <= FRESH_MS &&
|
|
['empty', 'in-progress'].includes(server.state) &&
|
|
server.players !== null &&
|
|
(watch.mode === 'empty'
|
|
? server.players === 0
|
|
: server.capacity !== null &&
|
|
server.capacity > 0 &&
|
|
server.players < server.capacity)
|
|
);
|
|
}
|
|
|
|
function alertMessage(watch: DiscordSubscription, server: DiscordPublicServer): DiscordMessage {
|
|
const safe = (value: string) => discordText(value, 100);
|
|
return {
|
|
content: `🍒 **${safe(server.map)}** ${watch.mode === 'empty' ? 'has an empty lobby' : 'has an open player slot'}: **${safe(server.label)}** (${server.players}/${server.capacity ?? DEFAULT_DISCORD_LOBBY_CAPACITY}).\nThis describes occupancy; joining an in-progress game may still be restricted.\nYour one-time map watch is complete. Use /servers to browse or /watch to watch again.`,
|
|
allowed_mentions: { parse: [] }
|
|
};
|
|
}
|
|
|
|
function nonce(id: string): string {
|
|
return createHash('sha256').update(id).digest('hex').slice(0, 24);
|
|
}
|
|
|
|
let worker: DiscordWorker | undefined;
|
|
let timer: ReturnType<typeof setInterval> | undefined;
|
|
export function tickDiscordWorker(): Promise<void> {
|
|
worker ??= new DiscordWorker({
|
|
store: discordStore(),
|
|
transport: discordRest,
|
|
fleet: publicFleet,
|
|
render: renderBoard
|
|
});
|
|
return worker.tick();
|
|
}
|
|
export function startDiscordWorker(): void {
|
|
if (timer) return;
|
|
void tickDiscordWorker();
|
|
timer = setInterval(() => {
|
|
void tickDiscordWorker();
|
|
}, TICK_MS);
|
|
timer.unref();
|
|
}
|
|
export function stopDiscordWorker(): void {
|
|
clearInterval(timer);
|
|
timer = undefined;
|
|
}
|
|
export function resetDiscordClient(): void {
|
|
client = undefined;
|
|
}
|
|
export function retryDiscordDelivery(): Promise<void> {
|
|
resetDiscordClient();
|
|
discordStore().retryFailed();
|
|
return tickDiscordWorker();
|
|
}
|