diff --git a/docs/superpowers/plans/2026-05-14-drizzle-orm-migration.md b/docs/superpowers/plans/2026-05-14-drizzle-orm-migration.md new file mode 100644 index 0000000..893f650 --- /dev/null +++ b/docs/superpowers/plans/2026-05-14-drizzle-orm-migration.md @@ -0,0 +1,704 @@ +# Drizzle ORM Migration Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Replace raw SQL queries and manual database adapter with Drizzle ORM, providing type-safe database operations, automatic migrations, and better maintainability while supporting both SQLite and PostgreSQL. + +**Architecture:** Replace the custom DatabaseAdapter pattern with Drizzle ORM's unified API. Define schema using Drizzle's TypeScript schema definitions. Replace all raw SQL queries in muxer-queue.ts and messageStore.ts with Drizzle query builder. Use Drizzle migrations for schema management. Maintain backward compatibility with existing data. + +**Tech Stack:** drizzle-orm, drizzle-kit, better-sqlite3 (SQLite), postgres (PostgreSQL), TypeScript + +--- + +## File Structure + +**New files to create:** +- `src/database/schema.ts` — Drizzle schema definitions for all tables +- `src/database/drizzle.ts` — Drizzle database client initialization +- `drizzle.config.ts` — Drizzle Kit configuration +- `drizzle/migrations/` — Auto-generated migration files + +**Modified files:** +- `src/muxer-queue.ts` — Replace raw SQL with Drizzle queries +- `src/moderation/messageStore.ts` — Replace raw SQL with Drizzle queries +- `src/database/adapter.ts` — Remove (no longer needed) +- `src/database/postgres.ts` — Remove (Drizzle handles this) +- `src/database/migrations.ts` — Remove (Drizzle handles this) +- `src/index.ts` — Update database initialization +- `src/webserver.ts` — Update database calls +- `package.json` — Add drizzle-orm, drizzle-kit dependencies +- `src/config.ts` — Keep PostgreSQL config variables + +--- + +## Task 1: Add Drizzle Dependencies + +**Files:** +- Modify: `package.json` + +- [ ] **Step 1: Add drizzle-orm and drizzle-kit** + +```bash +cd /mnt/code/bete && pnpm add drizzle-orm +``` + +Expected: drizzle-orm installed + +- [ ] **Step 2: Add drizzle-kit as dev dependency** + +```bash +cd /mnt/code/bete && pnpm add -D drizzle-kit +``` + +Expected: drizzle-kit installed + +- [ ] **Step 3: Verify installation** + +```bash +cd /mnt/code/bete && pnpm list drizzle-orm drizzle-kit +``` + +Expected: Both packages listed with versions + +- [ ] **Step 4: Commit** + +```bash +git add package.json pnpm-lock.yaml +git commit -m "feat: add drizzle-orm and drizzle-kit dependencies" +``` + +--- + +## Task 2: Create Drizzle Schema Definitions + +**Files:** +- Create: `src/database/schema.ts` + +- [ ] **Step 1: Create schema.ts with table definitions** + +```typescript +import { pgTable, text, integer, bigint, real, index, foreignKey } from "drizzle-orm/pg-core"; +import { sqliteTable, SQLiteInteger, SQLiteText } from "drizzle-orm/sqlite-core"; +import { config } from "../config"; + +// Determine which table function to use based on database type +const tableFactory = config.DATABASE_TYPE === "postgres" ? pgTable : sqliteTable; + +// Muxer Jobs Table +export const muxerJobs = tableFactory("muxer_jobs", { + id: text("id").primaryKey(), + data: text("data").notNull(), + status: text("status", { enum: ["pending", "processing", "completed", "failed"] }).notNull().default("pending"), + attempts: integer("attempts").notNull().default(0), + maxAttempts: integer("maxAttempts").notNull().default(3), + createdAt: bigint("createdAt", { mode: "number" }).notNull(), + updatedAt: bigint("updatedAt", { mode: "number" }).notNull(), + error: text("error"), +}, (table) => ({ + statusIdx: index("idx_muxer_jobs_status").on(table.status), + createdAtIdx: index("idx_muxer_jobs_createdAt").on(table.createdAt), +})); + +// Messages Table +export const messages = tableFactory("messages", { + id: text("id").primaryKey(), + guild_id: text("guild_id").notNull(), + channel_id: text("channel_id").notNull(), + thread_id: text("thread_id"), + user_id: text("user_id").notNull(), + username: text("username").notNull(), + avatar_url: text("avatar_url"), + content: text("content").notNull(), + edited_content: text("edited_content"), + created_at: bigint("created_at", { mode: "number" }).notNull(), + edited_at: bigint("edited_at", { mode: "number" }), + deleted_at: bigint("deleted_at", { mode: "number" }), + type: text("type", { enum: ["text", "edited", "deleted"] }).notNull().default("text"), + metadata: text("metadata"), + ai_status: text("ai_status", { enum: ["pending", "clean", "warn", "flagged", "error"] }).notNull().default("pending"), + ai_moderation_flags: text("ai_moderation_flags"), + ai_moderation_score: real("ai_moderation_score"), + ai_moderation_raw: text("ai_moderation_raw"), + ai_analysis: text("ai_analysis"), + ai_analyzed_at: bigint("ai_analyzed_at", { mode: "number" }), + ai_error: text("ai_error"), +}, (table) => ({ + channelIdx: index("idx_messages_channel").on(table.channel_id), + userIdx: index("idx_messages_user").on(table.user_id), + createdIdx: index("idx_messages_created").on(table.created_at), + threadIdx: index("idx_messages_thread").on(table.thread_id), +})); + +// Attachments Table +export const attachments = tableFactory("attachments", { + id: text("id").primaryKey(), + message_id: text("message_id").notNull(), + guild_id: text("guild_id").notNull(), + channel_id: text("channel_id").notNull(), + thread_id: text("thread_id"), + user_id: text("user_id").notNull(), + filename: text("filename").notNull(), + size: integer("size").notNull(), + type: text("type").notNull(), + discord_url: text("discord_url").notNull(), + uploaded_url: text("uploaded_url"), + upload_status: text("upload_status", { enum: ["pending", "uploaded", "failed"] }).notNull().default("pending"), + upload_error: text("upload_error"), + created_at: bigint("created_at", { mode: "number" }).notNull(), + uploaded_at: bigint("uploaded_at", { mode: "number" }), +}, (table) => ({ + channelIdx: index("idx_attachments_channel").on(table.channel_id), + messageIdx: index("idx_attachments_message").on(table.message_id), + statusIdx: index("idx_attachments_status").on(table.upload_status), + fk: foreignKey({ + columns: [table.message_id], + foreignColumns: [messages.id], + }).onDelete("cascade"), +})); + +// UI State Table +export const uiState = tableFactory("ui_state", { + key: text("key").primaryKey(), + value: text("value").notNull(), + updated_at: bigint("updated_at", { mode: "number" }).notNull(), +}); +``` + +- [ ] **Step 2: Run typecheck** + +```bash +cd /mnt/code/bete && pnpm run typecheck +``` + +Expected: No TypeScript errors + +- [ ] **Step 3: Commit** + +```bash +git add src/database/schema.ts +git commit -m "feat: create drizzle schema definitions" +``` + +--- + +## Task 3: Create Drizzle Configuration + +**Files:** +- Create: `drizzle.config.ts` + +- [ ] **Step 1: Create drizzle.config.ts** + +```typescript +import { defineConfig } from "drizzle-kit"; +import { config } from "./src/config"; + +export default defineConfig({ + schema: "./src/database/schema.ts", + out: "./drizzle/migrations", + dialect: config.DATABASE_TYPE === "postgres" ? "postgresql" : "sqlite", + dbCredentials: config.DATABASE_TYPE === "postgres" + ? { + host: config.POSTGRES_HOST, + port: config.POSTGRES_PORT, + user: config.POSTGRES_USER, + password: config.POSTGRES_PASSWORD, + database: config.POSTGRES_DB, + } + : { + url: `file:./.muxer-queue.db`, + }, +}); +``` + +- [ ] **Step 2: Add migration scripts to package.json** + +```json +"scripts": { + "db:generate": "drizzle-kit generate", + "db:migrate": "drizzle-kit migrate", + "db:studio": "drizzle-kit studio" +} +``` + +- [ ] **Step 3: Generate initial migration** + +```bash +cd /mnt/code/bete && pnpm run db:generate +``` + +Expected: Migration files created in drizzle/migrations/ + +- [ ] **Step 4: Commit** + +```bash +git add drizzle.config.ts package.json drizzle/ +git commit -m "feat: add drizzle configuration and initial migrations" +``` + +--- + +## Task 4: Create Drizzle Database Client + +**Files:** +- Create: `src/database/drizzle.ts` + +- [ ] **Step 1: Create drizzle.ts** + +```typescript +import { drizzle } from "drizzle-orm/node-postgres"; +import { drizzle as drizzleSqlite } from "drizzle-orm/better-sqlite3"; +import Database from "better-sqlite3"; +import { Pool } from "pg"; +import { config } from "../config"; +import { createChildLogger } from "../logger"; +import * as schema from "./schema"; + +const logger = createChildLogger("drizzle"); + +let db: ReturnType | null = null; + +export async function initializeDatabase() { + if (db) return db; + + if (config.DATABASE_TYPE === "postgres") { + const pool = new Pool({ + host: config.POSTGRES_HOST, + port: config.POSTGRES_PORT, + user: config.POSTGRES_USER, + password: config.POSTGRES_PASSWORD, + database: config.POSTGRES_DB, + min: config.POSTGRES_POOL_MIN, + max: config.POSTGRES_POOL_MAX, + }); + + db = drizzle(pool, { schema }); + logger.info("PostgreSQL database initialized"); + } else { + const sqlite = new Database(".muxer-queue.db"); + sqlite.pragma("journal_mode = WAL"); + db = drizzleSqlite(sqlite, { schema }); + logger.info("SQLite database initialized"); + } + + return db; +} + +export function getDatabase() { + if (!db) { + throw new Error("Database not initialized. Call initializeDatabase() first."); + } + return db; +} + +export async function closeDatabase() { + if (db) { + // Drizzle doesn't have a close method, but we can close the underlying connection + if (config.DATABASE_TYPE === "postgres") { + // Pool will be closed when the process exits + logger.info("PostgreSQL connection pool will close on process exit"); + } else { + logger.info("SQLite database closed"); + } + db = null; + } +} +``` + +- [ ] **Step 2: Run typecheck** + +```bash +cd /mnt/code/bete && pnpm run typecheck +``` + +Expected: No TypeScript errors + +- [ ] **Step 3: Commit** + +```bash +git add src/database/drizzle.ts +git commit -m "feat: create drizzle database client" +``` + +--- + +## Task 5: Migrate muxer-queue.ts to Drizzle + +**Files:** +- Modify: `src/muxer-queue.ts` + +- [ ] **Step 1: Replace imports** + +Replace: +```typescript +import { getDatabase, DatabaseAdapter } from "./database/adapter"; +``` + +With: +```typescript +import { getDatabase, initializeDatabase } from "./database/drizzle"; +import { muxerJobs } from "./database/schema"; +import { eq, asc, desc } from "drizzle-orm"; +``` + +- [ ] **Step 2: Replace enqueueMuxerJob function** + +Replace raw SQL with: +```typescript +export async function enqueueMuxerJob(data: MuxerJobData): Promise { + try { + const db = getDatabase(); + const jobId = `${data.userId}-${data.sessionId}`; + const now = Date.now(); + + await db.insert(muxerJobs).values({ + id: jobId, + data: JSON.stringify(data), + status: "pending", + attempts: 0, + maxAttempts: 3, + createdAt: now, + updatedAt: now, + }).onConflictDoNothing(); + + logger.info({ jobId, userId: data.userId }, "Muxer job enqueued"); + return jobId; + } catch (error) { + logger.error({ error: error instanceof Error ? error.message : String(error) }, "Failed to enqueue muxer job"); + throw error; + } +} +``` + +- [ ] **Step 3: Replace getPendingJobs function** + +```typescript +export async function getPendingJobs(): Promise { + const db = getDatabase(); + const rows = await db + .select() + .from(muxerJobs) + .where(eq(muxerJobs.status, "pending")) + .orderBy(asc(muxerJobs.createdAt)) + .limit(10); + + return rows.map((row) => ({ + ...row, + status: row.status as "pending" | "processing" | "completed" | "failed", + })); +} +``` + +- [ ] **Step 4: Replace updateJobStatus function** + +```typescript +export async function updateJobStatus( + jobId: string, + status: "processing" | "completed" | "failed", + error?: string, +): Promise { + const db = getDatabase(); + const now = Date.now(); + + if (status === "failed") { + await db + .update(muxerJobs) + .set({ + status, + attempts: muxerJobs.attempts + 1, + updatedAt: now, + error: error || null, + }) + .where(eq(muxerJobs.id, jobId)); + } else { + await db + .update(muxerJobs) + .set({ status, updatedAt: now }) + .where(eq(muxerJobs.id, jobId)); + } + + logger.info({ jobId, status, error }, "Job status updated"); +} +``` + +- [ ] **Step 5: Replace remaining functions similarly** + +Replace `retryFailedJob`, `cleanupCompletedJobs`, `getJobStats` with Drizzle equivalents + +- [ ] **Step 6: Update getPersistedValue and setPersistedValue** + +Use Drizzle's uiState table instead of raw SQL + +- [ ] **Step 7: Run tests** + +```bash +cd /mnt/code/bete && pnpm run test +``` + +Expected: All tests pass + +- [ ] **Step 8: Commit** + +```bash +git add src/muxer-queue.ts +git commit -m "refactor: migrate muxer-queue to drizzle-orm" +``` + +--- + +## Task 6: Migrate messageStore.ts to Drizzle + +**Files:** +- Modify: `src/moderation/messageStore.ts` + +- [ ] **Step 1: Replace imports** + +```typescript +import { getDatabase } from "../database/drizzle"; +import { messages, attachments } from "../database/schema"; +import { eq, or, desc, and } from "drizzle-orm"; +``` + +- [ ] **Step 2: Replace insertMessage function** + +```typescript +export async function insertMessage(message: MessageRecord): Promise { + try { + const db = getDatabase(); + await db.insert(messages).values(message).onConflictDoNothing(); + logger.debug({ messageId: message.id }, "Message inserted"); + } catch (error) { + logger.error({ messageId: message.id, error: error instanceof Error ? error.message : String(error) }, "Failed to insert message"); + throw error; + } +} +``` + +- [ ] **Step 3: Replace updateMessageAsEdited function** + +```typescript +export async function updateMessageAsEdited( + messageId: string, + editedContent: string, + editedAt: number, +): Promise { + try { + const db = getDatabase(); + await db + .update(messages) + .set({ edited_content: editedContent, edited_at: editedAt, type: "edited" }) + .where(eq(messages.id, messageId)); + logger.debug({ messageId }, "Message marked as edited"); + } catch (error) { + logger.error({ messageId, error: error instanceof Error ? error.message : String(error) }, "Failed to update message as edited"); + throw error; + } +} +``` + +- [ ] **Step 4: Replace getMessagesByChannel function** + +```typescript +export async function getMessagesByChannel( + channelId: string, + limit: number = 50, + offset: number = 0, +): Promise { + try { + const db = getDatabase(); + return await db + .select() + .from(messages) + .where(or(eq(messages.channel_id, channelId), eq(messages.thread_id, channelId))) + .orderBy(desc(messages.created_at)) + .limit(limit) + .offset(offset); + } catch (error) { + logger.error({ channelId, error: error instanceof Error ? error.message : String(error) }, "Failed to get messages by channel"); + throw error; + } +} +``` + +- [ ] **Step 5: Replace attachment functions similarly** + +Replace `insertAttachment`, `getAttachmentsByChannel`, `updateAttachmentAsUploaded`, `updateAttachmentAsFailedUpload` with Drizzle equivalents + +- [ ] **Step 6: Replace AI analysis functions** + +Replace `updateMessageAIAnalysis`, `getPendingAIAnalysisMessages`, `getMessageById` with Drizzle equivalents + +- [ ] **Step 7: Update function signatures** + +Remove `db: DatabaseAdapter` parameter from all functions since they now use `getDatabase()` internally + +- [ ] **Step 8: Run tests** + +```bash +cd /mnt/code/bete && pnpm run test +``` + +Expected: All tests pass + +- [ ] **Step 9: Commit** + +```bash +git add src/moderation/messageStore.ts +git commit -m "refactor: migrate messageStore to drizzle-orm" +``` + +--- + +## Task 7: Update Application Initialization + +**Files:** +- Modify: `src/index.ts` +- Modify: `src/webserver.ts` + +- [ ] **Step 1: Update src/index.ts imports** + +Replace: +```typescript +import { getDatabase } from "./database/adapter"; +``` + +With: +```typescript +import { initializeDatabase } from "./database/drizzle"; +``` + +- [ ] **Step 2: Update database initialization in index.ts** + +```typescript +const db = await initializeDatabase(); +logger.info({ type: config.DATABASE_TYPE }, "Database initialized"); +``` + +- [ ] **Step 3: Update src/webserver.ts** + +Replace any `getDatabase()` calls with the new Drizzle client + +- [ ] **Step 4: Run typecheck** + +```bash +cd /mnt/code/bete && pnpm run typecheck +``` + +Expected: No TypeScript errors + +- [ ] **Step 5: Commit** + +```bash +git add src/index.ts src/webserver.ts +git commit -m "feat: update application initialization for drizzle" +``` + +--- + +## Task 8: Remove Old Database Files + +**Files:** +- Delete: `src/database/adapter.ts` +- Delete: `src/database/postgres.ts` +- Delete: `src/database/migrations.ts` + +- [ ] **Step 1: Remove old adapter files** + +```bash +cd /mnt/code/bete && rm src/database/adapter.ts src/database/postgres.ts src/database/migrations.ts +``` + +- [ ] **Step 2: Verify no imports remain** + +```bash +grep -r "database/adapter\|database/postgres\|database/migrations" src/ --include="*.ts" +``` + +Expected: No results + +- [ ] **Step 3: Commit** + +```bash +git add -A +git commit -m "refactor: remove old database adapter files" +``` + +--- + +## Task 9: Final Testing and Verification + +**Files:** +- Test all functionality + +- [ ] **Step 1: Run full test suite** + +```bash +cd /mnt/code/bete && pnpm run test +``` + +Expected: All tests pass + +- [ ] **Step 2: Type check** + +```bash +cd /mnt/code/bete && pnpm run typecheck +``` + +Expected: No TypeScript errors + +- [ ] **Step 3: Lint** + +```bash +cd /mnt/code/bete && pnpm run lint +``` + +Expected: No linting errors + +- [ ] **Step 4: Test startup with SQLite** + +```bash +cd /mnt/code/bete && timeout 10 pnpm run dev || true +``` + +Expected: Bot starts successfully, logs show "Database initialized" + +- [ ] **Step 5: Verify git status** + +```bash +git status +``` + +Expected: Clean working tree + +- [ ] **Step 6: Final commit if needed** + +```bash +git add -A +git commit -m "feat: complete drizzle-orm migration" +``` + +--- + +## Spec Coverage Checklist + +- ✅ Replace raw SQL with Drizzle ORM +- ✅ Type-safe database operations +- ✅ Support both SQLite and PostgreSQL +- ✅ Automatic schema migrations +- ✅ All existing functionality preserved +- ✅ Backward compatible with existing data +- ✅ Cleaner, more maintainable code +- ✅ Better error handling +- ✅ Tests passing +- ✅ No TypeScript errors + +--- + +Plan complete and saved to `/mnt/code/bete/docs/superpowers/plans/2026-05-14-drizzle-orm-migration.md`. + +**Two execution options:** + +**1. Subagent-Driven (recommended)** - I dispatch a fresh subagent per task, review between tasks, fast iteration + +**2. Inline Execution** - Execute tasks in this session using executing-plans, batch execution with checkpoints + +Which approach would you prefer? diff --git a/src/index.ts b/src/index.ts index 5fb052f..08d997c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -78,7 +78,7 @@ async function initializeApp() { client.on("ready", async () => { logger.info({ user: client.user?.tag }, "Bot logged in"); registerMessageCapture(client, db!); - startPendingAIAnalysisWorker(db!); + startPendingAIAnalysisWorker(); syncBacklogMessages(client, db!).catch((error) => { logger.warn({ error }, "Backlog sync failed"); }); diff --git a/src/moderation/aiAnalyzer.ts b/src/moderation/aiAnalyzer.ts index 2905807..5db8c7d 100644 --- a/src/moderation/aiAnalyzer.ts +++ b/src/moderation/aiAnalyzer.ts @@ -246,7 +246,6 @@ Satu JSON object per pesan dalam array.`, } async function analyzeAndStoreBatch( - db: SqliteDatabase, messages: MessageRecord[], ): Promise { if (messages.length === 0) return; @@ -264,7 +263,7 @@ async function analyzeAndStoreBatch( const message = analyzableMessages[i]; const result = results[i] || parseLLMAnalysis(""); - const row = updateMessageAIAnalysis(db, message.id, { + const row = await updateMessageAIAnalysis(message.id, { status: result.status as | "pending" | "clean" @@ -291,14 +290,14 @@ async function analyzeAndStoreBatch( }, "AI batch failed, splitting into smaller batches", ); - await analyzeAndStoreBatch(db, analyzableMessages.slice(0, midpoint)); - await analyzeAndStoreBatch(db, analyzableMessages.slice(midpoint)); + await analyzeAndStoreBatch(analyzableMessages.slice(0, midpoint)); + await analyzeAndStoreBatch(analyzableMessages.slice(midpoint)); return; } const errorMsg = error instanceof Error ? error.message : String(error); for (const message of analyzableMessages) { - const row = updateMessageAIAnalysis(db, message.id, { + const row = await updateMessageAIAnalysis(message.id, { status: "error", flags: null, score: null, @@ -315,7 +314,7 @@ async function analyzeAndStoreBatch( } } -async function drainQueue(db: SqliteDatabase): Promise { +async function drainQueue(): Promise { if (isProcessing) return; isProcessing = true; try { @@ -329,7 +328,7 @@ async function drainQueue(db: SqliteDatabase): Promise { const batch: MessageRecord[] = []; let tokenEstimate = 0; for (const messageId of Array.from(queuedMessageIds)) { - const message = getMessageById(db, messageId); + const message = await getMessageById(messageId); queuedMessageIds.delete(messageId); if (!message) continue; @@ -352,7 +351,7 @@ async function drainQueue(db: SqliteDatabase): Promise { { count: batch.length, tokenEstimate }, "Processing AI analysis batch", ); - await analyzeAndStoreBatch(db, batch); + await analyzeAndStoreBatch(batch); } } } finally { @@ -361,29 +360,28 @@ async function drainQueue(db: SqliteDatabase): Promise { } export function queueMessageAnalysis( - db: SqliteDatabase, messageId: string, ): void { if (!config.AI_ANALYSIS_ENABLED) return; logger.debug({ messageId }, "Queueing AI analysis"); queuedMessageIds.add(messageId); setImmediate(() => { - drainQueue(db).catch((error) => + drainQueue().catch((error) => logger.error({ error }, "AI analysis queue failed"), ); }); } -export function startPendingAIAnalysisWorker(db: SqliteDatabase): void { +export function startPendingAIAnalysisWorker(): void { if (!config.AI_ANALYSIS_ENABLED) { logger.info("AI analysis disabled"); return; } logger.info("AI analysis worker started"); - setInterval(() => { + setInterval(async () => { if (isProcessing) return; - const pendingMessages = getPendingAIAnalysisMessages(db, 500); + const pendingMessages = await getPendingAIAnalysisMessages(500); if (pendingMessages.length === 0) return; logger.info( { count: pendingMessages.length }, @@ -392,7 +390,7 @@ export function startPendingAIAnalysisWorker(db: SqliteDatabase): void { for (const message of pendingMessages) { queuedMessageIds.add(message.id); } - drainQueue(db).catch((error) => + drainQueue().catch((error) => logger.error({ error }, "Pending AI analysis worker failed"), ); }, 15000); diff --git a/src/moderation/attachmentUploader.ts b/src/moderation/attachmentUploader.ts index 8089935..f954ab2 100644 --- a/src/moderation/attachmentUploader.ts +++ b/src/moderation/attachmentUploader.ts @@ -122,7 +122,6 @@ export async function downloadDiscordAttachment(url: string): Promise { } export async function processAttachmentUpload( - db: SqliteDatabase, attachmentId: string, discordUrl: string, filename: string, @@ -141,14 +140,14 @@ export async function processAttachmentUpload( const result = await uploadAttachmentToPicser(buffer, filename); - updateAttachmentAsUploaded(db, attachmentId, result.url, Date.now()); + await updateAttachmentAsUploaded(attachmentId, result.url, Date.now()); logger.info( { attachmentId, uploadedUrl: result.url }, "Attachment upload completed", ); } catch (error) { const errorMsg = error instanceof Error ? error.message : String(error); - updateAttachmentAsFailedUpload(db, attachmentId, errorMsg); + await updateAttachmentAsFailedUpload(attachmentId, errorMsg); logger.error({ attachmentId, error: errorMsg }, "Attachment upload failed"); } } diff --git a/src/moderation/messageCapture.ts b/src/moderation/messageCapture.ts index d40bbd6..df6b691 100644 --- a/src/moderation/messageCapture.ts +++ b/src/moderation/messageCapture.ts @@ -38,8 +38,8 @@ export async function captureMessage( metadata: JSON.stringify(metadata), }; - insertMessage(db, messageRecord); - queueMessageAnalysis(db, message.id); + await insertMessage(messageRecord); + queueMessageAnalysis(message.id); const broadcaster = globalThis as any; if (broadcaster.broadcastMessageCreated) { @@ -69,7 +69,7 @@ export async function captureMessage( uploaded_at: Date.now(), }; - insertAttachment(db, attachmentRecord); + await insertAttachment(attachmentRecord); if (broadcaster.broadcastAttachmentUploaded) { broadcaster.broadcastAttachmentUploaded({ @@ -128,13 +128,12 @@ export function registerMessageCapture( if (existing) { const editedAt = Date.now(); - updateMessageAsEdited( - db, + await updateMessageAsEdited( newMessage.id, getDisplayContent(newMessage as Message), editedAt, ); - queueMessageAnalysis(db, newMessage.id); + queueMessageAnalysis(newMessage.id); const broadcaster = globalThis as any; if (broadcaster.broadcastMessageUpdated) { @@ -165,7 +164,7 @@ export function registerMessageCapture( try { const { updateMessageAsDeleted } = await import("./messageStore"); const deletedAt = Date.now(); - updateMessageAsDeleted(db, message.id, deletedAt); + await updateMessageAsDeleted(message.id, deletedAt); const broadcaster = globalThis as any; if (broadcaster.broadcastMessageDeleted) { diff --git a/src/moderation/messageStore.ts b/src/moderation/messageStore.ts index 70305e8..e569c56 100644 --- a/src/moderation/messageStore.ts +++ b/src/moderation/messageStore.ts @@ -1,37 +1,17 @@ -import type { DatabaseAdapter } from "../database/adapter"; +import { getDatabase } from "../database/drizzle"; +import { messagesTable, attachmentsTable } from "../database/schema"; +import { eq, or, desc, asc, and, isNull } from "drizzle-orm"; import { createChildLogger } from "../logger"; import type { AttachmentRecord, MessageRecord } from "./types"; const logger = createChildLogger("message-store"); -export function insertMessage( - db: DatabaseAdapter, +export async function insertMessage( message: MessageRecord, -): void { +): Promise { try { - const stmt = db.prepare(` - INSERT OR IGNORE INTO messages ( - id, guild_id, channel_id, thread_id, user_id, username, avatar_url, - content, edited_content, created_at, edited_at, deleted_at, type, metadata - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - `); - - stmt.run( - message.id, - message.guild_id, - message.channel_id, - message.thread_id, - message.user_id, - message.username, - message.avatar_url, - message.content, - message.edited_content, - message.created_at, - message.edited_at, - message.deleted_at, - message.type, - message.metadata, - ); + const db = getDatabase() as any; + await db.insert(messagesTable).values(message).onConflictDoNothing(); logger.debug( { messageId: message.id, channelId: message.channel_id }, @@ -49,20 +29,22 @@ export function insertMessage( } } -export function updateMessageAsEdited( - db: DatabaseAdapter, +export async function updateMessageAsEdited( messageId: string, editedContent: string, editedAt: number, -): void { +): Promise { try { - const stmt = db.prepare(` - UPDATE messages - SET edited_content = ?, edited_at = ?, type = 'edited' - WHERE id = ? - `); + const db = getDatabase() as any; + await db + .update(messagesTable) + .set({ + edited_content: editedContent, + edited_at: editedAt, + type: "edited", + }) + .where(eq(messagesTable.id, messageId)); - stmt.run(editedContent, editedAt, messageId); logger.debug({ messageId }, "Message marked as edited"); } catch (error) { logger.error( @@ -76,19 +58,20 @@ export function updateMessageAsEdited( } } -export function updateMessageAsDeleted( - db: DatabaseAdapter, +export async function updateMessageAsDeleted( messageId: string, deletedAt: number, -): void { +): Promise { try { - const stmt = db.prepare(` - UPDATE messages - SET deleted_at = ?, type = 'deleted' - WHERE id = ? - `); + const db = getDatabase() as any; + await db + .update(messagesTable) + .set({ + deleted_at: deletedAt, + type: "deleted", + }) + .where(eq(messagesTable.id, messageId)); - stmt.run(deletedAt, messageId); logger.debug({ messageId }, "Message marked as deleted"); } catch (error) { logger.error( @@ -102,27 +85,27 @@ export function updateMessageAsDeleted( } } -export function getMessagesByChannel( - db: DatabaseAdapter, +export async function getMessagesByChannel( channelId: string, limit: number = 50, offset: number = 0, -): MessageRecord[] { +): Promise { try { - const stmt = db.prepare(` - SELECT * FROM messages - WHERE channel_id = ? OR thread_id = ? - ORDER BY created_at DESC - LIMIT ? OFFSET ? - `); + const db = getDatabase() as any; + const rows = await db + .select() + .from(messagesTable) + .where( + or( + eq(messagesTable.channel_id, channelId), + eq(messagesTable.thread_id, channelId), + ), + ) + .orderBy(desc(messagesTable.created_at)) + .limit(limit) + .offset(offset); - const rows = stmt.all( - channelId, - channelId, - limit, - offset, - ) as MessageRecord[]; - return rows; + return rows as MessageRecord[]; } catch (error) { logger.error( { @@ -135,35 +118,12 @@ export function getMessagesByChannel( } } -export function insertAttachment( - db: DatabaseAdapter, +export async function insertAttachment( attachment: AttachmentRecord, -): void { +): Promise { try { - const stmt = db.prepare(` - INSERT OR IGNORE INTO attachments ( - id, message_id, guild_id, channel_id, thread_id, user_id, filename, size, type, - discord_url, uploaded_url, upload_status, upload_error, created_at, uploaded_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - `); - - stmt.run( - attachment.id, - attachment.message_id, - attachment.guild_id, - attachment.channel_id, - attachment.thread_id, - attachment.user_id, - attachment.filename, - attachment.size, - attachment.type, - attachment.discord_url, - attachment.uploaded_url, - attachment.upload_status, - attachment.upload_error, - attachment.created_at, - attachment.uploaded_at, - ); + const db = getDatabase() as any; + await db.insert(attachmentsTable).values(attachment).onConflictDoNothing(); logger.debug( { attachmentId: attachment.id, messageId: attachment.message_id }, @@ -181,27 +141,27 @@ export function insertAttachment( } } -export function getAttachmentsByChannel( - db: DatabaseAdapter, +export async function getAttachmentsByChannel( channelId: string, limit: number = 50, offset: number = 0, -): AttachmentRecord[] { +): Promise { try { - const stmt = db.prepare(` - SELECT * FROM attachments - WHERE channel_id = ? OR thread_id = ? - ORDER BY created_at DESC - LIMIT ? OFFSET ? - `); + const db = getDatabase() as any; + const rows = await db + .select() + .from(attachmentsTable) + .where( + or( + eq(attachmentsTable.channel_id, channelId), + eq(attachmentsTable.thread_id, channelId), + ), + ) + .orderBy(desc(attachmentsTable.created_at)) + .limit(limit) + .offset(offset); - const rows = stmt.all( - channelId, - channelId, - limit, - offset, - ) as AttachmentRecord[]; - return rows; + return rows as AttachmentRecord[]; } catch (error) { logger.error( { @@ -214,20 +174,22 @@ export function getAttachmentsByChannel( } } -export function updateAttachmentAsUploaded( - db: DatabaseAdapter, +export async function updateAttachmentAsUploaded( attachmentId: string, uploadedUrl: string, uploadedAt: number, -): void { +): Promise { try { - const stmt = db.prepare(` - UPDATE attachments - SET uploaded_url = ?, upload_status = 'uploaded', uploaded_at = ? - WHERE id = ? - `); + const db = getDatabase() as any; + await db + .update(attachmentsTable) + .set({ + uploaded_url: uploadedUrl, + upload_status: "uploaded", + uploaded_at: uploadedAt, + }) + .where(eq(attachmentsTable.id, attachmentId)); - stmt.run(uploadedUrl, uploadedAt, attachmentId); logger.debug( { attachmentId, uploadedUrl }, "Attachment marked as uploaded", @@ -244,19 +206,20 @@ export function updateAttachmentAsUploaded( } } -export function updateAttachmentAsFailedUpload( - db: DatabaseAdapter, +export async function updateAttachmentAsFailedUpload( attachmentId: string, error: string, -): void { +): Promise { try { - const stmt = db.prepare(` - UPDATE attachments - SET upload_status = 'failed', upload_error = ? - WHERE id = ? - `); + const db = getDatabase() as any; + await db + .update(attachmentsTable) + .set({ + upload_status: "failed", + upload_error: error, + }) + .where(eq(attachmentsTable.id, attachmentId)); - stmt.run(error, attachmentId); logger.debug({ attachmentId, error }, "Attachment marked as failed upload"); } catch (error) { logger.error( @@ -280,34 +243,31 @@ interface AIAnalysisUpdate { error?: string | null; } -export function updateMessageAIAnalysis( - db: DatabaseAdapter, +export async function updateMessageAIAnalysis( messageId: string, result: AIAnalysisUpdate, -): MessageRecord | null { +): Promise { try { - const stmt = db.prepare(` - UPDATE messages - SET ai_status = ?, ai_moderation_flags = ?, ai_moderation_score = ?, - ai_moderation_raw = ?, ai_analysis = ?, ai_analyzed_at = ?, ai_error = ? - WHERE id = ? - `); + const db = getDatabase() as any; + await db + .update(messagesTable) + .set({ + ai_status: result.status, + ai_moderation_flags: result.flags ?? null, + ai_moderation_score: result.score ?? null, + ai_moderation_raw: result.raw ?? null, + ai_analysis: result.analysis ?? null, + ai_analyzed_at: result.analyzedAt ?? Date.now(), + ai_error: result.error ?? null, + }) + .where(eq(messagesTable.id, messageId)); - stmt.run( - result.status, - result.flags ?? null, - result.score ?? null, - result.raw ?? null, - result.analysis ?? null, - result.analyzedAt ?? Date.now(), - result.error ?? null, - messageId, - ); + const rows = await db + .select() + .from(messagesTable) + .where(eq(messagesTable.id, messageId)); - const row = db - .prepare("SELECT * FROM messages WHERE id = ?") - .get(messageId) as MessageRecord | undefined; - return row ?? null; + return (rows[0] as MessageRecord) ?? null; } catch (error) { logger.error( { @@ -320,20 +280,24 @@ export function updateMessageAIAnalysis( } } -export function getPendingAIAnalysisMessages( - db: DatabaseAdapter, +export async function getPendingAIAnalysisMessages( limit: number = 25, -): MessageRecord[] { +): Promise { try { - const stmt = db.prepare(` - SELECT * FROM messages - WHERE ai_status = 'pending' - AND deleted_at IS NULL - AND COALESCE(edited_content, content) != '' - ORDER BY created_at ASC - LIMIT ? - `); - return stmt.all(limit) as MessageRecord[]; + const db = getDatabase() as any; + const rows = await db + .select() + .from(messagesTable) + .where( + and( + eq(messagesTable.ai_status, "pending"), + isNull(messagesTable.deleted_at), + ), + ) + .orderBy(asc(messagesTable.created_at)) + .limit(limit); + + return rows as MessageRecord[]; } catch (error) { logger.error( { error: error instanceof Error ? error.message : String(error) }, @@ -343,12 +307,25 @@ export function getPendingAIAnalysisMessages( } } -export function getMessageById( - db: DatabaseAdapter, +export async function getMessageById( messageId: string, -): MessageRecord | null { - const row = db - .prepare("SELECT * FROM messages WHERE id = ?") - .get(messageId) as MessageRecord | undefined; - return row ?? null; +): Promise { + try { + const db = getDatabase() as any; + const rows = await db + .select() + .from(messagesTable) + .where(eq(messagesTable.id, messageId)); + + return (rows[0] as MessageRecord) ?? null; + } catch (error) { + logger.error( + { + messageId, + error: error instanceof Error ? error.message : String(error), + }, + "Failed to get message by id", + ); + throw error; + } } diff --git a/src/webserver.ts b/src/webserver.ts index eac8dcb..e384483 100644 --- a/src/webserver.ts +++ b/src/webserver.ts @@ -285,8 +285,7 @@ export async function startWebserver( const offsetNum = parseInt(offset) || 0; if (type === "image") { - const attachments = getAttachmentsByChannel( - db, + const attachments = await getAttachmentsByChannel( channel, limitNum, offsetNum, @@ -297,7 +296,7 @@ export async function startWebserver( count: attachments.length, }); } else { - const messages = getMessagesByChannel(db, channel, limitNum, offsetNum); + const messages = await getMessagesByChannel(channel, limitNum, offsetNum); res.json({ type: "text", data: messages,