| 1 | import * as Effect from "effect/Effect"; |
| 2 | import * as Layer from "effect/Layer"; |
| 3 | import { |
| 4 | createScheduleRoute, |
| 5 | createScheduleRouteId, |
| 6 | isScheduleEvent, |
| 7 | ScheduleEventSource as SchedulerScheduleEventSource, |
| 8 | type ScheduleDescriptor, |
| 9 | type ScheduleEvent, |
| 10 | type ScheduleEventSourceService, |
| 11 | } from "../Scheduler/ScheduleEventSource.ts"; |
| 12 | import * 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 | */ |
| 47 | export 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 | ); |