From 5aa57f884f144c86413480b8aafb7bf367a56a86 Mon Sep 17 00:00:00 2001 From: MythEclipse Date: Thu, 14 May 2026 04:02:25 +0700 Subject: [PATCH] feat: add API endpoint for syncing selected channel backlog --- src/moderation/backlogSync.ts | 153 +++++++++------------------------- src/webserver.ts | 27 ++++++ 2 files changed, 65 insertions(+), 115 deletions(-) diff --git a/src/moderation/backlogSync.ts b/src/moderation/backlogSync.ts index a3bd9e8..e480c0d 100644 --- a/src/moderation/backlogSync.ts +++ b/src/moderation/backlogSync.ts @@ -6,65 +6,6 @@ import { captureMessage } from "./messageCapture"; const logger = createChildLogger("backlog-sync"); -function isWatchableChannel(channel: { type?: string; messages?: unknown }): boolean { - return Boolean( - channel.messages && - ["GUILD_TEXT", "GUILD_PUBLIC_THREAD", "GUILD_PRIVATE_THREAD"].includes( - channel.type ?? "", - ), - ); -} - -async function collectWatchableChannels(guild: any): Promise { - const channels: any[] = []; - - // Fast pass: collect text channels from cache only - for (const channel of guild.channels.cache.values()) { - if (isWatchableChannel(channel)) { - channels.push(channel); - } - } - - // Slow pass: discover threads with timeout per channel (non-blocking to message sync) - const threadPromises: Promise[] = []; - for (const channel of guild.channels.cache.values()) { - if (!channel.threads?.fetch) continue; - - threadPromises.push( - (async () => { - for (const archived of [false, true]) { - try { - const controller = new AbortController(); - const timeout = setTimeout(() => controller.abort(), 5000); - const fetched = await Promise.race([ - channel.threads.fetch({ archived, limit: 100 }), - new Promise((_, reject) => controller.signal.addEventListener('abort', () => reject(new Error('timeout')))), - ]).catch(() => null); - clearTimeout(timeout); - - if (!fetched?.threads) continue; - for (const thread of fetched.threads.values()) { - if (isWatchableChannel(thread)) channels.push(thread); - } - } catch { - // Skip this channel's threads on timeout/error - } - } - })() - ); - } - - // Wait for all thread discoveries with overall timeout - await Promise.race([ - Promise.all(threadPromises), - new Promise((resolve) => setTimeout(resolve, 30000)), - ]).catch(() => { - logger.warn("Thread discovery timeout, proceeding with cached channels"); - }); - - return Array.from(new Map(channels.map((channel) => [channel.id, channel])).values()); -} - async function syncChannelMessages( db: SqliteDatabase, channel: any, @@ -116,63 +57,45 @@ export async function syncBacklogMessages( return; } - const cutoffTime = Date.now() - config.BACKLOG_SYNC_HOURS * 60 * 60 * 1000; - logger.info( - { guildId: guild.id, hours: config.BACKLOG_SYNC_HOURS }, - "Starting message backlog sync", - ); + logger.info({ guildId: guild.id }, "Backlog sync ready (will sync on-demand per selected channel)"); +} - logger.info({ guildId: guild.id }, "Fetching guild channels for backlog sync"); - await guild.channels.fetch().catch((error) => { - logger.warn( - { guildId: guild.id, error: error instanceof Error ? error.message : String(error) }, - "Failed to fetch guild channels before backlog sync", - ); - return null; - }); - - logger.info({ guildId: guild.id }, "Collecting watchable channels for backlog sync"); - const channels = await collectWatchableChannels(guild); - - let total = 0; - logger.info( - { guildId: guild.id, channels: channels.length, hours: config.BACKLOG_SYNC_HOURS }, - "Watchable channels collected for backlog sync", - ); - - // Sync channels in parallel with concurrency limit of 3 - const concurrency = 3; - const queue = [...channels]; - const active: Promise[] = []; - - while (queue.length > 0 || active.length > 0) { - while (active.length < concurrency && queue.length > 0) { - const channel = queue.shift()!; - const promise = (async () => { - try { - const count = await syncChannelMessages(db, channel as any, cutoffTime); - logger.info({ channelId: channel.id, count }, "Backlog channel sync completed"); - return count; - } catch (error) { - logger.warn( - { - channelId: channel.id, - error: error instanceof Error ? error.message : String(error), - }, - "Backlog channel sync failed", - ); - return 0; - } - })(); - active.push(promise); - } - - if (active.length > 0) { - const result = await Promise.race(active); - total += result; - active.splice(active.findIndex((p) => p === Promise.resolve(result)), 1); - } +export async function syncSelectedChannelBacklog( + client: Client, + db: SqliteDatabase, + guildId: string, + channelId: string, +): Promise { + const guild = client.guilds.cache.get(guildId); + if (!guild) { + logger.warn({ guildId }, "Guild not found for backlog sync"); + return 0; } - logger.info({ total }, "Message backlog sync completed"); + const channel = guild.channels.cache.get(channelId); + if (!channel) { + logger.warn({ guildId, channelId }, "Channel not found for backlog sync"); + return 0; + } + + const cutoffTime = Date.now() - config.BACKLOG_SYNC_HOURS * 60 * 60 * 1000; + logger.info( + { guildId, channelId, hours: config.BACKLOG_SYNC_HOURS }, + "Starting backlog sync for selected channel", + ); + + try { + const count = await syncChannelMessages(db, channel as any, cutoffTime); + logger.info({ channelId, count }, "Backlog sync completed for selected channel"); + return count; + } catch (error) { + logger.warn( + { + channelId, + error: error instanceof Error ? error.message : String(error), + }, + "Backlog sync failed for selected channel", + ); + return 0; + } } diff --git a/src/webserver.ts b/src/webserver.ts index e32b58d..fdc1350 100644 --- a/src/webserver.ts +++ b/src/webserver.ts @@ -12,6 +12,7 @@ import { discordPlayer } from "./player"; import type { VoiceController } from "./voiceController"; import { getDatabase, getPersistedValue, setPersistedValue } from "./muxer-queue"; import { getMessagesByChannel, getAttachmentsByChannel } from "./moderation/messageStore"; +import { syncSelectedChannelBacklog } from "./moderation/backlogSync"; const wsLogger = createChildLogger("webserver"); @@ -277,6 +278,32 @@ export function startWebserver( } }); + app.post("/api/backlog-sync", async (req, res, next) => { + try { + const { guildId, channelId } = req.body as { + guildId?: string; + channelId?: string; + }; + + if (!guildId || !channelId) { + throw new AppError( + "guildId and channelId are required", + "MISSING_BACKLOG_PARAMS", + 400, + ); + } + + const count = await syncSelectedChannelBacklog(_client, getDatabase(), guildId, channelId); + res.json({ + success: true, + channelId, + messagesSync: count, + }); + } catch (error) { + next(error); + } + }); + // Inbound: Discord PCM → tagged chunks → browser (global as any).broadcastPcmToWeb = (chunk: Buffer, userId: string) => { let hash = 0;