1import * as Neon from "@distilled.cloud/neon";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import * as Stream from "effect/Stream";
6import { Unowned } from "../AdoptPolicy.ts";
7import { isResolved } from "../Diff.ts";
8import * as ProviderLayer from "../Local/ProviderLayer.ts";
9import * as Provider from "../Provider.ts";
10import { Stack } from "../Stack.ts";
11import { Stage } from "../Stage.ts";
12import { sha256Object } from "../Util/sha256.ts";
13import { resolveBranchScope, type ResolvedBranchScope } from "./BranchScope.ts";
14import { Function, type FunctionAttributes } from "./Function.ts";
15import { buildFunctionArtifact } from "./FunctionArtifact.ts";
16import { functionEnvironment, functionSlug } from "./FunctionConfig.ts";
17import { LocalFunctionProvider } from "./LocalFunctionProvider.ts";
18
19export class FunctionDeploymentFailed extends Data.TaggedError(
20 "FunctionDeploymentFailed",
21)<{
22 slug: string;
23 deploymentId: number;
24 status: string;
25 message?: string;
26}> {}
27export class FunctionDeploymentNotReady extends Data.TaggedError(
28 "FunctionDeploymentNotReady",
29)<{ slug: string; deploymentId: number }> {}
30
31export const observeFunction = Effect.fn(function* (
32 scope: ResolvedBranchScope,
33 slug: string,
34) {
35 let cursor: string | undefined;
36 const seen = new Set<string>();
37 do {
38 const page = yield* Neon.listProjectBranchFunctions({
39 project_id: scope.projectId,
40 branch_id: scope.branchId,
41 cursor,
42 });
43 const found = page.functions.find((fn) => fn.slug === slug);
44 if (found) return found;
45 cursor = page.pagination?.next;
46 if (cursor && seen.has(cursor))
47 return yield* new FunctionDeploymentFailed({
48 slug,
49 deploymentId: 0,
50 status: "repeated-pagination-cursor",
51 message: "Neon Function listing returned a repeated pagination cursor",
52 });
53 if (cursor) seen.add(cursor);
54 } while (cursor);
55 return undefined;
56});
57
58const attributes = (
59 scope: ResolvedBranchScope,
60 fn: Neon.NeonFunction,
61 codeHash?: string,
62 environmentHash?: string,
63): FunctionAttributes => ({
64 ...scope,
65 functionId: fn.id,
66 slug: fn.slug,
67 name: fn.name,
68 url: fn.invocation_url,
69 currentDeploymentId: fn.current_deployment?.id,
70 activeDeploymentId: fn.active_deployment?.id,
71 status: fn.current_deployment?.status,
72 codeHash,
73 environmentHash,
74 environment: fn.active_deployment?.environment ?? [],
75});
76
77export const waitForFunctionDeployment = (
78 scope: ResolvedBranchScope,
79 slug: string,
80 deploymentId: number,
81) =>
82 Effect.gen(function* () {
83 const { function: fn } = yield* Neon.getProjectBranchFunction({
84 project_id: scope.projectId,
85 branch_id: scope.branchId,
86 slug,
87 });
88 if (
89 fn.current_deployment?.id === deploymentId &&
90 fn.current_deployment.status === "failed"
91 ) {
92 return yield* new FunctionDeploymentFailed({
93 slug,
94 deploymentId,
95 status: "failed",
96 message:
97 fn.current_deployment.error ?? "Neon Function deployment failed",
98 });
99 }
100 if (
101 fn.active_deployment?.id !== deploymentId ||
102 fn.active_deployment.status !== "completed" ||
103 !fn.invocation_url.startsWith("https://")
104 ) {
105 return yield* new FunctionDeploymentNotReady({ slug, deploymentId });
106 }
107 return fn;
108 }).pipe(
109 Effect.retry({
110 while: (error) => error._tag === "FunctionDeploymentNotReady",
111 schedule: Schedule.spaced("5 seconds"),
112 times: 9,
113 }),
114 );
115
116export class FunctionLogQueryError extends Data.TaggedError(
117 "FunctionLogQueryError",
118)<{
119 reason: "invalid-limit" | "invalid-cursor" | "pagination-limit";
120}> {}
121
122export const FunctionLogs = Effect.fn(function* (
123 output: FunctionAttributes,
124 options: Provider.LogsInput,
125) {
126 const limit = options.limit ?? 100;
127 if (!Number.isInteger(limit) || limit < 1 || limit > 8000)
128 return yield* new FunctionLogQueryError({ reason: "invalid-limit" });
129 // https://neon.com/docs/introduction/monitor-logs documents this service identity.
130 const service = `neon-function/${output.slug}`;
131 const query = yield* Effect.sync(() => ({
132 project_id: output.projectId,
133 branch_id: output.branchId,
134 source: "function" as const,
135 service_name: service,
136 ...(options.since
137 ? { start_time: options.since.toISOString() }
138 : { since: "1h" }),
139 end_time: new Date().toISOString(),
140 limit: Math.min(limit, 1000),
141 sort_order: "desc" as const,
142 }));
143 const records: Neon.ProjectBranchLogRecord[] = [];
144 const cursors = new Set<string>();
145 let cursor: string | undefined;
146 for (let page = 0; page < 8; page++) {
147 const response = yield* Neon.queryProjectBranchLogs({ ...query, cursor });
148 records.push(
149 ...response.logs.filter(
150 (line) => line.source === "function" && line.service_name === service,
151 ),
152 );
153 if (records.length >= limit || !response.is_truncated)
154 return yield* Effect.sync(() =>
155 records
156 .slice(0, limit)
157 .reverse()
158 .map((line) => ({
159 timestamp: new Date(line.timestamp),
160 message: line.message,
161 })),
162 );
163 cursor = response.next_cursor;
164 if (!cursor || cursors.has(cursor))
165 return yield* new FunctionLogQueryError({ reason: "invalid-cursor" });
166 cursors.add(cursor);
167 }
168 return yield* new FunctionLogQueryError({ reason: "pagination-limit" });
169}, Effect.timeout("45 seconds"));
170
171export const FunctionProvider = () =>
172 ProviderLayer.dual(Function, {
173 live: FunctionProviderLive,
174 local: LocalFunctionProvider,
175 });
176
177export const FunctionProviderLive = () =>
178 Provider.succeed(Function, {
179 stables: ["projectId", "branchId", "functionId", "slug", "url"],
180 list: Effect.fn(function* () {
181 const result: FunctionAttributes[] = [];
182 for (const project of yield* Neon.listProjects
183 .items({})
184 .pipe(Stream.runCollect)) {
185 for (const branch of yield* Neon.listProjectBranches
186 .items({ project_id: project.id })
187 .pipe(Stream.runCollect)) {
188 const scope = { projectId: project.id, branchId: branch.id };
189 for (const fn of yield* Neon.listProjectBranchFunctions
190 .items({ project_id: project.id, branch_id: branch.id })
191 .pipe(Stream.runCollect))
192 result.push(attributes(scope, fn));
193 }
194 }
195 return result;
196 }),
197 diff: Effect.fn(function* ({ id, news, output }) {
198 if (!isResolved(news)) return;
199 const scope = yield* resolveBranchScope(news);
200 if (
201 output &&
202 (scope.projectId !== output.projectId ||
203 scope.branchId !== output.branchId ||
204 (news.slug !== undefined && news.slug !== output.slug))
205 )
206 return { action: "replace" };
207 yield* functionSlug(id, news.slug);
208 const artifact = yield* buildFunctionArtifact(news);
209 if (artifact.codeHash !== output?.codeHash) return { action: "update" };
210 }),
211 read: Effect.fn(function* ({ id, olds, output }) {
212 if (
213 !output &&
214 !(olds?.branch?.projectId && olds.branch.branchId) &&
215 !olds?.project?.projectId
216 )
217 return undefined;
218 const scope = output ?? (yield* resolveBranchScope(olds!));
219 const slug = output?.slug ?? (yield* functionSlug(id, olds?.slug));
220 const found = yield* observeFunction(scope, slug);
221 if (!found) return undefined;
222 const attrs = attributes(
223 scope,
224 found,
225 output?.activeDeploymentId === found.active_deployment?.id
226 ? output?.codeHash
227 : undefined,
228 output?.activeDeploymentId === found.active_deployment?.id
229 ? output?.environmentHash
230 : undefined,
231 );
232 return output ? attrs : Unowned(attrs);
233 }),
234 reconcile: Effect.fn(function* ({ id, news, output, bindings }) {
235 const scope = yield* resolveBranchScope(news);
236 const slug = news.slug ?? output?.slug ?? (yield* functionSlug(id));
237 yield* functionSlug(id, slug);
238 const observed = yield* observeFunction(scope, slug);
239 const artifact = yield* buildFunctionArtifact(news);
240 const env = yield* functionEnvironment(news, bindings);
241 const stack = yield* Stack;
242 const stage = yield* Stage;
243 const managed = {
244 ...env,
245 ALCHEMY_STACK_NAME: stack.name,
246 ALCHEMY_STAGE: stage,
247 ALCHEMY_PHASE: "runtime",
248 };
249 const environment = {
250 ...Object.fromEntries(
251 (observed?.active_deployment?.environment ?? [])
252 .filter((key) => !(key in managed))
253 .map((key) => [key, ""]),
254 ),
255 ...managed,
256 };
257 const unchangedCode =
258 observed !== undefined &&
259 output?.activeDeploymentId === observed.active_deployment?.id &&
260 output?.codeHash === artifact.codeHash;
261 const environmentHash = yield* sha256Object(managed);
262 const unchangedEnvironment =
263 output?.activeDeploymentId === observed?.active_deployment?.id &&
264 output?.environmentHash === environmentHash;
265 let current = observed;
266 if (!current || !unchangedCode || !unchangedEnvironment) {
267 const zip = unchangedCode
268 ? undefined
269 : yield* Effect.sync(
270 () =>
271 new File([new Uint8Array(artifact.archive)], "function.zip", {
272 type: "application/zip",
273 }),
274 );
275 const deployment = yield* Neon.createProjectBranchFunctionDeployment({
276 project_id: scope.projectId,
277 branch_id: scope.branchId,
278 slug,
279 runtime: "nodejs24",
280 zip,
281 environment: JSON.stringify(environment),
282 });
283 current = yield* waitForFunctionDeployment(
284 scope,
285 slug,
286 deployment.deployment.id,
287 );
288 }
289 if (current.name !== (news.name ?? slug)) {
290 const updated = yield* Neon.updateProjectBranchFunction({
291 project_id: scope.projectId,
292 branch_id: scope.branchId,
293 slug,
294 name: news.name ?? null,
295 });
296 current = updated.function;
297 }
298 return attributes(scope, current, artifact.codeHash, environmentHash);
299 }),
300 delete: Effect.fn(function* ({ output }) {
301 if (!(yield* observeFunction(output, output.slug))) return;
302 yield* Neon.deleteProjectBranchFunction({
303 project_id: output.projectId,
304 branch_id: output.branchId,
305 slug: output.slug,
306 }).pipe(Effect.catchTag("NotFound", () => Effect.void));
307 }),
308 logs: ({ output, options }) => FunctionLogs(output, options),
309 tail: ({ output }) =>
310 Stream.unwrap(
311 Effect.sync(() => {
312 const seen = new Set<string>();
313 return Stream.fromEffectSchedule(
314 FunctionLogs(output, { limit: 1000 }),
315 Schedule.spaced("3 seconds"),
316 ).pipe(
317 Stream.flatMap((lines) => Stream.fromIterable(lines)),
318 Stream.filter((line) => {
319 const key = `${line.timestamp.toISOString()}\0${line.message}`;
320 if (seen.has(key)) return false;
321 seen.add(key);
322 if (seen.size > 1000) seen.delete(seen.values().next().value!);
323 return true;
324 }),
325 );
326 }),
327 ),
328 });