1import {
2 upgradeWebSocket,
3 waitUntil as nativeWaitUntil,
4} from "@neon/functions";
5import { Function } from "@/Neon/Function";
6import { FunctionRequest } from "@/Neon/FunctionEnvironment";
7import { waitUntil } from "@/Neon/waitUntil";
8import { upgrade } from "@/Neon/upgrade";
9import { Project } from "@/Neon/Project";
10import * as Config from "effect/Config";
11import * as Effect from "effect/Effect";
12import * as Stream from "effect/Stream";
13import { Postgres } from "@/SQL/Postgres";
14import { HttpServerRequest } from "effect/http/HttpServerRequest";
15import * as HttpServerResponse from "effect/http/HttpServerResponse";
16
17const diagnosticSql = Effect.gen(function* () {
18 return yield* Postgres({
19 url: Config.Redacted("DATABASE_URL"),
20 maxConnections: 1,
21 });
22});
23
24const recordLifecycle = (id: string, phase: string) =>
25 Effect.gen(function* () {
26 const sql = yield* diagnosticSql;
27 yield* sql`INSERT INTO alchemy_function_lifecycle (id, phase) VALUES (${id}, ${phase}) ON CONFLICT DO NOTHING`;
28 }).pipe(Effect.scoped);
29
30const recordNativeLifecycle = (id: string, phase: string) => {
31 console.info(JSON.stringify({ nativeLifecycle: { id, phase } }));
32 nativeWaitUntil(Effect.runPromise(recordLifecycle(id, phase)));
33};
34
35// This handler bypasses makeFunctionBridge entirely.
36export const nativeDiagnosticFetch = (request: Request) =>
37 Effect.gen(function* () {
38 const sql = yield* diagnosticSql;
39 yield* sql`CREATE TABLE IF NOT EXISTS alchemy_function_lifecycle (id text, phase text, PRIMARY KEY (id, phase))`;
40 const url = yield* Effect.sync(() => new URL(request.url));
41 const id = url.searchParams.get("id") ?? "native";
42 if (url.pathname === "/diagnostics") {
43 const rows =
44 yield* sql`SELECT id, phase FROM alchemy_function_lifecycle ORDER BY id, phase`;
45 return yield* Effect.sync(() => Response.json(rows));
46 }
47 yield* recordLifecycle(id, "entered");
48 yield* Effect.sync(() =>
49 request.signal.addEventListener(
50 "abort",
51 () => recordNativeLifecycle(id, "abort"),
52 { once: true },
53 ),
54 );
55 if (url.pathname === "/websocket") {
56 return yield* Effect.sync(() => {
57 const { socket, response } = upgradeWebSocket(request);
58 socket.addEventListener("open", () =>
59 recordNativeLifecycle(id, "open"),
60 );
61 socket.addEventListener("message", (event) => socket.send(event.data));
62 socket.addEventListener(
63 "close",
64 () => recordNativeLifecycle(id, "close"),
65 { once: true },
66 );
67 socket.addEventListener(
68 "error",
69 () => recordNativeLifecycle(id, "error"),
70 { once: true },
71 );
72 return response;
73 });
74 }
75 return yield* Effect.sync(() => {
76 const stream = new ReadableStream<Uint8Array>({
77 pull(controller) {
78 return Effect.runPromise(
79 Effect.sleep("100 millis").pipe(
80 Effect.andThen(
81 Effect.sync(() => {
82 controller.enqueue(
83 new TextEncoder().encode("data: tick\n\n"),
84 );
85 }),
86 ),
87 ),
88 );
89 },
90 cancel() {
91 recordNativeLifecycle(id, "cancel");
92 },
93 });
94 return new Response(stream, {
95 headers: url.searchParams.has("sse")
96 ? {
97 "content-type": "text/event-stream",
98 "cache-control": "no-cache, no-transform",
99 }
100 : undefined,
101 });
102 });
103 }).pipe(Effect.scoped, Effect.runPromise);
104
105export default class RuntimeFunction extends Function<RuntimeFunction>()(
106 "RuntimeFunction",
107 Effect.gen(function* () {
108 const project = yield* Project("RuntimeProject", {
109 region: "aws-us-east-2",
110 });
111 return {
112 project,
113 main: import.meta.url,
114 env: { FUNCTION_MESSAGE: "effect" },
115 };
116 }),
117 Effect.gen(function* () {
118 const message = yield* Config.String("FUNCTION_MESSAGE").pipe(
119 Config.withDefault("effect"),
120 );
121 const sql = yield* Postgres({
122 url: Config.Redacted("DATABASE_URL"),
123 maxConnections: 1,
124 });
125 let active = 0;
126 return {
127 fetch: Effect.gen(function* () {
128 const request = yield* HttpServerRequest;
129 const url = new URL(request.url, "https://function.test");
130 yield* sql`CREATE TABLE IF NOT EXISTS alchemy_function_finalizers (id text PRIMARY KEY)`.pipe(
131 Effect.orDie,
132 );
133 yield* sql`CREATE TABLE IF NOT EXISTS alchemy_function_lifecycle (id text, phase text, PRIMARY KEY (id, phase))`.pipe(
134 Effect.orDie,
135 );
136 if (url.pathname === "/diagnostics") {
137 const rows =
138 yield* sql`SELECT id, phase FROM alchemy_function_lifecycle ORDER BY id, phase`.pipe(
139 Effect.orDie,
140 );
141 return yield* HttpServerResponse.json(rows);
142 }
143 if (url.pathname === "/finalized") {
144 const rows = yield* sql<{
145 id: string;
146 }>`SELECT id FROM alchemy_function_finalizers`.pipe(Effect.orDie);
147 return yield* HttpServerResponse.json({
148 finalized: rows.map((row) => row.id),
149 active,
150 });
151 }
152 const id = url.searchParams.get("id") ?? "default";
153 yield* Effect.sync(() => {
154 active++;
155 });
156 yield* recordLifecycle(id, "entered").pipe(Effect.orDie);
157 const report = (phase: string) =>
158 Effect.sync(() =>
159 console.info(JSON.stringify({ functionFinalizer: { id, phase } })),
160 ).pipe(Effect.andThen(recordLifecycle(id, phase)), Effect.orDie);
161 yield* Effect.addFinalizer(() =>
162 report("started").pipe(
163 Effect.andThen(
164 sql`INSERT INTO alchemy_function_finalizers (id) VALUES (${id}) ON CONFLICT DO NOTHING`,
165 ),
166 Effect.tapCause(() => report("failed")),
167 Effect.orDie,
168 Effect.andThen(
169 Effect.sync(() => {
170 active--;
171 }),
172 ),
173 Effect.andThen(report("completed")),
174 ),
175 );
176 const nativeRequest = yield* FunctionRequest;
177 const onAbort = () => Effect.runFork(report("request-aborted"));
178 yield* Effect.sync(() =>
179 nativeRequest.signal.addEventListener("abort", onAbort, {
180 once: true,
181 }),
182 );
183 yield* Effect.addFinalizer(() =>
184 Effect.sync(() =>
185 nativeRequest.signal.removeEventListener("abort", onAbort),
186 ),
187 );
188 if (url.pathname === "/background") {
189 yield* waitUntil(
190 Effect.sleep("250 millis").pipe(
191 Effect.andThen(
192 sql`INSERT INTO alchemy_function_finalizers (id) VALUES ('background-work') ON CONFLICT DO NOTHING`,
193 ),
194 ),
195 );
196 return HttpServerResponse.text("scheduled");
197 }
198 if (url.pathname === "/websocket") {
199 const { socket, response } = yield* upgrade();
200 yield* Effect.sync(() => {
201 socket.addEventListener("message", (event) =>
202 socket.send(event.data),
203 );
204 socket.addEventListener(
205 "close",
206 () => Effect.runFork(report("socket-closed")),
207 { once: true },
208 );
209 });
210 return response;
211 }
212 if (url.pathname === "/stream-cancel")
213 return HttpServerResponse.stream(
214 Stream.fromEffectRepeat(
215 Effect.sleep("100 millis").pipe(Effect.as("tick")),
216 ).pipe(
217 Stream.encodeText,
218 Stream.ensuring(report("stream-released")),
219 ),
220 {
221 headers: url.searchParams.has("sse")
222 ? {
223 "content-type": "text/event-stream",
224 "cache-control": "no-cache, no-transform",
225 }
226 : undefined,
227 },
228 );
229 if (url.pathname === "/error")
230 return yield* Effect.die(
231 new Error("intentional effect function failure"),
232 );
233 if (url.pathname === "/empty")
234 return HttpServerResponse.empty({ status: 204 });
235 if (url.pathname === "/stream")
236 return HttpServerResponse.stream(
237 Stream.make("data: first\n\n", "data: second\n\n").pipe(
238 Stream.encodeText,
239 ),
240 { headers: { "content-type": "text/event-stream" } },
241 );
242 if (url.pathname === "/slow") yield* Effect.sleep("200 millis");
243 return HttpServerResponse.text(message);
244 }),
245 };
246 }),
247) {}