Release v0.7.8: queue SQLite writes, upgrade libsql to 0.18

cli/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@codeplans/cli",
-  "version": "0.7.7",
+  "version": "0.7.8",
   "description": "Command-line tool for CodePlans: script workspace setups, work with code plans, tasks and work items from your terminal.",
   "license": "MIT",
   "type": "module",

lib/db/index.ts

@@ -4,13 +4,18 @@ import { resolveDbSsl, sqliteFilePath } from '@/lib/runtime-env'
 import { getEnterpriseHooks } from '@/lib/ee/registry'
 import type { PostgresJsDatabase } from 'drizzle-orm/postgres-js'
 import type * as PgSchema from './schema.pg'
+import { queueWrites, WriteQueue } from './sqlite-writes'
 
 // TypeScript always sees db as the postgres type for full IDE / type-check support.
 // In SQLite mode the runtime object is a libsql Drizzle instance but the column names
 // and table names are identical, so all queries work correctly.
 type Db = PostgresJsDatabase<typeof PgSchema>
 
-function createDb(rawUrl: string): Db {
+// `writes` is set for a local SQLite file, where writes must run one at a time
+// (see lib/db/sqlite-writes.ts).
+type Handle = { instance: Db; writes: WriteQueue | null }
+
+function createDb(rawUrl: string): Handle {
   if (config.db.provider === 'sqlite') {
     const { createClient } = require('@libsql/client') as typeof import('@libsql/client')
     const { drizzle } = require('drizzle-orm/libsql') as typeof import('drizzle-orm/libsql')
@@ -28,34 +33,39 @@ function createDb(rawUrl: string): Db {
       mkdirSync(dirname(file), { recursive: true })
     }
     // DATABASE_AUTH_TOKEN is only needed for a hosted libsql server (Turso).
-    const client = createClient({ url, authToken: process.env.DATABASE_AUTH_TOKEN || undefined })
-    // WAL allows concurrent readers alongside a writer; busy_timeout retries
-    // instead of immediately throwing SQLITE_BUSY when a write lock is held.
+    // `timeout` is the busy timeout (ms) for every connection in libsql's
+    // pool: a write retries instead of failing with SQLITE_BUSY while another
+    // process holds the lock.
+    let client = createClient({ url, authToken: process.env.DATABASE_AUTH_TOKEN || undefined, timeout: 5000 })
+    // WAL allows concurrent readers alongside a writer. It's stored in the
+    // database file, so setting it on one connection covers them all.
     client.execute('PRAGMA journal_mode=WAL')
-    client.execute('PRAGMA busy_timeout=5000')
-    return drizzle(client, { schema }) as unknown as PostgresJsDatabase<typeof PgSchema>
+    // A hosted libsql server (Turso) arbitrates writes itself.
+    const writes = url === ':memory:' || /^file:/i.test(url) ? new WriteQueue() : null
+    if (writes) client = queueWrites(client, writes)
+    return { instance: drizzle(client, { schema }) as unknown as PostgresJsDatabase<typeof PgSchema>, writes }
   } else {
     const postgres = require('postgres') as typeof import('postgres')
     const { drizzle } = require('drizzle-orm/postgres-js') as typeof import('drizzle-orm/postgres-js')
     const schema = require('./schema.pg') as typeof PgSchema
     const client = postgres(rawUrl, { ssl: resolveDbSsl(rawUrl, process.env.DB_SSL) ? 'require' : false, prepare: false })
-    return drizzle(client, { schema })
+    return { instance: drizzle(client, { schema }), writes: null }
   }
 }
 
 // Drizzle methods that change data. A read-only workspace (see below) rejects
 // them before any SQL runs.
 const WRITE_METHODS = new Set<PropertyKey>(['insert', 'update', 'delete', 'transaction', 'run', 'execute', 'batch'])
 
-const opened = new Map<string, Db>()
+const opened = new Map<string, Handle>()
 
-function openDb(url: string): Db {
-  let instance = opened.get(url)
-  if (!instance) {
-    instance = createDb(url)
-    opened.set(url, instance)
+function openDb(url: string): Handle {
+  let handle = opened.get(url)
+  if (!handle) {
+    handle = createDb(url)
+    opened.set(url, handle)
   }
-  return instance
+  return handle
 }
 
 /**
@@ -64,10 +74,10 @@ function openDb(url: string): Db {
  * through the `workspaceDatabase` hook (lib/ee/types.ts); it is opened once and
  * reused.
  */
-function currentDb(): { instance: Db; readOnly: string | null } {
+function currentDb(): Handle & { readOnly: string | null } {
   const workspace = getEnterpriseHooks().workspaceDatabase()
-  if (!workspace) return { instance: openDb(config.db.url), readOnly: null }
-  return { instance: openDb(workspace.url), readOnly: workspace.readOnly ?? null }
+  if (!workspace) return { ...openDb(config.db.url), readOnly: null }
+  return { ...openDb(workspace.url), readOnly: workspace.readOnly ?? null }
 }
 
 /** Thrown when a read-only workspace tries to change data. */
@@ -83,10 +93,13 @@ export class ReadOnlyDatabaseError extends Error {
 // database, opened on first use, so behavior is the same as a plain instance.
 export const db: Db = new Proxy({} as Db, {
   get(_target, prop) {
-    const { instance, readOnly } = currentDb()
+    const { instance, writes, readOnly } = currentDb()
     if (readOnly && WRITE_METHODS.has(prop)) {
       return () => { throw new ReadOnlyDatabaseError(readOnly) }
     }
+    if (writes && prop === 'transaction') {
+      return (...args: Parameters<Db['transaction']>) => writes.transaction(() => instance.transaction(...args))
+    }
     const value = Reflect.get(instance, prop, instance)
     return typeof value === 'function' ? value.bind(instance) : value
   },

lib/db/sqlite-writes.ts

@@ -0,0 +1,86 @@
+import { AsyncLocalStorage } from 'node:async_hooks'
+import type { Client, InStatement } from '@libsql/client'
+
+/** Thrown when code inside db.transaction() writes through `db` instead of `tx`. */
+export class NestedWriteError extends Error {
+  constructor() {
+    super('Write through db inside db.transaction(); use the transaction (tx) passed to the callback')
+    this.name = 'NestedWriteError'
+  }
+}
+
+/**
+ * Lets one write at a time run against a local SQLite file.
+ *
+ * libsql's local driver runs each statement synchronously, but a transaction
+ * spans awaits. While one is open it holds the file's write lock on its own
+ * pooled connection, so a write that arrives meanwhile on another connection
+ * busy-waits synchronously, blocking the event loop the transaction needs in
+ * order to finish, and fails with SQLITE_BUSY when the timeout runs out.
+ * Queueing writes behind open transactions avoids that. Reads don't queue:
+ * WAL lets them run beside a writer.
+ */
+export class WriteQueue {
+  private tail: Promise<void> = Promise.resolve()
+  private readonly scope = new AsyncLocalStorage<{ open: boolean }>()
+
+  /** Runs fn once every write queued before it has finished. */
+  async run<T>(fn: () => Promise<T> | T): Promise<T> {
+    const previous = this.tail
+    let release!: () => void
+    this.tail = new Promise<void>((resolve) => { release = resolve })
+    await previous
+    try {
+      return await fn()
+    } finally {
+      release()
+    }
+  }
+
+  /** Runs a whole transaction as one queued write. */
+  transaction<T>(fn: () => Promise<T>): Promise<T> {
+    if (this.inTransaction()) return Promise.reject(new NestedWriteError())
+    return this.run(() => {
+      const state = { open: true }
+      return this.scope.run(state, async () => {
+        try {
+          return await fn()
+        } finally {
+          // Work started inside the callback that outlives it queues normally.
+          state.open = false
+        }
+      })
+    })
+  }
+
+  inTransaction(): boolean {
+    return this.scope.getStore()?.open === true
+  }
+}
+
+const isRead = (stmt: InStatement) => /^\s*select\b/i.test(typeof stmt === 'string' ? stmt : stmt.sql)
+
+/**
+ * Sends a local libsql client's writes through `queue`. db.transaction() must
+ * go through queue.transaction() (see lib/db/index.ts); the client's own
+ * transaction() then runs inside that queued slot.
+ */
+export function queueWrites(client: Client, queue: WriteQueue): Client {
+  const execute = client.execute.bind(client) as (...args: unknown[]) => ReturnType<Client['execute']>
+  const batch = client.batch.bind(client)
+  const migrate = client.migrate.bind(client)
+  const executeMultiple = client.executeMultiple.bind(client)
+
+  const write = <T>(fn: () => Promise<T>): Promise<T> =>
+    queue.inTransaction() ? Promise.reject(new NestedWriteError()) : queue.run(fn)
+
+  Object.assign(client, {
+    execute: (...args: unknown[]) =>
+      isRead(args[0] as InStatement) ? execute(...args) : write(() => execute(...args)),
+    batch: (stmts: Parameters<Client['batch']>[0], mode?: Parameters<Client['batch']>[1]) =>
+      mode === 'read' ? batch(stmts, mode) : write(() => batch(stmts, mode)),
+    migrate: (stmts: Parameters<Client['migrate']>[0]) => write(() => migrate(stmts)),
+    executeMultiple: (sql: string) => write(() => executeMultiple(sql)),
+  })
+  return client
+}

package.json

@@ -1,6 +1,6 @@
 {
   "name": "codeplans",
-  "version": "0.7.7",
+  "version": "0.7.8",
   "description": "Manage and track coordinated changes across your software architecture.",
   "author": "Sai Prakash <sylonzero@gmail.com>",
   "homepage": "https://codeplans.ai",
@@ -27,7 +27,7 @@
   "dependencies": {
     "@anthropic-ai/sdk": "0.111.0",
     "@hookform/resolvers": "^3.9.1",
-    "@libsql/client": "^0.17.3",
+    "@libsql/client": "^0.18.0",
     "@radix-ui/react-accordion": "1.2.12",
     "@radix-ui/react-alert-dialog": "1.1.15",
     "@radix-ui/react-aspect-ratio": "1.1.8",