1import * as IAM from "@/AWS/IAM";
2import * as Lambda from "@/AWS/Lambda";
3import * as Scheduler from "@/AWS/Scheduler";
4import * as SQS from "@/AWS/SQS";
5import * as Duration from "effect/Duration";
6import * as Effect from "effect/Effect";
7import * as Layer from "effect/Layer";
8import { HttpServerRequest } from "effect/http/HttpServerRequest";
9import * as HttpServerResponse from "effect/http/HttpServerResponse";
10import path from "pathe";
11
12const main = path.resolve(import.meta.dirname, "handler.ts");
13
14// Shared Lambda fixture for the EventBridge Scheduler runtime surface:
15// - CreateSchedule / GetSchedule / DeleteSchedule bindings, one HTTP route
16// per operation. Runtime-minted one-shot schedules target the sink queue,
17// so the test can observe the fire out-of-band.
18// - consumeSchedule (the cron-handler DX): a rate(1 minute) schedule invokes
19// this same Lambda; the handler forwards the typed event into the cron
20// queue for out-of-band observation.
21export class SchedulerTestFunction extends Lambda.Function<Lambda.Function>()(
22 "SchedulerTestFunction",
23) {}
24
25export default SchedulerTestFunction.make(
26 {
27 main,
28 functionUrl: true,
29 // AWS's 3s default intermittently kills invocations that fan out SDK
30 // calls (create + describe schedules) — always set an explicit timeout.
31 timeout: Duration.seconds(30),
32 },
33 Effect.gen(function* () {
34 // Queue that runtime-minted one-shot schedules deliver into.
35 const sinkQueue = yield* SQS.Queue("SchedulerSinkQueue");
36 // Queue the consumeSchedule cron handler forwards its typed events into.
37 const cronQueue = yield* SQS.Queue("SchedulerCronQueue");
38
39 // Execution role for runtime-minted schedules: assumable by EventBridge
40 // Scheduler, allowed to send into the sink queue. The CreateSchedule
41 // binding contributes iam:PassRole on this role to the host's policy.
42 const executionRole = yield* IAM.Role("SchedulerExecutionRole", {
43 assumeRolePolicyDocument: {
44 Version: "2012-10-17",
45 Statement: [
46 {
47 Effect: "Allow",
48 Principal: { Service: "scheduler.amazonaws.com" },
49 Action: ["sts:AssumeRole"],
50 },
51 ],
52 },
53 inlinePolicies: {
54 ScheduleTarget: {
55 Version: "2012-10-17",
56 Statement: [
57 {
58 Effect: "Allow",
59 Action: ["sqs:SendMessage"],
60 Resource: [sinkQueue.queueArn],
61 },
62 ],
63 },
64 },
65 });
66
67 const createSchedule = yield* Scheduler.CreateSchedule(executionRole);
68 const updateSchedule = yield* Scheduler.UpdateSchedule(executionRole);
69 const getSchedule = yield* Scheduler.GetSchedule();
70 const deleteSchedule = yield* Scheduler.DeleteSchedule();
71 const listSchedules = yield* Scheduler.ListSchedules();
72 const sendCron = yield* SQS.SendMessage(cronQueue);
73
74 // The cron-handler DX: consume this Lambda's own scheduled invocations
75 // and forward the typed event into the cron queue.
76 yield* Scheduler.consumeSchedule(Scheduler.every("1 minute"), (event) =>
77 sendCron({
78 MessageBody: JSON.stringify({
79 marker: "alch-schedtest-cron-fired",
80 scheduleId: event.scheduleId,
81 scheduleArn: event.scheduleArn,
82 scheduledTime: event.scheduledTime,
83 executionId: event.executionId,
84 }),
85 }).pipe(Effect.orDie, Effect.asVoid),
86 );
87
88 const sinkQueueArn = yield* sinkQueue.queueArn;
89 const sinkQueueUrl = yield* sinkQueue.queueUrl;
90 const cronQueueUrl = yield* cronQueue.queueUrl;
91
92 return {
93 fetch: Effect.gen(function* () {
94 const request = yield* HttpServerRequest;
95 const url = new URL(request.originalUrl);
96 const pathname = url.pathname;
97
98 if (request.method === "GET" && pathname === "/info") {
99 return yield* HttpServerResponse.json({
100 ok: true,
101 sinkQueueUrl: yield* sinkQueueUrl,
102 cronQueueUrl: yield* cronQueueUrl,
103 });
104 }
105
106 if (request.method === "GET" && pathname === "/schedules") {
107 const namePrefix = url.searchParams.get("namePrefix") ?? undefined;
108 // Surface the typed failure in the body (424 so the test's 5xx
109 // retry doesn't mask a genuine error) instead of dying to a 500.
110 return yield* listSchedules({ NamePrefix: namePrefix }).pipe(
111 Effect.flatMap((listed) =>
112 HttpServerResponse.json({
113 names: (listed.Schedules ?? []).flatMap((s) =>
114 s.Name ? [s.Name] : [],
115 ),
116 }),
117 ),
118 Effect.catch((e) =>
119 HttpServerResponse.json(
120 { error: e._tag, message: String(e) },
121 { status: 424 },
122 ),
123 ),
124 );
125 }
126
127 if (pathname.startsWith("/schedules/")) {
128 const name = pathname.slice("/schedules/".length);
129
130 if (request.method === "POST") {
131 const delaySeconds = Number(
132 url.searchParams.get("delaySeconds") ?? "30",
133 );
134 const fireAt = yield* Effect.sync(
135 () => new Date(Date.now() + delaySeconds * 1000),
136 );
137 // One-shot at() expressions take second granularity, no timezone
138 // suffix (UTC by default): at(yyyy-mm-ddThh:mm:ss)
139 const expression = `at(${fireAt.toISOString().slice(0, 19)})`;
140
141 const created = yield* createSchedule({
142 Name: name,
143 ScheduleExpression: expression,
144 ActionAfterCompletion: "DELETE",
145 Target: {
146 Arn: yield* sinkQueueArn,
147 Input: JSON.stringify({ marker: name }),
148 },
149 }).pipe(
150 Effect.catchTag("ConflictException", () =>
151 Effect.succeed(undefined),
152 ),
153 );
154
155 if (created === undefined) {
156 return yield* HttpServerResponse.json(
157 { error: "Schedule already exists", name },
158 { status: 409 },
159 );
160 }
161 return yield* HttpServerResponse.json({
162 scheduleArn: created.ScheduleArn,
163 expression,
164 });
165 }
166
167 if (request.method === "PUT") {
168 const delaySeconds = Number(
169 url.searchParams.get("delaySeconds") ?? "900",
170 );
171 const description =
172 url.searchParams.get("description") ?? undefined;
173 const fireAt = yield* Effect.sync(
174 () => new Date(Date.now() + delaySeconds * 1000),
175 );
176 const expression = `at(${fireAt.toISOString().slice(0, 19)})`;
177
178 // UpdateSchedule is a full PUT — resend the complete config.
179 const updated = yield* updateSchedule({
180 Name: name,
181 ScheduleExpression: expression,
182 Description: description,
183 ActionAfterCompletion: "DELETE",
184 Target: {
185 Arn: yield* sinkQueueArn,
186 Input: JSON.stringify({ marker: name }),
187 },
188 }).pipe(
189 Effect.catchTag("ResourceNotFoundException", () =>
190 Effect.succeed(undefined),
191 ),
192 );
193
194 if (updated === undefined) {
195 return yield* HttpServerResponse.json(
196 { error: "Not found", name },
197 { status: 404 },
198 );
199 }
200 return yield* HttpServerResponse.json({
201 scheduleArn: updated.ScheduleArn,
202 expression,
203 });
204 }
205
206 if (request.method === "GET") {
207 const found = yield* getSchedule({ Name: name }).pipe(
208 Effect.catchTag("ResourceNotFoundException", () =>
209 Effect.succeed(undefined),
210 ),
211 );
212 if (found === undefined) {
213 return yield* HttpServerResponse.json(
214 { error: "Not found", name },
215 { status: 404 },
216 );
217 }
218 return yield* HttpServerResponse.json({
219 arn: found.Arn,
220 name: found.Name,
221 state: found.State,
222 expression: found.ScheduleExpression,
223 actionAfterCompletion: found.ActionAfterCompletion,
224 });
225 }
226
227 if (request.method === "DELETE") {
228 const deleted = yield* deleteSchedule({ Name: name }).pipe(
229 Effect.as(true),
230 Effect.catchTag("ResourceNotFoundException", () =>
231 Effect.succeed(false),
232 ),
233 );
234 return yield* HttpServerResponse.json(
235 { deleted },
236 { status: deleted ? 200 : 404 },
237 );
238 }
239 }
240
241 return yield* HttpServerResponse.json(
242 { error: "Not found", method: request.method, pathname },
243 { status: 404 },
244 );
245 }).pipe(Effect.orDie),
246 };
247 }).pipe(
248 Effect.provide(
249 Layer.mergeAll(
250 Scheduler.CreateScheduleHttp,
251 Scheduler.UpdateScheduleHttp,
252 Scheduler.GetScheduleHttp,
253 Scheduler.DeleteScheduleHttp,
254 Scheduler.ListSchedulesHttp,
255 Lambda.ScheduleEventSource,
256 SQS.SendMessageHttp,
257 ),
258 ),
259 ),
260);