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