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
2026-05-14 14:55:21 +07:00
parent 84e20ae373
commit 94a3acf12e
4 changed files with 114 additions and 91 deletions
@@ -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<void> {
-    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<DatabaseAdapter> {
    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);
  }
 }
@@ -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<typeof dbPromise>;
+
+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);
+});
@@ -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<DatabaseAdapter> {
+  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<DatabaseAdapter> {
+  if (!dbAdapter) {
+    dbAdapter = await initializeDatabase();
  }
-  return db;
+  return dbAdapter;
 }

-export { getDatabase };
+// Export as getDatabase for backward compatibility
+export const getDatabase = getDatabaseAdapterInternal;

-export function getPersistedValue<T>(key: string, fallback: T): T {
-  const row = getDatabase()
+export async function getPersistedValue<T>(key: string, fallback: T): Promise<T> {
+  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<T>(key: string, fallback: T): T {
  }
 }

-export function setPersistedValue(key: string, value: unknown): void {
-  getDatabase()
+export async function setPersistedValue(key: string, value: unknown): Promise<void> {
+  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<string> {
  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<string> {
 }

 export async function getPendingJobs(): Promise<StoredJob[]> {
-  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<void> {
-  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<boolean> {
-  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<boolean> {
    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<boolean> {
 export async function cleanupCompletedJobs(
  olderThanMs: number = 24 * 60 * 60 * 1000,
 ): Promise<number> {
-  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<void> {
-  if (db) {
-    db.close();
-    db = null;
+  if (dbAdapter) {
+    await dbAdapter.close();
+    dbAdapter = null;
    logger.info("Muxer queue closed");
  }
 }
@@ -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,