diff --git a/packages/plugins/google-discovery/src/sdk/binding-store.ts b/packages/plugins/google-discovery/src/sdk/binding-store.ts index 3d1c35db9..012d9ebba 100644 --- a/packages/plugins/google-discovery/src/sdk/binding-store.ts +++ b/packages/plugins/google-discovery/src/sdk/binding-store.ts @@ -15,11 +15,7 @@ import { Effect, Schema } from "effect"; -import { - defineSchema, - type StorageDeps, - type StorageFailure, -} from "@executor-js/sdk/core"; +import { defineSchema, type StorageDeps, type StorageFailure } from "@executor-js/sdk/core"; import { GoogleDiscoveryMethodBinding, @@ -134,18 +130,45 @@ const decodeStoredSourceData = Schema.decodeUnknownSync(GoogleDiscoveryStoredSou const encodeBinding = Schema.encodeSync(GoogleDiscoveryMethodBinding); const decodeBinding = Schema.decodeUnknownSync(GoogleDiscoveryMethodBinding); -const toJsonRecord = (value: unknown): Record => value as Record; +const JsonRecord = Schema.Record(Schema.String, Schema.Unknown); +const decodeJsonRecord = Schema.decodeUnknownSync(JsonRecord); +const decodeJsonString = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown)); const decodeJson = (value: unknown): unknown => { if (value === null || value === undefined) return value; if (typeof value !== "string") return value; - try { - return JSON.parse(value); - } catch { - return value; - } + return decodeJsonString(value); }; +const SourceRow = Schema.Struct({ + id: Schema.String, + scope_id: Schema.String, + name: Schema.String, +}); +const decodeSourceRow = Schema.decodeUnknownSync(SourceRow); + +const BindingRow = Schema.Struct({ + id: Schema.String, + source_id: Schema.String, +}); +const decodeBindingRow = Schema.decodeUnknownSync(BindingRow); + +const CredentialValueRow = Schema.Struct({ + name: Schema.String, + kind: Schema.Union([Schema.Literal("text"), Schema.Literal("secret")]), + text_value: Schema.optional(Schema.NullOr(Schema.String)), + secret_id: Schema.optional(Schema.NullOr(Schema.String)), + secret_prefix: Schema.optional(Schema.NullOr(Schema.String)), +}); +const decodeCredentialValueRow = Schema.decodeUnknownSync(CredentialValueRow); + +const CredentialLookupRow = Schema.Struct({ + source_id: Schema.String, + scope_id: Schema.String, + name: Schema.String, +}); +const decodeCredentialLookupRow = Schema.decodeUnknownSync(CredentialLookupRow); + // --- auth column packing/unpacking ------------------------------------------ interface AuthColumns { @@ -239,14 +262,15 @@ const rowsToValueMap = ( ): Record => { const out: Record = {}; for (const row of rows) { - const name = row.name as string; - if (row.kind === "secret" && typeof row.secret_id === "string") { - const prefix = row.secret_prefix as string | undefined | null; + const decoded = decodeCredentialValueRow(row); + if (decoded.kind === "secret" && decoded.secret_id) { + const prefix = decoded.secret_prefix; + const name = decoded.name; out[name] = prefix - ? { secretId: row.secret_id, prefix } - : { secretId: row.secret_id }; - } else if (row.kind === "text" && typeof row.text_value === "string") { - out[name] = row.text_value; + ? { secretId: decoded.secret_id, prefix } + : { secretId: decoded.secret_id }; + } else if (decoded.kind === "text" && typeof decoded.text_value === "string") { + out[decoded.name] = decoded.text_value; } } return out; @@ -320,9 +344,7 @@ export interface GoogleDiscoveryStore { /** Source rows whose oauth2 auth columns reference the given secret id. * `slot` distinguishes client_id vs client_secret. */ - readonly findSourcesBySecret: ( - secretId: string, - ) => Effect.Effect< + readonly findSourcesBySecret: (secretId: string) => Effect.Effect< readonly { readonly namespace: string; readonly scope_id: string; @@ -333,9 +355,7 @@ export interface GoogleDiscoveryStore { >; /** Source rows whose oauth2 auth points at the given connection id. */ - readonly findSourcesByConnection: ( - connectionId: string, - ) => Effect.Effect< + readonly findSourcesByConnection: (connectionId: string) => Effect.Effect< readonly { readonly namespace: string; readonly scope_id: string; @@ -382,7 +402,7 @@ export const makeGoogleDiscoveryStore = ( }); if (!row) return null; const decoded = decodeBinding(decodeJson(row.binding)); - return { namespace: row.source_id as string, binding: decoded }; + return { namespace: decodeBindingRow(row).source_id, binding: decoded }; }), putBinding: (toolId, sourceId, scope, binding) => @@ -404,7 +424,7 @@ export const makeGoogleDiscoveryStore = ( id: toolId, scope_id: scope, source_id: sourceId, - binding: toJsonRecord(encodeBinding(binding)), + binding: decodeJsonRecord(encodeBinding(binding)), created_at: new Date(), }, forceAllowId: true, @@ -420,7 +440,7 @@ export const makeGoogleDiscoveryStore = ( { field: "scope_id", value: scope }, ], }); - const ids = rows.map((r) => r.id as string); + const ids = rows.map((r) => decodeBindingRow(r).id); yield* db.deleteMany({ model: "google_discovery_binding", where: [ @@ -442,7 +462,7 @@ export const makeGoogleDiscoveryStore = ( }); const out = new Map(); for (const row of rows) { - out.set(row.id as string, decodeBinding(decodeJson(row.binding))); + out.set(decodeBindingRow(row).id, decodeBinding(decodeJson(row.binding))); } return out; }), @@ -462,7 +482,7 @@ export const makeGoogleDiscoveryStore = ( yield* deleteSourceChildren(source.namespace, source.scope); const encoded = stripExtractedFields( - encodeStoredSourceData(source.config) as Record, + decodeJsonRecord(encodeStoredSourceData(source.config)), ); yield* db.create({ model: "google_discovery_source", @@ -470,18 +490,14 @@ export const makeGoogleDiscoveryStore = ( id: source.namespace, scope_id: source.scope, name: source.name, - config: toJsonRecord(encoded), + config: encoded, created_at: now, updated_at: now, ...authToColumns(source.config.auth), }, forceAllowId: true, }); - yield* writeCredentialRows( - source.namespace, - source.scope, - source.config.credentials, - ); + yield* writeCredentialRows(source.namespace, source.scope, source.config.credentials); }), updateSourceMeta: (sourceId, scope, update) => @@ -502,7 +518,7 @@ export const makeGoogleDiscoveryStore = ( { field: "scope_id", value: scope }, ], update: { - name: update.name ?? (row.name as string), + name: update.name ?? decodeSourceRow(row).name, updated_at: new Date(), ...authToColumns(auth), }, @@ -531,10 +547,11 @@ export const makeGoogleDiscoveryStore = ( ], }); if (!row) return null; + const sourceRow = decodeSourceRow(row); return { - namespace: row.id as string, - scope: row.scope_id as string, - name: row.name as string, + namespace: sourceRow.id, + scope: sourceRow.scope_id, + name: sourceRow.name, config: yield* hydrateStoredSourceData(row, sourceId, scope), }; }), @@ -558,15 +575,11 @@ export const makeGoogleDiscoveryStore = ( [ db.findMany({ model: "google_discovery_source", - where: [ - { field: "auth_client_id_secret_id", value: secretId }, - ], + where: [{ field: "auth_client_id_secret_id", value: secretId }], }), db.findMany({ model: "google_discovery_source", - where: [ - { field: "auth_client_secret_secret_id", value: secretId }, - ], + where: [{ field: "auth_client_secret_secret_id", value: secretId }], }), ], { concurrency: "unbounded" }, @@ -578,18 +591,20 @@ export const makeGoogleDiscoveryStore = ( readonly slot: string; }[] = []; for (const r of byClientId) { + const row = decodeSourceRow(r); out.push({ - namespace: r.id as string, - scope_id: r.scope_id as string, - name: r.name as string, + namespace: row.id, + scope_id: row.scope_id, + name: row.name, slot: "auth.oauth2.client_id", }); } for (const r of byClientSecret) { + const row = decodeSourceRow(r); out.push({ - namespace: r.id as string, - scope_id: r.scope_id as string, - name: r.name as string, + namespace: row.id, + scope_id: row.scope_id, + name: row.name, slot: "auth.oauth2.client_secret", }); } @@ -604,12 +619,15 @@ export const makeGoogleDiscoveryStore = ( }) .pipe( Effect.map((rows) => - rows.map((r) => ({ - namespace: r.id as string, - scope_id: r.scope_id as string, - name: r.name as string, - slot: "auth.oauth2.connection", - })), + rows.map((r) => { + const row = decodeSourceRow(r); + return { + namespace: row.id, + scope_id: row.scope_id, + name: row.name, + slot: "auth.oauth2.connection", + }; + }), ), ), @@ -629,18 +647,24 @@ export const makeGoogleDiscoveryStore = ( { concurrency: "unbounded" }, ); return [ - ...headers.map((r) => ({ - kind: "credential_header" as const, - source_id: r.source_id as string, - scope_id: r.scope_id as string, - name: r.name as string, - })), - ...params.map((r) => ({ - kind: "credential_query_param" as const, - source_id: r.source_id as string, - scope_id: r.scope_id as string, - name: r.name as string, - })), + ...headers.map((r) => { + const row = decodeCredentialLookupRow(r); + return { + kind: "credential_header" as const, + source_id: row.source_id, + scope_id: row.scope_id, + name: row.name, + }; + }), + ...params.map((r) => { + const row = decodeCredentialLookupRow(r); + return { + kind: "credential_query_param" as const, + source_id: row.source_id, + scope_id: row.scope_id, + name: row.name, + }; + }), ]; }), @@ -651,8 +675,9 @@ export const makeGoogleDiscoveryStore = ( const requested = new Set(keys); const out = new Map(); for (const r of rows) { - const key = `${r.scope_id as string}:${r.id as string}`; - if (requested.has(key)) out.set(key, r.name as string); + const row = decodeSourceRow(r); + const key = `${row.scope_id}:${row.id}`; + if (requested.has(key)) out.set(key, row.name); } return out; }), @@ -698,11 +723,7 @@ export const makeGoogleDiscoveryStore = ( forceAllowId: true, }); } - const paramRows = valueMapToRows( - sourceId, - scope, - credentials.queryParams, - ); + const paramRows = valueMapToRows(sourceId, scope, credentials.queryParams); if (paramRows.length > 0) { yield* db.createMany({ model: "google_discovery_source_credential_query_param", @@ -719,7 +740,7 @@ export const makeGoogleDiscoveryStore = ( scope: string, ): Effect.Effect { return Effect.gen(function* () { - const partial = decodeJson(row.config) as Record; + const partial = decodeJsonRecord(decodeJson(row.config)); const headerRows = yield* db.findMany({ model: "google_discovery_source_credential_header", where: [ @@ -737,8 +758,7 @@ export const makeGoogleDiscoveryStore = ( const headers = rowsToValueMap(headerRows); const queryParams = rowsToValueMap(paramRows); const credentials = - Object.keys(headers).length === 0 && - Object.keys(queryParams).length === 0 + Object.keys(headers).length === 0 && Object.keys(queryParams).length === 0 ? undefined : { ...(Object.keys(headers).length > 0 ? { headers } : {}), @@ -757,9 +777,7 @@ export const makeGoogleDiscoveryStore = ( // Strip auth/credentials from the encoded source-data shape. Those // moved to columns and child tables; the remaining structural fields // live in the `config` JSON. -const stripExtractedFields = ( - encoded: Record, -): Record => { +const stripExtractedFields = (encoded: Record): Record => { const { auth, credentials, ...rest } = encoded; void auth; void credentials; diff --git a/packages/plugins/mcp/src/sdk/binding-store.ts b/packages/plugins/mcp/src/sdk/binding-store.ts index 898340c29..9bbd69c8c 100644 --- a/packages/plugins/mcp/src/sdk/binding-store.ts +++ b/packages/plugins/mcp/src/sdk/binding-store.ts @@ -18,11 +18,7 @@ import { Effect, Schema } from "effect"; -import { - defineSchema, - type StorageDeps, - type StorageFailure, -} from "@executor-js/sdk/core"; +import { defineSchema, type StorageDeps, type StorageFailure } from "@executor-js/sdk/core"; import { McpToolBinding, @@ -120,15 +116,38 @@ const encodeSourceData = Schema.encodeSync(McpStoredSourceData); const decodeBinding = Schema.decodeUnknownSync(McpToolBinding); const encodeBinding = Schema.encodeSync(McpToolBinding); +const JsonRecord = Schema.Record(Schema.String, Schema.Unknown); +const decodeJsonRecord = Schema.decodeUnknownSync(JsonRecord); +const decodeJsonString = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown)); + const coerceJson = (value: unknown): unknown => { if (typeof value !== "string") return value; - try { - return JSON.parse(value); - } catch { - return value; - } + return decodeJsonString(value); }; +const SourceRow = Schema.Struct({ + id: Schema.String, + scope_id: Schema.String, + name: Schema.String, +}); +const decodeSourceRow = Schema.decodeUnknownSync(SourceRow); + +const ChildLookupRow = Schema.Struct({ + source_id: Schema.String, + scope_id: Schema.String, + name: Schema.String, +}); +const decodeChildLookupRow = Schema.decodeUnknownSync(ChildLookupRow); + +const SecretBackedValueRow = Schema.Struct({ + name: Schema.String, + kind: Schema.Union([Schema.Literal("text"), Schema.Literal("secret")]), + text_value: Schema.optional(Schema.NullOr(Schema.String)), + secret_id: Schema.optional(Schema.NullOr(Schema.String)), + secret_prefix: Schema.optional(Schema.NullOr(Schema.String)), +}); +const decodeSecretBackedValueRow = Schema.decodeUnknownSync(SecretBackedValueRow); + // --- auth column packing/unpacking ------------------------------------------ interface AuthColumns { @@ -162,26 +181,28 @@ const authToColumns = (auth: McpConnectionAuth): AuthColumns => { }; const columnsToAuth = (row: Record): McpConnectionAuth => { - const kind = row.auth_kind as string; + const kind = row.auth_kind; if (kind === "header" && typeof row.auth_secret_id === "string") { - const prefix = row.auth_secret_prefix as string | null | undefined; + const prefix = typeof row.auth_secret_prefix === "string" ? row.auth_secret_prefix : undefined; return { kind: "header", - headerName: (row.auth_header_name as string | null) ?? "", + headerName: typeof row.auth_header_name === "string" ? row.auth_header_name : "", secretId: row.auth_secret_id, ...(prefix ? { prefix } : {}), }; } if (kind === "oauth2" && typeof row.auth_connection_id === "string") { - const cid = row.auth_client_id_secret_id as string | null | undefined; - const csec = row.auth_client_secret_secret_id as string | null | undefined; + const cid = + typeof row.auth_client_id_secret_id === "string" ? row.auth_client_id_secret_id : undefined; + const csec = + typeof row.auth_client_secret_secret_id === "string" + ? row.auth_client_secret_secret_id + : undefined; return { kind: "oauth2", connectionId: row.auth_connection_id, ...(cid ? { clientIdSecretId: cid } : {}), - ...(csec !== undefined && csec !== null - ? { clientSecretSecretId: csec } - : {}), + ...(csec !== undefined && csec !== null ? { clientSecretSecretId: csec } : {}), }; } return { kind: "none" }; @@ -236,14 +257,15 @@ const rowsToValueMap = ( ): Record => { const out: Record = {}; for (const row of rows) { - const name = row.name as string; - if (row.kind === "secret" && typeof row.secret_id === "string") { - const prefix = row.secret_prefix as string | undefined | null; + const decoded = decodeSecretBackedValueRow(row); + if (decoded.kind === "secret" && decoded.secret_id) { + const prefix = decoded.secret_prefix; + const name = decoded.name; out[name] = prefix - ? { secretId: row.secret_id, prefix } - : { secretId: row.secret_id }; - } else if (row.kind === "text" && typeof row.text_value === "string") { - out[name] = row.text_value; + ? { secretId: decoded.secret_id, prefix } + : { secretId: decoded.secret_id }; + } else if (decoded.kind === "text" && typeof decoded.text_value === "string") { + out[decoded.name] = decoded.text_value; } } return out; @@ -320,10 +342,7 @@ export interface McpBindingStore { scope: string, ) => Effect.Effect; readonly putSource: (source: McpStoredSource) => Effect.Effect; - readonly removeSource: ( - namespace: string, - scope: string, - ) => Effect.Effect; + readonly removeSource: (namespace: string, scope: string) => Effect.Effect; // --------------------------------------------------------------------- // Usage lookups — back `usagesForSecret` / `usagesForConnection`. @@ -332,9 +351,7 @@ export interface McpBindingStore { /** Source rows whose flattened auth columns reference the given * secret id. The `slot` field on each result tags which column * matched so the caller can produce a precise Usage.slot. */ - readonly findSourcesBySecret: ( - secretId: string, - ) => Effect.Effect< + readonly findSourcesBySecret: (secretId: string) => Effect.Effect< readonly { readonly namespace: string; readonly scope_id: string; @@ -345,9 +362,7 @@ export interface McpBindingStore { >; /** Source rows whose oauth2 auth points at the given connection id. */ - readonly findSourcesByConnection: ( - connectionId: string, - ) => Effect.Effect< + readonly findSourcesByConnection: (connectionId: string) => Effect.Effect< readonly { readonly namespace: string; readonly scope_id: string; @@ -379,9 +394,7 @@ export interface McpBindingStore { // Factory // --------------------------------------------------------------------------- -export const makeMcpStore = ({ - adapter: db, -}: StorageDeps): McpBindingStore => { +export const makeMcpStore = ({ adapter: db }: StorageDeps): McpBindingStore => { return { listBindingsBySource: (namespace, scope) => Effect.gen(function* () { @@ -486,25 +499,18 @@ export const makeMcpStore = ({ yield* deleteSourceChildren(source.namespace, source.scope); const auth: McpConnectionAuth = - source.config.transport === "remote" - ? source.config.auth - : { kind: "none" }; + source.config.transport === "remote" ? source.config.auth : { kind: "none" }; const authCols = authToColumns(auth); - const headers = - source.config.transport === "remote" - ? source.config.headers - : undefined; + const headers = source.config.transport === "remote" ? source.config.headers : undefined; const queryParams = - source.config.transport === "remote" - ? source.config.queryParams - : undefined; + source.config.transport === "remote" ? source.config.queryParams : undefined; // The encoded config keeps every plugin-private field but // strips auth/headers/queryParams — those moved to columns/ // child tables. We round-trip through encodeSourceData so the // remaining fields stay in the same JSON shape decode expects. const encodedConfig = stripExtractedFields( - encodeSourceData(source.config) as Record, + decodeJsonRecord(encodeSourceData(source.config)), ); yield* db.create({ @@ -520,11 +526,7 @@ export const makeMcpStore = ({ forceAllowId: true, }); - const headerRows = valueMapToRows( - source.namespace, - source.scope, - headers, - ); + const headerRows = valueMapToRows(source.namespace, source.scope, headers); if (headerRows.length > 0) { yield* db.createMany({ model: "mcp_source_header", @@ -532,11 +534,7 @@ export const makeMcpStore = ({ forceAllowId: true, }); } - const paramRows = valueMapToRows( - source.namespace, - source.scope, - queryParams, - ); + const paramRows = valueMapToRows(source.namespace, source.scope, queryParams); if (paramRows.length > 0) { yield* db.createMany({ model: "mcp_source_query_param", @@ -579,36 +577,52 @@ export const makeMcpStore = ({ }), db.findMany({ model: "mcp_source", - where: [ - { field: "auth_client_id_secret_id", value: secretId }, - ], + where: [{ field: "auth_client_id_secret_id", value: secretId }], }), db.findMany({ model: "mcp_source", - where: [ - { field: "auth_client_secret_secret_id", value: secretId }, - ], + where: [{ field: "auth_client_secret_secret_id", value: secretId }], }), ], { concurrency: "unbounded" }, ); - const dedup = new Map>(); - for (const r of [...byHeader, ...byClientId, ...byClientSecret]) { - dedup.set(`${r.scope_id}:${r.id}`, r); + const dedup = new Map< + string, + { + readonly row: Record; + readonly slot: "auth.header" | "auth.oauth2.client_id" | "auth.oauth2.client_secret"; + } + >(); + for (const r of byHeader) { + const row = decodeSourceRow(r); + dedup.set(`${row.scope_id}:${row.id}`, { + row: r, + slot: "auth.header", + }); } - return [...dedup.values()].map((row) => ({ - namespace: row.id as string, - scope_id: row.scope_id as string, - name: row.name as string, - slot: - (byHeader as readonly Record[]).includes(row) - ? "auth.header" - : (byClientId as readonly Record[]).includes( - row, - ) - ? "auth.oauth2.client_id" - : "auth.oauth2.client_secret", - })); + for (const r of byClientId) { + const row = decodeSourceRow(r); + dedup.set(`${row.scope_id}:${row.id}`, { + row: r, + slot: "auth.oauth2.client_id", + }); + } + for (const r of byClientSecret) { + const row = decodeSourceRow(r); + dedup.set(`${row.scope_id}:${row.id}`, { + row: r, + slot: "auth.oauth2.client_secret", + }); + } + return [...dedup.values()].map(({ row: rawRow, slot }) => { + const row = decodeSourceRow(rawRow); + return { + namespace: row.id, + scope_id: row.scope_id, + name: row.name, + slot, + }; + }); }), findSourcesByConnection: (connectionId) => @@ -619,12 +633,15 @@ export const makeMcpStore = ({ }) .pipe( Effect.map((rows) => - rows.map((r) => ({ - namespace: r.id as string, - scope_id: r.scope_id as string, - name: r.name as string, - slot: "auth.oauth2.connection", - })), + rows.map((r) => { + const row = decodeSourceRow(r); + return { + namespace: row.id, + scope_id: row.scope_id, + name: row.name, + slot: "auth.oauth2.connection", + }; + }), ), ), @@ -644,18 +661,24 @@ export const makeMcpStore = ({ { concurrency: "unbounded" }, ); return [ - ...headers.map((r) => ({ - kind: "header" as const, - source_id: r.source_id as string, - scope_id: r.scope_id as string, - name: r.name as string, - })), - ...params.map((r) => ({ - kind: "query_param" as const, - source_id: r.source_id as string, - scope_id: r.scope_id as string, - name: r.name as string, - })), + ...headers.map((r) => { + const row = decodeChildLookupRow(r); + return { + kind: "header" as const, + source_id: row.source_id, + scope_id: row.scope_id, + name: row.name, + }; + }), + ...params.map((r) => { + const row = decodeChildLookupRow(r); + return { + kind: "query_param" as const, + source_id: row.source_id, + scope_id: row.scope_id, + name: row.name, + }; + }), ]; }), @@ -666,8 +689,9 @@ export const makeMcpStore = ({ const requested = new Set(keys); const out = new Map(); for (const r of rows) { - const key = `${r.scope_id as string}:${r.id as string}`; - if (requested.has(key)) out.set(key, r.name as string); + const row = decodeSourceRow(r); + const key = `${row.scope_id}:${row.id}`; + if (requested.has(key)) out.set(key, row.name); } return out; }), @@ -679,10 +703,7 @@ export const makeMcpStore = ({ function deleteSourceChildren(namespace: string, scope: string) { return Effect.gen(function* () { - for (const model of [ - "mcp_source_header", - "mcp_source_query_param", - ] as const) { + for (const model of ["mcp_source_header", "mcp_source_query_param"] as const) { yield* db.deleteMany({ model, where: [ @@ -704,7 +725,7 @@ export const makeMcpStore = ({ // moved to columns / child tables). We must rehydrate the full // shape BEFORE handing it to the schema decoder, because // `McpRemoteSourceData.auth` is required. - const partial = coerceJson(row.config) as Record; + const partial = decodeJsonRecord(coerceJson(row.config)); if (partial.transport !== "remote") { // stdio sources have no extracted fields — decode as-is. return decodeSourceData(partial); @@ -740,9 +761,7 @@ export const makeMcpStore = ({ // Keeps the remaining structural fields (transport, endpoint, etc.) in // the JSON config column. Per-transport: only the remote variant has // these fields, so this is a no-op for stdio. -const stripExtractedFields = ( - encoded: Record, -): Record => { +const stripExtractedFields = (encoded: Record): Record => { if (encoded.transport !== "remote") return encoded; const { auth, headers, queryParams, ...rest } = encoded; void auth;