1import * as Clock from "effect/Clock";
2import * as Context from "effect/Context";
3import * as Crypto from "effect/Crypto";
4import * as DateTime from "effect/DateTime";
5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";
7import * as Option from "effect/Option";
8import * as Redacted from "effect/Redacted";
9import * as Result from "effect/Result";
10import * as Schema from "effect/Schema";
11import * as Stream from "effect/Stream";
12import * as HttpClient from "effect/http/HttpClient";
13import type * as HttpServerRequest from "effect/http/HttpServerRequest";
14import * as HttpServerResponse from "effect/http/HttpServerResponse";
15import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
16import { EnvironmentId } from "@t3tools/contracts";
17import { RelayApi, type RelayHookDeliveryProofPayload } from "@t3tools/contracts/relay";
18import {
19 normalizeRelayIssuer,
20 RELAY_HOOK_DELIVERY_HEADER,
21 RELAY_HOOK_DELIVERY_TYP,
22 signRelayJwt,
23} from "@t3tools/shared/relayJwt";
24
25import * as RelayConfiguration from "../Config.ts";
26import { MANAGED_ENDPOINT_KEY_PATTERN } from "../deploymentConfig.ts";
27import * as HeldHooks from "./HeldHooks.ts";
28import * as HookInbox from "./HookInbox.ts";
29import { sendUpstream, TUNNEL_OFFLINE_STATUS } from "./upstream.ts";
30
31export const RELAY_HOOK_PATH_PREFIX = "/v1/hooks/";
32export const RELAY_HOOK_MAX_BODY_BYTES = 1_048_576;
33export const RELAY_HOOK_RATE_LIMIT = { limit: 60, periodSeconds: 60 } as const;
34/**
35 * Hook budgets are per URL, and a sender who knows an endpoint key can mint
36 * new URLs for free, so every endpoint also has one overall budget.
37 */
38export const RELAY_HOOK_ENDPOINT_RATE_LIMIT = { limit: 600, periodSeconds: 60 } as const;
39/**
40 * Upstream statuses that mean the environment did not take the request: the
41 * tunnel has no origin (530) or cloudflared cannot reach the local server
42 * (502, 503, 504) while it restarts.
43 */
44export const ENVIRONMENT_UNREACHABLE_STATUSES: ReadonlySet<number> = new Set([
45 502,
46 503,
47 504,
48 TUNNEL_OFFLINE_STATUS,
49]);
50
51const DROPPED_REQUEST_HEADERS = new Set([
52 "host",
53 "connection",
54 "keep-alive",
55 "transfer-encoding",
56 "te",
57 "upgrade",
58 "content-length",
59 "cookie",
60 "x-real-ip",
61 // Only the relay may set this; a sender could otherwise collide delivery ids.
62 "x-t3-relay-delivery-id",
63 "x-t3-relay-received-at",
64 RELAY_HOOK_DELIVERY_HEADER,
65 // The environment trusts trace context only from the relay, which sets its own.
66 "traceparent",
67 "tracestate",
68 "b3",
69]);
70const DROPPED_REQUEST_HEADER_PREFIXES = ["proxy-", "cf-", "x-forwarded-", "x-b3-"];
71
72export const isRelayHookPath = (url: string): boolean => url.startsWith(RELAY_HOOK_PATH_PREFIX);
73
74/** Replaces the hook token (and any query) so traces and logs never record the secret. */
75export const redactRelayHookUrl = (url: string): string => {
76 const path = url.split("?", 1)[0] ?? url;
77 const segments = path.split("/");
78 if (segments.length >= 6) {
79 // URL-safe, so span attributes built from the parsed URL keep it readable.
80 segments[5] = "redacted";
81 }
82 return segments.join("/");
83};
84
85/**
86 * Request budget for public hook forwarding, keyed by a hash of the hook URL
87 * (endpoint, hook and token). Requests with a wrong token get their own
88 * budget, so they cannot use up a real sender's; the environment rejects them.
89 * Built from decoded segments, because the environment decodes them too: two
90 * spellings of one token (`token`, `%74oken`) must share one budget.
91 */
92const hookBudgetKey = (hook: {
93 readonly endpointKey: string;
94 readonly hookId: string;
95 readonly token: string;
96}) =>
97 Effect.promise(() =>
98 crypto.subtle.digest(
99 "SHA-256",
100 // Length-prefixed, so no segment contents can make two keys collide.
101 new TextEncoder().encode(
102 [hook.endpointKey, hook.hookId, hook.token]
103 .map((part) => `${part.length}:${part}`)
104 .join(""),
105 ),
106 ),
107 ).pipe(
108 Effect.map((digest) =>
109 Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join(""),
110 ),
111 );
112
113export class HookRateLimiter extends Context.Service<
114 HookRateLimiter,
115 {
116 /** One hook URL's budget, keyed by `hookBudgetKey`. */
117 readonly allowHook: (key: string) => Effect.Effect<boolean>;
118 /** One endpoint's overall budget, keyed by its endpoint key. */
119 readonly allowEndpoint: (endpointKey: string) => Effect.Effect<boolean>;
120 }
121>()("t3code-relay/hooks/HookForwarder/HookRateLimiter") {}
122
123export class HookForwarder extends Context.Service<
124 HookForwarder,
125 {
126 readonly handle: (
127 request: HttpServerRequest.HttpServerRequest,
128 ) => Effect.Effect<HttpServerResponse.HttpServerResponse>;
129 }
130>()("t3code-relay/hooks/HookForwarder") {}
131
132class HookBodyTooLarge extends Schema.TaggedError<HookBodyTooLarge>()("HookBodyTooLarge", {}) {}
133
134const errorResponse = (status: number, error: string, headers?: Record<string, string>) =>
135 HttpServerResponse.jsonUnsafe({ error }, { status, ...(headers ? { headers } : {}) });
136
137const hookNotFound = () => errorResponse(404, "hook_not_found");
138
139/** Longer than the inbox holds a request, so a held delivery's proof still verifies. */
140const DELIVERY_PROOF_LIFETIME_SECONDS = 25 * 60 * 60;
141
142/** Methods a webhook can arrive with; HEAD reaches the GET route and is refused. */
143const FORWARDED_METHODS = new Set(["GET", "POST", "PUT", "PATCH"]);
144
145/**
146 * Whatever the environment answers is served from the relay's own origin, so
147 * a body must never render or run there.
148 */
149const SANDBOXED_RESPONSE_HEADERS = {
150 "x-content-type-options": "nosniff",
151 "content-security-policy": "sandbox; default-src 'none'",
152} as const;
153
154function parseHookPath(url: string) {
155 const queryIndex = url.indexOf("?");
156 const path = queryIndex === -1 ? url : url.slice(0, queryIndex);
157 const search = queryIndex === -1 ? "" : url.slice(queryIndex);
158 const segments = path.split("/");
159 // ["", "v1", "hooks", endpointKey, hookId, token]
160 if (segments.length !== 6) return null;
161 const [, , , endpointKey, rawHookId, rawToken] = segments;
162 if (!endpointKey || !MANAGED_ENDPOINT_KEY_PATTERN.test(endpointKey) || !rawHookId || !rawToken) {
163 return null;
164 }
165 try {
166 return {
167 endpointKey,
168 hookId: decodeURIComponent(rawHookId),
169 token: decodeURIComponent(rawToken),
170 // Forward the encoded segments byte-for-byte; the environment decodes them.
171 rawHookId,
172 rawToken,
173 search,
174 };
175 } catch {
176 return null;
177 }
178}
179
180function forwardedHeaders(headers: Readonly<Record<string, string>>): Record<string, string> {
181 const result: Record<string, string> = {};
182 // Headers the sender names in Connection are hop-by-hop too (RFC 9110 7.6.1).
183 const connectionValue =
184 Object.entries(headers).find(([name]) => name.toLowerCase() === "connection")?.[1] ?? "";
185 const nominated = new Set(
186 connectionValue
187 .split(",")
188 .map((name) => name.trim().toLowerCase())
189 .filter(Boolean),
190 );
191 for (const name in headers) {
192 const lower = name.toLowerCase();
193 if (
194 DROPPED_REQUEST_HEADERS.has(lower) ||
195 nominated.has(lower) ||
196 DROPPED_REQUEST_HEADER_PREFIXES.some((prefix) => lower.startsWith(prefix))
197 ) {
198 continue;
199 }
200 const value = headers[name];
201 if (value !== undefined) result[lower] = value;
202 }
203 return result;
204}
205
206const hasNoBody = (request: HttpServerRequest.HttpServerRequest) =>
207 request.source instanceof Request && request.source.body === null;
208
209const readCappedBody = (request: HttpServerRequest.HttpServerRequest) =>
210 Effect.suspend(() => {
211 if (hasNoBody(request)) {
212 return Effect.succeed(new Uint8Array(0));
213 }
214 const chunks: Array<Uint8Array> = [];
215 let total = 0;
216 return request.stream.pipe(
217 Stream.runForEach((chunk) => {
218 total += chunk.length;
219 if (total > RELAY_HOOK_MAX_BODY_BYTES) {
220 return Effect.fail(new HookBodyTooLarge());
221 }
222 chunks.push(chunk);
223 return Effect.void;
224 }),
225 Effect.map(() => {
226 const body = new Uint8Array(total);
227 let offset = 0;
228 for (const chunk of chunks) {
229 body.set(chunk, offset);
230 offset += chunk.length;
231 }
232 return body;
233 }),
234 );
235 });
236
237const make = Effect.gen(function* () {
238 const heldHooks = yield* HeldHooks.HeldHooks;
239 const settings = yield* RelayConfiguration.RelayConfiguration;
240 const httpClient = yield* HttpClient.HttpClient;
241 const rateLimiter = yield* HookRateLimiter;
242 const inbox = yield* HookInbox.HookInbox;
243 const crypto = yield* Crypto.Crypto;
244
245 const signDeliveryProof = (input: {
246 readonly environmentId: string;
247 readonly deliveryId: string;
248 readonly receivedAt: string;
249 readonly hookId: string;
250 readonly jti: string;
251 }) =>
252 Effect.gen(function* () {
253 const now = Math.floor((yield* Clock.currentTimeMillis) / 1_000);
254 return yield* signRelayJwt({
255 privateKey: Redacted.value(settings.cloudMintPrivateKey),
256 typ: RELAY_HOOK_DELIVERY_TYP,
257 payload: {
258 iss: normalizeRelayIssuer(settings.relayIssuer),
259 aud: `t3-env:${input.environmentId}`,
260 sub: input.environmentId,
261 jti: input.jti,
262 iat: now,
263 exp: now + DELIVERY_PROOF_LIFETIME_SECONDS,
264 environmentId: EnvironmentId.make(input.environmentId),
265 deliveryId: input.deliveryId,
266 receivedAt: input.receivedAt,
267 hookId: input.hookId,
268 } satisfies RelayHookDeliveryProofPayload,
269 });
270 }).pipe(Effect.orDie);
271
272 const handle = Effect.fn("relay.hooks.forward")(function* (
273 request: HttpServerRequest.HttpServerRequest,
274 ) {
275 const outcome = (value: string) => Effect.annotateCurrentSpan({ "relay.hook.outcome": value });
276 if (!FORWARDED_METHODS.has(request.method)) {
277 yield* outcome("method_not_allowed");
278 return errorResponse(405, "method_not_allowed", { allow: "GET, POST, PUT, PATCH" });
279 }
280 // When the sender called, not when the environment failed to answer.
281 const receivedAt = DateTime.formatIso(yield* DateTime.now);
282 const parsed = parseHookPath(request.url);
283 if (!parsed) {
284 yield* outcome("invalid_path");
285 return hookNotFound();
286 }
287 yield* Effect.annotateCurrentSpan({
288 "relay.hook.endpoint_key": parsed.endpointKey,
289 "relay.hook_id": parsed.hookId,
290 });
291 // A coarse budget per endpoint first, so minting new hook ids or tokens
292 // cannot buy unlimited lookups and forwards, or fill the inbox.
293 const endpointAllowed = yield* rateLimiter.allowEndpoint(parsed.endpointKey);
294 const hookAllowed =
295 endpointAllowed && (yield* rateLimiter.allowHook(yield* hookBudgetKey(parsed)));
296 if (!endpointAllowed || !hookAllowed) {
297 // Which budget ran out: the whole endpoint's, or this one hook URL's.
298 yield* Effect.annotateCurrentSpan({
299 "relay.hook.rate_limit": endpointAllowed ? "hook" : "endpoint",
300 });
301 yield* outcome("rate_limited");
302 return errorResponse(429, "rate_limited", {
303 "retry-after": String(RELAY_HOOK_RATE_LIMIT.periodSeconds),
304 });
305 }
306 const declaredLength = Number(request.headers["content-length"] ?? "0");
307 if (Number.isFinite(declaredLength) && declaredLength > RELAY_HOOK_MAX_BODY_BYTES) {
308 yield* outcome("payload_too_large");
309 return errorResponse(413, "payload_too_large");
310 }
311
312 const endpoint = yield* heldHooks.resolveEndpoint(parsed.endpointKey).pipe(
313 Effect.catch((error) =>
314 Effect.logWarning("Failed to resolve hook endpoint", {
315 endpointKey: parsed.endpointKey,
316 errorTag: error._tag,
317 }).pipe(Effect.as(null)),
318 ),
319 );
320 if (!endpoint) {
321 yield* outcome("not_found");
322 return hookNotFound();
323 }
324 yield* Effect.annotateCurrentSpan({
325 "relay.environment_id": endpoint.environmentId,
326 "relay.hook.hold_while_offline": endpoint.holdWhileOffline,
327 });
328
329 const body =
330 request.method === "GET"
331 ? Result.succeed(new Uint8Array(0))
332 : yield* readCappedBody(request).pipe(Effect.result);
333 if (Result.isFailure(body)) {
334 if (body.failure._tag === "HookBodyTooLarge") {
335 yield* outcome("payload_too_large");
336 return errorResponse(413, "payload_too_large");
337 }
338 yield* outcome("invalid_body");
339 return errorResponse(400, "invalid_body");
340 }
341
342 // One id per request, so a request that reached the environment before a
343 // timeout and is later delivered from the inbox runs only once.
344 yield* Effect.annotateCurrentSpan({ "relay.hook.body_bytes": body.success.byteLength });
345 const deliveryId = yield* crypto.randomUUIDv4().pipe(Effect.orDie);
346 // Proves to the environment that this delivery id, receive time and
347 // trace context came from the relay. Signed once here and stored with a
348 // held request, so the inbox never needs the signing key.
349 const proof = yield* signDeliveryProof({
350 environmentId: endpoint.environmentId,
351 deliveryId,
352 receivedAt,
353 hookId: parsed.hookId,
354 jti: yield* crypto.randomUUIDv4().pipe(Effect.orDie),
355 });
356 const hook = {
357 id: deliveryId,
358 receivedAt,
359 method: request.method,
360 rawHookId: parsed.rawHookId,
361 rawToken: parsed.rawToken,
362 hookKey: parsed.hookId,
363 query: parsed.search.replace(/^\?/, ""),
364 headers: { ...forwardedHeaders(request.headers), [RELAY_HOOK_DELIVERY_HEADER]: proof },
365 body: body.success,
366 };
367 // Held only for environments that opted in; otherwise the relay is a plain proxy.
368 const holdOrFail = (status: 503 | 504, error: string) =>
369 Effect.gen(function* () {
370 if (!endpoint.holdWhileOffline) {
371 yield* outcome(error);
372 return errorResponse(status, error);
373 }
374 const stored = yield* inbox
375 .hold({
376 endpointKey: parsed.endpointKey,
377 baseUrl: endpoint.httpBaseUrl,
378 hook,
379 })
380 .pipe(
381 Effect.catch((cause) =>
382 Effect.logWarning("Could not hold webhook request", {
383 environmentId: endpoint.environmentId,
384 errorTag: cause._tag,
385 }).pipe(Effect.as(null)),
386 ),
387 );
388 if (stored === null) {
389 yield* outcome(error);
390 return errorResponse(status, error);
391 }
392 if (!stored) {
393 // The inbox span carries which cap refused it (relay.inbox.refused).
394 yield* outcome("inbox_full");
395 return errorResponse(503, "inbox_full");
396 }
397 yield* outcome("held");
398 return HttpServerResponse.jsonUnsafe({ queued: true }, { status: 202 });
399 });
400
401 const upstream = yield* sendUpstream(endpoint.httpBaseUrl, hook).pipe(
402 Effect.provideService(HttpClient.HttpClient, httpClient),
403 Effect.result,
404 );
405 if (Result.isFailure(upstream)) {
406 yield* Effect.annotateCurrentSpan({ "relay.hook.upstream_error": upstream.failure._tag });
407 return yield* holdOrFail(503, "environment_unavailable");
408 }
409 if (Option.isNone(upstream.success)) {
410 return yield* holdOrFail(504, "environment_timeout");
411 }
412 const response = upstream.success.value;
413 if (ENVIRONMENT_UNREACHABLE_STATUSES.has(response.status)) {
414 yield* Effect.annotateCurrentSpan({ "relay.hook.upstream_status": response.status });
415 return yield* holdOrFail(503, "environment_unavailable");
416 }
417 yield* Effect.annotateCurrentSpan({
418 "relay.hook.outcome": "forwarded",
419 "relay.hook.upstream_status": response.status,
420 ...(response.outcome === undefined
421 ? {}
422 : { "relay.hook.upstream_outcome": response.outcome }),
423 });
424 // Only content-type is passed through: no location (redirects are never
425 // followed or relayed), no cookies, no upstream infrastructure headers.
426 const headers = {
427 ...SANDBOXED_RESPONSE_HEADERS,
428 ...(response.contentType ? { "content-type": response.contentType } : {}),
429 };
430 if (response.body.length === 0) {
431 return HttpServerResponse.empty({ status: response.status, headers });
432 }
433 return HttpServerResponse.uint8Array(response.body, {
434 status: response.status,
435 headers,
436 ...(response.contentType ? { contentType: response.contentType } : {}),
437 });
438 });
439
440 return HookForwarder.of({ handle });
441});
442
443export const layer = Layer.effect(HookForwarder, make);
444
445/**
446 * Implements the RelayApi `hooks` group. The endpoints are raw: the forwarder
447 * re-reads the encoded path segments from the request so the token and hook
448 * id reach the environment byte for byte, and streams the body itself.
449 */
450export const layerApi = HttpApiBuilder.group(
451 RelayApi,
452 "hooks",
453 Effect.fnUntraced(function* (handlers) {
454 const forwarder = yield* HookForwarder;
455 const forward = ({ request }: { readonly request: HttpServerRequest.HttpServerRequest }) =>
456 forwarder.handle(request);
457 return handlers
458 .handleRaw("forwardPost", forward)
459 .handleRaw("forwardPut", forward)
460 .handleRaw("forwardPatch", forward)
461 .handleRaw("forwardGet", forward);
462 }),
463);