diff --git a/packages/db/src/migrations/0183_connection_trigger_relay.sql b/packages/db/src/migrations/0183_connection_trigger_relay.sql deleted file mode 100644 index 642b3aba52a..00000000000 --- a/packages/db/src/migrations/0183_connection_trigger_relay.sql +++ /dev/null @@ -1,50 +0,0 @@ -CREATE TABLE "connection_triggers" ( - "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, - "company_id" uuid NOT NULL, - "connection_id" uuid NOT NULL, - "destination_type" text NOT NULL, - "destination_id" text NOT NULL, - "enabled" boolean DEFAULT true NOT NULL, - "config" jsonb DEFAULT '{}'::jsonb NOT NULL, - "created_at" timestamp with time zone DEFAULT now() NOT NULL, - "updated_at" timestamp with time zone DEFAULT now() NOT NULL, - CONSTRAINT "connection_triggers_destination_type_check" CHECK ("connection_triggers"."destination_type" in ('routine', 'issue_wake', 'plugin_worker')) -); ---> statement-breakpoint -CREATE TABLE "connection_trigger_deliveries" ( - "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, - "company_id" uuid NOT NULL, - "connection_id" uuid NOT NULL, - "delivery_id" text NOT NULL, - "provider_slug" text NOT NULL, - "status" text DEFAULT 'received' NOT NULL, - "attempt" integer DEFAULT 1 NOT NULL, - "envelope" jsonb NOT NULL, - "received_at" timestamp with time zone NOT NULL, - "forwarded_at" timestamp with time zone, - "delivered_at" timestamp with time zone, - "last_error" text, - "created_at" timestamp with time zone DEFAULT now() NOT NULL, - "updated_at" timestamp with time zone DEFAULT now() NOT NULL, - CONSTRAINT "connection_trigger_deliveries_status_check" CHECK ("connection_trigger_deliveries"."status" in ('received', 'forwarded', 'delivered', 'failed', 'dead_letter')) -); ---> statement-breakpoint -ALTER TABLE "connection_triggers" ADD CONSTRAINT "connection_triggers_company_id_companies_id_fk" FOREIGN KEY ("company_id") REFERENCES "public"."companies"("id") ON DELETE cascade ON UPDATE no action; ---> statement-breakpoint -ALTER TABLE "connection_triggers" ADD CONSTRAINT "connection_triggers_connection_id_tool_connections_id_fk" FOREIGN KEY ("connection_id") REFERENCES "public"."tool_connections"("id") ON DELETE cascade ON UPDATE no action; ---> statement-breakpoint -ALTER TABLE "connection_trigger_deliveries" ADD CONSTRAINT "connection_trigger_deliveries_company_id_companies_id_fk" FOREIGN KEY ("company_id") REFERENCES "public"."companies"("id") ON DELETE cascade ON UPDATE no action; ---> statement-breakpoint -ALTER TABLE "connection_trigger_deliveries" ADD CONSTRAINT "connection_trigger_deliveries_connection_id_tool_connections_id_fk" FOREIGN KEY ("connection_id") REFERENCES "public"."tool_connections"("id") ON DELETE cascade ON UPDATE no action; ---> statement-breakpoint -CREATE INDEX "connection_triggers_company_idx" ON "connection_triggers" USING btree ("company_id"); ---> statement-breakpoint -CREATE INDEX "connection_triggers_connection_idx" ON "connection_triggers" USING btree ("connection_id"); ---> statement-breakpoint -CREATE UNIQUE INDEX "connection_triggers_destination_uq" ON "connection_triggers" USING btree ("connection_id","destination_type","destination_id"); ---> statement-breakpoint -CREATE INDEX "connection_trigger_deliveries_company_idx" ON "connection_trigger_deliveries" USING btree ("company_id"); ---> statement-breakpoint -CREATE INDEX "connection_trigger_deliveries_connection_status_idx" ON "connection_trigger_deliveries" USING btree ("connection_id","status"); ---> statement-breakpoint -CREATE UNIQUE INDEX "connection_trigger_deliveries_connection_delivery_uq" ON "connection_trigger_deliveries" USING btree ("connection_id","delivery_id"); diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index e76366fe0d9..a500af86e34 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -1795,6 +1795,8 @@ export { createToolStdioCommandTemplateSchema, disableToolStdioCommandTemplateSchema, updateToolConnectionSchema, + createConnectionTriggerSchema, + updateConnectionTriggerSchema, putToolConnectionInstallsSchema, updateToolMcpGatewaySchema, createToolInvocationSchema, diff --git a/packages/shared/src/validators/index.ts b/packages/shared/src/validators/index.ts index 38bb67c9a7a..1092fbc9b4d 100644 --- a/packages/shared/src/validators/index.ts +++ b/packages/shared/src/validators/index.ts @@ -811,6 +811,8 @@ export { createToolStdioCommandTemplateSchema, disableToolStdioCommandTemplateSchema, updateToolConnectionSchema, + createConnectionTriggerSchema, + updateConnectionTriggerSchema, putToolConnectionInstallsSchema, updateToolMcpGatewaySchema, createToolInvocationSchema, @@ -844,6 +846,8 @@ export { type CreateToolStdioCommandTemplate, type DisableToolStdioCommandTemplate, type UpdateToolConnection, + type CreateConnectionTrigger, + type UpdateConnectionTrigger, type PutToolConnectionInstalls, type UpdateToolMcpGateway, type CreateToolInvocation, diff --git a/packages/shared/src/validators/tool-access.ts b/packages/shared/src/validators/tool-access.ts index fc2f2f79641..5b8e8b78ebd 100644 --- a/packages/shared/src/validators/tool-access.ts +++ b/packages/shared/src/validators/tool-access.ts @@ -167,6 +167,24 @@ export const updateToolConnectionSchema = createToolConnectionSchema.omit({ appl export type UpdateToolConnection = z.infer; +export const connectionTriggerDestinationTypeSchema = z.enum(["routine", "issue_wake", "plugin_worker"]); + +export const createConnectionTriggerSchema = z.object({ + destinationType: connectionTriggerDestinationTypeSchema, + destinationId: z.string().uuid(), + enabled: z.boolean().optional(), + config: z.record(z.string(), z.unknown()).optional(), +}); + +export type CreateConnectionTrigger = z.infer; + +export const updateConnectionTriggerSchema = createConnectionTriggerSchema.partial().refine( + (value) => Object.keys(value).length > 0, + { message: "At least one connection trigger field is required" }, +); + +export type UpdateConnectionTrigger = z.infer; + export const connectionGrantSchema = z.object({ id: z.string().uuid(), companyId: z.string().uuid(), diff --git a/server/src/routes/tool-access.ts b/server/src/routes/tool-access.ts index 4469846a1e0..3c02c0d3c42 100644 --- a/server/src/routes/tool-access.ts +++ b/server/src/routes/tool-access.ts @@ -624,6 +624,84 @@ export function toolAccessRoutes( res.json(await svc.getConnectionUsage(connection.id, range, connection.companyId)); }); + router.get("/tool-connections/:connectionId/triggers", async (req, res) => { + assertBoard(req); + const connection = await svc.getConnection(req.params.connectionId as string); + if (!hasCompanyAccess(req, connection.companyId)) throw notFound("Tool connection not found"); + assertCompanyAccess(req, connection.companyId); + res.json({ connectionId: connection.id, triggers: await triggerSvc.list(connection.companyId, connection.id) }); + }); + + router.post( + "/tool-connections/:connectionId/triggers", + validate(createConnectionTriggerSchema), + async (req, res) => { + assertBoard(req); + const connection = await svc.getConnection(req.params.connectionId as string); + await assertBoardToolPermission(req, connection.companyId, "tools:manage_connections"); + try { + const trigger = await triggerSvc.create({ + companyId: connection.companyId, + connectionId: connection.id, + ...req.body, + }); + await logActivity(db, { + companyId: connection.companyId, + actorType: "user", + actorId: req.actor.userId ?? "board", + action: "connection_trigger.created", + entityType: "connection_trigger", + entityId: trigger.id, + details: { connectionId: connection.id, destinationType: trigger.destinationType, destinationId: trigger.destinationId }, + }); + res.status(201).json(trigger); + } catch (error) { + if (error instanceof Error && error.message === "A connection may have at most 3 triggers") throw unprocessable(error.message); + throw error; + } + }, + ); + + router.patch( + "/tool-connections/:connectionId/triggers/:triggerId", + validate(updateConnectionTriggerSchema), + async (req, res) => { + assertBoard(req); + const connection = await svc.getConnection(req.params.connectionId as string); + await assertBoardToolPermission(req, connection.companyId, "tools:manage_connections"); + const trigger = await triggerSvc.update(connection.companyId, connection.id, req.params.triggerId as string, req.body); + if (!trigger) throw notFound("Connection trigger not found"); + await logActivity(db, { + companyId: connection.companyId, + actorType: "user", + actorId: req.actor.userId ?? "board", + action: "connection_trigger.updated", + entityType: "connection_trigger", + entityId: trigger.id, + details: { connectionId: connection.id }, + }); + res.json(trigger); + }, + ); + + router.delete("/tool-connections/:connectionId/triggers/:triggerId", async (req, res) => { + assertBoard(req); + const connection = await svc.getConnection(req.params.connectionId as string); + await assertBoardToolPermission(req, connection.companyId, "tools:manage_connections"); + const removed = await triggerSvc.remove(connection.companyId, connection.id, req.params.triggerId as string); + if (!removed) throw notFound("Connection trigger not found"); + await logActivity(db, { + companyId: connection.companyId, + actorType: "user", + actorId: req.actor.userId ?? "board", + action: "connection_trigger.deleted", + entityType: "connection_trigger", + entityId: req.params.triggerId as string, + details: { connectionId: connection.id }, + }); + res.status(204).end(); + }); + router.get("/tool-connections/:connectionId/installs", async (req, res) => { assertBoard(req); const connection = await svc.getConnection(req.params.connectionId as string); diff --git a/server/src/services/connection-relay.test.ts b/server/src/services/connection-relay.test.ts index e72d52eea77..e83d97f2e22 100644 --- a/server/src/services/connection-relay.test.ts +++ b/server/src/services/connection-relay.test.ts @@ -1,7 +1,9 @@ import { randomBytes } from "node:crypto"; -import { describe, expect, it } from "vitest"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { createRelaySignature, type RelayEnvelope } from "@paperclip/connect-protocol"; -import { pollRelayChannel, processAndDispatchConnectionRelay, processConnectionRelay, type ConnectionRelayStore } from "./connection-relay.js"; +import { companies, connectionTriggerDeliveries, connectionTriggers, createDb, toolApplications, toolConnections } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; +import { connectionRelayStore, pollRelayChannel, processAndDispatchConnectionRelay, processConnectionRelay, type ConnectionRelayStore } from "./connection-relay.js"; function fixture() { const envelope: RelayEnvelope = { @@ -136,7 +138,82 @@ describe("processConnectionRelay", () => { }, onEnvelope: async ({ body }) => { received.push(JSON.parse(body.toString("utf8")).deliveryId); }, }); - expect(calls.map((url) => new URL(url).pathname)).toEqual(["/v1/relay/channel", "/v1/relay/poll"]); + expect(calls.map((url) => new URL(url).pathname)).toEqual(["/v1/relay/stream", "/v1/relay/poll"]); expect(received).toEqual(["dl_01K0EXAMPLE"]); }); + + it("consumes SSE data and acknowledges successful deliveries", async () => { + const { rawBody } = fixture(); + const envelope = JSON.parse(rawBody.toString("utf8")); + const acknowledged: string[][] = []; + const received: string[] = []; + await pollRelayChannel({ + baseUrl: "https://connect.example", + createSession: async () => ({ channelToken: "short-lived-token", streamUrl: "/v1/relay/stream" }), + fetch: async () => new Response(`id: ${envelope.deliveryId}\ndata: ${JSON.stringify({ envelope, signature: "v1=test", timestamp: "1784657045" })}\n\n`, { + headers: { "content-type": "text/event-stream" }, + }), + onEnvelope: async ({ body }) => { received.push(JSON.parse(body.toString("utf8")).deliveryId); }, + acknowledge: async (deliveryIds) => { acknowledged.push(deliveryIds); }, + }); + expect(received).toEqual(["dl_01K0EXAMPLE"]); + expect(acknowledged).toEqual([["dl_01K0EXAMPLE"]]); + }); +}); + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +describeEmbeddedPostgres("connection relay persistence", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-connection-relay-"); + db = createDb(tempDb.connectionString); + }, 20_000); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + it("persists an observable delivery row and fires one routine destination", async () => { + const company = await db.insert(companies).values({ name: "Relay Test", issuePrefix: "RLY" }).returning().then((rows) => rows[0]!); + const application = await db.insert(toolApplications).values({ companyId: company.id, name: "Vercel", type: "mcp_http" }).returning().then((rows) => rows[0]!); + const connection = await db.insert(toolConnections).values({ + companyId: company.id, + applicationId: application.id, + name: "Vercel Relay", + uid: "vercel-relay", + transport: "rest_api", + enabled: true, + config: { relay: { publicRef: "cn_01K0EXAMPLE" } }, + }).returning().then((rows) => rows[0]!); + await db.insert(connectionTriggers).values({ + companyId: company.id, + connectionId: connection.id, + destinationType: "routine", + destinationId: "11111111-1111-4111-8111-111111111111", + }); + + const { rawBody, relaySecret } = fixture(); + const fired: string[] = []; + const result = await processAndDispatchConnectionRelay(connectionRelayStore(db), { + routine: async (trigger) => { fired.push(trigger.destinationId); }, + issue_wake: async () => {}, + plugin_worker: async () => {}, + }, { + rawBody, + signature: createRelaySignature({ body: rawBody, relaySecret }), + timestamp: "1784657045", + relaySecret, + now: new Date("2026-07-21T18:04:05.000Z"), + }); + + const rows = await db.select().from(connectionTriggerDeliveries); + expect(result.status).toBe("delivered"); + expect(fired).toEqual(["11111111-1111-4111-8111-111111111111"]); + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ deliveryId: "dl_01K0EXAMPLE", status: "delivered", connectionId: connection.id }); + }); }); diff --git a/server/src/services/connection-relay.ts b/server/src/services/connection-relay.ts index 670a09edf67..023bf8c7309 100644 --- a/server/src/services/connection-relay.ts +++ b/server/src/services/connection-relay.ts @@ -1,6 +1,6 @@ import { and, eq } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; -import { connectionTriggerDeliveries, connectionTriggers, toolConnections } from "@paperclipai/db"; +import { connectionTriggerDeliveries, connectionTriggers, issues, toolConnections } from "@paperclipai/db"; import { ConnectProtocolError, relayEnvelopeSchema, @@ -8,6 +8,9 @@ import { type RelayEnvelope, } from "@paperclip/connect-protocol"; import { secretService } from "./secrets.js"; +import { routineService } from "./routines.js"; +import { heartbeatService } from "./heartbeat.js"; +import type { PluginWorkerManager } from "./plugin-worker-manager.js"; export type RelayTrigger = { id: string; @@ -59,6 +62,62 @@ export function connectionRelaySecretResolver(db: Db) { export type ConnectionRelayDispatcher = Record Promise>; +export function connectionRelayDispatcher( + db: Db, + options: { pluginWorkerManager?: PluginWorkerManager } = {}, +): ConnectionRelayDispatcher { + const routines = routineService(db, { pluginWorkerManager: options.pluginWorkerManager }); + const heartbeat = heartbeatService(db, { pluginWorkerManager: options.pluginWorkerManager }); + return { + routine: async (trigger, envelope) => { + await routines.runRoutine(trigger.destinationId, { + source: "api", + payload: { connectionRelay: envelope }, + idempotencyKey: `connection-relay:${envelope.deliveryId}:${trigger.id}`, + }); + }, + issue_wake: async (trigger, envelope) => { + const issue = await db + .select({ id: issues.id, assigneeAgentId: issues.assigneeAgentId }) + .from(issues) + .where(eq(issues.id, trigger.destinationId)) + .then((rows) => rows[0] ?? null); + if (!issue?.assigneeAgentId) throw new Error("Relay issue destination has no assigned agent"); + await heartbeat.wakeup(issue.assigneeAgentId, { + source: "automation", + triggerDetail: "system", + reason: "connection_trigger", + idempotencyKey: `connection-relay:${envelope.deliveryId}:${trigger.id}`, + requestedByActorType: "system", + contextSnapshot: { + issueId: issue.id, + taskId: issue.id, + wakeReason: "connection_trigger", + connectionRelay: envelope, + }, + }); + }, + plugin_worker: async (trigger, envelope) => { + if (!options.pluginWorkerManager) throw new Error("Plugin worker manager is unavailable"); + const endpointKey = typeof trigger.config?.endpointKey === "string" ? trigger.config.endpointKey : "connection-relay"; + const rawBody = Buffer.from(envelope.provider.bodyB64, "base64").toString("utf8"); + let parsedBody: unknown; + try { + parsedBody = JSON.parse(rawBody); + } catch { + parsedBody = undefined; + } + await options.pluginWorkerManager.call(trigger.destinationId, "handleWebhook", { + endpointKey, + headers: envelope.provider.headers, + rawBody, + parsedBody, + requestId: envelope.deliveryId, + }); + }, + }; +} + export async function processConnectionRelay( store: ConnectionRelayStore, input: { @@ -112,36 +171,80 @@ export async function processAndDispatchConnectionRelay(store: ConnectionRelaySt } } -export async function pollRelayChannel(options: { baseUrl: string; createSession: () => Promise; onEnvelope: (input: { body: Buffer; signature: string; timestamp: string }) => Promise; fetch?: typeof globalThis.fetch }) { - const fetcher = options.fetch ?? globalThis.fetch; - const channelToken = await options.createSession(); - const headers = { authorization: `Bearer ${channelToken}`, accept: "text/event-stream, application/json" }; - let response = await fetcher(new URL("/v1/relay/channel", options.baseUrl), { headers }); - if (!response.ok || !response.body) response = await fetcher(new URL("/v1/relay/poll", options.baseUrl), { headers }); - if (!response.ok) throw new Error(`Relay channel failed (${response.status})`); - const payload = await response.json() as unknown; - for (const raw of Array.isArray(payload) ? payload : [payload]) { - if (!raw || typeof raw !== "object") continue; - const item = raw as { envelope?: unknown; signature?: unknown; timestamp?: unknown }; - if (typeof item.signature !== "string" || typeof item.timestamp !== "string") continue; - const envelope = relayEnvelopeSchema.parse(item.envelope); - await options.onEnvelope({ body: Buffer.from(JSON.stringify(envelope)), signature: item.signature, timestamp: item.timestamp }); +type RelayChannelItem = { envelope: RelayEnvelope; signature: string; timestamp: string }; + +function parseRelayChannelItem(raw: unknown): RelayChannelItem | null { + if (!raw || typeof raw !== "object") return null; + const item = raw as { envelope?: unknown; signature?: unknown; timestamp?: unknown }; + if (typeof item.signature !== "string" || typeof item.timestamp !== "string") return null; + return { envelope: relayEnvelopeSchema.parse(item.envelope), signature: item.signature, timestamp: item.timestamp }; +} + +async function readRelaySse(response: Response): Promise { + const text = await response.text(); + const items: RelayChannelItem[] = []; + for (const event of text.split(/\r?\n\r?\n/)) { + const data = event + .split(/\r?\n/) + .filter((line) => line.startsWith("data:")) + .map((line) => line.slice(5).trimStart()) + .join("\n"); + if (!data) continue; + const item = parseRelayChannelItem(JSON.parse(data)); + if (item) items.push(item); } + return items; +} + +export async function pollRelayChannel(options: { + baseUrl: string; + createSession: () => Promise; + onEnvelope: (input: { body: Buffer; signature: string; timestamp: string }) => Promise; + acknowledge?: (deliveryIds: string[]) => Promise; + lastEventId?: string; + fetch?: typeof globalThis.fetch; +}) { + const fetcher = options.fetch ?? globalThis.fetch; + const session = await options.createSession(); + const channelToken = typeof session === "string" ? session : session.channelToken; + const streamUrl = typeof session === "string" ? "/v1/relay/stream" : session.streamUrl; + const headers: Record = { authorization: `Bearer ${channelToken}`, accept: "text/event-stream, application/json" }; + if (options.lastEventId) headers["last-event-id"] = options.lastEventId; + let response = await fetcher(new URL(streamUrl, options.baseUrl), { headers }); + let items: RelayChannelItem[] = []; + if (response.ok && response.body && response.headers.get("content-type")?.includes("text/event-stream")) { + items = await readRelaySse(response); + } else { + response = await fetcher(new URL("/v1/relay/poll?waitSeconds=25", options.baseUrl), { headers }); + if (response.ok) { + const payload = await response.json() as unknown; + items = (Array.isArray(payload) ? payload : [payload]) + .map(parseRelayChannelItem) + .filter((item): item is RelayChannelItem => item !== null); + } + } + if (!response.ok) throw new Error(`Relay channel failed (${response.status})`); + const acknowledged: string[] = []; + for (const item of items) { + await options.onEnvelope({ body: Buffer.from(JSON.stringify(item.envelope)), signature: item.signature, timestamp: item.timestamp }); + acknowledged.push(item.envelope.deliveryId); + } + if (acknowledged.length > 0) await options.acknowledge?.(acknowledged); } export function connectionTriggerService(db: Db) { return { - list: (connectionId: string) => db.select().from(connectionTriggers).where(eq(connectionTriggers.connectionId, connectionId)), + list: (companyId: string, connectionId: string) => db.select().from(connectionTriggers).where(and(eq(connectionTriggers.companyId, companyId), eq(connectionTriggers.connectionId, connectionId))), async create(input: { companyId: string; connectionId: string; destinationType: RelayTrigger["destinationType"]; destinationId: string; enabled?: boolean; config?: Record }) { const existing = await db.select({ id: connectionTriggers.id }).from(connectionTriggers).where(eq(connectionTriggers.connectionId, input.connectionId)); if (existing.length >= 3) throw new Error("A connection may have at most 3 triggers"); return db.insert(connectionTriggers).values({ ...input, enabled: input.enabled ?? true, config: input.config ?? {} }).returning().then((rows) => rows[0]); }, - async update(id: string, patch: { destinationType?: RelayTrigger["destinationType"]; destinationId?: string; enabled?: boolean; config?: Record }) { - return db.update(connectionTriggers).set({ ...patch, updatedAt: new Date() }).where(eq(connectionTriggers.id, id)).returning().then((rows) => rows[0] ?? null); + async update(companyId: string, connectionId: string, id: string, patch: { destinationType?: RelayTrigger["destinationType"]; destinationId?: string; enabled?: boolean; config?: Record }) { + return db.update(connectionTriggers).set({ ...patch, updatedAt: new Date() }).where(and(eq(connectionTriggers.companyId, companyId), eq(connectionTriggers.connectionId, connectionId), eq(connectionTriggers.id, id))).returning().then((rows) => rows[0] ?? null); }, - async remove(id: string) { - return db.delete(connectionTriggers).where(eq(connectionTriggers.id, id)).returning({ id: connectionTriggers.id }).then((rows) => rows.length === 1); + async remove(companyId: string, connectionId: string, id: string) { + return db.delete(connectionTriggers).where(and(eq(connectionTriggers.companyId, companyId), eq(connectionTriggers.connectionId, connectionId), eq(connectionTriggers.id, id))).returning({ id: connectionTriggers.id }).then((rows) => rows.length === 1); }, }; }