src/store.js

Rama main · réplica de lectura del repositorio original.

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;
    }
  }
}