feat(db): Bun.SQL Kysely dialect for sqlite and postgres

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

packages/db/src/db.ts

@@ -0,0 +1,28 @@
+import { CompiledQuery, Kysely } from "kysely";
+import { BunSqlDialect, type Dialect } from "./dialect.ts";
+import type { Database } from "./schema.ts";
+
+export interface Db {
+  k: Kysely<Database>;
+  dialect: Dialect;
+  close(): Promise<void>;
+}
+
+/** Open a database from `sqlite://<path>`, `sqlite://:memory:` or `postgres://…`. */
+export async function createDb(url: string): Promise<Db> {
+  const d = new BunSqlDialect(url);
+  const k = new Kysely<Database>({ dialect: d });
+  try {
+    if (d.dialect === "sqlite") {
+      await k.executeQuery(CompiledQuery.raw("PRAGMA foreign_keys = ON"));
+      if (!url.includes(":memory:")) await k.executeQuery(CompiledQuery.raw("PRAGMA journal_mode = WAL"));
+    } else {
+      // Fail fast on a bad URL or unreachable server instead of at first query.
+      await k.executeQuery(CompiledQuery.raw("select 1"));
+    }
+  } catch (err) {
+    await k.destroy();
+    throw err;
+  }
+  return { k, dialect: d.dialect, close: () => k.destroy() };
+}

packages/db/src/dialect.ts

