feat(connections): finish instance relay dispatch

Co-Authored-By: Paperclip <noreply@paperclip.ing>

packages/db/src/migrations/0183_connection_trigger_relay.sql

@@ -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");

packages/shared/src/index.ts

@@ -1795,6 +1795,8 @@ export {
   createToolStdioCommandTemplateSchema,
   disableToolStdioCommandTemplateSchema,
   updateToolConnectionSchema,
+  createConnectionTriggerSchema,
+  updateConnectionTriggerSchema,
   putToolConnectionInstallsSchema,
   updateToolMcpGatewaySchema,
   createToolInvocationSchema,

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,

packages/shared/src/validators/tool-access.ts

@@ -167,6 +167,24 @@ export const updateToolConnectionSchema = createToolConnectionSchema.omit({ appl
 
 export type UpdateToolConnection = z.infer<typeof updateToolConnectionSchema>;
 
+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<typeof createConnectionTriggerSchema>;
+
+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<typeof updateConnectionTriggerSchema>;
+
 export const connectionGrantSchema = z.object({
   id: z.string().uuid(),
   companyId: z.string().uuid(),

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);

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<typeof createDb>;
+  let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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 });
+  });
 });

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<RelayTrigger["destinationType"], (trigger: RelayTrigger, envelope: RelayEnvelope) => Promise<void>>;
 
+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<string>; onEnvelope: (input: { body: Buffer; signature: string; timestamp: string }) => Promise<void>; 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<RelayChannelItem[]> {
+  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<string | { channelToken: string; streamUrl: string }>;
+  onEnvelope: (input: { body: Buffer; signature: string; timestamp: string }) => Promise<void>;
+  acknowledge?: (deliveryIds: string[]) => Promise<void>;
+  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<string, string> = { 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<string, unknown> }) {
       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<string, unknown> }) {
-      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<string, unknown> }) {
+      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);
     },
   };
 }