1import { adopt } from "@/AdoptPolicy";
2import * as AWS from "@/AWS";
3import { AWSEnvironment } from "@/AWS/Environment";
4import {
5 normalizePolicyDocument,
6 type PolicyStatement,
7} from "@/AWS/IAM/Policy";
8import { Queue } from "@/AWS/SQS";
9import * as Output from "@/Output";
10import * as Provider from "@/Provider";
11import { State } from "@/State";
12import * as Test from "@/Test/Alchemy";
13import * as SQS from "@distilled.cloud/aws/sqs";
14import { describe, expect } from "alchemy-test";
15import * as Console from "effect/Console";
16import * as Data from "effect/Data";
17import * as Duration from "effect/Duration";
18import * as Effect from "effect/Effect";
19import * as Schedule from "effect/Schedule";
20import * as HttpBody from "effect/http/HttpBody";
21import * as HttpClient from "effect/http/HttpClient";
22import { QueueSinkFunction, QueueSinkFunctionLive } from "./sink-handler";
23
24const { test } = Test.make({ providers: AWS.providers() });
25
26const provider = test.provider;
27const { test: adoptingTest } = Test.make({
28 providers: AWS.providers(),
29 adopt: true,
30});
31
32for (const fifo of [false, true]) {
33 provider(
34 `self-bound ${fifo ? "FIFO" : "standard"} queue creates, updates, and clears policies`,
35 (stack) =>
36 Effect.gen(function* () {
37 yield* stack.destroy();
38
39 const { accountId, region } = yield* AWSEnvironment.current;
40 const queueName = fifo
41 ? "alchemy-test-sqs-self-bound.fifo"
42 : "alchemy-test-sqs-self-bound";
43 const principal = { AWS: `arn:aws:iam::${accountId}:root` };
44 const explicit: PolicyStatement = {
45 Sid: "Explicit",
46 Effect: "Allow",
47 Principal: principal,
48 Action: ["sqs:GetQueueAttributes"],
49 Resource: `arn:aws:sqs:${region}:${accountId}:${queueName}`,
50 };
51 const deployQueue = (
52 action: string | undefined,
53 withPolicy: boolean,
54 visibilityTimeout: "30 seconds" | "60 seconds",
55 ) =>
56 stack.deploy(
57 Effect.gen(function* () {
58 const props = {
59 queueName,
60 visibilityTimeout,
61 policy: withPolicy
62 ? { Version: "2012-10-17", Statement: [explicit] }
63 : undefined,
64 redriveAllowPolicy: withPolicy
65 ? { redrivePermission: "denyAll" as const }
66 : undefined,
67 tags: { phase: visibilityTimeout },
68 };
69 const queue = yield* Queue(
70 "SelfBoundQueue",
71 fifo
72 ? {
73 ...props,
74 fifo: true,
75 contentBasedDeduplication: true,
76 deduplicationScope: "messageGroup",
77 fifoThroughputLimit: "perMessageGroupId",
78 }
79 : props,
80 );
81 if (action !== undefined) {
82 yield* queue.bind`SelfPolicy`({
83 policyStatements: [
84 {
85 Sid: "SelfBound",
86 Effect: "Allow",
87 Principal: principal,
88 Action: [action],
89 Resource: queue.queueArn,
90 },
91 ],
92 });
93 }
94 return queue;
95 }),
96 );
97 const boundStatement = (
98 queueArn: string,
99 action: string,
100 ): PolicyStatement => ({
101 Sid: "SelfBound",
102 Effect: "Allow",
103 Principal: principal,
104 Action: [action],
105 Resource: queueArn,
106 });
107
108 const created = yield* deployQueue(
109 "sqs:SendMessage",
110 true,
111 "30 seconds",
112 );
113 yield* waitForQueuePolicy(created.queueUrl, [
114 explicit,
115 boundStatement(created.queueArn, "sqs:SendMessage"),
116 ]);
117 yield* waitForQueueAttributeMatch(created.queueUrl, {
118 QueueArn: created.queueArn,
119 VisibilityTimeout: "30",
120 ...(fifo
121 ? {
122 FifoQueue: "true",
123 ContentBasedDeduplication: "true",
124 DeduplicationScope: "messageGroup",
125 FifoThroughputLimit: "perMessageGroupId",
126 }
127 : {}),
128 });
129 const tags = yield* SQS.listQueueTags({ QueueUrl: created.queueUrl });
130 expect(tags.Tags?.["alchemy::id"]).toBe("SelfBoundQueue");
131 expect(tags.Tags?.phase).toBe("30 seconds");
132
133 const updated = yield* deployQueue(
134 "sqs:ReceiveMessage",
135 true,
136 "30 seconds",
137 );
138 expect(updated).toEqual(created);
139 yield* waitForQueuePolicy(updated.queueUrl, [
140 explicit,
141 boundStatement(created.queueArn, "sqs:ReceiveMessage"),
142 ]);
143 const unbound = yield* deployQueue(undefined, true, "60 seconds");
144 expect(unbound).toEqual(created);
145 yield* waitForQueuePolicy(unbound.queueUrl, [explicit]);
146 yield* waitForQueueAttributeMatch(unbound.queueUrl, {
147 VisibilityTimeout: "60",
148 ...(fifo ? { FifoQueue: "true" } : {}),
149 });
150
151 const cleared = yield* deployQueue(undefined, false, "60 seconds");
152 expect(cleared).toEqual(created);
153 yield* waitForQueueAttributePredicate(
154 cleared.queueUrl,
155 (attrs) => !attrs.Policy && !attrs.RedriveAllowPolicy,
156 );
157
158 yield* stack.destroy();
159 yield* assertQueueDeleted(created.queueUrl);
160 }),
161 {
162 tags: ["provider:aws", "provider:aws:iam", "provider:aws:sqs", "live"],
163 timeout: 120_000,
164 },
165 );
166}
167
168for (const fifo of [false, true]) {
169 const run = fifo ? adoptingTest.provider : provider;
170 run(
171 `precreate preserves a foreign ${fifo ? "FIFO" : "standard"} queue when adoption is disabled on the resource`,
172 (stack) =>
173 Effect.gen(function* () {
174 yield* stack.destroy();
175 const ownerProps = {
176 visibilityTimeout: Duration.seconds(45),
177 tags: { purpose: "original-owner" },
178 };
179 const ownerProgram = Queue(
180 "OwnerQueue",
181 fifo ? { ...ownerProps, fifo: true } : ownerProps,
182 );
183 const owner = yield* stack.deploy(ownerProgram);
184 const originalTags = yield* SQS.listQueueTags({
185 QueueUrl: owner.queueUrl,
186 });
187 yield* SQS.sendMessage({
188 QueueUrl: owner.queueUrl,
189 MessageBody: "owned-message",
190 ...(fifo
191 ? { MessageGroupId: "owner", MessageDeduplicationId: "owner" }
192 : {}),
193 });
194
195 const conflict = yield* stack
196 .deploy(
197 Effect.gen(function* () {
198 yield* ownerProgram;
199 const dependency = yield* Queue("DependencyQueue");
200 const props = {
201 queueName: owner.queueName,
202 visibilityTimeout: Duration.seconds(90),
203 tags: { dependency: dependency.queueArn },
204 };
205 return yield* Queue(
206 "IntruderQueue",
207 fifo ? { ...props, fifo: true } : props,
208 ).pipe(adopt(false));
209 }),
210 )
211 .pipe(Effect.flip);
212 expect(JSON.stringify(conflict)).toContain("OwnedBySomeoneElse");
213 expect(JSON.stringify(conflict)).toContain("explicit adoption");
214
215 const attributes = yield* SQS.getQueueAttributes({
216 QueueUrl: owner.queueUrl,
217 AttributeNames: ["All"],
218 });
219 expect(attributes.Attributes?.QueueArn).toBe(owner.queueArn);
220 expect(attributes.Attributes?.VisibilityTimeout).toBe("45");
221 expect(attributes.Attributes?.FifoQueue === "true").toBe(fifo);
222 expect(
223 (yield* SQS.listQueueTags({ QueueUrl: owner.queueUrl })).Tags,
224 ).toEqual(originalTags.Tags);
225
226 // Removing the failed create must not delete the foreign queue.
227 yield* stack.deploy(ownerProgram);
228 expect(
229 (yield* SQS.listQueueTags({ QueueUrl: owner.queueUrl })).Tags,
230 ).toEqual(originalTags.Tags);
231 const messages = yield* SQS.receiveMessage({
232 QueueUrl: owner.queueUrl,
233 WaitTimeSeconds: 10,
234 MaxNumberOfMessages: 1,
235 });
236 expect(messages.Messages?.map((message) => message.Body)).toEqual([
237 "owned-message",
238 ]);
239 yield* stack.destroy();
240 yield* assertQueueDeleted(owner.queueUrl);
241 }),
242 { tags: ["provider:aws", "provider:aws:sqs", "live"], timeout: 120_000 },
243 );
244}
245
246describe.concurrent(
247 "queue mode changes",
248 { tags: ["provider:aws", "provider:aws:iam", "provider:aws:sqs", "live"] },
249 () => {
250 for (const initialFifo of [false, true]) {
251 for (const suffixed of [false, true]) {
252 provider(
253 `changing ${initialFifo ? "FIFO to standard" : "standard to FIFO"} with an explicit ${suffixed ? "suffixed" : "base"} name replaces the queue and updates references`,
254 (stack) =>
255 Effect.gen(function* () {
256 yield* stack.destroy();
257 const baseName = `alchemy-test-sqs-${stack.stage}-${initialFifo}-${suffixed}`;
258 const queueName = suffixed ? `${baseName}.fifo` : baseName;
259 const { accountId } = yield* AWSEnvironment.current;
260 const deployQueue = (
261 fifo: boolean,
262 visibilityTimeout: "30 seconds" | "45 seconds",
263 ) =>
264 stack.deploy(
265 Effect.gen(function* () {
266 const settings = yield* fifo === initialFifo
267 ? Effect.succeed(undefined)
268 : Queue("SettingsSource");
269 const props = {
270 queueName,
271 visibilityTimeout: settings
272 ? settings.queueArn.pipe(
273 Output.map(() => visibilityTimeout),
274 )
275 : visibilityTimeout,
276 };
277 const queue = yield* Queue(
278 "Queue",
279 fifo ? { ...props, fifo: true } : props,
280 );
281 yield* queue.bind`SelfPolicy`({
282 policyStatements: [
283 {
284 Effect: "Allow",
285 Principal: { AWS: `arn:aws:iam::${accountId}:root` },
286 Action: ["sqs:SendMessage"],
287 Resource: queue.queueArn,
288 },
289 ],
290 });
291 const reference = yield* Queue("Reference", {
292 tags: { target: queue.queueArn },
293 });
294 return { queue, reference };
295 }),
296 );
297 const original = yield* deployQueue(initialFifo, "30 seconds");
298 expect(original.queue.queueName).toBe(
299 initialFifo ? `${baseName}.fifo` : baseName,
300 );
301 const replacement = yield* deployQueue(
302 !initialFifo,
303 "30 seconds",
304 );
305 expect(replacement.queue.queueName).toBe(
306 initialFifo ? baseName : `${baseName}.fifo`,
307 );
308 expect(replacement.queue.queueArn).not.toBe(
309 original.queue.queueArn,
310 );
311 expect(replacement.reference).toEqual(original.reference);
312 yield* waitForQueueTags(
313 replacement.reference.queueUrl,
314 (tags) => tags.target === replacement.queue.queueArn,
315 );
316 yield* waitForQueueAttributeMatch(replacement.queue.queueUrl, {
317 QueueArn: replacement.queue.queueArn,
318 VisibilityTimeout: "30",
319 });
320 const attributes = yield* SQS.getQueueAttributes({
321 QueueUrl: replacement.queue.queueUrl,
322 AttributeNames: ["All"],
323 });
324 expect(attributes.Attributes?.FifoQueue === "true").toBe(
325 !initialFifo,
326 );
327 yield* waitForQueuePolicy(replacement.queue.queueUrl, [
328 {
329 Effect: "Allow",
330 Principal: { AWS: `arn:aws:iam::${accountId}:root` },
331 Action: ["sqs:SendMessage"],
332 Resource: replacement.queue.queueArn,
333 },
334 ]);
335 yield* assertQueueDeleted(original.queue.queueUrl);
336 const updated = yield* deployQueue(!initialFifo, "45 seconds");
337 expect(updated).toEqual(replacement);
338 yield* waitForQueueAttributeMatch(updated.queue.queueUrl, {
339 VisibilityTimeout: "45",
340 });
341 yield* SQS.sendMessage({
342 QueueUrl: updated.queue.queueUrl,
343 MessageBody: "replacement-message",
344 ...(!initialFifo
345 ? {
346 MessageGroupId: "replacement",
347 MessageDeduplicationId: "replacement",
348 }
349 : {}),
350 });
351 const messages = yield* SQS.receiveMessage({
352 QueueUrl: updated.queue.queueUrl,
353 WaitTimeSeconds: 10,
354 MaxNumberOfMessages: 1,
355 });
356 expect(messages.Messages?.map((message) => message.Body)).toEqual(
357 ["replacement-message"],
358 );
359 yield* stack.destroy();
360 yield* assertQueueDeleted(updated.queue.queueUrl);
361 yield* assertQueueDeleted(updated.reference.queueUrl);
362 }),
363 { timeout: 120_000 },
364 );
365 }
366 }
367 },
368);
369
370provider(
371 "create and delete queue with default props",
372 (stack) =>
373 Effect.gen(function* () {
374 yield* stack.destroy();
375
376 const queue = yield* stack.deploy(
377 Effect.gen(function* () {
378 return yield* Queue("DefaultQueue");
379 }),
380 );
381
382 expect(queue.queueName).toBeDefined();
383 expect(queue.queueUrl).toBeDefined();
384 expect(queue.queueArn).toBeDefined();
385
386 const queueAttributes = yield* SQS.getQueueAttributes({
387 QueueUrl: queue.queueUrl,
388 AttributeNames: ["All"],
389 });
390 expect(queueAttributes.Attributes?.QueueArn).toBe(queue.queueArn);
391 expect(queueAttributes.Attributes?.Policy).toBeFalsy();
392 expect(queueAttributes.Attributes?.FifoQueue).not.toBe("true");
393 expect(queueAttributes.Attributes?.VisibilityTimeout).toBe("30");
394
395 yield* SQS.sendMessage({
396 QueueUrl: queue.queueUrl,
397 MessageBody: "default-queue-message",
398 });
399 expect(yield* waitForQueueMessage(queue.queueUrl)).toBe(
400 "default-queue-message",
401 );
402
403 yield* stack.destroy();
404
405 yield* assertQueueDeleted(queue.queueUrl);
406 }),
407 { tags: ["provider:aws", "provider:aws:sqs", "live"] },
408);
409
410provider(
411 "create, update, delete standard queue",
412 (stack) =>
413 Effect.gen(function* () {
414 yield* stack.destroy();
415
416 const queue = yield* stack.deploy(
417 Effect.gen(function* () {
418 return yield* Queue("TestQueue", {
419 visibilityTimeout: "30 seconds",
420 delay: "0 seconds",
421 // exercise the non-string Duration.Input forms end-to-end
422 messageRetentionPeriod: Duration.days(4),
423 receiveMessageWaitTime: 10_000, // bare number = millis
424 });
425 }),
426 );
427
428 // Verify the queue was created
429 const queueAttributes = yield* SQS.getQueueAttributes({
430 QueueUrl: queue.queueUrl,
431 AttributeNames: ["All"],
432 });
433 expect(queueAttributes.Attributes?.VisibilityTimeout).toEqual("30");
434 expect(queueAttributes.Attributes?.DelaySeconds).toEqual("0");
435 expect(queueAttributes.Attributes?.MessageRetentionPeriod).toEqual(
436 "345600",
437 );
438 expect(queueAttributes.Attributes?.ReceiveMessageWaitTimeSeconds).toEqual(
439 "10",
440 );
441
442 // Update the queue
443 const updatedQueue = yield* stack.deploy(
444 Effect.gen(function* () {
445 return yield* Queue("TestQueue", {
446 visibilityTimeout: "60 seconds",
447 delay: "5 seconds",
448 messageRetentionPeriod: "5 days",
449 receiveMessageWaitTime: "20 seconds",
450 });
451 }),
452 );
453
454 // Verify the queue was updated (reads can lag briefly after SetQueueAttributes)
455 yield* waitForQueueAttributeMatch(updatedQueue.queueUrl, {
456 VisibilityTimeout: "60",
457 DelaySeconds: "5",
458 MessageRetentionPeriod: "432000",
459 ReceiveMessageWaitTimeSeconds: "20",
460 });
461
462 yield* stack.destroy();
463
464 yield* assertQueueDeleted(queue.queueUrl);
465 }),
466 { tags: ["provider:aws", "provider:aws:sqs", "live"] },
467);
468
469provider(
470 "create, update, delete fifo queue",
471 (stack) =>
472 Effect.gen(function* () {
473 yield* stack.destroy();
474
475 const queue = yield* stack.deploy(
476 Effect.gen(function* () {
477 return yield* Queue("TestFifoQueue", {
478 fifo: true,
479 contentBasedDeduplication: false,
480 visibilityTimeout: "30 seconds",
481 });
482 }),
483 );
484
485 // Verify the FIFO queue was created
486 expect(queue.queueUrl).toContain(".fifo");
487 expect(queue.queueName).toContain(".fifo");
488
489 const queueAttributes = yield* SQS.getQueueAttributes({
490 QueueUrl: queue.queueUrl,
491 AttributeNames: ["All"],
492 });
493 expect(queueAttributes.Attributes?.FifoQueue).toEqual("true");
494 expect(queueAttributes.Attributes?.ContentBasedDeduplication).toEqual(
495 "false",
496 );
497
498 // Update the FIFO queue to enable content-based deduplication
499 const updatedQueue = yield* stack.deploy(
500 Effect.gen(function* () {
501 return yield* Queue("TestFifoQueue", {
502 fifo: true,
503 contentBasedDeduplication: true,
504 visibilityTimeout: "60 seconds",
505 });
506 }),
507 );
508
509 // Verify the queue was updated (reads can lag briefly after SetQueueAttributes)
510 yield* waitForQueueAttributeMatch(updatedQueue.queueUrl, {
511 ContentBasedDeduplication: "true",
512 VisibilityTimeout: "60",
513 });
514
515 yield* stack.destroy();
516
517 yield* assertQueueDeleted(queue.queueUrl);
518 }),
519 { tags: ["provider:aws", "provider:aws:sqs", "live"] },
520);
521
522provider(
523 "create queue with custom name",
524 (stack) =>
525 Effect.gen(function* () {
526 yield* stack.destroy();
527
528 const queue = yield* stack.deploy(
529 Effect.gen(function* () {
530 return yield* Queue("CustomNameQueue", {
531 queueName: "my-custom-test-queue",
532 });
533 }),
534 );
535
536 expect(queue.queueName).toEqual("my-custom-test-queue");
537 expect(queue.queueUrl).toContain("my-custom-test-queue");
538
539 // Verify the queue exists
540 const queueAttributes = yield* SQS.getQueueAttributes({
541 QueueUrl: queue.queueUrl,
542 AttributeNames: ["All"],
543 });
544 expect(queueAttributes.Attributes).toBeDefined();
545
546 yield* stack.destroy();
547
548 yield* assertQueueDeleted(queue.queueUrl);
549 }),
550 { tags: ["provider:aws", "provider:aws:sqs", "live"] },
551);
552
553describe.concurrent(
554 "queue runtime and recovery",
555 { tags: ["provider:aws", "provider:aws:sqs", "live"] },
556 () => {
557 provider(
558 "QueueSink writes arbitrary messages through a deployed Lambda",
559 (stack) =>
560 Effect.gen(function* () {
561 yield* stack.destroy();
562
563 const apiFunction = yield* stack.deploy(
564 QueueSinkFunction.pipe(Effect.provide(QueueSinkFunctionLive)),
565 );
566 const baseUrl = apiFunction.functionUrl!.replace(/\/+$/, "");
567
568 const { queueUrl } = yield* waitForFunctionReady(`${baseUrl}/ready`);
569
570 // 25 messages > the SendMessageBatch limit of 10, so the batched sink
571 // must split the chunk into 3 sequential API calls (10 + 10 + 5).
572 const messages = Array.from(
573 { length: 25 },
574 (_, i) => `sink-${i}-${crypto.randomUUID()}`,
575 );
576 const response = yield* HttpClient.post(`${baseUrl}/sink`, {
577 body: yield* HttpBody.json({ messages }),
578 }).pipe(
579 // QueueSink can legitimately spend several seconds retrying a partial
580 // batch failure, but a stalled Function URL request must not consume
581 // the whole test timeout.
582 Effect.timeout("15 seconds"),
583 Effect.mapError(() => "not ready" as const),
584 Effect.flatMap((result) =>
585 result.status === 200
586 ? Effect.succeed(result)
587 : Effect.fail("not ready"),
588 ),
589 Effect.tapError(Console.log),
590 Effect.retry({
591 while: (error) => error === "not ready",
592 schedule: Schedule.fixed("3 seconds"),
593 times: 4,
594 }),
595 Effect.flatMap((result) => result.json),
596 );
597
598 expect((response as any).ok).toBe(true);
599 expect((response as any).count).toBe(messages.length);
600
601 const received = yield* waitForQueueMessages(
602 queueUrl,
603 messages.length,
604 );
605
606 expect(received.sort()).toEqual([...messages].sort());
607
608 yield* stack.destroy();
609
610 yield* assertQueueDeleted(queueUrl);
611 }),
612 { tags: ["provider:aws:lambda"], timeout: 240_000 },
613 );
614
615 provider(
616 "self-bound queue recovers a missing physical queue with persisted output",
617 (stack) =>
618 Effect.gen(function* () {
619 yield* stack.destroy();
620
621 const { accountId } = yield* AWSEnvironment.current;
622 const principal = { AWS: `arn:aws:iam::${accountId}:root` };
623 const deployQueue = (
624 visibilityTimeout: "30 seconds" | "60 seconds",
625 withBinding = true,
626 ) =>
627 stack.deploy(
628 Effect.gen(function* () {
629 const queue = yield* Queue("RecoveryQueue", {
630 visibilityTimeout,
631 tags: { purpose: "recovery" },
632 });
633 if (withBinding) {
634 yield* queue.bind`SelfPolicy`({
635 policyStatements: [
636 {
637 Effect: "Allow",
638 Principal: principal,
639 Action: ["sqs:SendMessage"],
640 Resource: queue.queueArn,
641 },
642 ],
643 });
644 }
645 return queue;
646 }),
647 );
648 const initial = yield* deployQueue("30 seconds");
649 yield* waitForQueuePolicy(initial.queueUrl, [
650 {
651 Effect: "Allow",
652 Principal: principal,
653 Action: ["sqs:SendMessage"],
654 Resource: initial.queueArn,
655 },
656 ]);
657
658 yield* SQS.deleteQueue({ QueueUrl: initial.queueUrl });
659 yield* assertQueueDeleted(initial.queueUrl);
660 const recovered = yield* deployQueue("60 seconds");
661 expect(recovered).toEqual(initial);
662 yield* waitForQueueAttributeMatch(recovered.queueUrl, {
663 QueueArn: initial.queueArn,
664 VisibilityTimeout: "60",
665 });
666 yield* waitForQueuePolicy(recovered.queueUrl, [
667 {
668 Effect: "Allow",
669 Principal: principal,
670 Action: ["sqs:SendMessage"],
671 Resource: recovered.queueArn,
672 },
673 ]);
674 const tags = yield* SQS.listQueueTags({
675 QueueUrl: recovered.queueUrl,
676 });
677 expect(tags.Tags?.purpose).toBe("recovery");
678 expect(tags.Tags?.["alchemy::id"]).toBe("RecoveryQueue");
679
680 const unbound = yield* deployQueue("60 seconds", false);
681 expect(unbound).toEqual(recovered);
682 yield* waitForQueueAttributePredicate(
683 unbound.queueUrl,
684 (attrs) => !attrs.Policy,
685 );
686
687 yield* stack.destroy();
688 yield* assertQueueDeleted(recovered.queueUrl);
689 }),
690 { tags: ["provider:aws:iam"], timeout: 120_000 },
691 );
692 },
693);
694
695// State-loss fixtures need cleanup even if adoption fails before state is restored.
696const ADOPT_QUEUE_NAME = "alchemy-test-sqs-adopt";
697const TAKEOVER_QUEUE_NAME = "alchemy-test-sqs-takeover";
698
699provider(
700 "owned queue (matching alchemy tags) is silently adopted without --adopt",
701 (stack) =>
702 Effect.gen(function* () {
703 yield* stack.destroy();
704
705 const queueName = ADOPT_QUEUE_NAME;
706
707 const { accountId, region } = yield* AWSEnvironment.current;
708 const principal = { AWS: `arn:aws:iam::${accountId}:root` };
709 const initial = yield* stack.deploy(
710 Effect.gen(function* () {
711 return yield* Queue("AdoptableQueue", {
712 queueName,
713 visibilityTimeout: "30 seconds",
714 redriveAllowPolicy: { redrivePermission: "denyAll" },
715 tags: { stale: "remove" },
716 policy: {
717 Statement: [
718 {
719 Sid: "Stale",
720 Effect: "Allow",
721 Principal: principal,
722 Action: ["sqs:GetQueueAttributes"],
723 Resource: `arn:aws:sqs:${region}:${accountId}:${queueName}`,
724 },
725 ],
726 },
727 });
728 }),
729 );
730 expect(initial.queueName).toEqual(queueName);
731 yield* waitForQueueAttributePredicate(
732 initial.queueUrl,
733 (attrs) => !!attrs.Policy && !!attrs.RedriveAllowPolicy,
734 );
735
736 // Wipe state — queue stays in SQS.
737 yield* Effect.gen(function* () {
738 const state = yield* yield* State;
739 yield* state.delete({
740 stack: stack.name,
741 stage: stack.stage,
742 fqn: "AdoptableQueue",
743 });
744 }).pipe(Effect.provide(stack.state));
745
746 const adopted = yield* stack.deploy(
747 Effect.gen(function* () {
748 const dependency = yield* Queue("AdoptionDependency");
749 const queue = yield* Queue("AdoptableQueue", {
750 queueName,
751 visibilityTimeout: "60 seconds",
752 tags: {
753 adopted: dependency.queueArn.pipe(Output.map(() => "true")),
754 },
755 });
756 yield* queue.bind`SelfPolicy`({
757 policyStatements: [
758 {
759 Effect: "Allow",
760 Principal: principal,
761 Action: ["sqs:SendMessage"],
762 Resource: queue.queueArn,
763 },
764 ],
765 });
766 return queue;
767 }),
768 );
769
770 expect(adopted.queueArn).toEqual(initial.queueArn);
771 expect(adopted.queueUrl).toEqual(initial.queueUrl);
772 yield* waitForQueuePolicy(adopted.queueUrl, [
773 {
774 Effect: "Allow",
775 Principal: principal,
776 Action: ["sqs:SendMessage"],
777 Resource: adopted.queueArn,
778 },
779 ]);
780 yield* waitForQueueAttributePredicate(
781 adopted.queueUrl,
782 (attrs) =>
783 attrs.VisibilityTimeout === "60" && !attrs.RedriveAllowPolicy,
784 );
785 yield* waitForQueueTags(
786 adopted.queueUrl,
787 (tags) =>
788 tags.adopted === "true" &&
789 tags.stale === undefined &&
790 tags["alchemy::id"] === "AdoptableQueue",
791 );
792
793 yield* stack.destroy();
794 yield* assertQueueDeleted(initial.queueUrl);
795 }).pipe(
796 Effect.ensuring(deleteQueueIfExists(ADOPT_QUEUE_NAME).pipe(Effect.orDie)),
797 ),
798 { tags: ["provider:aws", "provider:aws:iam", "provider:aws:sqs", "live"] },
799);
800
801describe.concurrent("queue adoption", () => {
802 for (const scope of ["resolved", "resource", "stack", "default"] as const) {
803 const queueName = `${TAKEOVER_QUEUE_NAME}-${scope}`;
804 const run = scope === "default" ? adoptingTest.provider : provider;
805 run(
806 `foreign-tagged queue converges settings and policies during ${scope} adoption`,
807 (stack) =>
808 Effect.gen(function* () {
809 yield* stack.destroy();
810 const { accountId } = yield* AWSEnvironment.current;
811 const principal = { AWS: `arn:aws:iam::${accountId}:root` };
812 const original = yield* stack.deploy(
813 Queue("Original", { queueName, tags: { stale: "remove" } }),
814 );
815 yield* SQS.sendMessage({
816 QueueUrl: original.queueUrl,
817 MessageBody: "adopted-message",
818 }).pipe(
819 Effect.retry({
820 while: (error) => error._tag === "QueueDoesNotExist",
821 schedule: Schedule.spaced("1 second"),
822 times: 10,
823 }),
824 );
825 yield* Effect.gen(function* () {
826 const state = yield* yield* State;
827 yield* state.delete({
828 stack: stack.name,
829 stage: stack.stage,
830 fqn: "Original",
831 });
832 }).pipe(Effect.provide(stack.state));
833 const deployment = stack.deploy(
834 Effect.gen(function* () {
835 const dependency = yield* Queue("AdoptionDependency");
836 const registration = Queue("Different", {
837 queueName,
838 visibilityTimeout:
839 scope === "resolved"
840 ? Duration.seconds(60)
841 : dependency.queueArn.pipe(
842 Output.map(() => Duration.seconds(60)),
843 ),
844 tags: {
845 adopted:
846 scope === "resolved"
847 ? "true"
848 : dependency.queueArn.pipe(Output.map(() => "true")),
849 },
850 });
851 const queue = yield* scope === "resource"
852 ? registration.pipe(adopt(true))
853 : registration;
854 yield* queue.bind`SelfPolicy`({
855 policyStatements: [
856 {
857 Effect: "Allow",
858 Principal: principal,
859 Action: ["sqs:SendMessage"],
860 Resource: queue.queueArn,
861 },
862 ],
863 });
864 return { queue, dependency };
865 }),
866 );
867 const { queue: takenOver, dependency } = yield* scope === "stack" ||
868 scope === "resolved"
869 ? deployment.pipe(adopt(true))
870 : deployment;
871 expect(takenOver.queueName).toBe(queueName);
872 expect(takenOver.queueUrl).toBe(original.queueUrl);
873 yield* waitForQueuePolicy(takenOver.queueUrl, [
874 {
875 Effect: "Allow",
876 Principal: principal,
877 Action: ["sqs:SendMessage"],
878 Resource: takenOver.queueArn,
879 },
880 ]);
881 yield* waitForQueueTags(
882 takenOver.queueUrl,
883 (tags) =>
884 tags["alchemy::id"] === "Different" &&
885 tags.adopted === "true" &&
886 tags.stale === undefined,
887 );
888 yield* waitForQueueAttributeMatch(takenOver.queueUrl, {
889 VisibilityTimeout: "60",
890 });
891 const messages = yield* SQS.receiveMessage({
892 QueueUrl: takenOver.queueUrl,
893 WaitTimeSeconds: 10,
894 MaxNumberOfMessages: 1,
895 });
896 expect(messages.Messages?.map((message) => message.Body)).toEqual([
897 "adopted-message",
898 ]);
899 yield* stack.destroy();
900 yield* assertQueueDeleted(takenOver.queueUrl);
901 yield* assertQueueDeleted(dependency.queueUrl);
902 }).pipe(
903 Effect.ensuring(deleteQueueIfExists(queueName).pipe(Effect.orDie)),
904 ),
905 { tags: ["provider:aws", "provider:aws:sqs", "live"], timeout: 120_000 },
906 );
907 }
908});
909
910provider(
911 "list enumerates the deployed queue",
912 (stack) =>
913 Effect.gen(function* () {
914 yield* stack.destroy();
915
916 const deployed = yield* stack.deploy(
917 Effect.gen(function* () {
918 return yield* Queue("ListQueue");
919 }),
920 );
921
922 const provider = yield* Provider.findProvider(Queue);
923
924 // SQS is eventually consistent: a freshly-created queue may not appear in
925 // listQueues immediately. Retry the list assertion on a bounded schedule.
926 yield* Effect.gen(function* () {
927 const all = yield* provider.list();
928 if (!all.some((q) => q.queueArn === deployed.queueArn)) {
929 return yield* Effect.fail(new QueueNotListed());
930 }
931 }).pipe(
932 Effect.retry({
933 while: (e) => e._tag === "QueueNotListed",
934 schedule: Schedule.spaced("3 seconds"),
935 times: 10,
936 }),
937 );
938
939 yield* stack.destroy();
940
941 yield* assertQueueDeleted(deployed.queueUrl);
942 }),
943 { tags: ["provider:aws", "provider:aws:sqs", "live"], timeout: 120_000 },
944);
945
946provider(
947 "DLQ redrive policy round-trips and can be removed",
948 (stack) =>
949 Effect.gen(function* () {
950 yield* stack.destroy();
951
952 // Create the DLQ and source together, keeping both deployed across
953 // steps to avoid the engine replace+remove-dependency deadlock.
954 const deployBoth = (withRedrive: boolean) =>
955 stack.deploy(
956 Effect.gen(function* () {
957 const dlq = yield* Queue("RedriveDLQ");
958 const source = yield* Queue(
959 "RedriveSource",
960 withRedrive
961 ? {
962 redrivePolicy: {
963 deadLetterTargetArn: dlq.queueArn,
964 maxReceiveCount: 3,
965 },
966 }
967 : {},
968 );
969 return { dlq, source };
970 }),
971 );
972
973 const { source } = yield* deployBoth(true);
974
975 yield* waitForQueueAttributePredicate(source.queueUrl, (attrs) => {
976 if (!attrs.RedrivePolicy) return false;
977 const parsed = JSON.parse(attrs.RedrivePolicy);
978 return parsed.maxReceiveCount === 3;
979 });
980
981 // Remove the redrive policy on update; it must be cleared.
982 const { source: updated } = yield* deployBoth(false);
983 yield* waitForQueueAttributePredicate(
984 updated.queueUrl,
985 (attrs) => !attrs.RedrivePolicy,
986 );
987
988 yield* stack.destroy();
989 yield* assertQueueDeleted(source.queueUrl);
990 }),
991 { tags: ["provider:aws", "provider:aws:sqs", "live"], timeout: 120_000 },
992);
993
994provider(
995 "redriveAllowPolicy is set on the dead-letter queue",
996 (stack) =>
997 Effect.gen(function* () {
998 yield* stack.destroy();
999
1000 const { dlq } = yield* stack.deploy(
1001 Effect.gen(function* () {
1002 const source = yield* Queue("AllowSource");
1003 const dlq = yield* Queue("AllowDLQ", {
1004 redriveAllowPolicy: {
1005 redrivePermission: "byQueue",
1006 sourceQueueArns: [source.queueArn],
1007 },
1008 });
1009 return { source, dlq };
1010 }),
1011 );
1012
1013 yield* waitForQueueAttributePredicate(dlq.queueUrl, (attrs) => {
1014 if (!attrs.RedriveAllowPolicy) return false;
1015 const parsed = JSON.parse(attrs.RedriveAllowPolicy);
1016 return parsed.redrivePermission === "byQueue";
1017 });
1018
1019 yield* stack.destroy();
1020 yield* assertQueueDeleted(dlq.queueUrl);
1021 }),
1022 { tags: ["provider:aws", "provider:aws:sqs", "live"], timeout: 120_000 },
1023);
1024
1025provider(
1026 "SSE-SQS encryption enables sqs-managed key",
1027 (stack) =>
1028 Effect.gen(function* () {
1029 yield* stack.destroy();
1030
1031 const queue = yield* stack.deploy(
1032 Effect.gen(function* () {
1033 return yield* Queue("SseSqsQueue", { sqsManagedSseEnabled: true });
1034 }),
1035 );
1036
1037 yield* waitForQueueAttributeMatch(queue.queueUrl, {
1038 SqsManagedSseEnabled: "true",
1039 });
1040
1041 yield* stack.destroy();
1042 yield* assertQueueDeleted(queue.queueUrl);
1043 }),
1044 { tags: ["provider:aws", "provider:aws:sqs", "live"] },
1045);
1046
1047provider(
1048 "SSE-KMS encryption with AWS-managed key",
1049 (stack) =>
1050 Effect.gen(function* () {
1051 yield* stack.destroy();
1052
1053 const queue = yield* stack.deploy(
1054 Effect.gen(function* () {
1055 return yield* Queue("KmsQueue", {
1056 kmsMasterKeyId: "alias/aws/sqs",
1057 kmsDataKeyReusePeriod: "300 seconds",
1058 });
1059 }),
1060 );
1061
1062 yield* waitForQueueAttributeMatch(queue.queueUrl, {
1063 KmsMasterKeyId: "alias/aws/sqs",
1064 KmsDataKeyReusePeriodSeconds: "300",
1065 });
1066
1067 yield* stack.destroy();
1068 yield* assertQueueDeleted(queue.queueUrl);
1069 }),
1070 { tags: ["provider:aws", "provider:aws:sqs", "live"] },
1071);
1072
1073provider(
1074 "SSE-KMS accepts an Output-valued false SSE-SQS setting during precreate",
1075 (stack) =>
1076 Effect.gen(function* () {
1077 yield* stack.destroy();
1078 const { source, queue } = yield* stack.deploy(
1079 Effect.gen(function* () {
1080 const source = yield* Queue("EncryptionSource");
1081 const queue = yield* Queue("OutputKmsQueue", {
1082 kmsMasterKeyId: "alias/aws/sqs",
1083 sqsManagedSseEnabled: source.queueArn.pipe(Output.map(() => false)),
1084 });
1085 return { source, queue };
1086 }),
1087 );
1088 yield* waitForQueueAttributeMatch(queue.queueUrl, {
1089 KmsMasterKeyId: "alias/aws/sqs",
1090 });
1091 yield* stack.destroy();
1092 yield* assertQueueDeleted(source.queueUrl);
1093 yield* assertQueueDeleted(queue.queueUrl);
1094 }),
1095 { tags: ["provider:aws", "provider:aws:sqs", "live"], timeout: 120_000 },
1096);
1097
1098provider(
1099 "kmsMasterKeyId and sqsManagedSseEnabled together fail fast",
1100 (stack) =>
1101 Effect.gen(function* () {
1102 yield* stack.destroy();
1103
1104 const result = yield* stack
1105 .deploy(
1106 Effect.gen(function* () {
1107 return yield* Queue("ConflictQueue", {
1108 kmsMasterKeyId: "alias/aws/sqs",
1109 sqsManagedSseEnabled: true,
1110 });
1111 }),
1112 )
1113 .pipe(Effect.flip);
1114
1115 // The typed validation error surfaces (possibly wrapped by the engine).
1116 expect(JSON.stringify(result)).toContain("SqsEncryptionConflict");
1117
1118 yield* stack.destroy();
1119 }),
1120 { tags: ["provider:aws", "provider:aws:sqs", "live"] },
1121);
1122
1123for (const property of ["fifo", "queueName"] as const) {
1124 provider(
1125 `rejects unresolved ${property} before creating a queue with the wrong identity`,
1126 (stack) =>
1127 Effect.gen(function* () {
1128 yield* stack.destroy();
1129 const queueName = `alchemy-test-sqs-unresolved-${property.toLowerCase()}`;
1130 const result = yield* stack
1131 .deploy(
1132 Effect.gen(function* () {
1133 const source = yield* Queue("IdentitySource");
1134 return yield* Queue(
1135 "UnresolvedIdentity",
1136 property === "fifo"
1137 ? {
1138 queueName,
1139 fifo: source.queueArn.pipe(
1140 Output.map(() => false as const),
1141 ),
1142 }
1143 : {
1144 queueName: source.queueArn.pipe(
1145 Output.map(() => queueName),
1146 ),
1147 },
1148 );
1149 }),
1150 )
1151 .pipe(Effect.flip);
1152 expect(JSON.stringify(result)).toContain("UnresolvedQueueIdentity");
1153 const queues = yield* SQS.listQueues({ QueueNamePrefix: queueName });
1154 expect(queues.QueueUrls ?? []).toEqual([]);
1155 yield* stack.destroy();
1156 }),
1157 { tags: ["provider:aws", "provider:aws:sqs", "live"], timeout: 120_000 },
1158 );
1159}
1160
1161provider(
1162 "missing dead-letter queues produce a typed validation error",
1163 (stack) =>
1164 Effect.gen(function* () {
1165 yield* stack.destroy();
1166 const queue = yield* stack.deploy(Queue("MissingDeadLetterQueueSource"));
1167 const { accountId, region } = yield* AWSEnvironment.current;
1168 const result = yield* SQS.setQueueAttributes({
1169 QueueUrl: queue.queueUrl,
1170 Attributes: {
1171 RedrivePolicy: JSON.stringify({
1172 deadLetterTargetArn: `arn:aws:sqs:${region}:${accountId}:alchemy-test-sqs-missing-dead-letter`,
1173 maxReceiveCount: 3,
1174 }),
1175 },
1176 }).pipe(Effect.flip);
1177 expect(result._tag).toBe("InvalidParameterValueException");
1178 expect(result.message).toContain("Dead letter target does not exist");
1179 yield* stack.destroy();
1180 yield* assertQueueDeleted(queue.queueUrl);
1181 }),
1182 { tags: ["provider:aws", "provider:aws:sqs", "live"] },
1183);
1184
1185provider(
1186 "user tags coexist with internal tags and can be removed",
1187 (stack) =>
1188 Effect.gen(function* () {
1189 yield* stack.destroy();
1190
1191 const withTags = yield* stack.deploy(
1192 Effect.gen(function* () {
1193 return yield* Queue("TaggedQueue", {
1194 tags: { team: "payments", env: "test" },
1195 });
1196 }),
1197 );
1198
1199 const tags1 = yield* SQS.listQueueTags({ QueueUrl: withTags.queueUrl });
1200 expect(tags1.Tags?.team).toEqual("payments");
1201 expect(tags1.Tags?.env).toEqual("test");
1202 expect(tags1.Tags?.["alchemy::id"]).toBeDefined();
1203
1204 // Remove one tag, change another.
1205 const updated = yield* stack.deploy(
1206 Effect.gen(function* () {
1207 return yield* Queue("TaggedQueue", {
1208 tags: { team: "platform" },
1209 });
1210 }),
1211 );
1212
1213 yield* Effect.gen(function* () {
1214 const tags = yield* SQS.listQueueTags({ QueueUrl: updated.queueUrl });
1215 const t = tags.Tags ?? {};
1216 if (t.team !== "platform" || t.env !== undefined) {
1217 return yield* Effect.fail(new QueueAttributesNotReady());
1218 }
1219 // internal tags survive untag.
1220 expect(t["alchemy::id"]).toBeDefined();
1221 }).pipe(
1222 Effect.retry({
1223 while: (e) => e._tag === "QueueAttributesNotReady",
1224 schedule: Schedule.spaced("2 seconds"),
1225 times: 10,
1226 }),
1227 );
1228
1229 yield* stack.destroy();
1230 yield* assertQueueDeleted(withTags.queueUrl);
1231 }),
1232 { tags: ["provider:aws", "provider:aws:sqs", "live"], timeout: 120_000 },
1233);
1234
1235provider(
1236 "FIFO source with FIFO dead-letter queue (no type mismatch)",
1237 (stack) =>
1238 Effect.gen(function* () {
1239 yield* stack.destroy();
1240
1241 const { source } = yield* stack.deploy(
1242 Effect.gen(function* () {
1243 const dlq = yield* Queue("FifoDLQ", { fifo: true });
1244 const source = yield* Queue("FifoSource", {
1245 fifo: true,
1246 redrivePolicy: {
1247 deadLetterTargetArn: dlq.queueArn,
1248 maxReceiveCount: 5,
1249 },
1250 });
1251 return { dlq, source };
1252 }),
1253 );
1254
1255 yield* waitForQueueAttributePredicate(source.queueUrl, (attrs) => {
1256 if (!attrs.RedrivePolicy) return false;
1257 return JSON.parse(attrs.RedrivePolicy).maxReceiveCount === 5;
1258 });
1259
1260 yield* stack.destroy();
1261 yield* assertQueueDeleted(source.queueUrl);
1262 }),
1263 { tags: ["provider:aws", "provider:aws:sqs", "live"], timeout: 120_000 },
1264);
1265
1266class QueueNotListed extends Data.TaggedError("QueueNotListed") {}
1267
1268class QueueStillExists extends Data.TaggedError("QueueStillExists") {}
1269
1270class FunctionNotReady extends Data.TaggedError("FunctionNotReady") {}
1271
1272class QueueMessageNotReady extends Data.TaggedError("QueueMessageNotReady") {}
1273
1274class QueueAttributesNotReady extends Data.TaggedError(
1275 "QueueAttributesNotReady",
1276) {}
1277
1278const waitForFunctionReady = (url: string) =>
1279 HttpClient.get(url).pipe(
1280 Effect.timeout("4 seconds"),
1281 Effect.mapError(() => new FunctionNotReady()),
1282 Effect.flatMap((response) =>
1283 response.status === 200
1284 ? (response.json as Effect.Effect<{ queueUrl: string }>)
1285 : Effect.fail(new FunctionNotReady()),
1286 ),
1287 // A freshly-deployed function can briefly serve a 200 before its captured
1288 // env vars (the queue URL) have finished propagating, so treat a missing
1289 // queueUrl as "not ready yet" and keep polling.
1290 Effect.flatMap((json: any) =>
1291 typeof json?.queueUrl === "string"
1292 ? Effect.succeed({ queueUrl: json.queueUrl as string })
1293 : Effect.fail(new FunctionNotReady()),
1294 ),
1295 Effect.retry({
1296 while: (error) => error._tag === "FunctionNotReady",
1297 schedule: Schedule.fixed("4 seconds"),
1298 times: 10,
1299 }),
1300 );
1301
1302/** Poll until GetQueueAttributes reflects SetQueueAttributes (SQS is eventually consistent). */
1303const waitForQueueAttributeMatch = Effect.fn(function* (
1304 queueUrl: string,
1305 expected: Record<string, string>,
1306) {
1307 yield* Effect.gen(function* () {
1308 const result = yield* SQS.getQueueAttributes({
1309 QueueUrl: queueUrl,
1310 AttributeNames: ["All"],
1311 });
1312 const attrs = result.Attributes ?? {};
1313 for (const [name, value] of Object.entries(expected)) {
1314 if (attrs[name] !== value) {
1315 return yield* Effect.fail(new QueueAttributesNotReady());
1316 }
1317 }
1318 }).pipe(
1319 Effect.retry({
1320 // SQS is eventually consistent: a freshly-created queue can briefly
1321 // 400 with `QueueDoesNotExist` on getQueueAttributes before it settles.
1322 while: (e) =>
1323 e._tag === "QueueAttributesNotReady" || e._tag === "QueueDoesNotExist",
1324 schedule: Schedule.spaced("2 seconds"),
1325 times: 10,
1326 }),
1327 );
1328});
1329
1330/** Poll until a predicate over the queue's attributes holds. */
1331const waitForQueueAttributePredicate = Effect.fn(function* (
1332 queueUrl: string,
1333 predicate: (attrs: Record<string, string | undefined>) => boolean,
1334) {
1335 yield* Effect.gen(function* () {
1336 const result = yield* SQS.getQueueAttributes({
1337 QueueUrl: queueUrl,
1338 AttributeNames: ["All"],
1339 });
1340 if (!predicate(result.Attributes ?? {})) {
1341 return yield* Effect.fail(new QueueAttributesNotReady());
1342 }
1343 }).pipe(
1344 Effect.retry({
1345 // See `waitForQueueAttributeMatch`: ride out the brief post-create
1346 // `QueueDoesNotExist` window as well as the predicate-not-yet-true case.
1347 while: (e) =>
1348 e._tag === "QueueAttributesNotReady" || e._tag === "QueueDoesNotExist",
1349 schedule: Schedule.spaced("4 seconds"),
1350 times: 10,
1351 }),
1352 );
1353});
1354
1355const waitForQueuePolicy = (queueUrl: string, statements: PolicyStatement[]) =>
1356 waitForQueueAttributePredicate(
1357 queueUrl,
1358 (attrs) =>
1359 attrs.Policy !== undefined &&
1360 normalizePolicyDocument(attrs.Policy) ===
1361 normalizePolicyDocument({
1362 Version: "2012-10-17",
1363 Statement: statements,
1364 }),
1365 );
1366
1367const waitForQueueTags = (
1368 queueUrl: string,
1369 predicate: (tags: Record<string, string | undefined>) => boolean,
1370) =>
1371 SQS.listQueueTags({ QueueUrl: queueUrl }).pipe(
1372 Effect.flatMap((result) =>
1373 predicate(result.Tags ?? {})
1374 ? Effect.void
1375 : Effect.fail(new QueueAttributesNotReady()),
1376 ),
1377 Effect.retry({
1378 while: (error) =>
1379 error._tag === "QueueAttributesNotReady" ||
1380 error._tag === "QueueDoesNotExist",
1381 schedule: Schedule.spaced("2 seconds"),
1382 times: 10,
1383 }),
1384 );
1385
1386const waitForQueueMessages = Effect.fn(function* (
1387 queueUrl: string,
1388 count: number,
1389) {
1390 const messages: string[] = [];
1391
1392 while (messages.length < count) {
1393 messages.push(yield* waitForQueueMessage(queueUrl));
1394 }
1395
1396 return messages;
1397});
1398
1399const waitForQueueMessage = (queueUrl: string) =>
1400 Effect.gen(function* () {
1401 const result = yield* SQS.receiveMessage({
1402 QueueUrl: queueUrl,
1403 MaxNumberOfMessages: 1,
1404 WaitTimeSeconds: 2,
1405 VisibilityTimeout: 5,
1406 });
1407
1408 const message = result.Messages?.[0];
1409 if (!message?.Body || !message.ReceiptHandle) {
1410 return yield* Effect.fail(new QueueMessageNotReady());
1411 }
1412
1413 const body = message.Body;
1414
1415 yield* SQS.deleteMessage({
1416 QueueUrl: queueUrl,
1417 ReceiptHandle: message.ReceiptHandle,
1418 });
1419
1420 return body;
1421 }).pipe(
1422 Effect.retry({
1423 while: (error) => error._tag === "QueueMessageNotReady",
1424 schedule: Schedule.spaced("2 seconds"),
1425 times: 10,
1426 }),
1427 );
1428
1429// Finalizer for adoption fixtures whose state has deliberately been removed.
1430const deleteQueueIfExists = (queueName: string) =>
1431 SQS.getQueueUrl({ QueueName: queueName }).pipe(
1432 Effect.flatMap((r) =>
1433 SQS.deleteQueue({ QueueUrl: r.QueueUrl! }).pipe(
1434 Effect.andThen(assertQueueDeleted(r.QueueUrl!)),
1435 ),
1436 ),
1437 Effect.catchTag("QueueDoesNotExist", () => Effect.void),
1438 Effect.retry({
1439 while: (error) => error._tag === "RequestThrottled",
1440 schedule: Schedule.spaced("2 seconds"),
1441 times: 5,
1442 }),
1443 );
1444
1445const assertQueueDeleted = (queueUrl: string) =>
1446 Effect.gen(function* () {
1447 const attributesAbsent = yield* SQS.getQueueAttributes({
1448 QueueUrl: queueUrl,
1449 AttributeNames: ["QueueArn"],
1450 }).pipe(
1451 Effect.as(false),
1452 Effect.catchTag("QueueDoesNotExist", () => Effect.succeed(true)),
1453 );
1454 const urlAbsent = yield* SQS.getQueueUrl({
1455 QueueName: queueUrl.slice(queueUrl.lastIndexOf("/") + 1),
1456 }).pipe(
1457 Effect.as(false),
1458 Effect.catchTag("QueueDoesNotExist", () => Effect.succeed(true)),
1459 );
1460 if (!attributesAbsent || !urlAbsent) {
1461 return yield* Effect.fail(new QueueStillExists());
1462 }
1463 }).pipe(
1464 Effect.retry({
1465 while: (error) => error._tag === "QueueStillExists",
1466 schedule: Schedule.spaced("6 seconds"),
1467 times: 10,
1468 }),
1469 );