@@ -0,0 +1,160 @@
+import { SQL, type ReservedSQL } from "bun";
+import {
+  CompiledQuery,
+  PostgresAdapter,
+  PostgresIntrospector,
+  PostgresQueryCompiler,
+  SqliteAdapter,
+  SqliteIntrospector,
+  SqliteQueryCompiler,
+  type DatabaseConnection,
+  type DatabaseIntrospector,
+  type DialectAdapter,
+  type Dialect as KyselyDialect,
+  type Driver,
+  type Kysely,
+  type QueryCompiler,
+  type QueryResult,
+  type TransactionSettings,
+} from "kysely";
+
+export type Dialect = "sqlite" | "postgres";
+
+export function dialectOf(url: string): Dialect {
+  if (url.startsWith("sqlite://")) return "sqlite";
+  if (url.startsWith("postgres://") || url.startsWith("postgresql://")) return "postgres";
+  throw new Error(`Unsupported database URL scheme: ${url.split(":")[0]}`);
+}
+
+/** Only these commands report a meaningful affected-row count. */
+const WRITE_COMMANDS = new Set(["INSERT", "UPDATE", "DELETE", "MERGE"]);
+
+type Runner = Pick<SQL, "unsafe">;
+
+class BunSqlConnection implements DatabaseConnection {
+  constructor(
+    readonly runner: Runner,
+    private readonly dialect: Dialect,
+  ) {}
+
+  async executeQuery<R>(compiled: CompiledQuery): Promise<QueryResult<R>> {
+    const result = await this.runner.unsafe(compiled.sql, compiled.parameters as unknown[]);
+    const rows = [...result] as R[];
+    const command: string | undefined = result.command;
+    if (!command || !WRITE_COMMANDS.has(command)) return { rows };
+    const numAffectedRows = BigInt(result.count ?? 0);
+    const lastId = result.lastInsertRowid;
+    if (this.dialect === "sqlite" && command === "INSERT" && lastId != null) {
+      return { rows, numAffectedRows, insertId: BigInt(lastId) };
+    }
+    return { rows, numAffectedRows };
+  }
+
+  async *streamQuery<R>(compiled: CompiledQuery): AsyncIterableIterator<QueryResult<R>> {
+    yield await this.executeQuery<R>(compiled);
+  }
+}
+
+/**
+ * Serialises access to the single SQLite connection. Bun.SQL's SQLite adapter
+ * does not support `reserve()`, so a held connection is the whole database.
+ */
+class Mutex {
+  private tail: Promise<void> = Promise.resolve();
+
+  async lock(): Promise<() => void> {
+    let unlock!: () => void;
+    const next = new Promise<void>((resolve) => (unlock = resolve));
+    const prev = this.tail;
+    this.tail = prev.then(() => next);
+    await prev;
+    return unlock;
+  }
+}
+
+class BunSqlDriver implements Driver {
+  private sql: SQL | undefined;
+  private readonly sqliteLock = new Mutex();
+  private readonly unlocks = new WeakMap<DatabaseConnection, () => void>();
+
+  constructor(
+    private readonly url: string,
+    private readonly dialect: Dialect,
+  ) {}
+
+  async init(): Promise<void> {
+    this.sql = new SQL(this.url);
+  }
+
+  private get client(): SQL {
+    if (!this.sql) throw new Error("Database driver is not initialised or already destroyed");
+    return this.sql;
+  }
+
+  async acquireConnection(): Promise<DatabaseConnection> {
+    if (this.dialect === "postgres") {
+      return new BunSqlConnection(await this.client.reserve(), this.dialect);
+    }
+    const unlock = await this.sqliteLock.lock();
+    const conn = new BunSqlConnection(this.client, this.dialect);
+    this.unlocks.set(conn, unlock);
+    return conn;
+  }
+
+  async releaseConnection(connection: DatabaseConnection): Promise<void> {
+    if (this.dialect === "postgres") {
+      ((connection as BunSqlConnection).runner as ReservedSQL).release();
+      return;
+    }
+    this.unlocks.get(connection)?.();
+    this.unlocks.delete(connection);
+  }
+
+  async beginTransaction(connection: DatabaseConnection, settings: TransactionSettings): Promise<void> {
+    let stmt = "begin";
+    if (this.dialect === "postgres") {
+      if (settings.isolationLevel) stmt += ` isolation level ${settings.isolationLevel}`;
+      if (settings.accessMode) stmt += ` ${settings.accessMode}`;
+    }
+    await connection.executeQuery(CompiledQuery.raw(stmt));
+  }
+
+  async commitTransaction(connection: DatabaseConnection): Promise<void> {
+    await connection.executeQuery(CompiledQuery.raw("commit"));
+  }
+
+  async rollbackTransaction(connection: DatabaseConnection): Promise<void> {
+    await connection.executeQuery(CompiledQuery.raw("rollback"));
+  }
+
+  async destroy(): Promise<void> {
+    const sql = this.sql;
+    this.sql = undefined;
+    await sql?.close();
+  }
+}
+
+/** One Kysely dialect over Bun.SQL for both `sqlite://` and `postgres://` URLs. */
+export class BunSqlDialect implements KyselyDialect {
+  readonly dialect: Dialect;
+
+  constructor(private readonly url: string) {
+    this.dialect = dialectOf(url);
+  }
+
+  createDriver(): Driver {
+    return new BunSqlDriver(this.url, this.dialect);
+  }
+
+  createQueryCompiler(): QueryCompiler {
+    return this.dialect === "sqlite" ? new SqliteQueryCompiler() : new PostgresQueryCompiler();
+  }
+
+  createAdapter(): DialectAdapter {
+    return this.dialect === "sqlite" ? new SqliteAdapter() : new PostgresAdapter();
+  }
+
+  createIntrospector(db: Kysely<any>): DatabaseIntrospector {
+    return this.dialect === "sqlite" ? new SqliteIntrospector(db) : new PostgresIntrospector(db);
+  }
+}

packages/db/src/ids.ts

@@ -0,0 +1,9 @@
+/** Time-sortable UUIDv7, generated in app code. */
+export function newId(): string {
+  return Bun.randomUUIDv7();
+}
+
+/** Current time as an ISO-8601 UTC string, the storage format for timestamps. */
+export function nowIso(): string {
+  return new Date().toISOString();
+}

packages/db/src/json.ts

@@ -0,0 +1,14 @@
+/** Serialise a value for a JSON column (stored as text on both dialects). */
+export function toJson(v: unknown): string {
+  return JSON.stringify(v);
+}
+
+/** Parse a JSON column, returning `fallback` for null, undefined or invalid text. */
+export function fromJson<T>(s: string | null | undefined, fallback: T): T {
+  if (s == null) return fallback;
+  try {
+    return JSON.parse(s) as T;
+  } catch {
+    return fallback;
+  }
+}

