diff --git a/.changeset/openapi-large-catalog-memory.md b/.changeset/openapi-large-catalog-memory.md new file mode 100644 index 0000000000..09ce7d772e --- /dev/null +++ b/.changeset/openapi-large-catalog-memory.md @@ -0,0 +1,6 @@ +--- +"@executor-js/sdk": patch +"@executor-js/plugin-openapi": patch +--- + +Reduce large OpenAPI import memory by compiling operation bindings in chunks and filtering integration storage prefixes in SQL. diff --git a/e2e/scenarios/openapi-large-catalog.test.ts b/e2e/scenarios/openapi-large-catalog.test.ts new file mode 100644 index 0000000000..9f84a9d2ba --- /dev/null +++ b/e2e/scenarios/openapi-large-catalog.test.ts @@ -0,0 +1,163 @@ +import { randomUUID } from "node:crypto"; +import { assert, expect } from "@effect/vitest"; +import { Effect } from "effect"; +import { composePluginApi } from "@executor-js/api/server"; +import { openApiHttpPlugin } from "@executor-js/plugin-openapi/api"; +import { AuthTemplateSlug, ConnectionName, IntegrationSlug } from "@executor-js/sdk/shared"; +import { scenario } from "../src/scenario"; +import { Api, Browser, Target } from "../src/services"; + +import { visit } from "../src/surfaces/browser"; + +const api = composePluginApi([openApiHttpPlugin()] as const); +const spec = (count: number) => + JSON.stringify({ + openapi: "3.0.3", + info: { title: "Large catalog", version: "1" }, + paths: Object.fromEntries( + Array.from({ length: count }, (_, i) => [ + `/items/${i}`, + { + get: { + operationId: `item${i}`, + responses: { + "200": { + description: "Item details", + content: { "application/json": { schema: { $ref: "#/components/schemas/Item" } } }, + }, + }, + }, + }, + ]), + ), + components: { + schemas: { + Item: { + type: "object", + properties: { + id: { type: "string" }, + description: { type: "string", description: "Item description. ".repeat(2000) }, + }, + }, + }, + }, + }); + +scenario( + "OpenAPI · large imports report complete counts and preserve unrelated catalogs", + {}, + Effect.gen(function* () { + const target = yield* Target; + const { client } = yield* Api; + const browser = yield* Browser; + const identity = yield* target.newIdentity(); + const http = yield* client(api, identity); + const slug = IntegrationSlug.make(`large-${randomUUID()}`); + const neighbor = IntegrationSlug.make(`neighbor-${randomUUID()}`); + yield* Effect.gen(function* () { + const add = (integration: string, count: number) => + http.openapi.addSpec({ + payload: { + slug: integration, + spec: { kind: "blob", value: spec(count) }, + baseUrl: "https://example.invalid", + authenticationTemplate: [], + }, + }); + expect((yield* add(neighbor, 1001)).toolCount).toBe(1001); + expect((yield* add(slug, 1201)).toolCount).toBe(1201); + yield* http.connections.create({ + payload: { + owner: "org", + integration: slug, + name: ConnectionName.make("main"), + template: AuthTemplateSlug.make("none"), + value: "catalog-fixture", + }, + }); + expect(yield* http.tools.list({ query: { integration: slug } })).toHaveLength(1201); + const updated = yield* http.openapi.updateSpec({ + params: { slug }, + payload: { + spec: { kind: "blob", value: spec(1002) }, + }, + }); + expect(updated.toolCount).toBe(1002); + expect(updated.removedTools).toHaveLength(199); + expect(updated.addedTools).toEqual([]); + const failed = yield* http.openapi + .updateSpec({ + params: { slug }, + payload: { + spec: { kind: "blob", value: "invalid" }, + }, + }) + .pipe(Effect.result); + expect(failed._tag).toBe("Failure"); + const missingPaths = yield* http.openapi + .updateSpec({ + params: { slug }, + payload: { + spec: { + kind: "blob", + value: JSON.stringify({ + openapi: "3.1.0", + info: { title: "No paths", version: "1" }, + }), + }, + }, + }) + .pipe(Effect.result); + expect(missingPaths._tag).toBe("Failure"); + yield* http.connections.refresh({ + params: { + owner: "org", + integration: slug, + name: ConnectionName.make("main"), + }, + }); + expect(yield* http.tools.list({ query: { integration: slug } })).toHaveLength(1002); + const intact = yield* http.openapi.updateSpec({ + params: { slug: neighbor }, + payload: { + spec: { kind: "blob", value: spec(1001) }, + }, + }); + expect(intact.toolCount).toBe(1001); + expect(intact.removedTools).toEqual([]); + const retry = yield* http.openapi.updateSpec({ + params: { slug }, + payload: { + spec: { kind: "blob", value: spec(1002) }, + }, + }); + expect(retry.toolCount).toBe(1002); + expect(retry.addedTools).toEqual([]); + expect(retry.removedTools).toEqual([]); + const catalog = yield* http.tools.list({ query: { integration: slug } }); + expect(catalog).toHaveLength(1002); + const first = catalog[0]; + assert(first); + const schema = yield* http.tools.schema({ + query: { address: first.address }, + }); + expect(JSON.stringify(schema)).toContain("description"); + yield* browser.session(identity, async ({ page, step }) => { + await step("Open the imported catalog after refresh", async () => { + await visit(page, `/integrations/${slug}?tab=tools`); + await page.getByPlaceholder("Filter 1002 tools…").waitFor(); + }); + }); + }).pipe( + Effect.ensuring( + Effect.forEach( + [slug, neighbor], + (slug) => http.openapi.removeSpec({ params: { slug } }).pipe(Effect.orDie), + { + discard: true, + }, + ), + ), + ); + }), +); diff --git a/e2e/vitest.config.ts b/e2e/vitest.config.ts index 2ed8ba01bd..880cbcafff 100644 --- a/e2e/vitest.config.ts +++ b/e2e/vitest.config.ts @@ -49,6 +49,7 @@ export default defineConfig({ include: [ "scenarios/browser-approval.test.ts", "scenarios/microsoft-graph-full.test.ts", + "scenarios/openapi-large-catalog.test.ts", "scenarios/toolkits-mcp.test.ts", "cloudflare/**/*.test.ts", ], diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 5d77abc207..f18925c206 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -2389,7 +2389,7 @@ const makePluginStorageFacade = (input: { list: (storageInput) => Effect.gen(function* () { const rows = yield* input.core.findMany("plugin_storage", { - where: whereFor(storageInput.collection), + where: whereForPrefix(storageInput.collection, storageInput.keyPrefix), }); return sortByOwnerPrecedence(rows) .filter((row) => diff --git a/packages/core/sdk/src/plugin-storage.test.ts b/packages/core/sdk/src/plugin-storage.test.ts index e7ebc7dd39..80876e0c54 100644 --- a/packages/core/sdk/src/plugin-storage.test.ts +++ b/packages/core/sdk/src/plugin-storage.test.ts @@ -76,6 +76,8 @@ const executionHistoryPlugin = definePlugin(() => ({ owner, entries: keys.map((key) => ({ collection: toolCalls.name, key })), }), + list: (keyPrefix?: string) => + ctx.storage.pluginStorage.list({ collection: toolCalls.name, keyPrefix }), get: (key: string) => ctx.storage.toolCalls.get({ key }), getMany: (keys: readonly string[]) => ctx.storage.toolCalls.getMany({ keys }), getForOwner: (owner: Owner, key: string) => ctx.storage.toolCalls.getForOwner({ owner, key }), @@ -164,6 +166,66 @@ const failPluginStorageBulkWriteAfterFirstRow = (db: FumaDb): FumaDb => { }; describe("plugin storage collections", () => { + it.effect("filters list prefixes in the database before materializing unrelated rows", () => + Effect.gen(function* () { + const config = makeTestConfig({ + backend: "sqlite", + plugins: [executionHistoryPlugin] as const, + }); + const reads: number[] = []; + const wrap = (db: FumaDb): FumaDb => + new Proxy(db, { + get(target, property, receiver) { + if (property === "withContext") { + const withContext = target.withContext; + return withContext === undefined + ? undefined + : (context: unknown) => wrap(withContext(context)); + } + if (property === "findMany") { + const findMany: FumaDb["findMany"] = async (table, options) => { + const rows = await target.findMany(table, options); + if (table === "plugin_storage") reads.push(rows.length); + return rows; + }; + return findMany; + } + return Reflect.get(target, property, receiver); + }, + }); + const executor = yield* Effect.acquireRelease( + createExecutor({ ...config, db: wrap(config.db) }), + (instance) => + instance + .close() + .pipe(Effect.orDie, Effect.ensuring(Effect.promise(() => config.testDb.close()))), + ); + const data = call({ + runId: "run", + toolId: "tool", + status: "ok", + startedAt: "2026-01-01T00:00:00Z", + }); + yield* executor.executionHistory.recordMany("org", [ + { key: "selected.one", data }, + ...Array.from({ length: 100 }, (_, i) => ({ key: `unrelated.${i}`, data })), + ]); + reads.length = 0; + expect((yield* executor.executionHistory.list("selected.")).map((row) => row.key)).toEqual([ + "selected.one", + ]); + expect(reads).toEqual([1]); + yield* executor.executionHistory.recordMany("org", [ + { key: "literal_%.one", data }, + { key: "literalXX.one", data }, + { key: "LITERAL_%.one", data }, + ]); + expect((yield* executor.executionHistory.list("literal_%.")).map((row) => row.key)).toEqual([ + "literal_%.one", + ]); + }), + ); + it.effect("queries declared indexes through the executor's SQLite FumaDB target", () => Effect.gen(function* () { const executor = yield* makeTestExecutor({ diff --git a/packages/plugins/openapi/src/sdk/plugin.test.ts b/packages/plugins/openapi/src/sdk/plugin.test.ts index 1617e8b555..a41182fad9 100644 --- a/packages/plugins/openapi/src/sdk/plugin.test.ts +++ b/packages/plugins/openapi/src/sdk/plugin.test.ts @@ -38,6 +38,7 @@ import { } from "@executor-js/sdk/testing"; import { openApiPlugin } from "./plugin"; +import { makeDefaultOpenapiStore } from "./store"; import { type AuthenticationInput } from "./types"; import { addOpenApiTestConnection, @@ -1125,7 +1126,26 @@ paths: "addSpec accepts Microsoft Graph-scale operation catalogs from one spec", () => Effect.gen(function* () { - const executor = yield* createExecutor(makeTestConfig({ plugins: testPlugins() })); + const batchSizes: number[] = []; + const executor = yield* createExecutor( + makeTestConfig({ + plugins: [ + openApiPlugin({ + storage: (deps) => + makeDefaultOpenapiStore({ + ...deps, + pluginStorage: { + ...deps.pluginStorage, + putMany: (input) => { + batchSizes.push(input.entries.length); + return deps.pluginStorage.putMany(input); + }, + }, + }), + }), + ] as const, + }), + ); const added = yield* executor.openapi.addSpec({ spec: { kind: "blob", value: microsoftGraphScaleSpecText() }, @@ -1134,6 +1154,21 @@ paths: }); expect(added.toolCount).toBe(MICROSOFT_GRAPH_V1_OPERATION_COUNT); + expect(batchSizes.length).toBeGreaterThan(1); + expect(Math.max(...batchSizes)).toBeLessThanOrEqual(500); + expect(batchSizes.reduce((sum, count) => sum + count, 0)).toBe( + MICROSOFT_GRAPH_V1_OPERATION_COUNT, + ); + batchSizes.length = 0; + const updated = yield* executor.openapi.updateSpec("microsoft_graph_scale", { + spec: { kind: "blob", value: microsoftGraphScaleSpecText() }, + }); + expect(updated.toolCount).toBe(MICROSOFT_GRAPH_V1_OPERATION_COUNT); + expect(batchSizes.length).toBeGreaterThan(1); + expect(Math.max(...batchSizes)).toBeLessThanOrEqual(500); + expect(batchSizes.reduce((sum, count) => sum + count, 0)).toBe( + MICROSOFT_GRAPH_V1_OPERATION_COUNT, + ); }), 30_000, ); diff --git a/packages/plugins/openapi/src/sdk/plugin.ts b/packages/plugins/openapi/src/sdk/plugin.ts index b0c0558236..6f9175982f 100644 --- a/packages/plugins/openapi/src/sdk/plugin.ts +++ b/packages/plugins/openapi/src/sdk/plugin.ts @@ -54,10 +54,10 @@ import { ApiKeyAuthTemplate, describeApiKeyAuthMethod } from "@executor-js/sdk/h import { checkHealthOpenApi, compileAndPersistOpenApiSpecStreaming, - compileOpenApiSpec, + buildDefsJson, + compileAndPersistOpenApiOperations, invokeOpenApiBackedTool, listHealthCheckCandidatesOpenApi, - openApiStoredOperationsFromCompiled, resolveOpenApiBackedAnnotations, resolveOpenApiBackedTools, validateOpenApiBackedToolArgs, @@ -784,9 +784,12 @@ export const openApiPlugin = definePlugin< // Resolve URL → text and parse BEFORE opening a transaction. Holding // `BEGIN` across a network fetch is the Hyperdrive deadlock path. const resolved = yield* resolveSpecForInput(config, httpClientLayer); - const compiled = resolved.keepPathItem - ? undefined - : yield* compileOpenApiSpec(resolved.specText); + const document = resolved.keepPathItem ? undefined : yield* parse(resolved.specText); + if (document && !document.paths) { + return yield* new OpenApiExtractionError({ + message: "OpenAPI document has no paths defined", + }); + } const adapter = yield* resolveSpecFormatAdapter( options?.specFormats ?? [], config.specFormat, @@ -906,21 +909,24 @@ export const openApiPlugin = definePlugin< // The content-addressed defs blob lets the serve path resolve the // shared `definitions` without re-parsing the spec. Same idempotent, // outside-the-transaction rationale as the spec blob. - if (compiled) { - yield* ctx.storage.putDefs(specHash, JSON.stringify(compiled.hoistedDefs)); + if (document) { + yield* ctx.storage.putDefs(specHash, buildDefsJson(document)); } - yield* ctx.transaction( + const persisted = yield* ctx.transaction( Effect.gen(function* () { yield* ctx.core.integrations.register({ slug, name: - config.name?.trim() || derivedIdentity?.name || compiled?.title || resolvedSlug, + config.name?.trim() || + derivedIdentity?.name || + document?.info.title || + resolvedSlug, description: config.description ?? derivedIdentity?.description ?? - compiled?.description ?? - compiled?.title ?? + document?.info.description ?? + document?.info.title ?? resolvedSlug, config: integrationConfig satisfies OpenApiIntegrationConfig as IntegrationConfig, canRemove: true, @@ -929,30 +935,24 @@ export const openApiPlugin = definePlugin< if (config.healthCheck) { yield* ctx.core.integrations.setHealthCheck(slug, config.healthCheck); } - if (compiled) { - yield* ctx.storage.putOperations( - resolvedSlug, - openApiStoredOperationsFromCompiled(resolvedSlug, compiled), - ); - return compiled.definitions.length; + if (document) { + return yield* compileAndPersistOpenApiOperations({ + doc: document, + integration: resolvedSlug, + storage: ctx.storage, + }); } - const persisted = yield* compileAndPersistOpenApiSpecStreaming({ + return yield* compileAndPersistOpenApiSpecStreaming({ specText: resolved.specText, integration: resolvedSlug, storage: ctx.storage, specHash, keepPathItem: resolved.keepPathItem, }); - return persisted.toolCount; }), ); - const toolCount = compiled - ? compiled.definitions.length - : yield* ctx.storage - .listOperations(resolvedSlug) - .pipe(Effect.map((operations) => operations.length)); - return { slug, toolCount }; + return { slug, toolCount: persisted.toolCount }; }); // Update the spec IN PLACE: re-resolve (stored source URL / bundle, or a @@ -994,82 +994,86 @@ export const openApiPlugin = definePlugin< }); } - // Resolve + compile BEFORE the transaction (same Hyperdrive-deadlock - // rule as addSpec: never hold BEGIN across a network fetch). - const resolved = yield* resolveSpecForInput( - { - spec: specInput, - specFormat: current.specFormat, - specOverrides: nextOverrides, - headers: current.headers, - queryParams: current.queryParams, - baseUrl: current.baseUrl, - authenticationTemplate: current.authenticationTemplate, - }, - httpClientLayer, - ); - const compiled = resolved.keepPathItem - ? undefined - : yield* compileOpenApiSpec(resolved.specText); - - const previousOperations = yield* ctx.storage.listOperations(rawSlug); - const previousNames = new Set(previousOperations.map((op) => op.toolName)); + const { nextNames, previousNames } = yield* Effect.gen(function* () { + // Resolve + parse BEFORE the transaction (same Hyperdrive-deadlock + // rule as addSpec: never hold BEGIN across a network fetch). + const resolved = yield* resolveSpecForInput( + { + spec: specInput, + specFormat: current.specFormat, + specOverrides: nextOverrides, + headers: current.headers, + queryParams: current.queryParams, + baseUrl: current.baseUrl, + authenticationTemplate: current.authenticationTemplate, + }, + httpClientLayer, + ); + const document = resolved.keepPathItem ? undefined : yield* parse(resolved.specText); + if (document && !document.paths) { + return yield* new OpenApiExtractionError({ + message: "OpenAPI document has no paths defined", + }); + } - // The resolved spec text lives in the plugin blob store keyed by its - // content hash (`spec/`); the config carries only the hash. Put - // the blob outside the transaction - re-puts are idempotent and an - // aborted config update just leaves an unreferenced blob. - const specHash = yield* sha256Hex(resolved.specText); - const sourceSpecHash = - nextOverrides.length > 0 ? yield* sha256Hex(resolved.sourceSpecText) : undefined; - yield* ctx.storage.putSpec(specHash, resolved.specText); - if (sourceSpecHash) { - yield* ctx.storage.putSpec(sourceSpecHash, resolved.sourceSpecText); - } - if (compiled) { - yield* ctx.storage.putDefs(specHash, JSON.stringify(compiled.hoistedDefs)); - } + const previousNames = new Set( + (yield* ctx.storage.listOperations(rawSlug)).map((op) => op.toolName), + ); - const { - sourceSpecHash: _currentSourceSpecHash, - specOverrides: _currentSpecOverrides, - ...currentWithoutOverrides - } = current; - const nextConfig: OpenApiIntegrationConfig = { - ...currentWithoutOverrides, - specHash, - ...((resolved.specUrl ?? specInputToSpecUrl(specInput)) !== undefined - ? { specUrl: resolved.specUrl ?? specInputToSpecUrl(specInput) } - : {}), - ...(nextOverrides.length > 0 ? { specOverrides: nextOverrides, sourceSpecHash } : {}), - }; + // The resolved spec text lives in the plugin blob store keyed by its + // content hash (`spec/`); the config carries only the hash. Put + // the blob outside the transaction - re-puts are idempotent and an + // aborted config update just leaves an unreferenced blob. + const specHash = yield* sha256Hex(resolved.specText); + const sourceSpecHash = + nextOverrides.length > 0 ? yield* sha256Hex(resolved.sourceSpecText) : undefined; + yield* ctx.storage.putSpec(specHash, resolved.specText); + if (sourceSpecHash) { + yield* ctx.storage.putSpec(sourceSpecHash, resolved.sourceSpecText); + } + if (document) { + yield* ctx.storage.putDefs(specHash, buildDefsJson(document)); + } - yield* ctx.transaction( - Effect.gen(function* () { - yield* ctx.core.integrations.update(slug, { - config: nextConfig satisfies OpenApiIntegrationConfig as IntegrationConfig, - }); - if (compiled) { - yield* ctx.storage.putOperations( - rawSlug, - openApiStoredOperationsFromCompiled(rawSlug, compiled), - ); - } else { - yield* compileAndPersistOpenApiSpecStreaming({ - specText: resolved.specText, - integration: rawSlug, - storage: ctx.storage, - specHash, - keepPathItem: resolved.keepPathItem, + const { + sourceSpecHash: _currentSourceSpecHash, + specOverrides: _currentSpecOverrides, + ...currentWithoutOverrides + } = current; + const nextConfig: OpenApiIntegrationConfig = { + ...currentWithoutOverrides, + specHash, + ...((resolved.specUrl ?? specInputToSpecUrl(specInput)) !== undefined + ? { specUrl: resolved.specUrl ?? specInputToSpecUrl(specInput) } + : {}), + ...(nextOverrides.length > 0 ? { specOverrides: nextOverrides, sourceSpecHash } : {}), + }; + + const persisted = yield* ctx.transaction( + Effect.gen(function* () { + yield* ctx.core.integrations.update(slug, { + config: nextConfig satisfies OpenApiIntegrationConfig as IntegrationConfig, }); - } - }), - ); + if (document) { + return yield* compileAndPersistOpenApiOperations({ + doc: document, + integration: rawSlug, + storage: ctx.storage, + }); + } else { + return yield* compileAndPersistOpenApiSpecStreaming({ + specText: resolved.specText, + integration: rawSlug, + storage: ctx.storage, + specHash, + keepPathItem: resolved.keepPathItem, + }); + } + }), + ); - const nextOperations = compiled - ? openApiStoredOperationsFromCompiled(rawSlug, compiled) - : yield* ctx.storage.listOperations(rawSlug); - const nextNames = new Set(nextOperations.map((op) => op.toolName)); + return { nextNames: new Set(persisted.toolNames), previousNames }; + }); // Rebuild each connection's tool rows from the new spec. Outside the // transaction: refresh opens its own, and a half-refreshed catalog diff --git a/packages/plugins/openapi/src/sdk/store.test.ts b/packages/plugins/openapi/src/sdk/store.test.ts index fb7021e5a2..6124df814f 100644 --- a/packages/plugins/openapi/src/sdk/store.test.ts +++ b/packages/plugins/openapi/src/sdk/store.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "@effect/vitest"; -import { Effect, Option } from "effect"; +import { Effect, Option, Schema } from "effect"; import { Subject, @@ -13,11 +13,14 @@ import { import { makeDefaultOpenapiStore } from "./store"; import { OperationBinding } from "./types"; +const encodeBinding = Schema.encodeSync(OperationBinding); + describe("OpenAPI operation store", () => { it.effect("bounds operation storage keys while preserving tool-name lookup", () => Effect.gen(function* () { const rows = new Map(); const capturedKeys: string[] = []; + const listedPrefixes: (string | undefined)[] = []; const storageKey = (collection: string, key: string) => `${collection}\0${key}`; const unexpectedCollectionCall = () => Effect.die("Unexpected collection storage call"); const now = new Date(); @@ -105,14 +108,19 @@ describe("OpenAPI operation store", () => { }), ), ), - list: (input: { readonly collection: string; readonly keyPrefix?: string }) => - Effect.succeed( + list: (input: { + readonly collection: string; + readonly keyPrefix?: string; + }) => { + listedPrefixes.push(input.keyPrefix); + return Effect.succeed( [...rows.values()].filter( (row) => row.collection === input.collection && (input.keyPrefix === undefined || row.key.startsWith(input.keyPrefix)), ) as PluginStorageEntry[], - ), + ); + }, put: (input: { readonly owner: "org" | "user"; readonly collection: string; @@ -177,6 +185,40 @@ describe("OpenAPI operation store", () => { }, ]); + const binding = OperationBinding.make({ + method: "get", + servers: [], + pathTemplate: "/legacy", + parameters: [], + requestBody: Option.none(), + responseBody: Option.none(), + }); + yield* pluginStorage.put({ + owner: "org", + collection: "operation", + key: "microsoft_graph.legacy", + data: { + integration: "microsoft_graph", + toolName: "legacy", + binding: encodeBinding(binding), + }, + }); + yield* pluginStorage.put({ + owner: "org", + collection: "operation", + key: "microsoft_graph.other.legacy", + data: { + integration: "microsoft_graph.other", + toolName: "legacy", + binding: encodeBinding(binding), + }, + }); + yield* pluginStorage.put({ + owner: "org", + collection: "operation", + key: "unrelated.invalid", + data: { integration: "unrelated", toolName: "invalid", binding: "invalid" }, + }); expect(capturedKeys).toHaveLength(1); expect(capturedKeys[0]!.length).toBeLessThanOrEqual(255); expect(capturedKeys[0]).not.toContain(toolName); @@ -184,6 +226,38 @@ describe("OpenAPI operation store", () => { const operation = yield* store.getOperation("microsoft_graph", toolName); expect(operation?.toolName).toBe(toolName); expect(operation?.binding.pathTemplate).toBe("/users/{userId}/messages"); + expect((yield* store.listOperations("microsoft_graph")).map((op) => op.toolName)).toEqual([ + toolName, + "legacy", + ]); + yield* store.removeOperations("microsoft_graph"); + expect(yield* store.listOperations("microsoft_graph")).toEqual([]); + expect((yield* store.getOperation("microsoft_graph.other", "legacy"))?.toolName).toBe( + "legacy", + ); + expect(rows.has(storageKey("operation", "unrelated.invalid"))).toBe(true); + expect(listedPrefixes).toEqual( + Array.from({ length: 4 }, () => [ + capturedKeys[0]!.slice(0, capturedKeys[0]!.lastIndexOf(".") + 1), + "microsoft_graph.", + ]).flat(), + ); + yield* store.appendOperations("op", [{ integration: "op", toolName: "current", binding }]); + yield* pluginStorage.put({ + owner: "org", + collection: "operation", + key: "op.legacy", + data: { integration: "op", toolName: "legacy", binding: encodeBinding(binding) }, + }); + expect((yield* store.listOperations("op")).map((op) => op.toolName)).toEqual([ + "current", + "legacy", + ]); + yield* store.removeOperations("op"); + expect(yield* store.listOperations("op")).toEqual([]); + expect((yield* store.getOperation("microsoft_graph.other", "legacy"))?.toolName).toBe( + "legacy", + ); }), ); }); diff --git a/packages/plugins/openapi/src/sdk/store.ts b/packages/plugins/openapi/src/sdk/store.ts index 548a521031..a63410c0db 100644 --- a/packages/plugins/openapi/src/sdk/store.ts +++ b/packages/plugins/openapi/src/sdk/store.ts @@ -84,8 +84,11 @@ const stableKeyHash = (value: string): string => { return hash.toString(36).padStart(13, "0"); }; +const operationKeyPrefix = (integration: string): string => + `${OPERATION_KEY_VERSION}.${stableKeyHash(integration)}.`; + const operationKey = (integration: string, toolName: string): string => - `${OPERATION_KEY_VERSION}.${stableKeyHash(integration)}.${stableKeyHash(toolName)}`; + `${operationKeyPrefix(integration)}${stableKeyHash(toolName)}`; const legacyOperationKey = (integration: string, toolName: string): string => `${integration}.${toolName}`; @@ -148,13 +151,23 @@ export const makeDefaultOpenapiStore = ({ pluginStorage, blobs }: StorageDeps): }); const listRows = (integration: string) => - pluginStorage - .list({ collection: OPERATION_COLLECTION }) - .pipe( - Effect.map((rows: readonly PluginStorageEntry[]) => - rows.filter((row) => rowToOperation(row)?.integration === integration), - ), + Effect.gen(function* () { + const currentPrefix = operationKeyPrefix(integration); + const current = yield* pluginStorage.list({ + collection: OPERATION_COLLECTION, + keyPrefix: currentPrefix, + }); + const legacy = yield* pluginStorage.list({ + collection: OPERATION_COLLECTION, + keyPrefix: `${integration}.`, + }); + return [...current, ...legacy.filter((row) => !row.key.startsWith(currentPrefix))].filter( + (row) => { + const decoded = decodeOperationStorage(row.data); + return Option.isSome(decoded) && decoded.value.integration === integration; + }, ); + }); const removeOperations = (integration: string) => Effect.gen(function* () {