1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
4import * as SQS from "@distilled.cloud/aws/sqs";
5import { describe, expect } from "alchemy-test";
6import * as Data from "effect/Data";
7import * as Effect from "effect/Effect";
8import * as Schedule from "effect/Schedule";
9import * as HttpClient from "effect/http/HttpClient";
10import * as HttpClientRequest from "effect/http/HttpClientRequest";
11import SQSTestFunctionLive, { SQSTestFunction } from "./handler";
12
13const testOptions = { providers: AWS.providers() };
14const { test, beforeAll, afterAll } = Test.make(testOptions);
15const sharedStack = Core.scratchStack(testOptions, "SQSBindings");
16
17// Lambda function URL cold-start (DNS, IAM propagation, init) can take well
18// over 60s on a fresh deploy under parallel-suite load. Budget ~150s of
19// readiness polling so we don't fail the whole suite on a slow init.
20const readinessPolicy = Schedule.max([
21 Schedule.fixed("4 seconds"),
22 Schedule.recurs(10),
23]);
24
25let baseUrl: string;
26let queueUrl: string;
27let dlqUrl: string;
28let moveSourceUrl: string;
29
30class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
31 readonly status: number;
32 readonly body: string;
33 readonly message: string;
34}> {}
35
36// The shared Lambda fixture occasionally answers a transient 5xx under
37// full-suite parallel load — a cold re-init or an eventually-consistent SQS
38// call that the handler's `Effect.orDie` surfaces as a 500. Retry only 5xx;
39// a genuine 4xx/assertion failure is returned immediately.
40const send = (request: HttpClientRequest.HttpClientRequest) =>
41 HttpClient.execute(request).pipe(
42 Effect.timeout("5 seconds"),
43 Effect.mapError(
44 (error) =>
45 new TransientUpstream({
46 status: 0,
47 body: String(error),
48 message: `Function URL request failed: ${String(error)}`,
49 }),
50 ),
51 Effect.flatMap((response) =>
52 response.status >= 500
53 ? response.text.pipe(
54 Effect.flatMap((body) =>
55 Effect.fail(
56 new TransientUpstream({
57 status: response.status,
58 body,
59 message: `Function returned ${response.status}: ${body}`,
60 }),
61 ),
62 ),
63 )
64 : Effect.succeed(response),
65 ),
66 Effect.retry({
67 while: (e) => e._tag === "TransientUpstream",
68 schedule: Schedule.max([
69 Schedule.exponential("500 millis"),
70 Schedule.recurs(6),
71 ]),
72 }),
73 );
74
75const post = (route: string, body: unknown) =>
76 send(
77 HttpClientRequest.bodyJsonUnsafe(
78 HttpClientRequest.post(`${baseUrl}${route}`),
79 body,
80 ),
81 ).pipe(Effect.flatMap((r) => r.json));
82
83interface ReceivedMessage {
84 messageId: string;
85 body: string;
86 receiptHandle: string;
87}
88
89class MessageNotReceived extends Data.TaggedError("MessageNotReceived") {}
90
91class MessageNotVisible extends Data.TaggedError("MessageNotVisible") {}
92
93class QueueStillExists extends Data.TaggedError("QueueStillExists") {}
94
95/** Poll (bounded) until GetQueueAttributes reports the queue gone. */
96const assertQueueDeleted = (url: string) =>
97 SQS.getQueueAttributes({
98 QueueUrl: url,
99 AttributeNames: ["All"],
100 }).pipe(
101 Effect.flatMap(() => Effect.fail(new QueueStillExists())),
102 Effect.retry({
103 // SQS DeleteQueue propagation is documented at ~60s; poll on a fixed
104 // cadence with a bounded budget.
105 while: (e) => e._tag === "QueueStillExists",
106 schedule: Schedule.max([
107 Schedule.spaced("3 seconds"),
108 Schedule.recurs(45),
109 ]),
110 }),
111 Effect.catchTag("QueueDoesNotExist", () => Effect.void),
112 );
113
114// Poll the fixture's /receive route (the ReceiveMessage binding) until every
115// body in `bodies` has been observed; returns a map body -> received message.
116// Bounded: ~20 polls, each an SQS long-poll of 2s.
117const receiveViaBindingUntil = (
118 bodies: string[],
119 options?: { visibilityTimeout?: number },
120) =>
121 Effect.gen(function* () {
122 const found = new Map<string, ReceivedMessage>();
123 yield* Effect.gen(function* () {
124 const response = (yield* post("/receive", {
125 maxNumberOfMessages: 10,
126 waitTimeSeconds: 2,
127 ...(options?.visibilityTimeout !== undefined
128 ? { visibilityTimeout: options.visibilityTimeout }
129 : {}),
130 })) as unknown as { messages: ReceivedMessage[] };
131 for (const message of response.messages) {
132 if (bodies.includes(message.body)) {
133 found.set(message.body, message);
134 }
135 }
136 if (found.size < bodies.length) {
137 return yield* Effect.fail(new MessageNotReceived());
138 }
139 }).pipe(
140 Effect.retry({
141 while: (e) => e._tag === "MessageNotReceived",
142 schedule: Schedule.max([
143 Schedule.fixed("1 second"),
144 Schedule.recurs(20),
145 ]),
146 }),
147 );
148 return found;
149 });
150
151// Out-of-band verification: poll the queue via distilled until every body in
152// `bodies` has been received, deleting each matching message as it arrives.
153const receiveAndDeleteViaDistilled = (bodies: string[], sourceUrl = queueUrl) =>
154 Effect.gen(function* () {
155 const found = new Set<string>();
156 yield* Effect.gen(function* () {
157 const result = yield* SQS.receiveMessage({
158 QueueUrl: sourceUrl,
159 MaxNumberOfMessages: 10,
160 WaitTimeSeconds: 2,
161 });
162 for (const message of result.Messages ?? []) {
163 if (
164 message.Body &&
165 message.ReceiptHandle &&
166 bodies.includes(message.Body)
167 ) {
168 found.add(message.Body);
169 yield* SQS.deleteMessage({
170 QueueUrl: sourceUrl,
171 ReceiptHandle: message.ReceiptHandle,
172 });
173 }
174 }
175 if (found.size < bodies.length) {
176 return yield* Effect.fail(new MessageNotReceived());
177 }
178 }).pipe(
179 Effect.retry({
180 while: (e) => e._tag === "MessageNotReceived",
181 schedule: Schedule.max([
182 Schedule.fixed("1 second"),
183 Schedule.recurs(20),
184 ]),
185 }),
186 );
187 });
188
189// Message-move tasks only operate on messages that actually entered a DLQ
190// through its source queue's redrive policy. Drive the dedicated source's
191// receive count, then use the clean DLQ's non-consuming approximate count as
192// readiness. Receiving from the DLQ, even with VisibilityTimeout=0, races the
193// asynchronous redrive scanner. The retry and final settle are bounded.
194const seedDlqViaRedrive = (body: string) =>
195 Effect.gen(function* () {
196 yield* SQS.sendMessage({
197 QueueUrl: moveSourceUrl,
198 MessageBody: body,
199 });
200
201 // Actively drive the receive count until SQS reports a visible DLQ
202 // message. VisibilityTimeout=0 increments the source receive count
203 // without hiding it between attempts.
204 yield* Effect.gen(function* () {
205 yield* SQS.receiveMessage({
206 QueueUrl: moveSourceUrl,
207 MaxNumberOfMessages: 10,
208 WaitTimeSeconds: 1,
209 VisibilityTimeout: 0,
210 });
211
212 const dlqAttributes = yield* SQS.getQueueAttributes({
213 QueueUrl: dlqUrl,
214 AttributeNames: ["ApproximateNumberOfMessages"],
215 });
216 if (
217 Number(dlqAttributes.Attributes?.ApproximateNumberOfMessages ?? "0") < 1
218 ) {
219 return yield* Effect.fail(new MessageNotVisible());
220 }
221 }).pipe(
222 Effect.retry({
223 while: (error) => error._tag === "MessageNotVisible",
224 schedule: Schedule.fixed("1 second"),
225 times: 15,
226 }),
227 );
228
229 // Let the redrive metadata observed through the approximate count settle
230 // before starting the asynchronous message-move scanner.
231 yield* Effect.sleep("2 seconds");
232 });
233
234// All five tests share ONE queue; a concurrent receive in test A steals (and
235// hides, for the visibility timeout) messages test B is polling for. The
236// global vitest `sequence: { concurrent: true }` must not apply here.
237describe.sequential(
238 "SQS Bindings",
239 { tags: ["provider:aws", "provider:aws:lambda", "provider:aws:sqs", "live"] },
240 () => {
241 beforeAll(
242 Effect.gen(function* () {
243 yield* Effect.logInfo("SQS test setup: destroying previous resources");
244 yield* sharedStack.destroy();
245
246 yield* Effect.logInfo("SQS test setup: deploying fixture");
247 const { functionUrl } = yield* sharedStack.deploy(
248 Effect.gen(function* () {
249 return yield* SQSTestFunction;
250 }).pipe(Effect.provide(SQSTestFunctionLive)),
251 );
252
253 expect(functionUrl).toBeTruthy();
254 baseUrl = functionUrl!.replace(/\/+$/, "");
255
256 yield* Effect.logInfo(
257 `SQS test setup: probing readiness at ${baseUrl}/ready`,
258 );
259
260 // A freshly-deployed function can briefly serve a 200 before its
261 // captured env vars (the queue URLs) finish propagating; keep polling
262 // until both queue URLs are populated.
263 const ready = yield* HttpClient.get(`${baseUrl}/ready`).pipe(
264 Effect.timeout("4 seconds"),
265 Effect.mapError(
266 () => new Error("Function URL readiness request timed out"),
267 ),
268 Effect.flatMap((response) =>
269 response.status === 200
270 ? (response.json as Effect.Effect<{
271 queueUrl?: string;
272 dlqUrl?: string;
273 moveSourceUrl?: string;
274 }>)
275 : Effect.fail(
276 new Error(`Function not ready: ${response.status}`),
277 ),
278 ),
279 Effect.flatMap((body) =>
280 typeof body?.queueUrl === "string" &&
281 body.queueUrl.length > 0 &&
282 typeof body?.dlqUrl === "string" &&
283 body.dlqUrl.length > 0 &&
284 typeof body?.moveSourceUrl === "string" &&
285 body.moveSourceUrl.length > 0
286 ? Effect.succeed(
287 body as {
288 queueUrl: string;
289 dlqUrl: string;
290 moveSourceUrl: string;
291 },
292 )
293 : Effect.fail(new Error("Function returned empty queue URLs")),
294 ),
295 Effect.tapError((error) =>
296 Effect.logWarning(
297 `SQS test setup: fixture not ready yet (${String(error)})`,
298 ),
299 ),
300 Effect.retry({ schedule: readinessPolicy }),
301 );
302 queueUrl = ready.queueUrl;
303 dlqUrl = ready.dlqUrl;
304 moveSourceUrl = ready.moveSourceUrl;
305
306 yield* Effect.logInfo(
307 `SQS test setup: fixture ready (queueUrl: ${queueUrl})`,
308 );
309 }),
310 { timeout: 240_000 },
311 );
312
313 afterAll(
314 Effect.gen(function* () {
315 yield* sharedStack.destroy();
316 // Out-of-band: the shared queue and DLQ are actually gone. The hook
317 // context has no provider environment (unlike `test.provider` bodies),
318 // so provide it explicitly for the distilled SQS calls.
319 yield* Core.withProviders(
320 Effect.gen(function* () {
321 if (queueUrl) yield* assertQueueDeleted(queueUrl);
322 if (dlqUrl) yield* assertQueueDeleted(dlqUrl);
323 if (moveSourceUrl) yield* assertQueueDeleted(moveSourceUrl);
324 }),
325 testOptions,
326 sharedStack.name,
327 );
328 }),
329 { timeout: 240_000 },
330 );
331
332 describe("SendMessage", () => {
333 test.provider("sends a message through the bound queue", (_stack) =>
334 Effect.gen(function* () {
335 const messageBody = `send-${crypto.randomUUID()}`;
336
337 const response = (yield* post("/send", { messageBody })) as {
338 messageId?: string;
339 };
340 expect(response.messageId).toBeTruthy();
341
342 // Out-of-band: the message is actually in the queue.
343 yield* receiveAndDeleteViaDistilled([messageBody]);
344 }),
345 );
346 });
347
348 describe("SendMessageBatch", () => {
349 test.provider(
350 "sends a batch of messages through the bound queue",
351 (_stack) =>
352 Effect.gen(function* () {
353 const bodies = [
354 `batch-${crypto.randomUUID()}`,
355 `batch-${crypto.randomUUID()}`,
356 `batch-${crypto.randomUUID()}`,
357 ];
358
359 const response = (yield* post("/send-batch", {
360 entries: bodies.map((messageBody, index) => ({
361 id: `entry-${index}`,
362 messageBody,
363 })),
364 })) as {
365 successful: { id: string; messageId: string }[];
366 failed: { id: string; code: string }[];
367 };
368
369 expect(response.failed).toHaveLength(0);
370 expect(response.successful).toHaveLength(3);
371 for (const entry of response.successful) {
372 expect(entry.messageId).toBeTruthy();
373 }
374
375 // Out-of-band: all three messages landed in the queue.
376 yield* receiveAndDeleteViaDistilled(bodies);
377 }),
378 );
379 });
380
381 describe("ReceiveMessage", () => {
382 test.provider("receives a message through the bound queue", (_stack) =>
383 Effect.gen(function* () {
384 const messageBody = `receive-${crypto.randomUUID()}`;
385
386 // Out-of-band send via distilled; receive via the binding.
387 yield* SQS.sendMessage({
388 QueueUrl: queueUrl,
389 MessageBody: messageBody,
390 });
391
392 const found = yield* receiveViaBindingUntil([messageBody]);
393 const message = found.get(messageBody)!;
394 expect(message.messageId).toBeTruthy();
395 expect(message.receiptHandle).toBeTruthy();
396
397 // Cleanup: delete with the receipt handle the binding returned.
398 yield* SQS.deleteMessage({
399 QueueUrl: queueUrl,
400 ReceiptHandle: message.receiptHandle,
401 });
402 }),
403 );
404 });
405
406 describe("DeleteMessage", () => {
407 test.provider(
408 "deletes a received message through the bound queue",
409 (_stack) =>
410 Effect.gen(function* () {
411 const messageBody = `delete-${crypto.randomUUID()}`;
412
413 yield* SQS.sendMessage({
414 QueueUrl: queueUrl,
415 MessageBody: messageBody,
416 });
417
418 // Receive with a short (5s) visibility timeout: long enough that the
419 // receipt handle is still the freshest when /delete runs, short
420 // enough that an undeleted message reappears within the absence poll.
421 const found = yield* receiveViaBindingUntil([messageBody], {
422 visibilityTimeout: 5,
423 });
424 const message = found.get(messageBody)!;
425
426 const response = (yield* post("/delete", {
427 receiptHandle: message.receiptHandle,
428 })) as { success: boolean };
429 expect(response.success).toBe(true);
430
431 // The deleted message must not reappear after its 1s visibility
432 // timeout lapses. Bounded absence poll (~10s of 1s long-polls).
433 yield* Effect.gen(function* () {
434 const result = yield* SQS.receiveMessage({
435 QueueUrl: queueUrl,
436 MaxNumberOfMessages: 10,
437 WaitTimeSeconds: 1,
438 VisibilityTimeout: 1,
439 });
440 const bodies = (result.Messages ?? []).map((m) => m.Body);
441 expect(bodies).not.toContain(messageBody);
442 }).pipe(
443 Effect.repeat({
444 schedule: Schedule.max([
445 Schedule.fixed("1 second"),
446 Schedule.recurs(5),
447 ]),
448 }),
449 );
450 }),
451 );
452 });
453
454 describe("DeleteMessageBatch", () => {
455 test.provider(
456 "deletes a batch of received messages through the bound queue",
457 (_stack) =>
458 Effect.gen(function* () {
459 const bodies = [
460 `delete-batch-${crypto.randomUUID()}`,
461 `delete-batch-${crypto.randomUUID()}`,
462 ];
463
464 yield* SQS.sendMessageBatch({
465 QueueUrl: queueUrl,
466 Entries: bodies.map((body, index) => ({
467 Id: `entry-${index}`,
468 MessageBody: body,
469 })),
470 });
471
472 // 15s visibility: long enough that both receipt handles stay valid
473 // while we collect the pair across polls, short enough that a message
474 // consumed by a failed invocation resurfaces within the poll budget.
475 const found = yield* receiveViaBindingUntil(bodies, {
476 visibilityTimeout: 15,
477 });
478
479 const response = (yield* post("/delete-batch", {
480 entries: bodies.map((body, index) => ({
481 id: `entry-${index}`,
482 receiptHandle: found.get(body)!.receiptHandle,
483 })),
484 })) as {
485 successful: string[];
486 failed: { id: string; code: string }[];
487 };
488
489 expect(response.failed).toHaveLength(0);
490 expect(response.successful.sort()).toEqual(["entry-0", "entry-1"]);
491 }),
492 );
493 });
494
495 describe("ChangeMessageVisibility", () => {
496 test.provider(
497 "releases an in-flight message back to the queue",
498 (_stack) =>
499 Effect.gen(function* () {
500 const messageBody = `visibility-${crypto.randomUUID()}`;
501
502 yield* SQS.sendMessage({
503 QueueUrl: queueUrl,
504 MessageBody: messageBody,
505 });
506
507 // Receive with a long visibility timeout so the message stays
508 // hidden unless the binding releases it.
509 const found = yield* receiveViaBindingUntil([messageBody], {
510 visibilityTimeout: 120,
511 });
512 const message = found.get(messageBody)!;
513
514 const response = (yield* post("/change-visibility", {
515 receiptHandle: message.receiptHandle,
516 visibilityTimeout: 0,
517 })) as { success: boolean };
518 expect(response.success).toBe(true);
519
520 // The message must reappear well before the 120s timeout —
521 // observable only if the binding actually reset visibility.
522 yield* receiveAndDeleteViaDistilled([messageBody]);
523 }),
524 );
525 });
526
527 describe("ChangeMessageVisibilityBatch", () => {
528 test.provider(
529 "releases a batch of in-flight messages back to the queue",
530 (_stack) =>
531 Effect.gen(function* () {
532 const bodies = [
533 `visibility-batch-${crypto.randomUUID()}`,
534 `visibility-batch-${crypto.randomUUID()}`,
535 ];
536
537 yield* SQS.sendMessageBatch({
538 QueueUrl: queueUrl,
539 Entries: bodies.map((body, index) => ({
540 Id: `entry-${index}`,
541 MessageBody: body,
542 })),
543 });
544
545 const found = yield* receiveViaBindingUntil(bodies, {
546 visibilityTimeout: 120,
547 });
548
549 const response = (yield* post("/change-visibility-batch", {
550 entries: bodies.map((body, index) => ({
551 id: `entry-${index}`,
552 receiptHandle: found.get(body)!.receiptHandle,
553 visibilityTimeout: 0,
554 })),
555 })) as {
556 successful: string[];
557 failed: { id: string; code: string }[];
558 };
559 expect(response.failed).toHaveLength(0);
560 expect(response.successful.sort()).toEqual(["entry-0", "entry-1"]);
561
562 // Both messages reappear well before the 120s timeout.
563 yield* receiveAndDeleteViaDistilled(bodies);
564 }),
565 );
566 });
567
568 describe("GetQueueAttributes", () => {
569 test.provider("reads live queue attributes", (_stack) =>
570 Effect.gen(function* () {
571 const response = (yield* post("/attributes", {
572 attributeNames: ["QueueArn", "ApproximateNumberOfMessages"],
573 })) as {
574 attributes: {
575 QueueArn?: string;
576 ApproximateNumberOfMessages?: string;
577 };
578 };
579
580 const queueName = queueUrl.split("/").pop()!;
581 expect(response.attributes.QueueArn).toMatch(/^arn:aws:sqs:/);
582 expect(response.attributes.QueueArn!.endsWith(queueName)).toBe(true);
583 expect(
584 Number(response.attributes.ApproximateNumberOfMessages),
585 ).toBeGreaterThanOrEqual(0);
586 }),
587 );
588 });
589
590 describe("MessageMoveTasks", () => {
591 test.provider(
592 "lists DLQ sources and redrives a message via a move task",
593 (_stack) =>
594 Effect.gen(function* () {
595 // ListDeadLetterSourceQueues on the DLQ sees the main queue, whose
596 // redrivePolicy targets it.
597 const sources = (yield* post("/dlq-sources", {})) as {
598 queueUrls: string[];
599 };
600 expect(sources.queueUrls).toContain(queueUrl);
601 expect(sources.queueUrls).toContain(moveSourceUrl);
602
603 // Redrive through the dedicated source so SQS records the metadata
604 // required for the message-move task.
605 const messageBody = `redrive-${crypto.randomUUID()}`;
606 yield* seedDlqViaRedrive(messageBody);
607
608 const started = (yield* post("/move/start", {})) as {
609 taskHandle?: string;
610 };
611 expect(started.taskHandle).toBeTruthy();
612
613 // Out-of-band data-plane proof: the message arrives back in its
614 // source queue. This is authoritative even while List's status and
615 // approximate counters lag.
616 yield* receiveAndDeleteViaDistilled([messageBody], moveSourceUrl);
617
618 // The exact task is visible via the ListMessageMoveTasks binding.
619 const listed = (yield* post("/move/list", {})) as {
620 tasks: { taskHandle?: string; status?: string }[];
621 };
622 expect(Array.isArray(listed.tasks)).toBe(true);
623 const task = listed.tasks.find(
624 (candidate) => candidate.taskHandle === started.taskHandle,
625 );
626 // SQS may omit a one-message task as soon as it completes. When it
627 // is retained, assert the decoded status; the destination receive
628 // above is the authoritative completion proof.
629 if (task !== undefined) {
630 expect(["RUNNING", "COMPLETING", "COMPLETED"]).toContain(
631 task.status,
632 );
633 }
634
635 // CancelMessageMoveTask rejects an already-completed task. The
636 // handler converts that exact typed terminal race to canceled:false.
637 const canceled = (yield* post("/move/cancel", {
638 taskHandle: started.taskHandle,
639 })) as { canceled: boolean };
640 expect(typeof canceled.canceled).toBe("boolean");
641 }),
642 );
643 });
644
645 // PurgeQueue runs LAST: it wipes the shared queue, and SQS allows only one
646 // purge per queue per 60 seconds.
647 describe("PurgeQueue", () => {
648 test.provider("purges every message in the queue", (_stack) =>
649 Effect.gen(function* () {
650 const bodies = [
651 `purge-${crypto.randomUUID()}`,
652 `purge-${crypto.randomUUID()}`,
653 ];
654 yield* SQS.sendMessageBatch({
655 QueueUrl: queueUrl,
656 Entries: bodies.map((body, index) => ({
657 Id: `entry-${index}`,
658 MessageBody: body,
659 })),
660 });
661
662 const response = (yield* post("/purge", {})) as { success: boolean };
663 expect(response.success).toBe(true);
664
665 // The purge deletes messages within up to 60s; poll until neither
666 // body is receivable anymore (bounded).
667 yield* Effect.gen(function* () {
668 const result = yield* SQS.receiveMessage({
669 QueueUrl: queueUrl,
670 MaxNumberOfMessages: 10,
671 WaitTimeSeconds: 1,
672 VisibilityTimeout: 1,
673 });
674 const received = (result.Messages ?? []).map((m) => m.Body);
675 if (bodies.some((body) => received.includes(body))) {
676 return yield* Effect.fail(new MessageNotReceived());
677 }
678 }).pipe(
679 Effect.retry({
680 while: (e) => e._tag === "MessageNotReceived",
681 schedule: Schedule.max([
682 Schedule.fixed("2 seconds"),
683 Schedule.recurs(15),
684 ]),
685 }),
686 );
687 }),
688 );
689 });
690 },
691);