History

Add guarded bulk database replacement

Rhys Sullivan committed 1 month ago

Commit e3a46ff1 parent 929b2337 files · +470 −0Browse files

Unified · Split

7 files
  1. .changeset/guarded-catalog-replacement.md +5 −0
  2. packages/core/fumadb/src/adapters/drizzle/query.ts +131 −0
  3. packages/core/fumadb/src/adapters/drizzle/replace-many-guard.test.ts +144 −0
  4. packages/core/fumadb/src/adapters/kysely/query.ts +78 −0
  5. packages/core/fumadb/src/adapters/memory/index.ts +27 −0
  6. packages/core/fumadb/src/query/index.ts +30 −0
  7. packages/core/fumadb/src/query/orm/index.ts +55 −0
.changeset/guarded-catalog-replacement.md+5−0
@@ -0,0 +1,5 @@
1---
2"@executor-js/fumadb": minor
3---
4
5Add 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+131−0
@@ -12,6 +12,15 @@ import { · 11 unmodified lines
12import type { SQLProvider } from "../../shared/providers";12import type { SQLProvider } from "../../shared/providers";
13import { type ColumnType, parseDrizzle, type TableType } from "./shared";13import { type ColumnType, parseDrizzle, type TableType } from "./shared";
14 14
15/** Thrown inside a `replaceMany` transaction to abort it when the guard
16 * matched no row; caught at the boundary and reported as `applied: false`. */
17class ReplaceGuardMiss extends Error {
18 constructor() {
19 super("replaceMany guard matched no row");
20 this.name = "ReplaceGuardMiss";
21 }
22}
23
15type P_TableType = PostgreSQL.PgTableWithColumns<PostgreSQL.TableConfig>;24type P_TableType = PostgreSQL.PgTableWithColumns<PostgreSQL.TableConfig>;
16type P_ColumnType = PostgreSQL.AnyPgColumn;25type P_ColumnType = PostgreSQL.AnyPgColumn;
17type P_DBType = PostgreSQL.PgDatabase<26type P_DBType = PostgreSQL.PgDatabase<
@@ -635,6 +644,128 @@ export function fromDrizzle( · 617 unmodified lines
635 644
636 await query;645 await query;
637 },646 },
647 async replaceMany(plan) {
648 // Every statement of the plan, built against one handle: the guard
649 // update first, then the deletes, then parameter-bounded insert batches.
650 const buildStatements = (handle: typeof db): unknown[] => {
651 const statements: unknown[] = [];
652 if (plan.guard) {
653 const guardTable = toDrizzle(plan.guard.table);
654 let update = handle.update(guardTable).set(mapValues(plan.guard.set, plan.guard.table));
655 if (plan.guard.where) {
656 update = update.where(buildWhere(toDrizzleColumn, plan.guard.where)) as any;
657 }
658 statements.push(update);
659 }
660 for (const del of plan.deletes) {
661 const drizzleTable = toDrizzle(del.table);
662 let query = handle.delete(drizzleTable);
663 if (del.where) query = query.where(buildWhere(toDrizzleColumn, del.where)) as any;
664 statements.push(query);
665 }
666 for (const ins of plan.inserts) {
667 if (ins.values.length === 0) continue;
668 const drizzleTable = toDrizzle(ins.table);
669 const values = ins.values.map((v) => mapValues(v, ins.table));
670 // Drizzle builds a multi-row INSERT over the UNION of every row's
671 // columns, so the widest row sets the per-row parameter count.
672 const columnsPerRow = Math.max(1, ...values.map((row) => Object.keys(row).length));
673 const batchSize = parameterBoundedBatchSize(ins.table, columnsPerRow, 0, maxBoundParameters);
674 for (let i = 0; i < values.length; i += batchSize) {
675 statements.push(handle.insert(drizzleTable).values(values.slice(i, i + batchSize)));
676 }
677 }
678 return statements;
679 };
680
681 // How many rows the guard matched. Drizzle hands back the driver's own
682 // result: libsql `rowsAffected`, better-sqlite3 `changes`, node-postgres
683 // `rowCount`, postgres.js `count` (a RowList), D1 `meta.changes`, and
684 // mysql2 a `[ResultSetHeader, FieldPacket[]]` tuple whose header
685 // carries `affectedRows`. A result with NONE of these is a driver this
686 // fence does not know, and a fence that cannot read its own guard is
687 // not a fence — so that is a hard error, never a silent "matched".
688 const guardMatched = (result: unknown): boolean => {
689 if (!plan.guard) return true;
690 const header =
691 Array.isArray(result) && result.length > 0 && result[0] && typeof result[0] === "object"
692 ? (result[0] as Record<string, unknown>)
693 : undefined;
694 if (header && typeof header["affectedRows"] === "number") {
695 return (header["affectedRows"] as number) > 0;
696 }
697 if (result && typeof result === "object") {
698 const r = result as Record<string, unknown>;
699 for (const key of ["rowsAffected", "changes", "rowCount", "count"]) {
700 if (typeof r[key] === "number") return (r[key] as number) > 0;
701 }
702 const meta = r["meta"];
703 if (meta && typeof meta === "object" && typeof (meta as Record<string, unknown>)["changes"] === "number") {
704 return ((meta as Record<string, unknown>)["changes"] as number) > 0;
705 }
706 }
707 // 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
708 throw new Error(
709 "[FumaDB Drizzle] replaceMany guard: the driver's update result carries no affected-row count.",
710 );
711 };
712
713 // D1: no interactive transactions, but the driver's native batch runs
714 // every statement in ONE transaction. A batch cannot make its later
715 // statements conditional on an earlier one's row count, so the guard
716 // runs ALONE first (one statement, atomic, tells us whether we own the
717 // row) and only then do the deletes + inserts go through one batch.
718 // That is guard-then-batch, not one unit: between the two another
719 // writer can re-claim. What a reader sees at each step stays
720 // consistent — after our guard the row's manifest names OUR build
721 // while the table still holds the old rows, which the reader refuses
722 // (count/generation mismatch); once our batch lands, manifest and rows
723 // agree and are served; if a re-claimer's guard lands in between, its
724 // manifest over our rows is refused until its own batch lands. No
725 // half-built or mismatched catalog is ever served.
726 const nativeBatch = db as unknown as {
727 readonly batch?: (statements: readonly unknown[]) => Promise<unknown[]>;
728 };
729 if (!interactiveTransactions) {
730 if (plan.guard) {
731 const guardTable = toDrizzle(plan.guard.table);
732 let update = db.update(guardTable).set(mapValues(plan.guard.set, plan.guard.table));
733 if (plan.guard.where) {
734 update = update.where(buildWhere(toDrizzleColumn, plan.guard.where)) as any;
735 }
736 const result = await update;
737 if (!guardMatched(result)) return { applied: false };
738 }
739 const rest = buildStatements(db).slice(plan.guard ? 1 : 0);
740 if (rest.length === 0) return { applied: true };
741 if (nativeBatch.batch) {
742 await nativeBatch.batch(rest);
743 } else {
744 for (const statement of rest) await statement;
745 }
746 return { applied: true };
747 }
748
749 // Interactive engines: one transaction, guard first; a guard that
750 // matched nothing rolls the transaction back untouched.
751 return runAtomically(async (handle) => {
752 const statements = buildStatements(handle);
753 if (plan.guard) {
754 const result = await statements[0];
755 if (!guardMatched(result)) {
756 // 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
757 throw new ReplaceGuardMiss();
758 }
759 for (const statement of statements.slice(1)) await statement;
760 return { applied: true };
761 }
762 for (const statement of statements) await statement;
763 return { applied: true };
764 }).catch((error: unknown) => {
765 if (error instanceof ReplaceGuardMiss) return { applied: false };
766 throw error;
767 });
768 },
638 async transaction(run) {769 async transaction(run) {
639 // Some SQLite-compatible engines (Cloudflare D1) reject interactive770 // Some SQLite-compatible engines (Cloudflare D1) reject interactive
640 // transactions — both raw BEGIN/COMMIT and the driver's `.transaction()`.771 // transactions — both raw BEGIN/COMMIT and the driver's `.transaction()`.
packages/core/fumadb/src/adapters/drizzle/replace-many-guard.test.ts+144−0
@@ -0,0 +1,144 @@
1import { expect, test } from "@effect/vitest";
2
3import { column, idColumn, schema, table } from "../../schema";
4import { fromDrizzle } from "./query";
5
6// `replaceMany`'s fence is only as good as its reading of the guard update's
7// affected-row count, and every driver spells that differently. A recording
8// fake of the Drizzle handle plays each driver's result shape so the fence
9// is proven to hold — and to refuse to fence at all — without a live server.
10const v1 = schema({
11 version: "1.0.0",
12 tables: {
13 owners: table("owners", {
14 id: idColumn("id", "varchar(255)"),
15 token: column("token", "string"),
16 }),
17 rows: table("rows", {
18 id: idColumn("id", "varchar(255)"),
19 value: column("value", "string"),
20 }),
21 },
22});
23
24interface FakeOptions {
25 /** What the guard UPDATE resolves to. */
26 readonly updateResult: unknown;
27}
28
29const createFakePgDb = (options: FakeOptions) => {
30 const events: string[] = [];
31 const fakeTables = {
32 owners: { id: { name: "id" }, token: { name: "token" } },
33 rows: { id: { name: "id" }, value: { name: "value" } },
34 };
35 const makeHandle = (label: string) => ({
36 _: { fullSchema: fakeTables },
37 update: () => ({
38 set: () => {
39 const builder = {
40 where: () => builder,
41 then: (resolve: (value: unknown) => void) => {
42 events.push(`${label}:update`);
43 resolve(options.updateResult);
44 },
45 };
46 return builder;
47 },
48 }),
49 delete: () => {
50 const builder = {
51 where: () => builder,
52 then: (resolve: (value: unknown) => void) => {
53 events.push(`${label}:delete`);
54 resolve(undefined);
55 },
56 };
57 return builder;
58 },
59 insert: () => ({
60 values: () => ({
61 then: (resolve: (value: unknown) => void) => {
62 events.push(`${label}:insert`);
63 resolve(undefined);
64 },
65 }),
66 }),
67 transaction: async <T>(callback: (tx: unknown) => Promise<T>): Promise<T> => {
68 events.push("transaction:begin");
69 // oxlint-disable-next-line executor/no-try-catch-or-throw -- fake driver mirrors commit/rollback
70 try {
71 const result = await callback(makeHandle("tx"));
72 events.push("transaction:commit");
73 return result;
74 } catch (error) {
75 events.push("transaction:rollback");
76 throw error;
77 }
78 },
79 });
80 return { db: makeHandle("root"), events };
81};
82
83const replace = (db: unknown) => {
84 const orm = fromDrizzle(v1, db, "postgresql");
85 const owners = v1.tables.owners;
86 const rows = v1.tables.rows;
87 expect(orm.internal.replaceMany).toBeDefined();
88 return orm.internal.replaceMany!({
89 guard: { table: owners, where: undefined, set: { token: null } },
90 deletes: [{ table: rows, where: undefined }],
91 inserts: [{ table: rows, values: [{ id: "r1", value: "a" }] }],
92 });
93};
94
95test("postgres.js `count: 0` is a guard miss: the transaction rolls back and nothing applies", async () => {
96 // postgres.js hands Drizzle a RowList whose affected count is `count`.
97 const { db, events } = createFakePgDb({ updateResult: Object.assign([], { count: 0 }) });
98 const out = await replace(db);
99 expect(out).toEqual({ applied: false });
100 expect(events).toEqual(["transaction:begin", "tx:update", "transaction:rollback"]);
101});
102
103test("postgres.js `count: 1` is a guard hit: deletes and inserts run in the same transaction", async () => {
104 const { db, events } = createFakePgDb({ updateResult: Object.assign([], { count: 1 }) });
105 const out = await replace(db);
106 expect(out).toEqual({ applied: true });
107 expect(events).toEqual([
108 "transaction:begin",
109 "tx:update",
110 "tx:delete",
111 "tx:insert",
112 "transaction:commit",
113 ]);
114});
115
116test("node-postgres `rowCount`, libsql `rowsAffected`, better-sqlite3 `changes`, D1 `meta.changes`, mysql2 `[header].affectedRows` are all read", async () => {
117 for (const updateResult of [
118 { rowCount: 0 },
119 { rowsAffected: 0 },
120 { changes: 0 },
121 { meta: { changes: 0 } },
122 [{ affectedRows: 0, fieldCount: 0 }, []],
123 ]) {
124 const { db } = createFakePgDb({ updateResult });
125 expect(await replace(db), JSON.stringify(updateResult)).toEqual({ applied: false });
126 }
127 for (const updateResult of [
128 { rowCount: 2 },
129 { rowsAffected: 1 },
130 { changes: 1 },
131 { meta: { changes: 1 } },
132 [{ affectedRows: 1, fieldCount: 0 }, []],
133 ]) {
134 const { db } = createFakePgDb({ updateResult });
135 expect(await replace(db), JSON.stringify(updateResult)).toEqual({ applied: true });
136 }
137});
138
139test("a driver result with no affected-row count refuses to fence rather than silently matching", async () => {
140 const { db, events } = createFakePgDb({ updateResult: { ok: true } });
141 await expect(replace(db)).rejects.toThrow(/affected-row count/);
142 // The transaction was aborted: nothing after the guard ran.
143 expect(events).toEqual(["transaction:begin", "tx:update", "transaction:rollback"]);
144});
packages/core/fumadb/src/adapters/kysely/query.ts+78−0
@@ -28,6 +28,25 @@ import { deserialize, serialize } from "../../schema/serialize"; · 27 unmodified lines
28import type { KyselyConfig } from "../../shared/config";28import type { KyselyConfig } from "../../shared/config";
29import type { SQLProvider } from "../../shared/providers";29import type { SQLProvider } from "../../shared/providers";
30 30
31/** Row cap per insert statement in `replaceMany`, before the parameter
32 * budget is applied. */
33const REPLACE_MANY_INSERT_BATCH_ROWS = 500;
34
35/** Conservative bound-parameter budget per statement for each provider. MSSQL
36 * caps at 2100; SQLite's historical floor is 999; Postgres and MySQL allow
37 * far more, but the same floor keeps a wide table from overflowing anywhere. */
38const replaceManyParameterBudget = (provider: SQLProvider): number =>
39 provider === "mssql" ? 2000 : 999;
40
41/** Thrown inside a `replaceMany` transaction to abort it when the guard
42 * matched no row; caught at the boundary and reported as `applied: false`. */
43class ReplaceGuardMiss extends Error {
44 constructor() {
45 super("replaceMany guard matched no row");
46 this.name = "ReplaceGuardMiss";
47 }
48}
49
31function fullSQLName(column: AnyColumn) {50function fullSQLName(column: AnyColumn) {
32 return `${column.table.names.sql}.${column.names.sql}`;51 return `${column.table.names.sql}.${column.names.sql}`;
33}52}
@@ -508,6 +527,65 @@ export function fromKysely( · 474 unmodified lines
508 }527 }
509 await query.execute();528 await query.execute();
510 },529 },
530 replaceMany(plan) {
531 // Every Kysely dialect has interactive transactions, so the whole plan
532 // is one transaction: the guard update first, and a guard that matched
533 // no row aborts it (rollback) before any delete or insert runs.
534 return kysely
535 .transaction()
536 .execute(async (tx) => {
537 if (plan.guard) {
538 let update = tx
539 .updateTable(plan.guard.table.names.sql)
540 .set(encodeValues(plan.guard.set, plan.guard.table, false));
541 if (plan.guard.where) {
542 const where = plan.guard.where;
543 update = update.where((eb) => buildWhere(where, eb, provider));
544 }
545 const result = await update.executeTakeFirst();
546 if (Number(result.numUpdatedRows) === 0) {
547 // 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
548 throw new ReplaceGuardMiss();
549 }
550 }
551 for (const del of plan.deletes) {
552 let query = tx.deleteFrom(del.table.names.sql);
553 if (del.where) {
554 const where = del.where;
555 query = query.where((eb) => buildWhere(where, eb, provider));
556 }
557 await query.execute();
558 }
559 for (const ins of plan.inserts) {
560 if (ins.values.length === 0) continue;
561 const encoded = ins.values.map((v) => encodeValues(v, ins.table, true));
562 // Engines cap bound parameters per statement (MSSQL 2100, older
563 // SQLite 999): chunk so `rows * columns` stays under the budget,
564 // all inside this one transaction.
565 // Kysely builds a multi-row INSERT over the UNION of every row's
566 // columns, so the widest row sets the per-row parameter count.
567 const columnsPerRow = Math.max(1, ...encoded.map((row) => Object.keys(row).length));
568 const rowsPerStatement = Math.max(
569 1,
570 Math.min(
571 REPLACE_MANY_INSERT_BATCH_ROWS,
572 Math.floor(replaceManyParameterBudget(provider) / columnsPerRow),
573 ),
574 );
575 for (let i = 0; i < encoded.length; i += rowsPerStatement) {
576 await tx
577 .insertInto(ins.table.names.sql)
578 .values(encoded.slice(i, i + rowsPerStatement))
579 .execute();
580 }
581 }
582 return { applied: true as const };
583 })
584 .catch((error: unknown) => {
585 if (error instanceof ReplaceGuardMiss) return { applied: false as const };
586 throw error;
587 });
588 },
511 transaction(run) {589 transaction(run) {
512 return kysely.transaction().execute((ctx) => {590 return kysely.transaction().execute((ctx) => {
513 const tx = fromKysely(schema, {591 const tx = fromKysely(schema, {
packages/core/fumadb/src/adapters/memory/index.ts+27−0
@@ -222,6 +222,33 @@ export function memoryAdapter(options: MemoryAdapterOptions = {}): FumaDBAdapter · 221 unmodified lines
222 const rows = tableRows(db, table);222 const rows = tableRows(db, table);
223 db[table.ormName] = rows.filter((row) => !matchesCondition(row, v.where));223 db[table.ormName] = rows.filter((row) => !matchesCondition(row, v.where));
224 },224 },
225 async replaceMany(plan) {
226 // In-memory: stage the whole result on a clone, then swap it in
227 // with no suspension point — so a concurrent reader or writer sees
228 // either the old state or the new one, never a partial plan.
229 const staged = cloneValue(db);
230 if (plan.guard) {
231 let matched = 0;
232 for (const row of tableRows(staged, plan.guard.table)) {
233 if (!matchesCondition(row, plan.guard.where)) continue;
234 Object.assign(row, cloneValue(plan.guard.set));
235 matched += 1;
236 }
237 if (matched === 0) return { applied: false };
238 }
239 for (const del of plan.deletes) {
240 staged[del.table.ormName] = tableRows(staged, del.table).filter(
241 (row) => !matchesCondition(row, del.where),
242 );
243 }
244 for (const ins of plan.inserts) {
245 const rows = tableRows(staged, ins.table);
246 for (const value of ins.values) rows.push(applyDefaults(ins.table, value));
247 }
248 for (const key of Object.keys(db)) delete db[key];
249 Object.assign(db, staged);
250 return { applied: true };
251 },
225 async transaction<T>(run: (transactionInstance: AbstractQuery<AnySchema>) => Promise<T>) {252 async transaction<T>(run: (transactionInstance: AbstractQuery<AnySchema>) => Promise<T>) {
226 const snapshot = cloneValue(db);253 const snapshot = cloneValue(db);
227 try {254 try {
packages/core/fumadb/src/query/index.ts+30−0
@@ -224,4 +224,34 @@ export interface AbstractQuery<S extends AnySchema> { · 223 unmodified lines
224 eb: ConditionBuilder<S["tables"][TableName]["columns"]>224 eb: ConditionBuilder<S["tables"][TableName]["columns"]>
225 ) => Condition | boolean;225 ) => Condition | boolean;
226 }) => Promise<void>;226 }) => Promise<void>;
227
228 /**
229 * Delete + insert across tables, optionally fenced by a guard update that
230 * must match a row for the rest to apply. Returns whether the guard
231 * matched; with no guard, always `applied: true`.
232 *
233 * Atomicity depends on the engine. With interactive transactions the whole
234 * plan — guard included — is one transaction: a guard miss or any failure
235 * rolls everything back. WITHOUT them (Cloudflare D1) the guard is its own
236 * committed statement and the deletes + inserts follow in one native batch
237 * (itself atomic): if the batch fails or the process dies between the two,
238 * the guard's update stays. Callers on such engines must be able to tell a
239 * guard-without-rows state apart on read (the executor's tool catalog does
240 * so with a per-connection manifest checked against row counts).
241 */
242 replaceMany: (plan: {
243 readonly guard?: {
244 readonly table: keyof S["tables"];
245 readonly where?: (eb: ConditionBuilder<AnyTable["columns"]>) => Condition | boolean;
246 readonly set: Record<string, unknown>;
247 };
248 readonly deletes: readonly {
249 readonly table: keyof S["tables"];
250 readonly where?: (eb: ConditionBuilder<AnyTable["columns"]>) => Condition | boolean;
251 }[];
252 readonly inserts: readonly {
253 readonly table: keyof S["tables"];
254 readonly values: readonly Record<string, unknown>[];
255 }[];
256 }) => Promise<{ readonly applied: boolean }>;
227}257}
packages/core/fumadb/src/query/orm/index.ts+55−0
@@ -383,6 +383,26 @@ export interface ORMAdapter<S extends AnySchema = AnySchema> { · 382 unmodified lines
383 },383 },
384 ) => Promise<void>;384 ) => Promise<void>;
385 385
386 /**
387 * Run a set of deletes and inserts, with an optional guard: a conditional
388 * update that must match at least one row for the rest to apply. With
389 * interactive transactions the whole plan is one transaction. Without them
390 * (Cloudflare D1) the guard is its own committed statement, followed by the
391 * deletes + inserts in one native batch — see `AbstractQuery.replaceMany`
392 * for the exact guarantee a caller gets on such engines.
393 */
394 replaceMany?: (
395 plan: {
396 readonly guard?: {
397 readonly table: AnyTable;
398 readonly where: Condition | undefined;
399 readonly set: Record<string, unknown>;
400 };
401 readonly deletes: readonly { readonly table: AnyTable; readonly where: Condition | undefined }[];
402 readonly inserts: readonly { readonly table: AnyTable; readonly values: Record<string, unknown>[] }[];
403 },
404 ) => Promise<{ readonly applied: boolean }>;
405
386 /**406 /**
387 * Override this to support native transaction, otherwise use soft transaction.407 * Override this to support native transaction, otherwise use soft transaction.
388 */408 */
@@ -610,6 +630,41 @@ export function toORM<S extends AnySchema>( · 221 unmodified lines
610 if (constrainedWhere === false) return;630 if (constrainedWhere === false) return;
611 return internal.updateMany(table, { set, where: constrainedWhere });631 return internal.updateMany(table, { set, where: constrainedWhere });
612 },632 },
633 async replaceMany(plan) {
634 if (!internal.replaceMany) {
635 // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: public query rejects an adapter without atomic replace
636 throw new Error("[FumaDB] This adapter does not support replaceMany.");
637 }
638 let guard: { table: AnyTable; where: Condition | undefined; set: Record<string, unknown> } | undefined;
639 if (plan.guard) {
640 const table = toTable(plan.guard.table);
641 let conditions = plan.guard.where ? buildCondition(table.columns, plan.guard.where) : undefined;
642 if (conditions === true) conditions = undefined;
643 if (conditions === false) return { applied: false };
644 const constrained = await applyUpdatePolicies(table, conditions, plan.guard.set, context, "update");
645 if (constrained === false) return { applied: false };
646 guard = { table, where: constrained, set: plan.guard.set };
647 }
648 const deletes: { table: AnyTable; where: Condition | undefined }[] = [];
649 for (const del of plan.deletes) {
650 const table = toTable(del.table);
651 let conditions = del.where ? buildCondition(table.columns, del.where) : undefined;
652 if (conditions === true) conditions = undefined;
653 if (conditions === false) continue;
654 const constrained = await applyDeletePolicies(table, conditions, context);
655 if (constrained === false) continue;
656 deletes.push({ table, where: constrained });
657 }
658 const inserts: { table: AnyTable; values: Record<string, unknown>[] }[] = [];
659 for (const ins of plan.inserts) {
660 const table = toTable(ins.table);
661 for (const value of ins.values) {
662 await runCreatePolicies(table, value, context);
663 }
664 if (ins.values.length > 0) inserts.push({ table, values: [...ins.values] });
665 }
666 return internal.replaceMany({ ...(guard ? { guard } : {}), deletes, inserts });
667 },
613 async transaction(run) {668 async transaction(run) {
614 return internal.transaction((transactionInstance) =>669 return internal.transaction((transactionInstance) =>
615 run(withQueryContext(transactionInstance, context)),670 run(withQueryContext(transactionInstance, context)),