Table

tcmts.server.Table · #xz6ahykf · class

A typed, durable table: rows of T in .tcm/data/store.sqlite, by id. The row type is erased at runtime, but tcm knows it where the table is declared and attaches its schema (with a hash) to the table. Every write is checked against it and records the schema it was written with. A read of a row written under another schema checks that it still fits, else migrates it, else refuses: stored data never silently disagrees with the code reading it. Rows from newer code mean the caller is a stale page (410).

import { compareFields, fieldSql, HttpError, isScalar, sqlValue } from "#codebase/tcmts/server";
import type { FindOptions, RowSchema, TableOptions } from "#codebase/tcmts/server";
import { checkValue, database, decode, encode, exclusive, randomId } from "#tcm/host";

export class Table<T> {
  readonly name: string;
  readonly #steps: ((old: unknown) => unknown)[];
  readonly #unique: string[];
  readonly #index: string[];
  readonly #renamedFrom?: string;
  #prepared?: ReturnType<typeof database>;
  constructor(name: string, opts: TableOptions<T> = {}) {
    if (!/^[\w.-]+$/.test(name)) throw new Error(`invalid table name '${name}'`);
    this.name = name;
    this.#steps = (
      opts.migrate === undefined ? [] : Array.isArray(opts.migrate) ? opts.migrate : [opts.migrate]
    ) as ((old: unknown) => unknown)[];
    if (this.#steps.some((f) => typeof f !== "function"))
      throw new TypeError(`table "${name}": migrate is a function, or an array of them`);
    this.#unique = [...(opts.unique ?? [])];
    this.#index = [...(opts.index ?? [])];
    for (const f of [...this.#unique, ...this.#index])
      if (!/^[A-Za-z_$][\w$]*$/.test(f))
        throw new Error(`table "${name}": can't index '${f}': fields are plain names`);
    if (
      opts.renamedFrom !== undefined &&
      (!/^[\w.-]+$/.test(opts.renamedFrom) || opts.renamedFrom === name)
    )
      throw new Error(`table "${name}": invalid renamedFrom '${opts.renamedFrom}'`);
    this.#renamedFrom = opts.renamedFrom;
  }
  #schema(): RowSchema {
    const s = (this as unknown as Record<symbol, RowSchema | undefined>)[Symbol.for("tcm.schema")];
    if (!s)
      throw new Error(
        `table "${this.name}" doesn't know its row type: declare it at the top level with one, e.g. export const ${this.name} = table<Row>("${this.name}")`,
      );
    return s;
  }
  #db(): ReturnType<typeof database> {
    const db = database("store");
    if (this.#prepared !== db) {
      const s = this.#schema();
      db.prepare(
        "INSERT OR IGNORE INTO schemas (tbl, hash, schema, first_seen) VALUES (?, ?, ?, ?)",
      ).run(this.name, s.hash, JSON.stringify({ s: s.s, defs: s.defs }), Date.now());
      const id = (f: string, kind: string) =>
        `tcm_${kind}_${this.name.replace(/[^A-Za-z0-9]/g, (c) => `_${c.charCodeAt(0).toString(16)}_`)}__${f}`;
      for (const f of this.#index)
        db.exec(
          `CREATE INDEX IF NOT EXISTS "${id(f, "index")}" ON rows (${fieldSql(f)}) WHERE tbl = '${this.name}'`,
        );
      for (const f of this.#unique) {
        try {
          db.exec(
            `CREATE UNIQUE INDEX IF NOT EXISTS "${id(f, "unique")}" ON rows (${fieldSql(f)}) WHERE tbl = '${this.name}'`,
          );
        } catch {
          const dup = db
            .prepare(
              `SELECT ${fieldSql(f)} AS v, count(*) AS n FROM rows WHERE tbl = '${this.name}' GROUP BY v HAVING n > 1 LIMIT 1`,
            )
            .get() as { v: unknown; n: number } | undefined;
          throw new Error(
            `table "${this.name}": can't make ${f} unique: ${dup ? `${dup.n} stored rows have ${f} = ${JSON.stringify(dup.v)}` : "stored rows share values"}`,
          );
        }
      }
      this.#prepared = db;
    }
    // rows still under the old name move over: all of them the first time, and any an older build writes since
    const old = this.#renamedFrom;
    if (old && db.prepare("SELECT 1 FROM rows WHERE tbl = ? LIMIT 1").get(old)) {
      this.#atomically(db, () => {
        db.prepare(
          "INSERT OR IGNORE INTO schemas (tbl, hash, schema, first_seen) SELECT ?, hash, schema, first_seen FROM schemas WHERE tbl = ?",
        ).run(this.name, old);
        db.prepare("UPDATE OR IGNORE rows SET tbl = ? WHERE tbl = ?").run(this.name, old);
      });
    }
    return db;
  }
  /** A table call: it waits for any transaction it isn't part of (the host's lock on the store). */
  #op<R>(fn: (db: ReturnType<typeof database>) => Promise<R>): Promise<R> {
    return exclusive(database("store"), () => fn(this.#db()));
  }
  #atomically<R>(db: ReturnType<typeof database>, fn: () => R): R {
    if (db.isTransaction) {
      db.exec("SAVEPOINT tcm_atomic");
      try {
        const r = fn();
        db.exec("RELEASE tcm_atomic");
        return r;
      } catch (e) {
        db.exec("ROLLBACK TO tcm_atomic");
        db.exec("RELEASE tcm_atomic");
        throw e;
      }
    }
    db.exec("BEGIN IMMEDIATE");
    try {
      const r = fn();
      db.exec("COMMIT");
      return r;
    } catch (e) {
      db.exec("ROLLBACK");
      throw e;
    }
  }
  #check(value: unknown, what: string) {
    const s = this.#schema();
    const bad = checkValue(s.s, value, what, s.defs);
    if (bad) throw new TypeError(`table "${this.name}": ${bad}`);
  }
  #fits(value: unknown): boolean {
    const s = this.#schema();
    return checkValue(s.s, value, "row", s.defs) === undefined;
  }
  #write(db: ReturnType<typeof database>, id: string, value: unknown) {
    // an upsert on the id only: a clash on a unique field is an error, never a silent replacement of the other row
    try {
      db.prepare(
        "INSERT INTO rows (tbl, id, data, schema_hash, updated) VALUES (?, ?, ?, ?, ?) ON CONFLICT (tbl, id) DO UPDATE SET data = excluded.data, schema_hash = excluded.schema_hash, updated = excluded.updated",
      ).run(this.name, id, encode(value, `a ${this.name} row`), this.#schema().hash, Date.now());
    } catch (e) {
      if (!/UNIQUE constraint failed/.test(String((e as Error)?.message))) throw e;
      for (const f of this.#unique) {
        const v = (value as Record<string, unknown>)[f];
        const other = db
          .prepare(
            `SELECT id FROM rows WHERE tbl = '${this.name}' AND ${fieldSql(f)} = ? AND id != ?`,
          )
          .get(sqlValue(v), id) as { id: string } | undefined;
        if (other)
          throw new HttpError(
            409,
            `table "${this.name}": another row (${other.id}) already has ${f} = ${JSON.stringify(v)}`,
          );
      }
      throw e;
    }
  }
  /** Run a value written as an older type through the migrate steps: each takes the values its parameter type fits. */
  async #migrate(value: unknown, id: string): Promise<unknown> {
    const params = this.#schema().steps ?? [];
    const refusal = (i: number, v: unknown) =>
      params[i] ? checkValue(params[i]!.s, v, "row", params[i]!.defs) : undefined;
    let v = value;
    for (let from = 0; ;) {
      let i = from;
      while (i < this.#steps.length && refusal(i, v) !== undefined) i++;
      if (i === this.#steps.length) {
        if (from > 0) return v; // the caller checks the result against the row type
        const why = this.#steps.map(
          (f, j) => `step ${j + 1}${f.name ? ` (${f.name})` : ""}: ${refusal(j, v)}`,
        );
        throw new Error(
          `table "${this.name}": row ${id} was written as an older type that no migrate step takes (${why.join("; ")})`,
        );
      }
      v = await this.#steps[i]!(v);
      from = i + 1;
      if (this.#fits(v)) return v;
    }
  }
  async #read(
    db: ReturnType<typeof database>,
    row: { id: string; data: string; schema_hash: string; updated: number },
  ): Promise<T> {
    const s = this.#schema();
    const value = decode(row.data);
    if (row.schema_hash === s.hash) return value as T;
    // written under another schema: fine if it still fits, otherwise migrate it (and store the result)
    if (this.#fits(value)) return value as T;
    const seen = db
      .prepare("SELECT hash, first_seen FROM schemas WHERE tbl = ? AND hash IN (?, ?)")
      .all(this.name, row.schema_hash, s.hash) as { hash: string; first_seen: number }[];
    const firstSeen = (h: string) => seen.find((x) => x.hash === h)?.first_seen ?? 0;
    if (firstSeen(row.schema_hash) > firstSeen(s.hash)) {
      throw new HttpError(
        410,
        `table "${this.name}": row ${row.id} was written by a newer version of this app; reload for the current version`,
      );
    }
    if (!this.#steps.length) {
      throw new Error(
        `table "${this.name}": row ${row.id} was written as an older type and doesn't fit the current one (${checkValue(s.s, value, "row", s.defs)}). Give table() a migrate function: table<Row>("${this.name}", { migrate: (old) => ... })`,
      );
    }
    const next = await this.#migrate(value, row.id);
    this.#check(next, `migrated row ${row.id}`);
    db.prepare(
      "UPDATE rows SET data = ?, schema_hash = ? WHERE tbl = ? AND id = ? AND schema_hash = ?",
    ).run(encode(next), s.hash, this.name, row.id, row.schema_hash);
    return next as T;
  }
  /** Rows not written as the current row type: they must be read (maybe migrated) before they can be compared. */
  #stale(
    db: ReturnType<typeof database>,
  ): { id: string; data: string; schema_hash: string; updated: number }[] {
    const h = this.#schema().hash;
    return db
      .prepare(
        "SELECT id, data, schema_hash, updated FROM rows WHERE tbl = ? AND (schema_hash < ? OR schema_hash > ?)",
      )
      .all(this.name, h, h) as { id: string; data: string; schema_hash: string; updated: number }[];
  }
  /** Store `value` under `id`, replacing what was there; it must fit the row type. */
  async put(id: string, value: T): Promise<void> {
    this.#check(value, "value");
    return this.#op(async (db) => this.#write(db, id, value));
  }
  /** Store a new row; returns its id. */
  async insert(value: T): Promise<string> {
    const id = randomId(9);
    await this.put(id, value);
    return id;
  }
  async get(id: string): Promise<T | undefined> {
    return this.#op(async (db) => {
      const row = db
        .prepare("SELECT id, data, schema_hash, updated FROM rows WHERE tbl = ? AND id = ?")
        .get(this.name, id) as
        | { id: string; data: string; schema_hash: string; updated: number }
        | undefined;
      return row ? this.#read(db, row) : undefined;
    });
  }
  async delete(id: string): Promise<boolean> {
    return this.#op(
      async (db) =>
        db.prepare("DELETE FROM rows WHERE tbl = ? AND id = ?").run(this.name, id).changes > 0,
    );
  }
  /**
   * Change one row atomically: `fn` gets its current value and returns the new one, synchronously, so nothing
   * can write in between. Returns the new value, or undefined if there's no such row.
   */
  async update(id: string, fn: (current: T) => T): Promise<T | undefined> {
    return this.#op(async (db) => {
      for (;;) {
        // bring the row up to the current type first: migrating may be async
        if ((await this.get(id)) === undefined) return undefined;
        const r = this.#atomically(db, () => {
          const row = db
            .prepare("SELECT data FROM rows WHERE tbl = ? AND id = ?")
            .get(this.name, id) as { data: string } | undefined;
          if (!row) return { done: true, value: undefined };
          const current = decode(row.data);
          if (!this.#fits(current)) return { done: false, value: undefined }; // rewritten as another type meanwhile: again
          const next = fn(current as T);
          if (typeof (next as { then?: unknown } | undefined)?.then === "function") {
            throw new TypeError(
              `table "${this.name}": update()'s function must return the new value, not a promise (it runs while the row is locked)`,
            );
          }
          this.#check(next, "value");
          this.#write(db, id, next);
          return { done: true, value: next };
        });
        if (r.done) return r.value;
      }
    });
  }
  /**
   * Rows whose fields equal `where`'s (all rows if omitted), oldest first or by `orderBy`. Fields in the table's
   * `index` or `unique` lists are looked up by index.
   */
  async find(
    where: Partial<T> = {},
    opts: FindOptions<T> = {},
  ): Promise<{ id: string; value: T }[]> {
    return this.#op(async (db) => {
      const entries = Object.entries(where as Record<string, unknown>);
      for (const f of [...entries.map(([k]) => k), ...(opts.orderBy ? [opts.orderBy] : [])])
        if (!/^[A-Za-z_$][\w$]*$/.test(f)) throw new Error(`table "${this.name}": no field '${f}'`);
      const wanted = entries.map(([k, v]) => [k, encode(v)] as const);
      const matches = (v: unknown) =>
        wanted.every(([k, x]) => encode((v as Record<string, unknown>)[k]) === x);
      const whole = (n: number | undefined) =>
        n === undefined ? undefined : Math.max(0, Math.floor(n));
      const limit = whole(opts.limit);
      const offset = whole(opts.offset) ?? 0;
      const stale = this.#stale(db);
      const scalar = entries.filter(([, v]) => isScalar(v));
      // with every row current and every condition expressible in SQL, SQLite does all of it (with indexes)
      const pushdown = !stale.length && scalar.length === entries.length;
      const order = opts.orderBy ? `${fieldSql(opts.orderBy)} ${opts.desc ? "DESC" : "ASC"}, ` : "";
      const page = pushdown && limit !== undefined ? ` LIMIT ${limit} OFFSET ${offset}` : "";
      const sql = `SELECT id, data, schema_hash, updated FROM rows WHERE tbl = '${this.name}' AND schema_hash = ?${scalar.map(([k]) => ` AND ${fieldSql(k)} = ?`).join("")} ORDER BY ${order}updated, id${page}`;
      const out: { id: string; value: T; updated: number }[] = [];
      for (const row of db
        .prepare(sql)
        .all(this.#schema().hash, ...scalar.map(([, v]) => sqlValue(v))) as {
        id: string;
        data: string;
        updated: number;
      }[]) {
        const value = decode(row.data);
        if (matches(value)) out.push({ id: row.id, value: value as T, updated: row.updated });
      }
      if (pushdown) return out.map(({ id, value }) => ({ id, value }));
      for (const row of stale) {
        const value = await this.#read(db, row);
        if (matches(value)) out.push({ id: row.id, value, updated: row.updated });
      }
      const by = opts.orderBy;
      const field = (r: { value: T }) => (r.value as Record<string, unknown>)[by!];
      out.sort(
        (a, b) =>
          (by ? compareFields(field(a), field(b)) * (opts.desc ? -1 : 1) : 0) ||
          a.updated - b.updated ||
          (a.id < b.id ? -1 : 1),
      );
      return out
        .slice(offset, limit === undefined ? undefined : offset + limit)
        .map(({ id, value }) => ({ id, value }));
    });
  }
  /** The first row `find` would return, or undefined. */
  async findOne(
    where: Partial<T> = {},
    opts: Omit<FindOptions<T>, "limit"> = {},
  ): Promise<{ id: string; value: T } | undefined> {
    return (await this.find(where, { ...opts, limit: 1 }))[0];
  }
  async count(where: Partial<T> = {}): Promise<number> {
    return this.#op(async (db) => {
      const entries = Object.entries(where as Record<string, unknown>);
      if (
        this.#stale(db).length ||
        !entries.every(([k, v]) => /^[A-Za-z_$][\w$]*$/.test(k) && isScalar(v))
      )
        return (await this.find(where)).length;
      const sql = `SELECT count(*) AS n FROM rows WHERE tbl = '${this.name}' AND schema_hash = ?${entries.map(([k]) => ` AND ${fieldSql(k)} = ?`).join("")}`;
      return (
        db.prepare(sql).get(this.#schema().hash, ...entries.map(([, v]) => sqlValue(v))) as {
          n: number;
        }
      ).n;
    });
  }
  /**
   * For the CLI (`tcm data migrate`): read every row written as another type now, storing it as the current one
   * (rows that fit are just restamped).
   */
  async [Symbol.for("tcm.migrateAll")](): Promise<{
    total: number;
    restamped: number;
    migrated: number;
    newer: number;
    failed: { id: string; why: string }[];
  }> {
    return this.#op(async (db) => {
      const s = this.#schema();
      const total = (
        db.prepare("SELECT count(*) AS n FROM rows WHERE tbl = ?").get(this.name) as { n: number }
      ).n;
      const r = {
        total,
        restamped: 0,
        migrated: 0,
        newer: 0,
        failed: [] as { id: string; why: string }[],
      };
      for (const row of this.#stale(db)) {
        if (this.#fits(decode(row.data))) {
          db.prepare(
            "UPDATE rows SET schema_hash = ? WHERE tbl = ? AND id = ? AND schema_hash = ?",
          ).run(s.hash, this.name, row.id, row.schema_hash);
          r.restamped++;
          continue;
        }
        try {
          await this.#read(db, row);
          r.migrated++;
        } catch (e) {
          if (e instanceof HttpError && e.status === 410) r.newer++;
          else r.failed.push({ id: row.id, why: (e as Error)?.message ?? String(e) });
        }
      }
      return r;
    });
  }
}

uses

• tcmts.server.FindOptions
• tcmts.server.HttpError
• tcmts.server.RowSchema
• tcmts.server.TableOptions
• tcmts.server.compareFields
• tcmts.server.fieldSql
• tcmts.server.isScalar
• tcmts.server.sqlValue

used by

• tcmts.server.table