diff --git a/apps/server/src/modules/agent/conversation/prompt/debug-prompt.ts b/apps/server/src/modules/agent/conversation/prompt/debug-prompt.ts index b923b209..c5649628 100644 --- a/apps/server/src/modules/agent/conversation/prompt/debug-prompt.ts +++ b/apps/server/src/modules/agent/conversation/prompt/debug-prompt.ts @@ -153,6 +153,9 @@ type SubstrateResolution = /** Where this module keeps one log per thread on a Disk substrate. */ function diskLogPath(substrate: SpaceSubstrate, threadId: string): string { + if (substrate.kind !== 'disk') { + throw new Error('Debug prompt logs require a Disk extension substrate'); + } return path.join( substrate.directory, `${sanitizeId(threadId, 'threadId')}.prompt.log`, diff --git a/apps/server/src/modules/agent/memory/trigger.ts b/apps/server/src/modules/agent/memory/trigger.ts index f2d4b1ce..28b8eb2e 100644 --- a/apps/server/src/modules/agent/memory/trigger.ts +++ b/apps/server/src/modules/agent/memory/trigger.ts @@ -45,6 +45,9 @@ const MEMORY_NAMESPACE = 'huabu.memory'; * the substrate, never a port member. */ function diskStatePath(substrate: SpaceSubstrate): string { + if (substrate.kind !== 'disk') { + throw new Error('Memory state requires a Disk extension substrate'); + } return path.join(substrate.directory, 'state.json'); } diff --git a/apps/server/src/modules/canvas/canvas.route.test.ts b/apps/server/src/modules/canvas/canvas.route.test.ts index fbe27685..1cbbee60 100644 --- a/apps/server/src/modules/canvas/canvas.route.test.ts +++ b/apps/server/src/modules/canvas/canvas.route.test.ts @@ -769,7 +769,9 @@ describe('Space export/import persistence', () => { createCanvas('c1', 'Private Export'); const promptStore = await space('c1').extension('huabu.prompt.log'); const memoryStore = await space('c1').extension('huabu.memory'); - if (!promptStore || !memoryStore) throw new Error('Expected Disk stores'); + if (promptStore?.kind !== 'disk' || memoryStore?.kind !== 'disk') { + throw new Error('Expected Disk stores'); + } writeFileSync( join(promptStore.directory, 'thread.prompt.log'), 'private system and user prompt', @@ -806,7 +808,7 @@ describe('Space export/import persistence', () => { const importedPrompt = await space(importedId).extension('huabu.prompt.log'); const importedMemory = await space(importedId).extension('huabu.memory'); - if (!importedPrompt || !importedMemory) { + if (importedPrompt?.kind !== 'disk' || importedMemory?.kind !== 'disk') { throw new Error('Expected imported Disk stores'); } expect( diff --git a/apps/server/src/modules/canvas/persistence-validation.ts b/apps/server/src/modules/canvas/persistence-validation.ts new file mode 100644 index 00000000..20b3613f --- /dev/null +++ b/apps/server/src/modules/canvas/persistence-validation.ts @@ -0,0 +1,45 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +/** Runtime validation shared by structured storage adapters. */ + +function finiteNumber(value: unknown): boolean { + return typeof value === 'number' && Number.isFinite(value); +} + +/** Return the first minimal CanvasFile shape violation, if any. */ +export function canvasFileShapeError( + value: unknown, + expectedCanvasId: string, +): string | null { + if (typeof value !== 'object' || value === null || Array.isArray(value)) { + return 'must be an object'; + } + + const record = value as Record; + if (record['canvasId'] !== expectedCanvasId) { + return `canvasId must equal ${JSON.stringify(expectedCanvasId)}`; + } + if (record['title'] !== null && typeof record['title'] !== 'string') { + return 'title must be a string or null'; + } + if (!finiteNumber(record['version'])) + return 'version must be a finite number'; + if (!finiteNumber(record['createdAt'])) { + return 'createdAt must be a finite number'; + } + if (!finiteNumber(record['updatedAt'])) { + return 'updatedAt must be a finite number'; + } + + const state = record['state']; + if (typeof state !== 'object' || state === null || Array.isArray(state)) { + return 'state must be an object'; + } + const stateRecord = state as Record; + if (!Array.isArray(stateRecord['nodes'])) + return 'state.nodes must be an array'; + if (!Array.isArray(stateRecord['edges'])) + return 'state.edges must be an array'; + return null; +} diff --git a/apps/server/src/modules/storage/backends/disk/space-nodes.test.ts b/apps/server/src/modules/storage/backends/disk/space-nodes.test.ts index 19cc23b7..ebf7d676 100644 --- a/apps/server/src/modules/storage/backends/disk/space-nodes.test.ts +++ b/apps/server/src/modules/storage/backends/disk/space-nodes.test.ts @@ -54,10 +54,12 @@ describeSpaceNodesContract('Disk', async () => { if (!created.ok) throw new Error('Node contract Space already exists'); const store = new DiskStructuredStore(); + const space = store.space('node-space'); return { - repository: store.space('node-space').nodes, + repository: space.nodes, missingRepository: store.space('missing-node-space').nodes, expectedCanvasId: 'node-space', + deletedNodePut: 'write-suppressed', cleanup: () => { vi.restoreAllMocks(); resetStorageCache(); diff --git a/apps/server/src/modules/storage/backends/disk/space-record-validation.ts b/apps/server/src/modules/storage/backends/disk/space-record-validation.ts index 1792458c..e31bb697 100644 --- a/apps/server/src/modules/storage/backends/disk/space-record-validation.ts +++ b/apps/server/src/modules/storage/backends/disk/space-record-validation.ts @@ -1,55 +1,14 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT license. -/** Runtime validation shared by strict Disk Space-record boundaries. */ +/** Runtime validation and strict reads for Disk Space-record boundaries. */ import { readJsonStrict } from '../../../../utils/fs.js'; +import { canvasFileShapeError } from '../../../canvas/persistence-validation.js'; import type { CanvasFile } from '../../../canvas/persistence-types.js'; -function finiteNumber(value: unknown): boolean { - return typeof value === 'number' && Number.isFinite(value); -} - -/** Return the first minimal {@link CanvasFile} shape violation, if any. */ -export function canvasFileShapeError( - value: unknown, - expectedCanvasId: string, -): string | null { - if (typeof value !== 'object' || value === null || Array.isArray(value)) { - return 'must be an object'; - } - - const record = value as Record; - if (record['canvasId'] !== expectedCanvasId) { - return `canvasId must equal ${JSON.stringify(expectedCanvasId)}`; - } - if (record['title'] !== null && typeof record['title'] !== 'string') { - return 'title must be a string or null'; - } - if (!finiteNumber(record['version'])) { - return 'version must be a finite number'; - } - if (!finiteNumber(record['createdAt'])) { - return 'createdAt must be a finite number'; - } - if (!finiteNumber(record['updatedAt'])) { - return 'updatedAt must be a finite number'; - } - - const state = record['state']; - if (typeof state !== 'object' || state === null || Array.isArray(state)) { - return 'state must be an object'; - } - const stateRecord = state as Record; - if (!Array.isArray(stateRecord['nodes'])) { - return 'state.nodes must be an array'; - } - if (!Array.isArray(stateRecord['edges'])) { - return 'state.edges must be an array'; - } - return null; -} +export { canvasFileShapeError } from '../../../canvas/persistence-validation.js'; /** * Strictly read and validate one indexed `space.json` path. diff --git a/apps/server/src/modules/storage/backends/disk/space-repository.test.ts b/apps/server/src/modules/storage/backends/disk/space-repository.test.ts index 250a6102..033cbb0b 100644 --- a/apps/server/src/modules/storage/backends/disk/space-repository.test.ts +++ b/apps/server/src/modules/storage/backends/disk/space-repository.test.ts @@ -141,9 +141,12 @@ describeSpaceExtensionContract('Disk', () => { // An owner of a Disk namespace writes files into its directory; nothing // about the shape is storage's business, so the suite borrows the // simplest one an owner could pick. - write: (substrate, value) => - writeFileSync(path.join(substrate.directory, 'value'), value, 'utf8'), + write: (substrate, value) => { + if (substrate.kind !== 'disk') throw new Error('Expected Disk substrate'); + writeFileSync(path.join(substrate.directory, 'value'), value, 'utf8'); + }, read: (substrate) => { + if (substrate.kind !== 'disk') throw new Error('Expected Disk substrate'); const file = path.join(substrate.directory, 'value'); return existsSync(file) ? readFileSync(file, 'utf8') : null; }, diff --git a/apps/server/src/modules/storage/backends/disk/structured-store.test.ts b/apps/server/src/modules/storage/backends/disk/structured-store.test.ts index 0f3dadd8..ef0ad077 100644 --- a/apps/server/src/modules/storage/backends/disk/structured-store.test.ts +++ b/apps/server/src/modules/storage/backends/disk/structured-store.test.ts @@ -28,6 +28,7 @@ import { import { DiskStructuredStore } from './structured-store.js'; import { toSafeFilename } from '../../../../utils/naming.js'; import { describeSpaceLogsContract } from '../../ports/contracts/space-logs.contract.js'; +import { describeSpaceTasksContract } from '../../ports/contracts/space-tasks.contract.js'; import { describeStructuredStoreContract } from '../../ports/contracts/structured-store.contract.js'; import type { CanvasFile } from '../../../canvas/persistence-types.js'; @@ -111,6 +112,8 @@ describe('Disk Space extension workspace binding', () => { try { const substrate = await pending; expect(substrate?.kind).toBe('disk'); + if (substrate?.kind !== 'disk') + throw new Error('Expected Disk substrate'); expect(substrate?.directory.startsWith(`${firstRoot}${path.sep}`)).toBe( true, ); @@ -125,6 +128,30 @@ describe('Disk Space extension workspace binding', () => { }); }); +describeSpaceTasksContract('Disk', () => { + const root = freshWorkspace('huabu-task-contract-'); + seedSpace(root, 'canvas-task', 'Canvas Task'); + const store = new DiskStructuredStore(); + return { + tasks: store.space('canvas-task').tasks, + concurrent: store.space('canvas-task').tasks, + canvasId: 'canvas-task', + missing: store.space('missing-canvas').tasks, + missingCanvasId: 'missing-canvas', + beginDelete: async () => { + const result = await store.spaces().beginDelete({ + canvasId: 'canvas-task', + }); + if (!result.ok) throw new Error('Ordinary Space must be deletable'); + return result.session; + }, + cleanup: () => { + resetStorageCache(); + rmSync(root, { recursive: true, force: true }); + }, + }; +}); + describe('Disk Space Tasks', () => { let root = ''; let store: DiskStructuredStore; @@ -141,132 +168,8 @@ describe('Disk Space Tasks', () => { rmSync(root, { recursive: true, force: true }); }); - it('serializes Task and Run mutations across independent handles', async () => { - const first = store.space('canvas-task').tasks; - const second = store.space('canvas-task').tasks; - await Promise.all([ - first.create({ - taskId: 'task-a', - canvasId: 'canvas-task', - goal: 'Goal A', - defaultRootProfileId: 'profile-a', - anchorNodeId: 'node-a', - createdAt: 1, - }), - second.create({ - taskId: 'task-b', - canvasId: 'canvas-task', - goal: 'Goal B', - defaultRootProfileId: 'profile-b', - anchorNodeId: 'node-b', - createdAt: 2, - }), - ]); - await first.runs.create({ - runId: 'run-a', - taskId: 'task-a', - canvasIdSnapshot: 'canvas-task', - goalSnapshot: 'Goal A', - rootProfileIdSnapshot: 'profile-a', - status: 'pending', - createdAt: 3, - }); - const updated = await second.runs.update('run-a', { - rootNodeId: 'node-root', - rootThreadId: 'thread-root', - status: 'running', - startedAt: 4, - }); - - expect(updated.status).toBe('running'); - await expect(first.read()).resolves.toMatchObject({ - version: 1, - tasks: [ - expect.objectContaining({ taskId: 'task-a' }), - expect.objectContaining({ taskId: 'task-b' }), - ], - runs: [ - expect.objectContaining({ - runId: 'run-a', - rootNodeId: 'node-root', - rootThreadId: 'thread-root', - }), - ], - }); - }); - - it('returns an empty versioned snapshot when no Task store exists', async () => { - await expect(store.space('canvas-empty').tasks.read()).resolves.toEqual({ - version: 1, - tasks: [], - runs: [], - }); - }); - - it('completes a running Run atomically and keeps its message immutable', async () => { - const runs = store.space('canvas-task').tasks.runs; - await expect( - runs.complete('task-a', 'run-a', { - completedAt: 5, - message: 'PR merged', - }), - ).resolves.toMatchObject({ - outcome: 'completed', - run: { - status: 'completed', - completion: { completedAt: 5, message: 'PR merged' }, - }, - }); - await expect( - runs.complete('task-a', 'run-a', { - completedAt: 6, - message: 'PR merged', - }), - ).resolves.toMatchObject({ - outcome: 'unchanged', - run: { completion: { completedAt: 5, message: 'PR merged' } }, - }); - await expect( - runs.complete('task-a', 'run-a', { - completedAt: 7, - message: 'Different result', - }), - ).resolves.toMatchObject({ outcome: 'completion_conflict' }); - - await runs.create({ - runId: 'run-pending', - taskId: 'task-b', - canvasIdSnapshot: 'canvas-task', - goalSnapshot: 'Goal B', - rootProfileIdSnapshot: 'profile-b', - status: 'pending', - createdAt: 8, - }); - await expect( - runs.complete('task-b', 'run-pending', { completedAt: 9 }), - ).resolves.toMatchObject({ outcome: 'run_not_running' }); - await expect( - runs.complete('missing-task', 'run-a', { completedAt: 9 }), - ).resolves.toEqual({ outcome: 'task_not_found' }); - await expect( - runs.complete('task-a', 'missing-run', { completedAt: 9 }), - ).resolves.toEqual({ outcome: 'run_not_found' }); - }); - - it('rejects mutations for a missing Space', async () => { - await expect( - store.space('missing-canvas').tasks.create({ - taskId: 'task-missing', - canvasId: 'missing-canvas', - goal: 'Missing', - defaultRootProfileId: 'profile-a', - anchorNodeId: 'node-missing', - createdAt: 1, - }), - ).rejects.toThrow(/cannot write a missing Space/); - }); - it('fails fast on malformed and internally inconsistent Task stores', async () => { + mkdirSync(path.dirname(tasksPath('canvas-task')), { recursive: true }); writeFileSync(tasksPath('canvas-task'), '{"version":1,"tasks":{}}'); await expect(store.space('canvas-task').tasks.read()).rejects.toThrow( /Invalid Task store/, diff --git a/apps/server/src/modules/storage/backends/sqlite/contracts.test.ts b/apps/server/src/modules/storage/backends/sqlite/contracts.test.ts new file mode 100644 index 00000000..0de7391b --- /dev/null +++ b/apps/server/src/modules/storage/backends/sqlite/contracts.test.ts @@ -0,0 +1,198 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +import { SqliteStructuredStore } from './structured-store.js'; +import { + createSqliteTestFile, + installDeltaAbortTrigger, + openEmptySqliteTestStore, + openSqliteTestStore, + readSqliteDeltaLog, +} from './test-support.js'; +import { describeSpaceExtensionContract } from '../../ports/contracts/space-extension.contract.js'; +import { describeSpaceLogsContract } from '../../ports/contracts/space-logs.contract.js'; +import { describeSpaceNodesContract } from '../../ports/contracts/space-nodes.contract.js'; +import { describeSpaceRepositoryContract } from '../../ports/contracts/space-repository.contract.js'; +import { describeSpaceTasksContract } from '../../ports/contracts/space-tasks.contract.js'; +import { describeSpaceWriteContract } from '../../ports/contracts/space-write.contract.js'; +import { describeStructuredStoreContract } from '../../ports/contracts/structured-store.contract.js'; + +import type { NodeContent } from '../../../canvas/persistence-types.js'; + +function note(nodeId: string, label: string, content: string): NodeContent { + return { nodeId, type: 'note', label, content }; +} + +async function createOrdinarySpace( + store: SqliteStructuredStore, + canvasId: string, + title: string, +): Promise { + const created = await store.spaces().create({ canvasId, title }); + if (!created.ok) throw new Error(`Could not create test Space ${canvasId}`); +} + +describeStructuredStoreContract('SQLite', () => { + const file = createSqliteTestFile('huabu-sqlite-structured-contract-'); + return { + store: new SqliteStructuredStore(file.filename), + cleanup: file.remove, + }; +}); + +describeSpaceRepositoryContract('SQLite', async () => { + const harness = await openSqliteTestStore( + 'huabu-sqlite-space-repository-contract-', + ); + const emptyStores: Array< + Awaited> + > = []; + return { + repository: harness.store.spaces(), + read: (canvasId: string) => harness.store.space(canvasId).read(), + worldCanvasId: harness.world.canvasId, + attemptMutation: (canvasId: string) => + harness.store.space(canvasId).nodes.put({ + nodeId: 'contract-delete-fence-node', + record: note( + 'contract-delete-fence-node', + 'Deletion fence node', + 'body', + ), + }), + openEmptyNamespace: async () => { + const empty = await openEmptySqliteTestStore( + 'huabu-sqlite-empty-namespace-contract-', + ); + emptyStores.push(empty); + return { + repository: empty.store.spaces(), + read: (canvasId: string) => empty.store.space(canvasId).read(), + }; + }, + cleanup: async () => { + for (const empty of emptyStores.splice(0)) await empty.cleanup(); + await harness.cleanup(); + }, + }; +}); + +describeSpaceNodesContract('SQLite', async () => { + const harness = await openSqliteTestStore('huabu-sqlite-nodes-contract-'); + const canvasId = 'sqlite-nodes-contract'; + await createOrdinarySpace(harness.store, canvasId, 'SQLite Nodes Contract'); + const space = harness.store.space(canvasId); + return { + repository: space.nodes, + missingRepository: harness.store.space('sqlite-nodes-missing').nodes, + expectedCanvasId: canvasId, + deletedNodePut: 'allowed', + cleanup: harness.cleanup, + }; +}); + +describeSpaceExtensionContract('SQLite', async () => { + const harness = await openSqliteTestStore('huabu-sqlite-extension-contract-'); + const table = 'contract_extension_values'; + return { + repository: harness.store.spaces(), + space: (canvasId: string) => harness.store.space(canvasId), + write: (substrate, value: string) => { + if (substrate.kind !== 'sqlite') { + throw new Error('Expected a SQLite substrate'); + } + substrate.database.exec( + `CREATE TABLE IF NOT EXISTS ${table} ( + extension_id INTEGER PRIMARY KEY, + value TEXT NOT NULL, + FOREIGN KEY (extension_id) REFERENCES space_extensions(extension_id) + ON DELETE CASCADE + ) STRICT`, + ); + substrate.database + .prepare( + `INSERT INTO ${table} (extension_id, value) VALUES (?, ?) + ON CONFLICT(extension_id) DO UPDATE SET value = excluded.value`, + ) + .run(substrate.extensionId, value); + }, + read: (substrate) => { + if (substrate.kind !== 'sqlite') { + throw new Error('Expected a SQLite substrate'); + } + const row = substrate.database + .prepare(`SELECT value FROM ${table} WHERE extension_id = ?`) + .get(substrate.extensionId); + return typeof row?.['value'] === 'string' ? row['value'] : null; + }, + cleanup: harness.cleanup, + }; +}); + +describeSpaceWriteContract('SQLite', async () => { + const harness = await openSqliteTestStore('huabu-sqlite-write-contract-'); + const canvasId = 'sqlite-write-contract'; + await createOrdinarySpace(harness.store, canvasId, 'SQLite Write Contract'); + const existingNode = note( + 'contract-existing-node', + 'Existing contract node', + 'before', + ); + const space = harness.store.space(canvasId); + const put = await space.nodes.put({ + nodeId: existingNode.nodeId, + record: existingNode, + }); + if (!put.ok) { + throw new Error(`Could not seed SQLite write contract: ${put.reason}`); + } + + return { + space, + concurrent: harness.store.space(canvasId), + missing: harness.store.space('sqlite-write-missing'), + existingNode, + newNode: note('contract-new-node', 'New contract node', 'after'), + readJournal: async () => readSqliteDeltaLog(harness.filename, canvasId), + failNextDeltaAppend: (error: Error) => + installDeltaAbortTrigger(harness.filename, error.message), + cleanup: harness.cleanup, + }; +}); + +describeSpaceLogsContract('SQLite', async () => { + const harness = await openSqliteTestStore('huabu-sqlite-logs-contract-'); + const canvasId = 'sqlite-logs-contract'; + await createOrdinarySpace(harness.store, canvasId, 'SQLite Logs Contract'); + const first = harness.store.space(canvasId); + const second = harness.store.space(canvasId); + return { + events: first.events, + changes: first.changes, + concurrent: { + events: second.events, + changes: second.changes, + }, + cleanup: harness.cleanup, + }; +}); + +describeSpaceTasksContract('SQLite', async () => { + const harness = await openSqliteTestStore('huabu-sqlite-tasks-contract-'); + const canvasId = 'sqlite-tasks-contract'; + const missingCanvasId = 'sqlite-tasks-missing'; + await createOrdinarySpace(harness.store, canvasId, 'SQLite Tasks Contract'); + return { + tasks: harness.store.space(canvasId).tasks, + concurrent: harness.store.space(canvasId).tasks, + canvasId, + missing: harness.store.space(missingCanvasId).tasks, + missingCanvasId, + beginDelete: async () => { + const result = await harness.store.spaces().beginDelete({ canvasId }); + if (!result.ok) throw new Error('Ordinary Space must be deletable'); + return result.session; + }, + cleanup: harness.cleanup, + }; +}); diff --git a/apps/server/src/modules/storage/backends/sqlite/database.ts b/apps/server/src/modules/storage/backends/sqlite/database.ts new file mode 100644 index 00000000..acd0e4af --- /dev/null +++ b/apps/server/src/modules/storage/backends/sqlite/database.ts @@ -0,0 +1,258 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +import { DatabaseSync } from 'node:sqlite'; + +import { + assertSpaceMutationAllowed, + beginSpaceDeleteAdmission, +} from '../../space-lifecycle-admission.js'; + +import type { StorageHealth } from '../../ports/common.js'; + +export const SQLITE_SCHEMA_VERSION = 1; +export const SQLITE_WORLD_COLLISION_KEY = '.world'; + +const SCHEMA_V1 = ` + CREATE TABLE spaces ( + canvas_id TEXT PRIMARY KEY, + title TEXT, + collision_key TEXT NOT NULL UNIQUE, + version INTEGER NOT NULL, + state_json TEXT NOT NULL CHECK (json_valid(state_json)), + created_at REAL NOT NULL, + updated_at REAL NOT NULL, + is_world INTEGER NOT NULL DEFAULT 0 CHECK (is_world IN (0, 1)) + ) STRICT; + + CREATE UNIQUE INDEX spaces_single_world + ON spaces(is_world) + WHERE is_world = 1; + + CREATE TABLE nodes ( + canvas_id TEXT NOT NULL, + node_id TEXT NOT NULL, + record_json TEXT NOT NULL CHECK (json_valid(record_json)), + revision TEXT NOT NULL CHECK (length(revision) > 0), + label_collision_key TEXT NOT NULL, + PRIMARY KEY (canvas_id, node_id), + UNIQUE (canvas_id, label_collision_key), + FOREIGN KEY (canvas_id) REFERENCES spaces(canvas_id) ON DELETE CASCADE + ) STRICT; + + CREATE TABLE events ( + event_id INTEGER PRIMARY KEY AUTOINCREMENT, + canvas_id TEXT NOT NULL, + event_json TEXT NOT NULL CHECK (json_valid(event_json)), + FOREIGN KEY (canvas_id) REFERENCES spaces(canvas_id) ON DELETE CASCADE + ) STRICT; + + CREATE INDEX events_by_canvas_order + ON events(canvas_id, event_id); + + CREATE TABLE changes ( + canvas_id TEXT NOT NULL, + thread_id TEXT NOT NULL, + snapshot_json TEXT NOT NULL CHECK (json_valid(snapshot_json)), + PRIMARY KEY (canvas_id, thread_id), + FOREIGN KEY (canvas_id) REFERENCES spaces(canvas_id) ON DELETE CASCADE + ) STRICT; + + CREATE TABLE tasks ( + canvas_id TEXT PRIMARY KEY, + snapshot_json TEXT NOT NULL CHECK (json_valid(snapshot_json)), + FOREIGN KEY (canvas_id) REFERENCES spaces(canvas_id) ON DELETE CASCADE + ) STRICT; + + CREATE TABLE space_extensions ( + extension_id INTEGER PRIMARY KEY AUTOINCREMENT, + canvas_id TEXT NOT NULL, + namespace TEXT NOT NULL, + UNIQUE (canvas_id, namespace), + FOREIGN KEY (canvas_id) REFERENCES spaces(canvas_id) ON DELETE CASCADE + ) STRICT; + + CREATE TABLE delta_log ( + canvas_id TEXT NOT NULL, + version INTEGER NOT NULL, + entry_json TEXT NOT NULL CHECK (json_valid(entry_json)), + PRIMARY KEY (canvas_id, version), + FOREIGN KEY (canvas_id) REFERENCES spaces(canvas_id) ON DELETE CASCADE + ) STRICT; +`; + +export interface SqliteMigration { + readonly version: number; + readonly sql: string; +} + +export const SQLITE_MIGRATIONS: readonly SqliteMigration[] = Object.freeze([ + Object.freeze({ version: 1, sql: SCHEMA_V1 }), +]); + +function readUserVersion(database: DatabaseSync): number { + const row = database.prepare('PRAGMA user_version').get(); + const version = row?.['user_version']; + if (typeof version !== 'number' || !Number.isSafeInteger(version)) { + throw new Error('SQLite returned an invalid PRAGMA user_version'); + } + return version; +} + +export function applySqliteMigrations( + database: DatabaseSync, + migrations: readonly SqliteMigration[] = SQLITE_MIGRATIONS, +): void { + for (let index = 0; index < migrations.length; index += 1) { + const expectedVersion = index + 1; + if (migrations[index]?.version !== expectedVersion) { + throw new Error( + `SQLite migrations must be contiguous from version 1; expected ${expectedVersion}`, + ); + } + } + const targetVersion = migrations.at(-1)?.version ?? 0; + const current = readUserVersion(database); + if (current > targetVersion) { + throw new Error( + `SQLite schema version ${current} is newer than supported version ${targetVersion}`, + ); + } + if (current === targetVersion) return; + + database.exec('BEGIN IMMEDIATE'); + try { + let version = readUserVersion(database); + for (const migration of migrations) { + if (migration.version <= version) continue; + if (migration.version !== version + 1) { + throw new Error( + `No SQLite migration path from schema version ${version} to ${targetVersion}`, + ); + } + database.exec(migration.sql); + database.exec(`PRAGMA user_version = ${migration.version}`); + version = migration.version; + } + if (version !== targetVersion) { + throw new Error( + `No SQLite migration path from schema version ${version} to ${targetVersion}`, + ); + } + database.exec('COMMIT'); + } catch (error) { + if (database.isTransaction) database.exec('ROLLBACK'); + throw error; + } +} + +/** One connection and all adapter-lifetime process-local state. */ +export class SqliteStoreContext { + readonly now: () => number; + + readonly #database: DatabaseSync; + readonly #admissionScope: string; + #state: 'new' | 'open' | 'closed' = 'new'; + + constructor(filename: string, now: () => number) { + this.now = now; + this.#admissionScope = `sqlite:${filename}`; + this.#database = new DatabaseSync(filename, { open: false }); + } + + init(): void { + if (this.#state === 'open') return; + if (this.#state === 'closed') { + throw new Error('SQLite store is closed'); + } + + try { + this.#database.open(); + this.#database.exec('PRAGMA foreign_keys = ON'); + const foreignKeys = this.#database.prepare('PRAGMA foreign_keys').get()?.[ + 'foreign_keys' + ]; + if (foreignKeys !== 1) { + throw new Error('Could not enable SQLite foreign key enforcement'); + } + applySqliteMigrations(this.#database); + this.#state = 'open'; + } catch (error) { + if (this.#database.isOpen) this.#database.close(); + this.#state = 'closed'; + throw error; + } + } + + health(kind: string): StorageHealth { + this.assertOpen(); + try { + const value = this.#database.prepare('SELECT 1 AS ok').get()?.['ok']; + return value === 1 + ? { ok: true, kind } + : { ok: false, kind, detail: 'SQLite liveness query returned no row' }; + } catch (error) { + return { + ok: false, + kind, + detail: error instanceof Error ? error.message : String(error), + }; + } + } + + close(): void { + if (this.#state === 'closed') return; + this.#state = 'closed'; + if (this.#database.isOpen) this.#database.close(); + } + + database(): DatabaseSync { + this.assertOpen(); + return this.#database; + } + + assertOpen(): void { + if (this.#state !== 'open') { + throw new Error( + this.#state === 'closed' + ? 'SQLite store is closed' + : 'SQLite store is not initialized', + ); + } + } + + assertMutationAllowed(canvasId: string): void { + this.assertOpen(); + assertSpaceMutationAllowed(this.#admissionScope, canvasId); + } + + async acquireDelete(canvasId: string): Promise<() => void> { + this.assertOpen(); + const releaseGate = await beginSpaceDeleteAdmission( + this.#admissionScope, + canvasId, + ); + try { + this.assertOpen(); + } catch (error) { + releaseGate(); + throw error; + } + return releaseGate; + } +} + +export function withImmediateTransaction( + database: DatabaseSync, + operation: () => T, +): T { + database.exec('BEGIN IMMEDIATE'); + try { + const result = operation(); + database.exec('COMMIT'); + return result; + } catch (error) { + if (database.isTransaction) database.exec('ROLLBACK'); + throw error; + } +} diff --git a/apps/server/src/modules/storage/backends/sqlite/fixtures/v1.sql b/apps/server/src/modules/storage/backends/sqlite/fixtures/v1.sql new file mode 100644 index 00000000..467674c0 --- /dev/null +++ b/apps/server/src/modules/storage/backends/sqlite/fixtures/v1.sql @@ -0,0 +1,117 @@ +-- Immutable SQLite structured-store schema v1 fixture. +-- Add a new fixture for later schema versions; do not rewrite this history. + +PRAGMA foreign_keys = ON; +BEGIN IMMEDIATE; + +CREATE TABLE spaces ( + canvas_id TEXT PRIMARY KEY, + title TEXT, + collision_key TEXT NOT NULL UNIQUE, + version INTEGER NOT NULL, + state_json TEXT NOT NULL CHECK (json_valid(state_json)), + created_at REAL NOT NULL, + updated_at REAL NOT NULL, + is_world INTEGER NOT NULL DEFAULT 0 CHECK (is_world IN (0, 1)) +) STRICT; + +CREATE UNIQUE INDEX spaces_single_world + ON spaces(is_world) + WHERE is_world = 1; + +CREATE TABLE nodes ( + canvas_id TEXT NOT NULL, + node_id TEXT NOT NULL, + record_json TEXT NOT NULL CHECK (json_valid(record_json)), + revision TEXT NOT NULL CHECK (length(revision) > 0), + label_collision_key TEXT NOT NULL, + PRIMARY KEY (canvas_id, node_id), + UNIQUE (canvas_id, label_collision_key), + FOREIGN KEY (canvas_id) REFERENCES spaces(canvas_id) ON DELETE CASCADE +) STRICT; + +CREATE TABLE events ( + event_id INTEGER PRIMARY KEY AUTOINCREMENT, + canvas_id TEXT NOT NULL, + event_json TEXT NOT NULL CHECK (json_valid(event_json)), + FOREIGN KEY (canvas_id) REFERENCES spaces(canvas_id) ON DELETE CASCADE +) STRICT; + +CREATE INDEX events_by_canvas_order + ON events(canvas_id, event_id); + +CREATE TABLE changes ( + canvas_id TEXT NOT NULL, + thread_id TEXT NOT NULL, + snapshot_json TEXT NOT NULL CHECK (json_valid(snapshot_json)), + PRIMARY KEY (canvas_id, thread_id), + FOREIGN KEY (canvas_id) REFERENCES spaces(canvas_id) ON DELETE CASCADE +) STRICT; + +CREATE TABLE tasks ( + canvas_id TEXT PRIMARY KEY, + snapshot_json TEXT NOT NULL CHECK (json_valid(snapshot_json)), + FOREIGN KEY (canvas_id) REFERENCES spaces(canvas_id) ON DELETE CASCADE +) STRICT; + +CREATE TABLE space_extensions ( + extension_id INTEGER PRIMARY KEY AUTOINCREMENT, + canvas_id TEXT NOT NULL, + namespace TEXT NOT NULL, + UNIQUE (canvas_id, namespace), + FOREIGN KEY (canvas_id) REFERENCES spaces(canvas_id) ON DELETE CASCADE +) STRICT; + +CREATE TABLE delta_log ( + canvas_id TEXT NOT NULL, + version INTEGER NOT NULL, + entry_json TEXT NOT NULL CHECK (json_valid(entry_json)), + PRIMARY KEY (canvas_id, version), + FOREIGN KEY (canvas_id) REFERENCES spaces(canvas_id) ON DELETE CASCADE +) STRICT; + +INSERT INTO spaces ( + canvas_id, title, collision_key, version, state_json, + created_at, updated_at, is_world +) VALUES ( + 'fixture-world', 'World', '.world', 0, + '{"nodes":[],"edges":[]}', 1, 1, 1 +); + +INSERT INTO spaces ( + canvas_id, title, collision_key, version, state_json, + created_at, updated_at, is_world +) VALUES ( + 'fixture-space', 'Fixture Space', 'fixture space', 3, + '{"nodes":[{"id":"fixture-node","type":"note"}],"edges":[]}', + 10, 13, 0 +); + +INSERT INTO nodes ( + canvas_id, node_id, record_json, revision, label_collision_key +) VALUES ( + 'fixture-space', 'fixture-node', + '{"nodeId":"fixture-node","type":"note","label":"Fixture Node","content":"fixture body"}', + 'fixture-revision', 'fixture node' +); + +INSERT INTO events (canvas_id, event_json) VALUES ( + 'fixture-space', + '{"payload":{"action":"node_selected","node":{"id":"fixture-node","type":"note","label":"Fixture Node"}},"ts":12}' +); + +INSERT INTO changes (canvas_id, thread_id, snapshot_json) VALUES ( + 'fixture-space', 'fixture-thread', '[]' +); + +INSERT INTO tasks (canvas_id, snapshot_json) VALUES ( + 'fixture-space', '{"version":1,"tasks":[],"runs":[]}' +); + +INSERT INTO delta_log (canvas_id, version, entry_json) VALUES ( + 'fixture-space', 3, + '{"version":3,"ts":13,"commands":[],"deltas":[],"originator":{"source":"system"}}' +); + +PRAGMA user_version = 1; +COMMIT; diff --git a/apps/server/src/modules/storage/backends/sqlite/identity.ts b/apps/server/src/modules/storage/backends/sqlite/identity.ts new file mode 100644 index 00000000..3a84a749 --- /dev/null +++ b/apps/server/src/modules/storage/backends/sqlite/identity.ts @@ -0,0 +1,92 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +/** + * Allocation of the names a Space or Node is filed under. + * + * The `collision_key` columns carry a UNIQUE constraint, so a title or label + * has to be de-duplicated before it reaches the database rather than after a + * failed insert. These rules are pure and share `utils/naming` with Disk, so + * both backends hand out the same ` (2)` suffixes for the same inputs — see + * `backends/disk/space-title.ts` for the directory-locator half. + */ + +import { + dedupeName, + normalizeForCompare, + toSafeFilename, +} from '../../../../utils/naming.js'; + +import type { NodeContent } from '../../../canvas/persistence-types.js'; + +function allocatedSpaceTitle( + requested: string | null, + canvasId: string, + allocatedName: string, +): string | null { + if (requested === null) return null; + const base = toSafeFilename(requested, canvasId); + if (allocatedName === base) return requested; + const candidate = `${requested}${allocatedName.slice(base.length)}`; + return toSafeFilename(candidate, canvasId) === allocatedName + ? candidate + : allocatedName; +} + +export function allocateSpaceIdentity( + requestedTitle: string | null, + canvasId: string, + occupiedCollisionKeys: Iterable, +): { readonly title: string | null; readonly collisionKey: string } { + const base = toSafeFilename(requestedTitle, canvasId); + const allocated = dedupeName(base, occupiedCollisionKeys); + return { + title: allocatedSpaceTitle(requestedTitle, canvasId, allocated), + collisionKey: normalizeForCompare(allocated), + }; +} + +export function collisionKeyForTitle( + title: string | null, + canvasId: string, +): string { + return normalizeForCompare(toSafeFilename(title, canvasId)); +} + +export function allocateNodeIdentity( + record: NodeContent, + nodeId: string, + existingCollisionKey: string | null, + occupiedCollisionKeys: Iterable, +): { + readonly record: NodeContent; + readonly collisionKey: string; + readonly desiredCollisionKey: string; +} { + const trimmedLabel = + typeof record.label === 'string' && record.label.trim().length > 0 + ? record.label + : null; + if (trimmedLabel === null && existingCollisionKey !== null) { + return { + record, + collisionKey: existingCollisionKey, + desiredCollisionKey: existingCollisionKey, + }; + } + + const desired = toSafeFilename(trimmedLabel, nodeId); + const allocated = dedupeName(desired, occupiedCollisionKeys); + const suffix = + allocated.length > desired.length && allocated.startsWith(desired) + ? allocated.slice(desired.length) + : ''; + return { + record: + suffix && trimmedLabel + ? { ...record, label: `${trimmedLabel}${suffix}` } + : record, + collisionKey: normalizeForCompare(allocated), + desiredCollisionKey: normalizeForCompare(desired), + }; +} diff --git a/apps/server/src/modules/storage/backends/sqlite/integration.test.ts b/apps/server/src/modules/storage/backends/sqlite/integration.test.ts new file mode 100644 index 00000000..b1b2fff6 --- /dev/null +++ b/apps/server/src/modules/storage/backends/sqlite/integration.test.ts @@ -0,0 +1,655 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +import { readFileSync } from 'node:fs'; + +import { afterEach, describe, expect, it } from 'vitest'; + +import { extractCanvasChanges } from '@huabu/shared/canvas-engine'; + +import { applySqliteMigrations, SQLITE_SCHEMA_VERSION } from './database.js'; +import { SqliteStructuredStore } from './structured-store.js'; +import { + createSqliteTestFile, + installDeltaAbortTrigger, + openSqliteTestStore, + readSqliteDeltaLog, + withTestDatabase, +} from './test-support.js'; + +import type { + CanvasFile, + DeltaLogEntry, + NodeContent, +} from '../../../canvas/persistence-types.js'; +import type { NodeSnapshot } from '../../ports/structured.js'; +import type { TaskRecord } from '@huabu/shared'; +import type { CanvasNode } from '@huabu/shared/canvas-engine'; + +const cleanups: Array<() => Promise | void> = []; + +afterEach(async () => { + for (const cleanup of cleanups.splice(0).reverse()) { + await cleanup(); + } +}); + +function trackedFile(prefix: string) { + const file = createSqliteTestFile(prefix); + cleanups.push(file.remove); + return file; +} + +function trackedStore(filename: string): SqliteStructuredStore { + const store = new SqliteStructuredStore(filename); + cleanups.push(() => store.close()); + return store; +} + +async function trackedOpenStore(prefix: string) { + const harness = await openSqliteTestStore(prefix); + cleanups.push(harness.cleanup); + return harness; +} + +function note(nodeId: string, label: string, content: string): NodeContent { + return { nodeId, type: 'note', label, content }; +} + +function nextRecord(current: CanvasFile): CanvasFile { + return { + ...current, + version: current.version + 1, + updatedAt: current.updatedAt + 1, + }; +} + +function delta(version: number, marker: string): DeltaLogEntry { + return { + version, + ts: version + 100, + commands: [{ marker }], + deltas: [{ marker }], + originator: { source: 'system' }, + }; +} + +async function createSpace( + store: SqliteStructuredStore, + canvasId: string, + title: string, +): Promise { + const result = await store.spaces().create({ canvasId, title }); + if (!result.ok) throw new Error(`Could not create test Space ${canvasId}`); + return result.record; +} + +describe('SqliteStructuredStore lifecycle and schema', () => { + it('rejects an empty database filename', () => { + expect(() => new SqliteStructuredStore('')).toThrow(/filename.*empty/i); + }); + + it('rejects before init and after close while lifecycle operations stay idempotent', async () => { + const file = trackedFile('huabu-sqlite-lifecycle-'); + const store = trackedStore(file.filename); + + await expect(store.health()).rejects.toThrow(/not initialized/); + await expect( + Promise.resolve().then(() => store.spaces().list()), + ).rejects.toThrow(/not initialized/); + await expect( + Promise.resolve().then(() => store.space('lifecycle-space').read()), + ).rejects.toThrow(/not initialized/); + await expect( + store.space('lifecycle-space').nodes.readMany([]), + ).rejects.toThrow(/not initialized/); + + await expect(store.init()).resolves.toBeUndefined(); + await expect(store.init()).resolves.toBeUndefined(); + await expect(store.health()).resolves.toEqual({ ok: true, kind: 'sqlite' }); + await expect(store.health()).resolves.toEqual({ ok: true, kind: 'sqlite' }); + + await expect(store.close()).resolves.toBeUndefined(); + await expect(store.close()).resolves.toBeUndefined(); + await expect(store.health()).rejects.toThrow(/closed/); + await expect( + Promise.resolve().then(() => store.spaces().list()), + ).rejects.toThrow(/closed/); + await expect( + Promise.resolve().then(() => store.space('lifecycle-space').read()), + ).rejects.toThrow(/closed/); + await expect( + store.space('lifecycle-space').nodes.readMany([]), + ).rejects.toThrow(/closed/); + await expect(store.init()).rejects.toThrow(/closed/); + }); + + it('creates the complete STRICT v1 schema in a fresh database', async () => { + const file = trackedFile('huabu-sqlite-fresh-schema-'); + const store = trackedStore(file.filename); + await store.init(); + + withTestDatabase(file.filename, (database) => { + expect(database.prepare('PRAGMA user_version').get()).toEqual({ + user_version: SQLITE_SCHEMA_VERSION, + }); + const expectedTables = [ + 'changes', + 'delta_log', + 'events', + 'nodes', + 'space_extensions', + 'spaces', + 'tasks', + ]; + const tableRows = database.prepare('PRAGMA table_list').all(); + const productionTables = tableRows.filter((row) => + expectedTables.includes(String(row['name'])), + ); + expect(productionTables.map((row) => row['name']).sort()).toEqual( + expectedTables, + ); + expect(productionTables.every((row) => row['strict'] === 1)).toBe(true); + expect( + database + .prepare('PRAGMA foreign_key_list(nodes)') + .all() + .map((row) => ({ + table: row['table'], + from: row['from'], + to: row['to'], + onDelete: row['on_delete'], + })), + ).toContainEqual({ + table: 'spaces', + from: 'canvas_id', + to: 'canvas_id', + onDelete: 'CASCADE', + }); + }); + }); + + it('opens the immutable v1 SQL fixture without rewriting its records', async () => { + const file = trackedFile('huabu-sqlite-v1-fixture-'); + const fixtureSql = readFileSync( + new URL('./fixtures/v1.sql', import.meta.url), + 'utf8', + ); + withTestDatabase(file.filename, (database) => database.exec(fixtureSql)); + + const store = trackedStore(file.filename); + await store.init(); + await expect(store.spaces().worldId()).resolves.toBe('fixture-world'); + await expect(store.spaces().list()).resolves.toEqual([ + { + canvasId: 'fixture-space', + title: 'Fixture Space', + nodeCount: 1, + createdAt: 10, + updatedAt: 13, + }, + ]); + const space = store.space('fixture-space'); + await expect(space.read()).resolves.toEqual({ + canvasId: 'fixture-space', + title: 'Fixture Space', + version: 3, + state: { + nodes: [{ id: 'fixture-node', type: 'note' }], + edges: [], + }, + createdAt: 10, + updatedAt: 13, + }); + await expect(space.nodes.read('fixture-node')).resolves.toEqual({ + record: note('fixture-node', 'Fixture Node', 'fixture body'), + revision: 'fixture-revision', + }); + await expect(space.events.read()).resolves.toEqual([ + { + payload: { + action: 'node_selected', + node: { id: 'fixture-node', type: 'note', label: 'Fixture Node' }, + }, + ts: 12, + }, + ]); + await expect(space.changes.read('fixture-thread')).resolves.toEqual([]); + await expect(space.tasks.read()).resolves.toEqual({ + version: 1, + tasks: [], + runs: [], + }); + expect(readSqliteDeltaLog(file.filename, 'fixture-space')).toEqual([ + { + version: 3, + ts: 13, + commands: [], + deltas: [], + originator: { source: 'system' }, + }, + ]); + }); + + it('rejects a database whose user_version is from the future', async () => { + const file = trackedFile('huabu-sqlite-future-schema-'); + withTestDatabase(file.filename, (database) => { + database.exec(`PRAGMA user_version = ${SQLITE_SCHEMA_VERSION + 1}`); + }); + const store = trackedStore(file.filename); + + await expect(store.init()).rejects.toThrow(/newer than supported/); + await expect(store.health()).rejects.toThrow(/closed/); + }); + + it('rolls every migration step and user_version back when a later step fails', () => { + const file = trackedFile('huabu-sqlite-migration-rollback-'); + withTestDatabase(file.filename, (database) => { + expect(() => + applySqliteMigrations(database, [ + { + version: 1, + sql: 'CREATE TABLE migration_v1 (id INTEGER PRIMARY KEY) STRICT;', + }, + { + version: 2, + sql: ` + CREATE TABLE migration_v2 (id INTEGER PRIMARY KEY) STRICT; + INSERT INTO missing_migration_table (id) VALUES (1); + `, + }, + ]), + ).toThrow(/missing_migration_table|no such table/); + + expect(database.prepare('PRAGMA user_version').get()).toEqual({ + user_version: 0, + }); + expect( + database + .prepare( + `SELECT name + FROM sqlite_schema + WHERE type = 'table' AND name LIKE 'migration_%'`, + ) + .all(), + ).toEqual([]); + }); + }); +}); + +describe('SqliteStructuredStore persistence and transactions', () => { + it('persists Space and Node records across close and reopen', async () => { + const harness = await trackedOpenStore('huabu-sqlite-reopen-'); + const canvasId = 'reopen-space'; + const created = await createSpace(harness.store, canvasId, 'Reopen Space'); + const record = note('reopen-node', 'Reopen Node', 'persisted body'); + const put = await harness.store.space(canvasId).nodes.put({ + nodeId: record.nodeId, + record, + }); + expect(put).toMatchObject({ ok: true, record }); + + await harness.store.close(); + const reopened = trackedStore(harness.filename); + await reopened.init(); + + await expect(reopened.spaces().worldId()).resolves.toBe( + harness.world.canvasId, + ); + await expect(reopened.space(canvasId).read()).resolves.toEqual(created); + await expect( + reopened.space(canvasId).nodes.read(record.nodeId), + ).resolves.toEqual(put.ok ? { record, revision: put.revision } : null); + }); + + it('rolls node, record, and delta state back on a real trigger abort', async () => { + const harness = await trackedOpenStore('huabu-sqlite-trigger-rollback-'); + const canvasId = 'trigger-rollback-space'; + const baseline = await createSpace( + harness.store, + canvasId, + 'Trigger Rollback Space', + ); + const oldNode = note('old-node', 'Old Node', 'before'); + const newNode = note('new-node', 'New Node', 'after'); + const oldPut = await harness.store.space(canvasId).nodes.put({ + nodeId: oldNode.nodeId, + record: oldNode, + }); + if (!oldPut.ok) throw new Error('Could not seed rollback node'); + + const next: CanvasFile = { + ...nextRecord(baseline), + state: { + nodes: [{ id: newNode.nodeId, type: newNode.type }], + edges: [], + }, + }; + const restore = installDeltaAbortTrigger( + harness.filename, + 'forced delta abort', + ); + try { + await expect( + harness.store.space(canvasId).write({ + expectedVersion: baseline.version, + nextRecord: next, + nodeMutations: [ + { kind: 'delete', nodeId: oldNode.nodeId }, + { + kind: 'put', + nodeId: newNode.nodeId, + record: newNode, + authoritativeInsert: true, + }, + ], + delta: delta(next.version, 'trigger-abort'), + }), + ).rejects.toThrow('forced delta abort'); + } finally { + restore(); + } + + const space = harness.store.space(canvasId); + await expect(space.read()).resolves.toEqual(baseline); + await expect(space.nodes.read(oldNode.nodeId)).resolves.toEqual({ + record: oldPut.record, + revision: oldPut.revision, + }); + await expect(space.nodes.read(newNode.nodeId)).resolves.toBeNull(); + expect(readSqliteDeltaLog(harness.filename, canvasId)).toEqual([]); + await expect( + space.nodes.put({ + nodeId: oldNode.nodeId, + expectedRevision: oldPut.revision, + record: { ...oldNode, content: 'still writable' }, + }), + ).resolves.toMatchObject({ ok: true }); + }); + + it('rejects sparse JSON arrays without changing the exact persisted Node', async () => { + const harness = await trackedOpenStore('huabu-sqlite-sparse-json-'); + const canvasId = 'sparse-json-space'; + await createSpace(harness.store, canvasId, 'Sparse JSON Space'); + const nodes = harness.store.space(canvasId).nodes; + const record = note('sparse-json-node', 'Sparse JSON Node', 'before'); + const baseline = await nodes.put({ nodeId: record.nodeId, record }); + if (!baseline.ok) throw new Error('Could not seed sparse JSON node'); + const sparse: unknown[] = []; + sparse[1] = 'present'; + expect(0 in sparse).toBe(false); + + await expect( + nodes.put({ + nodeId: record.nodeId, + expectedRevision: baseline.revision, + record: { ...record, metadata: sparse }, + }), + ).rejects.toThrow(/sparse array/i); + await expect(nodes.read(record.nodeId)).resolves.toEqual({ + record, + revision: baseline.revision, + }); + }); + + it('recovers malformed stored Node content through every read shape', async () => { + const harness = await trackedOpenStore('huabu-sqlite-node-recovery-'); + const canvasId = 'node-recovery-space'; + await createSpace(harness.store, canvasId, 'Node Recovery Space'); + const nodes = harness.store.space(canvasId).nodes; + const record = note('recoverable-node', 'Recoverable Node', 'before'); + const baseline = await nodes.put({ nodeId: record.nodeId, record }); + if (!baseline.ok) throw new Error('Could not seed recoverable Node'); + + withTestDatabase(harness.filename, (database) => { + database + .prepare( + `UPDATE nodes + SET record_json = ? + WHERE canvas_id = ? AND node_id = ?`, + ) + .run('{"content":"recoverable body"}', canvasId, record.nodeId); + }); + + const recovered: NodeSnapshot = { + record: { + nodeId: record.nodeId, + type: 'note', + label: null, + content: 'recoverable body', + }, + revision: baseline.revision, + }; + await expect(nodes.read(record.nodeId)).resolves.toEqual(recovered); + await expect(nodes.readMany([record.nodeId])).resolves.toEqual( + new Map([[record.nodeId, recovered]]), + ); + await expect(nodes.list()).resolves.toEqual( + new Map([[record.nodeId, recovered]]), + ); + const delivered: NodeSnapshot[] = []; + await expect( + nodes.stream((snapshot) => delivered.push(snapshot)), + ).resolves.toEqual(new Map([[record.nodeId, recovered]])); + expect(delivered).toEqual([recovered]); + + await expect( + nodes.put({ + nodeId: record.nodeId, + expectedRevision: baseline.revision, + record: { ...record, content: 'repaired' }, + }), + ).resolves.toMatchObject({ ok: true }); + }); + + it('releases deletion admission when post-acquire Space setup throws', async () => { + const harness = await trackedOpenStore('huabu-sqlite-delete-setup-'); + const canvasId = 'delete-setup-space'; + const record = await createSpace( + harness.store, + canvasId, + 'Delete Setup Space', + ); + const repository = harness.store.spaces(); + + const malformedAttempt = repository.beginDelete({ canvasId }); + withTestDatabase(harness.filename, (database) => { + database + .prepare('UPDATE spaces SET state_json = ? WHERE canvas_id = ?') + .run('[]', canvasId); + }); + await expect(malformedAttempt).rejects.toThrow(/Invalid Space/); + withTestDatabase(harness.filename, (database) => { + database + .prepare('UPDATE spaces SET state_json = ? WHERE canvas_id = ?') + .run(JSON.stringify(record.state), canvasId); + }); + + let secondResult: + | Awaited> + | undefined; + let secondError: unknown; + const secondSettled = repository.beginDelete({ canvasId }).then( + (result) => { + secondResult = result; + }, + (error: unknown) => { + secondError = error; + }, + ); + await new Promise((resolve) => setImmediate(resolve)); + + expect(secondError).toBeUndefined(); + expect(secondResult).toMatchObject({ ok: true }); + if (!secondResult?.ok) { + throw new Error('Deletion gate remained occupied after setup failure'); + } + await secondResult.session.abort(); + await secondSettled; + }); + + it('cascades every child record when a deletion session finishes', async () => { + const harness = await trackedOpenStore('huabu-sqlite-delete-session-'); + const canvasId = 'delete-session-space'; + const baseline = await createSpace( + harness.store, + canvasId, + 'Delete Session Space', + ); + const record = note('deleted-node', 'Deleted Node', 'stale body'); + const handle = harness.store.space(canvasId); + await handle.nodes.put({ nodeId: record.nodeId, record }); + await handle.events.append([ + { + payload: { + action: 'node_selected', + node: { + id: record.nodeId, + type: 'note', + label: record.label ?? undefined, + }, + }, + ts: 2, + }, + ]); + const changeNode: CanvasNode = { + id: 'change-node', + type: 'note', + position: { x: 0, y: 0 }, + data: { label: 'Change Node', content: 'change body' }, + } as CanvasNode; + await handle.changes.append( + 'delete-thread', + extractCanvasChanges([{ type: 'INSERT_NODE', node: changeNode }]), + ); + const task: TaskRecord = { + taskId: 'delete-task', + canvasId, + goal: 'Delete this fixture', + defaultRootProfileId: 'profile-delete', + anchorNodeId: record.nodeId, + createdAt: 3, + }; + await handle.tasks.create(task); + const next = nextRecord(baseline); + await expect( + handle.write({ + expectedVersion: baseline.version, + nextRecord: next, + nodeMutations: [], + delta: delta(next.version, 'delete-session'), + }), + ).resolves.toEqual({ ok: true }); + + withTestDatabase(harness.filename, (database) => { + for (const table of [ + 'nodes', + 'events', + 'changes', + 'tasks', + 'delta_log', + ]) { + expect( + database + .prepare( + `SELECT count(*) AS count FROM ${table} WHERE canvas_id = ?`, + ) + .get(canvasId)?.['count'], + ).toBe(1); + } + }); + + const started = await harness.store.spaces().beginDelete({ canvasId }); + if (!started.ok) throw new Error('Ordinary Space must be deletable'); + await expect(handle.read()).resolves.toEqual(next); + await expect(handle.nodes.read(record.nodeId)).resolves.toMatchObject({ + record, + }); + await expect(started.session.finish()).resolves.toEqual({ + ok: true, + reason: 'deleted', + }); + + withTestDatabase(harness.filename, (database) => { + for (const table of [ + 'nodes', + 'events', + 'changes', + 'tasks', + 'delta_log', + ]) { + expect( + database + .prepare( + `SELECT count(*) AS count FROM ${table} WHERE canvas_id = ?`, + ) + .get(canvasId)?.['count'], + ).toBe(0); + } + }); + + await expect(handle.read()).resolves.toBeNull(); + }); + + it('allows a first write after deleting an already absent node', async () => { + const harness = await trackedOpenStore('huabu-sqlite-absent-delete-'); + const canvasId = 'absent-delete-space'; + await createSpace(harness.store, canvasId, 'Absent Delete Space'); + const nodes = harness.store.space(canvasId).nodes; + const record = note('not-yet-created', 'Not Yet Created', 'body'); + + await expect(nodes.delete(record.nodeId)).resolves.toBe('absent'); + await expect( + nodes.put({ nodeId: record.nodeId, record }), + ).resolves.toMatchObject({ ok: true, record }); + }); + + it('allows immediate reuse of a deleted primary key across reopen', async () => { + const harness = await trackedOpenStore('huabu-sqlite-delete-reopen-'); + const canvasId = 'tombstone-reopen-space'; + await createSpace(harness.store, canvasId, 'Tombstone Reopen Space'); + const record = note('tombstoned-node', 'Tombstoned Node', 'before'); + const nodes = harness.store.space(canvasId).nodes; + const initial = await nodes.put({ nodeId: record.nodeId, record }); + if (!initial.ok) throw new Error('Could not create initial test Node'); + + await expect(nodes.delete(record.nodeId)).resolves.toBe('deleted'); + const recreated = await nodes.put({ + nodeId: record.nodeId, + record: { ...record, content: 'immediate replacement' }, + }); + if (!recreated.ok) throw new Error('Could not recreate test Node'); + expect(recreated.record).toEqual({ + ...record, + content: 'immediate replacement', + }); + expect(recreated.revision).not.toBe(initial.revision); + await expect( + nodes.put({ + nodeId: record.nodeId, + expectedRevision: initial.revision, + record: { ...record, content: 'stale replacement' }, + }), + ).resolves.toEqual({ + ok: false, + reason: 'revision-conflict', + currentRevision: recreated.revision, + }); + + await harness.store.close(); + const reopened = trackedStore(harness.filename); + await reopened.init(); + await expect( + reopened.space(canvasId).nodes.delete(record.nodeId), + ).resolves.toBe('deleted'); + await expect( + reopened.space(canvasId).nodes.put({ + nodeId: record.nodeId, + record: { ...record, content: 'after reopen' }, + }), + ).resolves.toMatchObject({ + ok: true, + record: { ...record, content: 'after reopen' }, + }); + }); +}); diff --git a/apps/server/src/modules/storage/backends/sqlite/rows.ts b/apps/server/src/modules/storage/backends/sqlite/rows.ts new file mode 100644 index 00000000..9cbe48f3 --- /dev/null +++ b/apps/server/src/modules/storage/backends/sqlite/rows.ts @@ -0,0 +1,302 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +/** + * Movement of persisted values between domain records and SQLite rows. + * + * Every column this backend stores is either JSON text or a scalar, so the + * codecs here are the single place that decides what a well-formed stored + * value looks like. Space and log reads reject malformed domain values. Node + * reads preserve the port's repair path by recovering malformed JSON values + * into a valid record whose content still exposes the stored value. + */ + +import { SQLITE_WORLD_COLLISION_KEY } from './database.js'; +import { canvasFileShapeError } from '../../../canvas/persistence-validation.js'; + +import type { + CanvasFile, + NodeContent, +} from '../../../canvas/persistence-types.js'; +import type { DatabaseSync } from 'node:sqlite'; + +type JsonPrimitive = null | boolean | number | string; +type JsonValue = JsonPrimitive | JsonValue[] | { [key: string]: JsonValue }; + +function assertJsonValue( + value: unknown, + context: string, + seen: Set, +): asserts value is JsonValue { + if ( + value === null || + typeof value === 'string' || + typeof value === 'boolean' + ) { + return; + } + if (typeof value === 'number') { + if (!Number.isFinite(value)) { + throw new TypeError(`${context} contains a non-finite number`); + } + return; + } + if (typeof value !== 'object') { + throw new TypeError(`${context} contains a non-JSON value`); + } + if (seen.has(value)) throw new TypeError(`${context} contains a cycle`); + seen.add(value); + try { + if (Array.isArray(value)) { + for (let index = 0; index < value.length; index += 1) { + if (!Object.prototype.hasOwnProperty.call(value, index)) { + throw new TypeError(`${context} contains a sparse array`); + } + assertJsonValue(value[index], `${context}[${index}]`, seen); + } + return; + } + const prototype = Object.getPrototypeOf(value); + if (prototype !== Object.prototype && prototype !== null) { + throw new TypeError(`${context} contains a non-plain object`); + } + for (const [key, entry] of Object.entries(value)) { + assertJsonValue(entry, `${context}.${key}`, seen); + } + } finally { + seen.delete(value); + } +} + +export function stringifyJson(value: unknown, context: string): string { + assertJsonValue(value, context, new Set()); + const encoded = JSON.stringify(value); + if (encoded === undefined) { + throw new TypeError(`${context} is not representable as JSON`); + } + return encoded; +} + +export function parseJson(value: unknown, context: string): unknown { + if (typeof value !== 'string') { + throw new SyntaxError(`${context} is not stored as JSON text`); + } + try { + return JSON.parse(value) as unknown; + } catch (error) { + throw new SyntaxError( + `Invalid JSON in ${context}: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + } +} + +function rowObject(value: unknown, context: string): Record { + if (typeof value !== 'object' || value === null || Array.isArray(value)) { + throw new SyntaxError(`Missing or malformed SQLite row for ${context}`); + } + return value as Record; +} + +function stringColumn( + row: Record, + column: string, + context: string, +): string { + const value = row[column]; + if (typeof value !== 'string') { + throw new SyntaxError(`Invalid ${column} in ${context}`); + } + return value; +} + +function nullableStringColumn( + row: Record, + column: string, + context: string, +): string | null { + const value = row[column]; + if (value !== null && typeof value !== 'string') { + throw new SyntaxError(`Invalid ${column} in ${context}`); + } + return value; +} + +function numberColumn( + row: Record, + column: string, + context: string, +): number { + const value = row[column]; + if (typeof value !== 'number' || !Number.isFinite(value)) { + throw new SyntaxError(`Invalid ${column} in ${context}`); + } + return value; +} + +export interface PersistedSpace { + readonly record: CanvasFile; + readonly collisionKey: string; + readonly isWorld: boolean; +} + +export function decodeSpaceRow(value: unknown): PersistedSpace { + const row = rowObject(value, 'Space'); + const canvasId = stringColumn(row, 'canvas_id', 'Space'); + const context = `Space ${JSON.stringify(canvasId)}`; + const record: CanvasFile = { + canvasId, + title: nullableStringColumn(row, 'title', context), + version: numberColumn(row, 'version', context), + state: parseJson( + row['state_json'], + `${context} state`, + ) as CanvasFile['state'], + createdAt: numberColumn(row, 'created_at', context), + updatedAt: numberColumn(row, 'updated_at', context), + }; + const shapeError = canvasFileShapeError(record, canvasId); + if (shapeError) throw new SyntaxError(`Invalid ${context}: ${shapeError}`); + const world = numberColumn(row, 'is_world', context); + if (world !== 0 && world !== 1) { + throw new SyntaxError(`Invalid is_world in ${context}`); + } + return { + record, + collisionKey: stringColumn(row, 'collision_key', context), + isWorld: world === 1, + }; +} + +export const SPACE_COLUMNS = + 'canvas_id, title, collision_key, version, state_json, created_at, updated_at, is_world'; + +export function readSpaceRow( + database: DatabaseSync, + canvasId: string, +): PersistedSpace | null { + const row = database + .prepare(`SELECT ${SPACE_COLUMNS} FROM spaces WHERE canvas_id = ?`) + .get(canvasId); + return row === undefined ? null : decodeSpaceRow(row); +} + +export function validateCanvasFile(record: CanvasFile, canvasId: string): void { + const shapeError = canvasFileShapeError(record, canvasId); + if (shapeError) { + throw new TypeError(`Invalid Space record: ${shapeError}`); + } + stringifyJson(record.state, `Space ${JSON.stringify(canvasId)} state`); +} + +export function insertSpaceRow( + database: DatabaseSync, + record: CanvasFile, + collisionKey: string, + isWorld = false, +): void { + validateCanvasFile(record, record.canvasId); + database + .prepare( + `INSERT INTO spaces ( + canvas_id, title, collision_key, version, state_json, + created_at, updated_at, is_world + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .run( + record.canvasId, + record.title, + isWorld ? SQLITE_WORLD_COLLISION_KEY : collisionKey, + record.version, + stringifyJson(record.state, `Space ${record.canvasId} state`), + record.createdAt, + record.updatedAt, + isWorld ? 1 : 0, + ); +} + +export function updateSpaceRow( + database: DatabaseSync, + record: CanvasFile, + expectedVersion: number, +): number { + validateCanvasFile(record, record.canvasId); + const result = database + .prepare( + `UPDATE spaces + SET version = ?, state_json = ?, updated_at = ? + WHERE canvas_id = ? AND version = ?`, + ) + .run( + record.version, + stringifyJson(record.state, `Space ${record.canvasId} state`), + record.updatedAt, + record.canvasId, + expectedVersion, + ); + return Number(result.changes); +} + +export function validateNodeContent( + record: NodeContent, + expectedNodeId: string, +): void { + if (typeof record !== 'object' || record === null || Array.isArray(record)) { + throw new TypeError('Node record must be an object'); + } + if (record.nodeId !== expectedNodeId) { + throw new Error( + `Node id mismatch: argument=${JSON.stringify(expectedNodeId)} ` + + `record=${JSON.stringify(record.nodeId)}`, + ); + } + if (typeof record.type !== 'string') { + throw new TypeError('Node record type must be a string'); + } + if (record.label !== null && typeof record.label !== 'string') { + throw new TypeError('Node record label must be a string or null'); + } + if (typeof record.content !== 'string') { + throw new TypeError('Node record content must be a string'); + } + stringifyJson(record, `Node ${JSON.stringify(expectedNodeId)} record`); +} + +export function decodeNodeRecord( + value: unknown, + expectedNodeId: string, +): NodeContent { + const parsed = parseJson(value, `Node ${JSON.stringify(expectedNodeId)}`); + try { + validateNodeContent(parsed as NodeContent, expectedNodeId); + return parsed as NodeContent; + } catch { + // A valid JSON value can still have a damaged Node shape after an + // out-of-band database edit. Keep it reachable so a normal put can repair + // it, matching the lenient content rule used by the Disk adapter. + const fields = + typeof parsed === 'object' && parsed !== null && !Array.isArray(parsed) + ? (parsed as Record) + : {}; + return { + ...fields, + nodeId: expectedNodeId, + type: typeof fields['type'] === 'string' ? fields['type'] : 'note', + label: typeof fields['label'] === 'string' ? fields['label'] : null, + content: + typeof fields['content'] === 'string' + ? fields['content'] + : stringifyJson(parsed, `Malformed Node ${expectedNodeId}`), + } as NodeContent; + } +} + +export function requireRevision(value: unknown, nodeId: string): string { + if (typeof value !== 'string' || value.length === 0) { + throw new SyntaxError( + `Invalid persisted revision for Node ${JSON.stringify(nodeId)}`, + ); + } + return value; +} diff --git a/apps/server/src/modules/storage/backends/sqlite/space-extension.ts b/apps/server/src/modules/storage/backends/sqlite/space-extension.ts new file mode 100644 index 00000000..51c86832 --- /dev/null +++ b/apps/server/src/modules/storage/backends/sqlite/space-extension.ts @@ -0,0 +1,58 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +/** SQLite connection point for one extension namespace in one Space. */ + +import { withImmediateTransaction } from './database.js'; +import { assertValidNamespace } from '../../ports/namespace.js'; + +import type { SqliteStoreContext } from './database.js'; +import type { SpaceHandle } from '../../ports/structured.js'; + +export function createSqliteSpaceExtension( + context: SqliteStoreContext, + canvasId: string, +): SpaceHandle['extension'] { + return async function extension(namespaceInput: string) { + const namespace = assertValidNamespace(namespaceInput); + context.assertMutationAllowed(canvasId); + const database = context.database(); + + return withImmediateTransaction(database, () => { + const exists = + database + .prepare('SELECT 1 AS present FROM spaces WHERE canvas_id = ?') + .get(canvasId)?.['present'] === 1; + if (!exists) return null; + + database + .prepare( + `INSERT INTO space_extensions (canvas_id, namespace) + VALUES (?, ?) + ON CONFLICT(canvas_id, namespace) DO NOTHING`, + ) + .run(canvasId, namespace); + const extensionId = database + .prepare( + `SELECT extension_id + FROM space_extensions + WHERE canvas_id = ? AND namespace = ?`, + ) + .get(canvasId, namespace)?.['extension_id']; + if ( + typeof extensionId !== 'number' || + !Number.isSafeInteger(extensionId) || + extensionId <= 0 + ) { + throw new Error( + `Could not resolve SQLite extension ${JSON.stringify(namespace)}`, + ); + } + return Object.freeze({ + kind: 'sqlite' as const, + database, + extensionId, + }); + }); + }; +} diff --git a/apps/server/src/modules/storage/backends/sqlite/space-logs.ts b/apps/server/src/modules/storage/backends/sqlite/space-logs.ts new file mode 100644 index 00000000..5fe88a84 --- /dev/null +++ b/apps/server/src/modules/storage/backends/sqlite/space-logs.ts @@ -0,0 +1,269 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +import { canvasEventInputSchema, canvasEventRecordSchema } from '@huabu/shared'; +import { + coalesceChanges, + type CanvasChangeRecord, +} from '@huabu/shared/canvas-engine'; + +import { withImmediateTransaction } from './database.js'; +import { parseJson, stringifyJson } from './rows.js'; +import { sanitizeId } from '../../../../utils/fs.js'; + +import type { SqliteStoreContext } from './database.js'; +import type { CanvasEvent } from '../../../canvas/persistence-types.js'; +import type { + NewCanvasEvent, + SpaceChanges, + SpaceEvents, +} from '../../ports/structured.js'; +import type { z } from 'zod'; + +function firstIssue(error: z.ZodError): string { + const issue = error.issues[0]; + if (!issue) return 'unknown schema violation'; + const location = issue.path.length > 0 ? issue.path.join('.') : ''; + return `${location}: ${issue.message}`; +} + +function requireSpace(context: SqliteStoreContext, canvasId: string): void { + context.assertMutationAllowed(canvasId); + if ( + context + .database() + .prepare('SELECT 1 AS present FROM spaces WHERE canvas_id = ?') + .get(canvasId)?.['present'] !== 1 + ) { + throw new Error( + `SQLite Space logs(${canvasId}) cannot mutate a missing Space`, + ); + } +} + +function decodeEvents(rows: readonly Record[]): CanvasEvent[] { + return rows.map((row, index) => { + const parsedJson = parseJson( + row['event_json'], + `Canvas event ${index + 1}`, + ); + const parsed = canvasEventRecordSchema.safeParse(parsedJson); + if (!parsed.success) { + throw new SyntaxError( + `Invalid persisted Canvas event ${index + 1}: ${firstIssue(parsed.error)}`, + ); + } + return parsedJson as CanvasEvent; + }); +} + +function decodeChanges( + value: unknown, + canvasId: string, + threadId: string, +): CanvasChangeRecord[] { + const parsed = parseJson( + value, + `changes for Space ${JSON.stringify(canvasId)} thread ${JSON.stringify(threadId)}`, + ); + if (!Array.isArray(parsed)) { + throw new SyntaxError( + `Persisted changes for Space ${canvasId} thread ${threadId} must be an array`, + ); + } + return coalesceChanges(parsed as CanvasChangeRecord[]); +} + +export interface SqliteSpaceLogs { + readonly events: SpaceEvents; + readonly changes: SpaceChanges; +} + +class SqliteSpaceLogCoordinator { + readonly #context: SqliteStoreContext; + readonly #canvasId: string; + + constructor(context: SqliteStoreContext, canvasId: string) { + this.#context = context; + this.#canvasId = canvasId; + } + + async readEvents(limit?: number): Promise { + const database = this.#context.database(); + if (limit !== undefined && !(limit > 0)) return []; + if (limit === undefined || !Number.isFinite(limit)) { + return decodeEvents( + database + .prepare( + `SELECT event_json + FROM events + WHERE canvas_id = ? + ORDER BY event_id ASC`, + ) + .all(this.#canvasId), + ); + } + const rows = database + .prepare( + `SELECT event_json + FROM events + WHERE canvas_id = ? + ORDER BY event_id DESC + LIMIT ?`, + ) + .all(this.#canvasId, Math.ceil(limit)) + .reverse(); + return decodeEvents(rows); + } + + async appendEvents(events: readonly NewCanvasEvent[]): Promise { + this.#context.assertOpen(); + if (events.length === 0) return; + const records: CanvasEvent[] = events.map((event, index) => { + const input = canvasEventInputSchema.safeParse(event); + if (!input.success) { + throw new TypeError( + `Invalid Canvas event append input at index ${index}: ${firstIssue(input.error)}`, + ); + } + const record = { + payload: event.payload, + ts: event.ts ?? this.#context.now(), + }; + const parsed = canvasEventRecordSchema.safeParse(record); + if (!parsed.success) { + throw new TypeError( + `Invalid Canvas event append record at index ${index}: ${firstIssue(parsed.error)}`, + ); + } + stringifyJson(record, `Canvas event append input ${index}`); + return record; + }); + + requireSpace(this.#context, this.#canvasId); + const database = this.#context.database(); + withImmediateTransaction(database, () => { + const insert = database.prepare( + 'INSERT INTO events (canvas_id, event_json) VALUES (?, ?)', + ); + for (const record of records) { + insert.run( + this.#canvasId, + stringifyJson(record, `Canvas event for ${this.#canvasId}`), + ); + } + }); + } + + async readChanges(threadIdInput: string): Promise { + const threadId = sanitizeId(threadIdInput, 'threadId'); + const row = this.#context + .database() + .prepare( + `SELECT snapshot_json + FROM changes + WHERE canvas_id = ? AND thread_id = ?`, + ) + .get(this.#canvasId, threadId); + return row === undefined + ? [] + : decodeChanges(row['snapshot_json'], this.#canvasId, threadId); + } + + async appendChanges( + threadIdInput: string, + records: readonly CanvasChangeRecord[], + ): Promise { + const threadId = sanitizeId(threadIdInput, 'threadId'); + stringifyJson(records, `Changes for thread ${JSON.stringify(threadId)}`); + requireSpace(this.#context, this.#canvasId); + const database = this.#context.database(); + return withImmediateTransaction(database, () => { + const current = database + .prepare( + `SELECT snapshot_json + FROM changes + WHERE canvas_id = ? AND thread_id = ?`, + ) + .get(this.#canvasId, threadId); + const existing = + current === undefined + ? [] + : decodeChanges(current['snapshot_json'], this.#canvasId, threadId); + const merged = coalesceChanges([...existing, ...records]); + database + .prepare( + `INSERT INTO changes (canvas_id, thread_id, snapshot_json) + VALUES (?, ?, ?) + ON CONFLICT(canvas_id, thread_id) DO UPDATE SET + snapshot_json = excluded.snapshot_json`, + ) + .run( + this.#canvasId, + threadId, + stringifyJson(merged, `Changes for thread ${threadId}`), + ); + return merged; + }); + } + + async deleteChange( + threadIdInput: string, + changeId: string, + ): Promise { + const threadId = sanitizeId(threadIdInput, 'threadId'); + requireSpace(this.#context, this.#canvasId); + const database = this.#context.database(); + return withImmediateTransaction(database, () => { + const current = database + .prepare( + `SELECT snapshot_json + FROM changes + WHERE canvas_id = ? AND thread_id = ?`, + ) + .get(this.#canvasId, threadId); + if (current === undefined) return null; + const existing = decodeChanges( + current['snapshot_json'], + this.#canvasId, + threadId, + ); + const index = existing.findIndex((record) => record.id === changeId); + if (index < 0) return null; + const [removed] = existing.splice(index, 1); + database + .prepare( + `UPDATE changes + SET snapshot_json = ? + WHERE canvas_id = ? AND thread_id = ?`, + ) + .run( + stringifyJson(existing, `Changes for thread ${threadId}`), + this.#canvasId, + threadId, + ); + return removed ?? null; + }); + } +} + +export function createSqliteSpaceLogs( + context: SqliteStoreContext, + canvasId: string, +): SqliteSpaceLogs { + const coordinator = new SqliteSpaceLogCoordinator(context, canvasId); + return Object.freeze({ + events: Object.freeze({ + read: (limit?: number) => coordinator.readEvents(limit), + append: (events: readonly NewCanvasEvent[]) => + coordinator.appendEvents(events), + }), + changes: Object.freeze({ + read: (threadId: string) => coordinator.readChanges(threadId), + append: (threadId: string, records: readonly CanvasChangeRecord[]) => + coordinator.appendChanges(threadId, records), + delete: (threadId: string, changeId: string) => + coordinator.deleteChange(threadId, changeId), + }), + }); +} diff --git a/apps/server/src/modules/storage/backends/sqlite/space-nodes.ts b/apps/server/src/modules/storage/backends/sqlite/space-nodes.ts new file mode 100644 index 00000000..acd17fd0 --- /dev/null +++ b/apps/server/src/modules/storage/backends/sqlite/space-nodes.ts @@ -0,0 +1,287 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +import { randomUUID } from 'node:crypto'; + +import { withImmediateTransaction } from './database.js'; +import { allocateNodeIdentity } from './identity.js'; +import { + decodeNodeRecord, + requireRevision, + stringifyJson, + validateNodeContent, +} from './rows.js'; +import { sanitizeId } from '../../../../utils/fs.js'; + +import type { SqliteStoreContext } from './database.js'; +import type { + NodeDeleteResult, + NodePutInput, + NodePutResult, + NodeSnapshot, + NodeStreamOptions, + SpaceNodes, +} from '../../ports/structured.js'; +import type { DatabaseSync } from 'node:sqlite'; + +interface NodeRow { + readonly record: NodeSnapshot['record']; + readonly revision: string; + readonly collisionKey: string; +} + +function decodeNodeRow(value: unknown, nodeId: string): NodeRow { + if (typeof value !== 'object' || value === null || Array.isArray(value)) { + throw new SyntaxError(`Malformed persisted Node ${JSON.stringify(nodeId)}`); + } + const row = value as Record; + const collisionKey = row['label_collision_key']; + if (typeof collisionKey !== 'string') { + throw new SyntaxError( + `Invalid collision key for Node ${JSON.stringify(nodeId)}`, + ); + } + return { + record: decodeNodeRecord(row['record_json'], nodeId), + revision: requireRevision(row['revision'], nodeId), + collisionKey, + }; +} + +function readNodeRow( + database: DatabaseSync, + canvasId: string, + nodeId: string, +): NodeRow | null { + const row = database + .prepare( + `SELECT record_json, revision, label_collision_key + FROM nodes + WHERE canvas_id = ? AND node_id = ?`, + ) + .get(canvasId, nodeId); + return row === undefined ? null : decodeNodeRow(row, nodeId); +} + +function spaceExists(database: DatabaseSync, canvasId: string): boolean { + return ( + database + .prepare('SELECT 1 AS present FROM spaces WHERE canvas_id = ?') + .get(canvasId)?.['present'] === 1 + ); +} + +function validatePut(input: NodePutInput): string { + const nodeId = sanitizeId(input.nodeId, 'nodeId'); + validateNodeContent(input.record, nodeId); + if ( + input.expectedRevision !== undefined && + input.expectedRevision !== null && + typeof input.expectedRevision !== 'string' + ) { + throw new TypeError('expectedRevision must be a string, null, or omitted'); + } + return nodeId; +} + +/** Apply one node put inside the caller's active transaction. */ +export function putSqliteNodeInTransaction( + database: DatabaseSync, + canvasId: string, + input: NodePutInput, +): NodePutResult { + const nodeId = validatePut(input); + if (!spaceExists(database, canvasId)) { + return { ok: false, reason: 'not-found' }; + } + + const current = readNodeRow(database, canvasId, nodeId); + const currentRevision = current?.revision ?? null; + if ( + input.expectedRevision !== undefined && + input.expectedRevision !== currentRevision + ) { + return { + ok: false, + reason: 'revision-conflict', + currentRevision, + }; + } + + const occupied = database + .prepare( + `SELECT label_collision_key + FROM nodes + WHERE canvas_id = ? AND node_id <> ?`, + ) + .all(canvasId, nodeId) + .map((row) => row['label_collision_key']) + .filter((value): value is string => typeof value === 'string'); + const allocation = allocateNodeIdentity( + input.record, + nodeId, + current?.collisionKey ?? null, + input.strictLabel === true ? [] : occupied, + ); + + if (input.strictLabel === true) { + const conflict = database + .prepare( + `SELECT node_id, record_json, label_collision_key + FROM nodes + WHERE canvas_id = ? + AND label_collision_key = ? + AND node_id <> ?`, + ) + .get(canvasId, allocation.desiredCollisionKey, nodeId); + if (conflict !== undefined) { + const conflictingNodeId = conflict['node_id']; + const collisionKey = conflict['label_collision_key']; + if (typeof conflictingNodeId !== 'string') { + throw new SyntaxError('Invalid conflicting SQLite Node id'); + } + const conflicting = decodeNodeRecord( + conflict['record_json'], + conflictingNodeId, + ); + return { + ok: false, + reason: 'label-conflict', + conflictingNodeId, + conflictingLabel: + typeof conflicting.label === 'string' + ? conflicting.label + : typeof collisionKey === 'string' + ? collisionKey + : conflictingNodeId, + }; + } + } + + const revision = randomUUID(); + database + .prepare( + `INSERT INTO nodes ( + canvas_id, node_id, record_json, revision, label_collision_key + ) VALUES (?, ?, ?, ?, ?) + ON CONFLICT(canvas_id, node_id) DO UPDATE SET + record_json = excluded.record_json, + revision = excluded.revision, + label_collision_key = excluded.label_collision_key`, + ) + .run( + canvasId, + nodeId, + stringifyJson(allocation.record, `Node ${JSON.stringify(nodeId)} record`), + revision, + allocation.collisionKey, + ); + return { + ok: true, + record: allocation.record, + revision, + }; +} + +export class SqliteSpaceNodes implements SpaceNodes { + readonly canvasId: string; + + readonly #context: SqliteStoreContext; + + constructor(context: SqliteStoreContext, canvasId: string) { + this.#context = context; + this.canvasId = canvasId; + } + + async read(nodeIdInput: string): Promise { + const nodeId = sanitizeId(nodeIdInput, 'nodeId'); + const current = readNodeRow( + this.#context.database(), + this.canvasId, + nodeId, + ); + return current === null + ? null + : { record: current.record, revision: current.revision }; + } + + async readMany( + nodeIds: readonly string[], + ): Promise> { + const database = this.#context.database(); + const snapshots = new Map(); + for (const nodeIdInput of new Set(nodeIds)) { + const nodeId = sanitizeId(nodeIdInput, 'nodeId'); + const row = readNodeRow(database, this.canvasId, nodeId); + if (row !== null) { + snapshots.set(nodeId, { + record: row.record, + revision: row.revision, + }); + } + } + return snapshots; + } + + async list(): Promise> { + const rows = this.#context + .database() + .prepare( + `SELECT node_id, record_json, revision, label_collision_key + FROM nodes + WHERE canvas_id = ?`, + ) + .all(this.canvasId); + const snapshots = new Map(); + for (const value of rows) { + const nodeId = value['node_id']; + if (typeof nodeId !== 'string') { + throw new SyntaxError('Invalid node_id in persisted SQLite Node'); + } + const row = decodeNodeRow(value, nodeId); + snapshots.set(nodeId, { + record: row.record, + revision: row.revision, + }); + } + return snapshots; + } + + async stream( + onNode: (snapshot: NodeSnapshot) => void, + options?: NodeStreamOptions, + ): Promise> { + const snapshots = await this.list(); + const delivered = new Map(); + for (const [nodeId, snapshot] of snapshots) { + if (options?.signal?.aborted) break; + onNode(snapshot); + delivered.set(nodeId, snapshot); + } + return delivered; + } + + async put(input: NodePutInput): Promise { + validatePut(input); + this.#context.assertMutationAllowed(this.canvasId); + const database = this.#context.database(); + return withImmediateTransaction(database, () => + putSqliteNodeInTransaction(database, this.canvasId, input), + ); + } + + async delete(nodeIdInput: string): Promise { + const nodeId = sanitizeId(nodeIdInput, 'nodeId'); + this.#context.assertMutationAllowed(this.canvasId); + const database = this.#context.database(); + return withImmediateTransaction(database, () => { + if (!spaceExists(database, this.canvasId)) return 'absent' as const; + const deleted = Number( + database + .prepare('DELETE FROM nodes WHERE canvas_id = ? AND node_id = ?') + .run(this.canvasId, nodeId).changes, + ); + return deleted === 1 ? ('deleted' as const) : ('absent' as const); + }); + } +} diff --git a/apps/server/src/modules/storage/backends/sqlite/space-repository.ts b/apps/server/src/modules/storage/backends/sqlite/space-repository.ts new file mode 100644 index 00000000..898a52b4 --- /dev/null +++ b/apps/server/src/modules/storage/backends/sqlite/space-repository.ts @@ -0,0 +1,272 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +import { randomUUID } from 'node:crypto'; + +import { withImmediateTransaction } from './database.js'; +import { allocateSpaceIdentity, collisionKeyForTitle } from './identity.js'; +import { + decodeSpaceRow, + insertSpaceRow, + readSpaceRow, + SPACE_COLUMNS, +} from './rows.js'; +import { sanitizeId } from '../../../../utils/fs.js'; + +import type { SqliteStoreContext } from './database.js'; +import type { CanvasFile } from '../../../canvas/persistence-types.js'; +import type { + SpaceBeginDeleteResult, + SpaceCreateInput, + SpaceCreateResult, + SpaceDeleteInput, + SpaceDeleteSession, + SpaceRenameInput, + SpaceRenameResult, + SpaceRepository, +} from '../../ports/structured.js'; +import type { CanvasSummary } from '@huabu/shared'; + +function validateTitle(title: unknown): asserts title is string | null { + if (title !== null && typeof title !== 'string') { + throw new TypeError('Space title must be a string or null'); + } +} + +export class SqliteSpaceRepository implements SpaceRepository { + readonly #context: SqliteStoreContext; + + constructor(context: SqliteStoreContext) { + this.#context = context; + } + + async list(): Promise { + const database = this.#context.database(); + return database + .prepare(`SELECT ${SPACE_COLUMNS} FROM spaces WHERE is_world = 0`) + .all() + .map((row) => { + const { record } = decodeSpaceRow(row); + return { + canvasId: record.canvasId, + title: record.title, + nodeCount: record.state.nodes.length, + createdAt: record.createdAt, + updatedAt: record.updatedAt, + }; + }); + } + + async worldId(): Promise { + const database = this.#context.database(); + const rows = database + .prepare(`SELECT ${SPACE_COLUMNS} FROM spaces WHERE is_world = 1`) + .all(); + if (rows.length !== 1) { + throw new Error( + rows.length === 0 + ? 'SQLite namespace has no World Space' + : 'SQLite namespace has multiple World Spaces', + ); + } + const world = decodeSpaceRow(rows[0]); + if (!world.isWorld) throw new Error('SQLite World Space is malformed'); + return sanitizeId(world.record.canvasId, 'world canvasId'); + } + + async ensureWorld(): Promise { + const database = this.#context.database(); + return withImmediateTransaction(database, () => { + const existing = database + .prepare(`SELECT ${SPACE_COLUMNS} FROM spaces WHERE is_world = 1`) + .all(); + if (existing.length > 1) { + throw new Error('SQLite namespace has multiple World Spaces'); + } + if (existing.length === 1) { + const world = decodeSpaceRow(existing[0]); + if (!world.isWorld) throw new Error('SQLite World Space is malformed'); + return sanitizeId(world.record.canvasId, 'world canvasId'); + } + + const canvasId = randomUUID(); + const timestamp = this.#context.now(); + if (!Number.isFinite(timestamp)) { + throw new TypeError('SQLite Space clock returned a non-finite value'); + } + insertSpaceRow( + database, + { + canvasId, + title: 'World', + version: 0, + state: { nodes: [], edges: [] }, + createdAt: timestamp, + updatedAt: timestamp, + }, + '', + true, + ); + return canvasId; + }); + } + + async create(input: SpaceCreateInput): Promise { + const canvasId = sanitizeId(input.canvasId, 'canvasId'); + validateTitle(input.title); + this.#context.assertMutationAllowed(canvasId); + const database = this.#context.database(); + + return withImmediateTransaction(database, () => { + if (readSpaceRow(database, canvasId) !== null) { + return { ok: false as const, reason: 'already-exists' as const }; + } + const occupied = database + .prepare('SELECT collision_key FROM spaces') + .all() + .map((row) => row['collision_key']) + .filter((value): value is string => typeof value === 'string'); + const identity = allocateSpaceIdentity(input.title, canvasId, occupied); + const timestamp = this.#context.now(); + if (!Number.isFinite(timestamp)) { + throw new TypeError('SQLite Space clock returned a non-finite value'); + } + const record: CanvasFile = { + canvasId, + title: identity.title, + version: 0, + state: { nodes: [], edges: [] }, + createdAt: timestamp, + updatedAt: timestamp, + }; + insertSpaceRow(database, record, identity.collisionKey); + return { ok: true as const, record }; + }); + } + + async beginDelete(input: SpaceDeleteInput): Promise { + const canvasId = sanitizeId(input.canvasId, 'canvasId'); + const beforeAdmission = readSpaceRow(this.#context.database(), canvasId); + if (beforeAdmission?.isWorld) { + return { ok: false, reason: 'world-forbidden' }; + } + + const release = await this.#context.acquireDelete(canvasId); + let sessionOwnsGate = false; + try { + const afterAdmission = readSpaceRow(this.#context.database(), canvasId); + if (afterAdmission?.isWorld) { + return { ok: false, reason: 'world-forbidden' }; + } + + let state: 'open' | 'finishing' | 'closed' = 'open'; + const close = (): void => { + if (state === 'closed') return; + state = 'closed'; + release(); + }; + const session: SpaceDeleteSession = Object.freeze({ + finish: async () => { + if (state !== 'open') { + throw new Error(`Space deletion session for ${canvasId} is closed`); + } + state = 'finishing'; + try { + this.#context.assertOpen(); + const database = this.#context.database(); + const result = withImmediateTransaction(database, () => { + const current = readSpaceRow(database, canvasId); + if (current?.isWorld) { + throw new Error(`Refusing to delete World Space ${canvasId}`); + } + if (current === null) { + return { + deleted: false, + }; + } + const deleted = Number( + database + .prepare('DELETE FROM spaces WHERE canvas_id = ?') + .run(canvasId).changes, + ); + return { deleted: deleted === 1 }; + }); + if (result.deleted) + return { ok: true as const, reason: 'deleted' as const }; + return { ok: false as const, reason: 'not-found' as const }; + } finally { + close(); + } + }, + abort: async () => { + if (state === 'finishing') { + throw new Error( + `Space deletion session for ${canvasId} is already finishing`, + ); + } + if (state === 'closed') return; + try { + this.#context.assertOpen(); + } finally { + close(); + } + }, + }); + sessionOwnsGate = true; + return { ok: true, session }; + } finally { + if (!sessionOwnsGate) release(); + } + } + + async rename(input: SpaceRenameInput): Promise { + const canvasId = sanitizeId(input.canvasId, 'canvasId'); + validateTitle(input.title); + this.#context.assertMutationAllowed(canvasId); + const database = this.#context.database(); + + return withImmediateTransaction(database, () => { + const current = readSpaceRow(database, canvasId); + if (current === null) return { ok: false, reason: 'not-found' } as const; + if (current.isWorld) { + return { ok: false, reason: 'world-forbidden' } as const; + } + if (current.record.title === input.title) { + return { ok: true, record: current.record } as const; + } + + const collisionKey = collisionKeyForTitle(input.title, canvasId); + if (collisionKey !== current.collisionKey) { + const conflict = database + .prepare( + `SELECT ${SPACE_COLUMNS} + FROM spaces + WHERE collision_key = ? AND canvas_id <> ?`, + ) + .get(collisionKey, canvasId); + if (conflict !== undefined) { + return { + ok: false, + reason: 'title-conflict', + conflictingTitle: decodeSpaceRow(conflict).record.title, + } as const; + } + } + + const result = database + .prepare( + `UPDATE spaces + SET title = ?, collision_key = ? + WHERE canvas_id = ?`, + ) + .run(input.title, collisionKey, canvasId); + if (Number(result.changes) !== 1) { + throw new Error(`Could not rename SQLite Space ${canvasId}`); + } + return { + ok: true, + record: { ...current.record, title: input.title }, + } as const; + }); + } +} diff --git a/apps/server/src/modules/storage/backends/sqlite/space-tasks.ts b/apps/server/src/modules/storage/backends/sqlite/space-tasks.ts new file mode 100644 index 00000000..4d27e2d1 --- /dev/null +++ b/apps/server/src/modules/storage/backends/sqlite/space-tasks.ts @@ -0,0 +1,250 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +import { + taskRecordSchema, + taskRunCompletionSchema, + taskRunRecordSchema, + taskStoreSnapshotSchema, + type TaskRecord, + type TaskRunCompletion, + type TaskRunRecord, + type TaskStoreSnapshot, +} from '@huabu/shared'; + +import { withImmediateTransaction } from './database.js'; +import { parseJson, stringifyJson } from './rows.js'; + +import type { SqliteStoreContext } from './database.js'; +import type { + SpaceTaskRuns, + SpaceTasks, + TaskRunCompletionResult, + TaskRunUpdate, +} from '../../ports/structured.js'; + +const EMPTY_TASKS: TaskStoreSnapshot = { + version: 1, + tasks: [], + runs: [], +}; + +function validateSnapshot(value: unknown, canvasId: string): TaskStoreSnapshot { + const parsed = taskStoreSnapshotSchema.safeParse(value); + if (!parsed.success) { + throw new SyntaxError( + `Invalid Task store for Canvas ${canvasId}: ${parsed.error.issues[0]?.message ?? 'schema violation'}`, + ); + } + const taskIds = new Set(); + for (const task of parsed.data.tasks) { + if (task.canvasId !== canvasId) { + throw new SyntaxError( + `Invalid Task store for Canvas ${canvasId}: Task ${task.taskId} belongs to Canvas ${task.canvasId}`, + ); + } + if (taskIds.has(task.taskId)) { + throw new SyntaxError( + `Invalid Task store for Canvas ${canvasId}: duplicate Task ${task.taskId}`, + ); + } + taskIds.add(task.taskId); + } + const runIds = new Set(); + for (const run of parsed.data.runs) { + if (run.canvasIdSnapshot !== canvasId) { + throw new SyntaxError( + `Invalid Task store for Canvas ${canvasId}: Run ${run.runId} belongs to Canvas ${run.canvasIdSnapshot}`, + ); + } + if (runIds.has(run.runId)) { + throw new SyntaxError( + `Invalid Task store for Canvas ${canvasId}: duplicate Run ${run.runId}`, + ); + } + if (!taskIds.has(run.taskId)) { + throw new SyntaxError( + `Invalid Task store for Canvas ${canvasId}: Run ${run.runId} references missing Task ${run.taskId}`, + ); + } + runIds.add(run.runId); + } + return parsed.data; +} + +function readSnapshot( + context: SqliteStoreContext, + canvasId: string, +): TaskStoreSnapshot { + const row = context + .database() + .prepare('SELECT snapshot_json FROM tasks WHERE canvas_id = ?') + .get(canvasId); + if (row === undefined) { + return { ...EMPTY_TASKS, tasks: [], runs: [] }; + } + return validateSnapshot( + parseJson(row['snapshot_json'], `Task store for Canvas ${canvasId}`), + canvasId, + ); +} + +export class SqliteSpaceTasks implements SpaceTasks { + readonly runs: SpaceTaskRuns; + + readonly #context: SqliteStoreContext; + readonly #canvasId: string; + + constructor(context: SqliteStoreContext, canvasId: string) { + this.#context = context; + this.#canvasId = canvasId; + this.runs = Object.freeze({ + create: (run: TaskRunRecord) => this.#createRun(run), + update: (runId: string, update: TaskRunUpdate) => + this.#updateRun(runId, update), + complete: ( + taskId: string, + runId: string, + completion: TaskRunCompletion, + ) => this.#completeRun(taskId, runId, completion), + }); + } + + async read(): Promise { + this.#context.assertOpen(); + return readSnapshot(this.#context, this.#canvasId); + } + + async create(task: TaskRecord): Promise { + const parsed = taskRecordSchema.safeParse(task); + if (!parsed.success || parsed.data.canvasId !== this.#canvasId) { + throw new TypeError(`Invalid Task record for Canvas ${this.#canvasId}`); + } + this.#mutate((snapshot) => { + if ( + snapshot.tasks.some( + (candidate) => candidate.taskId === parsed.data.taskId, + ) + ) { + throw new Error(`Task ${parsed.data.taskId} already exists`); + } + snapshot.tasks.push(parsed.data); + }); + } + + async #createRun(run: TaskRunRecord): Promise { + const parsed = taskRunRecordSchema.safeParse(run); + if (!parsed.success || parsed.data.canvasIdSnapshot !== this.#canvasId) { + throw new TypeError(`Invalid Run record for Canvas ${this.#canvasId}`); + } + this.#mutate((snapshot) => { + if ( + snapshot.runs.some((candidate) => candidate.runId === parsed.data.runId) + ) { + throw new Error(`Run ${parsed.data.runId} already exists`); + } + if ( + !snapshot.tasks.some( + (candidate) => candidate.taskId === parsed.data.taskId, + ) + ) { + throw new Error(`Task ${parsed.data.taskId} does not exist`); + } + snapshot.runs.push(parsed.data); + }); + } + + async #updateRun( + runId: string, + update: TaskRunUpdate, + ): Promise { + return this.#mutate((snapshot) => { + const index = snapshot.runs.findIndex((run) => run.runId === runId); + if (index < 0) throw new Error(`Run ${runId} does not exist`); + const parsed = taskRunRecordSchema.safeParse({ + ...snapshot.runs[index], + ...update, + }); + if (!parsed.success) { + throw new TypeError(`Invalid update for Run ${runId}`); + } + snapshot.runs[index] = parsed.data; + return parsed.data; + }); + } + + async #completeRun( + taskId: string, + runId: string, + completion: TaskRunCompletion, + ): Promise { + const parsedCompletion = taskRunCompletionSchema.safeParse(completion); + if (!parsedCompletion.success) { + throw new TypeError(`Invalid completion for Run ${runId}`); + } + return this.#mutate((snapshot) => { + if (!snapshot.tasks.some((task) => task.taskId === taskId)) { + return { outcome: 'task_not_found' }; + } + const index = snapshot.runs.findIndex((run) => run.runId === runId); + if (index < 0 || snapshot.runs[index]?.taskId !== taskId) { + return { outcome: 'run_not_found' }; + } + const current = snapshot.runs[index]; + if (!current) return { outcome: 'run_not_found' }; + if (current.status === 'completed') { + return current.completion?.message === parsedCompletion.data.message + ? { outcome: 'unchanged', run: current } + : { outcome: 'completion_conflict', run: current }; + } + if (current.status !== 'running') { + return { outcome: 'run_not_running', run: current }; + } + const parsedRun = taskRunRecordSchema.safeParse({ + ...current, + status: 'completed', + completion: parsedCompletion.data, + }); + if (!parsedRun.success) { + throw new TypeError(`Invalid completion update for Run ${runId}`); + } + snapshot.runs[index] = parsedRun.data; + return { outcome: 'completed', run: parsedRun.data }; + }); + } + + #mutate(apply: (snapshot: TaskStoreSnapshot) => T): T { + this.#context.assertMutationAllowed(this.#canvasId); + const database = this.#context.database(); + return withImmediateTransaction(database, () => { + if ( + database + .prepare('SELECT 1 AS present FROM spaces WHERE canvas_id = ?') + .get(this.#canvasId)?.['present'] !== 1 + ) { + throw new Error( + `Space Tasks(${this.#canvasId}) cannot write a missing Space`, + ); + } + const current = readSnapshot(this.#context, this.#canvasId); + const next: TaskStoreSnapshot = { + version: 1, + tasks: [...current.tasks], + runs: [...current.runs], + }; + const result = apply(next); + database + .prepare( + `INSERT INTO tasks (canvas_id, snapshot_json) + VALUES (?, ?) + ON CONFLICT(canvas_id) DO UPDATE SET + snapshot_json = excluded.snapshot_json`, + ) + .run( + this.#canvasId, + stringifyJson(next, `Task store for Canvas ${this.#canvasId}`), + ); + return result; + }); + } +} diff --git a/apps/server/src/modules/storage/backends/sqlite/space-write.ts b/apps/server/src/modules/storage/backends/sqlite/space-write.ts new file mode 100644 index 00000000..e97da423 --- /dev/null +++ b/apps/server/src/modules/storage/backends/sqlite/space-write.ts @@ -0,0 +1,181 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +import { withImmediateTransaction } from './database.js'; +import { allocateSpaceIdentity } from './identity.js'; +import { + insertSpaceRow, + readSpaceRow, + stringifyJson, + updateSpaceRow, + validateCanvasFile, + validateNodeContent, +} from './rows.js'; +import { putSqliteNodeInTransaction } from './space-nodes.js'; +import { sanitizeId } from '../../../../utils/fs.js'; + +import type { SqliteStoreContext } from './database.js'; +import type { + NodePutResult, + SpaceHandle, + SpaceNodeMutation, + SpaceWriteInput, + SpaceWriteResult, +} from '../../ports/structured.js'; + +function mutationError( + mutation: SpaceNodeMutation, + result: NodePutResult, +): Error { + const prefix = `Space write failed for node ${JSON.stringify(mutation.nodeId)}`; + if (result.ok) return new Error(`${prefix}: unexpected success result`); + switch (result.reason) { + case 'not-found': + return new Error(`${prefix}: Space does not exist`); + case 'revision-conflict': + return new Error(`${prefix}: unexpected revision conflict`); + case 'label-conflict': + return new Error( + `${prefix}: label conflicts with node ${JSON.stringify(result.conflictingNodeId)}`, + ); + case 'duplicate-node': + return new Error(`${prefix}: duplicate persisted node`); + case 'write-suppressed': + return new Error(`${prefix}: write is suppressed after deletion`); + } +} + +function validateInput(canvasId: string, input: SpaceWriteInput): void { + if (!Number.isFinite(input.expectedVersion)) { + throw new TypeError('expectedVersion must be a finite number'); + } + validateCanvasFile(input.nextRecord, canvasId); + if (input.nextRecord.version !== input.expectedVersion + 1) { + throw new Error( + `SpaceWrite(${canvasId}) expected nextRecord.version ` + + `${input.expectedVersion + 1}, received ${input.nextRecord.version}`, + ); + } + if ( + input.allowCreate === true && + (input.nodeMutations.length > 0 || input.delta !== undefined) + ) { + throw new Error( + 'allowCreate is valid only for a record-only structural write', + ); + } + if ( + input.delta !== undefined && + input.delta.version !== input.nextRecord.version + ) { + throw new Error( + 'delta.version must equal the committed Space record version', + ); + } + if (input.delta !== undefined) { + stringifyJson(input.delta, `Space ${JSON.stringify(canvasId)} delta`); + } + for (const mutation of input.nodeMutations) { + sanitizeId(mutation.nodeId, 'nodeId'); + if (mutation.kind === 'put') { + validateNodeContent(mutation.record, mutation.nodeId); + } + } +} + +/** Bind the atomic SQLite record/node/delta write to one Space. */ +export function createSqliteSpaceWrite( + context: SqliteStoreContext, + canvasId: string, +): SpaceHandle['write'] { + return async function write( + input: SpaceWriteInput, + ): Promise { + context.assertMutationAllowed(canvasId); + validateInput(canvasId, input); + const database = context.database(); + + const completed = withImmediateTransaction(database, () => { + const current = readSpaceRow(database, canvasId); + if (current === null) { + if (!input.allowCreate) { + return { ok: false, reason: 'not-found' } as const; + } + if (input.expectedVersion !== 0) { + throw new Error( + `SpaceWrite(${canvasId}) can create only from version 0`, + ); + } + const occupied = database + .prepare('SELECT collision_key FROM spaces') + .all() + .map((row) => row['collision_key']) + .filter((value): value is string => typeof value === 'string'); + const identity = allocateSpaceIdentity( + input.nextRecord.title, + canvasId, + occupied, + ); + insertSpaceRow( + database, + { ...input.nextRecord, title: identity.title }, + identity.collisionKey, + ); + return { ok: true } as const; + } + + if (current.record.version !== input.expectedVersion) { + return { + ok: false, + reason: 'version-conflict', + actualVersion: current.record.version, + } as const; + } + if (input.nextRecord.createdAt !== current.record.createdAt) { + throw new Error(`SpaceWrite(${canvasId}) refusing to change createdAt`); + } + if (input.nextRecord.title !== current.record.title) { + throw new Error( + `SpaceWrite(${canvasId}) cannot change title; ` + + 'use SpaceRepository.rename first', + ); + } + + for (const mutation of input.nodeMutations) { + if (mutation.kind === 'delete') { + database + .prepare('DELETE FROM nodes WHERE canvas_id = ? AND node_id = ?') + .run(canvasId, mutation.nodeId); + continue; + } + + const result = putSqliteNodeInTransaction(database, canvasId, { + nodeId: mutation.nodeId, + record: mutation.record, + strictLabel: mutation.strictLabel, + }); + if (!result.ok) throw mutationError(mutation, result); + } + + if ( + updateSpaceRow(database, input.nextRecord, input.expectedVersion) !== 1 + ) { + throw new Error(`SpaceWrite(${canvasId}) lost its version race`); + } + if (input.delta !== undefined) { + database + .prepare( + `INSERT INTO delta_log (canvas_id, version, entry_json) + VALUES (?, ?, ?)`, + ) + .run( + canvasId, + input.delta.version, + stringifyJson(input.delta, `Space ${canvasId} delta`), + ); + } + return { ok: true } as const; + }); + return completed; + }; +} diff --git a/apps/server/src/modules/storage/backends/sqlite/structured-store.ts b/apps/server/src/modules/storage/backends/sqlite/structured-store.ts new file mode 100644 index 00000000..a4844988 --- /dev/null +++ b/apps/server/src/modules/storage/backends/sqlite/structured-store.ts @@ -0,0 +1,70 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +import { SqliteStoreContext } from './database.js'; +import { readSpaceRow } from './rows.js'; +import { createSqliteSpaceExtension } from './space-extension.js'; +import { createSqliteSpaceLogs } from './space-logs.js'; +import { SqliteSpaceNodes } from './space-nodes.js'; +import { SqliteSpaceRepository } from './space-repository.js'; +import { SqliteSpaceTasks } from './space-tasks.js'; +import { createSqliteSpaceWrite } from './space-write.js'; +import { sanitizeId } from '../../../../utils/fs.js'; + +import type { StorageHealth } from '../../ports/common.js'; +import type { + SpaceHandle, + SpaceRepository, + StructuredStore, +} from '../../ports/structured.js'; + +/** Isolated structured-store adapter backed by one node:sqlite connection. */ +export class SqliteStructuredStore implements StructuredStore { + readonly kind = 'sqlite' as const; + + readonly #context: SqliteStoreContext; + + constructor(filename: string, now: () => number = Date.now) { + if (typeof filename !== 'string') { + throw new TypeError('SQLite filename must be a string'); + } + if (filename.length === 0) { + throw new TypeError('SQLite filename must not be empty'); + } + this.#context = new SqliteStoreContext(filename, now); + } + + async init(): Promise { + this.#context.init(); + } + + async health(): Promise { + return this.#context.health(this.kind); + } + + async close(): Promise { + this.#context.close(); + } + + spaces(): SpaceRepository { + return Object.freeze(new SqliteSpaceRepository(this.#context)); + } + + space(canvasIdInput: string): SpaceHandle { + const canvasId = sanitizeId(canvasIdInput, 'canvasId'); + const { events, changes } = createSqliteSpaceLogs(this.#context, canvasId); + const nodes = Object.freeze(new SqliteSpaceNodes(this.#context, canvasId)); + const tasks = Object.freeze(new SqliteSpaceTasks(this.#context, canvasId)); + return Object.freeze({ + canvasId, + read: async () => + readSpaceRow(this.#context.database(), canvasId)?.record ?? null, + write: createSqliteSpaceWrite(this.#context, canvasId), + nodes, + changes, + tasks, + events, + extension: createSqliteSpaceExtension(this.#context, canvasId), + }); + } +} diff --git a/apps/server/src/modules/storage/backends/sqlite/test-support.ts b/apps/server/src/modules/storage/backends/sqlite/test-support.ts new file mode 100644 index 00000000..4c5af703 --- /dev/null +++ b/apps/server/src/modules/storage/backends/sqlite/test-support.ts @@ -0,0 +1,201 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { DatabaseSync } from 'node:sqlite'; + +import { SQLITE_SCHEMA_VERSION } from './database.js'; +import { collisionKeyForTitle } from './identity.js'; +import { insertSpaceRow, parseJson } from './rows.js'; +import { SqliteStructuredStore } from './structured-store.js'; + +import type { + CanvasFile, + DeltaLogEntry, +} from '../../../canvas/persistence-types.js'; + +export const SQLITE_TEST_WORLD_ID = 'sqlite-test-world'; + +export interface SqliteTestFile { + readonly directory: string; + readonly filename: string; + readonly remove: () => void; +} + +export interface OpenSqliteTestStore extends SqliteTestFile { + readonly store: SqliteStructuredStore; + readonly world: CanvasFile; + readonly cleanup: () => Promise; +} + +export interface EmptySqliteTestStore extends SqliteTestFile { + readonly store: SqliteStructuredStore; + readonly cleanup: () => Promise; +} + +export function createSqliteTestFile(prefix = 'huabu-sqlite-'): SqliteTestFile { + const directory = mkdtempSync(path.join(tmpdir(), prefix)); + const filename = path.join(directory, 'structured.sqlite'); + let removed = false; + return { + directory, + filename, + remove: () => { + if (removed) return; + removed = true; + rmSync(directory, { recursive: true, force: true }); + }, + }; +} + +/** Run a short test-only query through a connection independent of the store. */ +export function withTestDatabase( + filename: string, + operation: (database: DatabaseSync) => T, +): T { + const database = new DatabaseSync(filename); + try { + database.exec('PRAGMA foreign_keys = ON'); + return operation(database); + } finally { + database.close(); + } +} + +/** + * Seed World without reaching through the adapter under test. + * + * The store first creates the production schema. This helper then opens a + * separate node:sqlite connection and uses the production row encoder, so a + * contract cannot pass because World creation accidentally shares private + * adapter state with the operation being exercised. + */ +export function seedSqliteWorld( + filename: string, + canvasId = SQLITE_TEST_WORLD_ID, +): CanvasFile { + const record: CanvasFile = { + canvasId, + title: 'World', + version: 0, + state: { nodes: [], edges: [] }, + createdAt: 1, + updatedAt: 1, + }; + withTestDatabase(filename, (database) => { + const version = database.prepare('PRAGMA user_version').get()?.[ + 'user_version' + ]; + if (version !== SQLITE_SCHEMA_VERSION) { + throw new Error( + `Expected production SQLite schema v${SQLITE_SCHEMA_VERSION}, got ${String(version)}`, + ); + } + insertSpaceRow( + database, + record, + collisionKeyForTitle(record.title, record.canvasId), + true, + ); + }); + return record; +} + +export async function openSqliteTestStore( + prefix = 'huabu-sqlite-', + now?: () => number, +): Promise { + const file = createSqliteTestFile(prefix); + const store = new SqliteStructuredStore(file.filename, now); + try { + await store.init(); + const world = seedSqliteWorld(file.filename); + return { + ...file, + store, + world, + cleanup: async () => { + await store.close(); + file.remove(); + }, + }; + } catch (error) { + await store.close(); + file.remove(); + throw error; + } +} + +export async function openEmptySqliteTestStore( + prefix = 'huabu-sqlite-empty-', + now?: () => number, +): Promise { + const file = createSqliteTestFile(prefix); + const store = new SqliteStructuredStore(file.filename, now); + try { + await store.init(); + return { + ...file, + store, + cleanup: async () => { + await store.close(); + file.remove(); + }, + }; + } catch (error) { + await store.close(); + file.remove(); + throw error; + } +} + +export function readSqliteDeltaLog( + filename: string, + canvasId: string, +): DeltaLogEntry[] { + return withTestDatabase(filename, (database) => + database + .prepare( + `SELECT entry_json + FROM delta_log + WHERE canvas_id = ? + ORDER BY version`, + ) + .all(canvasId) + .map( + (row, index) => + parseJson( + row['entry_json'], + `test delta row ${index} for ${canvasId}`, + ) as DeltaLogEntry, + ), + ); +} + +/** Install a real SQLite failure immediately before a delta row is inserted. */ +export function installDeltaAbortTrigger( + filename: string, + message: string, +): () => void { + const quotedMessage = message.split("'").join("''"); + withTestDatabase(filename, (database) => { + database.exec('DROP TRIGGER IF EXISTS test_abort_delta_insert'); + database.exec(` + CREATE TRIGGER test_abort_delta_insert + BEFORE INSERT ON delta_log + BEGIN + SELECT RAISE(ABORT, '${quotedMessage}'); + END + `); + }); + let restored = false; + return () => { + if (restored) return; + restored = true; + withTestDatabase(filename, (database) => { + database.exec('DROP TRIGGER IF EXISTS test_abort_delta_insert'); + }); + }; +} diff --git a/apps/server/src/modules/storage/capabilities.test.ts b/apps/server/src/modules/storage/capabilities.test.ts index 527dd8b2..12782907 100644 --- a/apps/server/src/modules/storage/capabilities.test.ts +++ b/apps/server/src/modules/storage/capabilities.test.ts @@ -58,7 +58,7 @@ describe('storage capability matrix', () => { expect(describeUnavailableCapabilities(DISK)).toEqual([]); }); - it('answers for a backend that has no adapter yet', () => { + it('answers for a backend whose adapter is not selectable yet', () => { const missing = unavailableCapabilities(TABLES); // Every entry is Disk-only today, so a structured backend that is not @@ -75,13 +75,12 @@ describe('storage capability matrix', () => { expect(hasStorageCapability(TABLES, 'something-portable')).toBe(true); }); - it('states a limitation without making it a misconfiguration', () => { - // A profile that merely offers fewer features must not fail validation — - // that is reserved for a backend that cannot serve at all. `sqlite` has - // no adapter yet, so it does fail; the distinction is which check - // rejects it. + it('reports capability gaps separately from profile selectability', () => { + // The matrix describes what SQLite lacks regardless of whether the + // preview can be selected. Validation rejects it at the separate + // production-readiness gate. expect(describeUnavailableCapabilities(TABLES).length).toBeGreaterThan(0); - expect(() => validateStorageProfile(TABLES)).toThrow(/not implemented/); + expect(() => validateStorageProfile(TABLES)).toThrow(/not selectable yet/); expect(() => validateStorageProfile(DISK)).not.toThrow(); }); diff --git a/apps/server/src/modules/storage/ports/contracts/space-nodes.contract.ts b/apps/server/src/modules/storage/ports/contracts/space-nodes.contract.ts index 3d70bbd8..f5649985 100644 --- a/apps/server/src/modules/storage/ports/contracts/space-nodes.contract.ts +++ b/apps/server/src/modules/storage/ports/contracts/space-nodes.contract.ts @@ -14,6 +14,8 @@ export interface SpaceNodesContractHarness { /** Repository scoped to a Space whose structural record is absent. */ readonly missingRepository: SpaceNodes; readonly expectedCanvasId: string; + /** Whether this adapter fences a deleted id against late standalone puts. */ + readonly deletedNodePut: 'allowed' | 'write-suppressed'; readonly cleanup?: () => Promise | void; } @@ -336,20 +338,24 @@ export function describeSpaceNodesContract( await expect(repository.delete(nodeId)).resolves.toBe('absent'); }); - it('suppresses a late standalone put after deletion', async () => { - const { repository } = await open(); + it('reports the adapter-defined result for a standalone put after deletion', async () => { + const { repository, deletedNodePut } = await open(); const nodeId = 'contract-late-put'; const record = note(nodeId, 'Contract late put', 'before'); await putSuccessfully(repository, { nodeId, record }); await repository.delete(nodeId); - await expect( - repository.put({ - nodeId, - record: { ...record, content: 'late resurrection' }, - }), - ).resolves.toEqual({ ok: false, reason: 'write-suppressed' }); - await expect(repository.read(nodeId)).resolves.toBeNull(); + const late = { ...record, content: 'late resurrection' }; + const result = await repository.put({ nodeId, record: late }); + if (deletedNodePut === 'write-suppressed') { + expect(result).toEqual({ ok: false, reason: 'write-suppressed' }); + await expect(repository.read(nodeId)).resolves.toBeNull(); + } else { + expect(result).toMatchObject({ ok: true, record: late }); + await expect(repository.read(nodeId)).resolves.toMatchObject({ + record: late, + }); + } }); }); } diff --git a/apps/server/src/modules/storage/ports/contracts/space-tasks.contract.ts b/apps/server/src/modules/storage/ports/contracts/space-tasks.contract.ts new file mode 100644 index 00000000..a02aba86 --- /dev/null +++ b/apps/server/src/modules/storage/ports/contracts/space-tasks.contract.ts @@ -0,0 +1,441 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +/** Reusable behavioral contract for {@link SpaceTasks} and its Runs. */ + +import { afterEach, describe, expect, it } from 'vitest'; + +import type { + SpaceDeleteSession, + SpaceTasks, + TaskRunUpdate, +} from '../structured.js'; +import type { TaskRecord, TaskRunRecord } from '@huabu/shared'; + +export interface SpaceTasksContractHarness { + /** Task ledger for an existing Space, initially empty. */ + readonly tasks: SpaceTasks; + /** A second retained handle for the same existing Space. */ + readonly concurrent: SpaceTasks; + readonly canvasId: string; + /** Task ledger scoped to a Space whose structural record is absent. */ + readonly missing: SpaceTasks; + readonly missingCanvasId: string; + /** Open a structured-deletion fence for `canvasId`. */ + readonly beginDelete: () => Promise; + readonly cleanup?: () => Promise | void; +} + +function task(canvasId: string, taskId: string, createdAt: number): TaskRecord { + return { + taskId, + canvasId, + goal: `Goal for ${taskId}`, + defaultRootProfileId: `profile-${taskId}`, + anchorNodeId: `anchor-${taskId}`, + createdAt, + }; +} + +function run( + canvasId: string, + taskId: string, + runId: string, + createdAt: number, +): TaskRunRecord { + return { + runId, + taskId, + canvasIdSnapshot: canvasId, + goalSnapshot: `Goal snapshot for ${taskId}`, + rootProfileIdSnapshot: `profile-${taskId}`, + status: 'pending', + createdAt, + }; +} + +export function describeSpaceTasksContract( + name: string, + createHarness: () => + | Promise + | SpaceTasksContractHarness, +): void { + describe(`SpaceTasks contract: ${name}`, () => { + let harness: SpaceTasksContractHarness | null = null; + + async function open(): Promise { + harness = await createHarness(); + return harness; + } + + afterEach(async () => { + await harness?.cleanup?.(); + harness = null; + }); + + it('reads an empty versioned snapshot', async () => { + const { tasks } = await open(); + + await expect(tasks.read()).resolves.toEqual({ + version: 1, + tasks: [], + runs: [], + }); + }); + + it('creates a Task and rejects a duplicate id without replacing it', async () => { + const { tasks, canvasId } = await open(); + const original = task(canvasId, 'task-duplicate', 1); + await tasks.create(original); + + await expect( + tasks.create({ ...original, goal: 'Replacement goal', createdAt: 2 }), + ).rejects.toThrow(); + await expect(tasks.read()).resolves.toEqual({ + version: 1, + tasks: [original], + runs: [], + }); + }); + + it('requires an existing Task before creating its Run', async () => { + const { tasks, canvasId } = await open(); + const owner = task(canvasId, 'task-owner', 1); + const ownedRun = run(canvasId, owner.taskId, 'run-owned', 2); + + await expect(tasks.runs.create(ownedRun)).rejects.toThrow(); + await expect(tasks.read()).resolves.toEqual({ + version: 1, + tasks: [], + runs: [], + }); + + await tasks.create(owner); + await tasks.runs.create(ownedRun); + await expect(tasks.read()).resolves.toEqual({ + version: 1, + tasks: [owner], + runs: [ownedRun], + }); + }); + + it('rejects a duplicate Run id without replacing it', async () => { + const { tasks, canvasId } = await open(); + const owner = task(canvasId, 'task-run-duplicate', 1); + const original = run(canvasId, owner.taskId, 'run-duplicate', 2); + await tasks.create(owner); + await tasks.runs.create(original); + + await expect( + tasks.runs.create({ + ...original, + status: 'running', + startedAt: 3, + }), + ).rejects.toThrow(); + await expect(tasks.read()).resolves.toEqual({ + version: 1, + tasks: [owner], + runs: [original], + }); + }); + + it('updates an existing Run and rejects a missing Run id', async () => { + const { tasks, canvasId } = await open(); + const owner = task(canvasId, 'task-update', 1); + const original = run(canvasId, owner.taskId, 'run-update', 2); + await tasks.create(owner); + await tasks.runs.create(original); + const update: TaskRunUpdate = { + rootNodeId: 'root-node', + rootThreadId: 'root-thread', + status: 'running', + startedAt: 3, + }; + + await expect(tasks.runs.update(original.runId, update)).resolves.toEqual({ + ...original, + ...update, + }); + await expect( + tasks.runs.update('run-missing', { status: 'running' }), + ).rejects.toThrow(); + await expect(tasks.read()).resolves.toEqual({ + version: 1, + tasks: [owner], + runs: [{ ...original, ...update }], + }); + }); + + it('completes only a running Run and keeps the first completion immutable', async () => { + const { tasks, concurrent, canvasId } = await open(); + const owner = task(canvasId, 'task-complete', 1); + const other = task(canvasId, 'task-complete-other', 2); + const original = run(canvasId, owner.taskId, 'run-complete', 3); + await tasks.create(owner); + await tasks.create(other); + await tasks.runs.create(original); + + await expect( + tasks.runs.complete(owner.taskId, original.runId, { completedAt: 4 }), + ).resolves.toMatchObject({ outcome: 'run_not_running', run: original }); + + await tasks.runs.update(original.runId, { + status: 'running', + startedAt: 5, + }); + await expect( + tasks.runs.complete(owner.taskId, original.runId, { + completedAt: 6, + message: 'Done', + }), + ).resolves.toMatchObject({ + outcome: 'completed', + run: { + status: 'completed', + completion: { completedAt: 6, message: 'Done' }, + }, + }); + await expect( + tasks.runs.complete(owner.taskId, original.runId, { + completedAt: 7, + message: 'Done', + }), + ).resolves.toMatchObject({ + outcome: 'unchanged', + run: { completion: { completedAt: 6, message: 'Done' } }, + }); + await expect( + tasks.runs.complete(owner.taskId, original.runId, { + completedAt: 8, + message: 'Different', + }), + ).resolves.toMatchObject({ outcome: 'completion_conflict' }); + await expect( + tasks.runs.complete('task-missing', original.runId, { completedAt: 9 }), + ).resolves.toEqual({ outcome: 'task_not_found' }); + await expect( + tasks.runs.complete(other.taskId, original.runId, { completedAt: 9 }), + ).resolves.toEqual({ outcome: 'run_not_found' }); + await expect( + tasks.runs.complete(owner.taskId, 'run-missing', { completedAt: 9 }), + ).resolves.toEqual({ outcome: 'run_not_found' }); + await expect(concurrent.read()).resolves.toEqual({ + version: 1, + tasks: [owner, other], + runs: [ + { + ...original, + status: 'completed', + startedAt: 5, + completion: { completedAt: 6, message: 'Done' }, + }, + ], + }); + }); + + it('serializes competing completions and persists exactly one winner', async () => { + const { tasks, concurrent, canvasId } = await open(); + const owner = task(canvasId, 'task-competing-completion', 1); + const original = run( + canvasId, + owner.taskId, + 'run-competing-completion', + 2, + ); + await tasks.create(owner); + await tasks.runs.create(original); + await tasks.runs.update(original.runId, { + status: 'running', + startedAt: 3, + }); + + const results = await Promise.all([ + tasks.runs.complete(owner.taskId, original.runId, { + completedAt: 4, + message: 'First candidate', + }), + concurrent.runs.complete(owner.taskId, original.runId, { + completedAt: 5, + message: 'Second candidate', + }), + ]); + expect(results.map((result) => result.outcome).sort()).toEqual([ + 'completed', + 'completion_conflict', + ]); + const completed = results.find( + (result) => result.outcome === 'completed', + ); + const conflict = results.find( + (result) => result.outcome === 'completion_conflict', + ); + if (completed?.outcome !== 'completed') { + throw new Error('Expected one completion winner'); + } + if (conflict?.outcome !== 'completion_conflict') { + throw new Error('Expected one completion conflict'); + } + expect(conflict.run).toEqual(completed.run); + await expect(concurrent.read()).resolves.toEqual({ + version: 1, + tasks: [owner], + runs: [completed.run], + }); + }); + + it('rejects Task and Run records scoped to another Space', async () => { + const { tasks, canvasId } = await open(); + const owner = task(canvasId, 'task-scope', 1); + + await expect( + tasks.create({ ...owner, canvasId: 'another-space' }), + ).rejects.toThrow(); + await tasks.create(owner); + await expect( + tasks.runs.create({ + ...run(canvasId, owner.taskId, 'run-scope', 2), + canvasIdSnapshot: 'another-space', + }), + ).rejects.toThrow(); + await expect(tasks.read()).resolves.toEqual({ + version: 1, + tasks: [owner], + runs: [], + }); + }); + + it('rejects malformed Task, Run, and Run-update input', async () => { + const { tasks, canvasId } = await open(); + const owner = task(canvasId, 'task-validation', 1); + const ownedRun = run(canvasId, owner.taskId, 'run-validation', 2); + + await expect(tasks.create({ ...owner, goal: '' })).rejects.toThrow(); + await tasks.create(owner); + await expect( + tasks.runs.create({ ...ownedRun, goalSnapshot: '' }), + ).rejects.toThrow(); + await tasks.runs.create(ownedRun); + await expect( + tasks.runs.update(ownedRun.runId, { startedAt: -1 }), + ).rejects.toThrow(); + await expect( + tasks.runs.complete(owner.taskId, ownedRun.runId, { + completedAt: -1, + }), + ).rejects.toThrow(); + await expect(tasks.read()).resolves.toEqual({ + version: 1, + tasks: [owner], + runs: [ownedRun], + }); + }); + + it('preserves concurrent mutations through two retained handles', async () => { + const { tasks, concurrent, canvasId } = await open(); + const taskA = task(canvasId, 'task-concurrent-a', 1); + const taskB = task(canvasId, 'task-concurrent-b', 2); + await Promise.all([tasks.create(taskA), concurrent.create(taskB)]); + + const runA = run(canvasId, taskA.taskId, 'run-concurrent-a', 3); + const runB = run(canvasId, taskB.taskId, 'run-concurrent-b', 4); + await Promise.all([ + tasks.runs.create(runA), + concurrent.runs.create(runB), + ]); + await Promise.all([ + concurrent.runs.update(runA.runId, { + status: 'running', + startedAt: 5, + }), + tasks.runs.update(runB.runId, { + status: 'running', + startedAt: 6, + }), + ]); + + const snapshot = await tasks.read(); + expect(snapshot.tasks.map((record) => record.taskId).sort()).toEqual([ + taskA.taskId, + taskB.taskId, + ]); + expect(snapshot.runs.map((record) => record.runId).sort()).toEqual([ + runA.runId, + runB.runId, + ]); + expect(snapshot.runs).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + runId: runA.runId, + status: 'running', + startedAt: 5, + }), + expect.objectContaining({ + runId: runB.runId, + status: 'running', + startedAt: 6, + }), + ]), + ); + }); + + it('rejects every mutation for a missing Space', async () => { + const { missing, missingCanvasId } = await open(); + const owner = task(missingCanvasId, 'task-missing-space', 1); + const ownedRun = run( + missingCanvasId, + owner.taskId, + 'run-missing-space', + 2, + ); + + await expect(missing.create(owner)).rejects.toThrow(); + await expect(missing.runs.create(ownedRun)).rejects.toThrow(); + await expect( + missing.runs.update(ownedRun.runId, { status: 'running' }), + ).rejects.toThrow(); + await expect( + missing.runs.complete(owner.taskId, ownedRun.runId, { + completedAt: 3, + }), + ).rejects.toThrow(); + }); + + it('rejects mutations while structured deletion is fenced', async () => { + const { tasks, canvasId, beginDelete } = await open(); + const owner = task(canvasId, 'task-delete-fence', 1); + const original = run(canvasId, owner.taskId, 'run-delete-fence', 2); + await tasks.create(owner); + await tasks.runs.create(original); + const before = await tasks.read(); + const session = await beginDelete(); + + try { + await expect( + tasks.create(task(canvasId, 'task-too-late', 3)), + ).rejects.toThrow(); + await expect( + tasks.runs.create(run(canvasId, owner.taskId, 'run-too-late', 4)), + ).rejects.toThrow(); + await expect( + tasks.runs.update(original.runId, { status: 'running' }), + ).rejects.toThrow(); + await expect( + tasks.runs.complete(owner.taskId, original.runId, { + completedAt: 5, + }), + ).rejects.toThrow(); + await expect(tasks.read()).resolves.toEqual(before); + } finally { + await session.abort(); + } + + await expect( + tasks.runs.update(original.runId, { + status: 'running', + startedAt: 5, + }), + ).resolves.toMatchObject({ status: 'running', startedAt: 5 }); + }); + }); +} diff --git a/apps/server/src/modules/storage/ports/contracts/structured-store.contract.ts b/apps/server/src/modules/storage/ports/contracts/structured-store.contract.ts index 9ede9188..42c08e6a 100644 --- a/apps/server/src/modules/storage/ports/contracts/structured-store.contract.ts +++ b/apps/server/src/modules/storage/ports/contracts/structured-store.contract.ts @@ -67,6 +67,7 @@ export function describeStructuredStoreContract( for (const method of [ 'list', 'worldId', + 'ensureWorld', 'create', 'beginDelete', 'rename', diff --git a/apps/server/src/modules/storage/ports/structured.ts b/apps/server/src/modules/storage/ports/structured.ts index 60332eeb..ee69234a 100644 --- a/apps/server/src/modules/storage/ports/structured.ts +++ b/apps/server/src/modules/storage/ports/structured.ts @@ -52,6 +52,7 @@ import type { TaskStoreSnapshot, } from '@huabu/shared'; import type { CanvasChangeRecord } from '@huabu/shared/canvas-engine'; +import type { DatabaseSync } from 'node:sqlite'; /** * Backends with a structured adapter today. @@ -61,7 +62,7 @@ import type { CanvasChangeRecord } from '@huabu/shared/canvas-engine'; * that are configurable but unimplemented — belongs to `profile.ts`, which * owns rejecting them with an actionable message. */ -export type StructuredBackendKind = 'disk'; +export type StructuredBackendKind = 'disk' | 'sqlite'; /** A connection to a structured backend. Process-wide; handles are derived. */ export interface StructuredStore { @@ -299,14 +300,23 @@ export interface SpaceHandle { * * One member per backend that exists, like {@link StructuredBackendKind} and * for the same reason: a union that named `sqlite` today would advertise a - * substrate no adapter can supply. It grows with each adapter — a table prefix - * for SQLite, a schema for Postgres — and an owner switches on `kind`. + * substrate no adapter can supply. It grows with each adapter — a scoped + * connection and parent id for SQLite, a schema for Postgres — and an owner + * switches on `kind`. */ -export type SpaceSubstrate = { - readonly kind: 'disk'; - /** A directory reserved for this namespace, created and ready to write. */ - readonly directory: string; -}; +export type SpaceSubstrate = + | { + readonly kind: 'disk'; + /** A directory reserved for this namespace, created and ready to write. */ + readonly directory: string; + } + | { + readonly kind: 'sqlite'; + /** The adapter connection on which the owner creates its own tables. */ + readonly database: DatabaseSync; + /** Stable parent row for owner tables to reference with ON DELETE CASCADE. */ + readonly extensionId: number; + }; // ─── The ordered Space write ───────────────────────────────────────────────── diff --git a/apps/server/src/modules/storage/profile.test.ts b/apps/server/src/modules/storage/profile.test.ts index 18be15c9..c5125500 100644 --- a/apps/server/src/modules/storage/profile.test.ts +++ b/apps/server/src/modules/storage/profile.test.ts @@ -64,7 +64,16 @@ describe('validateStorageProfile', () => { structured: { kind: 'postgres' }, blobs: { kind: 'disk' }, }), - ).toThrow(/not implemented yet.*disk/s); + ).toThrow(/not implemented yet.*disk, sqlite/s); + }); + + it('rejects an available preview adapter that is not selectable', () => { + expect(() => + validateStorageProfile({ + structured: { kind: 'sqlite' }, + blobs: { kind: 'disk' }, + }), + ).toThrow(/preview adapter.*not selectable yet.*Selectable: disk/s); }); it('rejects a known but unimplemented blob backend', () => { diff --git a/apps/server/src/modules/storage/profile.ts b/apps/server/src/modules/storage/profile.ts index f962e871..e045c72c 100644 --- a/apps/server/src/modules/storage/profile.ts +++ b/apps/server/src/modules/storage/profile.ts @@ -28,11 +28,20 @@ export interface StorageProfile { blobs: { kind: BlobBackendKind }; } +/** Backends with an adapter implementation, selectable or otherwise. */ +const AVAILABLE_STRUCTURED: readonly RequestedStructuredKind[] = [ + 'disk', + 'sqlite', +]; + /** - * Backends that exist today. Naming one that is not written yet must fail - * loudly rather than half-work. + * Backends whose complete capability matrix is safe for production use. + * + * SQLite deliberately stays out while product composition, Blob placement, + * Disk-only capabilities, and Workspace remounting still have one authority + * only in the Disk profile. */ -const IMPLEMENTED_STRUCTURED: readonly RequestedStructuredKind[] = ['disk']; +const SELECTABLE_STRUCTURED: readonly RequestedStructuredKind[] = ['disk']; const IMPLEMENTED_BLOBS: readonly BlobBackendKind[] = ['disk']; const STRUCTURED_KINDS: readonly RequestedStructuredKind[] = [ @@ -85,10 +94,11 @@ export function parseStorageProfile( /** * Reject profiles that cannot serve correctly, before any connection opens. * - * Today that means "named but not implemented". This is also where - * cross-axis rules belong as backends land — for example, Postgres paired - * with a node-local disk blob root is unsafe across replicas unless the - * path is a deliberately shared filesystem. + * Today that means either "named but not implemented" or "implemented only as + * an isolated preview". This is also where cross-axis rules belong as + * backends land — for example, Postgres paired with a node-local disk blob + * root is unsafe across replicas unless the path is a deliberately shared + * filesystem. * * A profile that merely offers *fewer features* is not rejected here. Those * are stated limitations rather than misconfigurations, and they are declared @@ -98,10 +108,17 @@ export function parseStorageProfile( * warning. */ export function validateStorageProfile(profile: StorageProfile): void { - if (!IMPLEMENTED_STRUCTURED.includes(profile.structured.kind)) { + if (!AVAILABLE_STRUCTURED.includes(profile.structured.kind)) { throw new StorageProfileError( `Structured backend "${profile.structured.kind}" is not implemented yet. ` + - `Available: ${IMPLEMENTED_STRUCTURED.join(', ')}.`, + `Adapters available: ${AVAILABLE_STRUCTURED.join(', ')}.`, + ); + } + if (!SELECTABLE_STRUCTURED.includes(profile.structured.kind)) { + throw new StorageProfileError( + `Structured backend "${profile.structured.kind}" has a preview adapter ` + + `but is not selectable yet. Required application capabilities still ` + + `depend on Disk. Selectable: ${SELECTABLE_STRUCTURED.join(', ')}.`, ); } if (!IMPLEMENTED_BLOBS.includes(profile.blobs.kind)) { diff --git a/docs/architecture/canvas-storage.md b/docs/architecture/canvas-storage.md index 2b5afc62..221ecb12 100644 --- a/docs/architecture/canvas-storage.md +++ b/docs/architecture/canvas-storage.md @@ -80,7 +80,7 @@ Key points: - The memory analyzer reads Space existence and at most 100 recent action events through one `SpaceHandle`. A missing Space skips the pass before reading memory files or calling the model; corrupt part data still fails the pass. Memory body/state files remain materialized workspace paths, while Agenetes-owned chat history is not part of the curator bundle. - **Chat history is Chat-V2, owned by Agenetes L2 — not `CanvasStore`.** The canonical per-thread conversation is a two-tier append-only log under `chat_v2/`: Tier-1 `.events.jsonl` (`AgentStreamEvent` deltas a running turn appends, written by `FileEventLogStore`) and Tier-2 `.turns.jsonl` (folded `AgentTurn`s, written by `FileTurnStore` — the only tier `history()` reads back). These files sit under the canvas `.history/` only because it is the Agenetes namespace `storage.root` (`canvasAcpNamespace(canvasId)`); `CanvasStore` never touches them. Do **not** confuse `chat_v2/.events.jsonl` (agent stream events) with the sibling `events.jsonl` (canvas action log) — same suffix, unrelated content. - Durable Agenetes workload records live in `.history/threads.json` (`agenetes-v2` schema, one record per thread; written by `FileThreadStore`). The host-local `namespace.storage.root` is never persisted: reads bind each record to the current Space namespace, so a Home synchronized across computers cannot redirect storage back to another machine's absolute path. -- Canonical Task and Run records live in `.history/tasks.json`, owned by Huabu Server through the async `SpaceTasks` ledger (`read`, `create`, and `runs.create`/`runs.update`). The Disk adapter validates the versioned snapshot and referential integrity on every read, rejects duplicate identifiers and Runs whose Task is absent, serializes read-modify-write operations with an independent per-Canvas process-local mutex, and atomically replaces the file. This mutex is intentionally separate from the Canvas topology write coordinator, so Task metadata does not participate in `space.json` version CAS. +- Canonical Task and Run records live in `.history/tasks.json`, owned by Huabu Server through the async `SpaceTasks` ledger (`read`, `create`, and `runs.create`/`runs.update`/`runs.complete`). The Disk adapter validates the versioned snapshot and referential integrity on every read, rejects duplicate identifiers and Runs whose Task is absent, serializes read-modify-write operations with an independent per-Canvas process-local mutex, and atomically replaces the file. This mutex is intentionally separate from the Canvas topology write coordinator, so Task metadata does not participate in `space.json` version CAS. - Legacy chat files are one-way migrated into `chat_v2/` at workspace activation and retired to `.bak`: the oldest pi-ai `Context` `chat/.json` via `migrate-chat-threads.ts` (hop 1), then the M5.6 `chat/.turns.jsonl` / `.active.json` via `migrate-chat-turns.ts` (hop 2). If hop 1 finds both formats after an interrupted launch, it completes a strict converted prefix atomically or preserves an existing tail when the full conversion is its prefix. Divergent logs are retained rather than guessed or overwritten; hop 2 skips the paired turn log while a valid same-thread legacy Context remains or its JSON cannot be read safely, so a later activation can retry both copies without blocking unrelated migrations. The obsolete `CanvasStore` chat methods and `chatPath()` helper were removed in Phase 2; `chatDir()` remains because change-review and agent-owned files still use that directory. ## 3. Storage composition and ownership @@ -103,7 +103,7 @@ Key points: | `index.ts` | Public exports only; application code imports here rather than reaching into an adapter. | | `canvas-store.ts`, `paths.ts`, `canvas-dirs.ts` | Deprecated forwarding shims with no logic, retained only for high-fanout compatibility imports. | -The Disk structured adapter and compatibility facade resolve the same cached legacy object, so migration does not create two in-memory authorities. All portable repository methods are async. `SpaceRepository` owns membership reads, structured create/rename, and an exclusive `beginDelete()` session; composition holds that session across the existing blob-first delete saga and then calls `finish()` or `abort()`. Every Space-record write goes through `SpaceHandle.write`, which is the version-checked replacement with the node and delta batch attached; `SpaceHandle.read` reads only. `SpaceNodes` returns complete records plus revision tokens without exposing filenames. `write` preserves the old node mutations → Space record → optional delta order. When a normal in-process node → record → delta batch rejects, the adapter must restore that batch's prestate before returning the rejection. An explicit title rename remains the preceding ordered, best-effort boundary and is not rolled back with the batch. The port does not promise process-crash or power-loss recovery, a determinate result after an unknown remote outcome, multi-process serialization, idempotent retry, or publication. Disk meets the in-process restoration requirement with its existing before-image rollback; a SQL adapter may use a native transaction. +The Disk structured adapter and compatibility facade resolve the same cached legacy object, so migration does not create two in-memory authorities. The SQLite adapter instead owns one explicit database filename and connection; retained handles stay bound to that connection, and its `init`, `health`, and `close` lifecycle is exercised only by direct tests. All portable repository methods are async. `SpaceRepository` owns membership reads, structured create/rename, and an exclusive `beginDelete()` session; composition holds that session across the existing blob-first delete saga and then calls `finish()` or `abort()`. Every Space-record write goes through `SpaceHandle.write`, which is the version-checked replacement with the node and delta batch attached; `SpaceHandle.read` reads only. `SpaceNodes` returns complete records plus revision tokens without exposing filenames. `write` preserves the old node mutations → Space record → optional delta order. When a normal in-process node → record → delta batch rejects, the adapter must restore that batch's prestate before returning the rejection. An explicit title rename remains the preceding ordered, best-effort boundary and is not rolled back with the batch. The port does not promise process-crash or power-loss recovery, a determinate result after an unknown remote outcome, multi-process serialization, idempotent retry, or publication. Disk meets the in-process restoration requirement with its existing before-image rollback; SQLite uses a native transaction. Canvas persistence DTOs and the write coordinator live under `modules/canvas/`; Workspace identity, durable membership, and Disk locators live under `modules/storage/`, while active-Workspace lifecycle and boot migrations remain under `modules/workspace/`; generic filesystem and Markdown codecs live under `utils/`. `canvasRoot()` validates the identifier and then verifies that the resolved Space directory remains a strict descendant of the active Workspace before any downstream Disk operation receives it. `module-boundaries.test.ts` enforces the storage dependency direction, prevents new consumers of the forwarding shims, and holds the neutrality guard: no production file outside `storage/` may import a Disk layout symbol or a legacy `CanvasStore` symbol. The check is import-level and symbol-level — a local variable that happens to be called `artifactPath` is not a violation, while importing `canvasRoot`, or reaching `getCanvasStore` through the barrel, is. Migrations are exempt because they rewrite frozen historical on-disk shapes; tests are exempt for the same reason they may name an adapter. @@ -133,7 +133,7 @@ The launch path deliberately has no compensation transaction. A launch failure l ### 3.3 Task Run completion -`RunCompletionService.complete()` validates the shared request and delegates the guarded transition to `SpaceTaskRuns.complete()`. The Disk adapter performs lookup, `running → completed`, and persistence under the existing per-Canvas Task mutation mutex, so HTTP and built-in-tool callers share one atomic transition rather than performing a read-then-update race. +`RunCompletionService.complete()` validates the shared request and delegates the guarded transition to `SpaceTaskRuns.complete()`. Both structured adapters perform lookup, `running → completed`, and persistence inside one Task-snapshot mutation boundary: Disk uses the per-Canvas Task mutex and atomic file replacement, while SQLite uses an immediate transaction. HTTP and built-in-tool callers therefore share one atomic transition rather than performing a read-then-update race. A completed Run stores immutable `completion.completedAt` and an optional trimmed caller-owned `completion.message`. The platform treats the message as untrusted text and does not interpret issue, pull-request, or URL semantics. A retry with the same normalized message is idempotent and preserves the original timestamp; a different message conflicts. A `pending` Run cannot complete, and Agent turn termination never implies Run completion. diff --git a/docs/proposals/multi-backend-storage.md b/docs/proposals/multi-backend-storage.md index c6d2371b..159e967c 100644 --- a/docs/proposals/multi-backend-storage.md +++ b/docs/proposals/multi-backend-storage.md @@ -1,7 +1,7 @@ # Multi-Backend Storage -Status: Phases 1–4.5 and §§12.6–12.8 implemented -Last updated: 2026-08-24 +Status: Phases 1–5 implemented; SQLite remains a contract preview +Last updated: 2026-09-04 > **Scope and decision confidence.** This proposal records the two-port > `StructuredStore` / `BlobStore` split and their target backend families as @@ -48,8 +48,7 @@ Last updated: 2026-08-24 > review are recorded in place, including the CAS race ordering (§12.2.5), > log-family interface segregation (§12.2.6), and retained-handle Workspace > guards (§12.2.4). Remaining Disk-only read and physical capabilities still -> keep non-Disk profiles unselectable. No SQLite, Postgres, or Azure adapter -> exists. +> keep non-Disk profiles unselectable. > > Phase 4.5 moved storage-owned Disk layout behind the storage boundary in > PR #93. What remains between the portable contracts and a second structured @@ -59,6 +58,11 @@ Last updated: 2026-08-24 > **implemented**), and §12.8 (the dispositions and the product-level > harness, **implemented**). §12 is the authoritative plan; > the decision table in §2 marks what each step has actually settled. +> +> Phase 5 is specified in §12.9 and is **implemented by this branch** as an +> isolated SQLite structured-store preview. It exercises the portable +> contracts with real SQLite files but is deliberately absent from runtime +> composition; Postgres and Azure adapters do not exist. --- @@ -88,7 +92,7 @@ built above these ports, but its form is intentionally unresolved here. | Topic | Status | Current position | | ------------------------------------------------------ | ------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | Separate authoritative structured and blob ports | **Accepted** (P1, merged) | Storage is composed from `StructuredStore` and `BlobStore`; there is no single backend interface that mixes both concerns. | -| Structured backend family | **Settled direction** | Support Disk, SQLite, and Postgres implementations. Only Disk exists. | +| Structured backend family | **Settled direction** | Support Disk, SQLite, and Postgres implementations. Disk is selectable; SQLite has an isolated contract-preview adapter but is not selectable; Postgres has no adapter. | | Blob backend family | **Settled direction** | Support Disk and Azure Blob implementations. Only Disk exists. | | Independent composition | **Accepted** (P1, merged) | `StorageProfile` has two env-parsed axes; `validateStorageProfile` fails fast on unimplemented kinds and is the extension point for combination rules. The lazy `getStorage()` path now rejects profiles whose adapters require awaited initialization (§12.1.1). | | Blob port contract | **Accepted** (P1, merged) | Connection → scope, stream-oriented, no permanent absolute path in the common contract; `materialize()` returns a bounded lease for the one consumer needing a file. Replacement atomicity and post-release lease semantics are contract terms, not adapter accidents (§6.2, §12.1.1). | @@ -144,8 +148,9 @@ external-note discovery watches `nodes/`, and export archives the entire Space directory. Therefore wrapping `CanvasStore` in a database adapter would not by itself make the application backend-neutral. -Canvas/Space persistence is currently Disk-only. SQLite, Postgres, and Azure -Blob adapters for this data do not yet exist. +Runtime Canvas/Space persistence remains Disk-only. An isolated SQLite +structured adapter exists for contract and integration tests, while Postgres +and Azure Blob adapters do not yet exist. ## 4. Goals @@ -166,7 +171,9 @@ Blob adapters for this data do not yet exist. ## 5. Non-goals -- Selecting an ORM, SQL query builder, Postgres driver, or SQLite driver. +- Selecting a production ORM, SQL query builder, Postgres driver, or final + SQLite driver. The isolated Phase 5 preview uses built-in `node:sqlite` + without making that production choice. - Defining the final relational schema or migration framework. - Choosing a VFS, FUSE, materialization, cache, or write-back design. - Replacing RFS or the canonical `SpaceQuery` / `CanvasCommand` contracts in @@ -175,9 +182,9 @@ Blob adapters for this data do not yet exist. their product semantics are defined. - Implementing online backend migration, replication, backup, or disaster recovery. -- Shipping any non-Disk adapter. The phases in §12 remove reasons why SQLite, - Postgres, and Azure _cannot_ be implemented; that is not the same as - implementing them. +- Making a non-Disk adapter runtime-selectable. Phase 5 proves an isolated + adapter against the contracts without registering it in composition or + changing product capabilities. ## 6. Settled backend split and implemented minimum contracts @@ -301,8 +308,9 @@ into place makes the failed write invisible instead of unremovable. ### 6.3 Composition -Configuration has two axes. The current shape carries only a backend kind per -axis, because no adapter yet needs more: +Configuration has two axes. The runtime-selectable profile carries only a +backend kind per axis. The isolated SQLite preview receives its explicit +database filename directly and is not constructed from this profile: ```ts interface StorageProfile { @@ -325,7 +333,8 @@ node-local DiskBlob implementation is unsafe in a multi-replica deployment unless the path is a deliberately shared and supported filesystem. SQLite on a network filesystem has different correctness and availability constraints from local SQLite. `validateStorageProfile()` is where such rules live; today it -rejects kinds that are named but not implemented, so an unsupported profile +rejects recognized kinds that are unavailable or deliberately unselectable, +including SQLite's preview-specific diagnostic, so an unsupported profile fails at startup with an actionable message rather than nondeterministically while serving data. @@ -684,7 +693,7 @@ exceptions: one names what it returns, the other opens a session. ```ts interface StructuredStore { - readonly kind: StructuredBackendKind; // 'disk' — implemented adapters only + readonly kind: StructuredBackendKind; // 'disk' | 'sqlite'; only Disk is selectable init(): Promise; health(): Promise; @@ -746,7 +755,7 @@ interface SpaceChanges { interface SpaceTasks { read(): Promise; // Tasks and Runs in one snapshot create(task: TaskRecord): Promise; - readonly runs: SpaceTaskRuns; // create(run), update(runId, patch) + readonly runs: SpaceTaskRuns; // create, update, and atomic complete } ``` @@ -943,9 +952,10 @@ explicitly: ## 12. Migration plan -Phases 1–4 are implemented and specified below. Phase 5 onward keeps the -provisional character of the original outline: those entries record intended -order, not approved designs. +Phases 1–4.5 are implemented and merged. Phase 5 is implemented by this +isolated contract preview. Phase 6 onward keeps the provisional character of +the original outline: those entries record intended order, not approved +designs. The current on-disk format remains readable throughout port extraction. A database adapter must not require Disk consumers to simulate tables, and the @@ -1807,12 +1817,16 @@ justify. footing and was left alone as Phase-1 surface. Not changed, deliberately: `authoritativeInsert` and the `write-suppressed` -put outcome remain in the portable shapes. Both exist for Disk's in-memory -deletion fence, and neither has a portable meaning a SQL adapter would -produce. They are now documented as adapter-shaped, the way `duplicate-node` -already was, rather than renamed or pushed behind the adapter — the honest -resolution needs a second adapter to say what the shared abstraction is, and -inventing one now would be the same speculative move this trim is undoing. +put outcome remain in the portable shapes. At this phase boundary, both +existed for Disk's in-memory deletion fence and a second adapter was still +needed to establish their shared meaning. + +**Resolved by Phase 5:** the SQLite contract preview supplies that second +adapter and confirms that these outcomes are adapter-shaped. Disk keeps its +process-local anti-resurrection fence and uses `authoritativeInsert` to lift +it. SQLite deletion is final at transaction commit, permits immediate reuse of +the primary key, and issues a fresh opaque revision so a token from the +deleted row cannot win a later compare-and-swap (§12.9.2). The review also asked composition to move default-title allocation ("Untitled", "Untitled (1)", …) into `create`, which would have removed the @@ -1961,10 +1975,11 @@ bearing: change-review records and Tasks are not history, whatever Disk's arrives, the group comes back — and `events` is where it was before, so nothing else has to move. -### 12.5 Phase 4.5 — storage-owned layout moves inside the boundary — **implemented** +### 12.5 Phase 4.5 — storage-owned layout moves inside the boundary — **merged** -Phase 5 adds a second structured backend. Before it does, the layout knowledge -that belongs to the _Disk_ backend has to stop living outside `storage/`. +Phase 5 would introduce a second structured backend. Before that work, the +layout knowledge that belongs to the _Disk_ backend had to stop living outside +`storage/`. Otherwise every later backend inherits a module named `disk` as the ambient description of where Spaces are, and each one pays to migrate the same callers again. @@ -2047,11 +2062,11 @@ substrate-specific but fails the test for the same reason — it exists so Windows can rename a Space _directory_ safely, and under SQLite there is no such rename. -`naming.ts` is misfiled in a different way: pure string logic with no I/O, -already re-exported rather than owned. It passes the test trivially (a second -backend needs the identical rules) but has no business behind a `disk` -segment. Phase 5 extracts it to `utils/naming.ts` as a side effect of needing -it twice; that extraction belongs here, where it is the point. +`naming.ts` was misfiled in a different way: pure string logic with no I/O, +already re-exported rather than owned. It passed the test trivially (a second +backend needs the identical rules) but had no business behind a `disk` +segment. Phase 4.5 extracted it to `utils/naming.ts`, where the shared rule has +a backend-neutral owner. Because the residue that survives the test is three setting helpers and `getWorkspacePath()` itself — none of it filesystem-specific — the target is a @@ -2128,8 +2143,9 @@ boundary test; behavior parity is asserted by the existing Disk suites, which must pass unchanged — a diff that alters a Disk test's expectations is out of scope by definition. -Phase 5 rebases onto this and drops its `utils/naming.ts` extraction, its -`workspace/disk/naming.ts` shim, and the corresponding roadmap edits. +Phase 5 builds on this merged result and carries none of its former +`utils/naming.ts` extraction, `workspace/disk/naming.ts` shim, or parallel +roadmap edits. **Landed for the Workspace-to-storage substrate move.** `modules/workspace/` is flat and holds `paths.ts` plus `migrations/`; the Disk record layout, blob @@ -2477,14 +2493,15 @@ temporary Workspace through the production lifecycle — prepared Workspace, opened connections, `ensureWorld()` — rather than swapping in a stub. A stub proves the application talks to an interface; only a real backend proves one serves the product, which is the half that decides whether a second adapter -works. `product-boundary.test.ts` runs the criterion against every profile in -`PRODUCT_STORAGE_PROFILES`, naming no directory, filename, or `space.json`; -Phase 5 adds one entry to that list and the same behaviours are covered for -SQLite. A guard reads the suite's own source and rejects a directory, a +works. `product-boundary.test.ts` runs the criterion against every selectable +profile in `PRODUCT_STORAGE_PROFILES`, naming no directory, filename, or +`space.json`. A guard reads the suite's own source and rejects a directory, a filename, or a `readFileSync` appearing in it, because the failure mode here is a helpful-looking assertion someone adds later. The records the suite reads back are built through the write engine, because a fixture that skips the -engine asserts nothing about what the product actually stores. +engine asserts nothing about what the product actually stores. The isolated +Phase 5 adapter runs the lower-level portable contracts; it does not enter +this product-profile harness until it becomes selectable. `closeStorage()` arrives with it, registered on graceful Server shutdown and used by the harness between profiles. On Disk it is close to a no-op — which @@ -2502,21 +2519,82 @@ bundle export, external-note claim), RFS's sidecar-to-record mapping (**B**, deferred until a second backend has a file plane at all), and the ACP session path that leaves with the Agenetes `Namespace` change. -Out of scope, unchanged: a SQLite adapter or schema, Disk→SQLite data -migration, SQLite profile registration, Postgres/Azure, the portable +Out of scope, unchanged: SQLite runtime composition and profile selection, +Disk→SQLite data migration, Postgres/Azure, the portable change-notification capability, RFS's backend-neutral path vocabulary, ACP session relocation, the rest of the Agenetes persistence migration, the portable export format, a writable general-purpose virtual filesystem or OS mount, protocol or UI changes, and stronger crash/distributed transaction guarantees. -### 12.9 Later phases — provisional +### 12.9 Phase 5 — SQLite contract preview — **implemented** + +Phase 5 adds one non-Disk structured adapter to test whether the boundary +survives a database implementation. It is an isolated implementation and test +target, not a product profile. The composition root does not construct or +export it, and `HUABU_STRUCTURED_BACKEND=sqlite` continues to fail during +profile validation with a preview-specific diagnostic. + +#### 12.9.1 Scope and lifecycle + +- The adapter uses built-in `node:sqlite`, owns one explicit database filename + and connection, and adds no package or native-addon dependency. +- Retained handles stay bound to that connection. `init`, `health`, and + `close` are real lifecycle operations; Workspace remounting and production + factory registration remain selectability work. +- The current portable surface is implemented: Space listing/lifecycle and + `ensureWorld`, record read/write, node read/readMany/list/stream and + mutations, events, changes, Tasks/Runs including atomic completion, and the + extension substrate. +- Postgres, Azure Blob, Disk-to-SQLite migration, RFS/file tools, external-note + watching, import/export, client/API changes, and product UI remain outside + this phase. + +#### 12.9.2 Schema and behavior + +Schema versioning uses `PRAGMA user_version`; migrations run transactionally, +reject databases from the future, and create `STRICT` tables with foreign keys +enabled. Version 1 stores Space records and World membership, complete node +JSON with opaque revision tokens, ordered events, coalesced changes, +Task/Run snapshots, extension namespaces, and the private delta journal. + +Every ordered Space write applies node mutations, record replacement, and the +optional delta insert in one immediate transaction. Same-baseline writers have +one winner. Space deletion uses the shared process-local admission coordinator +under a database-specific scope: reads remain available, mutations reject, +concurrent deletion sessions queue, and `finish()` removes all owned rows by +foreign-key cascade. No SQL transaction remains open across blob cleanup, and +no multi-process deletion fence is promised. + +SQLite does not emulate Disk's node tombstones. A committed delete immediately +frees the `(canvas_id, node_id)` key. Every successful put receives a new UUID +revision token, including a delete/recreate cycle, so a stale token from the +old row cannot match the replacement. `write-suppressed` and +`authoritativeInsert` remain valid adapter-specific parts of the common shape +for Disk rather than requirements every SQL adapter must reproduce. + +The extension substrate returns the shared connection plus a stable, +Space-owned namespace id. Owner tables can reference that id with +`ON DELETE CASCADE`, preserving namespace isolation and cleanup without +putting a generic key/value API in the storage port. + +#### 12.9.3 Proof + +The reusable structured contracts run against Disk and real temporary SQLite +files. They cover fresh World bootstrap, store lifecycle, Space deletion +admission, node CAS and read shapes, ordered transactional writes, event/change +ordering, Task/Run completion, and extension isolation and cleanup. SQLite +integration tests additionally cover strict schema creation, close/reopen +persistence, an immutable v1 fixture, future-version rejection, migration +rollback, SQL fault injection, foreign-key cascades, and revision safety across +delete/recreate. + +The preview changes only the storage implementation, contracts, focused type +narrowing for the new substrate union, and this documentation. Runtime +composition and product capability owners remain unchanged. + +### 12.10 Later phases — provisional -5. Add one new adapter at a time — SQLite, then Postgres, then Azure Blob — - running the same contract suites, migration fixtures, failure injection, - and concurrency tests against each. An adapter may exist for isolated - testing before its backend profile is selectable; profile validation keeps - rejecting it until the required capability matrix is satisfied. 6. Migrate the currently synchronous Agenetes persistence ports without changing their persist-before-notify, sequence, and fencing semantics. 7. Refactor RFS and built-in file tools only after a logical file-view contract @@ -2730,18 +2808,19 @@ Before a new backend is production-ready: persistence ownership, namespace, sequence, and replay invariants. - [Agenetes-Agentlet Gateway Consolidation](./agenetes-agentlet-gateway-consolidation.md) — records removal of the old Agentlet SQLite session store; it must not be - confused with the proposed SQLite structured backend. + confused with the SQLite structured contract-preview backend. ## 17. Code entry points | File/dir | Responsibility | | -------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| [`apps/server/src/modules/storage/`](../../apps/server/src/modules/storage/) | Ports, composition, adapters, compatibility, tests, and three forwarding shims — the canonical Phase-1–4 tree (§§12.1–12.4), guarded by `module-boundaries.test.ts`. | +| [`apps/server/src/modules/storage/`](../../apps/server/src/modules/storage/) | Ports, composition, adapters, compatibility, tests, and three forwarding shims — the canonical Phase-1–5 tree (§§12.1–12.9), guarded by `module-boundaries.test.ts`. | | [`apps/server/src/modules/storage/ports/`](../../apps/server/src/modules/storage/ports/) | The two ports; reusable suites live in `ports/contracts/`. `blob.ts` is normative (§7.1); `structured.ts` owns the Space collection and the per-Space handle: record read/write, nodes, changes, Tasks, and history. | | [`apps/server/src/modules/storage/storage.ts`](../../apps/server/src/modules/storage/storage.ts) | Composition root: maps profiles to adapters, guards blob puts, and holds a lifecycle deletion session across the blob-first cleanup saga. | | [`.../storage/backends/disk/legacy/canvas-store-cache.ts`](../../apps/server/src/modules/storage/backends/disk/legacy/canvas-store-cache.ts) | Bounded LRU of legacy Disk Space objects. The single owner both the adapter and the facade resolve through, and the real limit of `space(id)` identity (§12.2.4). | | [`apps/server/src/modules/storage/profile.ts`](../../apps/server/src/modules/storage/profile.ts) | Two-axis backend selection from env, and the fail-fast validation hook for unsupported combinations. | | [`apps/server/src/modules/storage/backends/disk/`](../../apps/server/src/modules/storage/backends/disk/) | Every Disk implementation: blob/structured stores, the Space collection, and the per-Space record, node, log, and Task adapters, in-process batch restoration, and the legacy class under `legacy/`. | +| [`apps/server/src/modules/storage/backends/sqlite/`](../../apps/server/src/modules/storage/backends/sqlite/) | Isolated `node:sqlite` structured adapter, strict schema and migrations, transaction-backed writes, and real-file contract/integration tests; available for proof but not runtime-selectable. | | [`.../storage/compatibility/canvas.ts`](../../apps/server/src/modules/storage/compatibility/canvas.ts) | Residual Disk read surface plus direct-module lifecycle test fixtures; production structured mutations enumerated in §12.4 use the portable ports. | | [`apps/server/src/modules/agent/memory/analyzer.ts`](../../apps/server/src/modules/agent/memory/analyzer.ts) | P3 repository consumer for strict Space existence, bounded action events, and intent episodes; physical chat and memory files remain Disk-specific. | | [`apps/server/src/modules/canvas/write-coordinator.ts`](../../apps/server/src/modules/canvas/write-coordinator.ts) | Canvas mutation coordinator and per-Space write lock, held across asynchronous node read, revision CAS, and put. |