feat(relay): forward webhook requests to the environment's tunnel

docs/internals/t3-connect.md

@@ -4,6 +4,12 @@ T3 Connect uses Clerk for cloud identity. The relay manages environment links,
 credentials for reaching environments, and managed tunnel allocations. After
 bootstrap, clients send application traffic through the environment's tunnel
 hostname; the relay Worker does not proxy their HTTP or WebSocket sessions.
+The one exception is automation webhooks: the relay forwards
+`/v1/hooks/:environmentId/:hookId/:token` statelessly to the environment's
+tunnel so senders get a stable URL. It stores nothing, keeps bodies and tokens
+out of its traces, and leaves token and signature verification to the
+environment
+([forwarder](../../infra/relay/src/hooks/HookForwarder.ts)).
 
 Clerk, deployment, and native authentication setup live in the
 [Connect setup runbook](../operations/connect-setup.md).

infra/relay/src/agentActivity/AgentActivityPublisher.test.ts

@@ -94,6 +94,7 @@ function makeEnvironmentLinks(
       ]),
     listForUser: () => Effect.succeed([]),
     getForUser: () => Effect.succeed(null),
+    findActiveManagedForEnvironment: () => Effect.succeed([]),
     revokeForUser: () => Effect.succeed(false),
     ...overrides,
   };

infra/relay/src/agentActivity/FcmDeliveries.test.ts

@@ -171,6 +171,7 @@ function harness() {
             : [],
         ),
       listForUser: () => Effect.succeed([]),
+      findActiveManagedForEnvironment: () => Effect.succeed([]),
       revokeForUser: () => Effect.succeed(false),
       getForUser: (input) =>
         Effect.sync(() =>

infra/relay/src/agentActivity/MobileRegistrations.test.ts

@@ -116,6 +116,7 @@ function makeEnvironmentLinks(
       ]),
     listForUser: () => Effect.succeed([]),
     getForUser: () => Effect.succeed(null),
+    findActiveManagedForEnvironment: () => Effect.succeed([]),
     revokeForUser: () => Effect.succeed(false),
     ...overrides,
   };

infra/relay/src/environments/EnvironmentConnector.test.ts

@@ -229,6 +229,7 @@ function makeLinks(
         environmentPublicKey: environmentKeyPair.publicKey,
         ...overrides,
       }),
+    findActiveManagedForEnvironment: () => Effect.succeed([]),
     revokeForUser: () => Effect.succeed(false),
   };
 }

infra/relay/src/environments/EnvironmentConnector.ts

