1import * as Effect from "effect/Effect";
2import * as Layer from "effect/Layer";
3import {
4 createScheduleRoute,
5 createScheduleRouteId,
6 isScheduleEvent,
7 ScheduleEventSource as SchedulerScheduleEventSource,
8 type ScheduleDescriptor,
9 type ScheduleEvent,
10 type ScheduleEventSourceService,
11} from "../Scheduler/ScheduleEventSource.ts";
12import * as Lambda from "./Function.ts";
13
14/**
15 * Lambda runtime implementation for `AWS.Scheduler.consumeSchedule(...)` —
16 * the "cron handler" DX where a Lambda consumes its own EventBridge Scheduler
17 * invocations.
18 *
19 * This layer does two things:
20 *
21 * 1. At deploy time it creates the backing `Schedule` targeting the current
22 * Lambda function, plus the synthesized execution role that allows
23 * EventBridge Scheduler to invoke it.
24 * 2. At runtime it matches incoming Lambda events against the schedule's
25 * typed envelope (`isScheduleEvent` + the stable route id) and dispatches
26 * them to the supplied handler.
27 * ### Consuming Scheduled Invocations
28 * **Example:** Run A Handler Every 5 Minutes
29 * ```typescript
30 * yield* AWS.Scheduler.consumeSchedule(
31 * AWS.Scheduler.every("5 minutes"),
32 * (event) => Effect.log(`fired at ${event.scheduledTime}`),
33 * );
34 * ```
35 *
36 * **Example:** Nightly Cron With An Explicit Route Id
37 * ```typescript
38 * yield* AWS.Scheduler.consumeSchedule(
39 * "NightlyCleanup",
40 * AWS.Scheduler.cron("cron(0 3 * * ? *)"),
41 * (event) => Effect.log(`cleanup ${event.executionId}`),
42 * );
43 * ```
44 *
45 * @binding
46 */
47export const ScheduleEventSource = Layer.effect(
48 SchedulerScheduleEventSource,
49 Effect.gen(function* () {
50 const host = yield* Lambda.Function;
51
52 return Effect.fn(function* <Req = never>(
53 descriptor: ScheduleDescriptor,
54 process: (event: ScheduleEvent) => Effect.Effect<void, never, Req>,
55 ) {
56 // Stable route id — computed identically at deploy time (names the
57 // backing Schedule + role) and at runtime (matches incoming events).
58 const routeId = yield* Effect.sync(() =>
59 createScheduleRouteId(descriptor, host),
60 );
61
62 // Deploy-time: create the backing Schedule + execution role targeting
63 // this function. Skipped once running inside the deployed Function (the
64 // global guard), where the only work is registering the runtime handler
65 // below.
66 if (!globalThis.__ALCHEMY_RUNTIME__) {
67 yield* createScheduleRoute(routeId, descriptor, host).pipe(
68 Effect.asVoid,
69 );
70 }
71
72 yield* host.listen(
73 Effect.sync(() => (event: any) => {
74 if (isScheduleEvent(event) && event.scheduleId === routeId) {
75 return process(event).pipe(Effect.orDie);
76 }
77 }),
78 );
79 }) as ScheduleEventSourceService;
80 }),
81);