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