feat(storage): sqlite backend — one database hosting all routed units
node:sqlite DatabaseSync with the session-persistence-sqlite open sequence (0o700 dir, exclusive 0o600 create, foreign_keys, configurable journal mode, user_version stamp-or-reject). STRICT tables throughout: units/unit_globals meta tables plus one document-per-row table per declared unit table, keeping per-key durable updates precise.
This commit is contained in:
imccyu
167
packages/storage/storage-sqlite/src/index.ts
Normal file
167
packages/storage/storage-sqlite/src/index.ts
Normal file
@@ -0,0 +1,167 @@
|
||||
/**
|
||||
* SQLite storage backend for the storage hub: one database file hosts every
|
||||
* routed unit, document-per-row (`key TEXT` / `value TEXT` JSON). Registers
|
||||
* as backend `sqlite`; the disposer unregisters first, then closes the medium.
|
||||
* @module @deepseek-ai/dsh-storage-sqlite
|
||||
*/
|
||||
|
||||
import type { Context } from 'cordis'
|
||||
import z from 'schemastery'
|
||||
import type { DatabaseSync } from 'node:sqlite'
|
||||
import { StorageError, UNIT_NAME_RE } from '@deepseek-ai/dsh-storage'
|
||||
import type { KvFacet, KvUnit, KvUnitDescriptor, StorageBackend } from '@deepseek-ai/dsh-storage'
|
||||
import { openDatabase, recordTableName, type JournalMode } from './schema.ts'
|
||||
import { SqliteKvUnit } from './unit.ts'
|
||||
|
||||
export { STORAGE_SQLITE_SCHEMA_VERSION, type JournalMode } from './schema.ts'
|
||||
|
||||
/** Cordis plugin name. */
|
||||
export const name = 'storage-sqlite'
|
||||
/** The backend registers on the storage hub. */
|
||||
export const inject = ['storage']
|
||||
|
||||
/** Plugin configuration. */
|
||||
export interface Config {
|
||||
/**
|
||||
* Filesystem path to the SQLite database file. The special value `:memory:`
|
||||
* opens an in-process database (tests). On filesystems with POSIX modes,
|
||||
* missing directories and databases are created owner-only; existing path
|
||||
* modes are preserved. Filesystem setup errors other than an existing
|
||||
* database fail the open. The backend does not protect confidentiality or
|
||||
* integrity when another principal can replace the database entry in its
|
||||
* parent directory.
|
||||
*/
|
||||
path: string
|
||||
/**
|
||||
* SQLite `journal_mode` pragma. `wal` (the default) suits local disks; pick
|
||||
* a rollback-journal mode (`delete`/`truncate`/`persist`) on filesystems
|
||||
* where WAL's shared-memory files do not work (network mounts). See
|
||||
* {@link JournalMode}.
|
||||
*/
|
||||
journalMode?: JournalMode
|
||||
}
|
||||
|
||||
/** Schemastery validator for {@link Config}. */
|
||||
export const Config: z<Config> = z.object({
|
||||
path: z.string().required(),
|
||||
journalMode: z.union(['wal', 'delete', 'truncate', 'persist'] as const).default('wal'),
|
||||
})
|
||||
|
||||
/**
|
||||
* The SQLite {@link StorageBackend}. Owns one `DatabaseSync` connection and
|
||||
* the open-unit table; `kv.open` validates names, enforces the per-unit
|
||||
* version stamp in `units`, and ensures the unit's record tables.
|
||||
*/
|
||||
export class SqliteStorageBackend implements StorageBackend {
|
||||
/** The key-value facet; the only shape this backend serves. */
|
||||
readonly kv: KvFacet = { open: descriptor => this.openUnit(descriptor) }
|
||||
|
||||
private readonly ready: Promise<DatabaseSync>
|
||||
/** Open (or still-opening) units by name; presence is the double-open guard. */
|
||||
private readonly units = new Map<string, Promise<SqliteKvUnit>>()
|
||||
private closing: Promise<void> | undefined
|
||||
|
||||
/**
|
||||
* @param config - Validated plugin configuration.
|
||||
*/
|
||||
constructor(config: Config) {
|
||||
this.ready = openDatabase(config.path, (config as Required<Config>).journalMode)
|
||||
// Mark the rejection handled: every primitive re-awaits `ready`, so an
|
||||
// open failure still surfaces to each caller; this guard only prevents an
|
||||
// unhandled-rejection crash when the failure precedes the first use.
|
||||
this.ready.catch(() => {})
|
||||
}
|
||||
|
||||
private openUnit(descriptor: KvUnitDescriptor): Promise<KvUnit> {
|
||||
if (this.closing !== undefined) {
|
||||
return Promise.reject(new StorageError('closed', 'sqlite storage backend is closed'))
|
||||
}
|
||||
if (!UNIT_NAME_RE.test(descriptor.name)) {
|
||||
return Promise.reject(new Error(`kv unit name '${descriptor.name}' violates ${UNIT_NAME_RE}`))
|
||||
}
|
||||
for (const table of descriptor.tables) {
|
||||
if (!UNIT_NAME_RE.test(table)) {
|
||||
return Promise.reject(new Error(`kv table name '${table}' in unit '${descriptor.name}' violates ${UNIT_NAME_RE}`))
|
||||
}
|
||||
}
|
||||
if (this.units.has(descriptor.name)) {
|
||||
return Promise.reject(new Error(`kv unit '${descriptor.name}' is already open (double-open is a caller bug)`))
|
||||
}
|
||||
// Reserve the name synchronously so a concurrent second open of the same
|
||||
// name rejects instead of racing past the guard during the awaits below.
|
||||
const pending = this.materializeUnit(descriptor)
|
||||
this.units.set(descriptor.name, pending)
|
||||
pending.catch(() => this.units.delete(descriptor.name))
|
||||
return pending
|
||||
}
|
||||
|
||||
private async materializeUnit(descriptor: KvUnitDescriptor): Promise<SqliteKvUnit> {
|
||||
const db = await this.ready
|
||||
const row = db.prepare('SELECT version FROM units WHERE name = ?').get(descriptor.name) as
|
||||
| { version: number }
|
||||
| undefined
|
||||
if (row === undefined) {
|
||||
db.prepare('INSERT INTO units (name, version) VALUES (?, ?)').run(descriptor.name, descriptor.version)
|
||||
} else if (row.version !== descriptor.version) {
|
||||
throw new StorageError(
|
||||
'version-mismatch',
|
||||
`kv unit '${descriptor.name}' is stamped version ${row.version} on the medium, incompatible with descriptor version ${descriptor.version}`,
|
||||
)
|
||||
}
|
||||
for (const table of descriptor.tables) {
|
||||
// Both segments passed UNIT_NAME_RE, so the identifier is safe in DDL.
|
||||
db.exec(`
|
||||
CREATE TABLE IF NOT EXISTS "${recordTableName(descriptor.name, table)}" (
|
||||
key TEXT PRIMARY KEY,
|
||||
value TEXT NOT NULL
|
||||
) STRICT
|
||||
`)
|
||||
}
|
||||
return new SqliteKvUnit(db, descriptor, () => {
|
||||
this.units.delete(descriptor.name)
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Close every open unit and release the database. Idempotent; concurrent
|
||||
* and repeated calls resolve once teardown finishes.
|
||||
* @returns resolution after the medium is released.
|
||||
*/
|
||||
close(): Promise<void> {
|
||||
this.closing ??= this.doClose()
|
||||
return this.closing
|
||||
}
|
||||
|
||||
private async doClose(): Promise<void> {
|
||||
let db: DatabaseSync
|
||||
try {
|
||||
db = await this.ready
|
||||
} catch {
|
||||
// The medium never opened; that failure already rejected the opener and
|
||||
// every unit call, so there is nothing left to release here.
|
||||
return
|
||||
}
|
||||
for (const pending of [...this.units.values()]) {
|
||||
const unit = await pending.catch(() => undefined)
|
||||
await unit?.close()
|
||||
}
|
||||
db.close()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Register the SQLite backend as `sqlite` on the storage hub. The disposer
|
||||
* unregisters the name first, then closes the backend.
|
||||
* @param ctx - Plugin context (must inject `storage`).
|
||||
* @param config - Validated plugin configuration.
|
||||
*/
|
||||
export function apply(ctx: Context, config: Config) {
|
||||
const backend = new SqliteStorageBackend(config)
|
||||
ctx.effect(() => {
|
||||
const dispose = ctx.storage.backend.register('sqlite', backend)
|
||||
return async () => {
|
||||
dispose()
|
||||
await backend.close()
|
||||
}
|
||||
}, 'storage-sqlite.registerBackend')
|
||||
}
|
||||
32
packages/storage/storage-sqlite/src/invariant.ts
Normal file
32
packages/storage/storage-sqlite/src/invariant.ts
Normal file
@@ -0,0 +1,32 @@
|
||||
/**
|
||||
* Package-owned invariant companion for `@deepseek-ai/dsh-storage-sqlite`.
|
||||
* @module @deepseek-ai/dsh-storage-sqlite/invariant
|
||||
*/
|
||||
|
||||
/* jscpd:ignore-start */
|
||||
import type { Context } from 'cordis'
|
||||
import type { InvariantInstaller } from '@deepseek-ai/dsh-invariants'
|
||||
|
||||
const PACKAGE_NAME = '@deepseek-ai/dsh-storage-sqlite'
|
||||
|
||||
/** Cordis companion plugin name. */
|
||||
export const name = 'storage-sqlite-invariant'
|
||||
/** Service required before the companion can reserve package ownership. */
|
||||
export const inject = ['invariants']
|
||||
|
||||
/**
|
||||
* No runtime invariant: schema-version and unit-version consistency are
|
||||
* open-time checks that reject before a unit exists, and durability needs the
|
||||
* backend round-trip tests in the shared KV conformance suite; this package
|
||||
* exposes no continuously observable in-process relation.
|
||||
*/
|
||||
const install: InvariantInstaller = () => {}
|
||||
|
||||
/**
|
||||
* Register this package's invariant companion.
|
||||
* @param ctx - Cordis context carrying the invariant service.
|
||||
* @returns the installed registration's disposer after setup succeeds.
|
||||
*/
|
||||
export const apply = (ctx: Context): Promise<() => void> =>
|
||||
Promise.resolve(ctx.invariants.register(PACKAGE_NAME, install))
|
||||
/* jscpd:ignore-end */
|
||||
112
packages/storage/storage-sqlite/src/schema.ts
Normal file
112
packages/storage/storage-sqlite/src/schema.ts
Normal file
@@ -0,0 +1,112 @@
|
||||
/**
|
||||
* Schema + open-time helpers for the SQLite storage backend: the physical
|
||||
* layout version, the database open/configure sequence (permissions, pragmas,
|
||||
* version stamp/reject), and the unit metadata tables. Unit record tables are
|
||||
* created per descriptor in `unit.ts`.
|
||||
* @module @deepseek-ai/dsh-storage-sqlite/schema
|
||||
*/
|
||||
|
||||
import { DatabaseSync } from 'node:sqlite'
|
||||
import { mkdir, open } from 'node:fs/promises'
|
||||
import { dirname, resolve } from 'node:path'
|
||||
import { StorageError } from '@deepseek-ai/dsh-storage'
|
||||
|
||||
/**
|
||||
* The on-disk physical layout version, stored in `PRAGMA user_version`.
|
||||
* Orthogonal to each unit's own `version` (stamped per unit in the `units`
|
||||
* row). Bumped only on a breaking change to the table layout; any other
|
||||
* stamped version rejects — this unreleased format has no migrations.
|
||||
*/
|
||||
export const STORAGE_SQLITE_SCHEMA_VERSION = 1
|
||||
|
||||
/**
|
||||
* Journal modes the backend will run under. `wal` is the default; the
|
||||
* rollback-journal modes (`delete`/`truncate`/`persist`) exist for
|
||||
* filesystems where WAL's shared-memory files do not work (network mounts).
|
||||
* `memory`/`off` are excluded: dropping journal durability silently
|
||||
* contradicts the durability clause of the KV backend contract.
|
||||
*/
|
||||
export type JournalMode = 'wal' | 'delete' | 'truncate' | 'persist'
|
||||
|
||||
/**
|
||||
* Exclusively create a missing database file with owner-only permissions.
|
||||
* Existing files retain their modes, and errors other than `EEXIST` propagate.
|
||||
* `DatabaseSync` reopens by path, so this does not protect confidentiality or
|
||||
* integrity when another principal can replace the database entry in its
|
||||
* parent directory.
|
||||
*/
|
||||
async function createDatabaseFile(path: string): Promise<void> {
|
||||
try {
|
||||
const handle = await open(path, 'wx', 0o600)
|
||||
await handle.close()
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== 'EEXIST') throw error
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Open the database and apply its schema and pragmas. Missing directories and
|
||||
* database files are created owner-only (`:memory:` skips filesystem setup).
|
||||
* A zero `user_version` is stamped with {@link STORAGE_SQLITE_SCHEMA_VERSION};
|
||||
* every other non-current version rejects rather than being migrated in place.
|
||||
* @param path - the SQLite database file to open, or `:memory:`.
|
||||
* @param journalMode - validated journal pragma.
|
||||
* @returns the open handle with pragmas applied and the unit metadata tables ensured.
|
||||
*/
|
||||
export async function openDatabase(path: string, journalMode: JournalMode): Promise<DatabaseSync> {
|
||||
const actual = path === ':memory:' ? path : resolve(path)
|
||||
if (actual !== ':memory:') {
|
||||
await mkdir(dirname(actual), { recursive: true, mode: 0o700 })
|
||||
await createDatabaseFile(actual)
|
||||
}
|
||||
const db = new DatabaseSync(actual)
|
||||
try {
|
||||
configureDatabase(db, actual, journalMode)
|
||||
return db
|
||||
} catch (error: unknown) {
|
||||
db.close()
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
function configureDatabase(db: DatabaseSync, path: string, journalMode: JournalMode): void {
|
||||
db.exec('PRAGMA foreign_keys = ON')
|
||||
// The validated union is safe to interpolate into a non-bindable PRAGMA.
|
||||
db.exec(`PRAGMA journal_mode = ${journalMode.toUpperCase()}`)
|
||||
// `PRAGMA user_version` always returns exactly one row { user_version }.
|
||||
const { user_version: onDisk } = db.prepare('PRAGMA user_version').get() as { user_version: number }
|
||||
if (onDisk !== 0 && onDisk !== STORAGE_SQLITE_SCHEMA_VERSION) {
|
||||
throw new StorageError(
|
||||
'version-mismatch',
|
||||
`storage database at "${path}" has schema version ${onDisk}, incompatible with this build (${STORAGE_SQLITE_SCHEMA_VERSION})`,
|
||||
)
|
||||
}
|
||||
if (onDisk === 0) {
|
||||
// Stamp fresh databases.
|
||||
db.exec(`PRAGMA user_version = ${STORAGE_SQLITE_SCHEMA_VERSION}`)
|
||||
}
|
||||
db.exec(`
|
||||
CREATE TABLE IF NOT EXISTS units (
|
||||
name TEXT PRIMARY KEY,
|
||||
version INTEGER NOT NULL
|
||||
) STRICT
|
||||
`)
|
||||
db.exec(`
|
||||
CREATE TABLE IF NOT EXISTS unit_globals (
|
||||
unit TEXT PRIMARY KEY REFERENCES units(name),
|
||||
value TEXT NOT NULL
|
||||
) STRICT
|
||||
`)
|
||||
}
|
||||
|
||||
/**
|
||||
* Physical table name for one unit table. Both segments are validated against
|
||||
* `UNIT_NAME_RE` before reaching this, so the result is safe to interpolate
|
||||
* into DDL and prepared-statement text.
|
||||
* @param unit - Validated unit name.
|
||||
* @param table - Validated table name.
|
||||
* @returns the `u_<unit>_<table>` identifier.
|
||||
*/
|
||||
export function recordTableName(unit: string, table: string): string {
|
||||
return `u_${unit}_${table}`
|
||||
}
|
||||
120
packages/storage/storage-sqlite/src/unit.ts
Normal file
120
packages/storage/storage-sqlite/src/unit.ts
Normal file
@@ -0,0 +1,120 @@
|
||||
/**
|
||||
* One opened SQLite KV unit: prepared per-table statements over the
|
||||
* `u_<unit>_<table>` record tables plus this unit's row in the shared
|
||||
* `unit_globals` table. Each primitive is a single statement, so atomicity
|
||||
* comes from SQLite itself — no explicit transactions, and no write queue
|
||||
* (write ordering is the caller's responsibility per the KV contract).
|
||||
* @module @deepseek-ai/dsh-storage-sqlite/unit
|
||||
*/
|
||||
|
||||
import type { DatabaseSync, StatementSync } from 'node:sqlite'
|
||||
import { StorageError } from '@deepseek-ai/dsh-storage'
|
||||
import type { KvUnit, KvUnitDescriptor } from '@deepseek-ai/dsh-storage'
|
||||
import { recordTableName } from './schema.ts'
|
||||
|
||||
/** Prepared statements for one declared table. */
|
||||
interface TableStatements {
|
||||
upsert: StatementSync
|
||||
remove: StatementSync
|
||||
selectAll: StatementSync
|
||||
}
|
||||
|
||||
/**
|
||||
* The SQLite {@link KvUnit}. Constructed by the backend AFTER the unit's
|
||||
* record tables exist; statements are prepared once here and reused for every
|
||||
* primitive. Values are stored as JSON text in the `value` column.
|
||||
*/
|
||||
export class SqliteKvUnit implements KvUnit {
|
||||
private readonly tables = new Map<string, TableStatements>()
|
||||
private readonly globalUpsert: StatementSync | undefined
|
||||
private readonly globalSelect: StatementSync | undefined
|
||||
private closed = false
|
||||
|
||||
/**
|
||||
* @param db - Open database handle owned by the backend (never closed here).
|
||||
* @param descriptor - Validated descriptor whose record tables already exist.
|
||||
* @param onClose - Backend callback releasing this unit's open-name slot.
|
||||
*/
|
||||
constructor(
|
||||
db: DatabaseSync,
|
||||
private readonly descriptor: KvUnitDescriptor,
|
||||
private readonly onClose: () => void,
|
||||
) {
|
||||
for (const table of descriptor.tables) {
|
||||
// Both name segments are validated against UNIT_NAME_RE by the backend,
|
||||
// so the physical identifier is safe to interpolate into statement text.
|
||||
const physical = recordTableName(descriptor.name, table)
|
||||
this.tables.set(table, {
|
||||
upsert: db.prepare(
|
||||
`INSERT INTO "${physical}" (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value`,
|
||||
),
|
||||
remove: db.prepare(`DELETE FROM "${physical}" WHERE key = ?`),
|
||||
selectAll: db.prepare(`SELECT key, value FROM "${physical}"`),
|
||||
})
|
||||
}
|
||||
this.globalUpsert = descriptor.hasGlobal
|
||||
? db.prepare(
|
||||
'INSERT INTO unit_globals (unit, value) VALUES (?, ?) ON CONFLICT(unit) DO UPDATE SET value = excluded.value',
|
||||
)
|
||||
: undefined
|
||||
this.globalSelect = descriptor.hasGlobal
|
||||
? db.prepare('SELECT value FROM unit_globals WHERE unit = ?')
|
||||
: undefined
|
||||
}
|
||||
|
||||
async loadAll(): Promise<{ tables: Record<string, Record<string, unknown>>; global: unknown | null }> {
|
||||
this.ensureOpen()
|
||||
const tables: Record<string, Record<string, unknown>> = {}
|
||||
for (const [name, statements] of this.tables) {
|
||||
const records: Record<string, unknown> = {}
|
||||
for (const row of statements.selectAll.all() as unknown as Array<{ key: string; value: string }>) {
|
||||
records[row.key] = JSON.parse(row.value)
|
||||
}
|
||||
tables[name] = records
|
||||
}
|
||||
let global: unknown = null
|
||||
if (this.globalSelect !== undefined) {
|
||||
const row = this.globalSelect.get(this.descriptor.name) as { value: string } | undefined
|
||||
if (row !== undefined) global = JSON.parse(row.value)
|
||||
}
|
||||
return { tables, global }
|
||||
}
|
||||
|
||||
async putRecord(table: string, key: string, value: unknown): Promise<void> {
|
||||
this.ensureOpen()
|
||||
this.statementsFor(table).upsert.run(key, JSON.stringify(value))
|
||||
}
|
||||
|
||||
async deleteRecord(table: string, key: string): Promise<void> {
|
||||
this.ensureOpen()
|
||||
this.statementsFor(table).remove.run(key)
|
||||
}
|
||||
|
||||
async setGlobal(value: unknown): Promise<void> {
|
||||
this.ensureOpen()
|
||||
if (this.globalUpsert === undefined) {
|
||||
throw new Error(`kv unit '${this.descriptor.name}' declared no global slot`)
|
||||
}
|
||||
this.globalUpsert.run(this.descriptor.name, JSON.stringify(value))
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
if (this.closed) return
|
||||
this.closed = true
|
||||
this.onClose()
|
||||
}
|
||||
|
||||
private ensureOpen(): void {
|
||||
if (this.closed) {
|
||||
throw new StorageError('closed', `kv unit '${this.descriptor.name}' is closed`)
|
||||
}
|
||||
}
|
||||
|
||||
private statementsFor(table: string): TableStatements {
|
||||
const statements = this.tables.get(table)
|
||||
if (statements === undefined) {
|
||||
throw new Error(`kv unit '${this.descriptor.name}' declared no table '${table}'`)
|
||||
}
|
||||
return statements
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user