From 75e1e80c3545a39c2e11fd2500024f705606d20b Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Wed, 15 Jul 2026 06:53:43 +0000 Subject: [PATCH] Prevent provider migration row ID collisions --- .../prevent-provider-migration-row-loss.md | 5 + .../provider-service-split/src/sqlite.test.ts | 100 +++++++++++++++++- .../provider-service-split/src/sqlite.ts | 74 +++++++------ 3 files changed, 143 insertions(+), 36 deletions(-) create mode 100644 .changeset/prevent-provider-migration-row-loss.md diff --git a/.changeset/prevent-provider-migration-row-loss.md b/.changeset/prevent-provider-migration-row-loss.md new file mode 100644 index 000000000..c6a48e7af --- /dev/null +++ b/.changeset/prevent-provider-migration-row-loss.md @@ -0,0 +1,5 @@ +--- +"executor": patch +--- + +Prevent provider service migration row loss caused by generated ID conflicts. diff --git a/packages/plugins/provider-service-split/src/sqlite.test.ts b/packages/plugins/provider-service-split/src/sqlite.test.ts index fca87a64c..743ca4a0c 100644 --- a/packages/plugins/provider-service-split/src/sqlite.test.ts +++ b/packages/plugins/provider-service-split/src/sqlite.test.ts @@ -4,6 +4,7 @@ import { collectTables, DataMigrationError, runSqliteDataMigrations, + sha256Hex, type SqliteDataMigrationClient, } from "@executor-js/sdk"; import { createSqliteTestFumaDb } from "@executor-js/sdk/testing"; @@ -47,7 +48,12 @@ const insertIntegration = ( ], }); -const insertConnection = (client: SqliteDataMigrationClient, tenant = "org_1") => +const insertConnection = ( + client: SqliteDataMigrationClient, + tenant = "org_1", + name = "main", + options: { readonly integration?: string; readonly rowId?: string } = {}, +) => client.execute({ sql: `INSERT INTO connection (integration, name, template, provider, item_ids, identity_label, description, @@ -56,8 +62,8 @@ const insertConnection = (client: SqliteDataMigrationClient, tenant = "org_1") = row_id, tenant, owner, subject) VALUES (?, ?, ?, ?, ?, ?, NULL, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, args: [ - "google", - "main", + options.integration ?? "google", + name, "googleOAuth", "vault", JSON.stringify({ token: "access_item", refresh: "refresh_item" }), @@ -72,7 +78,7 @@ const insertConnection = (client: SqliteDataMigrationClient, tenant = "org_1") = JSON.stringify({ token: "metadata" }), now, now, - `conn_google_${tenant}`, + options.rowId ?? `conn_google_${tenant}${name === "main" ? "" : `_${name}`}`, tenant, "org", "", @@ -288,6 +294,92 @@ describe("providerServiceSplitDataMigration", () => { }), ); + it.effect("preserves connections whose legacy migration row IDs collide", () => + Effect.gen(function* () { + const db = yield* Effect.promise(() => createSqliteTestFumaDb({ tables: collectTables() })); + const client = db.client; + + yield* Effect.promise(() => + insertIntegration(client, { + rowId: "google_row", + tenant: "org_1", + slug: "google", + pluginId: "google", + config: { + googleDiscoveryUrls: ["https://www.googleapis.com/discovery/v1/apis/calendar/v3/rest"], + specHash: "mono-hash", + }, + }), + ); + yield* Effect.promise(() => insertConnection(client, "org_1", "ypw3e5cb")); + yield* Effect.promise(() => insertConnection(client, "org_1", "al4jix0b")); + yield* Effect.promise(() => insertBlob(client, "spec/mono-hash")); + yield* Effect.promise(() => insertBlob(client, "defs/mono-hash")); + + expect(yield* runSqliteProviderServiceSplitMigration(client)).toBe(1); + + const connections = yield* Effect.promise(() => + client.execute("SELECT integration, name FROM connection ORDER BY name"), + ); + expect(connections.rows).toEqual([ + { integration: "google_calendar", name: "al4jix0b" }, + { integration: "google_calendar", name: "ypw3e5cb" }, + ]); + + yield* Effect.promise(() => db.close()); + }), + ); + + it.effect("rolls back instead of ignoring an unrelated row ID conflict", () => + Effect.gen(function* () { + const db = yield* Effect.promise(() => createSqliteTestFumaDb({ tables: collectTables() })); + const client = db.client; + const connectionRowId = `service_split_${yield* sha256Hex( + JSON.stringify(["connection", "org_1", "org", "", "google_calendar", "main"]), + )}`; + + yield* Effect.promise(() => + insertIntegration(client, { + rowId: "google_row", + tenant: "org_1", + slug: "google", + pluginId: "google", + config: { + googleDiscoveryUrls: ["https://www.googleapis.com/discovery/v1/apis/calendar/v3/rest"], + specHash: "mono-hash", + }, + }), + ); + yield* Effect.promise(() => insertConnection(client)); + yield* Effect.promise(() => + insertConnection(client, "org_1", "existing", { + integration: "google_calendar", + rowId: connectionRowId, + }), + ); + yield* Effect.promise(() => insertBlob(client, "spec/mono-hash")); + yield* Effect.promise(() => insertBlob(client, "defs/mono-hash")); + + const failure = yield* Effect.flip(runSqliteProviderServiceSplitMigration(client)); + expect(failure).toBeInstanceOf(DataMigrationError); + + const connections = yield* Effect.promise(() => + client.execute("SELECT integration, name FROM connection ORDER BY integration, name"), + ); + expect(connections.rows).toEqual([ + { integration: "google", name: "main" }, + { integration: "google_calendar", name: "existing" }, + ]); + + const integrations = yield* Effect.promise(() => + client.execute("SELECT slug, plugin_id FROM integration ORDER BY slug"), + ); + expect(integrations.rows).toEqual([{ slug: "google", plugin_id: "google" }]); + + yield* Effect.promise(() => db.close()); + }), + ); + it.effect("skips a specHash-less org intact and still migrates healthy orgs", () => Effect.gen(function* () { const db = yield* Effect.promise(() => createSqliteTestFumaDb({ tables: collectTables() })); diff --git a/packages/plugins/provider-service-split/src/sqlite.ts b/packages/plugins/provider-service-split/src/sqlite.ts index 37486b219..ce8c91483 100644 --- a/packages/plugins/provider-service-split/src/sqlite.ts +++ b/packages/plugins/provider-service-split/src/sqlite.ts @@ -1,6 +1,6 @@ /* oxlint-disable executor/no-try-catch-or-throw, executor/no-error-constructor, executor/no-json-parse -- boundary: one-shot provider service split data migration preserves legacy SQLite row coercion and typed DataMigrationError wrapping */ import { Effect } from "effect"; -import { DataMigrationError, type SqliteDataMigrationClient } from "@executor-js/sdk"; +import { DataMigrationError, sha256Hex, type SqliteDataMigrationClient } from "@executor-js/sdk"; import { operationStorageKey, @@ -53,14 +53,8 @@ const scrubJson = (value: unknown): unknown => { return value; }; -const stableId = (...parts: readonly string[]): string => { - const text = parts.join("\u0000"); - let hash = 0; - for (let index = 0; index < text.length; index += 1) { - hash = (hash * 31 + text.charCodeAt(index)) >>> 0; - } - return `service_split_${hash.toString(36)}`; -}; +const stableId = (...parts: readonly string[]): Effect.Effect => + sha256Hex(JSON.stringify(parts)).pipe(Effect.map((hash) => `service_split_${hash}`)); const tableExists = (client: SqliteDataMigrationClient, table: string) => execute(client, { @@ -293,6 +287,7 @@ const applyOrg = ( const now = Date.now(); for (const copy of org.blobCopies.filter((item) => item.backend === "database")) { + const rowId = yield* stableId("blob", org.tenant, copy.targetNamespace, copy.key); const source = yield* execute(client, { sql: "SELECT value FROM blob WHERE namespace = ? AND key = ? LIMIT 1", args: [copy.sourceNamespace, copy.key], @@ -305,24 +300,27 @@ const applyOrg = ( }); } yield* execute(client, { - sql: `INSERT OR IGNORE INTO blob (namespace, key, value, row_id, id) - VALUES (?, ?, ?, ?, ?)`, + sql: `INSERT INTO blob (namespace, key, value, row_id, id) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(id) DO NOTHING`, args: [ copy.targetNamespace, copy.key, sourceRow.value, - stableId("blob", org.tenant, copy.targetNamespace, copy.key), + rowId, JSON.stringify([copy.targetNamespace, copy.key]), ], }); } for (const row of org.integrations.filter((item) => item.action === "create")) { + const rowId = yield* stableId("integration", org.tenant, row.target.slug); yield* execute(client, { - sql: `INSERT OR IGNORE INTO integration + sql: `INSERT INTO integration (slug, plugin_id, name, description, config, health_check, config_revised_at, can_remove, can_refresh, created_at, updated_at, row_id, tenant) - VALUES (?, ?, ?, ?, ?, ?, NULL, 1, 1, ?, ?, ?, ?)`, + VALUES (?, ?, ?, ?, ?, ?, NULL, 1, 1, ?, ?, ?, ?) + ON CONFLICT(tenant, slug) DO NOTHING`, args: [ row.target.slug, row.target.pluginId, @@ -332,15 +330,23 @@ const applyOrg = ( row.healthCheck ? JSON.stringify(scrubJson(row.healthCheck)) : null, now, now, - stableId("integration", org.tenant, row.target.slug), + rowId, org.tenant, ], }); } for (const row of org.connections.filter((item) => item.action === "clone")) { + const rowId = yield* stableId( + "connection", + org.tenant, + row.source.owner, + row.source.subject, + row.targetIntegration, + row.source.name, + ); yield* execute(client, { - sql: `INSERT OR IGNORE INTO connection + sql: `INSERT INTO connection (integration, name, template, provider, item_ids, identity_label, description, last_health, tools_synced_at, oauth_client, oauth_client_owner, refresh_item_id, expires_at, oauth_scope, oauth_token_url, provider_state, created_at, updated_at, @@ -349,18 +355,12 @@ const applyOrg = ( NULL, NULL, oauth_client, oauth_client_owner, refresh_item_id, expires_at, oauth_scope, oauth_token_url, provider_state, created_at, ?, ?, tenant, owner, subject FROM connection - WHERE tenant = ? AND owner = ? AND subject = ? AND integration = ? AND name = ?`, + WHERE tenant = ? AND owner = ? AND subject = ? AND integration = ? AND name = ? + ON CONFLICT(tenant, owner, subject, integration, name) DO NOTHING`, args: [ row.targetIntegration, now, - stableId( - "connection", - org.tenant, - row.source.owner, - row.source.subject, - row.targetIntegration, - row.source.name, - ), + rowId, org.tenant, row.source.owner, row.source.subject, @@ -371,8 +371,15 @@ const applyOrg = ( } for (const integration of org.integrations) { + const toolRowIdPrefix = yield* stableId("tool", org.tenant, integration.target.slug); for (const contribution of integration.sourceContributions) { for (const toolName of contribution.operationToolNames) { + const operationRowId = yield* stableId( + "operation", + org.tenant, + integration.target.slug, + toolName, + ); const operation = yield* execute(client, { sql: `SELECT data, created_at, updated_at FROM plugin_storage @@ -424,23 +431,24 @@ const applyOrg = ( ), source.created_at, now, - stableId("operation", org.tenant, integration.target.slug, toolName), + operationRowId, org.tenant, ], }); yield* execute(client, { - sql: `INSERT OR IGNORE INTO tool + sql: `INSERT INTO tool (integration, connection, plugin_id, name, description, input_schema, output_schema, annotations, created_at, updated_at, row_id, tenant, owner, subject) SELECT ?, connection, ?, name, description, input_schema, output_schema, annotations, created_at, ?, ? || '_' || row_id, tenant, owner, subject FROM tool - WHERE tenant = ? AND integration = ? AND name = ?`, + WHERE tenant = ? AND integration = ? AND name = ? + ON CONFLICT(tenant, owner, subject, integration, connection, name) DO NOTHING`, args: [ integration.target.slug, integration.target.pluginId, now, - stableId("tool", org.tenant, integration.target.slug), + toolRowIdPrefix, org.tenant, contribution.source.slug, toolName, @@ -456,10 +464,12 @@ const applyOrg = ( args: [org.tenant, policy.policy.owner, policy.policy.subject, policy.policy.id], }); for (const [index, pattern] of policy.afterPatterns.entries()) { + const rowId = yield* stableId("policy", org.tenant, policy.policy.id, pattern); yield* execute(client, { - sql: `INSERT OR IGNORE INTO tool_policy + sql: `INSERT INTO tool_policy (id, pattern, action, position, created_at, updated_at, row_id, tenant, owner, subject) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(tenant, owner, subject, id) DO NOTHING`, args: [ policy.afterPatterns.length === 1 ? policy.policy.id @@ -469,7 +479,7 @@ const applyOrg = ( policy.policy.position, now, now, - stableId("policy", org.tenant, policy.policy.id, pattern), + rowId, org.tenant, policy.policy.owner, policy.policy.subject,