diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 71502e3..c4766a8 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -17,7 +17,7 @@ user layer or files a Changes-inbox event. | `apps/server` | `Bun.serve`: `/api/v1`, `/ingest/`, job worker, static web | planned | | `apps/web` | React app (HTML imports) | planned | | `packages/core` | domain services: entries, matching, change engine, history, export | planned | -| `packages/db` | Kysely, SQLite/Postgres dialects, migrations | planned | +| `packages/db` | Kysely over one Bun.SQL dialect (SQLite + Postgres), ids, JSON helpers, migrations, `forEachDialect` test harness | exists (schema pending) | | `packages/plugin-sdk` | plugin contract, `SourceRecord`, http/rate-limit helpers, fixture test harness | planned | | `plugins/` | one source integration each (hardcover, trakt, mangabaka, listenbrainz, lastfm, csv) | planned | | `docs/design/mockups` | approved UI mockups (reference only) | exists | diff --git a/docs/DECISIONS.md b/docs/DECISIONS.md index 12a102a..2a7b147 100644 --- a/docs/DECISIONS.md +++ b/docs/DECISIONS.md @@ -4,6 +4,41 @@ Append-only. Never edit or delete an entry. Supersede it with a new one that lin One entry per decision a future maintainer would otherwise have to reverse-engineer. No index: `grep -n '^## ' docs/DECISIONS.md` lists every entry (date, title, line number). +## 2026-10-30: A single Kysely dialect over Bun.SQL for SQLite and Postgres +**Status:** accepted +**Context:** Kysely ships no driver for Bun. `Bun.SQL` speaks both `sqlite://` and `postgres://` behind one API. +**Decision:** +- `packages/db/src/dialect.ts` is one Kysely `Dialect` whose driver wraps `new SQL(url)`, choosing Kysely's + SQLite or Postgres adapter, compiler and introspector by URL scheme. +- Postgres connections come from `sql.reserve()`. Bun's SQLite adapter does not support `reserve()`, so SQLite + uses the single `SQL` instance as its only connection, serialised by a mutex so transactions never interleave. +- Transactions are plain `begin`/`commit`/`rollback` on the held connection. + +**Alternatives:** +- Separate dialects over `bun:sqlite` and `Bun.sql`: two drivers to maintain for the same job. +- Community Bun dialect packages: an extra dependency, and none covers both databases. + +**Consequences:** SQLite is effectively single-connection (fine for a self-hosted app). Savepoints and real +streaming are not implemented; `streamQuery` returns the whole result as one chunk. + +## 2026-10-30: JSON columns are stored as text on both dialects +**Status:** accepted +**Context:** Spec §11 calls for `jsonb`. SQLite has no JSON type, so `jsonb` on Postgres would need +per-dialect read/write code in every service that touches a JSON column. +**Decision:** JSON columns are `text` everywhere, written with `toJson()` and read with `fromJson()` from +`@sttracker/db/json`. `fromJson` returns a caller-supplied fallback for null or invalid text. +**Alternatives:** `jsonb` on Postgres with per-dialect read/write code. +**Consequences:** No JSON operators or indexes in SQL; filtering on JSON content happens in app code. Can be +superseded later by a migration to `jsonb` if a query needs it. + +## 2026-10-30: UUIDv7 instead of ULID for IDs +**Status:** accepted +**Context:** Spec and `AGENTS.md` say IDs are ULIDs generated in app code. Bun has `Bun.randomUUIDv7()` built in. +**Decision:** `newId()` in `@sttracker/db/ids` returns `Bun.randomUUIDv7()`. IDs are stored as `text`. +**Alternatives:** ULID, which needs a dependency or hand-written encoder. +**Consequences:** IDs are equally time-sortable as strings with no extra dependency. They are 36 characters +instead of 26. Where `AGENTS.md` or the spec say "ULID", read "UUIDv7". + ## 2026-10-30: Product design decisions recorded in spec 001 **Status:** accepted **Context:** Brainstorming with the owner produced the v1 design. diff --git a/packages/db/src/db.ts b/packages/db/src/db.ts new file mode 100644 index 0000000..d7d2dde --- /dev/null +++ b/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; + dialect: Dialect; + close(): Promise; +} + +/** Open a database from `sqlite://`, `sqlite://:memory:` or `postgres://…`. */ +export async function createDb(url: string): Promise { + const d = new BunSqlDialect(url); + const k = new Kysely({ 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() }; +} diff --git a/packages/db/src/dialect.ts b/packages/db/src/dialect.ts new file mode 100644 index 0000000..f4a7560 --- /dev/null +++ b/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; + +class BunSqlConnection implements DatabaseConnection { + constructor( + readonly runner: Runner, + private readonly dialect: Dialect, + ) {} + + async executeQuery(compiled: CompiledQuery): Promise> { + 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(compiled: CompiledQuery): AsyncIterableIterator> { + yield await this.executeQuery(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 = Promise.resolve(); + + async lock(): Promise<() => void> { + let unlock!: () => void; + const next = new Promise((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 void>(); + + constructor( + private readonly url: string, + private readonly dialect: Dialect, + ) {} + + async init(): Promise { + 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 { + 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 { + 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 { + 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 { + await connection.executeQuery(CompiledQuery.raw("commit")); + } + + async rollbackTransaction(connection: DatabaseConnection): Promise { + await connection.executeQuery(CompiledQuery.raw("rollback")); + } + + async destroy(): Promise { + 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): DatabaseIntrospector { + return this.dialect === "sqlite" ? new SqliteIntrospector(db) : new PostgresIntrospector(db); + } +} diff --git a/packages/db/src/ids.ts b/packages/db/src/ids.ts new file mode 100644 index 0000000..40c91ad --- /dev/null +++ b/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(); +} diff --git a/packages/db/src/json.ts b/packages/db/src/json.ts new file mode 100644 index 0000000..90292a7 --- /dev/null +++ b/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(s: string | null | undefined, fallback: T): T { + if (s == null) return fallback; + try { + return JSON.parse(s) as T; + } catch { + return fallback; + } +} diff --git a/packages/db/src/migrate.ts b/packages/db/src/migrate.ts new file mode 100644 index 0000000..8b1e151 --- /dev/null +++ b/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) {} + async getMigrations(): Promise> { + 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 { + const migrator = new Migrator({ + db: db.k as Kysely, + 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 }); + } +} diff --git a/packages/db/src/migrations/index.ts b/packages/db/src/migrations/index.ts new file mode 100644 index 0000000..418b1a7 --- /dev/null +++ b/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 = {}; diff --git a/packages/db/src/schema.ts b/packages/db/src/schema.ts new file mode 100644 index 0000000..2310ae7 --- /dev/null +++ b/packages/db/src/schema.ts @@ -0,0 +1,2 @@ +/** Kysely table types. Placeholder until the schema migration lands. */ +export interface Database {} diff --git a/packages/db/src/testing.ts b/packages/db/src/testing.ts new file mode 100644 index 0000000..c8dade8 --- /dev/null +++ b/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(" []")` for SQLite, and for Postgres when + * `TEST_POSTGRES_URL` is set. Every test gets a fresh, migrated database: + * SQLite `:memory:`, or a new `st_` 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(url: string, f: (admin: SQL) => Promise): Promise { + const admin = new SQL(url); + try { + return await f(admin); + } finally { + await admin.close(); + } +} diff --git a/packages/db/test/db.test.ts b/packages/db/test/db.test.ts new file mode 100644 index 0000000..2e6f1c5 --- /dev/null +++ b/packages/db/test/db.test.ts @@ -0,0 +1,63 @@ +import { expect, test } from "bun:test"; +import { sql } from "kysely"; +import { newId } from "../src/ids.ts"; +import { fromJson, toJson } from "../src/json.ts"; +import { forEachDialect } from "../src/testing.ts"; + +forEachDialect("db basics", (getDb) => { + async function createTempTable() { + await sql`create table t (id text primary key, j text not null)`.execute(getDb().k); + } + + async function countRows(): Promise { + const { rows } = await sql<{ n: number | string }>`select count(*) as n from t`.execute(getDb().k); + return Number(rows[0]?.n); + } + + test("round-trips a row", async () => { + await createTempTable(); + const id = newId(); + await sql`insert into t (id, j) values (${id}, ${toJson({ a: 1 })})`.execute(getDb().k); + const { rows } = await sql<{ id: string; j: string }>`select id, j from t`.execute(getDb().k); + expect(rows).toHaveLength(1); + const row = rows[0]!; + expect(row.id).toBe(id); + expect(fromJson(row.j, {})).toEqual({ a: 1 }); + }); + + test("transactions roll back", async () => { + await createTempTable(); + await expect( + getDb() + .k.transaction() + .execute(async (tx) => { + await sql`insert into t (id, j) values (${newId()}, ${toJson(null)})`.execute(tx); + throw new Error("x"); + }), + ).rejects.toThrow("x"); + expect(await countRows()).toBe(0); + }); + + test("transactions commit", async () => { + await createTempTable(); + await getDb() + .k.transaction() + .execute(async (tx) => { + await sql`insert into t (id, j) values (${newId()}, ${toJson([1])})`.execute(tx); + }); + expect(await countRows()).toBe(1); + }); + + test("newId is time-sortable", () => { + const ids = Array.from({ length: 1000 }, () => newId()); + expect([...ids].sort()).toEqual(ids); + expect(new Set(ids).size).toBe(1000); + }); +}); + +test("fromJson falls back on null, undefined and invalid input", () => { + expect(fromJson(null, 5)).toBe(5); + expect(fromJson(undefined, "d")).toBe("d"); + expect(fromJson("{not json", [])).toEqual([]); + expect(fromJson(toJson({ b: [1, "x"] }), {})).toEqual({ b: [1, "x"] }); +});