import { readFileSync, readdirSync, existsSync, statSync, accessSync, constants, writeFileSync, renameSync, mkdirSync, rmSync } from 'node:fs'; import { open } from 'node:fs/promises'; import { dirname, join, resolve } from 'node:path'; import { uuid } from './ids.js'; let DatabaseSync = null; try { ({ DatabaseSync } = await import('node:sqlite')); } catch {} export class WeaveError extends Error { constructor(message, code = 'weave-error') { super(message); this.code = code; } } export async function streamToFileDurable(source, path, max) { mkdirSync(dirname(path), { recursive: true }); const handle = await open(path, 'w'); let size = 0; try { for await (const chunk of source) { size += chunk.length; if (size > max) throw new WeaveError(`Body too large: ${max} bytes at most`, 'too-large'); await handle.write(chunk); } await handle.sync(); } finally { await handle.close(); } return size; } export function writeFileDurable(path, bytes, options = {}) { mkdirSync(dirname(path), { recursive: true }); const temporary = `${path}.${uuid()}.tmp`; try { writeFileSync(temporary, bytes, { ...options, flush: true }); renameSync(temporary, path); } finally { rmSync(temporary, { force: true }); } } const SCHEMA = ` CREATE TABLE IF NOT EXISTS weave_meta (id INTEGER PRIMARY KEY CHECK (id = 1), json TEXT NOT NULL); CREATE TABLE IF NOT EXISTS spaces (id TEXT PRIMARY KEY, json TEXT NOT NULL); CREATE TABLE IF NOT EXISTS tables (id TEXT PRIMARY KEY, json TEXT NOT NULL); CREATE TABLE IF NOT EXISTS entities ( id TEXT PRIMARY KEY, db_id TEXT NOT NULL, public_id INTEGER, updated_at TEXT, json TEXT NOT NULL); CREATE INDEX IF NOT EXISTS idx_entities_db ON entities(db_id); CREATE UNIQUE INDEX IF NOT EXISTS idx_entities_pid ON entities(db_id, public_id); CREATE TABLE IF NOT EXISTS automations (id TEXT PRIMARY KEY, json TEXT NOT NULL); CREATE TABLE IF NOT EXISTS audit_log ( seq INTEGER PRIMARY KEY AUTOINCREMENT, at TEXT NOT NULL, actor TEXT, action TEXT, detail TEXT); CREATE TABLE IF NOT EXISTS undo_log ( seq INTEGER PRIMARY KEY AUTOINCREMENT, json TEXT NOT NULL, actor TEXT); CREATE TABLE IF NOT EXISTS doc_revisions ( seq INTEGER PRIMARY KEY AUTOINCREMENT, entity_id TEXT NOT NULL, field_id TEXT NOT NULL, at TEXT NOT NULL, actor TEXT, text TEXT NOT NULL, len INTEGER NOT NULL, restored_from INTEGER); CREATE INDEX IF NOT EXISTS idx_doc_revisions ON doc_revisions(entity_id, field_id, seq); CREATE TABLE IF NOT EXISTS file_storage (file_id TEXT PRIMARY KEY, at TEXT NOT NULL, json TEXT NOT NULL); `; const LOAD_ORDER = { automations: " ORDER BY json_extract(json, '$.seq') IS NULL, json_extract(json, '$.seq'), rowid", }; const UNDO_CAP = 200; export const DOC_REVISION_CAP = 200; function addRestoredFrom(exec) { try { exec('ALTER TABLE doc_revisions ADD COLUMN restored_from INTEGER'); } catch (err) { if (!/duplicate column/i.test(err.message)) throw err; } } function undoByActor(db) { if (!db.prepare("SELECT 1 FROM pragma_table_info('undo_log') WHERE name = 'actor'").get()) { db.exec('BEGIN IMMEDIATE'); try { db.exec('ALTER TABLE undo_log ADD COLUMN actor TEXT'); db.exec("UPDATE undo_log SET actor = json_extract(json, '$.actor')"); db.exec('COMMIT'); } catch (err) { db.exec('ROLLBACK'); throw err; } } db.exec('CREATE INDEX IF NOT EXISTS idx_undo_actor ON undo_log(actor, seq)'); } function uniquePublicIds(db) { const index = db.prepare("SELECT sql FROM sqlite_master WHERE type = 'index' AND name = 'idx_entities_pid'").get(); if (/UNIQUE/i.test(index?.sql ?? '')) return; db.exec('BEGIN IMMEDIATE'); try { const dupes = db.prepare(`SELECT e.id, e.db_id, e.public_id FROM entities e WHERE e.public_id IS NOT NULL AND EXISTS (SELECT 1 FROM entities o WHERE o.db_id = e.db_id AND o.public_id = e.public_id AND o.rowid < e.rowid) ORDER BY e.rowid`).all(); const top = db.prepare('SELECT MAX(public_id) AS n FROM entities WHERE db_id = ?'); const move = db.prepare("UPDATE entities SET public_id = ?, json = json_set(json, '$.publicId', ?) WHERE id = ?"); const counter = db.prepare(`UPDATE tables SET json = json_set(json, '$.publicIdCounter', ?) WHERE id = ? AND COALESCE(json_extract(json, '$.publicIdCounter'), 0) < ?`); const tableName = db.prepare("SELECT json_extract(json, '$.name') AS name FROM tables WHERE id = ?"); const audit = db.prepare("INSERT INTO audit_log (at, actor, action, detail) VALUES (?, NULL, 'public-id-renumbered', ?)"); const at = new Date().toISOString(); for (const { id, db_id, public_id } of dupes) { const next = (top.get(db_id).n ?? 0) + 1; move.run(next, next, id); counter.run(next, db_id, next); audit.run(at, JSON.stringify({ table: tableName.get(db_id)?.name ?? null, tableId: db_id, entity: id, from: public_id, to: next })); } db.exec('DROP INDEX IF EXISTS idx_entities_pid'); db.exec('CREATE UNIQUE INDEX idx_entities_pid ON entities(db_id, public_id)'); db.exec('COMMIT'); } catch (err) { db.exec('ROLLBACK'); throw err; } } function claimPublicIds(db, tables, entities, ids) { const holder = db.prepare('SELECT id FROM entities WHERE db_id = ? AND public_id = ?'); const top = db.prepare('SELECT MAX(public_id) AS n FROM entities WHERE db_id = ?'); const claimed = new Map(); let moved = 0; for (const id of ids) { const e = entities[id]; if (!e || e.publicId == null) continue; if (!claimed.has(e.dbId)) claimed.set(e.dbId, new Set()); const mine = claimed.get(e.dbId); const held = holder.get(e.dbId, e.publicId)?.id; const leaving = held && ids.has(held) && entities[held] === undefined; const table = tables[e.dbId]; const stored = top.get(e.dbId).n ?? 0; if ((held && held !== id && !leaving) || mine.has(e.publicId)) { e.publicId = Math.max(stored, table?.publicIdCounter ?? 0, ...mine) + 1; moved++; } mine.add(e.publicId); if (table) table.publicIdCounter = Math.max(table.publicIdCounter ?? 0, stored, e.publicId); } return moved; } function isWorkspaceShape(data) { return data && typeof data === 'object' && data.meta && (data.tables != null || data.databases != null); } export class Store { #db = null; #dataVersion = null; #cache = null; #batch = false; constructor(path = null) { this.legacyJsonPath = null; if (path && path.endsWith('.json')) { this.legacyJsonPath = path; path = path.slice(0, -5) + '.db'; } this.path = path; } load() { if (!this.path) return null; if (existsSync(this.path)) return this.#open(); let legacyState = null; if (this.legacyJsonPath && existsSync(this.legacyJsonPath)) { try { legacyState = JSON.parse(readFileSync(this.legacyJsonPath, 'utf8')); } catch { throw new WeaveError(`'${this.legacyJsonPath}' is not valid JSON`, 'invalid'); } if (!isWorkspaceShape(legacyState)) { throw new WeaveError(`'${this.legacyJsonPath}' is not a Weave workspace file`, 'invalid'); } } const state = this.#open(); if (legacyState && !state) { this.save(legacyState, { all: true }); return this.#loadState(); } return state; } #openFailure(err) { const dir = dirname(this.path); try { accessSync(dir, constants.W_OK); } catch { const who = typeof process.getuid === 'function' ? ` as uid ${process.getuid()}` : ''; const missing = !existsSync(dir); return `the data directory '${dir}' is ${missing ? 'missing' : 'not writable'}${who}, ` + `so weave cannot open '${this.path}' (${err.message}). ` + (missing ? 'Create it, or mount the volume there, and give it to the user weave runs as.' : 'Give it to the user weave runs as, or on Railway set RAILWAY_RUN_UID=0 on the service and redeploy.'); } return `'${this.path}' is not a SQLite database (${err.message})`; } #open() { if (!DatabaseSync) { throw new WeaveError( `weave requires Node >= 22.16 — node:sqlite is missing in ${process.version}. ` + 'Upgrade Node (24 LTS recommended) and retry.', 'invalid'); } const existed = existsSync(this.path); let db; try { db = new DatabaseSync(this.path); db.exec('PRAGMA busy_timeout = 5000'); db.exec('PRAGMA journal_mode = WAL'); db.exec('PRAGMA synchronous = FULL'); } catch (err) { throw new WeaveError(this.#openFailure(err), 'invalid'); } if (existed) { const names = db.prepare("SELECT name FROM sqlite_master WHERE type = 'table'").all().map((r) => r.name); if (names.length && !names.includes('weave_meta')) { db.close(); throw new WeaveError(`'${this.path}' is a SQLite database but not a Weave workspace`, 'invalid'); } } db.exec(SCHEMA); addRestoredFrom((q) => db.exec(q)); undoByActor(db); uniquePublicIds(db); if (db.prepare("SELECT 1 FROM sqlite_master WHERE name = 'entities_fts'").get()) db.exec('DROP TABLE entities_fts'); this.#db = db; this.#dataVersion = this.#pragmaDataVersion(); return this.#loadState(); } #loadState() { const db = this.#db; const metaRow = db.prepare('SELECT json FROM weave_meta WHERE id = 1').get(); if (!metaRow) { this.#cache = { meta: null, spaces: new Map(), tables: new Map(), automations: new Map() }; return null; } const state = { ...JSON.parse(metaRow.json), spaces: {}, tables: {}, entities: {}, automations: {} }; const cache = { meta: metaRow.json, spaces: new Map(), tables: new Map(), automations: new Map() }; for (const key of ['spaces', 'tables', 'automations']) { for (const row of db.prepare(`SELECT id, json FROM ${key}${LOAD_ORDER[key] ?? ''}`).all()) { state[key][row.id] = JSON.parse(row.json); cache[key].set(row.id, row.json); } } for (const row of db.prepare('SELECT id, json FROM entities').all()) { state.entities[row.id] = JSON.parse(row.json); } this.#cache = cache; return state; } #memAudit = []; audit(entry) { if (!this.#db) { this.#memAudit.push({ seq: this.#memAudit.length + 1, ...entry }); return; } this.#db.prepare('INSERT INTO audit_log (at, actor, action, detail) VALUES (?, ?, ?, ?)') .run(entry.at, entry.actor, entry.action, JSON.stringify(entry.detail ?? {})); } #memUndo = []; pushUndo(entry) { const actor = entry.actor ?? null; if (!this.#db) { this.#memUndo.push(entry); const mine = this.#memUndo.filter((u) => (u.actor ?? null) === actor); if (mine.length > UNDO_CAP) this.#memUndo.splice(this.#memUndo.indexOf(mine[0]), 1); return; } this.#db.prepare('INSERT INTO undo_log (json, actor) VALUES (?, ?)').run(JSON.stringify(entry), actor); this.#db.prepare(`DELETE FROM undo_log WHERE actor IS ? AND seq <= (SELECT seq FROM undo_log WHERE actor IS ? ORDER BY seq DESC LIMIT 1 OFFSET ?)`).run(actor, actor, UNDO_CAP); } popUndo(actor = null) { if (!this.#db) { const at = this.#memUndo.findLastIndex((u) => (u.actor ?? null) === actor); return at < 0 ? null : this.#memUndo.splice(at, 1)[0]; } const row = this.#db.prepare('SELECT seq, json FROM undo_log WHERE actor IS ? ORDER BY seq DESC LIMIT 1').get(actor); if (!row) return null; this.#db.prepare('DELETE FROM undo_log WHERE seq = ?').run(row.seq); return JSON.parse(row.json); } listUndo({ limit = 20, actor = null } = {}) { if (!this.#db) return this.#memUndo.filter((u) => (u.actor ?? null) === actor).slice(-limit).reverse(); return this.#db.prepare('SELECT json FROM undo_log WHERE actor IS ? ORDER BY seq DESC LIMIT ?').all(actor, limit) .map((r) => JSON.parse(r.json)); } #memRevs = []; #memRevSeq = 0; pushDocRevision({ entityId, fieldId, at, actor = null, text, restoredFrom = null }) { const len = text.length; if (!this.#db) { const row = { seq: ++this.#memRevSeq, entityId, fieldId, at, actor, text, len, restoredFrom }; this.#memRevs.push(row); const mine = this.#memRevs.filter((r) => r.entityId === entityId && r.fieldId === fieldId); if (mine.length > DOC_REVISION_CAP) { const drop = new Set(mine.slice(0, mine.length - DOC_REVISION_CAP).map((r) => r.seq)); this.#memRevs = this.#memRevs.filter((r) => !drop.has(r.seq)); } return { seq: row.seq }; } const { lastInsertRowid } = this.#db.prepare('INSERT INTO doc_revisions (entity_id, field_id, at, actor, text, len, restored_from) VALUES (?, ?, ?, ?, ?, ?, ?)') .run(entityId, fieldId, at, actor, text, len, restoredFrom); this.#db.prepare(`DELETE FROM doc_revisions WHERE entity_id = ? AND field_id = ? AND seq NOT IN (SELECT seq FROM doc_revisions WHERE entity_id = ? AND field_id = ? ORDER BY seq DESC LIMIT ?)`) .run(entityId, fieldId, entityId, fieldId, DOC_REVISION_CAP); return { seq: Number(lastInsertRowid) }; } replaceDocRevision(seq, { at, text }) { if (!this.#db) { const row = this.#memRevs.find((r) => r.seq === seq); if (row) Object.assign(row, { at, text, len: text.length }); return; } this.#db.prepare('UPDATE doc_revisions SET at = ?, text = ?, len = ? WHERE seq = ?').run(at, text, text.length, seq); } listDocRevisions(entityId, fieldId, { limit = 50 } = {}) { if (!this.#db) { return this.#memRevs.filter((r) => r.entityId === entityId && r.fieldId === fieldId) .slice(-limit).reverse().map(({ seq, at, actor, len, restoredFrom }) => ({ seq, at, actor, len, restoredFrom })); } return this.#db.prepare('SELECT seq, at, actor, len, restored_from AS restoredFrom FROM doc_revisions WHERE entity_id = ? AND field_id = ? ORDER BY seq DESC LIMIT ?') .all(entityId, fieldId, limit); } getDocRevision(entityId, fieldId, seq) { if (!this.#db) { const r = this.#memRevs.find((x) => x.seq === seq && x.entityId === entityId && x.fieldId === fieldId); return r ? { seq: r.seq, at: r.at, actor: r.actor, len: r.len, restoredFrom: r.restoredFrom, text: r.text } : null; } return this.#db.prepare('SELECT seq, at, actor, len, restored_from AS restoredFrom, text FROM doc_revisions WHERE seq = ? AND entity_id = ? AND field_id = ?') .get(seq, entityId, fieldId) ?? null; } deleteDocRevisions(entityId) { if (!this.#db) { this.#memRevs = this.#memRevs.filter((r) => r.entityId !== entityId); return; } this.#db.prepare('DELETE FROM doc_revisions WHERE entity_id = ?').run(entityId); } #memFileStorage = new Map(); getFileStorage(fileId) { if (!this.#db) return this.#memFileStorage.get(fileId) ?? null; return this.#db.prepare('SELECT at, json FROM file_storage WHERE file_id = ?').get(fileId) ?? null; } setFileStorage(fileId, { at, json }) { if (!this.#db) { this.#memFileStorage.set(fileId, { at, json }); return; } this.#db.prepare('INSERT INTO file_storage (file_id, at, json) VALUES (?, ?, ?) ON CONFLICT(file_id) DO UPDATE SET at = excluded.at, json = excluded.json').run(fileId, at, json); } deleteFileStorage(fileId) { if (!this.#db) { this.#memFileStorage.delete(fileId); return; } this.#db.prepare('DELETE FROM file_storage WHERE file_id = ?').run(fileId); } siblingNames(blobId) { if (!this.path || !DatabaseSync) return false; const own = resolve(this.path); const dir = dirname(own); let names; try { names = readdirSync(dir); } catch { return true; } for (const n of names) { const p = join(dir, n); if (!n.endsWith('.db') || p === own) continue; let db; try { db = new DatabaseSync(p, { readOnly: true }); db.exec('PRAGMA busy_timeout = 5000'); const tables = new Set(db.prepare("SELECT name FROM sqlite_master WHERE type = 'table'").all().map((r) => r.name)); if (tables.has('entities') && db.prepare('SELECT 1 FROM entities WHERE instr(json, ?) > 0 LIMIT 1').get(blobId)) return true; if (tables.has('weave_meta') && db.prepare('SELECT 1 FROM weave_meta WHERE instr(json, ?) > 0 LIMIT 1').get(blobId)) return true; } catch { return true; } finally { db?.close(); } } return false; } setAuditDetail(seq, detail) { if (!this.#db) { const row = this.#memAudit.find((r) => r.seq === seq); if (row) row.detail = structuredClone(detail); return; } this.#db.prepare('UPDATE audit_log SET detail = ? WHERE seq = ?').run(JSON.stringify(detail ?? {}), seq); } listAudit({ limit = 100, offset = 0, actions = null } = {}) { if (!this.#db) { const rows = this.#memAudit.slice().reverse().filter((r) => !actions || actions.includes(r.action)); return rows.slice(offset, limit < 0 ? undefined : offset + limit) .map((r) => ({ ...r, detail: structuredClone(r.detail ?? {}) })); } const where = actions?.length ? ` WHERE action IN (${actions.map(() => '?').join(', ')})` : ''; return this.#db.prepare(`SELECT seq, at, actor, action, detail FROM audit_log${where} ORDER BY seq DESC LIMIT ? OFFSET ?`) .all(...(where ? actions : []), limit, offset) .map((r) => ({ ...r, detail: JSON.parse(r.detail ?? '{}') })); } batch(fn) { if (!this.path || this.#batch) return fn(); if (!this.#db) this.#open(); this.#db.exec('BEGIN IMMEDIATE'); this.#batch = true; try { const result = fn(); this.#db.exec('COMMIT'); return result; } catch (err) { this.#db.exec('ROLLBACK'); this.#loadState(); throw err; } finally { this.#batch = false; } } save(state, { dirty = null, all = false } = {}) { if (!this.path) return; if (!this.#db) this.#open(); const db = this.#db; let moved = 0; if (!this.#batch) db.exec('BEGIN IMMEDIATE'); try { const { spaces, tables, entities, automations, ...rest } = state; const ids = all ? new Set([...Object.keys(entities), ...db.prepare('SELECT id FROM entities').all().map((r) => r.id)]) : (dirty ?? new Set()); moved = all ? 0 : claimPublicIds(db, tables, entities, ids); const metaJson = JSON.stringify(rest); if (metaJson !== this.#cache.meta) { db.prepare('INSERT INTO weave_meta (id, json) VALUES (1, ?) ON CONFLICT(id) DO UPDATE SET json = excluded.json').run(metaJson); this.#cache.meta = metaJson; } for (const [key, collection] of [['spaces', spaces], ['tables', tables], ['automations', automations]]) { const cache = this.#cache[key]; for (const [id, obj] of Object.entries(collection)) { const json = JSON.stringify(obj); if (cache.get(id) === json) continue; db.prepare(`INSERT INTO ${key} (id, json) VALUES (?, ?) ON CONFLICT(id) DO UPDATE SET json = excluded.json`).run(id, json); cache.set(id, json); } for (const id of [...cache.keys()]) { if (collection[id] === undefined) { db.prepare(`DELETE FROM ${key} WHERE id = ?`).run(id); cache.delete(id); } } } const dropEntity = db.prepare('DELETE FROM entities WHERE id = ?'); const putEntity = db.prepare(`INSERT INTO entities (id, db_id, public_id, updated_at, json) VALUES (?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET db_id = excluded.db_id, public_id = excluded.public_id, updated_at = excluded.updated_at, json = excluded.json`); for (const id of [...ids].sort((a, b) => (entities[a] === undefined ? 0 : 1) - (entities[b] === undefined ? 0 : 1))) { const e = entities[id]; if (e === undefined) dropEntity.run(id); else putEntity.run(e.id, e.dbId, e.publicId ?? null, e.updatedAt ?? null, JSON.stringify(e)); } if (!this.#batch) db.exec('COMMIT'); } catch (err) { if (!this.#batch) { db.exec('ROLLBACK'); this.#loadState(); } throw err; } return moved; } sizeBytes() { if (!this.path) return null; let total = 0; for (const p of [this.path, `${this.path}-wal`, `${this.path}-shm`]) { try { total += statSync(p).size; } catch {} } return total; } #pragmaDataVersion() { return this.#db.prepare('PRAGMA data_version').get().data_version; } changedExternally() { if (!this.#db) return false; const v = this.#pragmaDataVersion(); const changed = v !== this.#dataVersion; this.#dataVersion = v; return changed; } maxPublicId(dbId) { if (!this.#db) return 0; const row = this.#db.prepare(`SELECT MAX(COALESCE((SELECT MAX(public_id) FROM entities WHERE db_id = ?), 0), COALESCE((SELECT json_extract(json, '$.publicIdCounter') FROM tables WHERE id = ?), 0)) AS n`).get(dbId, dbId); return Number(row?.n ?? 0); } reload() { if (!this.#db) return null; return this.#loadState(); } close() { if (this.#db) { this.#db.close(); this.#db = null; } } }