1import type * as cf from "@cloudflare/workers-types";
2import * as Cause from "effect/Cause";
3import * as Effect from "effect/Effect";
4import { pipe } from "effect/Function";
5import * as Schedule from "effect/Schedule";
6import * as HttpBody from "effect/http/HttpBody";
7import * as HttpClient from "effect/http/HttpClient";
8import { HttpClientError, TransportError } from "effect/http/HttpClientError";
9import * as HttpClientRequest from "effect/http/HttpClientRequest";
10import * as HttpClientResponse from "effect/http/HttpClientResponse";
11import type { HttpServerError } from "effect/http/HttpServerError";
12import * as HttpServerRequest from "effect/http/HttpServerRequest";
13import * as HttpServerResponse from "effect/http/HttpServerResponse";
14import * as Socket from "effect/socket/Socket";
15
16export type SocketAddress = cf.SocketAddress;
17
18export type SocketOptions = cf.SocketOptions;
19
20/**
21 * Whether a native Web `Headers` and an Effect headers record carry the
22 * same entries. Lets {@link makeFetcher} keep forwarding the platform's
23 * own request object when `request.modify` left the headers unchanged,
24 * and rebuild only when they genuinely differ. Effect header keys are
25 * already lowercased and `Headers.get` is case-insensitive, so the
26 * comparison needs no normalization.
27 */
28const sameHeaders = (
29 web: Headers,
30 eff: Readonly<Record<string, string>>,
31): boolean => {
32 const keys = Object.keys(eff);
33 let webCount = 0;
34 for (const _ of web.keys()) {
35 webCount++;
36 if (webCount > keys.length) return false;
37 }
38 if (webCount !== keys.length) return false;
39 for (const key of keys) {
40 if (web.get(key) !== eff[key]) return false;
41 }
42 return true;
43};
44
45export interface Fetcher {
46 raw: cf.Fetcher;
47 fetch(
48 request: HttpClientRequest.HttpClientRequest,
49 ): Effect.Effect<HttpClientResponse.HttpClientResponse, HttpClientError>;
50 fetch(
51 request: HttpServerRequest.HttpServerRequest,
52 ): Effect.Effect<HttpServerResponse.HttpServerResponse, HttpServerError>;
53
54 connect(
55 address: SocketAddress | string,
56 options?: SocketOptions,
57 ): Socket.Socket;
58}
59
60export const toCloudflareFetcher = Effect.fn(function* (fetcher: Fetcher) {
61 const context = yield* Effect.context();
62 return {
63 fetch: (input, init) =>
64 fetcher
65 .fetch(
66 HttpServerRequest.fromWeb(
67 new Request(input as any, init as any) as any as Request,
68 ),
69 )
70 .pipe(
71 Effect.map(
72 (response) =>
73 HttpServerResponse.toWeb(response, {
74 context,
75 }) as any as cf.Response,
76 ),
77 Effect.provideContext(context),
78 Effect.runPromise,
79 ),
80 connect() {
81 // TODO
82 throw new Error("toCloudflareFetcher does not support connect()");
83 },
84 } satisfies cf.Fetcher;
85});
86
87export const fromCloudflareFetcher = (
88 fetcher: cf.Fetcher | globalThis.Fetcher,
89): Fetcher => {
90 const fetch = (request: Request) =>
91 Effect.suspend(() => {
92 // Clone per attempt, keeping `request` pristine: the HandlerNotReady
93 // retry below re-runs this suspend, and replaying a Request whose body
94 // the failed attempt already consumed makes workerd reject with
95 // "TypeError: Cannot reconstruct a Request with a used body".
96 const attempt = request.clone() as Request;
97 return Effect.promise((signal) =>
98 (fetcher as globalThis.Fetcher).fetch(attempt, {
99 signal: signal,
100 }),
101 );
102 }).pipe(
103 // The "Handler does not export a fetch()" window is a property of
104 // invoking a freshly-deployed Cloudflare binding, so it is ridden out
105 // HERE — the one adapter every binding flows through — rather than in any
106 // single higher-level wrapper (`toHttpClient`, the client/server
107 // overloads, the RPC DO transport all go through this). `Effect.promise`
108 // surfaces the rejection as a defect; lift it to a typed retryable error
109 // and back off until the new version propagates. The request never
110 // reached a handler, so nothing committed — safe to retry. This wraps
111 // only the promise, leaving the response (and its streaming body)
112 // untouched, so RPC/stream decoding is unaffected. After the budget is
113 // exhausted, re-raise the original defect unchanged.
114 Effect.catchCause((cause) => {
115 const squashed = Cause.squash(cause);
116 return isHandlerNotReady(squashed)
117 ? Effect.fail(new HandlerNotReady(squashed))
118 : Effect.failCause(cause);
119 }),
120 Effect.retry({
121 while: (error) => error instanceof HandlerNotReady,
122 schedule: Schedule.exponential("100 millis"),
123 times: 8,
124 }),
125 Effect.catch((error) =>
126 error instanceof HandlerNotReady
127 ? Effect.die(error.cause)
128 : Effect.failCause(Cause.fail(error)),
129 ),
130 );
131
132 return {
133 raw: fetcher as cf.Fetcher,
134 connect: (address, options) =>
135 fromCloudflareSocket(fetcher.connect(address, options)),
136 fetch: (
137 request:
138 | HttpClientRequest.HttpClientRequest
139 | HttpServerRequest.HttpServerRequest,
140 ): any =>
141 HttpClientRequest.isHttpClientRequest(request)
142 ? pipe(
143 HttpServerRequest.toWeb(
144 HttpServerRequest.fromClientRequest(request),
145 ),
146 Effect.flatMap(fetch),
147 Effect.map((response) =>
148 HttpClientResponse.fromWeb(request, response as any as Response),
149 ),
150 Effect.catch((error) =>
151 Effect.succeed(
152 HttpClientResponse.fromWeb(
153 request,
154 new Response(error.message, {
155 status:
156 error._tag === "InternalError"
157 ? 500
158 : error._tag === "RequestParseError"
159 ? 400
160 : 404,
161 }),
162 ),
163 ),
164 ),
165 )
166 : pipe(
167 HttpServerRequest.toWeb(request),
168 // `toWeb` hands back the raw native request untouched whenever
169 // there is one (always, on workerd) — the fast path we WANT:
170 // it forwards the platform's own edge request instead of
171 // cloning it. The catch is that it also ignores header
172 // overrides from `request.modify({ headers })`, so a caller
173 // that strips or mints an internal header before forwarding
174 // (e.g. a Worker sanitizing a trust header for its Durable
175 // Object) would be silently dropped. Rebuild ONLY when the
176 // effect request's headers actually differ from the native
177 // request's; an unmodified forward (or a modify that was a
178 // no-op) stays on the metal. `new Request(req, init)`
179 // preserves the method and un-consumed streaming body.
180 Effect.map((webRequest) =>
181 sameHeaders(webRequest.headers, request.headers)
182 ? webRequest
183 : new Request(webRequest, {
184 headers: request.headers as Record<string, string>,
185 }),
186 ),
187 Effect.flatMap(fetch),
188 Effect.map((response) => {
189 if ((response as any).status === 101) {
190 return HttpServerResponse.setBody(
191 HttpServerResponse.empty({ status: 101 }),
192 HttpBody.raw(response),
193 );
194 }
195 return HttpServerResponse.fromWeb(response as any as Response);
196 }),
197 ),
198 };
199};
200
201/**
202 * A freshly-deployed Durable Object / service script is eventually consistent:
203 * for a short window after deploy, workerd can route a `.fetch()` to a stale
204 * script version whose class has no fetch handler yet, surfacing as
205 * "Handler does not export a fetch() function." It clears within seconds once
206 * the new version propagates, so it is safe to retry.
207 */
208const isHandlerNotReady = (error: unknown): boolean => {
209 const message =
210 error instanceof Error
211 ? error.message
212 : typeof error === "string"
213 ? error
214 : typeof (error as { message?: unknown } | null)?.message === "string"
215 ? ((error as { message: string }).message as string)
216 : "";
217 return message.includes("does not export a fetch");
218};
219
220/**
221 * Adapt anything that exposes a server-shaped `fetch` (e.g. a Durable Object
222 * stub, a Worker service binding) into an Effect `HttpClient`. Lets HttpApi
223 * clients address bindings without a base URL via `transformClient`.
224 */
225class HandlerNotReady {
226 readonly _tag = "HandlerNotReady";
227 constructor(readonly cause: unknown) {}
228}
229
230export const toHttpClient = (fetcher: {
231 fetch: (
232 request: HttpServerRequest.HttpServerRequest,
233 ) => Effect.Effect<HttpServerResponse.HttpServerResponse, HttpServerError>;
234}) =>
235 HttpClient.make((request) => {
236 return Effect.suspend(() =>
237 // Rebuild the server request on every attempt so a retry re-serializes
238 // the body instead of replaying a consumed one.
239 fetcher
240 .fetch(HttpServerRequest.fromClientRequest(request))
241 .pipe(
242 Effect.map((response) =>
243 HttpClientResponse.fromWeb(
244 request,
245 HttpServerResponse.toWeb(response),
246 ),
247 ),
248 ),
249 ).pipe(
250 // The handler-not-ready window is already ridden out at the lowest level
251 // (the `fromCloudflareFetcher` promise retries `HandlerNotReady` before
252 // it ever reaches here). A `HandlerNotReady` still surfacing means the
253 // budget was exhausted — it arrives as a defect (re-raised rejected
254 // promise), so convert it (and any other failure cause) into a typed
255 // transport error rather than retrying again.
256 Effect.catchCause((cause) => {
257 const squashed = Cause.squash(cause);
258 return Effect.fail(
259 new HttpClientError({
260 reason: new TransportError({
261 request,
262 cause: isHandlerNotReady(squashed)
263 ? (squashed as { message?: unknown })
264 : squashed,
265 description: "Fetcher-backed HttpClient request failed",
266 }),
267 }),
268 );
269 }),
270 );
271 });
272
273export const fromCloudflareSocket = (
274 cfSocket: globalThis.Socket | cf.Socket,
275): Socket.Socket =>
276 // `fromTransformStream` snapshots fiber context, then waits to acquire
277 // the streams until a consumer opens the reader. `runSync` is only that
278 // snapshot — connection still happens on first `socket.reader`.
279 Effect.runSync(
280 Socket.fromTransformStream(
281 Effect.tryPromise({
282 try: () =>
283 Promise.resolve(cfSocket.opened).then(
284 () =>
285 ({
286 readable: cfSocket.readable,
287 writable: cfSocket.writable,
288 }) as Socket.InputTransformStream,
289 ),
290 catch: (cause) =>
291 new Socket.SocketError({
292 reason: new Socket.SocketOpenError({
293 kind: "Unknown",
294 cause,
295 }),
296 }),
297 }),
298 ),
299 );