6c7eed7580
Fresh history cut from cynarAI/evoo antigravity-handoff/project-timeline (Voice First + project timeline). Brand docs/titles as evOo; keep @rakazo/* packages as tech debt. Excludes node_modules, caches, secrets, voice blobs.
498 lines
16 KiB
TypeScript
498 lines
16 KiB
TypeScript
import { createHmac } from "node:crypto";
|
|
import type {
|
|
AdapterContext,
|
|
ConnectorCall,
|
|
ConnectorCatalogItem,
|
|
ConnectorEvent,
|
|
ConnectorTool,
|
|
ManagedConnectorProvider,
|
|
} from "@rakazo/adapter-kit";
|
|
import { collectPages, filterCatalog } from "./composio-connector.js";
|
|
import {
|
|
combineSignals,
|
|
redactConnectorPayload,
|
|
sanitizeConnectorError,
|
|
} from "./connector-safety.js";
|
|
import {
|
|
callRemoteMcpTool,
|
|
listRemoteMcpTools,
|
|
type RemoteTransportDependencies,
|
|
} from "./remote-mcp.js";
|
|
import { isVitestRuntime } from "./test-runtime.js";
|
|
import { readBodyCapped, withAbort } from "./web-ssrf.js";
|
|
|
|
const API_BASE = "https://api.pipedream.com";
|
|
const MCP_ENDPOINT = "https://remote.mcp.pipedream.net/v3";
|
|
const DIRECTORY_TTL_MS = 10 * 60_000;
|
|
const REQUEST_TIMEOUT_MS = 30_000;
|
|
export const MAX_PIPEDREAM_RESPONSE_BYTES = 2 * 1024 * 1024;
|
|
|
|
export interface PipedreamConnectorConfig {
|
|
clientId: string;
|
|
clientSecret: string;
|
|
projectId: string;
|
|
environment: "development" | "production";
|
|
identitySecret: string;
|
|
}
|
|
|
|
export interface PipedreamEnvironmentValues {
|
|
pipedreamClientId?: string;
|
|
pipedreamClientSecret?: string;
|
|
pipedreamProjectId?: string;
|
|
pipedreamEnvironment?: string;
|
|
encryptionKey: string;
|
|
}
|
|
|
|
export function pipedreamConfigFromEnv(
|
|
values: PipedreamEnvironmentValues,
|
|
): PipedreamConnectorConfig {
|
|
return {
|
|
clientId: values.pipedreamClientId ?? "",
|
|
clientSecret: values.pipedreamClientSecret ?? "",
|
|
projectId: values.pipedreamProjectId ?? "",
|
|
environment: values.pipedreamEnvironment === "production" ? "production" : "development",
|
|
identitySecret: values.encryptionKey,
|
|
};
|
|
}
|
|
|
|
export type PipedreamConnectorDependencies = RemoteTransportDependencies;
|
|
|
|
type PipedreamApp = {
|
|
id: string;
|
|
name_slug: string;
|
|
name: string;
|
|
img_src?: string;
|
|
};
|
|
|
|
type PipedreamAccount = {
|
|
id: string;
|
|
healthy?: boolean;
|
|
dead?: boolean;
|
|
app?: PipedreamApp;
|
|
};
|
|
|
|
export function isPipedreamEnabled(config: Partial<PipedreamConnectorConfig>): boolean {
|
|
return Boolean(
|
|
config.clientId &&
|
|
config.clientSecret &&
|
|
config.projectId &&
|
|
config.environment &&
|
|
config.identitySecret &&
|
|
!isVitestRuntime(),
|
|
);
|
|
}
|
|
|
|
function securePipedreamConnectUrl(value: string): URL {
|
|
let url: URL;
|
|
try {
|
|
url = new URL(value);
|
|
} catch {
|
|
throw new Error("Pipedream did not return a secure HTTPS connect URL");
|
|
}
|
|
if (url.protocol !== "https:" || url.username || url.password) {
|
|
throw new Error("Pipedream did not return a secure HTTPS connect URL");
|
|
}
|
|
return url;
|
|
}
|
|
|
|
export class PipedreamConnector implements ManagedConnectorProvider {
|
|
private accessToken?: { value: string; expiresAt: number };
|
|
private accessTokenRequest?: Promise<string>;
|
|
private directory?: { value: PipedreamApp[]; expiresAt: number };
|
|
private directoryRequest?: Promise<PipedreamApp[]>;
|
|
|
|
constructor(
|
|
private readonly config: PipedreamConnectorConfig,
|
|
private readonly dependencies: PipedreamConnectorDependencies = {},
|
|
) {}
|
|
|
|
describe() {
|
|
return {
|
|
id: "pipedream",
|
|
contractVersion: "1",
|
|
adapterVersion: "0.1.0",
|
|
capabilities: { discover: true, oauth: true, secretsBrokered: true },
|
|
};
|
|
}
|
|
|
|
async catalog(context: AdapterContext, query?: string): Promise<ConnectorCatalogItem[]> {
|
|
const [apps, connected] = await Promise.all([
|
|
this.apps(),
|
|
this.listConnectedExternalIds(context),
|
|
]);
|
|
const connectedSet = new Set(connected);
|
|
return filterCatalog(
|
|
apps.map((app) => ({
|
|
connectorId: "pipedream",
|
|
slug: app.name_slug,
|
|
name: app.name,
|
|
logo: app.img_src ?? null,
|
|
connected: connectedSet.has(app.name_slug),
|
|
noAuth: false,
|
|
})),
|
|
query ?? "",
|
|
);
|
|
}
|
|
|
|
async warmDirectory(): Promise<void> {
|
|
await this.apps();
|
|
}
|
|
|
|
async listConnectedExternalIds(context: AdapterContext): Promise<string[]> {
|
|
const accounts = await this.accounts(context);
|
|
return [
|
|
...new Set(
|
|
accounts
|
|
.filter((account) => account.healthy !== false && account.dead !== true)
|
|
.map((account) => account.app?.name_slug)
|
|
.filter((slug): slug is string => Boolean(slug)),
|
|
),
|
|
];
|
|
}
|
|
|
|
async discoverTools(context: AdapterContext): Promise<ConnectorTool[]> {
|
|
const apps =
|
|
context.connectedConnections
|
|
?.filter((connection) => connection.connectorId === "pipedream")
|
|
.map((connection) => connection.externalId) ?? [];
|
|
if (apps.length === 0) return [];
|
|
const token = await this.token();
|
|
const groups = await Promise.all(
|
|
[...new Set(apps)].slice(0, 20).map(async (app) => {
|
|
const tools = await listRemoteMcpTools({
|
|
endpoint: MCP_ENDPOINT,
|
|
headers: this.mcpHeaders(context, app, token),
|
|
signal: context.signal,
|
|
fetch: this.dependencies.fetch,
|
|
resolveHostname: this.dependencies.resolveHostname,
|
|
});
|
|
return tools.map((tool) => ({
|
|
...tool,
|
|
route: { connectorId: "pipedream", resourceId: app, toolName: tool.name },
|
|
}));
|
|
}),
|
|
);
|
|
return groups.flat();
|
|
}
|
|
|
|
async *execute(call: ConnectorCall, context: AdapterContext): AsyncIterable<ConnectorEvent> {
|
|
const app = call.route?.resourceId;
|
|
if (!app) {
|
|
yield { type: "error", message: "Pipedream app route is missing" };
|
|
return;
|
|
}
|
|
let token: string | undefined;
|
|
try {
|
|
token = await this.token();
|
|
const result = await callRemoteMcpTool(
|
|
{
|
|
endpoint: MCP_ENDPOINT,
|
|
headers: this.mcpHeaders(context, app, token),
|
|
signal: context.signal,
|
|
fetch: this.dependencies.fetch,
|
|
resolveHostname: this.dependencies.resolveHostname,
|
|
},
|
|
call.route?.toolName ?? call.tool,
|
|
call.args,
|
|
);
|
|
yield {
|
|
type: "result",
|
|
data: redactConnectorPayload(result, [token, this.config.clientSecret]),
|
|
};
|
|
} catch (error) {
|
|
yield {
|
|
type: "error",
|
|
message: sanitizeConnectorError(
|
|
error,
|
|
[token, this.config.clientSecret].filter((value): value is string => Boolean(value)),
|
|
),
|
|
};
|
|
}
|
|
}
|
|
|
|
async begin(
|
|
request: { provider: string; redirectUrl: string },
|
|
context: AdapterContext,
|
|
): Promise<{ authorizationUrl: string | null; state: string }> {
|
|
const response = await this.request<{ connect_link_url: string }>(
|
|
`/v1/connect/${encodeURIComponent(this.config.projectId)}/tokens`,
|
|
{
|
|
method: "POST",
|
|
body: JSON.stringify({
|
|
external_user_id: this.externalUserId(context),
|
|
success_redirect_uri: request.redirectUrl,
|
|
error_redirect_uri: request.redirectUrl,
|
|
scope: "connect:accounts:read connect:accounts:write",
|
|
}),
|
|
},
|
|
context.signal,
|
|
);
|
|
const url = securePipedreamConnectUrl(response.connect_link_url);
|
|
url.searchParams.set("app", request.provider);
|
|
return { authorizationUrl: url.toString(), state: request.provider };
|
|
}
|
|
|
|
async complete(request: { state: string }): Promise<{ connectionRef: string }> {
|
|
return { connectionRef: request.state };
|
|
}
|
|
|
|
/**
|
|
* Prefer a concrete Pipedream account id over the app slug so multi-account
|
|
* revoke can target one authorization. spaceId is required for the hashed
|
|
* external user id Pipedream uses to scope accounts.
|
|
*/
|
|
async resolveConnectedAccountId(
|
|
userId: string,
|
|
slug: string,
|
|
currentRef: string | null | undefined,
|
|
excludeIds: string[] = [],
|
|
spaceId?: string,
|
|
): Promise<string | undefined> {
|
|
const excluded = new Set(excludeIds.filter(Boolean));
|
|
const current = currentRef?.trim() || undefined;
|
|
if (!spaceId) {
|
|
return current && current !== slug && !excluded.has(current) ? current : undefined;
|
|
}
|
|
// Concrete account ids are already authoritative; do not reassign them.
|
|
if (current && current !== slug) {
|
|
return excluded.has(current) ? undefined : current;
|
|
}
|
|
const context = {
|
|
userId,
|
|
spaceId,
|
|
signal: AbortSignal.timeout(30_000),
|
|
} as AdapterContext;
|
|
const accounts = await this.accounts(context, slug);
|
|
const ids = accounts
|
|
.filter((account) => account.healthy !== false && account.dead !== true)
|
|
.map((account) => account.id)
|
|
.filter(Boolean);
|
|
return ids.find((id) => !excluded.has(id));
|
|
}
|
|
|
|
async connectionReady(context: AdapterContext, externalId: string): Promise<boolean> {
|
|
const accounts = await this.accounts(context, externalId);
|
|
return accounts.some(
|
|
(account) =>
|
|
account.app?.name_slug === externalId && account.healthy !== false && account.dead !== true,
|
|
);
|
|
}
|
|
|
|
async revoke(externalId: string, context: AdapterContext): Promise<void> {
|
|
let accounts: PipedreamAccount[];
|
|
try {
|
|
accounts = await this.accounts(context);
|
|
} catch (error) {
|
|
// Listing failed before any DELETE — mark so callers can restore local retry state.
|
|
if (error && typeof error === "object") {
|
|
(error as { remoteRevokePreDelete?: boolean }).remoteRevokePreDelete = true;
|
|
}
|
|
throw error;
|
|
}
|
|
const targets = accounts.filter(
|
|
(account) => account.id === externalId || account.app?.name_slug === externalId,
|
|
);
|
|
await Promise.all(
|
|
targets.map((account) =>
|
|
this.request(
|
|
`/v1/connect/${encodeURIComponent(this.config.projectId)}/accounts/${encodeURIComponent(account.id)}`,
|
|
{ method: "DELETE" },
|
|
context.signal,
|
|
),
|
|
),
|
|
);
|
|
}
|
|
|
|
/**
|
|
* Delete remote accounts for `slug` that no local row still references. Used when a
|
|
* begin loses a race to revoke and only the app slug is available — never revoke by
|
|
* slug alone, or every sibling authorization would be deleted.
|
|
*/
|
|
async revokeUnreferencedAccounts(
|
|
slug: string,
|
|
keepAccountIds: string[],
|
|
context: AdapterContext,
|
|
): Promise<void> {
|
|
const keep = new Set(keepAccountIds.filter(Boolean));
|
|
const accounts = await this.accounts(context, slug);
|
|
const orphans = accounts.filter(
|
|
(account) =>
|
|
account.app?.name_slug === slug &&
|
|
account.healthy !== false &&
|
|
account.dead !== true &&
|
|
!keep.has(account.id),
|
|
);
|
|
await Promise.all(
|
|
orphans.map((account) =>
|
|
this.request(
|
|
`/v1/connect/${encodeURIComponent(this.config.projectId)}/accounts/${encodeURIComponent(account.id)}`,
|
|
{ method: "DELETE" },
|
|
context.signal,
|
|
),
|
|
),
|
|
);
|
|
}
|
|
|
|
private externalUserId(context: AdapterContext): string {
|
|
return `rkz_${createHmac("sha256", this.config.identitySecret)
|
|
.update(`${context.spaceId}:${context.userId}`)
|
|
.digest("hex")}`;
|
|
}
|
|
|
|
private mcpHeaders(
|
|
context: AdapterContext,
|
|
app: string,
|
|
accessToken: string,
|
|
): Record<string, string> {
|
|
return {
|
|
authorization: `Bearer ${accessToken}`,
|
|
"x-pd-project-id": this.config.projectId,
|
|
"x-pd-environment": this.config.environment,
|
|
"x-pd-external-user-id": this.externalUserId(context),
|
|
"x-pd-app-slug": app,
|
|
};
|
|
}
|
|
|
|
private async apps(): Promise<PipedreamApp[]> {
|
|
if (this.directory && this.directory.expiresAt > Date.now()) return this.directory.value;
|
|
if (this.directoryRequest) return this.directoryRequest;
|
|
this.directoryRequest = this.loadApps();
|
|
try {
|
|
return await this.directoryRequest;
|
|
} finally {
|
|
this.directoryRequest = undefined;
|
|
}
|
|
}
|
|
|
|
private async loadApps(): Promise<PipedreamApp[]> {
|
|
const apps: PipedreamApp[] = [];
|
|
let after: string | undefined;
|
|
for (let page = 0; page < 100; page += 1) {
|
|
const query = new URLSearchParams({ limit: "100", has_actions: "true" });
|
|
if (after) query.set("after", after);
|
|
const response = await this.request<{
|
|
data?: PipedreamApp[];
|
|
page_info?: { end_cursor?: string };
|
|
}>(`/v1/connect/apps?${query}`);
|
|
apps.push(...(response.data ?? []));
|
|
after = response.page_info?.end_cursor;
|
|
if (!after || apps.length >= 5_000) break;
|
|
}
|
|
this.directory = { value: apps, expiresAt: Date.now() + DIRECTORY_TTL_MS };
|
|
return apps;
|
|
}
|
|
|
|
private async accounts(context: AdapterContext, app?: string): Promise<PipedreamAccount[]> {
|
|
return collectPages(async (after) => {
|
|
const query = new URLSearchParams({
|
|
external_user_id: this.externalUserId(context),
|
|
limit: "100",
|
|
});
|
|
if (app) query.set("app", app);
|
|
if (after) query.set("after", after);
|
|
const response = await this.request<{
|
|
data?: PipedreamAccount[];
|
|
page_info?: { end_cursor?: string };
|
|
}>(
|
|
`/v1/connect/${encodeURIComponent(this.config.projectId)}/accounts?${query}`,
|
|
{},
|
|
context.signal,
|
|
);
|
|
return { items: response.data ?? [], cursor: response.page_info?.end_cursor };
|
|
}, 20);
|
|
}
|
|
|
|
private async request<T = unknown>(
|
|
path: string,
|
|
init: RequestInit = {},
|
|
signal?: AbortSignal,
|
|
): Promise<T> {
|
|
const token = await this.token();
|
|
const requestAbort = combineSignals(signal, AbortSignal.timeout(REQUEST_TIMEOUT_MS));
|
|
const response = await (this.dependencies.fetch ?? globalThis.fetch)(`${API_BASE}${path}`, {
|
|
...init,
|
|
headers: {
|
|
authorization: `Bearer ${token}`,
|
|
"content-type": "application/json",
|
|
"x-pd-environment": this.config.environment,
|
|
...init.headers,
|
|
},
|
|
signal: requestAbort,
|
|
});
|
|
if (response.status === 401) this.accessToken = undefined;
|
|
const body = await readPipedreamBody(response, requestAbort);
|
|
if (!response.ok) {
|
|
throw new Error(
|
|
sanitizeConnectorError(`Pipedream returned HTTP ${response.status}: ${body}`, [token]),
|
|
);
|
|
}
|
|
return (body ? JSON.parse(body) : {}) as T;
|
|
}
|
|
|
|
private async token(): Promise<string> {
|
|
if (this.accessToken && this.accessToken.expiresAt > Date.now() + 60_000) {
|
|
return this.accessToken.value;
|
|
}
|
|
if (this.accessTokenRequest) return this.accessTokenRequest;
|
|
this.accessTokenRequest = this.fetchToken();
|
|
try {
|
|
return await this.accessTokenRequest;
|
|
} finally {
|
|
this.accessTokenRequest = undefined;
|
|
}
|
|
}
|
|
|
|
private async fetchToken(): Promise<string> {
|
|
const requestAbort = AbortSignal.timeout(REQUEST_TIMEOUT_MS);
|
|
const response = await (this.dependencies.fetch ?? globalThis.fetch)(
|
|
`${API_BASE}/v1/oauth/token`,
|
|
{
|
|
method: "POST",
|
|
headers: { "content-type": "application/json" },
|
|
body: JSON.stringify({
|
|
grant_type: "client_credentials",
|
|
client_id: this.config.clientId,
|
|
client_secret: this.config.clientSecret,
|
|
scope: "connect:*",
|
|
}),
|
|
signal: requestAbort,
|
|
},
|
|
);
|
|
const body = JSON.parse(await readPipedreamBody(response, requestAbort)) as {
|
|
access_token?: string;
|
|
expires_in?: number;
|
|
error?: string;
|
|
};
|
|
if (!response.ok || !body.access_token) {
|
|
throw new Error(`Pipedream authentication failed: ${body.error ?? response.status}`);
|
|
}
|
|
this.accessToken = {
|
|
value: body.access_token,
|
|
expiresAt: Date.now() + (body.expires_in ?? 3_600) * 1_000,
|
|
};
|
|
return body.access_token;
|
|
}
|
|
}
|
|
|
|
async function readPipedreamBody(response: Response, signal: AbortSignal): Promise<string> {
|
|
const declared = Number(response.headers.get("content-length") ?? 0);
|
|
if (Number.isFinite(declared) && declared > MAX_PIPEDREAM_RESPONSE_BYTES) {
|
|
const cancel = response.body?.cancel() ?? Promise.resolve();
|
|
// A hanging cancel must not outlive the shared deadline.
|
|
await withAbort(
|
|
cancel.catch(() => undefined),
|
|
signal,
|
|
).catch(() => undefined);
|
|
throw new Error("Pipedream response is too large.");
|
|
}
|
|
try {
|
|
const bytes = await readBodyCapped(response, MAX_PIPEDREAM_RESPONSE_BYTES, signal);
|
|
return new TextDecoder().decode(bytes);
|
|
} catch (error) {
|
|
if (error instanceof Error && error.message === "Response is too large") {
|
|
throw new Error("Pipedream response is too large.");
|
|
}
|
|
throw error;
|
|
}
|
|
}
|