feat(connections): add instance relay runtime

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

server/src/services/connection-relay.test.ts

@@ -1,7 +1,7 @@
 import { randomBytes } from "node:crypto";
 import { describe, expect, it } from "vitest";
 import { createRelaySignature, type RelayEnvelope } from "@paperclip/connect-protocol";
-import { processConnectionRelay, type ConnectionRelayStore } from "./connection-relay.js";
+import { pollRelayChannel, processAndDispatchConnectionRelay, processConnectionRelay, type ConnectionRelayStore } from "./connection-relay.js";
 
 function fixture() {
   const envelope: RelayEnvelope = {
@@ -19,10 +19,12 @@ function fixture() {
   return { rawBody, relaySecret };
 }
 
-function store(): ConnectionRelayStore & { deliveries: Set<string> } {
+function store(): ConnectionRelayStore & { deliveries: Set<string>; statuses: string[] } {
   const deliveries = new Set<string>();
+  const statuses: string[] = [];
   return {
     deliveries,
+    statuses,
     async findConnectionByPublicRef(publicRef) {
       return publicRef === "cn_01K0EXAMPLE" ? { id: "connection-1", companyId: "company-1", enabled: true } : null;
     },
@@ -34,6 +36,9 @@ function store(): ConnectionRelayStore & { deliveries: Set<string> } {
     async listEnabledTriggers() {
       return [{ id: "trigger-1", destinationType: "routine", destinationId: "routine-1" }];
     },
+    async updateDelivery({ status }) {
+      statuses.push(status);
+    },
   };
 }
 
@@ -50,7 +55,7 @@ describe("processConnectionRelay", () => {
       timestamp: "1784657045",
       relaySecret,
       now: new Date("2026-07-21T18:04:05.000Z"),
-    })).rejects.toMatchObject({ code: "invalid_relay_signature", status: 401 });
+    })).rejects.toMatchObject({ code: "invalid_envelope", status: 401 });
     expect(relayStore.deliveries.size).toBe(0);
   });
 
@@ -72,4 +77,66 @@ describe("processConnectionRelay", () => {
     await expect(processConnectionRelay(relayStore, input)).resolves.toMatchObject({ status: "duplicate", triggers: [] });
     expect(relayStore.deliveries.size).toBe(1);
   });
+
+  it("dispatches a routine and records observable status transitions", async () => {
+    const relayStore = store();
+    const { rawBody, relaySecret } = fixture();
+    const fired: string[] = [];
+    const result = await processAndDispatchConnectionRelay(relayStore, {
+      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"),
+    });
+
+    expect(result.status).toBe("delivered");
+    expect(fired).toEqual(["routine-1"]);
+    expect(relayStore.deliveries).toEqual(new Set(["dl_01K0EXAMPLE"]));
+    expect(relayStore.statuses).toEqual(["forwarded", "delivered"]);
+  });
+
+  it("records a dead letter after the final failed attempt", async () => {
+    const relayStore = store();
+    const { rawBody: originalBody, relaySecret } = fixture();
+    const envelope = { ...JSON.parse(originalBody.toString("utf8")), attempt: 10 };
+    const rawBody = Buffer.from(JSON.stringify(envelope));
+    const result = await processAndDispatchConnectionRelay(relayStore, {
+      routine: async () => { throw new Error("routine unavailable"); },
+      issue_wake: async () => {},
+      plugin_worker: async () => {},
+    }, {
+      rawBody,
+      signature: createRelaySignature({ body: rawBody, relaySecret }),
+      timestamp: "1784657045",
+      relaySecret,
+      now: new Date("2026-07-21T18:04:05.000Z"),
+    });
+    expect(result.status).toBe("dead_letter");
+    expect(relayStore.statuses).toEqual(["forwarded", "dead_letter"]);
+  });
+
+  it("falls back to long-poll and forwards channel envelopes", async () => {
+    const { rawBody } = fixture();
+    const envelope = JSON.parse(rawBody.toString("utf8"));
+    const calls: string[] = [];
+    const received: string[] = [];
+    await pollRelayChannel({
+      baseUrl: "https://connect.example",
+      createSession: async () => "short-lived-token",
+      fetch: async (url) => {
+        calls.push(String(url));
+        return calls.length === 1
+          ? new Response(null, { status: 503 })
+          : Response.json([{ envelope, signature: "sha256=test", timestamp: "1784657045" }]);
+      },
+      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(received).toEqual(["dl_01K0EXAMPLE"]);
+  });
 });

