1import * as Cloudflare from "@/Cloudflare/index.ts";
2import * as Alchemy from "@/index.ts";
3import * as Test from "@/Test/Alchemy.ts";
4import { describe, expect } from "alchemy-test";
5import * as Data from "effect/Data";
6import * as Deferred from "effect/Deferred";
7import * as Effect from "effect/Effect";
8import * as Exit from "effect/Exit";
9import * as Queue from "effect/Queue";
10import * as Schedule from "effect/Schedule";
11import * as Schema from "effect/Schema";
12import * as HttpClientRequest from "effect/http/HttpClientRequest";
13import { requestWorker } from "../Utils/WorkerRequest.ts";
14import AttachmentWorker from "./fixtures/hibernating-websocket/worker.ts";
15
16class HandshakeFailed extends Data.TaggedError("HandshakeFailed")<{
17 readonly message: string;
18}> {}
19
20const connect = Effect.fn(
21 function* (url: string) {
22 const opened = yield* Deferred.make<void, HandshakeFailed>();
23 const closed = yield* Deferred.make<{ code: number; reason: string }>();
24 const messages = yield* Queue.unbounded<string>();
25 const socket = yield* Effect.acquireRelease(
26 Effect.sync(() => {
27 const socket = new WebSocket(url.replace(/^http/, "ws"));
28 socket.addEventListener("open", () =>
29 Deferred.doneUnsafe(opened, Exit.void),
30 );
31 socket.addEventListener("error", (event) =>
32 Deferred.doneUnsafe(
33 opened,
34 Exit.fail(
35 new HandshakeFailed({
36 message: `WebSocket handshake failed: ${event.type}`,
37 }),
38 ),
39 ),
40 );
41 socket.addEventListener("message", (event) =>
42 Queue.offerUnsafe(messages, String(event.data)),
43 );
44 socket.addEventListener("close", (event) =>
45 Deferred.doneUnsafe(
46 closed,
47 Exit.succeed({ code: event.code, reason: event.reason }),
48 ),
49 );
50 return socket;
51 }),
52 (socket) => Effect.sync(() => socket.close()),
53 );
54 yield* Deferred.await(opened).pipe(
55 Effect.timeout("10 seconds"),
56 Effect.onError(() => Effect.sync(() => socket.close())),
57 );
58 return {
59 socket,
60 send: (message: string) => Effect.sync(() => socket.send(message)),
61 receive: Queue.take(messages).pipe(
62 Effect.flatMap(
63 Schema.decodeUnknownEffect(Schema.fromJsonString(Schema.Unknown)),
64 ),
65 Effect.timeout("10 seconds"),
66 ),
67 closed: Deferred.await(closed).pipe(Effect.timeout("10 seconds")),
68 };
69 },
70 Effect.retry({
71 while: (error) => error._tag === "HandshakeFailed",
72 schedule: Schedule.spaced("1 second"),
73 times: 8,
74 }),
75);
76
77const Stats = Schema.Struct({
78 boots: Schema.Number,
79 restored: Schema.Array(Schema.Number),
80 rejected: Schema.Array(Schema.String),
81 count: Schema.Number,
82 joined: Schema.String,
83 isDate: Schema.Boolean,
84});
85const timestamp = "2026-01-02T03:04:05.000Z";
86
87describe.concurrent.each([
88 { dev: true, stage: "native-websocket-local" },
89 { dev: false, stage: "native-websocket-live" },
90])(
91 "Native WebSocket attachments (dev: $dev)",
92 ({ dev, stage }) => {
93 const state = dev ? Alchemy.inMemoryState() : Cloudflare.state();
94 const Stack = Alchemy.Stack(
95 "HibernatingWebSocketStack",
96 { providers: Cloudflare.providers(), state },
97 Effect.gen(function* () {
98 const worker = yield* AttachmentWorker;
99 return { url: worker.url.as<string>() };
100 }),
101 );
102 const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
103 providers: Cloudflare.providers(),
104 state,
105 dev,
106 stage,
107 });
108 const stack = beforeAll(
109 Effect.gen(function* () {
110 yield* destroy(Stack);
111 const output = yield* deploy(Stack);
112 const response = yield* requestWorker(
113 HttpClientRequest.get(`${output.url}/ready`),
114 );
115 expect(response.status).toBe(200);
116 expect(yield* response.text).toBe("ready");
117 yield* connect(`${output.url}/socket/ready`).pipe(Effect.scoped);
118 return output;
119 }),
120 { timeout: 120_000 },
121 );
122 afterAll(destroy(Stack), { timeout: 30_000 });
123
124 const request = Effect.fn(function* (command: string) {
125 const { url } = yield* stack;
126 const connection = yield* connect(`${url}/socket/${command}`);
127 yield* connection.send(command);
128 const response = yield* connection.receive;
129 expect(connection.socket.readyState).toBe(WebSocket.OPEN);
130 yield* connection.send("codec");
131 expect(yield* connection.receive).toEqual({
132 encoded: { version: 1, count: "42", joined: timestamp },
133 count: 42,
134 joined: timestamp,
135 isDate: true,
136 });
137 return response;
138 });
139
140 test(
141 "encodes number and Date codecs and exposes the encoded value unchecked",
142 Effect.gen(function* () {
143 expect(yield* request("codec")).toEqual({
144 encoded: { version: 1, count: "42", joined: timestamp },
145 count: 42,
146 joined: timestamp,
147 isDate: true,
148 });
149 }).pipe(Effect.scoped),
150 { tags: [...(dev ? ["local"] : ["live"])] },
151 );
152
153 test(
154 "uses distinct encoding and decoding services in the caller's context",
155 Effect.gen(function* () {
156 expect(yield* request("services")).toEqual({
157 encoded: "37",
158 decoded: 42,
159 });
160 }).pipe(Effect.scoped),
161 { tags: [...(dev ? ["local"] : ["live"])] },
162 );
163
164 test(
165 "decodes existing unchecked attachments without an envelope",
166 Effect.gen(function* () {
167 expect(yield* request("unchecked")).toEqual({
168 count: 17,
169 joined: timestamp,
170 isDate: true,
171 });
172 }).pipe(Effect.scoped),
173 { tags: [...(dev ? ["local"] : ["live"])] },
174 );
175
176 test(
177 "reports native absence even for Schema.Unknown",
178 Effect.gen(function* () {
179 const response = yield* request("missing");
180 yield* Effect.sync(() =>
181 console.log("Native attachment absence", { dev, response }),
182 );
183 expect(response).toMatchObject({
184 _tag: "WebSocketAttachmentError",
185 reason: "missing",
186 absence: "null",
187 });
188 }).pipe(Effect.scoped),
189 { tags: [...(dev ? ["local"] : ["live"])] },
190 );
191
192 test(
193 "returns malformed encoded values as recoverable decode errors",
194 Effect.gen(function* () {
195 expect(yield* request("malformed")).toMatchObject({
196 _tag: "WebSocketAttachmentError",
197 reason: "decode",
198 causeIsError: true,
199 });
200 }).pipe(Effect.scoped),
201 { tags: [...(dev ? ["local"] : ["live"])] },
202 );
203
204 test(
205 "encode failure preserves the previously written attachment",
206 Effect.gen(function* () {
207 expect(yield* request("encode")).toMatchObject({
208 _tag: "WebSocketAttachmentError",
209 reason: "encode",
210 previous: "7",
211 causeIsError: true,
212 });
213 }).pipe(Effect.scoped),
214 { tags: [...(dev ? ["local"] : ["live"])] },
215 );
216
217 test(
218 "native oversize serialization is a recoverable write failure",
219 Effect.gen(function* () {
220 const response = yield* request("oversize");
221 yield* Effect.sync(() =>
222 console.log("Native attachment size limit", { dev, response }),
223 );
224 expect(response).toMatchObject({
225 _tag: "WebSocketAttachmentError",
226 reason: "write",
227 acceptedLength: 16_000,
228 previous: "small",
229 cause:
230 "A WebSocket 'attachment' cannot be larger than 16384 bytes.'attachment' was 32774 bytes.",
231 causeIsError: true,
232 });
233 }).pipe(Effect.scoped),
234 { tags: [...(dev ? ["local"] : ["live"])] },
235 );
236
237 test(
238 "native structured clone preserves Dates, Maps, and bytes without aliasing",
239 Effect.gen(function* () {
240 expect(yield* request("clone")).toEqual({
241 date: timestamp,
242 count: 42,
243 bytes: [1, 2, 3],
244 });
245 }).pipe(Effect.scoped),
246 { tags: [...(dev ? ["local"] : ["live"])] },
247 );
248
249 test(
250 "a schema-valid uncloneable value reports the genuine native error",
251 Effect.gen(function* () {
252 expect(yield* request("uncloneable")).toMatchObject({
253 _tag: "WebSocketAttachmentError",
254 reason: "write",
255 causeIsError: true,
256 });
257 }).pipe(Effect.scoped),
258 { tags: [...(dev ? ["local"] : ["live"])] },
259 );
260
261 test(
262 "real hibernation restores typed attachments and closes an obsolete schema by application policy",
263 Effect.gen(function* () {
264 const { url } = yield* stack;
265 const valid = yield* connect(`${url}/socket/idle`);
266 yield* valid.send("codec");
267 yield* valid.receive;
268 yield* valid.send("stats");
269 const before = yield* valid.receive.pipe(
270 Effect.flatMap(Schema.decodeUnknownEffect(Stats)),
271 );
272 const obsolete = yield* connect(`${url}/socket/idle`);
273 yield* obsolete.send("obsolete");
274 expect(yield* obsolete.receive).toEqual({
275 boots: before.boots,
276 obsolete: true,
277 });
278 const sameSocket = valid.socket;
279 const after = yield* Effect.gen(function* () {
280 yield* Effect.sleep("15 seconds");
281 yield* valid.send("stats");
282 return yield* valid.receive.pipe(
283 Effect.flatMap(Schema.decodeUnknownEffect(Stats)),
284 );
285 }).pipe(
286 Effect.repeat({
287 times: 2,
288 until: (value) => value.boots > before.boots,
289 }),
290 );
291 expect(after.boots).toBeGreaterThan(before.boots);
292 expect(after.restored).toEqual([42]);
293 expect(after.rejected).toEqual(["decode"]);
294 expect(after.count).toBe(42);
295 expect(after.joined).toBe(timestamp);
296 expect(after.isDate).toBe(true);
297 expect(valid.socket).toBe(sameSocket);
298 expect(valid.socket.readyState).toBe(WebSocket.OPEN);
299 expect(yield* obsolete.receive).toMatchObject({
300 _tag: "WebSocketAttachmentError",
301 reason: "decode",
302 boots: after.boots,
303 });
304 expect(yield* obsolete.closed).toEqual({
305 code: 1008,
306 reason: "Invalid attachment",
307 });
308 expect(obsolete.socket.readyState).toBe(WebSocket.CLOSED);
309 yield* valid.send("stats");
310 expect(yield* valid.receive).toEqual(after);
311 }).pipe(Effect.scoped),
312 { tags: [...(dev ? ["local"] : ["live"])], timeout: 110_000 },
313 );
314 },
315 { tags: ["provider:cloudflare", "provider:cloudflare:worker"] },
316);