feat(storage): implement SQLite adapter with better-sqlite3

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-04-02 20:51:13 -05:00
parent 7bb878718d
commit 25383ea645
5 changed files with 768 additions and 15 deletions
--- a/packages/storage/src/adapters/sqlite.ts
+++ b/packages/storage/src/adapters/sqlite.ts
@@ -0,0 +1,283 @@
+import Database from 'better-sqlite3';
+import { randomUUID } from 'node:crypto';
+import type { StorageAdapter, StorageConfig } from '../types.js';
+
+/* eslint-disable @typescript-eslint/no-explicit-any */
+
+const COLLECTIONS = [
+  'users',
+  'sessions',
+  'accounts',
+  'projects',
+  'missions',
+  'tasks',
+  'agents',
+  'conversations',
+  'messages',
+  'preferences',
+  'insights',
+  'skills',
+  'events',
+  'routing_rules',
+  'provider_credentials',
+  'agent_logs',
+  'teams',
+  'team_members',
+  'mission_tasks',
+  'tickets',
+  'summarization_jobs',
+  'appreciations',
+  'verifications',
+] as const;
+
+function buildFilterClause(filter?: Record<string, unknown>): {
+  clause: string;
+  params: unknown[];
+} {
+  if (!filter || Object.keys(filter).length === 0) return { clause: '', params: [] };
+  const conditions: string[] = [];
+  const params: unknown[] = [];
+  for (const [key, value] of Object.entries(filter)) {
+    if (key === 'id') {
+      conditions.push('id = ?');
+      params.push(value);
+    } else {
+      conditions.push(`json_extract(data_json, '$.${key}') = ?`);
+      params.push(typeof value === 'object' ? JSON.stringify(value) : value);
+    }
+  }
+  return { clause: ` WHERE ${conditions.join(' AND ')}`, params };
+}
+
+export class SqliteAdapter implements StorageAdapter {
+  readonly name = 'sqlite';
+  private db: Database.Database;
+
+  constructor(config: Extract<StorageConfig, { type: 'sqlite' }>) {
+    this.db = new Database(config.path);
+    this.db.pragma('journal_mode = WAL');
+    this.db.pragma('foreign_keys = ON');
+  }
+
+  async create<T extends Record<string, unknown>>(
+    collection: string,
+    data: T,
+  ): Promise<T & { id: string }> {
+    const id = (data as any).id ?? randomUUID();
+    const now = new Date().toISOString();
+    const rest = Object.fromEntries(Object.entries(data).filter(([k]) => k !== 'id'));
+    this.db
+      .prepare(
+        `INSERT INTO ${collection} (id, data_json, created_at, updated_at) VALUES (?, ?, ?, ?)`,
+      )
+      .run(id, JSON.stringify(rest), now, now);
+    return { ...data, id } as T & { id: string };
+  }
+
+  async read<T extends Record<string, unknown>>(collection: string, id: string): Promise<T | null> {
+    const row = this.db.prepare(`SELECT * FROM ${collection} WHERE id = ?`).get(id) as any;
+    if (!row) return null;
+    return { id: row.id, ...JSON.parse(row.data_json as string) } as T;
+  }
+
+  async update(collection: string, id: string, data: Record<string, unknown>): Promise<boolean> {
+    const existing = this.db
+      .prepare(`SELECT data_json FROM ${collection} WHERE id = ?`)
+      .get(id) as any;
+    if (!existing) return false;
+    const merged = { ...JSON.parse(existing.data_json as string), ...data };
+    const now = new Date().toISOString();
+    const result = this.db
+      .prepare(`UPDATE ${collection} SET data_json = ?, updated_at = ? WHERE id = ?`)
+      .run(JSON.stringify(merged), now, id);
+    return result.changes > 0;
+  }
+
+  async delete(collection: string, id: string): Promise<boolean> {
+    const result = this.db.prepare(`DELETE FROM ${collection} WHERE id = ?`).run(id);
+    return result.changes > 0;
+  }
+
+  async find<T extends Record<string, unknown>>(
+    collection: string,
+    filter?: Record<string, unknown>,
+    opts?: { limit?: number; offset?: number; orderBy?: string; order?: 'asc' | 'desc' },
+  ): Promise<T[]> {
+    const { clause, params } = buildFilterClause(filter);
+    let query = `SELECT * FROM ${collection}${clause}`;
+    if (opts?.orderBy) {
+      const dir = opts.order === 'desc' ? 'DESC' : 'ASC';
+      const col =
+        opts.orderBy === 'id' || opts.orderBy === 'created_at' || opts.orderBy === 'updated_at'
+          ? opts.orderBy
+          : `json_extract(data_json, '$.${opts.orderBy}')`;
+      query += ` ORDER BY ${col} ${dir}`;
+    }
+    if (opts?.limit) {
+      query += ` LIMIT ?`;
+      params.push(opts.limit);
+    }
+    if (opts?.offset) {
+      query += ` OFFSET ?`;
+      params.push(opts.offset);
+    }
+    const rows = this.db.prepare(query).all(...params) as any[];
+    return rows.map((row) => ({ id: row.id, ...JSON.parse(row.data_json as string) }) as T);
+  }
+
+  async findOne<T extends Record<string, unknown>>(
+    collection: string,
+    filter: Record<string, unknown>,
+  ): Promise<T | null> {
+    const results = await this.find<T>(collection, filter, { limit: 1 });
+    return results[0] ?? null;
+  }
+
+  async count(collection: string, filter?: Record<string, unknown>): Promise<number> {
+    const { clause, params } = buildFilterClause(filter);
+    const row = this.db
+      .prepare(`SELECT COUNT(*) as count FROM ${collection}${clause}`)
+      .get(...params) as any;
+    return row?.count ?? 0;
+  }
+
+  async transaction<T>(fn: (tx: StorageAdapter) => Promise<T>): Promise<T> {
+    const txAdapter = new SqliteTxAdapter(this.db);
+    this.db.exec('BEGIN');
+    try {
+      const result = await fn(txAdapter);
+      this.db.exec('COMMIT');
+      return result;
+    } catch (err) {
+      this.db.exec('ROLLBACK');
+      throw err;
+    }
+  }
+
+  async migrate(): Promise<void> {
+    const createTable = (name: string) =>
+      this.db.exec(`
+        CREATE TABLE IF NOT EXISTS ${name} (
+          id TEXT PRIMARY KEY,
+          data_json TEXT NOT NULL DEFAULT '{}',
+          created_at TEXT NOT NULL DEFAULT (datetime('now')),
+          updated_at TEXT NOT NULL DEFAULT (datetime('now'))
+        )
+      `);
+    for (const collection of COLLECTIONS) {
+      createTable(collection);
+    }
+  }
+
+  async close(): Promise<void> {
+    this.db.close();
+  }
+}
+
+/**
+ * Transaction wrapper that uses the same db handle — better-sqlite3 transactions
+ * are connection-level, so all statements on the same Database instance within
+ * a db.transaction() callback participate in the transaction.
+ */
+class SqliteTxAdapter implements StorageAdapter {
+  readonly name = 'sqlite';
+  private db: Database.Database;
+
+  constructor(db: Database.Database) {
+    this.db = db;
+  }
+
+  async create<T extends Record<string, unknown>>(
+    collection: string,
+    data: T,
+  ): Promise<T & { id: string }> {
+    const id = (data as any).id ?? randomUUID();
+    const now = new Date().toISOString();
+    const rest = Object.fromEntries(Object.entries(data).filter(([k]) => k !== 'id'));
+    this.db
+      .prepare(
+        `INSERT INTO ${collection} (id, data_json, created_at, updated_at) VALUES (?, ?, ?, ?)`,
+      )
+      .run(id, JSON.stringify(rest), now, now);
+    return { ...data, id } as T & { id: string };
+  }
+
+  async read<T extends Record<string, unknown>>(collection: string, id: string): Promise<T | null> {
+    const row = this.db.prepare(`SELECT * FROM ${collection} WHERE id = ?`).get(id) as any;
+    if (!row) return null;
+    return { id: row.id, ...JSON.parse(row.data_json as string) } as T;
+  }
+
+  async update(collection: string, id: string, data: Record<string, unknown>): Promise<boolean> {
+    const existing = this.db
+      .prepare(`SELECT data_json FROM ${collection} WHERE id = ?`)
+      .get(id) as any;
+    if (!existing) return false;
+    const merged = { ...JSON.parse(existing.data_json as string), ...data };
+    const now = new Date().toISOString();
+    const result = this.db
+      .prepare(`UPDATE ${collection} SET data_json = ?, updated_at = ? WHERE id = ?`)
+      .run(JSON.stringify(merged), now, id);
+    return result.changes > 0;
+  }
+
+  async delete(collection: string, id: string): Promise<boolean> {
+    const result = this.db.prepare(`DELETE FROM ${collection} WHERE id = ?`).run(id);
+    return result.changes > 0;
+  }
+
+  async find<T extends Record<string, unknown>>(
+    collection: string,
+    filter?: Record<string, unknown>,
+    opts?: { limit?: number; offset?: number; orderBy?: string; order?: 'asc' | 'desc' },
+  ): Promise<T[]> {
+    const { clause, params } = buildFilterClause(filter);
+    let query = `SELECT * FROM ${collection}${clause}`;
+    if (opts?.orderBy) {
+      const dir = opts.order === 'desc' ? 'DESC' : 'ASC';
+      const col =
+        opts.orderBy === 'id' || opts.orderBy === 'created_at' || opts.orderBy === 'updated_at'
+          ? opts.orderBy
+          : `json_extract(data_json, '$.${opts.orderBy}')`;
+      query += ` ORDER BY ${col} ${dir}`;
+    }
+    if (opts?.limit) {
+      query += ` LIMIT ?`;
+      params.push(opts.limit);
+    }
+    if (opts?.offset) {
+      query += ` OFFSET ?`;
+      params.push(opts.offset);
+    }
+    const rows = this.db.prepare(query).all(...params) as any[];
+    return rows.map((row) => ({ id: row.id, ...JSON.parse(row.data_json as string) }) as T);
+  }
+
+  async findOne<T extends Record<string, unknown>>(
+    collection: string,
+    filter: Record<string, unknown>,
+  ): Promise<T | null> {
+    const results = await this.find<T>(collection, filter, { limit: 1 });
+    return results[0] ?? null;
+  }
+
+  async count(collection: string, filter?: Record<string, unknown>): Promise<number> {
+    const { clause, params } = buildFilterClause(filter);
+    const row = this.db
+      .prepare(`SELECT COUNT(*) as count FROM ${collection}${clause}`)
+      .get(...params) as any;
+    return row?.count ?? 0;
+  }
+
+  async transaction<T>(fn: (tx: StorageAdapter) => Promise<T>): Promise<T> {
+    return fn(this);
+  }
+
+  async migrate(): Promise<void> {
+    // No-op inside transaction
+  }
+
+  async close(): Promise<void> {
+    // No-op inside transaction
+  }
+}