From fe2c29371f1fa2258b84a3e8901c44f7cb56a0ca Mon Sep 17 00:00:00 2001 From: legop3 Date: Fri, 18 Sep 2026 14:41:11 -0400 Subject: [PATCH] fix discord bot config restart and startup connection retry --- .../services/discordBotService/channelIO.js | 11 +- .../services/discordBotService/connection.js | 94 +++ .../discordBotService/connection.test.js | 146 +++++ .../src/services/discordBotService/index.js | 554 +++++++++--------- .../discordBotService/integrations/index.js | 30 +- .../services/discordBotService/presence.js | 7 + 6 files changed, 559 insertions(+), 283 deletions(-) create mode 100644 server/src/services/discordBotService/connection.js create mode 100644 server/src/services/discordBotService/connection.test.js diff --git a/server/src/services/discordBotService/channelIO.js b/server/src/services/discordBotService/channelIO.js index 0882b32a..8b2956ca 100644 --- a/server/src/services/discordBotService/channelIO.js +++ b/server/src/services/discordBotService/channelIO.js @@ -6,12 +6,14 @@ const { MessageFlags } = require('discord.js'); function createChannelIO({ client, logger, sanitizeMentions }) { const channelCache = new Map(); const typingMessageCache = new Map(); + let stopped = false; async function fetchChannel(id) { - if (!id) return null; + if (stopped || !client.isReady() || !id) return null; if (channelCache.has(id)) return channelCache.get(id); try { const channel = await client.channels.fetch(id); + if (stopped) return null; if (channel) { channelCache.set(id, channel); return channel; @@ -68,6 +70,7 @@ function createChannelIO({ client, logger, sanitizeMentions }) { const content = `-# *${username} is typing...*`; try { const message = await channel.send({ content, allowedMentions: { parse: [] }, flags: [MessageFlags.SuppressNotifications]}); + if (stopped) return; const timeoutId = setTimeout(() => { clearTypingMessage(entry.guildId, typingId); }, 20000); @@ -78,6 +81,12 @@ function createChannelIO({ client, logger, sanitizeMentions }) { } return { + stop() { + stopped = true; + for (const record of typingMessageCache.values()) clearTimeout(record.timeoutId); + typingMessageCache.clear(); + channelCache.clear(); + }, fetchChannel, sendToChannel, clearTypingMessage, diff --git a/server/src/services/discordBotService/connection.js b/server/src/services/discordBotService/connection.js new file mode 100644 index 00000000..d6f1f460 --- /dev/null +++ b/server/src/services/discordBotService/connection.js @@ -0,0 +1,94 @@ +// Owns connection replacement and recovery from failed initial logins. +function createConnectionManager({ createRuntime, logger }) { + let desired = { enabled: false, token: '' }; + let revision = 0; + let runtime = null; + let queue = Promise.resolve(); + let retryTimer = null; + let retryDelay = 5000; + + function enqueue(operation) { + const pending = queue.then(operation); + // Report failures without poisoning subsequent configuration changes. + queue = pending.catch((err) => logger.error('Discord connection update failed', err.message)); + return pending; + } + + async function stop() { + const previous = runtime; + runtime = null; + if (!previous) return; + previous.stop(); + // discord.js cannot cancel login while gateway discovery is pending. + // Let that attempt settle before destroying it, otherwise it can open a + // socket after destroy() has already run and leave an orphan connection. + await previous.login; + await previous.client.destroy(); + } + + function scheduleRetry(version) { + if (version !== revision || !desired.enabled) return; + const delay = retryDelay; + retryDelay = Math.min(retryDelay * 2, 60000); + logger.warn('Retrying Discord login', { delayMs: delay }); + retryTimer = setTimeout(() => { + retryTimer = null; + enqueue(() => start(version)); + }, delay); + retryTimer.unref?.(); + } + + async function start(version) { + if (version !== revision || !desired.enabled) return; + await stop(); + if (version !== revision || !desired.enabled) return; + const current = createRuntime(); + runtime = current; + const { client } = current; + client.on('error', (err) => logger.error('Discord client error', err.message)); + client.on('shardError', (err, shardId) => { + logger.warn('Discord connection error', { shardId, error: err.message }); + }); + client.on('shardDisconnect', (event, shardId) => { + logger.warn('Discord disconnected', { shardId, code: event.code }); + }); + client.on('clientReady', () => { retryDelay = 5000; }); + + // Login must not block configuration saves while the network is down. + // Each attempt has its own client, so stale login completion/cleanup cannot + // clear the credentials or stop the connection of its replacement. + current.login = client.login(desired.token).catch((err) => { + if (version !== revision || runtime !== current) return; + logger.error('Discord login failed', err.message); + enqueue(async () => { + if (version !== revision || runtime !== current) return; + await stop(); + // Credentials require an operator edit; retry network/service failures. + if (err.code === 'TokenInvalid' || err.status === 401) return; + scheduleRetry(version); + }); + }); + } + + function configure({ enabled, token }) { + const next = { enabled: Boolean(enabled), token }; + if (next.enabled === desired.enabled && next.token === desired.token) return queue; + desired = next; + const version = ++revision; + clearTimeout(retryTimer); + retryTimer = null; + retryDelay = 5000; + return enqueue(async () => { + if (version !== revision) return; + if (desired.enabled) await start(version); + else await stop(); + }); + } + + return { + configure, + refresh() { runtime?.refresh(); }, + }; +} + +module.exports = { createConnectionManager }; diff --git a/server/src/services/discordBotService/connection.test.js b/server/src/services/discordBotService/connection.test.js new file mode 100644 index 00000000..65caec23 --- /dev/null +++ b/server/src/services/discordBotService/connection.test.js @@ -0,0 +1,146 @@ +const test = require('node:test'); +const assert = require('node:assert/strict'); +const { EventEmitter } = require('node:events'); +const { createConnectionManager } = require('./connection'); + +const flush = () => new Promise((resolve) => setImmediate(resolve)); + +function deferred() { + let resolve; + let reject; + const promise = new Promise((yes, no) => { resolve = yes; reject = no; }); + return { promise, resolve, reject }; +} + +function setup(t, attempts = []) { + t.mock.timers.enable({ apis: ['setTimeout'] }); + const runtimes = []; + const logs = []; + const manager = createConnectionManager({ + logger: { + warn: (...args) => logs.push(args), + error: (...args) => logs.push(args), + }, + createRuntime() { + const attempt = attempts[runtimes.length] || {}; + const client = new EventEmitter(); + const runtime = { + client, + stopped: false, + destroyed: false, + refresh() {}, + stop() { runtime.stopped = true; }, + }; + client.login = async (token) => { + runtime.token = token; + await attempt.login?.(); + }; + client.destroy = async () => { + await attempt.destroy?.(); + runtime.destroyed = true; + }; + runtimes.push(runtime); + return runtime; + }, + }); + return { manager, runtimes, logs }; +} + +test('retries a network failure at startup with a fresh client', async (t) => { + const { manager, runtimes } = setup(t, [{ login: () => { throw new Error('getaddrinfo EAI_AGAIN'); } }]); + await manager.configure({ enabled: true, token: 'token' }); + await flush(); + assert.equal(runtimes[0].stopped, true); + assert.equal(runtimes[0].destroyed, true); + t.mock.timers.tick(4999); + await flush(); + assert.equal(runtimes.length, 1); + t.mock.timers.tick(1); + await flush(); + assert.equal(runtimes.length, 2); + assert.equal(runtimes[1].token, 'token'); + assert.notEqual(runtimes[0].client, runtimes[1].client); +}); + +test('backs off repeated failures and cancels retries when disabled', async (t) => { + const failed = { login: () => { throw new Error('Network unavailable'); } }; + const { manager, runtimes, logs } = setup(t, Array(8).fill(failed)); + await manager.configure({ enabled: true, token: 'token' }); + await flush(); + for (const delay of [5000, 10000, 20000, 40000, 60000]) { + t.mock.timers.tick(delay); + await flush(); + } + assert.deepEqual(logs.filter(([message]) => message === 'Retrying Discord login').map(([, data]) => data.delayMs), + [5000, 10000, 20000, 40000, 60000, 60000]); + await manager.configure({ enabled: false, token: 'token' }); + t.mock.timers.tick(120000); + await flush(); + assert.equal(runtimes.length, 6); +}); + +test('waits for logout before logging in with a changed token', async (t) => { + const logout = deferred(); + const { manager, runtimes } = setup(t, [{ destroy: () => logout.promise }]); + await manager.configure({ enabled: true, token: 'old' }); + const change = manager.configure({ enabled: true, token: 'new' }); + await flush(); + assert.equal(runtimes[0].stopped, true); + assert.equal(runtimes.length, 1); + logout.resolve(); + await change; + assert.equal(runtimes[0].destroyed, true); + assert.equal(runtimes[1].token, 'new'); +}); + +test('retires a pending login before replacement and ignores its stale failure', async (t) => { + const login = deferred(); + const { manager, runtimes } = setup(t, [{ login: () => login.promise }]); + await manager.configure({ enabled: true, token: 'old' }); + const change = manager.configure({ enabled: true, token: 'new' }); + await flush(); + assert.equal(runtimes[0].stopped, true); + assert.equal(runtimes[0].destroyed, false); + login.reject(new Error('Old connection failed')); + await change; + assert.equal(runtimes[0].destroyed, true); + assert.equal(runtimes[1].token, 'new'); + t.mock.timers.tick(120000); + await flush(); + assert.equal(runtimes.length, 2); +}); + +test('re-enabling creates a fresh client; unrelated config edits do not reconnect', async (t) => { + const { manager, runtimes } = setup(t); + await manager.configure({ enabled: true, token: 'token' }); + await manager.configure({ enabled: true, token: 'token', channels: { general: 'new-channel' } }); + assert.equal(runtimes.length, 1); + await manager.configure({ enabled: false, token: 'token' }); + assert.equal(runtimes[0].destroyed, true); + await manager.configure({ enabled: true, token: 'token' }); + assert.equal(runtimes.length, 2); +}); + +test('invalid credentials wait for a token edit instead of retrying', async (t) => { + const { manager, runtimes } = setup(t, [{ login: () => { + throw Object.assign(new Error('Invalid token'), { code: 'TokenInvalid' }); + } }]); + await manager.configure({ enabled: true, token: 'bad' }); + await flush(); + t.mock.timers.tick(120000); + await flush(); + assert.equal(runtimes.length, 1); + await manager.configure({ enabled: true, token: 'fixed' }); + assert.equal(runtimes[1].token, 'fixed'); +}); + +test('client and gateway errors are logged without throwing', async (t) => { + const { manager, runtimes, logs } = setup(t); + await manager.configure({ enabled: true, token: 'token' }); + runtimes[0].client.emit('error', new Error('Client failure')); + runtimes[0].client.emit('shardError', new Error('Gateway failure'), 0); + runtimes[0].client.emit('shardDisconnect', { code: 1006 }, 0); + assert.deepEqual(logs.map(([message]) => message), [ + 'Discord client error', 'Discord connection error', 'Discord disconnected', + ]); +}); diff --git a/server/src/services/discordBotService/index.js b/server/src/services/discordBotService/index.js index 643bc5fc..4e7085a6 100644 --- a/server/src/services/discordBotService/index.js +++ b/server/src/services/discordBotService/index.js @@ -67,6 +67,7 @@ const { } = require('../privateRoverAccessRequestService'); const { subscribe } = require('../eventBus'); const { createPresenceManager } = require('./presence'); +const { createConnectionManager } = require('./connection'); const { createChannelIO } = require('./channelIO'); const { createCommandHandlers } = require('../operatorCommandService'); const greenModeService = require('../greenModeService'); @@ -86,9 +87,7 @@ const { buildStatusMessage, } = require('../replayDeliveryService/workflow'); -// Discord helper modules retain references to these objects. Mutating those -// references on configuration application updates command and integration -// behavior without registering a second tree of Discord/event listeners. +// Keep configuration object identities stable for the active runtime's helpers. const config = structuredClone(loadConfig()); const discordConfig = config.discord || {}; let enabled = Boolean(discordConfig.enabled); @@ -106,11 +105,6 @@ const intents = [ GatewayIntentBits.MessageContent, ]; -const client = new Client({ - intents, - partials: [Partials.Channel, Partials.Message, Partials.Reaction, Partials.User], -}); - function sanitizeMentions(text) { if (!text) return ''; return String(text) @@ -156,300 +150,310 @@ function countReady() { return { ready, total }; } -const channelIO = createChannelIO({ - client, - logger, - sanitizeMentions, -}); +// Each connection owns its listeners, timers, and cached Discord objects. +// Rebuilding them prevents a reconnect from retaining a destroyed client. +function createDiscordRuntime() { + const client = new Client({ + intents, + partials: [Partials.Channel, Partials.Message, Partials.Reaction, Partials.User], + }); + const channelIO = createChannelIO({ + client, + logger, + sanitizeMentions, + }); -const presence = createPresenceManager({ - client, - logger, - getMode, - getGlobalObjective, - countReady, -}); + const presence = createPresenceManager({ + client, + logger, + getMode, + getGlobalObjective, + countReady, + }); -const replayCaption = createReplayCaptionBuilder({ - io, - rovers, - getActiveDrivers, - getNickname, - sanitizeMentions, -}); + const replayCaption = createReplayCaptionBuilder({ + io, + rovers, + getActiveDrivers, + getNickname, + sanitizeMentions, + }); -// Discord is the preferred replay host only while this optional feature is -// active. The core replay delivery service owns generation and automatically -// falls back to its local media store when any operation below fails. -registerPreferredDeliveryProvider({ - async begin(job) { - const channelId = discordConfig.channels?.replay; - if (!enabled || !channelId) throw new Error('Discord replay delivery is disabled'); - const progressMessage = await channelIO.sendToChannel(channelId, buildAcceptedMessage(job), {}, DEFAULT_ALLOWED_MENTIONS); - if (!progressMessage) throw new Error('Discord replay progress message could not be sent'); - const channel = await channelIO.fetchChannel(channelId); - return { - channelId, - progressMessage, - stopTyping: startDiscordTypingLoop(channel, logger, 'web replay delivery'), - }; - }, - async deliver({ job, context, buffer, usedSources = job.sources, missingSources = [] }) { - const progressMessage = context?.progressMessage; - try { - if (progressMessage?.edit) { - await progressMessage.edit({ content: buildStatusMessage(job, 'uploading'), allowedMentions: DEFAULT_ALLOWED_MENTIONS }); + // Discord is the preferred replay host only while this optional feature is + // active. The core replay delivery service owns generation and automatically + // falls back to its local media store when any operation below fails. + const unregisterReplayProvider = registerPreferredDeliveryProvider({ + async begin(job) { + const channelId = discordConfig.channels?.replay; + if (!enabled || !channelId) throw new Error('Discord replay delivery is disabled'); + const progressMessage = await channelIO.sendToChannel(channelId, buildAcceptedMessage(job), {}, DEFAULT_ALLOWED_MENTIONS); + if (!progressMessage) throw new Error('Discord replay progress message could not be sent'); + const channel = await channelIO.fetchChannel(channelId); + return { + channelId, + progressMessage, + stopTyping: startDiscordTypingLoop(channel, logger, 'web replay delivery'), + }; + }, + async deliver({ job, context, buffer, usedSources = job.sources, missingSources = [] }) { + const progressMessage = context?.progressMessage; + try { + if (progressMessage?.edit) { + await progressMessage.edit({ content: buildStatusMessage(job, 'uploading'), allowedMentions: DEFAULT_ALLOWED_MENTIONS }); + } + // Every delivery path uses the job creation time, so a Discord upload + // and a server-hosted fallback always expose the same replay filename. + const attachment = new AttachmentBuilder(buffer, { name: buildReplayFilename(job) }); + const body = replayCaption.build({ job, usedSources, missingSources }); + const uploadMessage = await channelIO.sendToChannel(context.channelId, body, { files: [attachment] }, DEFAULT_ALLOWED_MENTIONS); + if (!uploadMessage) throw new Error('Discord upload did not return a message'); + const media = buildDiscordReplayMediaPayload({ message: uploadMessage, attachment: firstAttachmentFromMessage(uploadMessage), job }); + if (!media) throw new Error('Discord upload did not include a replay attachment URL'); + if (progressMessage?.edit) { + // The attachment URL is already durable once Discord returns it. A + // cosmetic progress-edit failure must not trigger a duplicate local + // replay or replace the successful media payload sent to clients. + await progressMessage.edit({ content: buildStatusMessage(job, 'ready'), allowedMentions: DEFAULT_ALLOWED_MENTIONS }).catch((err) => { + logger.warn('Discord replay uploaded but progress message update failed', { jobId: job.id, error: err.message }); + }); + } + return media; + } catch (err) { + err.progressMessage = progressMessage; + throw err; + } finally { + if (context?.stopTyping) context.stopTyping(); } - // Every delivery path uses the job creation time, so a Discord upload - // and a server-hosted fallback always expose the same replay filename. - const attachment = new AttachmentBuilder(buffer, { name: buildReplayFilename(job) }); - const body = replayCaption.build({ job, usedSources, missingSources }); - const uploadMessage = await channelIO.sendToChannel(context.channelId, body, { files: [attachment] }, DEFAULT_ALLOWED_MENTIONS); - if (!uploadMessage) throw new Error('Discord upload did not return a message'); - const media = buildDiscordReplayMediaPayload({ message: uploadMessage, attachment: firstAttachmentFromMessage(uploadMessage), job }); - if (!media) throw new Error('Discord upload did not include a replay attachment URL'); - if (progressMessage?.edit) { - // The attachment URL is already durable once Discord returns it. A - // cosmetic progress-edit failure must not trigger a duplicate local - // replay or replace the successful media payload sent to clients. - await progressMessage.edit({ content: buildStatusMessage(job, 'ready'), allowedMentions: DEFAULT_ALLOWED_MENTIONS }).catch((err) => { - logger.warn('Discord replay uploaded but progress message update failed', { jobId: job.id, error: err.message }); + }, + async completeFallback({ context, media }) { + const publicBaseUrl = String(config.publicUrl || '').replace(/\/$/, ''); + const publicUrl = publicBaseUrl ? `${publicBaseUrl}${media.url}` : media.url; + if (context?.progressMessage?.reply) { + await context.progressMessage.reply({ + content: `Replay hosted by the rover server: ${publicUrl}`, + allowedMentions: DEFAULT_ALLOWED_MENTIONS, }); } - return media; - } catch (err) { - err.progressMessage = progressMessage; - throw err; - } finally { - if (context?.stopTyping) context.stopTyping(); - } - }, - async completeFallback({ context, media }) { - const publicBaseUrl = String(config.publicUrl || '').replace(/\/$/, ''); - const publicUrl = publicBaseUrl ? `${publicBaseUrl}${media.url}` : media.url; - if (context?.progressMessage?.reply) { - await context.progressMessage.reply({ - content: `Replay hosted by the rover server: ${publicUrl}`, - allowedMentions: DEFAULT_ALLOWED_MENTIONS, - }); - } - }, -}); + }, + }); -const commandDependencies = { - logger, - client, - io, - rovers, - roverManager, - getMode, - MODES, - setMode, - lockRover, - getNickname, - getActiveDrivers, - buildReplayVideo, - getReplaySources, - getDefaultDiscordSources, - validateSources, - tryTriggerReplay, - getGlobalObjective, - setGlobalObjective, - clearGlobalObjective, - getAdminReason, - setAdminReason, - clearAdminReason, - // Room-light lock commands must use the same Home Assistant service instance - // as sockets, HA button triggers, and idle/darkness policies. Passing the - // service into the shared command router keeps Discord and mirrored web-chat - // command behavior aligned without duplicating Home Assistant calls here. - homeAssistantService, - homeAssistantActivitiesService, - greenModeService, - liftService, - neatoService, - isFeatureEnabled, - getGuildConfig, - setGuildConfig, - removeGuildConfig, - normalizeMode, - VALID_MODES, - listVerifiedUsers, - removeVerifiedUser, - listDeterredUsers, - listMutedUsers, - deterUser, - undeterUser, - muteUser, - unmuteUser, - listUsersForAdmin, - listUsersWithPermission, - listRegisteredPermissions, - setUserPermission, - sanitizeMentions, - sendToChannel: channelIO.sendToChannel, - isAdminUser, - isLockdownAdminUser, - discordConfig, - config, -}; -commandDependencies.transportHandlers = createDiscordTransportHandlers(commandDependencies); -const commands = createCommandHandlers(commandDependencies); + const commandDependencies = { + logger, + client, + io, + rovers, + roverManager, + getMode, + MODES, + setMode, + lockRover, + getNickname, + getActiveDrivers, + buildReplayVideo, + getReplaySources, + getDefaultDiscordSources, + validateSources, + tryTriggerReplay, + getGlobalObjective, + setGlobalObjective, + clearGlobalObjective, + getAdminReason, + setAdminReason, + clearAdminReason, + // Room-light lock commands must use the same Home Assistant service instance + // as sockets, HA button triggers, and idle/darkness policies. Passing the + // service into the shared command router keeps Discord and mirrored web-chat + // command behavior aligned without duplicating Home Assistant calls here. + homeAssistantService, + homeAssistantActivitiesService, + greenModeService, + liftService, + neatoService, + isFeatureEnabled, + getGuildConfig, + setGuildConfig, + removeGuildConfig, + normalizeMode, + VALID_MODES, + listVerifiedUsers, + removeVerifiedUser, + listDeterredUsers, + listMutedUsers, + deterUser, + undeterUser, + muteUser, + unmuteUser, + listUsersForAdmin, + listUsersWithPermission, + listRegisteredPermissions, + setUserPermission, + sanitizeMentions, + sendToChannel: channelIO.sendToChannel, + isAdminUser, + isLockdownAdminUser, + discordConfig, + config, + }; + commandDependencies.transportHandlers = createDiscordTransportHandlers(commandDependencies); + const commands = createCommandHandlers(commandDependencies); const integrations = createIntegrations({ - logger, - client, - config, - discordConfig, - rovers, - roverManager, - getMode, - MODES, - getGlobalObjective, - getActiveDrivers, - getNickname, - subscribe, - sendExternalMessage, - sendExternalTyping, - getGuildConfig, - listGuildConfigs, - attachDmMessage, - getRequestByMessageId, - approveRequest, - denyRequest, - attachPrivateAccessDmMessage, - getPrivateAccessRequestByMessageId, - approvePrivateAccessRequest, - denyPrivateAccessRequest, - getLockdownAdminIds, - isAdminUser, - isLockdownAdminUser, - sendToChannel: channelIO.sendToChannel, - fetchChannel: channelIO.fetchChannel, - clearTypingMessage: channelIO.clearTypingMessage, - sendTypingMessage: channelIO.sendTypingMessage, - schedulePresenceRotation: presence.schedulePresenceRotation, - buildReplayVideo, - sanitizeMentions, -}); - -const integrationHandlers = integrations.register(); - -function isTextCommand(content) { - // Both transports share this parser so command detection cannot drift from - // the dispatcher when an installation changes its prefix. - return parseCommandText(content, config).matched; -} - -function isBridgeChannelMessage(message) { - if (!message?.guild?.id || !message?.channelId) return false; - const guildConfig = getGuildConfig(message.guild.id); - return Boolean(guildConfig?.channelId && String(message.channelId) === String(guildConfig.channelId)); -} - -function createBridgeMirroredCommandMessage(message) { - if (!isBridgeChannelMessage(message) || !isTextCommand(message.content)) return message; - const originalReply = message.reply.bind(message); - return Object.assign(Object.create(message), { - reply: async (payload) => { - const sent = await originalReply(payload); - const text = sanitizeMentions(commandReplyToText(payload)); - if (text) { - // Discord command replies are mirrored to web chat by the Discord - // command adapter, not by the chat bridge. The bridge continues to - // ignore bot-authored Discord messages, which prevents typing helper - // messages and bot replies from feeding back into chat. - sendExternalMessage({ - text, - nickname: client.user?.username || 'Rover bot', - role: 'admin', - roverId: null, - discordGuildId: message.guild.id, - discordGuildName: message.guild.name, - discordGuildIconUrl: message.guild.iconURL?.({ extension: 'png', size: 64 }) || null, - discordChannelId: message.channelId, - discordUserId: client.user?.id || null, - discordUserName: client.user?.username || 'Rover bot', - discordUserAvatarUrl: client.user?.displayAvatarURL?.({ extension: 'png', size: 64 }) || null, - bot: true, - profileImage: client.user?.displayAvatarURL?.({ extension: 'png', size: 64 }) || null, - }); - } - return sent; - }, - }); -} - -client.on('messageCreate', async (message) => { - try { - await integrationHandlers.handleBridgeInbound(message); - const commandMessage = createBridgeMirroredCommandMessage(message); - await commands.handleCommand(createDiscordCommandRequest(commandMessage, { isAdminUser, isLockdownAdminUser })); - } catch (err) { - logger.warn('Error handling Discord message', err.message); - } -}); - -let fleetDailyReports = null; - -function restartFleetDailyReports() { - fleetDailyReports?.stop(); - fleetDailyReports = createFleetDailyReports({ logger, + client, + config, discordConfig, - fleetConfig: config.fleetReports || {}, - fleetReportService, + rovers, roverManager, + getMode, + MODES, + getGlobalObjective, + getActiveDrivers, + getNickname, + subscribe, + sendExternalMessage, + sendExternalTyping, + getGuildConfig, + listGuildConfigs, + attachDmMessage, + getRequestByMessageId, + approveRequest, + denyRequest, + attachPrivateAccessDmMessage, + getPrivateAccessRequestByMessageId, + approvePrivateAccessRequest, + denyPrivateAccessRequest, + getLockdownAdminIds, + isAdminUser, + isLockdownAdminUser, sendToChannel: channelIO.sendToChannel, + fetchChannel: channelIO.fetchChannel, + clearTypingMessage: channelIO.clearTypingMessage, + sendTypingMessage: channelIO.sendTypingMessage, + schedulePresenceRotation: presence.schedulePresenceRotation, + buildReplayVideo, + sanitizeMentions, }); - fleetDailyReports.start(); + + const integrationHandlers = integrations.register(); + + function isTextCommand(content) { + // Both transports share this parser so command detection cannot drift from + // the dispatcher when an installation changes its prefix. + return parseCommandText(content, config).matched; + } + + function isBridgeChannelMessage(message) { + if (!message?.guild?.id || !message?.channelId) return false; + const guildConfig = getGuildConfig(message.guild.id); + return Boolean(guildConfig?.channelId && String(message.channelId) === String(guildConfig.channelId)); + } + + function createBridgeMirroredCommandMessage(message) { + if (!isBridgeChannelMessage(message) || !isTextCommand(message.content)) return message; + const originalReply = message.reply.bind(message); + return Object.assign(Object.create(message), { + reply: async (payload) => { + const sent = await originalReply(payload); + const text = sanitizeMentions(commandReplyToText(payload)); + if (text) { + // Discord command replies are mirrored to web chat by the Discord + // command adapter, not by the chat bridge. The bridge continues to + // ignore bot-authored Discord messages, which prevents typing helper + // messages and bot replies from feeding back into chat. + sendExternalMessage({ + text, + nickname: client.user?.username || 'Rover bot', + role: 'admin', + roverId: null, + discordGuildId: message.guild.id, + discordGuildName: message.guild.name, + discordGuildIconUrl: message.guild.iconURL?.({ extension: 'png', size: 64 }) || null, + discordChannelId: message.channelId, + discordUserId: client.user?.id || null, + discordUserName: client.user?.username || 'Rover bot', + discordUserAvatarUrl: client.user?.displayAvatarURL?.({ extension: 'png', size: 64 }) || null, + bot: true, + profileImage: client.user?.displayAvatarURL?.({ extension: 'png', size: 64 }) || null, + }); + } + return sent; + }, + }); + } + + client.on('messageCreate', async (message) => { + try { + await integrationHandlers.handleBridgeInbound(message); + const commandMessage = createBridgeMirroredCommandMessage(message); + await commands.handleCommand(createDiscordCommandRequest(commandMessage, { isAdminUser, isLockdownAdminUser })); + } catch (err) { + logger.warn('Error handling Discord message', err.message); + } + }); + + let fleetDailyReports = null; + + function restartFleetDailyReports() { + fleetDailyReports?.stop(); + fleetDailyReports = createFleetDailyReports({ + logger, + discordConfig, + fleetConfig: config.fleetReports || {}, + fleetReportService, + roverManager, + sendToChannel: channelIO.sendToChannel, + }); + fleetDailyReports.start(); + } + + client.on('clientReady', () => { + logger.info('Discord bot logged in', { tag: client.user?.tag }); + presence.schedulePresenceRotation(); + // Discord is only a delivery consumer. Starting its scheduler after the bot + // is ready avoids failed sends during login while the collector continues to + // operate independently of Discord availability. + restartFleetDailyReports(); + }); + + return { + client, + refresh() { + if (client.isReady()) restartFleetDailyReports(); + }, + stop() { + for (const event of ['clientReady', 'messageCreate', 'typingStart', 'messageReactionAdd']) { + client.removeAllListeners(event); + } + integrationHandlers.stop(); + unregisterReplayProvider(); + presence.stop(); + channelIO.stop(); + fleetDailyReports?.stop(); + }, + }; } -client.on('ready', () => { - logger.info('Discord bot logged in', { tag: client.user?.tag }); - presence.schedulePresenceRotation(); - // Discord is only a delivery consumer. Starting its scheduler after the bot - // is ready avoids failed sends during login while the collector continues to - // operate independently of Discord availability. - restartFleetDailyReports(); -}); +const connection = createConnectionManager({ createRuntime: createDiscordRuntime, logger }); function replaceObject(target, source = {}) { Object.keys(target).forEach((key) => delete target[key]); Object.assign(target, structuredClone(source)); } -function applyDiscordConfig(nextDiscordConfig = {}) { - const wasEnabled = enabled; - const previousToken = discordConfig.token; +async function applyDiscordConfig(nextDiscordConfig = {}) { replaceObject(discordConfig, nextDiscordConfig); config.discord = discordConfig; enabled = Boolean(discordConfig.enabled); - - if (!enabled) { - fleetDailyReports?.stop(); - fleetDailyReports = null; - if (wasEnabled) client.destroy(); - return; - } - if (!wasEnabled || previousToken !== discordConfig.token) { - if (wasEnabled) client.destroy(); - // Login health is reported by Discord itself; do not hold the committed - // configuration request open while an external network service connects. - client.login(discordConfig.token).catch((err) => { - logger.error('Discord login failed after configuration change', err.message); - }); - } else if (client.isReady()) { - restartFleetDailyReports(); - } + await connection.configure(discordConfig); + connection.refresh(); } function applySharedConfigSection(section, value) { config[section] = structuredClone(value); // Fleet delivery owns a timer derived from both Discord and fleet settings. // Reconnecting is unnecessary; rebuild only that scheduler when ready. - if (section === 'fleetReports' && client.isReady()) { - restartFleetDailyReports(); + if (section === 'fleetReports') { + connection.refresh(); } } @@ -458,10 +462,8 @@ registerConfigurationHandler('discord', applyDiscordConfig); registerConfigurationHandler(section, (value) => applySharedConfigSection(section, value)); }); -if (enabled) { - client.login(discordConfig.token).catch((err) => { - logger.error('Discord login failed', err.message); - }); -} +connection.configure(discordConfig).catch((err) => { + logger.error('Discord startup failed', err.message); +}); module.exports = {}; diff --git a/server/src/services/discordBotService/integrations/index.js b/server/src/services/discordBotService/integrations/index.js index 603e2c4e..70d0313f 100644 --- a/server/src/services/discordBotService/integrations/index.js +++ b/server/src/services/discordBotService/integrations/index.js @@ -24,6 +24,7 @@ function createIntegrations(deps) { const userAnnouncements = createUserAnnouncements({ ...deps, sendToChannel, schedulePresenceRotation }); function register() { + let stopped = false; client.on('typingStart', (typing) => { chat.handleDiscordTypingStart(typing).catch((err) => logger.warn('Error handling Discord typing', err.message)); }); @@ -33,14 +34,31 @@ function createIntegrations(deps) { dm.handlePrivateAccessReaction(reaction, user).catch((err) => logger.warn('Error handling private access reaction', err.message)); }); - subscribe('*', handleBusEvent); - subscribe('*', userAnnouncements.handleBusEvent); - subscribe('verification.requested', dm.sendVerificationRequestDms); - subscribe('privateRoverAccess.requested', dm.sendPrivateRoverAccessRequestDms); - subscribe('chat:message', chat.handleChatBridgeOutbound); - subscribe('chat:typing', chat.handleChatTypingOutbound); + // Only the active, ready connection consumes server events. Retain each + // unsubscribe function so replacement clients never duplicate deliveries. + const unsubscribe = []; + function subscribeWhileReady(type, handler) { + unsubscribe.push(subscribe(type, (event) => { + Promise.resolve().then(() => { + if (!stopped && client.isReady()) return handler(event); + }).catch((err) => { + logger.warn('Error handling Discord integration event', { type, error: err.message }); + }); + })); + } + + subscribeWhileReady('*', handleBusEvent); + subscribeWhileReady('*', userAnnouncements.handleBusEvent); + subscribeWhileReady('verification.requested', dm.sendVerificationRequestDms); + subscribeWhileReady('privateRoverAccess.requested', dm.sendPrivateRoverAccessRequestDms); + subscribeWhileReady('chat:message', chat.handleChatBridgeOutbound); + subscribeWhileReady('chat:typing', chat.handleChatTypingOutbound); return { + stop() { + stopped = true; + unsubscribe.forEach((remove) => remove()); + }, handleBridgeInbound: chat.handleBridgeInbound, handleBusEvent, }; diff --git a/server/src/services/discordBotService/presence.js b/server/src/services/discordBotService/presence.js index 1b478c27..227dbc5c 100644 --- a/server/src/services/discordBotService/presence.js +++ b/server/src/services/discordBotService/presence.js @@ -7,6 +7,7 @@ function createPresenceManager({ client, logger, getMode, getGlobalObjective, co const PRESENCE_ROTATE_MS = 20000; let presenceInterval = null; let presenceShowObjective = false; + let stopped = false; function truncatePresenceText(text, maxLength) { if (!text) return ''; @@ -40,6 +41,7 @@ function createPresenceManager({ client, logger, getMode, getGlobalObjective, co } function schedulePresenceRotation() { + if (stopped) return; if (presenceInterval) { clearInterval(presenceInterval); presenceInterval = null; @@ -59,6 +61,11 @@ function createPresenceManager({ client, logger, getMode, getGlobalObjective, co } return { + stop() { + stopped = true; + clearInterval(presenceInterval); + presenceInterval = null; + }, schedulePresenceRotation, }; }