This commit is contained in:
@@ -0,0 +1,415 @@
|
||||
import { createHash } from 'node:crypto';
|
||||
import type {
|
||||
DiscordGame,
|
||||
DiscordMessage,
|
||||
DiscordPublicServer,
|
||||
DiscordSettings
|
||||
} from '../../discord.js';
|
||||
import { requireDiscordCredentials } from './config.js';
|
||||
import { 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 ?? '?'}).\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();
|
||||
}
|
||||
Reference in New Issue
Block a user