1import { makeFunctionBridge } from "@/Neon/FunctionBridge";
2import { FunctionRequest } from "@/Neon/FunctionEnvironment";
3import { makeFunctionRuntimeContext } from "@/Neon/FunctionRuntimeContext";
4import { FunctionUpgradeSockets } from "@/Neon/FunctionUpgrade";
5import { expect, test } from "alchemy-test";
6import * as ConfigProvider from "effect/ConfigProvider";
7import * as Deferred from "effect/Deferred";
8import * as Effect from "effect/Effect";
9import * as Stream from "effect/Stream";
10import * as Scope from "effect/Scope";
11import { HttpServerRequest } from "effect/http/HttpServerRequest";
12import * as HttpServerResponse from "effect/http/HttpServerResponse";
13
14for (const cancellation of ["abort", "body"] as const)
15 test.live(
16 `Function bridge closes the handler scope on ${cancellation} cancellation`,
17 () =>
18 Effect.gen(function* () {
19 const finalized = yield* Deferred.make<void>();
20 const streamFinalized = yield* Deferred.make<void>();
21 const runtime = yield* Effect.sync(() =>
22 makeFunctionRuntimeContext("Bridge"),
23 );
24 let closed = false;
25 yield* runtime.route(
26 "/stream",
27 Effect.gen(function* () {
28 const scope = yield* Effect.scope;
29 const request = yield* FunctionRequest;
30 expect((yield* HttpServerRequest).source).toBe(request);
31 yield* Effect.addFinalizer(() =>
32 Effect.sync(() => {
33 closed = true;
34 }).pipe(Effect.andThen(Deferred.succeed(finalized, undefined))),
35 );
36 return HttpServerResponse.stream(
37 Stream.fromEffect(
38 Effect.gen(function* () {
39 expect(yield* Effect.scope).toBe(scope);
40 return yield* Effect.sync(() =>
41 new TextEncoder().encode("first"),
42 );
43 }),
44 ).pipe(
45 Stream.concat(Stream.never),
46 Stream.ensuring(Deferred.succeed(streamFinalized, undefined)),
47 Stream.provideService(Scope.Scope, scope),
48 ),
49 );
50 }),
51 );
52 const bridge = yield* Effect.sync(() =>
53 makeFunctionBridge(Effect.succeed({ RuntimeContext: runtime })),
54 );
55 const controller = yield* Effect.sync(() => new AbortController());
56 const request = yield* Effect.sync(
57 () =>
58 new Request("https://function.test/stream", {
59 signal: controller.signal,
60 }),
61 );
62 const response = yield* Effect.tryPromise(() => bridge.fetch(request));
63 const reader = yield* Effect.sync(() => response.body!.getReader());
64 const first = yield* Effect.tryPromise(() => reader.read());
65 expect(first.done).toBe(false);
66 expect(closed).toBe(false);
67 if (cancellation === "abort")
68 yield* Effect.sync(() => controller.abort());
69 else yield* Effect.tryPromise(() => reader.cancel());
70 yield* Effect.all([
71 Deferred.await(finalized),
72 Deferred.await(streamFinalized),
73 ]).pipe(
74 Effect.timeout("1 second"),
75 Effect.ensuring(
76 Effect.tryPromise(() => reader.cancel()).pipe(Effect.ignore),
77 ),
78 );
79 expect(closed).toBe(true);
80 }),
81 { tags: ["unit", "provider:neon", "provider:neon:function", "local"] },
82 );
83
84for (const cancellation of ["abort", "body"] as const)
85 test.live(
86 `Function bridge preserves backpressure until ${cancellation} cancellation`,
87 () =>
88 Effect.gen(function* () {
89 const runtime = yield* Effect.sync(() =>
90 makeFunctionRuntimeContext("Backpressure"),
91 );
92 const finalized = yield* Deferred.make<void>();
93 const streamFinalized = yield* Deferred.make<void>();
94 let produced = 0;
95 let closed = false;
96 yield* runtime
97 .route(
98 "/",
99 Effect.gen(function* () {
100 yield* Effect.addFinalizer(() =>
101 Effect.sync(() => {
102 closed = true;
103 }).pipe(Effect.andThen(Deferred.succeed(finalized, undefined))),
104 );
105 return HttpServerResponse.stream(
106 Stream.fromEffectRepeat(
107 Effect.sync(() =>
108 new TextEncoder().encode(String(++produced)),
109 ),
110 ).pipe(
111 Stream.ensuring(Deferred.succeed(streamFinalized, undefined)),
112 ),
113 );
114 }),
115 )
116 .pipe(
117 Effect.provideService(
118 ConfigProvider.ConfigProvider,
119 // The removed watchdog must not interpret a paused consumer as a disconnect.
120 ConfigProvider.fromUnknown({
121 ALCHEMY_NEON_STREAM_IDLE_TIMEOUT: "1 millis",
122 }),
123 ),
124 );
125 const bridge = yield* Effect.sync(() =>
126 makeFunctionBridge(Effect.succeed({ RuntimeContext: runtime })),
127 );
128 const controller = yield* Effect.sync(() => new AbortController());
129 const request = yield* Effect.sync(
130 () =>
131 new Request("https://function.test/", {
132 signal: controller.signal,
133 }),
134 );
135 const response = yield* Effect.tryPromise(() => bridge.fetch(request));
136 const reader = yield* Effect.sync(() => response.body!.getReader());
137 yield* Effect.gen(function* () {
138 const first = yield* Effect.tryPromise(() => reader.read());
139 expect(new TextDecoder().decode(first.value)).toBe("1");
140 yield* Effect.sleep("1200 millis");
141 expect(closed).toBe(false);
142 const second = yield* Effect.tryPromise(() => reader.read());
143 expect(new TextDecoder().decode(second.value)).toBe("2");
144 if (cancellation === "abort")
145 yield* Effect.sync(() => controller.abort());
146 else yield* Effect.tryPromise(() => reader.cancel());
147 yield* Deferred.await(streamFinalized).pipe(
148 Effect.timeout("1 second"),
149 );
150 yield* Deferred.await(finalized).pipe(Effect.timeout("1 second"));
151 expect(closed).toBe(true);
152 }).pipe(
153 Effect.ensuring(
154 Effect.tryPromise(() => reader.cancel()).pipe(Effect.ignore),
155 ),
156 );
157 }),
158 { tags: ["unit", "provider:neon", "provider:neon:function", "local"] },
159 );
160
161for (const outcome of ["complete", "error"] as const)
162 test.live(
163 `Function bridge finalizes a streamed ${outcome} once`,
164 () =>
165 Effect.gen(function* () {
166 const runtime = yield* Effect.sync(() =>
167 makeFunctionRuntimeContext("Completion"),
168 );
169 const events: string[] = [];
170 yield* runtime.route(
171 "/",
172 Effect.gen(function* () {
173 yield* Effect.addFinalizer(() =>
174 Effect.sync(() => {
175 events.push("request");
176 }),
177 );
178 return HttpServerResponse.stream(
179 (outcome === "complete"
180 ? Stream.make(new Uint8Array([1, 2, 3]))
181 : Stream.fail(new Error("stream failure"))
182 ).pipe(
183 Stream.ensuring(
184 Effect.sync(() => {
185 events.push("stream");
186 }),
187 ),
188 ),
189 { status: 201, headers: { "x-stream-test": "preserved" } },
190 );
191 }),
192 );
193 const bridge = yield* Effect.sync(() =>
194 makeFunctionBridge(Effect.succeed({ RuntimeContext: runtime })),
195 );
196 const request = yield* Effect.sync(
197 () => new Request("https://function.test/"),
198 );
199 const response = yield* Effect.tryPromise(() => bridge.fetch(request));
200 expect(response.status).toBe(201);
201 expect(response.headers.get("x-stream-test")).toBe("preserved");
202 const result = yield* Effect.tryPromise(() =>
203 response.arrayBuffer(),
204 ).pipe(Effect.exit);
205 expect(result._tag).toBe(
206 outcome === "complete" ? "Success" : "Failure",
207 );
208 expect(events).toEqual(["stream", "request"]);
209 }),
210 { tags: ["unit", "provider:neon", "provider:neon:function", "local"] },
211 );
212
213for (const cancellation of ["close", "abort"] as const)
214 test.live(
215 `Function bridge preserves upgrade identity and finalizes once on ${cancellation}`,
216 () =>
217 Effect.gen(function* () {
218 const runtime = yield* Effect.sync(() =>
219 makeFunctionRuntimeContext("Upgrade"),
220 );
221 const finalized = yield* Deferred.make<void>();
222 const release = yield* Deferred.make<void>();
223 let count = 0;
224 const socket = yield* Effect.sync(
225 () =>
226 Object.assign(new EventTarget(), {
227 readyState: 1,
228 CLOSED: 3,
229 }) as WebSocket,
230 );
231 const native = yield* Effect.sync(() => new Response(null));
232 yield* Effect.sync(() => FunctionUpgradeSockets.set(native, socket));
233 yield* runtime.route(
234 "/",
235 Effect.gen(function* () {
236 yield* Effect.addFinalizer(() =>
237 Deferred.await(release).pipe(
238 Effect.andThen(
239 Effect.sync(() => {
240 count++;
241 }),
242 ),
243 Effect.andThen(Deferred.succeed(finalized, undefined)),
244 ),
245 );
246 return HttpServerResponse.raw(native);
247 }),
248 );
249 const bridge = yield* Effect.sync(() =>
250 makeFunctionBridge(Effect.succeed({ RuntimeContext: runtime })),
251 );
252 const controller = yield* Effect.sync(() => new AbortController());
253 const request = yield* Effect.sync(
254 () =>
255 new Request("https://function.test/", {
256 signal: controller.signal,
257 }),
258 );
259 const response = yield* Effect.tryPromise(() => bridge.fetch(request));
260 expect(response).toBe(native);
261 expect(count).toBe(0);
262 yield* Effect.sync(() =>
263 cancellation === "close"
264 ? socket.dispatchEvent(new Event("close"))
265 : controller.abort(),
266 );
267 expect(count).toBe(0);
268 yield* Deferred.succeed(release, undefined);
269 yield* Deferred.await(finalized).pipe(Effect.timeout("2 seconds"));
270 yield* Effect.sync(() => {
271 socket.dispatchEvent(new Event("close"));
272 controller.abort();
273 });
274 expect(count).toBe(1);
275 }),
276 { tags: ["unit", "provider:neon", "provider:neon:function", "local"] },
277 );
278
279test.effect(
280 "Function bridge closes bodyless request scopes without consuming a stream",
281 () =>
282 Effect.gen(function* () {
283 const runtime = yield* Effect.sync(() =>
284 makeFunctionRuntimeContext("Bodyless"),
285 );
286 let finalized = 0;
287 let consumed = 0;
288 yield* runtime.route(
289 "/",
290 Effect.gen(function* () {
291 yield* Effect.addFinalizer(() =>
292 Effect.sync(() => {
293 finalized++;
294 }),
295 );
296 return HttpServerResponse.stream(
297 Stream.fromEffect(
298 Effect.sync(() => {
299 consumed++;
300 return new TextEncoder().encode("body");
301 }),
302 ),
303 );
304 }),
305 );
306 const bridge = yield* Effect.sync(() =>
307 makeFunctionBridge(Effect.succeed({ RuntimeContext: runtime })),
308 );
309 const request = yield* Effect.sync(
310 () => new Request("https://function.test/", { method: "HEAD" }),
311 );
312 const response = yield* Effect.tryPromise(() => bridge.fetch(request));
313 expect(response.body).toBe(null);
314 expect(finalized).toBe(1);
315 expect(consumed).toBe(0);
316 }),
317 { tags: ["unit", "provider:neon", "provider:neon:function", "local"] },
318);