packages/db/src/migrate.ts

@@ -0,0 +1,24 @@
+import type { Kysely } from "kysely";
+import { Migrator, type Migration, type MigrationProvider } from "kysely/migration";
+import type { Db } from "./db.ts";
+import { migrations } from "./migrations/index.ts";
+
+class StaticMigrationProvider implements MigrationProvider {
+  constructor(private readonly all: Record<string, Migration>) {}
+  async getMigrations(): Promise<Record<string, Migration>> {
+    return this.all;
+  }
+}
+
+/** Apply every pending migration; throws on the first failure. A no-op when there are none. */
+export async function migrateToLatest(db: Db): Promise<void> {
+  const migrator = new Migrator({
+    db: db.k as Kysely<any>,
+    provider: new StaticMigrationProvider(migrations),
+  });
+  const { error, results } = await migrator.migrateToLatest();
+  if (error) {
+    const failed = results?.find((r) => r.status === "Error")?.migrationName;
+    throw new Error(`Migration ${failed ?? "(unknown)"} failed: ${String(error)}`, { cause: error });
+  }
+}

packages/db/src/migrations/index.ts

@@ -0,0 +1,4 @@
+import type { Migration } from "kysely/migration";
+
+/** All migrations, keyed by sortable name. Imported statically so bundles need no filesystem. */
+export const migrations: Record<string, Migration> = {};

packages/db/src/schema.ts

@@ -0,0 +1,2 @@
+/** Kysely table types. Placeholder until the schema migration lands. */
+export interface Database {}

packages/db/src/testing.ts

@@ -0,0 +1,63 @@
+import { SQL } from "bun";
+import { afterAll, afterEach, beforeEach, describe } from "bun:test";
+import { createDb, type Db } from "./db.ts";
+import type { Dialect } from "./dialect.ts";
+import { migrateToLatest } from "./migrate.ts";
+
+/**
+ * Register `describe("<name> [<dialect>]")` for SQLite, and for Postgres when
+ * `TEST_POSTGRES_URL` is set. Every test gets a fresh, migrated database:
+ * SQLite `:memory:`, or a new `st_<random>` database on the Postgres server.
+ */
+export function forEachDialect(name: string, fn: (getDb: () => Db) => void): void {
+  const dialects: Dialect[] = ["sqlite"];
+  const pgUrl = process.env.TEST_POSTGRES_URL;
+  if (pgUrl) dialects.push("postgres");
+
+  for (const dialect of dialects) {
+    describe(`${name} [${dialect}]`, () => {
+      let current: Db | undefined;
+      const created: string[] = [];
+
+      beforeEach(async () => {
+        let url = "sqlite://:memory:";
+        if (dialect === "postgres") {
+          const dbName = `st_${crypto.randomUUID().replaceAll("-", "")}`;
+          await withAdmin(pgUrl!, (admin) => admin.unsafe(`create database "${dbName}"`));
+          created.push(dbName);
+          const u = new URL(pgUrl!);
+          u.pathname = `/${dbName}`;
+          url = u.toString();
+        }
+        current = await createDb(url);
+        await migrateToLatest(current);
+      });
+
+      afterEach(async () => {
+        await current?.close();
+        current = undefined;
+      });
+
+      afterAll(async () => {
+        if (created.length === 0) return;
+        await withAdmin(pgUrl!, async (admin) => {
+          for (const dbName of created) await admin.unsafe(`drop database if exists "${dbName}" with (force)`);
+        });
+      });
+
+      fn(() => {
+        if (!current) throw new Error("getDb() called outside a test");
+        return current;
+      });
+    });
+  }
+}
+
+async function withAdmin<T>(url: string, f: (admin: SQL) => Promise<T>): Promise<T> {
+  const admin = new SQL(url);
+  try {
+    return await f(admin);
+  } finally {
+    await admin.close();
+  }
+}