From 3e990a88cb762fc867c8498e9a30c33f9a4ef61a Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Tue, 23 Jun 2026 20:53:19 +0530 Subject: [PATCH 01/10] feat(fumadb): add bulk upsert queries --- .../core/fumadb/src/adapters/drizzle/query.ts | 74 +++++++++++ .../core/fumadb/src/adapters/memory/index.ts | 25 ++++ packages/core/fumadb/src/query/index.ts | 16 +++ packages/core/fumadb/src/query/orm/index.ts | 118 +++++++++++++++++- .../fumadb/src/query/table-policy.test.ts | 67 +++++++++- packages/core/sdk/src/fuma-runtime.ts | 1 + packages/core/sdk/src/test-config.ts | 9 ++ 7 files changed, 307 insertions(+), 3 deletions(-) diff --git a/packages/core/fumadb/src/adapters/drizzle/query.ts b/packages/core/fumadb/src/adapters/drizzle/query.ts index 9471ecda0d..27330d8e67 100644 --- a/packages/core/fumadb/src/adapters/drizzle/query.ts +++ b/packages/core/fumadb/src/adapters/drizzle/query.ts @@ -121,6 +121,16 @@ function buildWhere( ); } +function countConditionParameters(condition: Condition): number { + if (condition.type === ConditionType.Compare) { + if (condition.b instanceof Column) return 0; + if (Array.isArray(condition.b)) return condition.b.length; + return 1; + } + if (condition.type === ConditionType.Not) return countConditionParameters(condition.item); + return condition.items.reduce((count, item) => count + countConditionParameters(item), 0); +} + function mapValues( values: Record, table: AnyTable @@ -303,6 +313,70 @@ export function fromDrizzle( await this.createMany(table, [v.create]); } }, + async upsertMany(table, v) { + if (v.values.length === 0) return; + if (v.update.length === 0) { + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: adapter rejects invalid upsert shape + throw new Error("[FumaDB] upsertMany requires at least one update column."); + } + if (provider !== "sqlite" && provider !== "postgresql") { + for (const value of v.values) { + const targetCondition: Condition = { + type: ConditionType.And, + items: v.target.map((column) => ({ + type: ConditionType.Compare, + a: column, + operator: "=", + b: value[column.ormName], + })), + }; + await this.upsert(table, { + where: v.where + ? { type: ConditionType.And, items: [targetCondition, v.where] } + : targetCondition, + update: Object.fromEntries( + v.update.map((column) => [column.ormName, value[column.ormName]]), + ), + create: value, + }); + } + return; + } + + const drizzleTable = toDrizzle(table); + const values = v.values.map((value) => mapValues(value, table)); + const where = v.where ? buildWhere(toDrizzleColumn, v.where) : undefined; + const whereParameters = v.where ? countConditionParameters(v.where) : 0; + const columnsPerRow = values.length > 0 ? Math.max(1, Object.keys(values[0]!).length) : 1; + const batchSize = maxBoundParameters + ? Math.max( + 1, + Math.min( + CREATE_MANY_BATCH_SIZE, + Math.floor(Math.max(1, maxBoundParameters - whereParameters) / columnsPerRow), + ), + ) + : CREATE_MANY_BATCH_SIZE; + const target = v.target.map((column) => drizzleTable[column.names.drizzle]); + const set = Object.fromEntries( + v.update.map((column) => [ + column.names.drizzle, + Drizzle.sql.raw(`excluded.${column.names.sql}`), + ]), + ); + + for (let i = 0; i < values.length; i += batchSize) { + const batch = values.slice(i, i + batchSize); + await (db as any) + .insert(drizzleTable) + .values(batch) + .onConflictDoUpdate({ + target, + set, + ...(where === undefined ? {} : { where }), + }); + } + }, async findMany(table, v) { return ( await db.query[table.names.drizzle].findMany(buildQueryConfig(table, v)) diff --git a/packages/core/fumadb/src/adapters/memory/index.ts b/packages/core/fumadb/src/adapters/memory/index.ts index a596373f5c..eebb973f1d 100644 --- a/packages/core/fumadb/src/adapters/memory/index.ts +++ b/packages/core/fumadb/src/adapters/memory/index.ts @@ -174,6 +174,31 @@ export function memoryAdapter(options: MemoryAdapterOptions = {}): FumaDBAdapter } await this.create(table, v.create); }, + async upsertMany(table, v) { + if (v.update.length === 0) { + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: adapter rejects invalid upsert shape + throw new Error("[FumaDB] upsertMany requires at least one update column."); + } + for (const value of v.values) { + const existing = tableRows(db, table).find( + (row) => + matchesCondition(row, v.where) && + v.target.every((column) => row[column.ormName] === value[column.ormName]), + ); + if (existing) { + Object.assign( + existing, + cloneValue( + Object.fromEntries( + v.update.map((column) => [column.ormName, value[column.ormName]]), + ), + ), + ); + continue; + } + await this.create(table, value); + } + }, async create(table, values) { const row = applyDefaults(table, values); tableRows(db, table).push(row); diff --git a/packages/core/fumadb/src/query/index.ts b/packages/core/fumadb/src/query/index.ts index 2128adb0ef..45633dfb94 100644 --- a/packages/core/fumadb/src/query/index.ts +++ b/packages/core/fumadb/src/query/index.ts @@ -174,6 +174,22 @@ export interface AbstractQuery { } ) => Promise; + /** + * Bulk upsert rows by a unique column target. + * + * Adapters with native conflict support should implement this as one or more + * `INSERT ... ON CONFLICT ... DO UPDATE` statements. Other adapters may fall + * back to per-row `upsert`. + */ + upsertMany: ( + table: TableName, + v: { + target: (keyof S["tables"][TableName]["columns"])[]; + update: (keyof S["tables"][TableName]["columns"])[]; + values: TableToInsertValues[]; + } + ) => Promise; + /** * Note: you cannot update the id of a row, some databases don't support that (including MongoDB). */ diff --git a/packages/core/fumadb/src/query/orm/index.ts b/packages/core/fumadb/src/query/orm/index.ts index f0bac42e78..2435c87cd8 100644 --- a/packages/core/fumadb/src/query/orm/index.ts +++ b/packages/core/fumadb/src/query/orm/index.ts @@ -4,6 +4,7 @@ import type { AnySchema, AnyTable, } from "../../schema"; +import { Column } from "../../schema"; import type { AbstractQuery, AnySelectClause, @@ -12,7 +13,12 @@ import type { JoinBuilder, OrderBy, } from ".."; -import { buildCondition, createBuilder, type Condition } from "../condition-builder"; +import { + buildCondition, + createBuilder, + type Condition, + ConditionType, +} from "../condition-builder"; export interface CompiledJoin { relation: AnyRelation; @@ -231,6 +237,27 @@ const applyUpdatePolicies = async ( return nextWhere; }; +const conditionKey = (condition: Condition | undefined): string => { + if (!condition) return "none"; + if (condition.type === ConditionType.Compare) { + const right = + condition.b instanceof Column ? { column: condition.b.ormName } : { value: condition.b }; + return JSON.stringify({ + type: "compare", + left: condition.a.ormName, + operator: condition.operator, + right, + }); + } + if (condition.type === ConditionType.Not) { + return JSON.stringify({ type: "not", item: conditionKey(condition.item) }); + } + return JSON.stringify({ + type: condition.type === ConditionType.And ? "and" : "or", + items: condition.items.map(conditionKey), + }); +}; + const applyDeletePolicies = async ( table: AnyTable, where: Condition | undefined, @@ -284,6 +311,16 @@ export interface ORMAdapter { }, ) => Promise; + upsertMany?: ( + table: AnyTable, + v: { + target: AnyColumn[]; + update: AnyColumn[]; + values: Record[]; + where?: Condition; + }, + ) => Promise; + create: ( table: AnyTable, values: Record, @@ -367,6 +404,85 @@ export function toORM( ...options, }); }, + async upsertMany(name, { target, update, values }) { + const table = toTable(name); + if (values.length === 0) return; + + const targetColumns = target.map((columnName) => { + const column = table.columns[columnName as string]; + if (!column) throw new Error(`[FumaDB] unknown column name ${String(columnName)}.`); + return column; + }); + const updateColumns = update.map((columnName) => { + const column = table.columns[columnName as string]; + if (!column) throw new Error(`[FumaDB] unknown column name ${String(columnName)}.`); + return column; + }); + + const builder = createBuilder(table.columns); + const permittedRows: { + readonly value: Record; + readonly where: Condition | undefined; + }[] = []; + for (const value of values) { + const updateValues = Object.fromEntries( + updateColumns.map((column) => [column.ormName, value[column.ormName]]), + ); + const constrainedWhere = await applyUpdatePolicies( + table, + undefined, + updateValues, + context, + "upsert", + value, + ); + if (constrainedWhere === false) continue; + await runCreatePolicies(table, value, context); + permittedRows.push({ value, where: constrainedWhere }); + } + if (permittedRows.length === 0) return; + + if (internal.upsertMany) { + const groups = new Map< + string, + { readonly where: Condition | undefined; readonly values: Record[] } + >(); + for (const row of permittedRows) { + const key = conditionKey(row.where); + const group = groups.get(key); + if (group) { + group.values.push(row.value); + } else { + groups.set(key, { where: row.where, values: [row.value] }); + } + } + for (const group of groups.values()) { + await internal.upsertMany(table, { + target: targetColumns, + update: updateColumns, + values: group.values, + where: group.where, + }); + } + return; + } + + for (const row of permittedRows) { + const value = row.value; + const targetWhere = builder.and( + ...targetColumns.map((column) => builder(column.ormName, "=", value[column.ormName])), + ); + const where = builder.and(targetWhere, row.where ?? true); + if (where === false) continue; + await internal.upsert(table, { + where: where === true ? undefined : where, + update: Object.fromEntries( + updateColumns.map((column) => [column.ormName, value[column.ormName]]), + ), + create: value, + }); + } + }, async create(name, values) { const table = toTable(name); await runCreatePolicies(table, values, context); diff --git a/packages/core/fumadb/src/query/table-policy.test.ts b/packages/core/fumadb/src/query/table-policy.test.ts index 51d0c2b8c5..a16d4459bb 100644 --- a/packages/core/fumadb/src/query/table-policy.test.ts +++ b/packages/core/fumadb/src/query/table-policy.test.ts @@ -382,6 +382,24 @@ describe("FumaDB table policies", () => { title: "A Three", }, }); + await tenantA.upsertMany("posts", { + target: ["id"], + update: ["title"], + values: [ + { + id: "post-a-1", + tenantId: "tenant-a", + authorId: "author-a", + title: "tenant-a-bulk-upserted", + }, + { + id: "post-a-4", + tenantId: "tenant-a", + authorId: "author-a", + title: "A Four", + }, + ], + }); await expect( tenantA.findMany("posts", { @@ -391,7 +409,7 @@ describe("FumaDB table policies", () => { ).resolves.toEqual([ { id: "post-a-1", - title: "tenant-a-updated", + title: "tenant-a-bulk-upserted", }, { id: "post-a-2", @@ -401,6 +419,10 @@ describe("FumaDB table policies", () => { id: "post-a-3", title: "A Three", }, + { + id: "post-a-4", + title: "A Four", + }, ]); expect(tenantAContext.observed).toEqual( @@ -471,7 +493,7 @@ describe("FumaDB table policies", () => { ); it.effect( - "rejects out-of-context writes across createMany, updateMany, upsert, and transactions", + "rejects out-of-context writes across createMany, updateMany, upsert, upsertMany, and transactions", () => useHarness(async (orm) => { await seedTenants(orm); @@ -524,6 +546,47 @@ describe("FumaDB table policies", () => { }), ).rejects.toThrow("tenant tenant-b is not allowed for posts"); + await expect( + tenantA.upsertMany("posts", { + target: ["id"], + update: ["title"], + values: [ + { + id: "post-a-bulk-upsert", + tenantId: "tenant-a", + authorId: "author-a", + title: "A bulk upsert", + }, + { + id: "post-b-bulk-upsert", + tenantId: "tenant-b", + authorId: "author-b", + title: "B bulk upsert", + }, + ], + }), + ).rejects.toThrow("tenant tenant-b is not allowed for posts"); + await expect( + tenantA.findFirst("posts", { + where: (builder) => builder("id", "=", "post-a-bulk-upsert"), + }), + ).resolves.toBeNull(); + + await expect( + tenantA.upsertMany("posts", { + target: ["id"], + update: ["tenantId"], + values: [ + { + id: "post-a-1", + tenantId: "tenant-b", + authorId: "author-b", + title: "tenant move", + }, + ], + }), + ).rejects.toThrow("tenant tenant-b is not allowed for posts"); + await expect( tenantA.transaction(async (tx) => { await tx.create("posts", { diff --git a/packages/core/sdk/src/fuma-runtime.ts b/packages/core/sdk/src/fuma-runtime.ts index ae5c4ebcc5..19852d48d3 100644 --- a/packages/core/sdk/src/fuma-runtime.ts +++ b/packages/core/sdk/src/fuma-runtime.ts @@ -285,6 +285,7 @@ const makeSafeFumaQuery = ( db.transaction((transactionDb) => run(makeSafeFumaQuery(transactionDb, options))), updateMany: (name, value) => db.updateMany(table(name), value), upsert: (name, value) => db.upsert(table(name), value), + upsertMany: (name, value) => db.upsertMany(table(name), value), }; return Object.freeze(query); diff --git a/packages/core/sdk/src/test-config.ts b/packages/core/sdk/src/test-config.ts index f452b94db4..cbb64e1dad 100644 --- a/packages/core/sdk/src/test-config.ts +++ b/packages/core/sdk/src/test-config.ts @@ -53,6 +53,14 @@ const makeLazyTestFumaDb = (options: { transaction: async (run) => (await start()).db.internal.transaction(run), updateMany: async (table, value) => (await start()).db.internal.updateMany(table, value), upsert: async (table, value) => (await start()).db.internal.upsert(table, value), + upsertMany: async (table, value) => { + const actual = await start(); + if (!actual.db.internal.upsertMany) { + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: lazy test DB must expose the current FumaDB adapter surface + throw new Error("[FumaDB] upsertMany is not supported by this adapter."); + } + return actual.db.internal.upsertMany(table, value); + }, }; const queryMethods = new Set([ @@ -65,6 +73,7 @@ const makeLazyTestFumaDb = (options: { "transaction", "updateMany", "upsert", + "upsertMany", ]); const makeDb = (context?: ExecutorOwnerPolicyContext): FumaDb => From 22b67d18d50821d3408bd47b678380d1d9a7e0fc Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Tue, 23 Jun 2026 20:53:27 +0530 Subject: [PATCH 02/10] perf(sdk): upsert plugin storage bulk writes --- .changeset/plugin-storage-bulk-upserts.md | 7 ++ packages/core/sdk/src/executor.ts | 77 +++++++++++++------- packages/core/sdk/src/plugin-storage.test.ts | 12 +-- packages/core/sdk/src/plugin-storage.ts | 21 ++++++ 4 files changed, 82 insertions(+), 35 deletions(-) create mode 100644 .changeset/plugin-storage-bulk-upserts.md diff --git a/.changeset/plugin-storage-bulk-upserts.md b/.changeset/plugin-storage-bulk-upserts.md new file mode 100644 index 0000000000..4265a60e98 --- /dev/null +++ b/.changeset/plugin-storage-bulk-upserts.md @@ -0,0 +1,7 @@ +--- +"@executor-js/fumadb": patch +"@executor-js/sdk": patch +--- + +Add a FumaDB bulk upsert query path and route plugin-storage bulk writes through +it so existing rows are updated without delete/reinsert churn. diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index f5ad432e79..5e56b7fe61 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -183,7 +183,6 @@ import { makeShapeMemory, observedShapeToJsonSchema, SHAPE_MEMORY_PLUGIN_ID } fr import { isUnauthorizedToolFailure } from "./auth-tool-failure"; const PLUGIN_STORAGE_DELETE_KEY_BATCH_SIZE = 90; -const PLUGIN_STORAGE_CREATE_ROW_BATCH_SIZE = 90; const MAX_APPROVAL_ARGUMENT_PREVIEW_CHARS = 4_000; // --------------------------------------------------------------------------- @@ -974,6 +973,14 @@ type LooseStorageDb = { tableName: string, rows: readonly Record[], ) => Promise; + readonly upsertMany: ( + tableName: string, + options: { + readonly target: readonly string[]; + readonly update: readonly string[]; + readonly values: readonly Record[]; + }, + ) => Promise; readonly deleteMany: (tableName: string, options?: unknown) => Promise; readonly findFirst: ( tableName: string, @@ -1010,6 +1017,19 @@ const makeCoreDb = (fuma: ReturnType) => ({ : fuma .use(`${tableName}.createMany`, (db) => asLooseStorageDb(db).createMany(tableName, rows)) .pipe(Effect.asVoid), + upsertMany: ( + tableName: TName, + options: { + readonly target: readonly string[]; + readonly update: readonly string[]; + readonly values: readonly Record[]; + }, + ): Effect.Effect => + options.values.length === 0 + ? Effect.void + : fuma.use(`${tableName}.upsertMany`, (db) => + asLooseStorageDb(db).upsertMany(tableName, options), + ), deleteMany: ( tableName: TName, options: { readonly where?: CoreWhere } = {}, @@ -1359,33 +1379,22 @@ const makePluginStorageFacade = (input: { const uniqueEntries = [...entriesById.values()]; if (uniqueEntries.length === 0) return; - yield* deleteManyImpl(owner, os.subject, uniqueEntries); - const now = new Date(); - for ( - let offset = 0; - offset < uniqueEntries.length; - offset += PLUGIN_STORAGE_CREATE_ROW_BATCH_SIZE - ) { - const batchEntries = uniqueEntries.slice( - offset, - offset + PLUGIN_STORAGE_CREATE_ROW_BATCH_SIZE, - ); - yield* input.core.createMany( - "plugin_storage", - batchEntries.map((entry) => ({ - tenant, - owner: os.owner, - subject: os.subject, - plugin_id: input.pluginId, - collection: entry.collection, - key: entry.key, - data: entry.data, - created_at: now, - updated_at: now, - })), - ); - } + yield* input.core.upsertMany("plugin_storage", { + target: ["tenant", "owner", "subject", "plugin_id", "collection", "key"], + update: ["data", "updated_at"], + values: uniqueEntries.map((entry) => ({ + tenant, + owner: os.owner, + subject: os.subject, + plugin_id: input.pluginId, + collection: entry.collection, + key: entry.key, + data: entry.data, + created_at: now, + updated_at: now, + })), + }); }); const removeManyImpl = ( @@ -1482,10 +1491,24 @@ const makePluginStorageFacade = (input: { PluginStorageEntry>, StorageFailure >, + putMany: (storageInput) => + putManyImpl( + storageInput.owner, + storageInput.entries.map((entry) => ({ + collection: definition.name, + key: entry.key, + data: entry.data, + })), + ), query: (storageInput) => queryCollection(definition, storageInput), count: (storageInput) => queryCollection(definition, storageInput).pipe(Effect.map((rows) => rows.length)), remove: (storageInput) => removeImpl(storageInput.owner, definition.name, storageInput.key), + removeMany: (storageInput) => + removeManyImpl( + storageInput.owner, + storageInput.keys.map((key) => ({ collection: definition.name, key })), + ), }), get: (storageInput) => getVisible(storageInput.collection, storageInput.key), getForOwner: (storageInput) => diff --git a/packages/core/sdk/src/plugin-storage.test.ts b/packages/core/sdk/src/plugin-storage.test.ts index 95c7c0b714..8a4b919db3 100644 --- a/packages/core/sdk/src/plugin-storage.test.ts +++ b/packages/core/sdk/src/plugin-storage.test.ts @@ -65,18 +65,14 @@ const executionHistoryPlugin = definePlugin(() => ({ owner: Owner, rows: readonly { readonly key: string; readonly data: ToolCall }[], ) => - ctx.pluginStorage.putMany({ + ctx.storage.toolCalls.putMany({ owner, - entries: rows.map((row) => ({ - collection: toolCalls.name, - key: row.key, - data: row.data, - })), + entries: rows, }), removeMany: (owner: Owner, keys: readonly string[]) => - ctx.pluginStorage.removeMany({ + ctx.storage.toolCalls.removeMany({ owner, - entries: keys.map((key) => ({ collection: toolCalls.name, key })), + keys, }), get: (key: string) => ctx.storage.toolCalls.get({ key }), getForOwner: (owner: Owner, key: string) => ctx.storage.toolCalls.getForOwner({ owner, key }), diff --git a/packages/core/sdk/src/plugin-storage.ts b/packages/core/sdk/src/plugin-storage.ts index e854feb51b..1c0eec0464 100644 --- a/packages/core/sdk/src/plugin-storage.ts +++ b/packages/core/sdk/src/plugin-storage.ts @@ -135,6 +135,21 @@ export interface PluginStorageCollectionScopedKeyInput extends PluginStorageColl readonly owner: Owner; } +export interface PluginStorageCollectionPutManyEntry { + readonly key: string; + readonly data: TData; +} + +export interface PluginStorageCollectionPutManyInput { + readonly owner: Owner; + readonly entries: readonly PluginStorageCollectionPutManyEntry[]; +} + +export interface PluginStorageCollectionRemoveManyInput { + readonly owner: Owner; + readonly keys: readonly string[]; +} + export interface PluginStorageCollectionListInput { readonly keyPrefix?: string; } @@ -188,6 +203,9 @@ export interface PluginStorageCollectionFacade< readonly put: ( input: PluginStorageCollectionPutInput>, ) => Effect.Effect>, StorageFailure>; + readonly putMany: ( + input: PluginStorageCollectionPutManyInput>, + ) => Effect.Effect; readonly query: ( input?: PluginStorageCollectionQueryInput, ) => Effect.Effect< @@ -200,6 +218,9 @@ export interface PluginStorageCollectionFacade< readonly remove: ( input: PluginStorageCollectionScopedKeyInput, ) => Effect.Effect; + readonly removeMany: ( + input: PluginStorageCollectionRemoveManyInput, + ) => Effect.Effect; } export interface PluginStorageFacade { From 92f45b6d8750428e55a2eb1496669826a39ab0d9 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Wed, 24 Jun 2026 11:51:36 +0530 Subject: [PATCH 03/10] test(openapi): update plugin storage mock facade Add collection-level bulk methods to the OpenAPI store test stub so it satisfies the expanded PluginStorageCollectionFacade interface. --- packages/plugins/openapi/src/sdk/store.test.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/packages/plugins/openapi/src/sdk/store.test.ts b/packages/plugins/openapi/src/sdk/store.test.ts index ef4a4ed2cf..24b7494f08 100644 --- a/packages/plugins/openapi/src/sdk/store.test.ts +++ b/packages/plugins/openapi/src/sdk/store.test.ts @@ -49,9 +49,11 @@ describe("OpenAPI operation store", () => { data: input.data, }), ), + putMany: () => Effect.void, query: () => Effect.succeed([]), count: () => Effect.succeed(0), remove: () => Effect.void, + removeMany: () => Effect.void, }), get: (input: { readonly collection: string; readonly key: string }) => Effect.succeed( From da20d2a477715a9cfcbeec2c43ba89ae3a803de0 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Wed, 24 Jun 2026 12:35:22 +0530 Subject: [PATCH 04/10] fix(fumadb): validate bulk upsert conflict shapes --- .../core/fumadb/src/adapters/drizzle/query.ts | 24 +++++++++----- .../core/fumadb/src/adapters/memory/index.ts | 4 +++ packages/core/fumadb/src/query/orm/index.ts | 8 +++++ .../fumadb/src/query/table-policy.test.ts | 31 +++++++++++++++++++ 4 files changed, 59 insertions(+), 8 deletions(-) diff --git a/packages/core/fumadb/src/adapters/drizzle/query.ts b/packages/core/fumadb/src/adapters/drizzle/query.ts index 27330d8e67..20eb39efbc 100644 --- a/packages/core/fumadb/src/adapters/drizzle/query.ts +++ b/packages/core/fumadb/src/adapters/drizzle/query.ts @@ -315,6 +315,10 @@ export function fromDrizzle( }, async upsertMany(table, v) { if (v.values.length === 0) return; + if (v.target.length === 0) { + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: adapter rejects invalid upsert shape + throw new Error("[FumaDB] upsertMany requires at least one target column."); + } if (v.update.length === 0) { // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: adapter rejects invalid upsert shape throw new Error("[FumaDB] upsertMany requires at least one update column."); @@ -367,14 +371,18 @@ export function fromDrizzle( for (let i = 0; i < values.length; i += batchSize) { const batch = values.slice(i, i + batchSize); - await (db as any) - .insert(drizzleTable) - .values(batch) - .onConflictDoUpdate({ - target, - set, - ...(where === undefined ? {} : { where }), - }); + const insert = db.insert(drizzleTable).values(batch) as unknown as { + onConflictDoUpdate: (input: { + readonly target: typeof target; + readonly set: typeof set; + readonly where?: typeof where; + }) => Promise; + }; + await insert.onConflictDoUpdate({ + target, + set, + ...(where === undefined ? {} : { where }), + }); } }, async findMany(table, v) { diff --git a/packages/core/fumadb/src/adapters/memory/index.ts b/packages/core/fumadb/src/adapters/memory/index.ts index eebb973f1d..5c34debc5c 100644 --- a/packages/core/fumadb/src/adapters/memory/index.ts +++ b/packages/core/fumadb/src/adapters/memory/index.ts @@ -175,6 +175,10 @@ export function memoryAdapter(options: MemoryAdapterOptions = {}): FumaDBAdapter await this.create(table, v.create); }, async upsertMany(table, v) { + if (v.target.length === 0) { + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: adapter rejects invalid upsert shape + throw new Error("[FumaDB] upsertMany requires at least one target column."); + } if (v.update.length === 0) { // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: adapter rejects invalid upsert shape throw new Error("[FumaDB] upsertMany requires at least one update column."); diff --git a/packages/core/fumadb/src/query/orm/index.ts b/packages/core/fumadb/src/query/orm/index.ts index 2435c87cd8..fae5374450 100644 --- a/packages/core/fumadb/src/query/orm/index.ts +++ b/packages/core/fumadb/src/query/orm/index.ts @@ -407,6 +407,14 @@ export function toORM( async upsertMany(name, { target, update, values }) { const table = toTable(name); if (values.length === 0) return; + if (target.length === 0) { + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: public query rejects invalid upsert shape + throw new Error("[FumaDB] upsertMany requires at least one target column."); + } + if (update.length === 0) { + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: public query rejects invalid upsert shape + throw new Error("[FumaDB] upsertMany requires at least one update column."); + } const targetColumns = target.map((columnName) => { const column = table.columns[columnName as string]; diff --git a/packages/core/fumadb/src/query/table-policy.test.ts b/packages/core/fumadb/src/query/table-policy.test.ts index a16d4459bb..688ba06f27 100644 --- a/packages/core/fumadb/src/query/table-policy.test.ts +++ b/packages/core/fumadb/src/query/table-policy.test.ts @@ -482,6 +482,37 @@ describe("FumaDB table policies", () => { }), ); + it.effect("rejects invalid bulk upsert conflict shapes", () => + useHarness(async (orm) => { + await seedTenants(orm); + const tenantA = withQueryContext(orm, makeContext(["tenant-a"], "tenant-a")); + const values = [ + { + id: "post-a-bulk-upsert", + tenantId: "tenant-a", + authorId: "author-a", + title: "A bulk upsert", + }, + ]; + + await expect( + tenantA.upsertMany("posts", { + target: [], + update: ["title"], + values, + }), + ).rejects.toThrow("[FumaDB] upsertMany requires at least one target column."); + + await expect( + tenantA.upsertMany("posts", { + target: ["id"], + update: [], + values, + }), + ).rejects.toThrow("[FumaDB] upsertMany requires at least one update column."); + }), + ); + it.effect("fails closed when a query wrapper does not forward context rebinding", () => useHarness(async (orm) => { const wrapped = { ...orm }; From 1ca307dec0c3ab5686659e5eb8839cf44673155b Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Fri, 28 Aug 2026 01:06:46 +0530 Subject: [PATCH 05/10] fix(fumadb): batch bounded D1 upserts atomically --- .../core/fumadb/src/adapters/drizzle/query.ts | 31 ++++++-- .../fumadb/src/query/table-policy.test.ts | 74 ++++++++++++++++++- 2 files changed, 97 insertions(+), 8 deletions(-) diff --git a/packages/core/fumadb/src/adapters/drizzle/query.ts b/packages/core/fumadb/src/adapters/drizzle/query.ts index 20eb39efbc..72d6942f3b 100644 --- a/packages/core/fumadb/src/adapters/drizzle/query.ts +++ b/packages/core/fumadb/src/adapters/drizzle/query.ts @@ -369,6 +369,7 @@ export function fromDrizzle( ]), ); + const statements: unknown[] = []; for (let i = 0; i < values.length; i += batchSize) { const batch = values.slice(i, i + batchSize); const insert = db.insert(drizzleTable).values(batch) as unknown as { @@ -376,13 +377,31 @@ export function fromDrizzle( readonly target: typeof target; readonly set: typeof set; readonly where?: typeof where; - }) => Promise; + }) => unknown; }; - await insert.onConflictDoUpdate({ - target, - set, - ...(where === undefined ? {} : { where }), - }); + statements.push( + insert.onConflictDoUpdate({ + target, + set, + ...(where === undefined ? {} : { where }), + }), + ); + } + + // D1 rejects interactive transactions but its native batch API executes + // prepared statements as one transaction. Drizzle exposes that API on + // the database handle, so keep parameter-bounded upserts atomic instead + // of auto-committing each statement independently. + const nativeBatch = db as unknown as { + readonly batch?: (statements: readonly unknown[]) => Promise; + }; + if (!interactiveTransactions && statements.length > 1 && nativeBatch.batch) { + await nativeBatch.batch(statements); + return; + } + + for (const statement of statements) { + await statement; } }, async findMany(table, v) { diff --git a/packages/core/fumadb/src/query/table-policy.test.ts b/packages/core/fumadb/src/query/table-policy.test.ts index 688ba06f27..38ae98ed27 100644 --- a/packages/core/fumadb/src/query/table-policy.test.ts +++ b/packages/core/fumadb/src/query/table-policy.test.ts @@ -142,7 +142,10 @@ const makeContext = ( observed: [], }); -const makeHarness = async () => { +const makeHarness = async (options?: { + readonly nativeBatch?: boolean; + readonly maxBoundParameters?: number; +}) => { const sqlite = new Database(":memory:"); sqlite.pragma("foreign_keys = ON"); const runtimeSchema = createDrizzleRuntimeSchemaFromTables({ @@ -152,6 +155,16 @@ const makeHarness = async () => { provider: "sqlite", }); const drizzleDb = drizzle(sqlite, { schema: runtimeSchema }); + let batchCalls = 0; + + if (options?.nativeBatch) { + Object.assign(drizzleDb, { + batch: async (queries: readonly { run: () => unknown }[]) => { + batchCalls += 1; + return sqlite.transaction(() => queries.map((query) => query.run()))(); + }, + }); + } for (const statement of createDrizzleRuntimeSchemaSqlFromTables({ tables: v1.tables, @@ -166,11 +179,14 @@ const makeHarness = async () => { drizzleAdapter({ db: drizzleDb, provider: "sqlite", + interactiveTransactions: options?.nativeBatch ? false : undefined, + maxBoundParameters: options?.maxBoundParameters, }), ); return { orm: client.orm("1.0.0"), + getBatchCalls: () => batchCalls, close: async () => { sqlite.close(); }, @@ -179,11 +195,23 @@ const makeHarness = async () => { const useHarness = (run: (orm: TablePolicyQuery) => Promise) => Effect.acquireUseRelease( - Effect.promise(makeHarness), + Effect.promise(() => makeHarness()), ({ orm }) => Effect.promise(() => run(orm)), ({ close }) => Effect.promise(close), ); +const useNativeBatchHarness = ( + run: (harness: { + readonly orm: TablePolicyQuery; + readonly getBatchCalls: () => number; + }) => Promise, +) => + Effect.acquireUseRelease( + Effect.promise(() => makeHarness({ nativeBatch: true, maxBoundParameters: 8 })), + (harness) => Effect.promise(() => run(harness)), + ({ close }) => Effect.promise(close), + ); + const seedTenants = async (orm: TablePolicyQuery) => { const seed = withQueryContext(orm, makeContext(["tenant-a", "tenant-b"], "seed")); @@ -513,6 +541,48 @@ describe("FumaDB table policies", () => { }), ); + it.effect("rolls back every bounded upsert statement when a native batch fails", () => + useNativeBatchHarness(async ({ orm, getBatchCalls }) => { + await seedTenants(orm); + const tenantA = withQueryContext(orm, makeContext(["tenant-a"], "tenant-a")); + + await expect( + tenantA.upsertMany("posts", { + target: ["id"], + update: ["title"], + values: [ + { + id: "post-a-batch-1", + tenantId: "tenant-a", + authorId: "author-a", + title: "A batch one", + }, + { + id: "post-a-batch-2", + tenantId: "tenant-a", + authorId: "author-a", + title: "A batch two", + }, + { + id: "post-a-batch-invalid", + tenantId: "tenant-a", + authorId: "missing-author", + title: "Must roll back", + }, + ], + }), + ).rejects.toThrow(); + + expect(getBatchCalls()).toBe(1); + await expect( + tenantA.findMany("posts", { + where: (builder) => builder("id", "starts with", "post-a-batch-"), + select: ["id"], + }), + ).resolves.toEqual([]); + }), + ); + it.effect("fails closed when a query wrapper does not forward context rebinding", () => useHarness(async (orm) => { const wrapped = { ...orm }; From aa9c25ec2474e663cbfdf37b5bc30ead74097ef7 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Fri, 28 Aug 2026 02:13:29 +0530 Subject: [PATCH 06/10] fix(sdk): transact bulk plugin storage writes --- packages/core/sdk/src/executor.ts | 76 ++++++++------- packages/core/sdk/src/plugin-storage.test.ts | 98 +++++++++++++++++++- 2 files changed, 136 insertions(+), 38 deletions(-) diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 5e56b7fe61..0186e329e5 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -996,6 +996,8 @@ type LooseStorageDb = { const asLooseStorageDb = (db: unknown): LooseStorageDb => db as LooseStorageDb; const makeCoreDb = (fuma: ReturnType) => ({ + transaction: (effect: Effect.Effect): Effect.Effect => + fuma.transaction(effect), count: ( tableName: TName, options?: { readonly where?: CoreWhere }, @@ -1358,44 +1360,46 @@ const makePluginStorageFacade = (input: { readonly data: unknown; }[], ) => - Effect.gen(function* () { - const os = ownerSubject(owner); - if (!os) { - return yield* new StorageError({ - message: `Cannot write plugin storage for owner "user": executor has no subject.`, - cause: undefined, - }); - } - const entriesById = new Map( - entries.map((entry) => [ - pluginStorageId({ - pluginId: input.pluginId, + input.core.transaction( + Effect.gen(function* () { + const os = ownerSubject(owner); + if (!os) { + return yield* new StorageError({ + message: `Cannot write plugin storage for owner "user": executor has no subject.`, + cause: undefined, + }); + } + const entriesById = new Map( + entries.map((entry) => [ + pluginStorageId({ + pluginId: input.pluginId, + collection: entry.collection, + key: entry.key, + }), + entry, + ]), + ); + const uniqueEntries = [...entriesById.values()]; + if (uniqueEntries.length === 0) return; + + const now = new Date(); + yield* input.core.upsertMany("plugin_storage", { + target: ["tenant", "owner", "subject", "plugin_id", "collection", "key"], + update: ["data", "updated_at"], + values: uniqueEntries.map((entry) => ({ + tenant, + owner: os.owner, + subject: os.subject, + plugin_id: input.pluginId, collection: entry.collection, key: entry.key, - }), - entry, - ]), - ); - const uniqueEntries = [...entriesById.values()]; - if (uniqueEntries.length === 0) return; - - const now = new Date(); - yield* input.core.upsertMany("plugin_storage", { - target: ["tenant", "owner", "subject", "plugin_id", "collection", "key"], - update: ["data", "updated_at"], - values: uniqueEntries.map((entry) => ({ - tenant, - owner: os.owner, - subject: os.subject, - plugin_id: input.pluginId, - collection: entry.collection, - key: entry.key, - data: entry.data, - created_at: now, - updated_at: now, - })), - }); - }); + data: entry.data, + created_at: now, + updated_at: now, + })), + }); + }), + ); const removeManyImpl = ( owner: Owner, diff --git a/packages/core/sdk/src/plugin-storage.test.ts b/packages/core/sdk/src/plugin-storage.test.ts index 8a4b919db3..c8ce24d1e8 100644 --- a/packages/core/sdk/src/plugin-storage.test.ts +++ b/packages/core/sdk/src/plugin-storage.test.ts @@ -1,7 +1,8 @@ import { describe, expect, it } from "@effect/vitest"; import { Cause, Effect, Exit, Schema } from "effect"; -import { StorageError } from "./fuma-runtime"; +import { createExecutor } from "./executor"; +import { StorageError, type FumaDb } from "./fuma-runtime"; import { Owner } from "./ids"; import { definePlugin } from "./plugin"; import { @@ -10,7 +11,7 @@ import { type PluginStorageCollectionQueryInput, type PluginStorageCollectionWhere, } from "./plugin-storage"; -import { makeTestExecutor } from "./testing"; +import { makeTestConfig, makeTestExecutor } from "./testing"; const ToolCall = Schema.Struct({ runId: Schema.String, @@ -109,6 +110,48 @@ const call = (input: { durationMs: input.durationMs ?? 0, }); +const failPluginStorageBulkWriteAfterFirstRow = (db: FumaDb): FumaDb => { + const wrap = (source: FumaDb, failBulkWrite: boolean): FumaDb => + new Proxy(source, { + get(target, property, receiver) { + if (property === "withContext") { + const withContext = target.withContext; + return withContext === undefined + ? undefined + : (context: unknown) => wrap(withContext(context), failBulkWrite); + } + if (property === "transaction") { + const transaction: FumaDb["transaction"] = (run) => + target.transaction((transactionDb) => run(wrap(transactionDb, true))); + return transaction; + } + if (property === "upsertMany" && failBulkWrite) { + const upsertMany: FumaDb["upsertMany"] = async (table, options) => { + if (table !== "plugin_storage" || options.values.length < 2) { + return target.upsertMany(table, options); + } + + await target.upsertMany(table, { + ...options, + values: options.values.slice(0, 1), + }); + // oxlint-disable-next-line executor/no-promise-reject -- boundary: fault-injecting FumaDB adapter must reject to exercise transaction rollback + return Promise.reject( + new StorageError({ + message: "Injected plugin storage bulk-write failure.", + cause: undefined, + }), + ); + }; + return upsertMany; + } + return Reflect.get(target, property, receiver); + }, + }); + + return wrap(db, false); +}; + describe("plugin storage collections", () => { it.effect("queries declared indexes through the executor's SQLite FumaDB target", () => Effect.gen(function* () { @@ -218,6 +261,57 @@ describe("plugin storage collections", () => { }), ); + it.effect("rolls back every plugin storage row when a bulk write fails", () => + Effect.gen(function* () { + const config = makeTestConfig({ + backend: "sqlite", + plugins: [executionHistoryPlugin] as const, + }); + const executor = yield* Effect.acquireRelease( + createExecutor({ + ...config, + db: failPluginStorageBulkWriteAfterFirstRow(config.db), + }), + (instance) => + instance + .close() + .pipe( + Effect.ignore, + Effect.andThen(Effect.promise(() => config.testDb.close()).pipe(Effect.ignore)), + ), + ); + + const exit = yield* Effect.exit( + executor.executionHistory.recordMany("org", [ + { + key: "call-first", + data: call({ + runId: "run-rollback", + toolId: "browser", + status: "ok", + startedAt: "2026-05-29T12:00:00.000Z", + }), + }, + { + key: "call-second", + data: call({ + runId: "run-rollback", + toolId: "shell", + status: "ok", + startedAt: "2026-05-29T12:01:00.000Z", + }), + }, + ]), + ); + expect(Exit.isFailure(exit)).toBe(true); + + const stored = yield* executor.executionHistory.query({ + where: { runId: "run-rollback" }, + }); + expect(stored).toEqual([]); + }), + ); + it.effect( "bulk puts large plugin storage row sets in bounded batches", () => From fcd13d1b9cb8e30974b85f645323528761569609 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Fri, 28 Aug 2026 02:11:05 -0700 Subject: [PATCH 07/10] Reconcile bulk plugin-storage upserts with the landed putMany API The collection-level putMany/removeMany the branch added duplicate no production consumer; both call sites use the facade. Drop them and keep the branch to its real contribution, the atomic upsert implementation. Add a case proving a mid-batch failure leaves pre-existing rows intact, and make the fault injector unconditional so removing the transaction fails the test on the data rather than silently disarming it. --- packages/core/sdk/src/executor.ts | 14 --- packages/core/sdk/src/plugin-storage.test.ts | 99 +++++++++++++++++-- packages/core/sdk/src/plugin-storage.ts | 21 ---- .../plugins/openapi/src/sdk/store.test.ts | 2 - 4 files changed, 90 insertions(+), 46 deletions(-) diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 26006ed85d..e59ebafb29 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -1530,24 +1530,10 @@ const makePluginStorageFacade = (input: { PluginStorageEntry>, StorageFailure >, - putMany: (storageInput) => - putManyImpl( - storageInput.owner, - storageInput.entries.map((entry) => ({ - collection: definition.name, - key: entry.key, - data: entry.data, - })), - ), query: (storageInput) => queryCollection(definition, storageInput), count: (storageInput) => queryCollection(definition, storageInput).pipe(Effect.map((rows) => rows.length)), remove: (storageInput) => removeImpl(storageInput.owner, definition.name, storageInput.key), - removeMany: (storageInput) => - removeManyImpl( - storageInput.owner, - storageInput.keys.map((key) => ({ collection: definition.name, key })), - ), }), get: (storageInput) => getVisible(storageInput.collection, storageInput.key), getForOwner: (storageInput) => diff --git a/packages/core/sdk/src/plugin-storage.test.ts b/packages/core/sdk/src/plugin-storage.test.ts index c8ce24d1e8..e42c20408e 100644 --- a/packages/core/sdk/src/plugin-storage.test.ts +++ b/packages/core/sdk/src/plugin-storage.test.ts @@ -66,14 +66,18 @@ const executionHistoryPlugin = definePlugin(() => ({ owner: Owner, rows: readonly { readonly key: string; readonly data: ToolCall }[], ) => - ctx.storage.toolCalls.putMany({ + ctx.pluginStorage.putMany({ owner, - entries: rows, + entries: rows.map((row) => ({ + collection: toolCalls.name, + key: row.key, + data: row.data, + })), }), removeMany: (owner: Owner, keys: readonly string[]) => - ctx.storage.toolCalls.removeMany({ + ctx.pluginStorage.removeMany({ owner, - keys, + entries: keys.map((key) => ({ collection: toolCalls.name, key })), }), get: (key: string) => ctx.storage.toolCalls.get({ key }), getForOwner: (owner: Owner, key: string) => ctx.storage.toolCalls.getForOwner({ owner, key }), @@ -110,22 +114,29 @@ const call = (input: { durationMs: input.durationMs ?? 0, }); +// A FumaDB that commits the FIRST row of a multi-row `plugin_storage` bulk +// write and then fails. Injecting the fault mid-write, rather than before it, +// is what makes the two rollback cases below meaningful: the row is really on +// disk when the failure lands, so only the enclosing transaction can take it +// back. The fault is unconditional (not armed by entering a transaction) so +// that dropping the transaction is a visible failure and not a silently +// disarmed test. const failPluginStorageBulkWriteAfterFirstRow = (db: FumaDb): FumaDb => { - const wrap = (source: FumaDb, failBulkWrite: boolean): FumaDb => + const wrap = (source: FumaDb): FumaDb => new Proxy(source, { get(target, property, receiver) { if (property === "withContext") { const withContext = target.withContext; return withContext === undefined ? undefined - : (context: unknown) => wrap(withContext(context), failBulkWrite); + : (context: unknown) => wrap(withContext(context)); } if (property === "transaction") { const transaction: FumaDb["transaction"] = (run) => - target.transaction((transactionDb) => run(wrap(transactionDb, true))); + target.transaction((transactionDb) => run(wrap(transactionDb))); return transaction; } - if (property === "upsertMany" && failBulkWrite) { + if (property === "upsertMany") { const upsertMany: FumaDb["upsertMany"] = async (table, options) => { if (table !== "plugin_storage" || options.values.length < 2) { return target.upsertMany(table, options); @@ -149,7 +160,7 @@ const failPluginStorageBulkWriteAfterFirstRow = (db: FumaDb): FumaDb => { }, }); - return wrap(db, false); + return wrap(db); }; describe("plugin storage collections", () => { @@ -312,6 +323,76 @@ describe("plugin storage collections", () => { }), ); + // The hazard this whole change exists to remove: the previous implementation + // deleted every target key and only then re-created the rows, so a failure + // between the two halves destroyed data the caller never meant to touch. An + // upsert inside a transaction cannot lose a row it did not successfully + // replace, so the ORIGINAL values must still be readable after the failure — + // not merely absent-and-consistent like the rolled-back insert above. + it.effect("leaves pre-existing rows intact when a bulk overwrite fails mid-batch", () => + Effect.gen(function* () { + const config = makeTestConfig({ + backend: "sqlite", + plugins: [executionHistoryPlugin] as const, + }); + const executor = yield* Effect.acquireRelease( + createExecutor({ + ...config, + db: failPluginStorageBulkWriteAfterFirstRow(config.db), + }), + (instance) => + instance + .close() + .pipe( + Effect.ignore, + Effect.andThen(Effect.promise(() => config.testDb.close()).pipe(Effect.ignore)), + ), + ); + + // Seeded one row at a time, so the seeding itself never goes through the + // bulk path the fault injector breaks. + const original = [ + { + key: "call-first", + data: call({ + runId: "run-preexisting", + toolId: "browser", + status: "ok", + startedAt: "2026-05-29T12:00:00.000Z", + }), + }, + { + key: "call-second", + data: call({ + runId: "run-preexisting", + toolId: "shell", + status: "ok", + startedAt: "2026-05-29T12:01:00.000Z", + }), + }, + ]; + for (const row of original) { + yield* executor.executionHistory.record("org", row.key, row.data); + } + + const exit = yield* Effect.exit( + executor.executionHistory.recordMany( + "org", + original.map((row) => ({ + key: row.key, + data: { ...row.data, toolId: "overwritten", status: "failed" as const }, + })), + ), + ); + expect(Exit.isFailure(exit)).toBe(true); + + const first = yield* executor.executionHistory.get("call-first"); + const second = yield* executor.executionHistory.get("call-second"); + expect(first?.data).toEqual(original[0]!.data); + expect(second?.data).toEqual(original[1]!.data); + }), + ); + it.effect( "bulk puts large plugin storage row sets in bounded batches", () => diff --git a/packages/core/sdk/src/plugin-storage.ts b/packages/core/sdk/src/plugin-storage.ts index 1c0eec0464..e854feb51b 100644 --- a/packages/core/sdk/src/plugin-storage.ts +++ b/packages/core/sdk/src/plugin-storage.ts @@ -135,21 +135,6 @@ export interface PluginStorageCollectionScopedKeyInput extends PluginStorageColl readonly owner: Owner; } -export interface PluginStorageCollectionPutManyEntry { - readonly key: string; - readonly data: TData; -} - -export interface PluginStorageCollectionPutManyInput { - readonly owner: Owner; - readonly entries: readonly PluginStorageCollectionPutManyEntry[]; -} - -export interface PluginStorageCollectionRemoveManyInput { - readonly owner: Owner; - readonly keys: readonly string[]; -} - export interface PluginStorageCollectionListInput { readonly keyPrefix?: string; } @@ -203,9 +188,6 @@ export interface PluginStorageCollectionFacade< readonly put: ( input: PluginStorageCollectionPutInput>, ) => Effect.Effect>, StorageFailure>; - readonly putMany: ( - input: PluginStorageCollectionPutManyInput>, - ) => Effect.Effect; readonly query: ( input?: PluginStorageCollectionQueryInput, ) => Effect.Effect< @@ -218,9 +200,6 @@ export interface PluginStorageCollectionFacade< readonly remove: ( input: PluginStorageCollectionScopedKeyInput, ) => Effect.Effect; - readonly removeMany: ( - input: PluginStorageCollectionRemoveManyInput, - ) => Effect.Effect; } export interface PluginStorageFacade { diff --git a/packages/plugins/openapi/src/sdk/store.test.ts b/packages/plugins/openapi/src/sdk/store.test.ts index 24b7494f08..ef4a4ed2cf 100644 --- a/packages/plugins/openapi/src/sdk/store.test.ts +++ b/packages/plugins/openapi/src/sdk/store.test.ts @@ -49,11 +49,9 @@ describe("OpenAPI operation store", () => { data: input.data, }), ), - putMany: () => Effect.void, query: () => Effect.succeed([]), count: () => Effect.succeed(0), remove: () => Effect.void, - removeMany: () => Effect.void, }), get: (input: { readonly collection: string; readonly key: string }) => Effect.succeed( From b5c57f9ba69494590cb274165479537c17e755aa Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Fri, 28 Aug 2026 11:56:57 -0700 Subject: [PATCH 08/10] Fix bulk upsert batching, predicate grouping, and memory conflict semantics --- .../core/fumadb/src/adapters/drizzle/query.ts | 45 +++-- .../core/fumadb/src/adapters/memory/index.ts | 16 +- packages/core/fumadb/src/query/orm/index.ts | 86 ++++++--- .../fumadb/src/query/table-policy.test.ts | 171 +++++++++++++++++- packages/core/sdk/src/plugin-storage.test.ts | 37 ++++ 5 files changed, 306 insertions(+), 49 deletions(-) diff --git a/packages/core/fumadb/src/adapters/drizzle/query.ts b/packages/core/fumadb/src/adapters/drizzle/query.ts index 72d6942f3b..84cd8710c6 100644 --- a/packages/core/fumadb/src/adapters/drizzle/query.ts +++ b/packages/core/fumadb/src/adapters/drizzle/query.ts @@ -22,6 +22,28 @@ type P_DBType = PostgreSQL.PgDatabase< const CREATE_MANY_BATCH_SIZE = 500; +// A multi-row write binds (rows * columns) parameters in one statement, and +// engines cap bound parameters per statement (older SQLite: 999, Cloudflare +// D1: 100). When the adapter does not advertise its limit, budget against the +// conservative 999 floor so a wide table cannot overflow with "too many SQL +// variables" on stricter engines. +const DEFAULT_MAX_BOUND_PARAMETERS = 999; + +function parameterBoundedBatchSize( + columnsPerRow: number, + reservedParameters: number, + maxBoundParameters: number | undefined +): number { + const budget = maxBoundParameters ?? DEFAULT_MAX_BOUND_PARAMETERS; + return Math.max( + 1, + Math.min( + CREATE_MANY_BATCH_SIZE, + Math.floor(Math.max(1, budget - reservedParameters) / columnsPerRow) + ) + ); +} + function buildWhere( toDrizzle: (col: AnyColumn) => ColumnType, condition: Condition @@ -352,15 +374,11 @@ export function fromDrizzle( const where = v.where ? buildWhere(toDrizzleColumn, v.where) : undefined; const whereParameters = v.where ? countConditionParameters(v.where) : 0; const columnsPerRow = values.length > 0 ? Math.max(1, Object.keys(values[0]!).length) : 1; - const batchSize = maxBoundParameters - ? Math.max( - 1, - Math.min( - CREATE_MANY_BATCH_SIZE, - Math.floor(Math.max(1, maxBoundParameters - whereParameters) / columnsPerRow), - ), - ) - : CREATE_MANY_BATCH_SIZE; + const batchSize = parameterBoundedBatchSize( + columnsPerRow, + whereParameters, + maxBoundParameters, + ); const target = v.target.map((column) => drizzleTable[column.names.drizzle]); const set = Object.fromEntries( v.update.map((column) => [ @@ -457,15 +475,8 @@ export function fromDrizzle( const idField = table.getIdColumn().names.drizzle; const drizzleTable = toDrizzle(table); values = values.map((v) => mapValues(v, table)); - // A multi-row insert binds (rows * columns) parameters in one statement. - // Some engines cap bound parameters per query (Cloudflare D1: 100), so - // size the batch by PARAMETER count, not row count — otherwise a wide - // table (e.g. tools) overflows with "too many SQL variables". Engines - // without a tight cap keep the row-count batch. const columnsPerRow = values.length > 0 ? Math.max(1, Object.keys(values[0]!).length) : 1; - const batchSize = maxBoundParameters - ? Math.max(1, Math.min(CREATE_MANY_BATCH_SIZE, Math.floor(maxBoundParameters / columnsPerRow))) - : CREATE_MANY_BATCH_SIZE; + const batchSize = parameterBoundedBatchSize(columnsPerRow, 0, maxBoundParameters); const batches: (typeof values)[] = []; for (let i = 0; i < values.length; i += batchSize) { batches.push(values.slice(i, i + batchSize)); diff --git a/packages/core/fumadb/src/adapters/memory/index.ts b/packages/core/fumadb/src/adapters/memory/index.ts index 5c34debc5c..205655e9fa 100644 --- a/packages/core/fumadb/src/adapters/memory/index.ts +++ b/packages/core/fumadb/src/adapters/memory/index.ts @@ -184,14 +184,18 @@ export function memoryAdapter(options: MemoryAdapterOptions = {}): FumaDBAdapter throw new Error("[FumaDB] upsertMany requires at least one update column."); } for (const value of v.values) { - const existing = tableRows(db, table).find( - (row) => - matchesCondition(row, v.where) && - v.target.every((column) => row[column.ormName] === value[column.ormName]), + // Mirror SQL `ON CONFLICT ... DO UPDATE ... WHERE`: the unique + // target alone detects the conflict, and the predicate only + // decides whether the conflicting row may be updated. A + // policy-excluded conflict skips the row; it never inserts a + // duplicate of the unique target. + const conflicting = tableRows(db, table).find((row) => + v.target.every((column) => row[column.ormName] === value[column.ormName]), ); - if (existing) { + if (conflicting) { + if (!matchesCondition(conflicting, v.where)) continue; Object.assign( - existing, + conflicting, cloneValue( Object.fromEntries( v.update.map((column) => [column.ormName, value[column.ormName]]), diff --git a/packages/core/fumadb/src/query/orm/index.ts b/packages/core/fumadb/src/query/orm/index.ts index fae5374450..0796e4c848 100644 --- a/packages/core/fumadb/src/query/orm/index.ts +++ b/packages/core/fumadb/src/query/orm/index.ts @@ -237,25 +237,62 @@ const applyUpdatePolicies = async ( return nextWhere; }; -const conditionKey = (condition: Condition | undefined): string => { - if (!condition) return "none"; - if (condition.type === ConditionType.Compare) { - const right = - condition.b instanceof Column ? { column: condition.b.ormName } : { value: condition.b }; - return JSON.stringify({ - type: "compare", - left: condition.a.ormName, - operator: condition.operator, - right, - }); +// Structural equality over predicate values. Policy predicates may carry any +// column value shape — string, number, bigint, boolean, null, Date, binary, +// JSON objects, and arrays of those — so serializing them to string keys +// (e.g. JSON.stringify) either throws (bigint) or collides (values with the +// same serialized form). Comparing structurally is unambiguous; a false +// negative only splits a group and never merges rows under the wrong +// predicate. +const predicateValuesEqual = (a: unknown, b: unknown): boolean => { + if (Object.is(a, b)) return true; + if (a instanceof Date || b instanceof Date) { + return a instanceof Date && b instanceof Date && a.getTime() === b.getTime(); } - if (condition.type === ConditionType.Not) { - return JSON.stringify({ type: "not", item: conditionKey(condition.item) }); + if (a instanceof Uint8Array || b instanceof Uint8Array) { + return ( + a instanceof Uint8Array && + b instanceof Uint8Array && + a.length === b.length && + a.every((byte, index) => byte === b[index]) + ); } - return JSON.stringify({ - type: condition.type === ConditionType.And ? "and" : "or", - items: condition.items.map(conditionKey), - }); + if (Array.isArray(a) || Array.isArray(b)) { + return ( + Array.isArray(a) && + Array.isArray(b) && + a.length === b.length && + a.every((item, index) => predicateValuesEqual(item, b[index])) + ); + } + if (isRecord(a) && isRecord(b)) { + const aKeys = Object.keys(a); + const bKeys = Object.keys(b); + return ( + aKeys.length === bKeys.length && + aKeys.every((key) => Object.hasOwn(b, key) && predicateValuesEqual(a[key], b[key])) + ); + } + return false; +}; + +const conditionsEqual = (a: Condition | undefined, b: Condition | undefined): boolean => { + if (a === undefined || b === undefined) return a === b; + if (a.type === ConditionType.Compare || b.type === ConditionType.Compare) { + if (a.type !== ConditionType.Compare || b.type !== ConditionType.Compare) return false; + if (a.a !== b.a || a.operator !== b.operator) return false; + if (a.b instanceof Column || b.b instanceof Column) return a.b === b.b; + return predicateValuesEqual(a.b, b.b); + } + if (a.type === ConditionType.Not || b.type === ConditionType.Not) { + if (a.type !== ConditionType.Not || b.type !== ConditionType.Not) return false; + return conditionsEqual(a.item, b.item); + } + return ( + a.type === b.type && + a.items.length === b.items.length && + a.items.every((item, index) => conditionsEqual(item, b.items[index])) + ); }; const applyDeletePolicies = async ( @@ -451,20 +488,19 @@ export function toORM( if (permittedRows.length === 0) return; if (internal.upsertMany) { - const groups = new Map< - string, - { readonly where: Condition | undefined; readonly values: Record[] } - >(); + const groups: { + readonly where: Condition | undefined; + readonly values: Record[]; + }[] = []; for (const row of permittedRows) { - const key = conditionKey(row.where); - const group = groups.get(key); + const group = groups.find((candidate) => conditionsEqual(candidate.where, row.where)); if (group) { group.values.push(row.value); } else { - groups.set(key, { where: row.where, values: [row.value] }); + groups.push({ where: row.where, values: [row.value] }); } } - for (const group of groups.values()) { + for (const group of groups) { await internal.upsertMany(table, { target: targetColumns, update: updateColumns, diff --git a/packages/core/fumadb/src/query/table-policy.test.ts b/packages/core/fumadb/src/query/table-policy.test.ts index 38ae98ed27..b0c6d8bb2b 100644 --- a/packages/core/fumadb/src/query/table-policy.test.ts +++ b/packages/core/fumadb/src/query/table-policy.test.ts @@ -8,6 +8,7 @@ import { createDrizzleRuntimeSchemaSqlFromTables, drizzleAdapter, } from "@executor-js/fumadb/adapters/drizzle"; +import { memoryAdapter } from "@executor-js/fumadb/adapters/memory"; import { withQueryContext, type AbstractQuery } from "@executor-js/fumadb/query"; import { column, idColumn, schema, table } from "@executor-js/fumadb/schema"; @@ -16,6 +17,7 @@ interface TenantPolicyContext { readonly deniedTables: ReadonlySet; readonly marker: string; readonly observed: string[]; + readonly allowedRegionId?: bigint; } const observe = (context: TenantPolicyContext, event: string) => { @@ -103,12 +105,39 @@ const comments = table("policy_comments", { }, }); +// Policy predicate carries a bigint value; string-keyed grouping of these +// predicates (e.g. JSON.stringify) throws on bigint. +const quotas = table("policy_quotas", { + id: idColumn("id", "varchar(255)"), + region: column("region", "bigint"), + label: column("label", "string"), +}).policy({ + name: "tenant.quotas", + onUpdate: ({ builder, context }) => builder("region", "=", context.allowedRegionId ?? BigInt(-1)), +}); + +// Ten columns, so a row-count-sized batch would bind ten parameters per row. +const wideRows = table("policy_wide_rows", { + id: idColumn("id", "varchar(255)"), + c1: column("c1", "string"), + c2: column("c2", "string"), + c3: column("c3", "string"), + c4: column("c4", "string"), + c5: column("c5", "string"), + c6: column("c6", "string"), + c7: column("c7", "string"), + c8: column("c8", "string"), + c9: column("c9", "string"), +}); + const v1 = schema({ version: "1.0.0", tables: { authors, posts, comments, + quotas, + wideRows, }, relations: { authors: ({ many }) => ({ @@ -154,7 +183,15 @@ const makeHarness = async (options?: { version: "1.0.0", provider: "sqlite", }); - const drizzleDb = drizzle(sqlite, { schema: runtimeSchema }); + const statements: { readonly sql: string; readonly paramCount: number }[] = []; + const drizzleDb = drizzle(sqlite, { + schema: runtimeSchema, + logger: { + logQuery: (query: string, params: unknown[]) => { + statements.push({ sql: query, paramCount: params.length }); + }, + }, + }); let batchCalls = 0; if (options?.nativeBatch) { @@ -187,6 +224,8 @@ const makeHarness = async (options?: { return { orm: client.orm("1.0.0"), getBatchCalls: () => batchCalls, + getStatements: (): readonly { readonly sql: string; readonly paramCount: number }[] => + statements, close: async () => { sqlite.close(); }, @@ -200,6 +239,18 @@ const useHarness = (run: (orm: TablePolicyQuery) => Promise) => ({ close }) => Effect.promise(close), ); +const useStatementHarness = ( + run: (harness: { + readonly orm: TablePolicyQuery; + readonly getStatements: () => readonly { readonly sql: string; readonly paramCount: number }[]; + }) => Promise, +) => + Effect.acquireUseRelease( + Effect.promise(() => makeHarness()), + (harness) => Effect.promise(() => run(harness)), + ({ close }) => Effect.promise(close), + ); + const useNativeBatchHarness = ( run: (harness: { readonly orm: TablePolicyQuery; @@ -541,6 +592,124 @@ describe("FumaDB table policies", () => { }), ); + it.effect("keeps every bulk upsert statement inside the bound-variable budget", () => + useStatementHarness(async ({ orm, getStatements }) => { + // 250 rows * 10 columns = 2,500 bound variables in a single statement — + // over older SQLite's 999-variable cap. The adapter advertises no limit + // here, so batching must fall back to the conservative 999 budget. + const values = Array.from({ length: 250 }, (_, index) => ({ + id: `wide-${String(index).padStart(3, "0")}`, + c1: `value-${index}-1`, + c2: `value-${index}-2`, + c3: `value-${index}-3`, + c4: `value-${index}-4`, + c5: `value-${index}-5`, + c6: `value-${index}-6`, + c7: `value-${index}-7`, + c8: `value-${index}-8`, + c9: `value-${index}-9`, + })); + + await orm.upsertMany("wideRows", { + target: ["id"], + update: ["c1"], + values, + }); + + const inserts = getStatements().filter((statement) => + statement.sql.toLowerCase().startsWith("insert into"), + ); + expect(inserts.length).toBeGreaterThanOrEqual(3); + for (const statement of inserts) { + expect(statement.paramCount).toBeLessThanOrEqual(999); + } + await expect(orm.count("wideRows")).resolves.toBe(250); + }), + ); + + it.effect("bulk upserts through a policy predicate that contains a bigint", () => + useHarness(async (orm) => { + const region = withQueryContext(orm, { + ...makeContext(["tenant-a"], "region"), + allowedRegionId: BigInt(7), + }); + + await region.createMany("quotas", [ + { id: "quota-in-region", region: BigInt(7), label: "before" }, + { id: "quota-out-of-region", region: BigInt(9), label: "before" }, + ]); + + await region.upsertMany("quotas", { + target: ["id"], + update: ["label"], + values: [ + { id: "quota-in-region", region: BigInt(7), label: "after" }, + { id: "quota-out-of-region", region: BigInt(9), label: "after" }, + { id: "quota-created", region: BigInt(7), label: "created" }, + ], + }); + + await expect( + region.findMany("quotas", { + select: ["id", "label"], + orderBy: ["id", "asc"], + }), + ).resolves.toEqual([ + { id: "quota-created", label: "created" }, + { id: "quota-in-region", label: "after" }, + { id: "quota-out-of-region", label: "before" }, + ]); + }), + ); + + it.effect("skips policy-excluded conflicts identically on drizzle and memory adapters", () => { + // `post-b-1` already exists for tenant-b, so its unique target conflicts, + // and the tenant-a update policy excludes the conflicting row. SQL detects + // the conflict first and the predicate only gates the update, so the row + // is skipped — never duplicated. + const upsertAcrossPolicyExcludedConflict = async (orm: TablePolicyQuery) => { + await seedTenants(orm); + const tenantA = withQueryContext(orm, makeContext(["tenant-a"], "tenant-a")); + await tenantA.upsertMany("posts", { + target: ["id"], + update: ["title"], + values: [ + { + id: "post-b-1", + tenantId: "tenant-a", + authorId: "author-a", + title: "hijack attempt", + }, + { + id: "post-a-9", + tenantId: "tenant-a", + authorId: "author-a", + title: "A Nine", + }, + ], + }); + const allTenants = withQueryContext(orm, makeContext(["tenant-a", "tenant-b"], "all")); + return allTenants.findMany("posts", { + select: ["id", "tenantId", "title"], + orderBy: ["id", "asc"], + }); + }; + + const expected = [ + { id: "post-a-1", tenantId: "tenant-a", title: "A One" }, + { id: "post-a-2", tenantId: "tenant-a", title: "A Two" }, + { id: "post-a-9", tenantId: "tenant-a", title: "A Nine" }, + { id: "post-b-1", tenantId: "tenant-b", title: "B One" }, + ]; + + return useHarness(async (orm) => { + await expect(upsertAcrossPolicyExcludedConflict(orm)).resolves.toEqual(expected); + + const memoryOrm = tablePolicyDB.client(memoryAdapter()).orm("1.0.0"); + await expect(upsertAcrossPolicyExcludedConflict(memoryOrm)).resolves.toEqual(expected); + }); + }); + it.effect("rolls back every bounded upsert statement when a native batch fails", () => useNativeBatchHarness(async ({ orm, getBatchCalls }) => { await seedTenants(orm); diff --git a/packages/core/sdk/src/plugin-storage.test.ts b/packages/core/sdk/src/plugin-storage.test.ts index e42c20408e..3cc1d0793e 100644 --- a/packages/core/sdk/src/plugin-storage.test.ts +++ b/packages/core/sdk/src/plugin-storage.test.ts @@ -272,6 +272,43 @@ describe("plugin storage collections", () => { }), ); + it.effect("stores and overwrites every row when a bulk write spans multiple batches", () => + Effect.gen(function* () { + const executor = yield* makeTestExecutor({ + backend: "sqlite", + plugins: [executionHistoryPlugin] as const, + }); + // A plugin_storage row binds ~9 values, so 300 rows exceed one + // 999-bound-variable statement budget and must span several batches. + const entries = (status: "ok" | "failed") => + Array.from({ length: 300 }, (_, index) => ({ + key: `batched-call-${String(index).padStart(3, "0")}`, + data: call({ + runId: "run-batched", + toolId: "browser", + status, + startedAt: new Date(Date.UTC(2026, 4, 29, 12, 0, index)).toISOString(), + }), + })); + + yield* executor.executionHistory.recordMany("org", entries("ok")); + const total = yield* executor.executionHistory.count({ + where: { runId: "run-batched" }, + }); + expect(total).toBe(300); + + yield* executor.executionHistory.recordMany("org", entries("failed")); + const failed = yield* executor.executionHistory.count({ + where: { runId: "run-batched", status: "failed" }, + }); + expect(failed).toBe(300); + const totalAfterOverwrite = yield* executor.executionHistory.count({ + where: { runId: "run-batched" }, + }); + expect(totalAfterOverwrite).toBe(300); + }), + ); + it.effect("rolls back every plugin storage row when a bulk write fails", () => Effect.gen(function* () { const config = makeTestConfig({ From 514b77dba9512bb7d7ae2733b4afe4607c215e75 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Fri, 28 Aug 2026 12:23:41 -0700 Subject: [PATCH 09/10] Make split bulk writes atomic and fail fast on parameter budget overflow --- .../core/fumadb/src/adapters/drizzle/query.ts | 208 +++++++++++++----- packages/core/fumadb/src/query/orm/index.ts | 50 +++-- .../query/orm/predicate-values-equal.test.ts | 34 +++ .../fumadb/src/query/table-policy.test.ts | 154 +++++++++++++ 4 files changed, 372 insertions(+), 74 deletions(-) create mode 100644 packages/core/fumadb/src/query/orm/predicate-values-equal.test.ts diff --git a/packages/core/fumadb/src/adapters/drizzle/query.ts b/packages/core/fumadb/src/adapters/drizzle/query.ts index 84cd8710c6..6c86b82d59 100644 --- a/packages/core/fumadb/src/adapters/drizzle/query.ts +++ b/packages/core/fumadb/src/adapters/drizzle/query.ts @@ -30,18 +30,25 @@ const CREATE_MANY_BATCH_SIZE = 500; const DEFAULT_MAX_BOUND_PARAMETERS = 999; function parameterBoundedBatchSize( + table: AnyTable, columnsPerRow: number, reservedParameters: number, maxBoundParameters: number | undefined ): number { const budget = maxBoundParameters ?? DEFAULT_MAX_BOUND_PARAMETERS; - return Math.max( - 1, - Math.min( - CREATE_MANY_BATCH_SIZE, - Math.floor(Math.max(1, budget - reservedParameters) / columnsPerRow) - ) - ); + const rowsPerStatement = Math.floor((budget - reservedParameters) / columnsPerRow); + if (rowsPerStatement < 1) { + // Even a single row cannot fit the advertised budget. Clamping to one row + // anyway would silently emit a statement the engine may reject, so fail + // fast and name the numbers instead. + const reservedNote = + reservedParameters > 0 ? ` and the predicate reserves ${reservedParameters} more` : ""; + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: adapter rejects writes that cannot fit the engine's bound-parameter budget + throw new Error( + `[FumaDB Drizzle] Cannot write table "${table.ormName}": one row binds ${columnsPerRow} bound parameters${reservedNote}, which exceeds the ${budget}-parameter budget per statement.` + ); + } + return Math.min(CREATE_MANY_BATCH_SIZE, rowsPerStatement); } function buildWhere( @@ -203,7 +210,8 @@ export function fromDrizzle( _db: unknown, provider: SQLProvider, interactiveTransactions: boolean = true, - maxBoundParameters?: number + maxBoundParameters?: number, + transactionDepth: number = 0 ): AbstractQuery { const [db, drizzleTables] = parseDrizzle(_db); @@ -227,6 +235,38 @@ export function fromDrizzle( throw new Error("[FumaDB Drizzle] Database cannot execute raw transaction statements."); } + // Runs `fn` atomically when the engine allows it. SQLite drivers get raw + // BEGIN/COMMIT on the shared connection — or a SAVEPOINT when this adapter + // instance already lives inside a transaction, so nested use (e.g. a bulk + // upsert splitting into predicate groups inside a caller's transaction) + // does not issue a second BEGIN. Other providers get a driver transaction + // whose handle must be used for every statement inside `fn`. When the + // engine rejects interactive transactions (Cloudflare D1), statements + // auto-commit — that engine constraint is documented on `transaction`. + async function runAtomically(fn: (handle: typeof db) => Promise): Promise { + if (!interactiveTransactions) { + return fn(db); + } + if (provider === "sqlite") { + const savepoint = transactionDepth > 0 ? `fumadb_tx_${transactionDepth}` : undefined; + await executeRaw(savepoint ? `SAVEPOINT ${savepoint}` : "BEGIN"); + try { + const result = await fn(db); + await executeRaw(savepoint ? `RELEASE SAVEPOINT ${savepoint}` : "COMMIT"); + return result; + } catch (e) { + if (savepoint) { + await executeRaw(`ROLLBACK TO SAVEPOINT ${savepoint}`); + await executeRaw(`RELEASE SAVEPOINT ${savepoint}`); + } else { + await executeRaw("ROLLBACK"); + } + throw e; + } + } + return db.transaction((tx) => fn(tx as unknown as typeof db)); + } + function toDrizzle(v: AnyTable): TableType { const out = drizzleTables[v.names.drizzle]; if (out) return out; @@ -375,6 +415,7 @@ export function fromDrizzle( const whereParameters = v.where ? countConditionParameters(v.where) : 0; const columnsPerRow = values.length > 0 ? Math.max(1, Object.keys(values[0]!).length) : 1; const batchSize = parameterBoundedBatchSize( + table, columnsPerRow, whereParameters, maxBoundParameters, @@ -387,24 +428,33 @@ export function fromDrizzle( ]), ); - const statements: unknown[] = []; - for (let i = 0; i < values.length; i += batchSize) { - const batch = values.slice(i, i + batchSize); - const insert = db.insert(drizzleTable).values(batch) as unknown as { - onConflictDoUpdate: (input: { - readonly target: typeof target; - readonly set: typeof set; - readonly where?: typeof where; - }) => unknown; - }; - statements.push( - insert.onConflictDoUpdate({ - target, - set, - ...(where === undefined ? {} : { where }), - }), - ); - } + const buildStatements = (handle: typeof db): unknown[] => { + const statements: unknown[] = []; + for (let i = 0; i < values.length; i += batchSize) { + const batch = values.slice(i, i + batchSize); + const insert = handle.insert(drizzleTable).values(batch) as unknown as { + onConflictDoUpdate: (input: { + readonly target: typeof target; + readonly set: typeof set; + readonly where?: typeof where; + }) => unknown; + }; + statements.push( + insert.onConflictDoUpdate({ + target, + set, + ...(where === undefined ? {} : { where }), + }), + ); + } + return statements; + }; + const executeStatements = async (handle: typeof db) => { + for (const statement of buildStatements(handle)) { + await statement; + } + }; + const statementCount = Math.ceil(values.length / batchSize); // D1 rejects interactive transactions but its native batch API executes // prepared statements as one transaction. Drizzle exposes that API on @@ -413,14 +463,19 @@ export function fromDrizzle( const nativeBatch = db as unknown as { readonly batch?: (statements: readonly unknown[]) => Promise; }; - if (!interactiveTransactions && statements.length > 1 && nativeBatch.batch) { - await nativeBatch.batch(statements); + if (!interactiveTransactions && statementCount > 1 && nativeBatch.batch) { + await nativeBatch.batch(buildStatements(db)); return; } - for (const statement of statements) { - await statement; + if (statementCount === 1) { + await executeStatements(db); + return; } + + // One logical upsert split into several parameter-bounded statements + // stays atomic: run them inside one transaction. + await runAtomically(executeStatements); }, async findMany(table, v) { return ( @@ -472,36 +527,48 @@ export function fromDrizzle( }, async createMany(table, values) { + if (values.length === 0) return []; const idField = table.getIdColumn().names.drizzle; const drizzleTable = toDrizzle(table); values = values.map((v) => mapValues(v, table)); - const columnsPerRow = values.length > 0 ? Math.max(1, Object.keys(values[0]!).length) : 1; - const batchSize = parameterBoundedBatchSize(columnsPerRow, 0, maxBoundParameters); + const columnsPerRow = Math.max(1, Object.keys(values[0]!).length); + const batchSize = parameterBoundedBatchSize(table, columnsPerRow, 0, maxBoundParameters); const batches: (typeof values)[] = []; for (let i = 0; i < values.length; i += batchSize) { batches.push(values.slice(i, i + batchSize)); } - if (provider === "sqlite" || provider === "postgresql") { - const out: { _id: unknown }[] = []; + const insertBatches = async (handle: typeof db): Promise<{ _id: unknown }[]> => { + if (provider === "sqlite" || provider === "postgresql") { + const out: { _id: unknown }[] = []; + for (const batch of batches) { + out.push( + ...(await (handle as unknown as P_DBType) + .insert(drizzleTable as unknown as P_TableType) + .values(batch) + .returning({ + _id: (drizzleTable as unknown as P_TableType)[idField], + })), + ); + } + return out; + } + + const results: Record[] = []; for (const batch of batches) { - out.push( - ...(await (db as unknown as P_DBType) - .insert(drizzleTable as unknown as P_TableType) - .values(batch) - .returning({ - _id: (drizzleTable as unknown as P_TableType)[idField], - })), - ); + results.push(...(await handle.insert(drizzleTable).values(batch).$returningId())); } - return out; - } + return results.map((result) => ({ _id: result[idField] })); + }; - const results: Record[] = []; - for (const batch of batches) { - results.push(...(await db.insert(drizzleTable).values(batch).$returningId())); - } - return results.map((result) => ({ _id: result[idField] })); + if (batches.length === 1) return insertBatches(db); + // One logical insert split into parameter-bounded statements must stay + // atomic: a later batch's constraint failure rolls back the earlier + // batches. (Engines without interactive transactions auto-commit each + // statement; `createMany` needs the inserted ids back, which the native + // batch API's result shape does not guarantee across drivers, so those + // engines keep sequential statements — their documented constraint.) + return runAtomically(insertBatches); }, async deleteMany(table, v) { @@ -521,23 +588,44 @@ export function fromDrizzle( // each statement auto-commits, so there is no atomic rollback (the // engine's constraint, not ours). libSQL/Postgres keep real transactions. if (!interactiveTransactions) { - return run(fromDrizzle(schema, _db, provider, interactiveTransactions, maxBoundParameters)); + return run( + fromDrizzle( + schema, + _db, + provider, + interactiveTransactions, + maxBoundParameters, + transactionDepth + ) + ); } if (provider === "sqlite") { - await executeRaw("BEGIN"); - try { - const result = await run(fromDrizzle(schema, _db, provider, interactiveTransactions, maxBoundParameters)); - await executeRaw("COMMIT"); - return result; - } catch (e) { - await executeRaw("ROLLBACK"); - throw e; - } + return runAtomically(() => + run( + fromDrizzle( + schema, + _db, + provider, + interactiveTransactions, + maxBoundParameters, + transactionDepth + 1 + ) + ) + ); } return db.transaction((tx) => - run(fromDrizzle(schema, tx, provider, interactiveTransactions, maxBoundParameters)) + run( + fromDrizzle( + schema, + tx, + provider, + interactiveTransactions, + maxBoundParameters, + transactionDepth + 1 + ) + ) ); }, }); diff --git a/packages/core/fumadb/src/query/orm/index.ts b/packages/core/fumadb/src/query/orm/index.ts index 0796e4c848..1136b5e318 100644 --- a/packages/core/fumadb/src/query/orm/index.ts +++ b/packages/core/fumadb/src/query/orm/index.ts @@ -244,7 +244,7 @@ const applyUpdatePolicies = async ( // same serialized form). Comparing structurally is unambiguous; a false // negative only splits a group and never merges rows under the wrong // predicate. -const predicateValuesEqual = (a: unknown, b: unknown): boolean => { +export const predicateValuesEqual = (a: unknown, b: unknown): boolean => { if (Object.is(a, b)) return true; if (a instanceof Date || b instanceof Date) { return a instanceof Date && b instanceof Date && a.getTime() === b.getTime(); @@ -258,12 +258,16 @@ const predicateValuesEqual = (a: unknown, b: unknown): boolean => { ); } if (Array.isArray(a) || Array.isArray(b)) { - return ( - Array.isArray(a) && - Array.isArray(b) && - a.length === b.length && - a.every((item, index) => predicateValuesEqual(item, b[index])) - ); + if (!Array.isArray(a) || !Array.isArray(b) || a.length !== b.length) return false; + // Compare by index instead of `.every`, which skips holes in sparse + // arrays — `Array(1)` would otherwise equal `[123]` and merge rows under + // the wrong predicate group. A hole and a present element are unequal. + for (let index = 0; index < a.length; index += 1) { + const aHas = index in a; + if (aHas !== index in b) return false; + if (aHas && !predicateValuesEqual(a[index], b[index])) return false; + } + return true; } if (isRecord(a) && isRecord(b)) { const aKeys = Object.keys(a); @@ -500,14 +504,32 @@ export function toORM( groups.push({ where: row.where, values: [row.value] }); } } - for (const group of groups) { - await internal.upsertMany(table, { - target: targetColumns, - update: updateColumns, - values: group.values, - where: group.where, - }); + const runGroups = async (adapter: ORMAdapter): Promise => { + if (!adapter.upsertMany) { + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: a transaction adapter must mirror the base adapter's upsertMany support + throw new Error("[FumaDB] Transaction adapter does not support upsertMany."); + } + for (const group of groups) { + await adapter.upsertMany(table, { + target: targetColumns, + update: updateColumns, + values: group.values, + where: group.where, + }); + } + }; + + // A single group is one adapter call, which is as atomic as the + // engine allows. Multiple predicate groups are separate adapter + // calls, so run them inside one transaction: a later group's failure + // must not leave earlier groups committed. + if (groups.length === 1) { + await runGroups(internal); + return; } + await internal.transaction(async (transactionInstance) => { + await runGroups(transactionInstance.internal); + }); return; } diff --git a/packages/core/fumadb/src/query/orm/predicate-values-equal.test.ts b/packages/core/fumadb/src/query/orm/predicate-values-equal.test.ts new file mode 100644 index 0000000000..21c5aaff0c --- /dev/null +++ b/packages/core/fumadb/src/query/orm/predicate-values-equal.test.ts @@ -0,0 +1,34 @@ +import { describe, expect, it } from "@effect/vitest"; + +import { predicateValuesEqual } from "./index"; + +describe("predicateValuesEqual", () => { + it("treats a sparse-array hole as unequal to a present element, in both orders", () => { + // oxlint-disable-next-line unicorn/no-new-array -- the sparse array is the case under test + expect(predicateValuesEqual(Array(1), [123])).toBe(false); + // oxlint-disable-next-line unicorn/no-new-array -- the sparse array is the case under test + expect(predicateValuesEqual([123], Array(1))).toBe(false); + }); + + it("treats arrays with matching holes as equal", () => { + // oxlint-disable-next-line unicorn/no-new-array -- the sparse array is the case under test + expect(predicateValuesEqual(Array(2), Array(2))).toBe(true); + }); + + it("keeps structural equality for dense arrays", () => { + expect( + predicateValuesEqual([BigInt(7), new Date(5), Number.NaN], [BigInt(7), new Date(5), Number.NaN]), + ).toBe(true); + expect(predicateValuesEqual([1, 2], [1, 3])).toBe(false); + expect(predicateValuesEqual([1, undefined], [1, undefined])).toBe(true); + }); + + it("keeps the existing Date, Uint8Array, and record handling", () => { + expect(predicateValuesEqual(new Date(5), new Date(5))).toBe(true); + expect(predicateValuesEqual(new Date(5), new Date(6))).toBe(false); + expect(predicateValuesEqual(new Uint8Array([1, 2]), new Uint8Array([1, 2]))).toBe(true); + expect(predicateValuesEqual(new Uint8Array([1, 2]), new Uint8Array([1, 3]))).toBe(false); + expect(predicateValuesEqual({ a: BigInt(1) }, { a: BigInt(1) })).toBe(true); + expect(predicateValuesEqual({ a: 1 }, { a: 1, b: 2 })).toBe(false); + }); +}); diff --git a/packages/core/fumadb/src/query/table-policy.test.ts b/packages/core/fumadb/src/query/table-policy.test.ts index b0c6d8bb2b..225ddd2d5f 100644 --- a/packages/core/fumadb/src/query/table-policy.test.ts +++ b/packages/core/fumadb/src/query/table-policy.test.ts @@ -116,6 +116,18 @@ const quotas = table("policy_quotas", { onUpdate: ({ builder, context }) => builder("region", "=", context.allowedRegionId ?? BigInt(-1)), }); +// The update policy predicate depends on the row itself, so rows with +// different `shard` values compile to different predicates and split one +// upsertMany call into separate predicate groups. +const shards = table("policy_shards", { + id: idColumn("id", "varchar(255)"), + shard: column("shard", "string"), + authorId: column("author_id", "varchar(255)"), +}).policy({ + name: "tenant.shards", + onUpdate: ({ builder, set }) => builder("shard", "=", set.shard ?? ""), +}); + // Ten columns, so a row-count-sized batch would bind ten parameters per row. const wideRows = table("policy_wide_rows", { id: idColumn("id", "varchar(255)"), @@ -137,11 +149,16 @@ const v1 = schema({ posts, comments, quotas, + shards, wideRows, }, relations: { authors: ({ many }) => ({ posts: many("posts"), + shards: many("shards"), + }), + shards: ({ one }) => ({ + author: one("authors", ["authorId", "id"]).foreignKey(), }), posts: ({ one, many }) => ({ author: one("authors", ["authorId", "id"]).foreignKey(), @@ -251,6 +268,16 @@ const useStatementHarness = ( ({ close }) => Effect.promise(close), ); +const useBudgetHarness = ( + maxBoundParameters: number, + run: (orm: TablePolicyQuery) => Promise, +) => + Effect.acquireUseRelease( + Effect.promise(() => makeHarness({ maxBoundParameters })), + ({ orm }) => Effect.promise(() => run(orm)), + ({ close }) => Effect.promise(close), + ); + const useNativeBatchHarness = ( run: (harness: { readonly orm: TablePolicyQuery; @@ -752,6 +779,133 @@ describe("FumaDB table policies", () => { }), ); + it.effect("rolls back earlier predicate groups when a later group fails", () => + useHarness(async (orm) => { + await seedTenants(orm); + const writer = withQueryContext(orm, makeContext(["tenant-a"], "groups")); + + // Two distinct per-row predicates split this call into two internal + // groups. The second group's row references a missing author, so its + // insert violates the foreign key — the first group's rows must roll + // back with it instead of staying committed. + await expect( + writer.upsertMany("shards", { + target: ["id"], + update: ["shard"], + values: [ + { id: "shard-a-1", shard: "a", authorId: "author-a" }, + { id: "shard-b-1", shard: "b", authorId: "missing-author" }, + ], + }), + ).rejects.toThrow(); + + await expect(orm.count("shards")).resolves.toBe(0); + }), + ); + + it.effect("rolls back earlier createMany batches when a later batch fails", () => + useBudgetHarness(8, async (orm) => { + await seedTenants(orm); + const tenantA = withQueryContext(orm, makeContext(["tenant-a"], "tenant-a")); + + // Four columns per row against an 8-parameter budget forces two-row + // batches, so this insert runs as two statements. The last row reuses + // the seeded `post-a-1` id, so the second statement violates the + // primary key — the first statement's rows must roll back with it. + await expect( + tenantA.createMany("posts", [ + { id: "post-a-n1", tenantId: "tenant-a", authorId: "author-a", title: "N1" }, + { id: "post-a-n2", tenantId: "tenant-a", authorId: "author-a", title: "N2" }, + { id: "post-a-n3", tenantId: "tenant-a", authorId: "author-a", title: "N3" }, + { id: "post-a-1", tenantId: "tenant-a", authorId: "author-a", title: "Duplicate" }, + ]), + ).rejects.toThrow(); + + await expect( + tenantA.findMany("posts", { + where: (builder) => builder("id", "starts with", "post-a-n"), + select: ["id"], + }), + ).resolves.toEqual([]); + }), + ); + + it.effect("fails fast when a single row exceeds the bound-parameter budget", () => + useBudgetHarness(8, async (orm) => { + await expect( + orm.createMany("wideRows", [ + { + id: "wide-overflow", + c1: "1", + c2: "2", + c3: "3", + c4: "4", + c5: "5", + c6: "6", + c7: "7", + c8: "8", + c9: "9", + }, + ]), + ).rejects.toThrow( + 'one row binds 10 bound parameters, which exceeds the 8-parameter budget', + ); + + await expect(orm.count("wideRows")).resolves.toBe(0); + }), + ); + + it.effect("fails fast when reserved predicate parameters consume the budget", () => + useBudgetHarness(4, async (orm) => { + const seed = withQueryContext(orm, makeContext(["tenant-a"], "seed")); + await seed.createMany("authors", [{ id: "author-a", tenantId: "tenant-a", name: "Ada" }]); + + const tenantA = withQueryContext(orm, makeContext(["tenant-a"], "tenant-a")); + // A posts row alone fits the 4-parameter budget exactly, but the update + // policy predicate reserves one more bound parameter per statement. + await expect( + tenantA.upsertMany("posts", { + target: ["id"], + update: ["title"], + values: [{ id: "post-a-r1", tenantId: "tenant-a", authorId: "author-a", title: "R1" }], + }), + ).rejects.toThrow("the predicate reserves 1 more"); + + await expect(tenantA.count("posts")).resolves.toBe(0); + }), + ); + + it.effect("applies intra-call conflicts identically on drizzle and memory adapters", () => { + // The second value's unique target conflicts with a row created by the + // first value in the SAME upsertMany call: both adapters must update the + // freshly created row instead of duplicating or dropping it. + const upsertIntraCallConflict = async (orm: TablePolicyQuery) => { + await seedTenants(orm); + const tenantA = withQueryContext(orm, makeContext(["tenant-a"], "tenant-a")); + await tenantA.upsertMany("posts", { + target: ["id"], + update: ["title"], + values: [ + { id: "post-a-dup", tenantId: "tenant-a", authorId: "author-a", title: "first write" }, + { id: "post-a-dup", tenantId: "tenant-a", authorId: "author-a", title: "second write" }, + ], + }); + return tenantA.findMany("posts", { + where: (builder) => builder("id", "=", "post-a-dup"), + select: ["id", "title"], + }); + }; + + const expected = [{ id: "post-a-dup", title: "second write" }]; + + return useHarness(async (orm) => { + await expect(upsertIntraCallConflict(orm)).resolves.toEqual(expected); + + const memoryOrm = tablePolicyDB.client(memoryAdapter()).orm("1.0.0"); + await expect(upsertIntraCallConflict(memoryOrm)).resolves.toEqual(expected); + }); + }); + it.effect("fails closed when a query wrapper does not forward context rebinding", () => useHarness(async (orm) => { const wrapped = { ...orm }; From 357e1d94774b2a464c6cf3edf91cccb314665c97 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Fri, 28 Aug 2026 12:45:49 -0700 Subject: [PATCH 10/10] Quote excluded columns, serialize SQLite transaction scopes, and transact generic bulk upserts --- .../core/fumadb/src/adapters/drizzle/query.ts | 124 +++++++++++++----- .../drizzle/upsert-many-generic.test.ts | 120 +++++++++++++++++ .../fumadb/src/query/table-policy.test.ts | 79 +++++++++++ 3 files changed, 288 insertions(+), 35 deletions(-) create mode 100644 packages/core/fumadb/src/adapters/drizzle/upsert-many-generic.test.ts diff --git a/packages/core/fumadb/src/adapters/drizzle/query.ts b/packages/core/fumadb/src/adapters/drizzle/query.ts index 6c86b82d59..21c7bc1e95 100644 --- a/packages/core/fumadb/src/adapters/drizzle/query.ts +++ b/packages/core/fumadb/src/adapters/drizzle/query.ts @@ -51,6 +51,48 @@ function parameterBoundedBatchSize( return Math.min(CREATE_MANY_BATCH_SIZE, rowsPerStatement); } +// SQLite transaction scopes run as raw BEGIN/SAVEPOINT statements on one +// shared connection, so two scopes started concurrently would interleave +// their control statements: two top-level BEGINs collide, and sibling +// savepoints at one nesting depth reuse each other's names — one sibling's +// ROLLBACK TO can undo work the other already released. Serialize scope +// execution per connection handle and nesting depth: same-depth scopes run +// one after another, while a parent scope at depth N can still open its +// child scope at depth N + 1 without deadlocking. SQLite is single-writer, +// so this serialization costs no real concurrency. +const transactionScopeQueues = new WeakMap>>(); + +async function runSerializedScope( + handle: object, + depth: number, + fn: () => Promise, +): Promise { + let queues = transactionScopeQueues.get(handle); + if (!queues) { + queues = new Map(); + transactionScopeQueues.set(handle, queues); + } + const previous = queues.get(depth) ?? Promise.resolve(); + let release!: () => void; + queues.set( + depth, + new Promise((resolve) => { + release = resolve; + }), + ); + await previous; + // oxlint-disable-next-line executor/no-try-catch-or-throw -- scope release must survive the wrapped failure + try { + return await fn(); + } finally { + release(); + } +} + +// Savepoint names are invocation-unique as defense in depth: even if two +// scopes ever interleave, a ROLLBACK TO can only target its own savepoint. +let savepointSequence = 0; + function buildWhere( toDrizzle: (col: AnyColumn) => ColumnType, condition: Condition @@ -248,21 +290,24 @@ export function fromDrizzle( return fn(db); } if (provider === "sqlite") { - const savepoint = transactionDepth > 0 ? `fumadb_tx_${transactionDepth}` : undefined; - await executeRaw(savepoint ? `SAVEPOINT ${savepoint}` : "BEGIN"); - try { - const result = await fn(db); - await executeRaw(savepoint ? `RELEASE SAVEPOINT ${savepoint}` : "COMMIT"); - return result; - } catch (e) { - if (savepoint) { - await executeRaw(`ROLLBACK TO SAVEPOINT ${savepoint}`); - await executeRaw(`RELEASE SAVEPOINT ${savepoint}`); - } else { - await executeRaw("ROLLBACK"); + return runSerializedScope(db, transactionDepth, async () => { + savepointSequence += 1; + const savepoint = transactionDepth > 0 ? `fumadb_tx_${savepointSequence}` : undefined; + await executeRaw(savepoint ? `SAVEPOINT ${savepoint}` : "BEGIN"); + try { + const result = await fn(db); + await executeRaw(savepoint ? `RELEASE SAVEPOINT ${savepoint}` : "COMMIT"); + return result; + } catch (e) { + if (savepoint) { + await executeRaw(`ROLLBACK TO SAVEPOINT ${savepoint}`); + await executeRaw(`RELEASE SAVEPOINT ${savepoint}`); + } else { + await executeRaw("ROLLBACK"); + } + throw e; } - throw e; - } + }); } return db.transaction((tx) => fn(tx as unknown as typeof db)); } @@ -386,26 +431,32 @@ export function fromDrizzle( throw new Error("[FumaDB] upsertMany requires at least one update column."); } if (provider !== "sqlite" && provider !== "postgresql") { - for (const value of v.values) { - const targetCondition: Condition = { - type: ConditionType.And, - items: v.target.map((column) => ({ - type: ConditionType.Compare, - a: column, - operator: "=", - b: value[column.ormName], - })), - }; - await this.upsert(table, { - where: v.where - ? { type: ConditionType.And, items: [targetCondition, v.where] } - : targetCondition, - update: Object.fromEntries( - v.update.map((column) => [column.ormName, value[column.ormName]]), - ), - create: value, - }); - } + // This path issues several statements per row, so a failure halfway + // through must not leave a prefix of the rows committed. These + // providers have native driver transactions; `transaction` routes + // every per-row statement through one of them. + await this.transaction(async (scoped) => { + for (const value of v.values) { + const targetCondition: Condition = { + type: ConditionType.And, + items: v.target.map((column) => ({ + type: ConditionType.Compare, + a: column, + operator: "=", + b: value[column.ormName], + })), + }; + await scoped.internal.upsert(table, { + where: v.where + ? { type: ConditionType.And, items: [targetCondition, v.where] } + : targetCondition, + update: Object.fromEntries( + v.update.map((column) => [column.ormName, value[column.ormName]]), + ), + create: value, + }); + } + }); return; } @@ -424,7 +475,10 @@ export function fromDrizzle( const set = Object.fromEntries( v.update.map((column) => [ column.names.drizzle, - Drizzle.sql.raw(`excluded.${column.names.sql}`), + // `sql.identifier` quotes the physical column name; a raw + // interpolation would emit e.g. `excluded.display-name` unquoted, + // which the engine parses as an expression. + Drizzle.sql`excluded.${Drizzle.sql.identifier(column.names.sql)}`, ]), ); diff --git a/packages/core/fumadb/src/adapters/drizzle/upsert-many-generic.test.ts b/packages/core/fumadb/src/adapters/drizzle/upsert-many-generic.test.ts new file mode 100644 index 0000000000..333d4a2f7d --- /dev/null +++ b/packages/core/fumadb/src/adapters/drizzle/upsert-many-generic.test.ts @@ -0,0 +1,120 @@ +import { expect, test } from "@effect/vitest"; + +import { column, idColumn, schema, table } from "../../schema"; +import { fromDrizzle } from "./query"; + +// The generic bulk-upsert path (providers without ON CONFLICT support, e.g. +// MySQL) issues several statements per row. The unit harness has no MySQL +// server, so a recording fake of the Drizzle handle asserts the transaction +// routing instead: every per-row statement must run on the driver +// transaction's handle, and a failing row must roll the earlier rows back. +const v1 = schema({ + version: "1.0.0", + tables: { + rows: table("generic_rows", { + id: idColumn("id", "varchar(255)"), + value: column("value", "string"), + }), + }, +}); + +interface FakeDbOptions { + // A row whose `value` matches makes its INSERT fail. + readonly failOnValue?: string; +} + +const createFakeMysqlDb = (options: FakeDbOptions = {}) => { + const committed: Record[] = []; + const events: string[] = []; + const fakeTable = { id: { name: "id" }, value: { name: "value" } }; + + const makeHandle = (label: string, sink: Record[]) => ({ + _: { fullSchema: { rows: fakeTable } }, + select: () => { + const builder = { + from: () => builder, + limit: () => builder, + where: () => builder, + execute: async (): Promise[]> => [], + }; + return builder; + }, + insert: () => ({ + values: (rows: Record[]) => ({ + $returningId: async () => { + events.push(`${label}:insert`); + for (const row of rows) { + if (options.failOnValue !== undefined && row["value"] === options.failOnValue) { + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- test forces a row's statement to fail + throw new Error("statement failure"); + } + sink.push(row); + } + return rows.map((row) => ({ id: row["id"] })); + }, + }), + }), + transaction: async (callback: (tx: unknown) => Promise): Promise => { + events.push("transaction:begin"); + const staged: Record[] = []; + const tx = makeHandle("tx", staged); + // oxlint-disable-next-line executor/no-try-catch-or-throw -- fake driver mirrors commit/rollback + try { + const result = await callback(tx); + committed.push(...staged); + events.push("transaction:commit"); + return result; + } catch (error) { + events.push("transaction:rollback"); + throw error; + } + }, + }); + + // Statements on the root handle auto-commit, exactly like a driver outside + // an explicit transaction. + return { db: makeHandle("root", committed), committed, events }; +}; + +const upsertAll = async (db: unknown, values: readonly { id: string; value: string }[]) => { + const orm = fromDrizzle(v1, db, "mysql"); + const rows = v1.tables.rows; + expect(orm.internal.upsertMany).toBeDefined(); + await orm.internal.upsertMany?.(rows, { + target: [rows.columns.id], + update: [rows.columns.value], + values: values.map((value) => ({ ...value })), + }); +}; + +test("generic bulk upsert runs every row inside one driver transaction", async () => { + const { db, committed, events } = createFakeMysqlDb(); + + await upsertAll(db, [ + { id: "r1", value: "a" }, + { id: "r2", value: "b" }, + { id: "r3", value: "c" }, + ]); + + expect(committed.map((row) => row["id"])).toEqual(["r1", "r2", "r3"]); + expect(events).toContain("transaction:begin"); + expect(events).toContain("transaction:commit"); + // No row statement may run on the auto-committing root handle. + expect(events).not.toContain("root:insert"); +}); + +test("generic bulk upsert rolls earlier rows back when a later row fails", async () => { + const { db, committed, events } = createFakeMysqlDb({ failOnValue: "poison" }); + + await expect( + upsertAll(db, [ + { id: "r1", value: "a" }, + { id: "r2", value: "b" }, + { id: "r3", value: "poison" }, + ]), + ).rejects.toThrow("statement failure"); + + // A single logical bulk upsert must not leave a partial prefix committed. + expect(committed).toEqual([]); + expect(events).toContain("transaction:rollback"); +}); diff --git a/packages/core/fumadb/src/query/table-policy.test.ts b/packages/core/fumadb/src/query/table-policy.test.ts index 225ddd2d5f..df3d39c9a7 100644 --- a/packages/core/fumadb/src/query/table-policy.test.ts +++ b/packages/core/fumadb/src/query/table-policy.test.ts @@ -142,6 +142,13 @@ const wideRows = table("policy_wide_rows", { c9: column("c9", "string"), }); +// The physical column name needs identifier quoting; the bulk upsert conflict +// SET clause references it as `excluded.` and must quote it. +const quotedNames = table("policy_quoted_names", { + id: idColumn("id", "varchar(255)"), + displayName: column("display-name", "string"), +}); + const v1 = schema({ version: "1.0.0", tables: { @@ -151,6 +158,7 @@ const v1 = schema({ quotas, shards, wideRows, + quotedNames, }, relations: { authors: ({ many }) => ({ @@ -654,6 +662,77 @@ describe("FumaDB table policies", () => { }), ); + it.effect("bulk upserts a column whose physical name needs identifier quoting", () => + useHarness(async (orm) => { + await orm.createMany("quotedNames", [{ id: "row-1", displayName: "before" }]); + + // The conflict SET clause references the update column through the + // `excluded` pseudo-table; an unquoted physical name like + // `display-name` is a syntax error there. + await orm.upsertMany("quotedNames", { + target: ["id"], + update: ["displayName"], + values: [ + { id: "row-1", displayName: "after" }, + { id: "row-2", displayName: "created" }, + ], + }); + + await expect( + orm.findMany("quotedNames", { + select: ["id", "displayName"], + orderBy: ["id", "asc"], + }), + ).resolves.toEqual([ + { id: "row-1", displayName: "after" }, + { id: "row-2", displayName: "created" }, + ]); + }), + ); + + it.effect("keeps a successful sibling nested transaction when the other rolls back", () => + useHarness(async (orm) => { + const writer = withQueryContext(orm, makeContext(["tenant-a"], "siblings")); + + await writer.transaction(async (tx) => { + // Two sibling savepoint scopes started concurrently on the one shared + // SQLite connection: the failing sibling's ROLLBACK TO must not undo + // work the successful sibling already released. + const failing = tx.transaction(async (inner) => { + await inner.create("authors", { + id: "author-rolled-back", + tenantId: "tenant-a", + name: "Rolled Back", + }); + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- test forces one sibling to roll back + throw new Error("sibling failure"); + }); + const succeeding = tx.transaction(async (inner) => { + await inner.create("authors", { + id: "author-committed", + tenantId: "tenant-a", + name: "Committed", + }); + }); + + const [failed, succeeded] = await Promise.allSettled([failing, succeeding]); + expect(failed.status).toBe("rejected"); + expect(succeeded.status).toBe("fulfilled"); + }); + + await expect( + writer.findMany("authors", { + select: ["id"], + where: (builder) => + builder.or( + builder("id", "=", "author-rolled-back"), + builder("id", "=", "author-committed"), + ), + }), + ).resolves.toEqual([{ id: "author-committed" }]); + }), + ); + it.effect("bulk upserts through a policy predicate that contains a bigint", () => useHarness(async (orm) => { const region = withQueryContext(orm, {