Add guarded bulk database replacement

.changeset/guarded-catalog-replacement.md

@@ -0,0 +1,5 @@
+---
+"@executor-js/fumadb": minor
+---
+
+Add guarded bulk replacement to the database adapters. Callers can replace rows only while they still own a rebuild claim, and inserts respect driver parameter limits.

packages/core/fumadb/src/adapters/drizzle/query.ts

@@ -12,6 +12,15 @@ import {
 import type { SQLProvider } from "../../shared/providers";
 import { type ColumnType, parseDrizzle, type TableType } from "./shared";
 
+/** Thrown inside a `replaceMany` transaction to abort it when the guard
+ *  matched no row; caught at the boundary and reported as `applied: false`. */
+class ReplaceGuardMiss extends Error {
+  constructor() {
+    super("replaceMany guard matched no row");
+    this.name = "ReplaceGuardMiss";
+  }
+}
+
 type P_TableType = PostgreSQL.PgTableWithColumns<PostgreSQL.TableConfig>;
 type P_ColumnType = PostgreSQL.AnyPgColumn;
 type P_DBType = PostgreSQL.PgDatabase<
@@ -635,6 +644,128 @@ export function fromDrizzle(
 
       await query;
     },
+    async replaceMany(plan) {
+      // Every statement of the plan, built against one handle: the guard
+      // update first, then the deletes, then parameter-bounded insert batches.
+      const buildStatements = (handle: typeof db): unknown[] => {
+        const statements: unknown[] = [];
+        if (plan.guard) {
+          const guardTable = toDrizzle(plan.guard.table);
+          let update = handle.update(guardTable).set(mapValues(plan.guard.set, plan.guard.table));
+          if (plan.guard.where) {
+            update = update.where(buildWhere(toDrizzleColumn, plan.guard.where)) as any;
+          }
+          statements.push(update);
+        }
+        for (const del of plan.deletes) {
+          const drizzleTable = toDrizzle(del.table);
+          let query = handle.delete(drizzleTable);
+          if (del.where) query = query.where(buildWhere(toDrizzleColumn, del.where)) as any;
+          statements.push(query);
+        }
+        for (const ins of plan.inserts) {
+          if (ins.values.length === 0) continue;
+          const drizzleTable = toDrizzle(ins.table);
+          const values = ins.values.map((v) => mapValues(v, ins.table));
+          // Drizzle builds a multi-row INSERT over the UNION of every row's
+          // columns, so the widest row sets the per-row parameter count.
+          const columnsPerRow = Math.max(1, ...values.map((row) => Object.keys(row).length));
+          const batchSize = parameterBoundedBatchSize(ins.table, columnsPerRow, 0, maxBoundParameters);
+          for (let i = 0; i < values.length; i += batchSize) {
+            statements.push(handle.insert(drizzleTable).values(values.slice(i, i + batchSize)));
+          }
+        }
+        return statements;
+      };
+
+      // How many rows the guard matched. Drizzle hands back the driver's own
+      // result: libsql `rowsAffected`, better-sqlite3 `changes`, node-postgres
+      // `rowCount`, postgres.js `count` (a RowList), D1 `meta.changes`, and
+      // mysql2 a `[ResultSetHeader, FieldPacket[]]` tuple whose header
+      // carries `affectedRows`. A result with NONE of these is a driver this
+      // fence does not know, and a fence that cannot read its own guard is
+      // not a fence — so that is a hard error, never a silent "matched".
+      const guardMatched = (result: unknown): boolean => {
+        if (!plan.guard) return true;
+        const header =
+          Array.isArray(result) && result.length > 0 && result[0] && typeof result[0] === "object"
+            ? (result[0] as Record<string, unknown>)
+            : undefined;
+        if (header && typeof header["affectedRows"] === "number") {
+          return (header["affectedRows"] as number) > 0;
+        }
+        if (result && typeof result === "object") {
+          const r = result as Record<string, unknown>;
+          for (const key of ["rowsAffected", "changes", "rowCount", "count"]) {
+            if (typeof r[key] === "number") return (r[key] as number) > 0;
+          }
+          const meta = r["meta"];
+          if (meta && typeof meta === "object" && typeof (meta as Record<string, unknown>)["changes"] === "number") {
+            return ((meta as Record<string, unknown>)["changes"] as number) > 0;
+          }
+        }
+        // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: adapter refuses to fence on a driver whose update result it cannot read
+        throw new Error(
+          "[FumaDB Drizzle] replaceMany guard: the driver's update result carries no affected-row count.",
+        );
+      };
+
+      // D1: no interactive transactions, but the driver's native batch runs
+      // every statement in ONE transaction. A batch cannot make its later
+      // statements conditional on an earlier one's row count, so the guard
+      // runs ALONE first (one statement, atomic, tells us whether we own the
+      // row) and only then do the deletes + inserts go through one batch.
+      // That is guard-then-batch, not one unit: between the two another
+      // writer can re-claim. What a reader sees at each step stays
+      // consistent — after our guard the row's manifest names OUR build
+      // while the table still holds the old rows, which the reader refuses
+      // (count/generation mismatch); once our batch lands, manifest and rows
+      // agree and are served; if a re-claimer's guard lands in between, its
+      // manifest over our rows is refused until its own batch lands. No
+      // half-built or mismatched catalog is ever served.
+      const nativeBatch = db as unknown as {
+        readonly batch?: (statements: readonly unknown[]) => Promise<unknown[]>;
+      };
+      if (!interactiveTransactions) {
+        if (plan.guard) {
+          const guardTable = toDrizzle(plan.guard.table);
+          let update = db.update(guardTable).set(mapValues(plan.guard.set, plan.guard.table));
+          if (plan.guard.where) {
+            update = update.where(buildWhere(toDrizzleColumn, plan.guard.where)) as any;
+          }
+          const result = await update;
+          if (!guardMatched(result)) return { applied: false };
+        }
+        const rest = buildStatements(db).slice(plan.guard ? 1 : 0);
+        if (rest.length === 0) return { applied: true };
+        if (nativeBatch.batch) {
+          await nativeBatch.batch(rest);
+        } else {
+          for (const statement of rest) await statement;
+        }
+        return { applied: true };
+      }
+
+      // Interactive engines: one transaction, guard first; a guard that
+      // matched nothing rolls the transaction back untouched.
+      return runAtomically(async (handle) => {
+        const statements = buildStatements(handle);
+        if (plan.guard) {
+          const result = await statements[0];
+          if (!guardMatched(result)) {
+            // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: abort the driver transaction so nothing after a failed guard commits
+            throw new ReplaceGuardMiss();
+          }
+          for (const statement of statements.slice(1)) await statement;
+          return { applied: true };
+        }
+        for (const statement of statements) await statement;
+        return { applied: true };
+      }).catch((error: unknown) => {
+        if (error instanceof ReplaceGuardMiss) return { applied: false };
+        throw error;
+      });
+    },
     async transaction(run) {
       // Some SQLite-compatible engines (Cloudflare D1) reject interactive
       // transactions — both raw BEGIN/COMMIT and the driver's `.transaction()`.

packages/core/fumadb/src/adapters/drizzle/replace-many-guard.test.ts

@@ -0,0 +1,144 @@
+import { expect, test } from "@effect/vitest";
+
+import { column, idColumn, schema, table } from "../../schema";
+import { fromDrizzle } from "./query";
+
+// `replaceMany`'s fence is only as good as its reading of the guard update's
+// affected-row count, and every driver spells that differently. A recording
+// fake of the Drizzle handle plays each driver's result shape so the fence
+// is proven to hold — and to refuse to fence at all — without a live server.
+const v1 = schema({
+  version: "1.0.0",
+  tables: {
+    owners: table("owners", {
+      id: idColumn("id", "varchar(255)"),
+      token: column("token", "string"),
+    }),
+    rows: table("rows", {
+      id: idColumn("id", "varchar(255)"),
+      value: column("value", "string"),
+    }),
+  },
+});
+
+interface FakeOptions {
+  /** What the guard UPDATE resolves to. */
+  readonly updateResult: unknown;
+}
+
+const createFakePgDb = (options: FakeOptions) => {
+  const events: string[] = [];
+  const fakeTables = {
+    owners: { id: { name: "id" }, token: { name: "token" } },
+    rows: { id: { name: "id" }, value: { name: "value" } },
+  };
+  const makeHandle = (label: string) => ({
+    _: { fullSchema: fakeTables },
+    update: () => ({
+      set: () => {
+        const builder = {
+          where: () => builder,
+          then: (resolve: (value: unknown) => void) => {
+            events.push(`${label}:update`);
+            resolve(options.updateResult);
+          },
+        };
+        return builder;
+      },
+    }),
+    delete: () => {
+      const builder = {
+        where: () => builder,
+        then: (resolve: (value: unknown) => void) => {
+          events.push(`${label}:delete`);
+          resolve(undefined);
+        },
+      };
+      return builder;
+    },
+    insert: () => ({
+      values: () => ({
+        then: (resolve: (value: unknown) => void) => {
+          events.push(`${label}:insert`);
+          resolve(undefined);
+        },
+      }),
+    }),
+    transaction: async <T>(callback: (tx: unknown) => Promise<T>): Promise<T> => {
+      events.push("transaction:begin");
+      // oxlint-disable-next-line executor/no-try-catch-or-throw -- fake driver mirrors commit/rollback
+      try {
+        const result = await callback(makeHandle("tx"));
+        events.push("transaction:commit");
+        return result;
+      } catch (error) {
+        events.push("transaction:rollback");
+        throw error;
+      }
+    },
+  });
+  return { db: makeHandle("root"), events };
+};
+
+const replace = (db: unknown) => {
+  const orm = fromDrizzle(v1, db, "postgresql");
+  const owners = v1.tables.owners;
+  const rows = v1.tables.rows;
+  expect(orm.internal.replaceMany).toBeDefined();
+  return orm.internal.replaceMany!({
+    guard: { table: owners, where: undefined, set: { token: null } },
+    deletes: [{ table: rows, where: undefined }],
+    inserts: [{ table: rows, values: [{ id: "r1", value: "a" }] }],
+  });
+};
+
+test("postgres.js `count: 0` is a guard miss: the transaction rolls back and nothing applies", async () => {
+  // postgres.js hands Drizzle a RowList whose affected count is `count`.
+  const { db, events } = createFakePgDb({ updateResult: Object.assign([], { count: 0 }) });
+  const out = await replace(db);
+  expect(out).toEqual({ applied: false });
+  expect(events).toEqual(["transaction:begin", "tx:update", "transaction:rollback"]);
+});
+
+test("postgres.js `count: 1` is a guard hit: deletes and inserts run in the same transaction", async () => {
+  const { db, events } = createFakePgDb({ updateResult: Object.assign([], { count: 1 }) });
+  const out = await replace(db);
+  expect(out).toEqual({ applied: true });
+  expect(events).toEqual([
+    "transaction:begin",
+    "tx:update",
+    "tx:delete",
+    "tx:insert",
+    "transaction:commit",
+  ]);
+});
+
+test("node-postgres `rowCount`, libsql `rowsAffected`, better-sqlite3 `changes`, D1 `meta.changes`, mysql2 `[header].affectedRows` are all read", async () => {
+  for (const updateResult of [
+    { rowCount: 0 },
+    { rowsAffected: 0 },
+    { changes: 0 },
+    { meta: { changes: 0 } },
+    [{ affectedRows: 0, fieldCount: 0 }, []],
+  ]) {
+    const { db } = createFakePgDb({ updateResult });
+    expect(await replace(db), JSON.stringify(updateResult)).toEqual({ applied: false });
+  }
+  for (const updateResult of [
+    { rowCount: 2 },
+    { rowsAffected: 1 },
+    { changes: 1 },
+    { meta: { changes: 1 } },
+    [{ affectedRows: 1, fieldCount: 0 }, []],
+  ]) {
+    const { db } = createFakePgDb({ updateResult });
+    expect(await replace(db), JSON.stringify(updateResult)).toEqual({ applied: true });
+  }
+});
+
+test("a driver result with no affected-row count refuses to fence rather than silently matching", async () => {
+  const { db, events } = createFakePgDb({ updateResult: { ok: true } });
+  await expect(replace(db)).rejects.toThrow(/affected-row count/);
+  // The transaction was aborted: nothing after the guard ran.
+  expect(events).toEqual(["transaction:begin", "tx:update", "transaction:rollback"]);
+});

packages/core/fumadb/src/adapters/kysely/query.ts

@@ -28,6 +28,25 @@ import { deserialize, serialize } from "../../schema/serialize";
 import type { KyselyConfig } from "../../shared/config";
 import type { SQLProvider } from "../../shared/providers";
 
+/** Row cap per insert statement in `replaceMany`, before the parameter
+ *  budget is applied. */
+const REPLACE_MANY_INSERT_BATCH_ROWS = 500;
+
+/** Conservative bound-parameter budget per statement for each provider. MSSQL
+ *  caps at 2100; SQLite's historical floor is 999; Postgres and MySQL allow
+ *  far more, but the same floor keeps a wide table from overflowing anywhere. */
+const replaceManyParameterBudget = (provider: SQLProvider): number =>
+  provider === "mssql" ? 2000 : 999;
+
+/** Thrown inside a `replaceMany` transaction to abort it when the guard
+ *  matched no row; caught at the boundary and reported as `applied: false`. */
+class ReplaceGuardMiss extends Error {
+  constructor() {
+    super("replaceMany guard matched no row");
+    this.name = "ReplaceGuardMiss";
+  }
+}
+
 function fullSQLName(column: AnyColumn) {
   return `${column.table.names.sql}.${column.names.sql}`;
 }
@@ -508,6 +527,65 @@ export function fromKysely(
       }
       await query.execute();
     },
+    replaceMany(plan) {
+      // Every Kysely dialect has interactive transactions, so the whole plan
+      // is one transaction: the guard update first, and a guard that matched
+      // no row aborts it (rollback) before any delete or insert runs.
+      return kysely
+        .transaction()
+        .execute(async (tx) => {
+          if (plan.guard) {
+            let update = tx
+              .updateTable(plan.guard.table.names.sql)
+              .set(encodeValues(plan.guard.set, plan.guard.table, false));
+            if (plan.guard.where) {
+              const where = plan.guard.where;
+              update = update.where((eb) => buildWhere(where, eb, provider));
+            }
+            const result = await update.executeTakeFirst();
+            if (Number(result.numUpdatedRows) === 0) {
+              // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: abort the driver transaction so nothing after a failed guard commits
+              throw new ReplaceGuardMiss();
+            }
+          }
+          for (const del of plan.deletes) {
+            let query = tx.deleteFrom(del.table.names.sql);
+            if (del.where) {
+              const where = del.where;
+              query = query.where((eb) => buildWhere(where, eb, provider));
+            }
+            await query.execute();
+          }
+          for (const ins of plan.inserts) {
+            if (ins.values.length === 0) continue;
+            const encoded = ins.values.map((v) => encodeValues(v, ins.table, true));
+            // Engines cap bound parameters per statement (MSSQL 2100, older
+            // SQLite 999): chunk so `rows * columns` stays under the budget,
+            // all inside this one transaction.
+            // Kysely builds a multi-row INSERT over the UNION of every row's
+            // columns, so the widest row sets the per-row parameter count.
+            const columnsPerRow = Math.max(1, ...encoded.map((row) => Object.keys(row).length));
+            const rowsPerStatement = Math.max(
+              1,
+              Math.min(
+                REPLACE_MANY_INSERT_BATCH_ROWS,
+                Math.floor(replaceManyParameterBudget(provider) / columnsPerRow),
+              ),
+            );
+            for (let i = 0; i < encoded.length; i += rowsPerStatement) {
+              await tx
+                .insertInto(ins.table.names.sql)
+                .values(encoded.slice(i, i + rowsPerStatement))
+                .execute();
+            }
+          }
+          return { applied: true as const };
+        })
+        .catch((error: unknown) => {
+          if (error instanceof ReplaceGuardMiss) return { applied: false as const };
+          throw error;
+        });
+    },
     transaction(run) {
       return kysely.transaction().execute((ctx) => {
         const tx = fromKysely(schema, {