1import * as Duration from "effect/Duration";
2import * as Effect from "effect/Effect";
3import * as Option from "effect/Option";
4import * as Schema from "effect/Schema";
5import * as Stream from "effect/Stream";
6import * as HttpClient from "effect/http/HttpClient";
7import * as HttpClientRequest from "effect/http/HttpClientRequest";
8import * as HttpMethod from "effect/http/HttpMethod";
9import * as HttpTraceContext from "effect/http/HttpTraceContext";
10import type * as HttpClientResponse from "effect/http/HttpClientResponse";
11
12import { withoutRedirects } from "../environments/EnvironmentConnector.ts";
13
14/** Set by the relay on every forward; the environment uses it as the delivery id. */
15const RELAY_DELIVERY_ID_HEADER = "x-t3-relay-delivery-id";
16/** When the relay received the request; the environment trusts it only with the delivery id. */
17const RELAY_RECEIVED_AT_HEADER = "x-t3-relay-received-at";
18/** The environment answers with a small JSON status; anything past this is cut off. */
19const MAX_RESPONSE_BYTES = 64 * 1024;
20export const RELAY_HOOK_UPSTREAM_TIMEOUT_MS = 8_000;
21// Cloudflare answers 530 when the tunnel for a hostname has no connected origin.
22export const TUNNEL_OFFLINE_STATUS = 530;
23
24/** A webhook request as the relay sends it on to the environment. */
25export interface UpstreamHook {
26 readonly id: string;
27 readonly receivedAt: string;
28 readonly method: string;
29 /** Path segments exactly as the sender sent them; the environment decodes them. */
30 readonly rawHookId: string;
31 readonly rawToken: string;
32 /** Without the leading `?`. */
33 readonly query: string;
34 readonly headers: Readonly<Record<string, string>>;
35 readonly body: Uint8Array;
36}
37
38export interface UpstreamResponse {
39 readonly status: number;
40 readonly contentType: string | undefined;
41 /** What the environment did with the request (`x-t3-hook-outcome`). */
42 readonly outcome: string | undefined;
43 readonly body: Uint8Array;
44}
45
46/** Outcome names are short identifiers; anything else is not recorded. */
47const OUTCOME_PATTERN = /^[a-z_]{1,32}$/;
48
49class ResponseTooLarge extends Schema.TaggedError<ResponseTooLarge>()("ResponseTooLarge", {}) {}
50
51/** Reads a response body, failing once it passes the cap rather than buffering it all. */
52const readCapped = (response: HttpClientResponse.HttpClientResponse) =>
53 Effect.suspend(() => {
54 const chunks: Array<Uint8Array> = [];
55 let total = 0;
56 return response.stream.pipe(
57 Stream.runForEach((chunk) => {
58 total += chunk.length;
59 if (total > MAX_RESPONSE_BYTES) return Effect.fail(new ResponseTooLarge());
60 chunks.push(chunk);
61 return Effect.void;
62 }),
63 Effect.map(() => {
64 const body = new Uint8Array(total);
65 let offset = 0;
66 for (const chunk of chunks) {
67 body.set(chunk, offset);
68 offset += chunk.length;
69 }
70 return body;
71 }),
72 // A response without a body, such as a redirect, has no stream at all.
73 Effect.catchIf(
74 (error) => error._tag === "HttpClientError" && error.reason._tag === "EmptyBodyError",
75 () => Effect.succeed(new Uint8Array(0)),
76 ),
77 );
78 });
79
80/**
81 * Sends a webhook request through the environment's tunnel. Fails when the
82 * environment cannot be reached, and succeeds with None on timeout.
83 */
84export const sendUpstream = (baseUrl: string, hook: UpstreamHook) =>
85 Effect.gen(function* () {
86 const httpClient = yield* HttpClient.HttpClient;
87 const base = baseUrl.endsWith("/") ? baseUrl : `${baseUrl}/`;
88 // The environment's span joins this trace. Set by hand: the client span
89 // that would propagate it is off, because it records the token in url.full.
90 const parent = yield* Effect.currentSpan.pipe(Effect.option);
91 // `hook.headers` carries the signed delivery proof, set once when the
92 // relay received the request, so a held request sends the same proof.
93 const headers: Record<string, string> = {
94 ...hook.headers,
95 ...(Option.isSome(parent) ? HttpTraceContext.toHeaders(parent.value) : {}),
96 [RELAY_DELIVERY_ID_HEADER]: hook.id,
97 [RELAY_RECEIVED_AT_HEADER]: hook.receivedAt,
98 };
99 let request = HttpClientRequest.make(hook.method as "GET" | "POST" | "PUT" | "PATCH")(
100 `${base}api/hooks/${hook.rawHookId}/${hook.rawToken}${hook.query ? `?${hook.query}` : ""}`,
101 { headers },
102 );
103 if (HttpMethod.hasBody(request.method)) {
104 request = HttpClientRequest.bodyUint8Array(request, hook.body, headers["content-type"]);
105 }
106 return yield* httpClient.execute(request).pipe(
107 Effect.flatMap((response) =>
108 readCapped(response).pipe(
109 Effect.map((body): UpstreamResponse => {
110 const outcome = response.headers["x-t3-hook-outcome"];
111 return {
112 status: response.status,
113 contentType: response.headers["content-type"],
114 outcome: outcome !== undefined && OUTCOME_PATTERN.test(outcome) ? outcome : undefined,
115 body,
116 };
117 }),
118 ),
119 ),
120 withoutRedirects,
121 // The client span would record url.full, which carries the token.
122 Effect.provideService(HttpClient.TracerDisabledWhen, () => true),
123 Effect.timeoutOption(Duration.millis(RELAY_HOOK_UPSTREAM_TIMEOUT_MS)),
124 );
125 });