server/src/services/connection-relay.ts

@@ -1,22 +1,64 @@
+import { and, eq } from "drizzle-orm";
+import type { Db } from "@paperclipai/db";
+import { connectionTriggerDeliveries, connectionTriggers, toolConnections } from "@paperclipai/db";
 import {
   ConnectProtocolError,
   relayEnvelopeSchema,
   verifyRelaySignature,
   type RelayEnvelope,
 } from "@paperclip/connect-protocol";
+import { secretService } from "./secrets.js";
 
 export type RelayTrigger = {
   id: string;
   destinationType: "routine" | "issue_wake" | "plugin_worker";
   destinationId: string;
+  config?: Record<string, unknown>;
 };
 
 export type ConnectionRelayStore = {
   findConnectionByPublicRef(publicRef: string): Promise<{ id: string; companyId: string; enabled: boolean } | null>;
   createDeliveryIfAbsent(input: { companyId: string; connectionId: string; envelope: RelayEnvelope }): Promise<boolean>;
   listEnabledTriggers(connectionId: string): Promise<RelayTrigger[]>;
+  updateDelivery?(input: { connectionId: string; deliveryId: string; status: "forwarded" | "delivered" | "failed" | "dead_letter"; error?: string | null; now?: Date }): Promise<void>;
 };
 
