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