Open History to see the last commit
1import { AlchemyContext } from "alchemy/AlchemyContext";
2import * as Namespace from "alchemy/Namespace";
3import { ProviderModePolicy } from "alchemy/ProviderMode";
4import * as Effect from "effect/Effect";
5import * as HttpServerResponse from "effect/http/HttpServerResponse";
6import * as Layer from "effect/Layer";
7import {
8 CronEventSource,
9 type CronEventSourceProps,
10 type CronEventSourceService,
11} from "./CronEventSource.ts";
12import { Function } from "./Function.ts";
13import { FunctionRequest } from "./FunctionEnvironment.ts";
14import { FunctionTrigger } from "./FunctionTrigger.ts";
15import { decodeFunctionTriggerEvent, type CronEvent } from "./FunctionTriggerEvent.ts";
16
17/**
18 * Neon Function HTTP schedule dispatch and deployment wiring.
19 *
20 * ### Subscribe to a schedule
21 * **Example:** Provide the subscription implementation
22 * ```typescript
23 * const application = Effect.gen(function* () {
24 * yield* Neon.CronEventSource("Nightly", { cron: "0 2 * * *" }, event =>
25 * Effect.log(event.invocationId),
26 * );
27 * return { fetch: Effect.succeed(HttpServerResponse.text("ok")) };
28 * }).pipe(Effect.provide(Neon.CronEventSourceHttp));
29 * ```
30 *
31 * @layer
32 * @product Function
33 * @provides Neon.CronEventSource
34 */
35export const CronEventSourceHttp = Layer.effect(
36 CronEventSource,
37 Effect.gen(function* () {
38 const host = yield* Function;
39 const Trigger = yield* FunctionTrigger;
40 return Effect.fn(function* (
41 name: string,
42 props: CronEventSourceProps,
43 handler: (event: CronEvent) => Effect.Effect<void, unknown>,
44 ) {
45 const path = `/__alchemy/neon/cron/${encodeURIComponent(name)}`;
46 const triggerName = `${host.FQN}:${name}`;
47 if (!globalThis.__ALCHEMY_RUNTIME__) {
48 const context = yield* AlchemyContext;
49 const remote = yield* ProviderModePolicy;
50 if (!context.dev || remote)
51 yield* Namespace.push(
52 host.LogicalId,
53 Trigger(name, {
54 function: host,
55 type: "schedule",
56 name: triggerName,
57 schedule: { cron: props.cron },
58 path,
59 enabled: props.enabled,
60 }),
61 );
62 }
63 yield* host.route(
64 path,
65 Effect.gen(function* () {
66 const event = yield* decodeFunctionTriggerEvent(yield* FunctionRequest);
67 if (
68 event.trigger.type !== "schedule" ||
69 event.trigger.name !== triggerName ||
70 !("scheduled_at" in event.data)
71 )
72 return HttpServerResponse.empty({ status: 400 });
73 yield* handler({
74 invocationId: event.invocation_id,
75 triggerId: event.trigger.id,
76 name,
77 scheduledAt: event.data.scheduled_at,
78 }).pipe(Effect.orDie);
79 return HttpServerResponse.empty({ status: 204 });
80 }).pipe(
81 Effect.catchTag("FunctionTriggerEventError", (error) =>
82 Effect.succeed(HttpServerResponse.empty({ status: error.status })),
83 ),
84 ),
85 );
86 }) as CronEventSourceService;
87 }),
88);