From 401155b50176910d39d43ecfb5229d80b06166c0 Mon Sep 17 00:00:00 2001 From: asepharyana Date: Wed, 23 Sep 2026 22:01:48 +0700 Subject: [PATCH] fix: auto-fix code quality [skip ci] --- services/backend/src/e2e.test.ts | 11 +- .../modules/messages/messages.repository.ts | 147 ++++++++++-------- .../src/modules/messages/messages.service.ts | 44 ++++-- services/backend/src/orpc/router.ts | 2 - services/backend/src/shared/config/index.ts | 5 +- services/backend/src/ws/broadcast.ts | 8 +- services/backend/src/ws/redis-bridge.ts | 17 +- services/backend/src/ws/server.ts | 47 +++--- services/discord-gateway/src/app/bootstrap.ts | 106 +------------ services/discord-gateway/src/app/retention.ts | 26 +++- services/discord-gateway/src/app/shutdown.ts | 9 +- .../modules/command-handler/commandHandler.ts | 13 +- 12 files changed, 200 insertions(+), 235 deletions(-) diff --git a/services/backend/src/e2e.test.ts b/services/backend/src/e2e.test.ts index 6145a5c3..7d6fb8a8 100644 --- a/services/backend/src/e2e.test.ts +++ b/services/backend/src/e2e.test.ts @@ -6,7 +6,13 @@ import { describe, expect, it } from "vitest"; const BASE = process.env.API_BASE ?? "http://localhost:4001/api"; -async function api(path: string, init?: RequestInit) { +/** Result of a JSON API call: status + parsed body (or null for 204/empty). */ +interface ApiResult { + status: number; + body: Record | null; +} + +async function api(path: string, init?: RequestInit): Promise { const res = await fetch(`${BASE}${path}`, { ...init, headers: { "Content-Type": "application/json", ...init?.headers }, @@ -23,7 +29,8 @@ describe("API Health", () => { }); it("GET /metrics returns prometheus text", async () => { - const res = await fetch(`${BASE.replace("/api", "")}/api/metrics`); + const base = BASE.endsWith("/api") ? BASE.slice(0, -4) : BASE; + const res = await fetch(`${base}/api/metrics`); expect(res.status).toBe(200); const text = await res.text(); expect(text).toContain("nodejs"); diff --git a/services/backend/src/modules/messages/messages.repository.ts b/services/backend/src/modules/messages/messages.repository.ts index aba4fc9f..c1d82c50 100644 --- a/services/backend/src/modules/messages/messages.repository.ts +++ b/services/backend/src/modules/messages/messages.repository.ts @@ -51,10 +51,51 @@ export interface AttachmentResult { uploaded_at: number | null; } +type MessageRow = ReturnType; + +export type { MessageRow }; + +/** + * Build the NULL-safe "exclude spam threads" condition. Non-thread messages + * (NULL thread_id) are always kept; thread messages are kept only when their + * thread is not in the configured exclusion list. + */ +function excludeSpamThreads(): SQL | undefined { + if (EXCLUDED_THREAD_IDS.length === 0) return undefined; + return or( + isNull(pgMessagesTable.thread_id), + notInArray(pgMessagesTable.thread_id, EXCLUDED_THREAD_IDS), + ); +} + +/** Normalize a raw attachment DB row to the API shape. */ +function mapAttachmentRow(r: Record): AttachmentResult { + return { + id: String(r.id ?? ""), + message_id: String(r.message_id ?? ""), + guild_id: String(r.guild_id ?? ""), + channel_id: String(r.channel_id ?? ""), + thread_id: (r.thread_id as string | null) ?? null, + user_id: String(r.user_id ?? ""), + filename: String(r.filename ?? ""), + size: Number(r.size ?? 0), + type: String(r.type ?? ""), + discord_url: String(r.discord_url ?? ""), + uploaded_url: (r.uploaded_url as string | null) ?? null, + upload_status: String(r.upload_status ?? "pending"), + upload_error: (r.upload_error as string | null) ?? null, + created_at: Number(r.created_at ?? 0), + uploaded_at: (r.uploaded_at as number | null) ?? null, + }; +} + +/** Select the first `limit + 1` rows so the caller can derive the next cursor. */ +function cursorLimit(limit: number): number { + return limit + 1; +} + export class MessagesRepository { - async findMany( - query: MessageQuery, - ): Promise>> { + async findMany(query: MessageQuery): Promise> { const db = getDatabase(); const limit = query.limit ?? 50; const conditions: SQL[] = []; @@ -76,13 +117,8 @@ export class MessagesRepository { } // Exclude spam threads (NULL-safe: non-thread messages are kept) - if (EXCLUDED_THREAD_IDS.length > 0) { - const excludeThreads = or( - isNull(pgMessagesTable.thread_id), - notInArray(pgMessagesTable.thread_id, EXCLUDED_THREAD_IDS), - ); - if (excludeThreads) conditions.push(excludeThreads); - } + const excludeThreads = excludeSpamThreads(); + if (excludeThreads) conditions.push(excludeThreads); const where = conditions.length > 0 ? and(...conditions) : undefined; const rows = await db @@ -90,7 +126,7 @@ export class MessagesRepository { .from(pgMessagesTable) .where(where) .orderBy(desc(pgMessagesTable.created_at)) - .limit(limit + 1); + .limit(cursorLimit(limit)); const data = rows .slice(0, limit) @@ -138,7 +174,7 @@ export class MessagesRepository { async findByChannel( channelId: string, query: MessageQuery, - ): Promise>> { + ): Promise> { const db = getDatabase(); const limit = query.limit ?? 50; const conditions: SQL[] = [eq(pgMessagesTable.channel_id, channelId)]; @@ -148,20 +184,15 @@ export class MessagesRepository { } // Exclude spam threads (NULL-safe) - if (EXCLUDED_THREAD_IDS.length > 0) { - const excludeThreads = or( - isNull(pgMessagesTable.thread_id), - notInArray(pgMessagesTable.thread_id, EXCLUDED_THREAD_IDS), - ); - if (excludeThreads) conditions.push(excludeThreads); - } + const excludeThreads = excludeSpamThreads(); + if (excludeThreads) conditions.push(excludeThreads); const rows = await db .select() .from(pgMessagesTable) .where(and(...conditions)) .orderBy(desc(pgMessagesTable.created_at)) - .limit(limit + 1); + .limit(cursorLimit(limit)); const data = rows .slice(0, limit) @@ -181,7 +212,7 @@ export class MessagesRepository { async *streamMany( query: MessageQuery, pageSize = 50, - ): AsyncGenerator, void, unknown> { + ): AsyncGenerator { const conditions: SQL[] = []; if (query.guildId) { @@ -196,13 +227,8 @@ export class MessagesRepository { if (query.status) { conditions.push(eq(pgMessagesTable.ai_status, query.status)); } - if (EXCLUDED_THREAD_IDS.length > 0) { - const excludeThreads = or( - isNull(pgMessagesTable.thread_id), - notInArray(pgMessagesTable.thread_id, EXCLUDED_THREAD_IDS), - ); - if (excludeThreads) conditions.push(excludeThreads); - } + const excludeThreads = excludeSpamThreads(); + if (excludeThreads) conditions.push(excludeThreads); const where = conditions.length > 0 ? and(...conditions) : undefined; let cursor: string | undefined = query.cursor; @@ -221,7 +247,7 @@ export class MessagesRepository { .from(pgMessagesTable) .where(pageWhere) .orderBy(desc(pgMessagesTable.created_at)) - .limit(pageSize + 1); + .limit(cursorLimit(pageSize)); if (rows.length === 0) return; @@ -366,31 +392,30 @@ export class MessagesRepository { async getImageMessages( guildId: string, limit: number = 50, - ): Promise>> { + ): Promise> { const db = getDatabase(); // Subquery: find distinct message_ids from attachments with image MIME type + const attachmentConditions: SQL[] = [ + eq(pgAttachmentsTable.guild_id, guildId), + like(pgAttachmentsTable.type, "image/%"), + ]; + // Exclude spam threads (NULL-safe for non-thread messages) + const excludeThreads = + EXCLUDED_THREAD_IDS.length > 0 + ? or( + isNull(pgAttachmentsTable.thread_id), + notInArray(pgAttachmentsTable.thread_id, EXCLUDED_THREAD_IDS), + ) + : undefined; + if (excludeThreads) attachmentConditions.push(excludeThreads); + const imageMsgIds = db .select({ id: pgAttachmentsTable.message_id }) .from(pgAttachmentsTable) - .where( - and( - eq(pgAttachmentsTable.guild_id, guildId), - like(pgAttachmentsTable.type, "image/%"), - // Exclude spam threads (NULL-safe for non-thread messages) - ...(EXCLUDED_THREAD_IDS.length > 0 - ? (() => { - const excludeThreads = or( - isNull(pgAttachmentsTable.thread_id), - notInArray(pgAttachmentsTable.thread_id, EXCLUDED_THREAD_IDS), - ); - return excludeThreads ? [excludeThreads] : []; - })() - : []), - ), - ) + .where(and(...attachmentConditions)) .orderBy(desc(pgAttachmentsTable.created_at)) - .limit(limit + 1); + .limit(cursorLimit(limit)); // Fetch full message rows for those IDs const rows = await db @@ -398,7 +423,7 @@ export class MessagesRepository { .from(pgMessagesTable) .where(inArray(pgMessagesTable.id, imageMsgIds)) .orderBy(desc(pgMessagesTable.created_at)) - .limit(limit + 1); + .limit(cursorLimit(limit)); const data = rows .slice(0, limit) @@ -433,28 +458,16 @@ export class MessagesRepository { .from(pgAttachmentsTable) .where(and(...conditions)) .orderBy(desc(pgAttachmentsTable.created_at)) - .limit(limit + 1); + .limit(cursorLimit(limit)); - const data = rows.map((r) => ({ - id: String(r.id ?? ""), - message_id: String(r.message_id ?? ""), - guild_id: String(r.guild_id ?? ""), - channel_id: String(r.channel_id ?? ""), - thread_id: (r.thread_id as string | null) ?? null, - user_id: String(r.user_id ?? ""), - filename: String(r.filename ?? ""), - size: Number(r.size ?? 0), - type: String(r.type ?? ""), - discord_url: String(r.discord_url ?? ""), - uploaded_url: (r.uploaded_url as string | null) ?? null, - upload_status: String(r.upload_status ?? "pending"), - upload_error: (r.upload_error as string | null) ?? null, - created_at: Number(r.created_at ?? 0), - uploaded_at: (r.uploaded_at as number | null) ?? null, - })); + const data = rows.map((r) => + mapAttachmentRow(r as Record), + ); + // nextCursor derives from the fetched-but-untrimmed overflow row (index + // `limit`), matching the other cursor-paginated queries. const nextCursor = - data.length > limit ? String(data[limit].created_at) : null; + rows.length > limit ? String(rows[limit].created_at) : null; const trimmed = data.slice(0, limit); return { data: trimmed, nextCursor }; diff --git a/services/backend/src/modules/messages/messages.service.ts b/services/backend/src/modules/messages/messages.service.ts index e4aa84bc..dfe5e452 100644 --- a/services/backend/src/modules/messages/messages.service.ts +++ b/services/backend/src/modules/messages/messages.service.ts @@ -2,14 +2,16 @@ import { config } from "@/shared/config/index"; import { NotFoundError, ValidationError } from "@/shared/errors/index"; import { createChildLogger } from "@/shared/logger/index"; import { embedQuery } from "./embed.js"; -import { messagesRepository } from "./messages.repository.js"; +import { type MessageRow, messagesRepository } from "./messages.repository.js"; import type { MessageQuery, SemanticSearchQuery } from "./messages.schema.js"; import { searchArchive } from "./qdrant.js"; const logger = createChildLogger("messages.service"); export class MessagesService { - async listMessages(query: MessageQuery) { + async listMessages( + query: MessageQuery, + ): Promise>> { if (!query.channelId && !query.guildId) { throw new ValidationError("Either channelId or guildId is required"); } @@ -26,7 +28,10 @@ export class MessagesService { return messagesRepository.streamMany(query, pageSize); } - async getMessagesByChannel(channelId: string, query: MessageQuery) { + async getMessagesByChannel( + channelId: string, + query: MessageQuery, + ): Promise>> { if (!channelId) { throw new ValidationError("channelId is required"); } @@ -35,7 +40,14 @@ export class MessagesService { return messagesRepository.findByChannel(channelId, query); } - async getMessageById(id: string) { + async getMessageById(id: string): Promise< + NonNullable>> & { + edit_count: number; + edit_history: Awaited< + ReturnType + >; + } + > { if (!id) { throw new ValidationError("message ID is required"); } @@ -56,7 +68,12 @@ export class MessagesService { }; } - async getAttachmentsByChannel(channelId: string, query: MessageQuery) { + async getAttachmentsByChannel( + channelId: string, + query: MessageQuery, + ): Promise< + Awaited> + > { if (!channelId) { throw new ValidationError("channelId is required"); } @@ -111,22 +128,31 @@ export class MessagesService { return { results, nextCursor: null }; } - async getActivity(days = 30) { + async getActivity( + days = 30, + ): Promise>> { return messagesRepository.getActivity(days); } - async getRecentEdits(limit = 50, channelId?: string) { + async getRecentEdits( + limit = 50, + channelId?: string, + ): Promise>> { logger.debug({ limit, channelId }, "Getting recent message edits"); return messagesRepository.getRecentEdits(limit, channelId); } /** Distinct guilds present in the message archive (guild picker). */ - async getGuilds() { + async getGuilds(): Promise< + Awaited> + > { return messagesRepository.listGuilds(); } /** Text channels for a guild (channel picker). */ - async getTextChannels(guildId: string) { + async getTextChannels( + guildId: string, + ): Promise>> { return messagesRepository.listTextChannels(guildId); } } diff --git a/services/backend/src/orpc/router.ts b/services/backend/src/orpc/router.ts index 14aad4ba..6ff4e4b1 100644 --- a/services/backend/src/orpc/router.ts +++ b/services/backend/src/orpc/router.ts @@ -222,8 +222,6 @@ const moderationRouter = { .handler(({ input }) => moderationService.getCoverage(input.days)), }; -// ── Recordings (removed) / Voice (removed) ───────────────────── -// Music/media playback (media.mjs) was removed with the voice feature. // ── Analysis (search) ────────────────────────────────────────────── const analysisRouter = { search: os diff --git a/services/backend/src/shared/config/index.ts b/services/backend/src/shared/config/index.ts index fd1606c6..a9482fc4 100644 --- a/services/backend/src/shared/config/index.ts +++ b/services/backend/src/shared/config/index.ts @@ -229,9 +229,8 @@ export const configSchema = z .default(true), }) .superRefine((value, ctx) => { - if (!value.AI_ANALYSIS_ENABLED) { - // skip: AI analysis not enabled - } else if (!value.AI_LLM_API_KEY) { + // AI analysis requires credentials to run. + if (value.AI_ANALYSIS_ENABLED && !value.AI_LLM_API_KEY) { ctx.addIssue({ code: z.ZodIssueCode.custom, path: ["AI_LLM_API_KEY"], diff --git a/services/backend/src/ws/broadcast.ts b/services/backend/src/ws/broadcast.ts index a53347f8..1a98d75b 100644 --- a/services/backend/src/ws/broadcast.ts +++ b/services/backend/src/ws/broadcast.ts @@ -46,17 +46,11 @@ export function clearBroadcastFunctions(): void { logger.info("Broadcast functions cleared"); } -function shouldLog(type: string): boolean { - if (!_enabled) return false; - // Avoid logging high-volume events - return true; -} - /** * Broadcast a JSON event to all connected WebSocket clients. */ export function broadcastEvent(type: string, data: unknown): void { - if (shouldLog(type)) { + if (_enabled) { logger.debug({ event: type }, "Broadcasting event"); } _broadcast?.(type, data); diff --git a/services/backend/src/ws/redis-bridge.ts b/services/backend/src/ws/redis-bridge.ts index 8ed8aad1..c62cb374 100644 --- a/services/backend/src/ws/redis-bridge.ts +++ b/services/backend/src/ws/redis-bridge.ts @@ -9,6 +9,14 @@ const logger = createChildLogger("ws.redis-bridge"); /** Channels we subscribe to = all keys in DISCORD_CHANNEL_TO_WS_EVENT */ const SUBSCRIPTION_CHANNELS = Object.keys(DISCORD_CHANNEL_TO_WS_EVENT); +/** Shape of the DiscordGatewayEvent envelope published by the gateway. */ +interface GatewayEnvelope { + type?: string; + data?: unknown; + timestamp?: number; + source?: string; +} + let subscriber: Redis | null = null; function createSubscriber(): Redis { @@ -22,14 +30,9 @@ function handleSubscriptionMessage(channel: string, message: string): void { return; } - let envelope: { - type?: string; - data?: unknown; - timestamp?: number; - source?: string; - }; + let envelope: GatewayEnvelope; try { - envelope = JSON.parse(message); + envelope = JSON.parse(message) as GatewayEnvelope; } catch (err) { logger.error({ channel, err }, "Failed to parse Redis message as JSON"); return; diff --git a/services/backend/src/ws/server.ts b/services/backend/src/ws/server.ts index bdc892ec..84bc6a4d 100644 --- a/services/backend/src/ws/server.ts +++ b/services/backend/src/ws/server.ts @@ -1,6 +1,7 @@ import type { IncomingMessage, Server } from "node:http"; import type { Duplex } from "node:stream"; import { WebSocket, WebSocketServer } from "ws"; +import type { MessageQuery } from "../modules/messages/messages.schema.js"; import { messagesService } from "../modules/messages/messages.service.js"; import { createChildLogger } from "../shared/logger/index.js"; import { setBroadcastFunctions } from "./broadcast.js"; @@ -18,6 +19,14 @@ interface JsonMessage { payload?: Record; } +/** Payload accepted by the `stream_messages` JSON command. */ +interface StreamMessagesPayload { + guildId?: string; + channelId?: string; + cursor?: string; + limit?: number; +} + // Track the active WebSocket server for lifecycle management let _wss: WebSocketServer | null = null; @@ -81,12 +90,7 @@ export function createWebSocketServer(server: Server): WebSocketServer { jsonHandlers.set("stream_messages", async (ws, message) => { if (ws.readyState !== WebSocket.OPEN) return; - const payload = (message.payload ?? {}) as { - guildId?: string; - channelId?: string; - cursor?: string; - limit?: number; - }; + const payload = (message.payload ?? {}) as StreamMessagesPayload; const guildId = payload.guildId; const channelId = payload.channelId; if (!guildId && !channelId) { @@ -97,6 +101,17 @@ export function createWebSocketServer(server: Server): WebSocketServer { const pageSize = 50; // internal DB page size; still emitted one frame at a time const maxFrames = Math.min(payload.limit ?? 200, 500); + /** Send the end-of-stream frame, reporting sent count + next cursor. */ + function sendEnd(data: Record): void { + if (ws.readyState !== WebSocket.OPEN) return; + ws.send( + JSON.stringify({ + type: "message_snapshot_end", + data, + }), + ); + } + let sent = 0; let nextCursor: string | null = null; try { @@ -105,7 +120,7 @@ export function createWebSocketServer(server: Server): WebSocketServer { guildId, channelId, cursor: payload.cursor, - } as never, + } as MessageQuery, pageSize, )) { if (ws.readyState !== WebSocket.OPEN) break; @@ -122,24 +137,10 @@ export function createWebSocketServer(server: Server): WebSocketServer { sent++; if (sent >= maxFrames) break; } - if (ws.readyState === WebSocket.OPEN) { - ws.send( - JSON.stringify({ - type: "message_snapshot_end", - data: { sent, nextCursor }, - }), - ); - } + sendEnd({ sent, nextCursor }); } catch (err) { logger.error({ err }, "stream_messages failed"); - if (ws.readyState === WebSocket.OPEN) { - ws.send( - JSON.stringify({ - type: "message_snapshot_end", - data: { sent, nextCursor, error: true }, - }), - ); - } + sendEnd({ sent, nextCursor, error: true }); } }); diff --git a/services/discord-gateway/src/app/bootstrap.ts b/services/discord-gateway/src/app/bootstrap.ts index 5e12e47d..3fd33b09 100644 --- a/services/discord-gateway/src/app/bootstrap.ts +++ b/services/discord-gateway/src/app/bootstrap.ts @@ -1,6 +1,4 @@ import { Client } from "discord.js-selfbot-v13"; -import { inArray, lt } from "drizzle-orm"; -import type { NodePgDatabase } from "drizzle-orm/node-postgres"; import { ConfigError, DatabaseError } from "@/shared/errors/index"; import { createChildLogger } from "@/shared/logger/index"; import { @@ -36,117 +34,15 @@ import { registerPresenceCapture } from "../modules/user-presence/index.js"; import { config } from "../shared/config/config.js"; import { closeDatabase, - getDatabase, initializeDatabase, } from "../shared/database/drizzle.js"; import { runMigrations } from "../shared/database/migrate.js"; -import type * as schema from "../shared/database/schema.js"; -import { attachmentsTable, messagesTable } from "../shared/database/schema.js"; import { createDiscordClientOptions } from "../shared/discord/clientOptions.js"; +import { startRetentionCleanup } from "./retention.js"; import { createGracefulShutdown } from "./shutdown.js"; const logger = createChildLogger("discord-gateway"); -// ─── Retention Cleanup ───────────────────────────────────────────────────── - -async function deleteExpiredRecords( - table: any, - timestampField: any, - days: number | undefined, - dryRun: boolean, - label: string, -): Promise { - if (!days || days <= 0) { - logger.debug({ label }, `Retention disabled for ${label}`); - return; - } - - const cutoff = Date.now() - days * 24 * 60 * 60 * 1000; - const db = getDatabase() as unknown as NodePgDatabase; - - const expired = await db - .select({ id: table.id }) - .from(table) - .where(lt(timestampField, cutoff)) - .limit(1000); - - if (expired.length === 0) { - logger.debug({ label }, `No expired ${label} found`); - return; - } - - logger.info({ count: expired.length, label }, `Found expired ${label}`); - - if (dryRun) { - logger.info( - { count: expired.length, label }, - `[DRY RUN] Would delete ${expired.length} ${label}`, - ); - return; - } - - try { - await db.delete(table).where( - inArray( - table.id, - expired.map((r) => r.id), - ), - ); - logger.info({ count: expired.length, label }, `Deleted expired ${label}`); - } catch (err) { - logger.error({ err, label }, `Failed to delete expired ${label}`); - } -} - -function startRetentionCleanup(): void { - const intervalMs = config.RETENTION_CLEANUP_INTERVAL_MS; - const dryRun = config.RETENTION_DRY_RUN; - - logger.info( - { - intervalMs, - dryRun, - messagesDays: config.RETENTION_MESSAGES_DAYS, - attachmentsDays: config.RETENTION_ATTACHMENTS_DAYS, - }, - "Starting retention cleanup scheduler", - ); - - async function runCleanupTick(): Promise { - await deleteExpiredRecords( - messagesTable, - messagesTable.created_at, - config.RETENTION_MESSAGES_DAYS, - dryRun, - "messages", - ); - await deleteExpiredRecords( - attachmentsTable, - attachmentsTable.created_at, - config.RETENTION_ATTACHMENTS_DAYS, - dryRun, - "attachments", - ); - } - - // Run immediately on start, then schedule - runCleanupTick().catch((error) => { - logger.error( - { error: error instanceof Error ? error.message : String(error) }, - "Initial retention cleanup tick failed", - ); - }); - - setInterval(() => { - runCleanupTick().catch((error) => { - logger.error( - { error: error instanceof Error ? error.message : String(error) }, - "Retention cleanup tick failed", - ); - }); - }, intervalMs); -} - // ─── Bootstrap ───────────────────────────────────────────────────────────── export async function initializeDiscordGateway() { diff --git a/services/discord-gateway/src/app/retention.ts b/services/discord-gateway/src/app/retention.ts index 94a2adbd..d8a6104b 100644 --- a/services/discord-gateway/src/app/retention.ts +++ b/services/discord-gateway/src/app/retention.ts @@ -1,5 +1,8 @@ import { inArray, lt } from "drizzle-orm"; -import type { NodePgDatabase } from "drizzle-orm/node-postgres"; +import type { + NodePgDatabase, + NodePgQueryResultHKT, +} from "drizzle-orm/node-postgres"; import { createChildLogger } from "@/shared/logger/index"; import { config } from "../shared/config/config.js"; import { getDatabase } from "../shared/database/drizzle.js"; @@ -8,11 +11,26 @@ import { attachmentsTable, messagesTable } from "../shared/database/schema.js"; const logger = createChildLogger("discord-gateway"); +/** DB handle typed with the full schema so table/column refs resolve. */ +type GatewayDatabase = NodePgDatabase; + +/** Tables eligible for retention cleanup: string `id` + numeric `created_at`. */ +type RetentionTable = typeof messagesTable | typeof attachmentsTable; + +type RetentionTimestampColumn = + | typeof messagesTable.created_at + | typeof attachmentsTable.created_at; + // ─── Retention Cleanup ───────────────────────────────────────────────────── +/** + * Delete rows older than `days` in `table`, in batches of up to 1000 ids. + * When `dryRun` is set, logs what would be deleted without deleting. + * Returns immediately (no-op) when `days` is unset or <= 0. + */ async function deleteExpiredRecords( - table: any, - timestampField: any, + table: RetentionTable, + timestampField: RetentionTimestampColumn, days: number | undefined, dryRun: boolean, label: string, @@ -23,7 +41,7 @@ async function deleteExpiredRecords( } const cutoff = Date.now() - days * 24 * 60 * 60 * 1000; - const db = getDatabase() as unknown as NodePgDatabase; + const db = getDatabase() as unknown as GatewayDatabase; const expired = await db .select({ id: table.id }) diff --git a/services/discord-gateway/src/app/shutdown.ts b/services/discord-gateway/src/app/shutdown.ts index 94aaed0c..6debd98a 100644 --- a/services/discord-gateway/src/app/shutdown.ts +++ b/services/discord-gateway/src/app/shutdown.ts @@ -18,10 +18,15 @@ export interface GracefulShutdownOptions { stopMetricsServer?: StopMetricsServer; } -export function createGracefulShutdown(options: GracefulShutdownOptions) { +export type GracefulShutdown = (signal: string) => Promise; + +/** Create a shutdown handler that can only be triggered once. */ +export function createGracefulShutdown( + options: GracefulShutdownOptions, +): GracefulShutdown { let isShuttingDown = false; - return async function gracefulShutdown(signal: string) { + return async function gracefulShutdown(signal: string): Promise { if (isShuttingDown) { options.logger.warn(`Already shutting down, ignoring ${signal}`); return; diff --git a/services/discord-gateway/src/modules/command-handler/commandHandler.ts b/services/discord-gateway/src/modules/command-handler/commandHandler.ts index a7324a27..d9750219 100644 --- a/services/discord-gateway/src/modules/command-handler/commandHandler.ts +++ b/services/discord-gateway/src/modules/command-handler/commandHandler.ts @@ -32,9 +32,9 @@ export class CommandHandler { // connection for SUBSCRIBE mode — a subscribed connection cannot perform // publish/set operations. This connection listens on backend:command for // inbound requests from the backend. - this.redisSub = new Redis(config.REDIS_URL); // Dedicated Redis connection needed because: Redis requires a dedicated - // PUBLISH connection (cannot share with redisSub which is in SUBSCRIBE mode). - // Handles command reply publishing and voice/media status key updates. + this.redisSub = new Redis(config.REDIS_URL); + // A second dedicated connection for PUBLISH — a connection in SUBSCRIBE + // mode cannot publish, so replies go out on this one. this.redisPub = new Redis(config.REDIS_URL); this.redisSub.on("error", (err) => { @@ -85,6 +85,11 @@ export class CommandHandler { // ---- Command dispatch ---- + /** Normalize an unknown thrown value to a readable message. */ + private static errorMessage(err: unknown): string { + return err instanceof Error ? err.message : String(err); + } + private async handleCommand(raw: string): Promise { let cmd: CommandMessage; try { @@ -112,7 +117,7 @@ export class CommandHandler { }; } } catch (err) { - const message = err instanceof Error ? err.message : String(err); + const message = CommandHandler.errorMessage(err); logger.error( { commandId: cmd.id, error: message }, "Command execution failed",