1import type * as scheduler from "@distilled.cloud/aws/scheduler";
2import * as Effect from "effect/Effect";
3import { createHash } from "node:crypto";
4import * as Binding from "../../Binding.ts";
5import * as IAM from "../IAM/index.ts";
6import type { Function as LambdaFunction } from "../Lambda/Function.ts";
7import type { ScheduleBuilder, ScheduleBuilderState } from "./builders.ts";
8import { Schedule } from "./Schedule.ts";
9import type { ScheduleGroup } from "./ScheduleGroup.ts";
10
11/**
12 * The typed event a consumed schedule delivers to its handler. EventBridge
13 * Scheduler substitutes the `<aws.scheduler.*>` context attributes into the
14 * schedule's `Input` template at invocation time; `scheduleId` is the stable
15 * route id used to match the invocation to the registered handler.
16 */
17export interface ScheduleEvent {
18 source: "alchemy.scheduler";
19 /**
20 * Stable route id of the `consumeSchedule` registration that created the
21 * schedule.
22 */
23 scheduleId: string;
24 /**
25 * ARN of the schedule (`<aws.scheduler.schedule-arn>`).
26 */
27 scheduleArn: string;
28 /**
29 * The time the invocation was scheduled for, ISO 8601
30 * (`<aws.scheduler.scheduled-time>`).
31 */
32 scheduledTime: string;
33 /**
34 * Unique id of this delivery attempt (`<aws.scheduler.execution-id>`).
35 */
36 executionId: string;
37 /**
38 * Delivery attempt counter, 1-based (`<aws.scheduler.attempt-number>`).
39 */
40 attemptNumber: string;
41}
42
43/**
44 * Narrow an arbitrary Lambda invocation payload to a schedule event produced
45 * by `consumeSchedule`.
46 */
47export const isScheduleEvent = (event: any): event is ScheduleEvent =>
48 event?.source === "alchemy.scheduler" &&
49 typeof event?.scheduleId === "string";
50
51export interface ScheduleRouteProps {
52 /**
53 * Optional schedule group the backing schedule is created in.
54 * @default the AWS default group
55 */
56 group?: ScheduleGroup;
57 /**
58 * Optional description on the backing schedule.
59 */
60 description?: string;
61 /**
62 * Desired schedule state (`ENABLED` / `DISABLED`).
63 */
64 state?: string;
65 /**
66 * Timezone for cron or at expressions.
67 */
68 timezone?: string;
69 /**
70 * Optional start date.
71 */
72 startDate?: Date;
73 /**
74 * Optional end date.
75 */
76 endDate?: Date;
77 /**
78 * Optional KMS key ARN.
79 */
80 kmsKeyArn?: string;
81 /**
82 * Flexible time window configuration.
83 * @default { Mode: "OFF" }
84 */
85 flexibleTimeWindow?: scheduler.FlexibleTimeWindow;
86 /**
87 * Action after a one-time schedule completes.
88 */
89 actionAfterCompletion?: string;
90 /**
91 * Retry policy for failed invocations.
92 */
93 retryPolicy?: scheduler.RetryPolicy;
94 /**
95 * Dead-letter queue for undeliverable invocations.
96 */
97 deadLetterConfig?: scheduler.DeadLetterConfig;
98}
99
100export interface ScheduleDescriptor {
101 /**
102 * Stable route id. Defaults to a hash of the schedule expression and the
103 * host Function's logical id.
104 */
105 id?: string;
106 /**
107 * Schedule expression: `rate(...)`, `cron(...)`, or `at(...)`.
108 */
109 expression: string;
110 props?: ScheduleRouteProps;
111}
112
113/**
114 * The cron-handler DX for EventBridge Scheduler: a Lambda consumes its own
115 * scheduled invocations. `consumeSchedule(every("5 minutes"), handler)`
116 * provisions the backing `Schedule` (plus the synthesized execution role that
117 * lets Scheduler invoke the host) and registers the runtime handler with a
118 * typed event guard.
119 *
120 * @binding
121 */
122export interface ScheduleEventSource extends Binding.Service<
123 ScheduleEventSource,
124 "AWS.Scheduler.ScheduleEventSource",
125 ScheduleEventSourceService
126> {}
127export const ScheduleEventSource = Binding.Service<ScheduleEventSource>(
128 "AWS.Scheduler.ScheduleEventSource",
129);
130
131export type ScheduleEventSourceService = <Req = never>(
132 descriptor: ScheduleDescriptor,
133 process: (event: ScheduleEvent) => Effect.Effect<void, never, Req>,
134) => Effect.Effect<void, never, never>;
135
136/**
137 * Consume a schedule's invocations on the host Lambda Function — the "cron
138 * handler" DX. Build the cadence with `every`/`cron`/`at` and pass the handler
139 * as the last argument; the runtime layer provisions the backing schedule,
140 * the execution role, and the typed event dispatch.
141 *
142 * @example Run a handler every 5 minutes
143 * ```typescript
144 * yield* AWS.Scheduler.consumeSchedule(
145 * AWS.Scheduler.every("5 minutes"),
146 * (event) => Effect.log(`fired at ${event.scheduledTime}`),
147 * );
148 * ```
149 *
150 * @example Name the backing schedule route deterministically
151 * ```typescript
152 * yield* AWS.Scheduler.consumeSchedule(
153 * "NightlyCleanup",
154 * AWS.Scheduler.cron("cron(0 3 * * ? *)"),
155 * (event) => Effect.log(`cleanup ${event.executionId}`),
156 * );
157 * ```
158 */
159export function consumeSchedule<Req = never>(
160 builder: ScheduleBuilder,
161 process: (event: ScheduleEvent) => Effect.Effect<void, never, Req>,
162): Effect.Effect<void, never, ScheduleEventSource | Req>;
163export function consumeSchedule<Req = never>(
164 id: string,
165 builder: ScheduleBuilder,
166 process: (event: ScheduleEvent) => Effect.Effect<void, never, Req>,
167): Effect.Effect<void, never, ScheduleEventSource | Req>;
168export function consumeSchedule(...args: any[]) {
169 const process = args[args.length - 1] as (
170 event: ScheduleEvent,
171 ) => Effect.Effect<void, never, never>;
172 const head = args.slice(0, -1);
173 const id = typeof head[0] === "string" ? (head[0] as string) : undefined;
174 const builder = (id === undefined ? head[0] : head[1]) as ScheduleBuilder;
175 return ScheduleEventSource.use((source) =>
176 source(toScheduleDescriptor(id, builder.state), process),
177 );
178}
179
180const toScheduleDescriptor = (
181 id: string | undefined,
182 state: ScheduleBuilderState,
183): ScheduleDescriptor => ({
184 id: id ?? state.name,
185 expression: state.expression,
186 props: {
187 group: state.group,
188 description: state.description,
189 state: state.state,
190 timezone: state.timezone,
191 startDate: state.startDate,
192 endDate: state.endDate,
193 kmsKeyArn: state.kmsKeyArn,
194 flexibleTimeWindow: state.flexibleTimeWindow,
195 actionAfterCompletion: state.actionAfterCompletion,
196 },
197});
198
199/**
200 * Derive the stable route id for a schedule descriptor: the explicit id when
201 * given, otherwise a hash of the expression and the host Function's logical
202 * id. Computed identically at deploy time (to name the backing resources) and
203 * at runtime (to match incoming events to the handler).
204 */
205export const createScheduleRouteId = (
206 descriptor: ScheduleDescriptor,
207 fn: LambdaFunction,
208): string =>
209 descriptor.id ??
210 `Scheduler${createHash("sha1")
211 .update(
212 JSON.stringify({
213 expression: descriptor.expression,
214 host: fn.LogicalId,
215 }),
216 )
217 .digest("hex")
218 .slice(0, 10)}`;
219
220/**
221 * Deploy-time half of `consumeSchedule`: synthesize the execution role that
222 * lets EventBridge Scheduler invoke the host Function and create the backing
223 * `Schedule` whose `Input` template carries the typed event envelope.
224 *
225 * @binding
226 */
227export const createScheduleRoute = (
228 routeId: string,
229 descriptor: ScheduleDescriptor,
230 fn: LambdaFunction,
231) =>
232 Effect.gen(function* () {
233 const props = descriptor.props ?? {};
234
235 const role = yield* IAM.Role(`${routeId}Role`, {
236 assumeRolePolicyDocument: {
237 Version: "2012-10-17",
238 Statement: [
239 {
240 Effect: "Allow",
241 Principal: { Service: "scheduler.amazonaws.com" },
242 Action: ["sts:AssumeRole"],
243 },
244 ],
245 },
246 inlinePolicies: {
247 InvokeHost: {
248 Version: "2012-10-17",
249 Statement: [
250 {
251 Effect: "Allow",
252 Action: ["lambda:InvokeFunction"],
253 Resource: [fn.functionArn],
254 },
255 ],
256 },
257 },
258 });
259
260 return yield* Schedule(routeId, {
261 groupName: props.group?.scheduleGroupName,
262 scheduleExpression: descriptor.expression,
263 description: props.description,
264 scheduleExpressionTimezone: props.timezone,
265 startDate: props.startDate,
266 endDate: props.endDate,
267 state: props.state,
268 kmsKeyArn: props.kmsKeyArn,
269 flexibleTimeWindow: props.flexibleTimeWindow ?? { Mode: "OFF" },
270 actionAfterCompletion: props.actionAfterCompletion,
271 target: {
272 Arn: fn.functionArn as any,
273 RoleArn: role.roleArn as any,
274 // EventBridge Scheduler substitutes the <aws.scheduler.*> context
275 // attributes into this template at invocation time; the resulting
276 // JSON is the Lambda event payload matched by `isScheduleEvent`.
277 Input: JSON.stringify({
278 source: "alchemy.scheduler",
279 scheduleId: routeId,
280 scheduleArn: "<aws.scheduler.schedule-arn>",
281 scheduledTime: "<aws.scheduler.scheduled-time>",
282 executionId: "<aws.scheduler.execution-id>",
283 attemptNumber: "<aws.scheduler.attempt-number>",
284 }),
285 RetryPolicy: props.retryPolicy,
286 DeadLetterConfig: props.deadLetterConfig,
287 },
288 });
289 });