1import * as Data from "effect/Data";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Schedule from "effect/Schedule";
5import * as FetchHttpClient from "effect/http/FetchHttpClient";
6import * as HttpClient from "effect/http/HttpClient";
7import { DecodeError, HttpClientError } from "effect/http/HttpClientError";
8import * as HttpClientRequest from "effect/http/HttpClientRequest";
9import type { HttpClientResponse } from "effect/http/HttpClientResponse";
10import * as RpcClient from "effect/rpc/RpcClient";
11import * as RpcSerialization from "effect/rpc/RpcSerialization";
12
13/**
14 * A freshly-deployed Cloudflare Worker is not instantly reachable over HTTP.
15 * Its `workers.dev` route, the script, and each binding (R2 / D1 / DO / Secrets
16 * Store) propagate to the edge independently and asynchronously, so the first
17 * requests to a new URL can transiently return:
18 *
19 * - `404` while the `workers.dev` subdomain / route is still propagating
20 * (Cloudflare serves its "There is nothing here yet" placeholder), or
21 * - `5xx` while the script is up but a binding it depends on isn't ready yet.
22 *
23 * This is ordinary eventual consistency that belongs at the call site, not in
24 * the resource provider — the provider returning before every edge PoP has
25 * converged is correct. Consumers ride out the window by retrying the request.
26 */
27export class WorkerNotReady extends Data.TaggedError("WorkerNotReady")<{
28 status: number;
29}> {}
30
31/**
32 * Status codes that indicate the edge hasn't finished converging on a fresh
33 * deploy. Deliberate client errors (`401`, `403`, `400`, …) are NOT in this
34 * set, so assertions on those statuses still observe them immediately rather
35 * than being retried away.
36 */
37const isColdStartStatus = (status: number): boolean =>
38 status === 404 || status >= 500;
39
40export interface WhenReadyOptions {
41 /** Max retry attempts before surfacing {@link WorkerNotReady}. Default `20`. */
42 times?: number;
43}
44
45/**
46 * Execute an arbitrary {@link HttpClientRequest.HttpClientRequest}, retrying
47 * through the Cloudflare cold-start window ({@link isColdStartStatus}) until the
48 * Worker serves a non-transient response. The returned response can carry any
49 * non-cold-start status (e.g. `200`, `202`, `401`) for the caller to assert on.
50 */
51export const executeWhenReady = (
52 request: HttpClientRequest.HttpClientRequest,
53 options?: WhenReadyOptions,
54): Effect.Effect<HttpClientResponse, unknown, HttpClient.HttpClient> =>
55 Effect.gen(function* () {
56 const client = yield* HttpClient.HttpClient;
57 return yield* client.execute(request).pipe(
58 Effect.flatMap((response) =>
59 isColdStartStatus(response.status)
60 ? Effect.fail(new WorkerNotReady({ status: response.status }))
61 : Effect.succeed(response),
62 ),
63 Effect.retry({
64 while: (error) => error instanceof WorkerNotReady,
65 schedule: Schedule.max([
66 Schedule.exponential("500 millis"),
67 Schedule.recurs(options?.times ?? 20),
68 ]),
69 }),
70 );
71 });
72
73/**
74 * Convenience wrapper over {@link executeWhenReady} for a plain `GET`.
75 */
76export const getWhenReady = (
77 url: string,
78 options?: WhenReadyOptions,
79): Effect.Effect<HttpClientResponse, unknown, HttpClient.HttpClient> =>
80 executeWhenReady(HttpClientRequest.get(url), options);
81
82/**
83 * Options for the edge-transient response guard applied by
84 * {@link guardContentType} / {@link rpcClientLayer}. Pass per suite via
85 * `Test.make({ http: { ... } })` or per call site.
86 */
87export interface EdgeGuardOptions {
88 /**
89 * Retry schedule for edge-transient responses (and transport errors).
90 * Default: exponential from 500ms capped at 3s.
91 */
92 schedule?: Schedule.Schedule<unknown, unknown>;
93 /** Max transport-level retry attempts. Default `5`. */
94 times?: number;
95}
96
97const defaultGuardSchedule = Schedule.min([
98 Schedule.exponential("500 millis"),
99 Schedule.spaced("3 seconds"),
100]);
101
102/**
103 * Guard an `HttpClient` against edge-generated bodies on a freshly deployed
104 * Worker URL.
105 *
106 * Protocol clients like effect RPC's `layerProtocolHttp` never inspect the
107 * response status or content-type — they pipe the raw body straight into the
108 * serialization parser, so every edge-generated HTML page (the workers.dev
109 * placeholder, which serves with HTTP **200**; 1101/1102 error pages;
110 * 429/1015 rate limits) surfaces as an opaque decode defect (e.g.
111 * `RpcClientDefect: Error decoding HTTP response`).
112 *
113 * This transform rejects any response whose `content-type` doesn't match the
114 * expected one as a typed `HttpClientError` that records the status + a body
115 * snippet — so a failure names its cause — and retries the request through a
116 * bounded schedule so callers ride out the cold-start window
117 * ({@link WorkerNotReady} documents why this belongs at the call site).
118 */
119export const guardContentType =
120 (contentType: string, options?: EdgeGuardOptions) =>
121 (client: HttpClient.HttpClient): HttpClient.HttpClient =>
122 client.pipe(
123 HttpClient.transform((effect, request) =>
124 Effect.flatMap(effect, (response) => {
125 const observed = response.headers["content-type"] ?? "";
126 if (observed.includes(contentType)) {
127 return Effect.succeed(response);
128 }
129 return response.text.pipe(
130 Effect.orElseSucceed(() => "<unreadable body>"),
131 Effect.flatMap((body) =>
132 Effect.fail(
133 new HttpClientError({
134 reason: new DecodeError({
135 request,
136 response,
137 description: `expected "${contentType}", got "${observed}" (status ${response.status}): ${body.slice(0, 200)}`,
138 }),
139 }),
140 ),
141 ),
142 );
143 }),
144 ),
145 HttpClient.retry({
146 schedule: options?.schedule ?? defaultGuardSchedule,
147 times: options?.times ?? 5,
148 }),
149 );
150
151/**
152 * A fetch-backed `HttpClient` layer wrapped with {@link guardContentType}.
153 *
154 * Deliberately a standalone layer rather than a transform of the ambient
155 * test `HttpClient`: the ambient client also serves the engine's own cloud
156 * API calls during `deploy`/`destroy` and distilled SDK calls in
157 * `test.provider` bodies, which must NOT be subjected to this guard.
158 */
159export const guardedFetchLayer = (
160 contentType: string,
161 options?: EdgeGuardOptions,
162): Layer.Layer<HttpClient.HttpClient> =>
163 Layer.effect(HttpClient.HttpClient)(
164 Effect.gen(function* () {
165 const client = yield* HttpClient.HttpClient;
166 return guardContentType(contentType, options)(client);
167 }),
168 ).pipe(Layer.provide(FetchHttpClient.layer));
169
170/**
171 * Complete RPC-over-HTTP client transport for driving a deployed Worker:
172 * `RpcClient.layerProtocolHttp` + the given serialization (ndjson by
173 * default) + a {@link guardedFetchLayer} expecting that serialization's
174 * content-type.
175 *
176 * ```ts
177 * const client = yield* RpcClient.make(WorkerRpcs);
178 * // ...
179 * }).pipe(Effect.scoped, Effect.provide(Test.Http.rpcClientLayer(url)));
180 * ```
181 */
182export const rpcClientLayer = (
183 url: string,
184 options?: EdgeGuardOptions & {
185 /** RPC wire serialization. Default {@link RpcSerialization.ndjson}. */
186 serialization?: RpcSerialization.RpcSerialization["Service"];
187 /**
188 * Set `false` to skip the edge guard entirely and use a plain fetch
189 * transport — e.g. for a test that asserts on the raw edge behavior
190 * itself. Default `true`.
191 */
192 guard?: boolean;
193 },
194): Layer.Layer<RpcClient.Protocol> => {
195 const serialization = options?.serialization ?? RpcSerialization.ndjson;
196 return RpcClient.layerProtocolHttp({ url }).pipe(
197 Layer.provide(
198 options?.guard === false
199 ? FetchHttpClient.layer
200 : guardedFetchLayer(serialization.contentType, options),
201 ),
202 Layer.provide(
203 Layer.succeed(RpcSerialization.RpcSerialization, serialization),
204 ),
205 );
206};