1import * as Alchemy from "@/index";
2import { Function } from "@/Neon/Function";
3import { NeonAuth } from "@/Neon/AuthProvider";
4import { fromAuthProvider } from "@/Neon/Credentials";
5import * as Layer from "effect/Layer";
6import * as NeonApi from "@distilled.cloud/neon";
7import * as Exit from "effect/Exit";
8import { providers } from "@/Neon/Providers";
9import { FunctionLogs } from "@/Neon/FunctionProvider";
10import * as Test from "@/Test/Alchemy";
11import { expect } from "alchemy-test";
12import * as Effect from "effect/Effect";
13import * as Schedule from "effect/Schedule";
14import * as Option from "effect/Option";
15import * as Stream from "effect/Stream";
16import * as ChildProcess from "effect/process/ChildProcess";
17import * as HttpClient from "effect/http/HttpClient";
18import RuntimeFunction from "./fixtures/function-effect.ts";
19
20const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
21 providers: providers(),
22});
23const Stack = Alchemy.Stack(
24 "NeonFunctionRuntime",
25 { providers: providers(), state: Alchemy.localState() },
26 Effect.gen(function* () {
27 const api = yield* RuntimeFunction;
28 const main = yield* Effect.sync(
29 () =>
30 new URL("./fixtures/function-native-lifecycle.ts", import.meta.url)
31 .href,
32 );
33 const native = yield* Function("NativeLifecycle", {
34 branch: { projectId: api.projectId, branchId: api.branchId },
35 main,
36 });
37 return { url: api.url, api, native };
38 }),
39);
40const stack = beforeAll(destroy(Stack).pipe(Effect.andThen(deploy(Stack))));
41afterAll(
42 Effect.gen(function* () {
43 const outputs = yield* Effect.exit(stack);
44 yield* destroy(Stack);
45 if (Exit.isFailure(outputs) || !outputs.value) return;
46 const { api, native } = outputs.value;
47 for (const resource of [api, native]) {
48 expect(
49 yield* NeonApi.getProjectBranchFunction({
50 project_id: resource.projectId,
51 branch_id: resource.branchId,
52 slug: resource.slug,
53 }).pipe(
54 Effect.as(false),
55 Effect.catchTag("NotFound", () => Effect.succeed(true)),
56 ),
57 ).toBe(true);
58 }
59 expect(
60 yield* NeonApi.getProject({ project_id: api.projectId }).pipe(
61 Effect.as(false),
62 Effect.catchTag("NotFound", () => Effect.succeed(true)),
63 ),
64 ).toBe(true);
65 yield* Effect.logInfo(
66 "Neon lifecycle cleanup: both functions and project independently absent",
67 );
68 }).pipe(Effect.provide(fromAuthProvider().pipe(Layer.provide(NeonAuth)))),
69);
70const reportRuntimeLogs = Effect.gen(function* () {
71 const { api, native } = yield* stack;
72 const client = yield* HttpClient.HttpClient;
73 for (const resource of [api, native]) {
74 const lifecycle = yield* client
75 .get(`${resource.url}diagnostics`)
76 .pipe(Effect.flatMap((response) => response.json));
77 yield* Effect.logInfo(
78 JSON.stringify({ native: resource === native, lifecycle }),
79 );
80 const lines = yield* FunctionLogs(resource, { limit: 1000 });
81 yield* Effect.logInfo(
82 JSON.stringify({
83 native: resource === native,
84 functionRuntimeLogs: lines,
85 }),
86 );
87 }
88});
89
90test(
91 "Effect class preserves request isolation, streams and bodyless finalizers",
92 Effect.gen(function* () {
93 const { url } = yield* stack;
94 const client = yield* HttpClient.HttpClient;
95 expect(yield* (yield* client.get(url)).text).toBe("effect");
96 const stream = yield* client.get(`${url}stream?id=stream`);
97 expect(yield* stream.text).toBe("data: first\n\ndata: second\n\n");
98 expect((yield* client.head(`${url}?id=head`)).status).toBe(200);
99 expect((yield* client.get(`${url}empty?id=empty`)).status).toBe(204);
100 const parallel = yield* Effect.all(
101 Array.from({ length: 4 }, (_, i) =>
102 client
103 .get(`${url}slow?id=parallel${i}`)
104 .pipe(Effect.flatMap((response) => response.text)),
105 ),
106 { concurrency: 4 },
107 );
108 expect(parallel).toEqual(["effect", "effect", "effect", "effect"]);
109 const finalized = yield* client
110 .get(`${url}finalized`)
111 .pipe(Effect.flatMap((response) => response.json));
112 expect(finalized).toMatchObject({ active: 0 });
113 expect(finalized).toMatchObject({
114 finalized: expect.arrayContaining([
115 "stream",
116 "head",
117 "empty",
118 "parallel0",
119 "parallel1",
120 "parallel2",
121 "parallel3",
122 ]),
123 });
124 }),
125 {
126 tags: [
127 "provider:neon",
128 "provider:neon:function",
129 "provider:neon:project",
130 "live",
131 ],
132 timeout: 120_000,
133 },
134);
135
136test(
137 "waitUntil outlives the response with an independent request scope",
138 Effect.gen(function* () {
139 const { url } = yield* stack;
140 const client = yield* HttpClient.HttpClient;
141 expect(
142 yield* (yield* client.get(`${url}background?id=background-request`)).text,
143 ).toBe("scheduled");
144 const completed = yield* client.get(`${url}finalized`).pipe(
145 Effect.flatMap((response) => response.json),
146 Effect.repeat({
147 schedule: Schedule.spaced("500 millis"),
148 times: 8,
149 until: (body) => JSON.stringify(body).includes("background-work"),
150 }),
151 );
152 expect(completed).toMatchObject({
153 finalized: expect.arrayContaining([
154 "background-request",
155 "background-work",
156 ]),
157 });
158 }),
159 {
160 tags: [
161 "provider:neon",
162 "provider:neon:function",
163 "provider:neon:project",
164 "live",
165 ],
166 timeout: 120_000,
167 },
168);
169
170// Waits up to two minutes for durable finalizer evidence; skip under --fast.
171test.provider.skipIf(!!process.env.FAST)(
172 "WebSocket upgrade preserves native metadata and closes the request scope",
173 () =>
174 Effect.gen(function* () {
175 const { url, native } = yield* stack;
176 for (const [target, id] of [
177 [native.url, "native-websocket"],
178 [url, "websocket"],
179 ]) {
180 const echoed = yield* Effect.callback<string, Error>((resume) => {
181 const socket = new WebSocket(
182 `${target.replace(/^http/, "ws")}websocket?id=${id}`,
183 );
184 let message: string | undefined;
185 socket.addEventListener("open", () => socket.send("native-upgrade"));
186 socket.addEventListener(
187 "message",
188 (event) => {
189 message = String(event.data);
190 socket.close(1000);
191 },
192 { once: true },
193 );
194 socket.addEventListener(
195 "close",
196 (event) =>
197 resume(
198 message !== undefined && event.wasClean
199 ? Effect.succeed(message)
200 : Effect.fail(
201 new Error(
202 `Neon WebSocket closed unexpectedly (${event.code})`,
203 ),
204 ),
205 ),
206 { once: true },
207 );
208 socket.addEventListener(
209 "error",
210 () =>
211 resume(Effect.fail(new Error("Neon WebSocket handshake failed"))),
212 { once: true },
213 );
214 return Effect.sync(() => socket.close());
215 }).pipe(Effect.timeout("15 seconds"));
216 expect(echoed).toBe("native-upgrade");
217 }
218 const client = yield* HttpClient.HttpClient;
219 const completed = yield* client.get(`${url}finalized`).pipe(
220 Effect.flatMap((response) => response.json),
221 Effect.repeat({
222 schedule: Schedule.spaced("1 second"),
223 until: (body) => JSON.stringify(body).includes("websocket"),
224 }),
225 Effect.timeout("2 minutes"),
226 );
227 const lifecycle = yield* (yield* client.get(`${native.url}diagnostics`))
228 .json;
229 yield* Effect.logInfo(JSON.stringify({ lifecycle }));
230 expect(lifecycle).toEqual(
231 expect.arrayContaining([{ id: "native-websocket", phase: "close" }]),
232 );
233 expect(completed).toMatchObject({
234 finalized: expect.arrayContaining(["websocket"]),
235 });
236 }).pipe(Effect.ensuring(reportRuntimeLogs.pipe(Effect.orDie))),
237 {
238 tags: [
239 "provider:neon",
240 "provider:neon:function",
241 "provider:neon:project",
242 "live",
243 ],
244 timeout: 180_000,
245 },
246);
247
248// Waits up to two minutes for durable finalizer evidence; skip under --fast.
249test.provider.skipIf(!!process.env.FAST)(
250 "cancelling a streamed response releases its request scope",
251 () =>
252 Effect.gen(function* () {
253 const { url, native } = yield* stack;
254 for (const [target, id, suffix] of [
255 [native.url, "native-stream", ""],
256 [url, "cancelled-stream", ""],
257 [native.url, "native-sse", "&sse"],
258 [url, "cancelled-sse", "&sse"],
259 ]) {
260 yield* Effect.gen(function* () {
261 const child = yield* ChildProcess.make("curl", [
262 "--silent",
263 "--show-error",
264 "--fail",
265 "--no-buffer",
266 "--http1.1",
267 "--noproxy",
268 "*",
269 "--max-time",
270 "15",
271 "-H",
272 "Connection: close",
273 `${target}stream-cancel?id=${id}${suffix}`,
274 ]);
275 const first = yield* child.stdout.pipe(
276 Stream.decodeText,
277 Stream.filter((chunk) => chunk.includes("tick")),
278 Stream.runHead,
279 Effect.timeout("10 seconds"),
280 );
281 expect(Option.isSome(first)).toBe(true);
282 // Terminating the client proves the transport closed, not just its reader.
283 if (yield* child.isRunning) yield* child.kill();
284 expect(yield* child.isRunning).toBe(false);
285 }).pipe(Effect.scoped);
286 }
287 const client = yield* HttpClient.HttpClient;
288 const completed = yield* client.get(`${url}finalized`).pipe(
289 Effect.flatMap((response) => response.json),
290 Effect.repeat({
291 schedule: Schedule.spaced("1 second"),
292 until: (body) =>
293 ["cancelled-stream", "cancelled-sse"].every((id) =>
294 JSON.stringify(body).includes(id),
295 ),
296 }),
297 Effect.timeout("2 minutes"),
298 );
299 const lifecycle = (yield* (yield* client.get(`${native.url}diagnostics`))
300 .json) as { id: string; phase: string }[];
301 yield* Effect.logInfo(JSON.stringify({ lifecycle }));
302 for (const id of ["native-stream", "native-sse"]) {
303 expect(lifecycle).toEqual(
304 expect.arrayContaining([{ id, phase: "entered" }]),
305 );
306 expect(
307 lifecycle.some(
308 (row) =>
309 row.id === id &&
310 (row.phase === "abort" || row.phase === "cancel"),
311 ),
312 ).toBe(true);
313 }
314 expect(completed).toMatchObject({
315 finalized: expect.arrayContaining([
316 "cancelled-stream",
317 "cancelled-sse",
318 ]),
319 });
320 }).pipe(Effect.ensuring(reportRuntimeLogs.pipe(Effect.orDie))),
321 {
322 tags: [
323 "provider:neon",
324 "provider:neon:function",
325 "provider:neon:project",
326 "live",
327 ],
328 timeout: 180_000,
329 },
330);
331
332test(
333 "Effect defects produce non-success responses without leaking errors",
334 Effect.gen(function* () {
335 const { url } = yield* stack;
336 const client = yield* HttpClient.HttpClient;
337 const response = yield* client.get(`${url}error`);
338 expect(response.status).toBe(500);
339 expect(yield* response.text).not.toContain("intentional");
340 }),
341 {
342 tags: [
343 "provider:neon",
344 "provider:neon:function",
345 "provider:neon:project",
346 "live",
347 ],
348 timeout: 120_000,
349 },
350);