1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
4import * as scheduler from "@distilled.cloud/aws/scheduler";
5import * as SQS from "@distilled.cloud/aws/sqs";
6import { describe, expect } from "alchemy-test";
7import * as Data from "effect/Data";
8import * as Effect from "effect/Effect";
9import * as Schedule from "effect/Schedule";
10import * as HttpClient from "effect/http/HttpClient";
11import * as HttpClientRequest from "effect/http/HttpClientRequest";
12import SchedulerTestFunctionLive, { SchedulerTestFunction } from "./handler";
13
14const testOptions = { providers: AWS.providers() };
15const { test, beforeAll, afterAll } = Test.make(testOptions);
16const sharedStack = Core.scratchStack(testOptions, "SchedulerBindings");
17
18// Every schedule the fixture mints at runtime carries this deterministic
19// prefix so leftovers can be purged out-of-band before and after the suite.
20const SCHEDULE_PREFIX = "alch-schedtest-";
21
22// Lambda function URL cold-start (DNS, IAM propagation, init) can take well
23// over 60s on a fresh deploy under parallel-suite load.
24const readinessPolicy = Schedule.max([
25 Schedule.fixed("2 seconds"),
26 Schedule.recurs(75),
27]);
28
29let baseUrl: string;
30let sinkQueueUrl: string;
31let cronQueueUrl: string;
32
33class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
34 readonly status: number;
35 readonly body: string;
36}> {}
37
38class MessageNotDelivered extends Data.TaggedError("MessageNotDelivered") {}
39
40class ScheduleStillExists extends Data.TaggedError("ScheduleStillExists") {}
41
42// The shared Lambda fixture occasionally answers a transient 5xx under
43// parallel load (cold re-init, IAM propagation). Retry 5xx only; a genuine
44// 4xx/assertion failure surfaces immediately.
45const send = (request: HttpClientRequest.HttpClientRequest) =>
46 HttpClient.execute(request).pipe(
47 Effect.flatMap((response) =>
48 response.status >= 500
49 ? response.text.pipe(
50 Effect.flatMap((body) =>
51 Effect.fail(
52 new TransientUpstream({ status: response.status, body }),
53 ),
54 ),
55 )
56 : Effect.succeed(response),
57 ),
58 Effect.retry({
59 while: (e) => e._tag === "TransientUpstream",
60 schedule: Schedule.max([
61 Schedule.exponential("500 millis"),
62 Schedule.recurs(6),
63 ]),
64 }),
65 );
66
67// Out-of-band purge of runtime-minted schedules (default group, deterministic
68// prefix) so a crashed previous run can't ConflictException this one. Wrapped
69// in Core.withProviders because the distilled ops need the AWS environment
70// (Credentials/Region) that test.provider bodies get implicitly but
71// beforeAll/afterAll hooks do not.
72const purgeTestSchedules = Core.withProviders(
73 Effect.gen(function* () {
74 const listed = yield* scheduler.listSchedules({
75 NamePrefix: SCHEDULE_PREFIX,
76 });
77 const names = (listed.Schedules ?? []).flatMap((s) =>
78 s.Name ? [s.Name] : [],
79 );
80 if (names.length > 0) {
81 yield* Effect.logInfo(
82 `Scheduler test cleanup: deleting leftover schedules ${names.join(", ")}`,
83 );
84 }
85 yield* Effect.forEach(
86 names,
87 (name) =>
88 scheduler
89 .deleteSchedule({ Name: name })
90 .pipe(
91 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
92 ),
93 { concurrency: 5 },
94 );
95 }),
96 testOptions,
97 "SchedulerBindingsPurge",
98);
99
100// Bounded poll until a message whose JSON body matches the predicate shows up
101// on the queue; matched messages are deleted so re-polls stay clean.
102const receiveMatching = (
103 queueUrl: string,
104 matches: (body: any) => boolean,
105 options?: { times?: number },
106) =>
107 Effect.gen(function* () {
108 const result = yield* SQS.receiveMessage({
109 QueueUrl: queueUrl,
110 MaxNumberOfMessages: 10,
111 WaitTimeSeconds: 5,
112 VisibilityTimeout: 15,
113 });
114 const match = (result.Messages ?? []).find((message) => {
115 try {
116 return message.Body !== undefined && matches(JSON.parse(message.Body));
117 } catch {
118 return false;
119 }
120 });
121 if (!match?.ReceiptHandle) {
122 return yield* Effect.fail(new MessageNotDelivered());
123 }
124 yield* SQS.deleteMessage({
125 QueueUrl: queueUrl,
126 ReceiptHandle: match.ReceiptHandle,
127 });
128 return JSON.parse(match.Body!) as any;
129 }).pipe(
130 Effect.retry({
131 while: (error) => error._tag === "MessageNotDelivered",
132 schedule: Schedule.max([
133 Schedule.fixed("1 seconds"),
134 Schedule.recurs(options?.times ?? 12),
135 ]),
136 }),
137 );
138
139// Typed wait-until-gone: retries while GetSchedule still succeeds.
140const waitUntilScheduleGone = (name: string) =>
141 Effect.gen(function* () {
142 const found = yield* scheduler
143 .getSchedule({ Name: name })
144 .pipe(
145 Effect.catchTag("ResourceNotFoundException", () =>
146 Effect.succeed(undefined),
147 ),
148 );
149 if (found !== undefined) {
150 return yield* Effect.fail(new ScheduleStillExists());
151 }
152 }).pipe(
153 Effect.retry({
154 while: (error) => error._tag === "ScheduleStillExists",
155 schedule: Schedule.max([
156 Schedule.fixed("3 seconds"),
157 Schedule.recurs(15),
158 ]),
159 }),
160 );
161
162describe(
163 "Scheduler Bindings",
164 {
165 tags: [
166 "provider:aws",
167 "provider:aws:iam",
168 "provider:aws:lambda",
169 "provider:aws:scheduler",
170 "provider:aws:sqs",
171 "live",
172 ],
173 },
174 () => {
175 beforeAll(
176 Effect.gen(function* () {
177 yield* Effect.logInfo(
178 "Scheduler test setup: destroying previous resources",
179 );
180 yield* sharedStack.destroy();
181 yield* purgeTestSchedules;
182
183 yield* Effect.logInfo("Scheduler test setup: deploying fixture");
184 const { functionUrl } = yield* sharedStack.deploy(
185 Effect.gen(function* () {
186 return yield* SchedulerTestFunction;
187 }).pipe(Effect.provide(SchedulerTestFunctionLive)),
188 );
189
190 expect(functionUrl).toBeTruthy();
191 baseUrl = functionUrl!.replace(/\/+$/, "");
192 const readinessUrl = `${baseUrl}/info`;
193
194 yield* Effect.logInfo(
195 `Scheduler test setup: probing readiness at ${readinessUrl}`,
196 );
197 const info = yield* HttpClient.get(readinessUrl).pipe(
198 Effect.flatMap((response) =>
199 response.status === 200
200 ? (response.json as Effect.Effect<{
201 sinkQueueUrl?: string;
202 cronQueueUrl?: string;
203 }>)
204 : Effect.fail(
205 new Error(`Function not ready: ${response.status}`),
206 ),
207 ),
208 Effect.flatMap((body) =>
209 body.sinkQueueUrl && body.cronQueueUrl
210 ? Effect.succeed(
211 body as { sinkQueueUrl: string; cronQueueUrl: string },
212 )
213 : Effect.fail(new Error("Function returned empty queue urls")),
214 ),
215 Effect.tapError((error) =>
216 Effect.logWarning(
217 `Scheduler test setup: fixture not ready yet (${String(error)})`,
218 ),
219 ),
220 Effect.retry({ schedule: readinessPolicy }),
221 );
222 sinkQueueUrl = info.sinkQueueUrl;
223 cronQueueUrl = info.cronQueueUrl;
224 }),
225 { timeout: 240_000 },
226 );
227
228 afterAll(
229 Effect.gen(function* () {
230 // NO_DESTROY=1 keeps the deployment (and its log group) around while
231 // iterating locally — standard escape hatch, default is full cleanup.
232 if (process.env.NO_DESTROY) return;
233 // Leak guard: no runtime-minted schedules may outlive the suite.
234 yield* purgeTestSchedules;
235 yield* sharedStack.destroy();
236 }),
237 { timeout: 120_000 },
238 );
239
240 describe("CreateSchedule", () => {
241 test.provider(
242 "mints a one-shot at() schedule whose target fires (iam:PassRole end-to-end)",
243 (_stack) =>
244 Effect.gen(function* () {
245 const name = `${SCHEDULE_PREFIX}oneshot`;
246
247 const created = (yield* send(
248 HttpClientRequest.post(
249 `${baseUrl}/schedules/${name}?delaySeconds=15`,
250 ),
251 ).pipe(Effect.flatMap((r) => r.json))) as {
252 scheduleArn: string;
253 expression: string;
254 };
255
256 expect(created.scheduleArn).toContain(`:schedule/default/${name}`);
257 expect(created.expression).toMatch(/^at\(/);
258
259 // Out-of-band: the schedule exists with the one-shot config the
260 // fixture requested.
261 const observed = yield* scheduler.getSchedule({ Name: name });
262 expect(observed.Name).toBe(name);
263 expect(observed.ActionAfterCompletion).toBe("DELETE");
264 expect(observed.Target?.RoleArn).toBeTruthy();
265
266 // The schedule fires ~15s after creation; the execution role (which
267 // the binding contributed iam:PassRole for) delivers the marker into
268 // the sink queue. Bounded: ~15s delay + poll ≤ ~75s.
269 const message = yield* receiveMatching(
270 sinkQueueUrl,
271 (body) => body.marker === name,
272 );
273 expect(message.marker).toBe(name);
274
275 // ActionAfterCompletion=DELETE reaps the fired one-shot: typed
276 // wait-until-gone doubles as the no-leak proof.
277 yield* waitUntilScheduleGone(name);
278 }),
279 { timeout: 150_000 },
280 );
281 });
282
283 describe("GetSchedule", () => {
284 test.provider("reads a pending schedule through the binding", (_stack) =>
285 Effect.gen(function* () {
286 const name = `${SCHEDULE_PREFIX}get`;
287
288 yield* send(
289 HttpClientRequest.post(
290 `${baseUrl}/schedules/${name}?delaySeconds=900`,
291 ),
292 );
293
294 const found = (yield* send(
295 HttpClientRequest.get(`${baseUrl}/schedules/${name}`),
296 ).pipe(Effect.flatMap((r) => r.json))) as {
297 arn: string;
298 name: string;
299 state: string;
300 expression: string;
301 };
302
303 expect(found.name).toBe(name);
304 expect(found.state).toBe("ENABLED");
305 expect(found.expression).toMatch(/^at\(/);
306
307 // cleanup via the DeleteSchedule binding route
308 const deleted = yield* send(
309 HttpClientRequest.delete(`${baseUrl}/schedules/${name}`),
310 );
311 expect(deleted.status).toBe(200);
312 }),
313 );
314
315 test.provider("surfaces the typed not-found as a 404", (_stack) =>
316 Effect.gen(function* () {
317 const response = yield* send(
318 HttpClientRequest.get(
319 `${baseUrl}/schedules/${SCHEDULE_PREFIX}missing`,
320 ),
321 );
322 expect(response.status).toBe(404);
323 }),
324 );
325 });
326
327 describe("UpdateSchedule", () => {
328 test.provider(
329 "reschedules a pending schedule (full PUT + iam:PassRole)",
330 (_stack) =>
331 Effect.gen(function* () {
332 const name = `${SCHEDULE_PREFIX}upd`;
333
334 yield* send(
335 HttpClientRequest.post(
336 `${baseUrl}/schedules/${name}?delaySeconds=900`,
337 ),
338 );
339
340 // Update: push the fire time out further and set a description.
341 const updated = (yield* send(
342 HttpClientRequest.put(
343 `${baseUrl}/schedules/${name}?delaySeconds=1800&description=rescheduled`,
344 ),
345 ).pipe(Effect.flatMap((r) => r.json))) as {
346 scheduleArn: string;
347 expression: string;
348 };
349 expect(updated.scheduleArn).toContain(`:schedule/default/${name}`);
350
351 // Out-of-band: the new expression and description landed.
352 const observed = yield* scheduler.getSchedule({ Name: name });
353 expect(observed.ScheduleExpression).toBe(updated.expression);
354 expect(observed.Description).toBe("rescheduled");
355
356 // cleanup via the DeleteSchedule binding route
357 const deleted = yield* send(
358 HttpClientRequest.delete(`${baseUrl}/schedules/${name}`),
359 );
360 expect(deleted.status).toBe(200);
361 }),
362 { timeout: 120_000 },
363 );
364
365 test.provider("surfaces the typed not-found as a 404", (_stack) =>
366 Effect.gen(function* () {
367 const response = yield* send(
368 HttpClientRequest.put(
369 `${baseUrl}/schedules/${SCHEDULE_PREFIX}missing-upd`,
370 ),
371 );
372 expect(response.status).toBe(404);
373 }),
374 );
375 });
376
377 describe("ListSchedules", () => {
378 test.provider(
379 "lists runtime-minted schedules by prefix (default-group scoped)",
380 (_stack) =>
381 Effect.gen(function* () {
382 const nameA = `${SCHEDULE_PREFIX}list-a`;
383 const nameB = `${SCHEDULE_PREFIX}list-b`;
384
385 yield* send(
386 HttpClientRequest.post(
387 `${baseUrl}/schedules/${nameA}?delaySeconds=900`,
388 ),
389 );
390 yield* send(
391 HttpClientRequest.post(
392 `${baseUrl}/schedules/${nameB}?delaySeconds=900`,
393 ),
394 );
395
396 const listed = (yield* send(
397 HttpClientRequest.get(
398 `${baseUrl}/schedules?namePrefix=${SCHEDULE_PREFIX}list-`,
399 ),
400 ).pipe(Effect.flatMap((r) => r.json))) as {
401 names: string[];
402 error?: string;
403 message?: string;
404 };
405
406 // A typed failure inside the Lambda surfaces here with its tag.
407 expect(listed.error, listed.message).toBeUndefined();
408 expect(listed.names).toContain(nameA);
409 expect(listed.names).toContain(nameB);
410
411 // cleanup via the DeleteSchedule binding route
412 for (const name of [nameA, nameB]) {
413 const deleted = yield* send(
414 HttpClientRequest.delete(`${baseUrl}/schedules/${name}`),
415 );
416 expect(deleted.status).toBe(200);
417 }
418 }),
419 { timeout: 120_000 },
420 );
421 });
422
423 describe("DeleteSchedule", () => {
424 test.provider(
425 "deletes a pending schedule and reports not-found on repeat",
426 (_stack) =>
427 Effect.gen(function* () {
428 const name = `${SCHEDULE_PREFIX}del`;
429
430 yield* send(
431 HttpClientRequest.post(
432 `${baseUrl}/schedules/${name}?delaySeconds=900`,
433 ),
434 );
435
436 // out-of-band: it exists before deletion
437 const observed = yield* scheduler.getSchedule({ Name: name });
438 expect(observed.Name).toBe(name);
439
440 const first = yield* send(
441 HttpClientRequest.delete(`${baseUrl}/schedules/${name}`),
442 );
443 expect(first.status).toBe(200);
444 expect(((yield* first.json) as any).deleted).toBe(true);
445
446 // out-of-band typed wait-until-gone
447 yield* waitUntilScheduleGone(name);
448
449 // idempotent repeat surfaces the typed ResourceNotFoundException
450 const second = yield* send(
451 HttpClientRequest.delete(`${baseUrl}/schedules/${name}`),
452 );
453 expect(second.status).toBe(404);
454 }),
455 { timeout: 120_000 },
456 );
457 });
458
459 describe("ScheduleEventSource", () => {
460 test.provider(
461 "the cron handler consumes its own scheduled invocations",
462 (_stack) =>
463 Effect.gen(function* () {
464 // The fixture registered consumeSchedule(every("1 minute"), ...) at
465 // deploy time; the first fire lands within a minute of schedule
466 // creation (usually before the tests start), forwarded into the
467 // cron queue as the typed ScheduleEvent envelope. Bounded: long-poll
468 // ~90s to ride out one full rate window.
469 const event = yield* receiveMatching(
470 cronQueueUrl,
471 (body) => body.marker === "alch-schedtest-cron-fired",
472 { times: 15 },
473 );
474
475 expect(event.scheduleArn).toContain(":schedule/default/");
476 expect(event.scheduleId).toMatch(/^Scheduler[0-9a-f]{10}$/);
477 expect(event.scheduledTime).toBeTruthy();
478 expect(event.executionId).toBeTruthy();
479 }),
480 { timeout: 120_000 },
481 );
482 });
483 },
484);