1import { EnvironmentHttpApi } from "@t3tools/contracts";
2import * as ByteSize from "effect/ByteSize";
3import * as Effect from "effect/Effect";
4import * as Option from "effect/Option";
5import * as Tracer from "effect/Tracer";
6import * as HttpIncomingMessage from "effect/http/HttpIncomingMessage";
7import type * as HttpServerRequest from "effect/http/HttpServerRequest";
8import * as HttpServerResponse from "effect/http/HttpServerResponse";
9import * as HttpTraceContext from "effect/http/HttpTraceContext";
10import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
11
12import * as Metrics from "../observability/Metrics.ts";
13import * as RelayDeliveryProof from "./RelayDeliveryProof.ts";
14import * as ScheduledTaskService from "./ScheduledTaskService.ts";
15
16/** Largest request body a webhook accepts. The relay enforces the same cap. */
17export const WEBHOOK_MAX_BODY_BYTES = 1024 * 1024;
18
19/** Response header naming what happened to the request; the relay records it. */
20const WEBHOOK_OUTCOME_HEADER = "x-t3-hook-outcome";
21
22const json = (status: number, body: Record<string, string>, outcome: string) =>
23 HttpServerResponse.jsonUnsafe(body, { status, headers: { [WEBHOOK_OUTCOME_HEADER]: outcome } });
24
25export const layer = HttpApiBuilder.group(
26 EnvironmentHttpApi,
27 "webhooks",
28 Effect.fnUntraced(function* (handlers) {
29 const scheduledTasks = yield* ScheduledTaskService.ScheduledTaskService;
30 const relayDeliveryProof = yield* RelayDeliveryProof.RelayDeliveryProof;
31 /**
32 * Handles `/api/hooks/:hookId/:token` for every accepted method. The endpoint
33 * is raw so the signature is checked over the exact body bytes; the service
34 * checks the token and signature. It is reachable directly, over the managed
35 * tunnel, or through the relay's stable `/v1/hooks/:endpointKey/:hookId/:token`
36 * URL, where the endpoint key is this environment's managed tunnel key.
37 */
38 const handler = ({
39 params,
40 request,
41 }: {
42 readonly params: { readonly hookId: string; readonly token: string };
43 readonly request: HttpServerRequest.HttpServerRequest;
44 }) =>
45 Effect.gen(function* () {
46 const contentLength = Number(request.headers["content-length"] ?? "0");
47 if (!Number.isFinite(contentLength) || contentLength > WEBHOOK_MAX_BODY_BYTES) {
48 return json(413, { error: "body_too_large" }, "body_too_large");
49 }
50 // Chunked requests carry no content-length, so the reader itself is capped.
51 const body = yield* request.arrayBuffer.pipe(
52 Effect.map((buffer) => new Uint8Array(buffer)),
53 Effect.provideService(
54 HttpIncomingMessage.MaxBodySize,
55 ByteSize.bytes(WEBHOOK_MAX_BODY_BYTES),
56 ),
57 Effect.option,
58 );
59 // Refused before a task is looked up, so the service never sees them.
60 const tooLarge = (error: string) =>
61 Metrics.increment(Metrics.webhookDeliveriesTotal, {
62 outcome: "body_too_large",
63 source: request.headers["x-t3-relay-delivery-id"] ? "relay" : "direct",
64 }).pipe(Effect.as(json(413, { error }, "body_too_large")));
65 if (Option.isNone(body)) return yield* tooLarge("body_too_large_or_unreadable");
66 if (body.value.byteLength > WEBHOOK_MAX_BODY_BYTES) {
67 return yield* tooLarge("body_too_large");
68 }
69
70 const headers: Record<string, string> = {};
71 for (const [name, value] of Object.entries(request.headers)) {
72 if (typeof value === "string") headers[name.toLowerCase()] = value;
73 }
74 const queryIndex = request.url.indexOf("?");
75 // The relay's delivery id and receive time count only with its signed
76 // proof; the URL can also be called directly. The receive time matters
77 // for requests the relay held while we were offline.
78 const relay = yield* relayDeliveryProof.verify({ headers, hookId: params.hookId });
79 const relayDeliveryId = Option.isSome(relay) ? relay.value.deliveryId : undefined;
80 const relayReceivedAt = Option.isSome(relay) ? relay.value.receivedAt : undefined;
81
82 // A relay delivery joins the relay's trace, and goes to the T3 Connect
83 // tracer with it. Anyone else's traceparent is never trusted.
84 const relayParent = Option.isSome(relay)
85 ? HttpTraceContext.fromHeaders(request.headers)
86 : Option.none<Tracer.ExternalSpan>();
87 const result = yield* scheduledTasks
88 .triggerWebhook({
89 hookId: params.hookId,
90 token: params.token,
91 method: request.method,
92 path: `${ScheduledTaskService.WEBHOOK_ROUTE_PREFIX}/${encodeURIComponent(params.hookId)}`,
93 query: queryIndex === -1 ? "" : request.url.slice(queryIndex + 1),
94 headers,
95 body: body.value,
96 bodyText: new TextDecoder().decode(body.value),
97 ...(relayDeliveryId ? { relayDeliveryId } : {}),
98 ...(relayReceivedAt ? { receivedAt: relayReceivedAt } : {}),
99 })
100 .pipe(
101 Option.isSome(relayParent)
102 ? (effect) => effect.pipe(Effect.withParentSpan(relayParent.value))
103 : (effect) => effect,
104 // Defects too, so the sender only ever sees the fixed error body.
105 Effect.catchCause((cause) =>
106 Effect.logWarning("Webhook delivery failed").pipe(
107 Effect.annotateLogs({ hookId: params.hookId }),
108 Effect.andThen(Effect.logDebug("Webhook delivery failure cause", { cause })),
109 Effect.as({ _tag: "error" as const }),
110 ),
111 ),
112 );
113
114 switch (result._tag) {
115 case "accepted":
116 return json(202, { deliveryId: result.deliveryId }, result.outcome);
117 case "not_found":
118 return json(404, { error: "hook_not_found" }, "not_found");
119 case "rejected_signature":
120 return json(401, { error: "invalid_signature" }, "rejected_signature");
121 case "disabled":
122 return json(409, { error: "hook_disabled" }, "disabled");
123 case "rate_limited":
124 return json(429, { error: "rate_limited" }, result.outcome);
125 case "expired":
126 return json(410, { error: "delivery_too_old" }, "expired");
127 case "error":
128 return json(500, { error: "internal_error" }, "error");
129 }
130 });
131 return handlers
132 .handleRaw("webhookPost", handler)
133 .handleRaw("webhookPut", handler)
134 .handleRaw("webhookPatch", handler)
135 .handleRaw("webhookGet", handler);
136 }),
137);