1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { waitUntil as nativeWaitUntil } from "@neon/functions";
3import * as ConfigProvider from "effect/ConfigProvider";
4import * as Context from "effect/Context";
5import * as Deferred from "effect/Deferred";
6import * as Effect from "effect/Effect";
7import * as Exit from "effect/Exit";
8import * as Layer from "effect/Layer";
9import * as Scope from "effect/Scope";
10import * as FetchHttpClient from "effect/http/FetchHttpClient";
11import * as EffectHttp from "effect/http/HttpEffect";
12import * as HttpServerRequest from "effect/http/HttpServerRequest";
13import * as HttpServerResponse from "effect/http/HttpServerResponse";
14import { safeHttpEffect } from "../Http.ts";
15import { makeEntrypointLayer, reifyBoundConfigProvider } from "../Runtime.ts";
16import { RuntimeContext } from "../RuntimeContext.ts";
17import { Self } from "../Self.ts";
18import { StackContext } from "../StackContext.ts";
19import { FunctionEnvironment, FunctionRequest } from "./FunctionEnvironment.ts";
20import type { FunctionRuntimeContext } from "./FunctionRuntimeContext.ts";
21import { FunctionUpgradeSockets } from "./FunctionUpgrade.ts";
22
23const closeRequestScope = (scope: Scope.Closeable) => {
24 nativeWaitUntil(
25 Effect.runPromise(
26 Scope.close(scope, Exit.void).pipe(
27 Effect.timeout("15 seconds"),
28 Effect.ignoreCause({ log: "Error" }),
29 ),
30 ),
31 );
32};
33
34/** Build a Node-only Fetch bridge once per process, with a fresh scope for every request. */
35export const makeFunctionBridge = (entrypoint: unknown) => {
36 const instanceScope = Scope.makeUnsafe();
37 const tag = Self as unknown as Context.Service<
38 never,
39 { RuntimeContext: FunctionRuntimeContext }
40 >;
41 const platform = Layer.mergeAll(
42 NodeServices.layer,
43 FetchHttpClient.layer,
44 Layer.succeed(FunctionEnvironment, process.env),
45 Layer.succeed(
46 ConfigProvider.ConfigProvider,
47 reifyBoundConfigProvider(
48 ConfigProvider.orElse(
49 ConfigProvider.fromUnknown({ ALCHEMY_PHASE: "runtime" }),
50 ConfigProvider.fromEnv(),
51 ),
52 process.env,
53 ),
54 ),
55 Layer.succeed(StackContext, {
56 name: process.env.ALCHEMY_STACK_NAME ?? "NeonFunction",
57 stage: process.env.ALCHEMY_STAGE ?? "runtime",
58 bindings: {},
59 resources: {},
60 actions: {},
61 }),
62 );
63 const build = Layer.buildWithScope(
64 makeEntrypointLayer(tag, entrypoint).pipe(Layer.provideMerge(platform)),
65 instanceScope,
66 ).pipe(
67 Effect.flatMap((context) =>
68 tag.pipe(
69 Effect.flatMap((host) =>
70 host.RuntimeContext.handler.pipe(
71 Effect.map((handler) => ({
72 ...handler,
73 runtime: host.RuntimeContext,
74 built: context,
75 })),
76 ),
77 ),
78 Effect.provideContext(context),
79 ),
80 ),
81 Effect.cachedWithTTL((exit) => (Exit.isSuccess(exit) ? Infinity : 0)),
82 Effect.runSync,
83 );
84 const close = Scope.close(instanceScope, Exit.void).pipe(
85 Effect.timeout("4 seconds"),
86 Effect.ignoreCause({ log: "Error" }),
87 );
88 process.once("SIGINT", () => {
89 Effect.runFork(close);
90 });
91
92 return {
93 fetch: (request: Request) =>
94 Effect.gen(function* () {
95 const built = yield* build;
96 const response = yield* Deferred.make<Response>();
97 const services = Context.mergeAll(
98 built.built,
99 built.context,
100 Context.make(RuntimeContext, built.runtime),
101 Context.make(FunctionRequest, request),
102 Context.make(
103 HttpServerRequest.HttpServerRequest,
104 HttpServerRequest.fromWeb(request),
105 ),
106 );
107 const handler = safeHttpEffect(
108 built.dispatch(new URL(request.url).pathname),
109 ).pipe(Effect.interruptible);
110 return yield* EffectHttp.toHandled(handler, (req, res) =>
111 Effect.gen(function* () {
112 const scope = yield* Effect.scope;
113 const context = yield* Effect.context<never>();
114 if (res.body._tag === "Raw" && res.body.body instanceof Response) {
115 // Neon attaches upgrade metadata to this exact object.
116 const socket = FunctionUpgradeSockets.get(res.body.body);
117 if (socket && socket.readyState !== socket.CLOSED) {
118 EffectHttp.scopeDisableClose(scope);
119 const onClose = () =>
120 closeRequestScope(scope as Scope.Closeable);
121 socket.addEventListener("close", onClose, { once: true });
122 request.signal.addEventListener("abort", onClose, {
123 once: true,
124 });
125 yield* Effect.addFinalizer(() =>
126 Effect.sync(() => {
127 socket.removeEventListener("close", onClose);
128 request.signal.removeEventListener("abort", onClose);
129 }),
130 );
131 if (request.signal.aborted) onClose();
132 }
133 yield* Deferred.succeed(response, res.body.body);
134 return;
135 }
136 const withoutBody = req.method === "HEAD";
137 const transferred = HttpServerResponse.omitsBody(res, withoutBody)
138 ? res
139 : EffectHttp.scopeTransferToStream(res);
140 const web = HttpServerResponse.toWeb(transferred, {
141 withoutBody,
142 context,
143 });
144 // The body outlives fetch; propagate actual request aborts to its producer.
145 yield* Deferred.succeed(
146 response,
147 res.body._tag === "Stream" && web.body
148 ? new Response(
149 web.body.pipeThrough(
150 new TransformStream<Uint8Array, Uint8Array>(),
151 {
152 signal: request.signal,
153 },
154 ),
155 {
156 status: web.status,
157 statusText: web.statusText,
158 headers: web.headers,
159 },
160 )
161 : web,
162 );
163 }),
164 ).pipe(
165 Effect.andThen(Deferred.await(response)),
166 Effect.provideContext(services),
167 );
168 }).pipe((effect) =>
169 Effect.runPromise(effect, { signal: request.signal }),
170 ),
171 };
172};