+type RelayConnectionConfig = { relay?: { publicRef?: string; secretId?: string; secretVersion?: string } };
+
+export function connectionRelayStore(db: Db): ConnectionRelayStore {
+  return {
+    async findConnectionByPublicRef(publicRef) {
+      const rows = await db.select({ id: toolConnections.id, companyId: toolConnections.companyId, enabled: toolConnections.enabled, config: toolConnections.config }).from(toolConnections);
+      return rows.find((row) => (row.config as RelayConnectionConfig).relay?.publicRef === publicRef) ?? null;
+    },
+    async createDeliveryIfAbsent({ companyId, connectionId, envelope }) {
+      const inserted = await db.insert(connectionTriggerDeliveries).values({ companyId, connectionId, deliveryId: envelope.deliveryId, providerSlug: envelope.providerSlug, attempt: envelope.attempt, envelope, receivedAt: new Date(envelope.receivedAt) }).onConflictDoNothing().returning({ id: connectionTriggerDeliveries.id });
+      return inserted.length === 1;
+    },
+    async listEnabledTriggers(connectionId) {
+      return db.select({ id: connectionTriggers.id, destinationType: connectionTriggers.destinationType, destinationId: connectionTriggers.destinationId, config: connectionTriggers.config }).from(connectionTriggers).where(and(eq(connectionTriggers.connectionId, connectionId), eq(connectionTriggers.enabled, true)));
+    },
+    async updateDelivery({ connectionId, deliveryId, status, error, now = new Date() }) {
+      await db.update(connectionTriggerDeliveries).set({ status, forwardedAt: status === "forwarded" ? now : undefined, deliveredAt: status === "delivered" ? now : undefined, lastError: error ?? null, updatedAt: now }).where(and(eq(connectionTriggerDeliveries.connectionId, connectionId), eq(connectionTriggerDeliveries.deliveryId, deliveryId)));
+    },
+  };
+}
+
+export function connectionRelaySecretResolver(db: Db) {
+  const secrets = secretService(db);
+  return async (publicRef: string): Promise<Buffer> => {
+    const rows = await db.select({ companyId: toolConnections.companyId, config: toolConnections.config }).from(toolConnections);
+    const connection = rows.find((row) => (row.config as RelayConnectionConfig).relay?.publicRef === publicRef);
+    const relay = connection && (connection.config as RelayConnectionConfig).relay;
+    if (!connection || !relay?.secretId) throw new Error("Relay connection or secret not found");
+    const parsedVersion = relay.secretVersion && /^\d+$/.test(relay.secretVersion) ? Number(relay.secretVersion) : "latest";
+    const value = await secrets.resolveSecretValue(connection.companyId, relay.secretId, parsedVersion);
+    return Buffer.from(value, "base64");
+  };
+}
+
+export type ConnectionRelayDispatcher = Record<RelayTrigger["destinationType"], (trigger: RelayTrigger, envelope: RelayEnvelope) => Promise<void>>;
+
 export async function processConnectionRelay(
   store: ConnectionRelayStore,
   input: {
@@ -29,28 +71,77 @@ export async function processConnectionRelay(
 ) {
   if (!input.signature || !input.timestamp || !verifyRelaySignature({
     body: input.rawBody,
-    relaySecret: input.relaySecret,
+    relaySecret: typeof input.relaySecret === "string" ? Buffer.from(input.relaySecret, "base64") : input.relaySecret,
     signature: input.signature,
     timestamp: input.timestamp,
     now: input.now,
   })) {
-    throw new ConnectProtocolError("invalid_relay_signature", 401);
+    throw new ConnectProtocolError("invalid_envelope", 401);
   }
 
   let envelope: RelayEnvelope;
   try {
     envelope = relayEnvelopeSchema.parse(JSON.parse(input.rawBody.toString("utf8")));
   } catch {
-    throw new ConnectProtocolError("invalid_relay_envelope", 400);
+    throw new ConnectProtocolError("invalid_envelope", 401);
   }
 
   const connection = await store.findConnectionByPublicRef(envelope.connectionPublicRef);
   if (!connection || !connection.enabled) {
-    throw new ConnectProtocolError("connection_not_found", 404);
+    throw new Error("Relay connection not found");
   }
 
   const inserted = await store.createDeliveryIfAbsent({ companyId: connection.companyId, connectionId: connection.id, envelope });
-  if (!inserted) return { status: "duplicate" as const, envelope, triggers: [] as RelayTrigger[] };
+  if (!inserted) return { status: "duplicate" as const, envelope, connection, triggers: [] as RelayTrigger[] };
 
-  return { status: "accepted" as const, envelope, triggers: await store.listEnabledTriggers(connection.id) };
+  return { status: "accepted" as const, envelope, connection, triggers: await store.listEnabledTriggers(connection.id) };
+}
+
+export async function processAndDispatchConnectionRelay(store: ConnectionRelayStore, dispatcher: ConnectionRelayDispatcher, input: Parameters<typeof processConnectionRelay>[1]) {
+  const result = await processConnectionRelay(store, input);
+  if (result.status === "duplicate") return result;
+  await store.updateDelivery?.({ connectionId: result.connection.id, deliveryId: result.envelope.deliveryId, status: "forwarded", now: input.now });
+  try {
+    for (const trigger of result.triggers) await dispatcher[trigger.destinationType](trigger, result.envelope);
+    await store.updateDelivery?.({ connectionId: result.connection.id, deliveryId: result.envelope.deliveryId, status: "delivered", now: input.now });
+    return { ...result, status: "delivered" as const };
+  } catch (error) {
+    const status = result.envelope.attempt >= 10 ? "dead_letter" : "failed";
+    await store.updateDelivery?.({ connectionId: result.connection.id, deliveryId: result.envelope.deliveryId, status, error: error instanceof Error ? error.message : "Relay destination dispatch failed", now: input.now });
+    return { ...result, status };
+  }
+}
+
+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 });
+  }
+}
+
+export function connectionTriggerService(db: Db) {
+  return {
+    list: (connectionId: string) => db.select().from(connectionTriggers).where(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 remove(id: string) {
+      return db.delete(connectionTriggers).where(eq(connectionTriggers.id, id)).returning({ id: connectionTriggers.id }).then((rows) => rows.length === 1);
+    },
+  };
 }