@@ -16,6 +16,7 @@ import {
   RelayEnvironmentConnectNotAuthorizedReason,
   type RelayEnvironmentConnectResponse,
   type RelayEnvironmentStatusResponse,
+  type RelayManagedEndpoint,
 } from "@t3tools/contracts/relay";
 import {
   normalizeRelayIssuer,
@@ -35,6 +36,7 @@ import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 import * as Option from "effect/Option";
 import * as Redacted from "effect/Redacted";
+import * as Result from "effect/Result";
 import * as Schema from "effect/Schema";
 import * as FetchHttpClient from "effect/http/FetchHttpClient";
 import * as HttpClient from "effect/http/HttpClient";
@@ -185,7 +187,7 @@ const currentTraceId = Effect.currentSpan.pipe(
   Effect.orElseSucceed(() => "unavailable"),
 );
 
-const withoutRedirects = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
+export const withoutRedirects = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
   effect.pipe(Effect.provideService(FetchHttpClient.RequestInit, { redirect: "manual" }));
 
 const verifyWithEnvironmentKeys = Effect.fnUntraced(function* <A, E>(input: {
@@ -288,6 +290,92 @@ function verifyEnvironmentHealthResponse(input: {
   );
 }
 
+export interface ManagedEndpointValidationFailure {
+  readonly reason: Exclude<
+    RelayEnvironmentConnectNotAuthorizedReason,
+    "client_proof_key_thumbprint_missing" | "environment_link_not_found"
+  >;
+  /** Diagnostic span attributes; never contains secrets. */
+  readonly attributes?: Record<string, string | boolean>;
+}
+
+/**
+ * Checks that a link's relay-managed endpoint is backed by a ready allocation
+ * on the configured base domain and still matches what the link recorded.
+ */
+export function validateManagedEndpoint(input: {
+  readonly link: EnvironmentLinks.RelayLinkedEnvironmentRecord;
+  readonly allocation: ManagedEndpointAllocations.ManagedEndpointAllocation | null;
+  readonly baseDomain: string | undefined;
+}): Result.Result<RelayManagedEndpoint, ManagedEndpointValidationFailure> {
+  const { link, allocation, baseDomain } = input;
+  if (link.endpoint.providerKind !== "cloudflare_tunnel") {
+    return Result.fail({
+      reason: "endpoint_provider_not_managed",
+      attributes: { "relay.authorization.endpoint_provider_kind": link.endpoint.providerKind },
+    });
+  }
+  if (!allocation) {
+    return Result.fail({ reason: "managed_endpoint_allocation_not_found" });
+  }
+  const allocationAttributes = {
+    "relay.authorization.allocation_hostname": allocation.hostname,
+    "relay.authorization.allocation_has_ready_at": allocation.readyAt !== null,
+    "relay.authorization.allocation_has_tunnel_id": allocation.tunnelId !== null,
+    "relay.authorization.allocation_has_dns_record_id": allocation.dnsRecordId !== null,
+  };
+  if (!baseDomain) {
+    return Result.fail({
+      reason: "managed_endpoint_base_domain_not_configured",
+      attributes: allocationAttributes,
+    });
+  }
+  if (
+    allocation.readyAt === null ||
+    allocation.tunnelId === null ||
+    allocation.dnsRecordId === null
+  ) {
+    return Result.fail({
+      reason: "managed_endpoint_allocation_not_ready",
+      attributes: allocationAttributes,
+    });
+  }
+  if (!isManagedEndpointHostname(allocation.hostname, baseDomain)) {
+    return Result.fail({
+      reason: "managed_endpoint_hostname_invalid",
+      attributes: {
+        ...allocationAttributes,
+        "relay.authorization.managed_endpoint_base_domain": baseDomain,
+      },
+    });
+  }
+  const endpoint = ManagedEndpointAllocations.resolveReadyManagedEndpoint({
+    allocation,
+    baseDomain,
+  });
+  if (
+    endpoint === null ||
+    endpoint.httpBaseUrl !== link.endpoint.httpBaseUrl ||
+    endpoint.wsBaseUrl !== link.endpoint.wsBaseUrl
+  ) {
+    return Result.fail({
+      reason: "managed_endpoint_mismatch",
+      attributes: {
+        ...allocationAttributes,
+        "relay.authorization.linked_http_base_url": link.endpoint.httpBaseUrl,
+        "relay.authorization.linked_ws_base_url": link.endpoint.wsBaseUrl,
+        ...(endpoint
+          ? {
+              "relay.authorization.resolved_http_base_url": endpoint.httpBaseUrl,
+              "relay.authorization.resolved_ws_base_url": endpoint.wsBaseUrl,
+            }
+          : {}),
+      },
+    });
+  }
+  return Result.succeed(endpoint);
+}
+
 const make = Effect.gen(function* () {
   const links = yield* EnvironmentLinks.EnvironmentLinks;
   const allocations = yield* ManagedEndpointAllocations.ManagedEndpointAllocations;
@@ -305,89 +393,22 @@ const make = Effect.gen(function* () {
       readonly link: EnvironmentLinks.RelayLinkedEnvironmentRecord;
       readonly allocation: ManagedEndpointAllocations.ManagedEndpointAllocation | null;
     }) {
-      if (input.link.endpoint.providerKind !== "cloudflare_tunnel") {
-        yield* Effect.annotateCurrentSpan({
-          "relay.authorization.endpoint_provider_kind": input.link.endpoint.providerKind,
-        });
-        return yield* new EnvironmentConnectNotAuthorized({
-          environmentId: input.link.environmentId,
-          operation: input.operation,
-          reason: "endpoint_provider_not_managed",
-        });
-      }
-      if (!input.allocation) {
-        return yield* new EnvironmentConnectNotAuthorized({
-          environmentId: input.link.environmentId,
-          operation: input.operation,
-          reason: "managed_endpoint_allocation_not_found",
-        });
-      }
-      const allocationAttributes = {
-        "relay.authorization.allocation_hostname": input.allocation.hostname,
-        "relay.authorization.allocation_has_ready_at": input.allocation.readyAt !== null,
-        "relay.authorization.allocation_has_tunnel_id": input.allocation.tunnelId !== null,
-        "relay.authorization.allocation_has_dns_record_id": input.allocation.dnsRecordId !== null,
-      } as const;
-      if (!settings.managedEndpointBaseDomain) {
-        yield* Effect.annotateCurrentSpan(allocationAttributes);
-        return yield* new EnvironmentConnectNotAuthorized({
-          environmentId: input.link.environmentId,
-          operation: input.operation,
-          reason: "managed_endpoint_base_domain_not_configured",
-        });
-      }
-      if (
-        input.allocation.readyAt === null ||
-        input.allocation.tunnelId === null ||
-        input.allocation.dnsRecordId === null
-      ) {
-        yield* Effect.annotateCurrentSpan(allocationAttributes);
-        return yield* new EnvironmentConnectNotAuthorized({
-          environmentId: input.link.environmentId,
-          operation: input.operation,
-          reason: "managed_endpoint_allocation_not_ready",
-        });
-      }
-      if (
-        !isManagedEndpointHostname(input.allocation.hostname, settings.managedEndpointBaseDomain)
-      ) {
-        yield* Effect.annotateCurrentSpan({
-          ...allocationAttributes,
-          "relay.authorization.managed_endpoint_base_domain": settings.managedEndpointBaseDomain,
-        });
-        return yield* new EnvironmentConnectNotAuthorized({
-          environmentId: input.link.environmentId,
-          operation: input.operation,
-          reason: "managed_endpoint_hostname_invalid",
-        });
-      }
-      const endpoint = ManagedEndpointAllocations.resolveReadyManagedEndpoint({
+      const result = validateManagedEndpoint({
+        link: input.link,
         allocation: input.allocation,
         baseDomain: settings.managedEndpointBaseDomain,
       });
-      if (
-        endpoint === null ||
-        endpoint.httpBaseUrl !== input.link.endpoint.httpBaseUrl ||
-        endpoint.wsBaseUrl !== input.link.endpoint.wsBaseUrl
-      ) {
-        yield* Effect.annotateCurrentSpan({
-          ...allocationAttributes,
-          "relay.authorization.linked_http_base_url": input.link.endpoint.httpBaseUrl,
-          "relay.authorization.linked_ws_base_url": input.link.endpoint.wsBaseUrl,
-          ...(endpoint
-            ? {
-                "relay.authorization.resolved_http_base_url": endpoint.httpBaseUrl,
-                "relay.authorization.resolved_ws_base_url": endpoint.wsBaseUrl,
-              }
-            : {}),
-        });
-        return yield* new EnvironmentConnectNotAuthorized({
-          environmentId: input.link.environmentId,
-          operation: input.operation,
-          reason: "managed_endpoint_mismatch",
-        });
+      if (Result.isSuccess(result)) {
+        return result.success;
       }
-      return endpoint;
+      if (result.failure.attributes) {
+        yield* Effect.annotateCurrentSpan(result.failure.attributes);
+      }
+      return yield* new EnvironmentConnectNotAuthorized({
+        environmentId: input.link.environmentId,
+        operation: input.operation,
+        reason: result.failure.reason,
+      });
     },
   );
 

infra/relay/src/environments/EnvironmentLinker.test.ts

@@ -126,6 +126,7 @@ function testLayer(input?: {
           listDeliveryUsersForEnvironment: () => Effect.succeed([]),
           listForUser: () => Effect.succeed([]),
           getForUser: () => Effect.succeed(null),
+          findActiveManagedForEnvironment: () => Effect.succeed([]),
           revokeForUser: () => Effect.succeed(false),
         }),
         Layer.succeed(EnvironmentCredentials.EnvironmentCredentials, {

infra/relay/src/environments/EnvironmentLinks.ts

@@ -75,6 +75,18 @@ export class EnvironmentLinkLookupPersistenceError extends Schema.TaggedError<En
   }
 }
 
+export class EnvironmentLinkEnvironmentLookupPersistenceError extends Schema.TaggedError<EnvironmentLinkEnvironmentLookupPersistenceError>()(
+  "EnvironmentLinkEnvironmentLookupPersistenceError",
+  {
+    environmentId: Schema.String,
+    cause: Schema.Defect(),
+  },
+) {
+  override get message(): string {
+    return `Failed to look up active managed links for environment '${this.environmentId}'`;
+  }
+}
+
 export class EnvironmentLinkRevokePersistenceError extends Schema.TaggedError<EnvironmentLinkRevokePersistenceError>()(
   "EnvironmentLinkRevokePersistenceError",
   {
@@ -114,6 +126,13 @@ export class EnvironmentLinks extends Context.Service<
       readonly userId: string;
       readonly environmentId: string;
     }) => Effect.Effect<RelayLinkedEnvironmentRecord | null, EnvironmentLinkLookupPersistenceError>;
+    /** Active relay-managed links for an environment, across all users (webhook forwarding). */
+    readonly findActiveManagedForEnvironment: (input: {
+      readonly environmentId: string;
+    }) => Effect.Effect<
+      ReadonlyArray<RelayLinkedEnvironmentRecord & { readonly userId: string }>,
+      EnvironmentLinkEnvironmentLookupPersistenceError
+    >;
     readonly revokeForUser: (input: {
       readonly userId: string;
       readonly environmentId: string;
@@ -322,6 +341,59 @@ const make = Effect.gen(function* () {
         );
     }),
 
+    findActiveManagedForEnvironment: Effect.fn(
+      "relay.environment_links.find_active_managed_for_environment",
+    )(function* (input) {
+      yield* Effect.annotateCurrentSpan({ "relay.environment_id": input.environmentId });
+      return yield* db
+        .select({
+          userId: relayEnvironmentLinks.userId,
+          environmentId: relayEnvironmentLinks.environmentId,
+          environmentLabel: relayEnvironmentLinks.environmentLabel,
+          environmentPublicKey: relayEnvironmentLinks.environmentPublicKey,
+          endpointHttpBaseUrl: relayEnvironmentLinks.endpointHttpBaseUrl,
+          endpointWsBaseUrl: relayEnvironmentLinks.endpointWsBaseUrl,
+          endpointProviderKind: relayEnvironmentLinks.endpointProviderKind,
+          createdAt: relayEnvironmentLinks.createdAt,
+        })
+        .from(relayEnvironmentLinks)
+        .where(
+          and(
+            eq(relayEnvironmentLinks.environmentId, input.environmentId),
+            isNull(relayEnvironmentLinks.revokedAt),
+            eq(relayEnvironmentLinks.endpointProviderKind, "cloudflare_tunnel"),
+            eq(relayEnvironmentLinks.managedTunnelsEnabled, true),
+          ),
+        )
+        // One row per user who linked this environment; every row is checked
+        // until one has a ready endpoint, so none may be cut off.
+        .pipe(
+          Effect.map((rows) =>
+            rows.map((row) => ({
+              userId: row.userId,
+              environmentId: row.environmentId as RelayClientEnvironmentRecord["environmentId"],
+              label:
+                row.environmentLabel.trim().length > 0 ? row.environmentLabel : row.environmentId,
+              endpoint: {
+                httpBaseUrl: row.endpointHttpBaseUrl,
+                wsBaseUrl: row.endpointWsBaseUrl,
+                providerKind:
+                  row.endpointProviderKind as RelayClientEnvironmentRecord["endpoint"]["providerKind"],
+              },
+              environmentPublicKey: row.environmentPublicKey,
+              linkedAt: row.createdAt,
+            })),
+          ),
+          Effect.mapError(
+            (cause) =>
+              new EnvironmentLinkEnvironmentLookupPersistenceError({
+                environmentId: input.environmentId,
+                cause,
+              }),
+          ),
+        );
+    }),
+
     revokeForUser: Effect.fn("relay.environment_links.revoke_for_user")(function* (input) {
       yield* Effect.annotateCurrentSpan({
         "relay.environment_id": input.environmentId,