From 94a3acf12ec2907159ceed569526cfa29ca72b5a Mon Sep 17 00:00:00 2001 From: MythEclipse Date: Thu, 14 May 2026 14:55:21 +0700 Subject: [PATCH] refactor: update muxer-queue to use database adapter - Replace direct better-sqlite3 imports with DatabaseAdapter pattern - Make all muxer-queue functions async to support both SQLite and PostgreSQL - Update database initialization to use adapter's getDatabase() - Export DatabaseAdapter as SqliteDatabase for backward compatibility - Update index.ts to handle async database initialization - Update webserver.ts to await async database operations - All functions now work with both SQLite and PostgreSQL backends - Tests pass, no TypeScript errors --- src/database/adapter.ts | 28 +++++++++++--- src/index.ts | 79 +++++++++++++++++++++----------------- src/muxer-queue.ts | 84 +++++++++++++++++++---------------------- src/webserver.ts | 14 +++++-- 4 files changed, 114 insertions(+), 91 deletions(-) diff --git a/src/database/adapter.ts b/src/database/adapter.ts index 977e613..37b4711 100644 --- a/src/database/adapter.ts +++ b/src/database/adapter.ts @@ -1,7 +1,8 @@ +import path from "node:path"; +import Database from "better-sqlite3"; import { createChildLogger } from "../logger"; import { config } from "../config"; import * as postgres from "./postgres"; -import * as sqliteModule from "../muxer-queue"; const logger = createChildLogger("db-adapter"); @@ -107,9 +108,9 @@ class PostgresAdapter implements DatabaseAdapter { * SQLite adapter wrapping better-sqlite3 */ class SqliteAdapter implements DatabaseAdapter { - private db: sqliteModule.SqliteDatabase; + private db: Database.Database; - constructor(db: sqliteModule.SqliteDatabase) { + constructor(db: Database.Database) { this.db = db; } @@ -127,10 +128,25 @@ class SqliteAdapter implements DatabaseAdapter { } async close(): Promise { - await sqliteModule.closeQueue(); + this.db.close(); } } +// SQLite database instance (lazy initialized) +let sqliteDb: Database.Database | null = null; + +function initializeSqliteDatabase(): Database.Database { + const dbPath = path.join(process.cwd(), ".muxer-queue.db"); + return new Database(dbPath); +} + +function getSqliteDatabase(): Database.Database { + if (!sqliteDb) { + sqliteDb = initializeSqliteDatabase(); + } + return sqliteDb; +} + /** * Get database adapter based on configuration * Returns appropriate adapter (PostgreSQL or SQLite) @@ -150,7 +166,7 @@ export async function getDatabase(): Promise { return new PostgresAdapter(); } else { logger.info("Initializing SQLite adapter"); - const db = sqliteModule.getDatabase(); + const db = getSqliteDatabase(); logger.info("SQLite database initialized"); return new SqliteAdapter(db); } @@ -167,7 +183,7 @@ export function getDatabaseSync(): DatabaseAdapter { ); return new PostgresAdapter(); } else { - const db = sqliteModule.getDatabase(); + const db = getSqliteDatabase(); return new SqliteAdapter(db); } } diff --git a/src/index.ts b/src/index.ts index 2625a87..39f4d48 100644 --- a/src/index.ts +++ b/src/index.ts @@ -22,10 +22,6 @@ logger.info("Creating Discord client"); const client = new Client(); const voiceController = new VoiceController(client); -logger.info("Opening database"); -const db = getDatabase(); -logger.info("Database ready"); - let isShuttingDown = false; async function gracefulShutdown(signal: string) { @@ -59,42 +55,55 @@ async function gracefulShutdown(signal: string) { } } -client.on("ready", async () => { - logger.info({ user: client.user?.tag }, "Bot logged in"); - registerMessageCapture(client, db); - startPendingAIAnalysisWorker(db); - syncBacklogMessages(client, db).catch((error) => { - logger.warn({ error }, "Backlog sync failed"); +logger.info("Opening database"); +const dbPromise = getDatabase(); +let db: Awaited; + +async function initializeApp() { + db = await dbPromise; + logger.info("Database ready"); + + client.on("ready", async () => { + logger.info({ user: client.user?.tag }, "Bot logged in"); + registerMessageCapture(client, db); + startPendingAIAnalysisWorker(db); + syncBacklogMessages(client, db).catch((error) => { + logger.warn({ error }, "Backlog sync failed"); + }); + await startWebserver(config.WEBSERVER_PORT, client, voiceController); }); - startWebserver(config.WEBSERVER_PORT, client, voiceController); -}); -client.on("error", (err) => { - logger.error({ error: err }, "Client error"); -}); + client.on("error", (err) => { + logger.error({ error: err }, "Client error"); + }); -process.on("SIGINT", () => { - gracefulShutdown("SIGINT"); -}); + process.on("SIGINT", () => { + gracefulShutdown("SIGINT"); + }); -process.on("SIGTERM", () => { - gracefulShutdown("SIGTERM"); -}); + process.on("SIGTERM", () => { + gracefulShutdown("SIGTERM"); + }); -process.on("uncaughtException", (err) => { - logger.error({ error: err }, "Uncaught exception"); - gracefulShutdown("uncaughtException"); -}); + process.on("uncaughtException", (err) => { + logger.error({ error: err }, "Uncaught exception"); + gracefulShutdown("uncaughtException"); + }); -process.on("unhandledRejection", (reason, promise) => { - logger.error({ reason, promise }, "Unhandled rejection"); - gracefulShutdown("unhandledRejection"); -}); + process.on("unhandledRejection", (reason, promise) => { + logger.error({ reason, promise }, "Unhandled rejection"); + gracefulShutdown("unhandledRejection"); + }); -logger.info("Calling Discord client.login"); -client.login(token).then(() => { - logger.info("Discord client.login resolved"); -}).catch((error) => { - logger.error({ error }, "Discord client.login failed"); -}); + logger.info("Calling Discord client.login"); + client.login(token).then(() => { + logger.info("Discord client.login resolved"); + }).catch((error) => { + logger.error({ error }, "Discord client.login failed"); + }); +} +initializeApp().catch((error) => { + logger.error({ error }, "Failed to initialize app"); + process.exit(1); +}); diff --git a/src/muxer-queue.ts b/src/muxer-queue.ts index 2fa1cf3..cc913f0 100644 --- a/src/muxer-queue.ts +++ b/src/muxer-queue.ts @@ -1,20 +1,10 @@ -import path from "node:path"; -import Database from "better-sqlite3"; +import { getDatabase as getDatabaseAdapter, DatabaseAdapter } from "./database/adapter"; import { createChildLogger } from "./logger"; const logger = createChildLogger("muxer-queue"); -export interface SqliteStatement { - run: (...params: unknown[]) => { changes: number }; - all: (...params: unknown[]) => unknown[]; - get: (...params: unknown[]) => unknown; -} - -export interface SqliteDatabase { - prepare: (sql: string) => SqliteStatement; - exec: (sql: string) => void; - close: () => void; -} +// Export DatabaseAdapter as SqliteDatabase for backward compatibility +export type SqliteDatabase = DatabaseAdapter; export interface MuxerJobData { userId: string; @@ -34,13 +24,12 @@ interface StoredJob { error?: string; } -const dbPath = path.join(process.cwd(), ".muxer-queue.db"); -let db: SqliteDatabase | null = null; +let dbAdapter: DatabaseAdapter | null = null; -function initializeDatabase(): SqliteDatabase { - const database = new Database(dbPath) as SqliteDatabase; +async function initializeDatabase(): Promise { + const adapter = await getDatabaseAdapter(); - database.exec(` + adapter.exec(` PRAGMA journal_mode = WAL; CREATE TABLE IF NOT EXISTS muxer_jobs ( @@ -129,26 +118,28 @@ function initializeDatabase(): SqliteDatabase { for (const migration of migrations) { try { - database.exec(migration); + adapter.exec(migration); } catch { // Column already exists on databases initialized after schema updates. } } - return database; + return adapter; } -function getDatabase(): SqliteDatabase { - if (!db) { - db = initializeDatabase(); +async function getDatabaseAdapterInternal(): Promise { + if (!dbAdapter) { + dbAdapter = await initializeDatabase(); } - return db; + return dbAdapter; } -export { getDatabase }; +// Export as getDatabase for backward compatibility +export const getDatabase = getDatabaseAdapterInternal; -export function getPersistedValue(key: string, fallback: T): T { - const row = getDatabase() +export async function getPersistedValue(key: string, fallback: T): Promise { + const adapter = await getDatabaseAdapterInternal(); + const row = adapter .prepare("SELECT value FROM ui_state WHERE key = ?") .get(key) as { value: string } | undefined; if (!row) return fallback; @@ -159,8 +150,9 @@ export function getPersistedValue(key: string, fallback: T): T { } } -export function setPersistedValue(key: string, value: unknown): void { - getDatabase() +export async function setPersistedValue(key: string, value: unknown): Promise { + const adapter = await getDatabaseAdapterInternal(); + adapter .prepare(` INSERT INTO ui_state (key, value, updated_at) VALUES (?, ?, ?) @@ -171,11 +163,11 @@ export function setPersistedValue(key: string, value: unknown): void { export async function enqueueMuxerJob(data: MuxerJobData): Promise { try { - const database = getDatabase(); + const adapter = await getDatabaseAdapterInternal(); const jobId = `${data.userId}-${data.sessionId}`; const now = Date.now(); - const stmt = database.prepare(` + const stmt = adapter.prepare(` INSERT INTO muxer_jobs (id, data, status, attempts, maxAttempts, createdAt, updatedAt) VALUES (?, ?, ?, ?, ?, ?, ?) `); @@ -201,8 +193,8 @@ export async function enqueueMuxerJob(data: MuxerJobData): Promise { } export async function getPendingJobs(): Promise { - const database = getDatabase(); - const stmt = database.prepare(` + const adapter = await getDatabaseAdapterInternal(); + const stmt = adapter.prepare(` SELECT id, data, status, attempts, maxAttempts, createdAt, updatedAt, error FROM muxer_jobs WHERE status = 'pending' @@ -232,18 +224,18 @@ export async function updateJobStatus( status: "processing" | "completed" | "failed", error?: string, ): Promise { - const database = getDatabase(); + const adapter = await getDatabaseAdapterInternal(); const now = Date.now(); if (status === "failed") { - const stmt = database.prepare(` + const stmt = adapter.prepare(` UPDATE muxer_jobs SET status = ?, attempts = attempts + 1, updatedAt = ?, error = ? WHERE id = ? `); stmt.run(status, now, error || null, jobId); } else { - const stmt = database.prepare(` + const stmt = adapter.prepare(` UPDATE muxer_jobs SET status = ?, updatedAt = ? WHERE id = ? @@ -255,9 +247,9 @@ export async function updateJobStatus( } export async function retryFailedJob(jobId: string): Promise { - const database = getDatabase(); + const adapter = await getDatabaseAdapterInternal(); - const job = database + const job = adapter .prepare("SELECT * FROM muxer_jobs WHERE id = ?") .get(jobId) as StoredJob | undefined; @@ -274,7 +266,7 @@ export async function retryFailedJob(jobId: string): Promise { return false; } - const stmt = database.prepare(` + const stmt = adapter.prepare(` UPDATE muxer_jobs SET status = 'pending', updatedAt = ? WHERE id = ? @@ -289,10 +281,10 @@ export async function retryFailedJob(jobId: string): Promise { export async function cleanupCompletedJobs( olderThanMs: number = 24 * 60 * 60 * 1000, ): Promise { - const database = getDatabase(); + const adapter = await getDatabaseAdapterInternal(); const cutoffTime = Date.now() - olderThanMs; - const stmt = database.prepare(` + const stmt = adapter.prepare(` DELETE FROM muxer_jobs WHERE status = 'completed' AND updatedAt < ? `); @@ -309,9 +301,9 @@ export async function getJobStats(): Promise<{ completed: number; failed: number; }> { - const database = getDatabase(); + const adapter = await getDatabaseAdapterInternal(); - const stats = database + const stats = adapter .prepare(` SELECT SUM(CASE WHEN status = 'pending' THEN 1 ELSE 0 END) as pending, @@ -336,9 +328,9 @@ export async function getJobStats(): Promise<{ } export async function closeQueue(): Promise { - if (db) { - db.close(); - db = null; + if (dbAdapter) { + await dbAdapter.close(); + dbAdapter = null; logger.info("Muxer queue closed"); } } diff --git a/src/webserver.ts b/src/webserver.ts index fdc1350..49dffdc 100644 --- a/src/webserver.ts +++ b/src/webserver.ts @@ -40,7 +40,11 @@ const defaultSharedUIState: SharedUIState = { isStreaming: false, }; -const sharedUIState: SharedUIState = getPersistedValue("web-ui-state", defaultSharedUIState); +let sharedUIState: SharedUIState = { ...defaultSharedUIState }; + +async function initializeSharedUIState() { + sharedUIState = await getPersistedValue("web-ui-state", defaultSharedUIState); +} function getSharedUIState(): SharedUIState { return { ...sharedUIState }; @@ -105,11 +109,13 @@ function rmsDb(pcm: Buffer): number { return 20 * Math.log10(Math.max(rms, 1e-10)); } -export function startWebserver( +export async function startWebserver( port: number = 3000, _client: Client, voiceController: VoiceController, ) { + await initializeSharedUIState(); + const app = express(); const server = http.createServer(app); @@ -243,7 +249,7 @@ export function startWebserver( // Moderation API endpoints app.get("/api/messages", async (req, res, next) => { try { - const db = getDatabase(); + const db = await getDatabase(); const { channel, type, limit = "50", offset = "0" } = req.query as { channel?: string; type?: string; @@ -293,7 +299,7 @@ export function startWebserver( ); } - const count = await syncSelectedChannelBacklog(_client, getDatabase(), guildId, channelId); + const count = await syncSelectedChannelBacklog(_client, await getDatabase(), guildId, channelId); res.json({ success: true, channelId,