Skip to content

Prevent provider migration row ID collisions #1416

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
morluto wants to merge 1 commit into
base: main
Choose a base branch
from
+143 −36
Open

Prevent provider migration row ID collisions #1416

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/prevent-provider-migration-row-loss.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"executor": patch
---

Prevent provider service migration row loss caused by generated ID conflicts.
100 changes: 96 additions & 4 deletions packages/plugins/provider-service-split/src/sqlite.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import {
collectTables,
DataMigrationError,
runSqliteDataMigrations,
sha256Hex,
type SqliteDataMigrationClient,
} from "@executor-js/sdk";
import { createSqliteTestFumaDb } from "@executor-js/sdk/testing";
Expand Down Expand Up @@ -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,
Expand All @@ -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" }),
Expand All @@ -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",
"",
Expand Down Expand Up @@ -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() }));
Expand Down
74 changes: 42 additions & 32 deletions packages/plugins/provider-service-split/src/sqlite.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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,
Expand Down Expand Up @@ -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<string> =>
sha256Hex(JSON.stringify(parts)).pipe(Effect.map((hash) => `service_split_${hash}`));

const tableExists = (client: SqliteDataMigrationClient, table: string) =>
execute(client, {
Expand Down Expand Up @@ -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],
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -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
Expand Down Expand Up @@ -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,
Expand All @@ -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
Expand All @@ -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,
Expand Down