1import * as sqs from "@distilled.cloud/aws/sqs";
2import * as Config from "effect/Config";
3import * as Data from "effect/Data";
4import type * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Option from "effect/Option";
7import * as Schedule from "effect/Schedule";
8import * as Stream from "effect/Stream";
9import { AdoptPolicy, OwnedBySomeoneElse, Unowned } from "../../AdoptPolicy.ts";
10import { AlchemyContext } from "../../AlchemyContext.ts";
11import { isResolved } from "../../Diff.ts";
12import { isOutput } from "../../Output.ts";
13import { createPhysicalName } from "../../PhysicalName.ts";
14import * as Provider from "../../Provider.ts";
15import { Resource, type ResourceBinding } from "../../Resource.ts";
16import { Stack } from "../../Stack.ts";
17import { createInternalTags, diffTags, hasAlchemyTags } from "../../Tags.ts";
18import { toWireSeconds } from "../../Util/Duration.ts";
19import { AWSEnvironment, type AccountID } from "../Environment.ts";
20import type { PolicyStatement } from "../IAM/Policy.ts";
21import type { Providers } from "../Providers.ts";
22import type { RegionID } from "../Region.ts";
23
24export type QueueName = string;
25export type QueueArn = `arn:aws:sqs:${RegionID}:${AccountID}:${QueueName}`;
26export type QueueUrl = string;
27
28export type QueueProps = {
29 /**
30 * Name of the queue. Must be a literal value, available before precreation.
31 * The reserved `.fifo` suffix is added for FIFO queues and removed for standard queues.
32 * @default ${app}-${stage}-${id}?.fifo
33 */
34 queueName?: string;
35 /**
36 * Delay applied to all messages in the queue (`0` - `900` seconds).
37 * Accepts any `Duration.Input` (e.g. `"30 seconds"`,
38 * `Duration.seconds(30)`; a bare number is milliseconds); the wire unit
39 * is whole seconds.
40 * @default 0
41 */
42 delay?: Duration.Input;
43 /**
44 * Maximum message size in bytes (`1,024` - `1,048,576`).
45 * @default 1048576
46 */
47 maximumMessageSize?: number;
48 /**
49 * How long messages are retained (`60` - `1,209,600` seconds). Accepts
50 * any `Duration.Input` (e.g. `"4 days"`, `Duration.days(4)`; a bare
51 * number is milliseconds); the wire unit is whole seconds.
52 * @default 4 days
53 */
54 messageRetentionPeriod?: Duration.Input;
55 /**
56 * How long `ReceiveMessage` waits for a message (`0` - `20` seconds).
57 * Accepts any `Duration.Input` (e.g. `"20 seconds"`,
58 * `Duration.seconds(20)`; a bare number is milliseconds); the wire unit
59 * is whole seconds.
60 * @default 0
61 */
62 receiveMessageWaitTime?: Duration.Input;
63 /**
64 * Visibility timeout (`0` - `43,200` seconds). Accepts any
65 * `Duration.Input` (e.g. `"30 seconds"`, `Duration.seconds(30)`; a bare
66 * number is milliseconds); the wire unit is whole seconds.
67 * @default 30 seconds
68 */
69 visibilityTimeout?: Duration.Input;
70 /**
71 * Dead-letter queue redrive policy. Failed messages are moved to the
72 * dead-letter queue after `maxReceiveCount` receive attempts. The
73 * dead-letter queue must be the same type (a FIFO source requires a
74 * FIFO dead-letter queue).
75 */
76 redrivePolicy?: {
77 /**
78 * The ARN of the dead-letter queue that failed messages are moved to.
79 */
80 deadLetterTargetArn: string;
81 /**
82 * The number of times a message is received before it is moved to the
83 * dead-letter queue (`1` - `1000`).
84 */
85 maxReceiveCount: number;
86 };
87 /**
88 * Redrive-allow policy. Set on the **dead-letter queue** to authorize
89 * which source queues may use it.
90 */
91 redriveAllowPolicy?: {
92 /**
93 * Whether all, none, or a specified list of source queues may use this
94 * queue as a dead-letter queue.
95 */
96 redrivePermission: "allowAll" | "denyAll" | "byQueue";
97 /**
98 * The ARNs of the source queues permitted to use this dead-letter
99 * queue. Only valid (and required) when `redrivePermission` is
100 * `byQueue` (up to 10 ARNs).
101 */
102 sourceQueueArns?: string[];
103 };
104 /**
105 * An access-control policy document (IAM policy JSON) attached to the
106 * queue. Provided as a JSON string or a plain object. Merged with any
107 * policy statements contributed by capability bindings.
108 */
109 policy?: string | Record<string, any>;
110 /**
111 * The ID, alias, or ARN of a KMS key for server-side encryption (SSE-KMS).
112 * Use `alias/aws/sqs` for the AWS-managed SQS key. Mutually exclusive with
113 * `sqsManagedSseEnabled`.
114 */
115 kmsMasterKeyId?: string;
116 /**
117 * How long SQS reuses a data key before calling KMS again (`60` -
118 * `86,400` seconds). Accepts any `Duration.Input` (e.g. `"5 minutes"`,
119 * `Duration.minutes(5)`; a bare number is milliseconds); the wire unit
120 * is whole seconds. Only meaningful with `kmsMasterKeyId`.
121 * @default 5 minutes
122 */
123 kmsDataKeyReusePeriod?: Duration.Input;
124 /**
125 * Enables server-side encryption using SQS-owned keys (SSE-SQS).
126 * Mutually exclusive with `kmsMasterKeyId`.
127 * @default false
128 */
129 sqsManagedSseEnabled?: boolean;
130 /**
131 * Tags to apply to the queue. Merged with internal Alchemy tags.
132 */
133 tags?: Record<string, string>;
134} & (
135 | {
136 /** Standard queue mode. Must be known before precreation. */
137 fifo?: false;
138 contentBasedDeduplication?: undefined;
139 deduplicationScope?: undefined;
140 fifoThroughputLimit?: undefined;
141 }
142 | {
143 /** FIFO queue mode. Must be known before precreation. */
144 fifo: true;
145 /**
146 * Enables content-based deduplication for FIFO queues. Only valid when `fifo` is `true`.
147 * @default false
148 */
149 contentBasedDeduplication?: boolean;
150 /**
151 * Specifies whether message deduplication occurs at the message group or queue level.
152 * Valid values are `messageGroup` and `queue`. Only valid when `fifo` is `true`.
153 */
154 deduplicationScope?: "messageGroup" | "queue";
155 /**
156 * Specifies whether the FIFO queue throughput quota applies to the entire queue or per message group.
157 * Valid values are `perQueue` and `perMessageGroupId`. Only valid when `fifo` is `true`.
158 */
159 fifoThroughputLimit?: "perQueue" | "perMessageGroupId";
160 }
161);
162
163export interface Queue extends Resource<
164 "AWS.SQS.Queue",
165 QueueProps,
166 {
167 queueUrl: string;
168 queueName: QueueName;
169 queueArn: QueueArn;
170 },
171 {
172 policyStatements: PolicyStatement[];
173 },
174 Providers
175> {}
176
177class QueueStillExists extends Data.TaggedError("QueueStillExists")<{
178 readonly queueUrl: string;
179}> {}
180
181/**
182 * An Amazon SQS queue for reliable, decoupled message processing.
183 *
184 * `Queue` owns the lifecycle of a standard or FIFO SQS queue. A queue name
185 * is auto-generated from the app, stage, and logical ID unless you provide
186 * one explicitly. The `.fifo` suffix follows the `fifo` setting, including
187 * for explicit names. Changing queue mode replaces the physical queue and
188 * updates downstream references; existing messages are not migrated.
189 *
190 * `queueName` and `fifo` must be known before the queue is precreated; unresolved
191 * resource outputs in these identity properties fail with `UnresolvedQueueIdentity`.
192 * Mutable settings and policy bindings may reference other resource outputs,
193 * including the queue's own ARN.
194 *
195 * ### Creating Queues
196 * **Example:** Standard Queue
197 * ```typescript
198 * import * as SQS from "alchemy/AWS/SQS";
199 *
200 * const queue = yield* SQS.Queue("OrdersQueue");
201 * ```
202 *
203 * **Example:** FIFO Queue
204 * ```typescript
205 * const queue = yield* SQS.Queue("OrdersFifoQueue", {
206 * fifo: true,
207 * contentBasedDeduplication: true,
208 * });
209 * ```
210 *
211 * **Example:** Queue with Custom Settings
212 * ```typescript
213 * const queue = yield* SQS.Queue("ProcessingQueue", {
214 * visibilityTimeout: "2 minutes",
215 * messageRetentionPeriod: "1 day",
216 * receiveMessageWaitTime: "20 seconds",
217 * });
218 * ```
219 *
220 * ### Adopting an Existing Queue
221 * **Example:** Adopt a queue with output-valued settings
222 * ```typescript
223 * import * as Alchemy from "alchemy";
224 * import * as SQS from "alchemy/AWS/SQS";
225 *
226 * const audit = yield* SQS.Queue("Audit");
227 * const orders = yield* SQS.Queue("Orders", {
228 * queueName: "existing-orders",
229 * tags: { auditQueue: audit.queueArn },
230 * }).pipe(Alchemy.adopt(true));
231 * ```
232 * Explicit adoption preserves the queue identity and messages while reconciling
233 * settings, policies, and tags, even when mutable properties depend on other resources.
234 *
235 * ### Dead-Letter Queues
236 * **Example:** Route failures to a dead-letter queue
237 * ```typescript
238 * const dlq = yield* SQS.Queue("OrdersDLQ");
239 * const orders = yield* SQS.Queue("Orders", {
240 * redrivePolicy: {
241 * deadLetterTargetArn: dlq.queueArn,
242 * maxReceiveCount: 3,
243 * },
244 * });
245 * ```
246 *
247 * **Example:** Authorize source queues on the dead-letter queue
248 * ```typescript
249 * const dlq = yield* SQS.Queue("OrdersDLQ", {
250 * redriveAllowPolicy: {
251 * redrivePermission: "byQueue",
252 * sourceQueueArns: [orders.queueArn],
253 * },
254 * });
255 * ```
256 *
257 * ### Encryption
258 * **Example:** SSE-SQS (SQS-managed keys)
259 * ```typescript
260 * const queue = yield* SQS.Queue("SecureQueue", {
261 * sqsManagedSseEnabled: true,
262 * });
263 * ```
264 *
265 * **Example:** SSE-KMS (AWS-managed key)
266 * ```typescript
267 * const queue = yield* SQS.Queue("KmsQueue", {
268 * kmsMasterKeyId: "alias/aws/sqs",
269 * kmsDataKeyReusePeriod: "5 minutes",
270 * });
271 * ```
272 *
273 * ### Sending Messages
274 * Bind send operations in the init phase and use them in runtime
275 * handlers.
276 *
277 * **Example:** Send a message from a handler
278 * ```typescript
279 * // init
280 * const sendMessage = yield* SQS.SendMessage(queue);
281 *
282 * return {
283 * fetch: Effect.gen(function* () {
284 * // runtime
285 * yield* sendMessage({
286 * MessageBody: JSON.stringify({ orderId: "123" }),
287 * });
288 * return HttpServerResponse.text("Queued");
289 * }),
290 * };
291 * ```
292 *
293 * ### Event Sources
294 * Process messages from a queue using a Lambda event source mapping.
295 * Messages are automatically deleted after successful processing.
296 *
297 * **Example:** Process queue messages
298 * ```typescript
299 * // init
300 * yield* SQS.consumeQueueMessages(queue,
301 * Effect.fn(function* (message) {
302 * yield* Effect.log(`Received: ${message.body}`);
303 * }),
304 * );
305 * ```
306 *
307 * @resource
308 */
309export const Queue = Resource<Queue>("AWS.SQS.Queue");
310
311/**
312 * Raised when a `Queue` is configured with both `kmsMasterKeyId` (SSE-KMS)
313 * and `sqsManagedSseEnabled` (SSE-SQS). The two encryption modes are
314 * mutually exclusive.
315 */
316export class SqsEncryptionConflict extends Data.TaggedError(
317 "SqsEncryptionConflict",
318)<{ message: string }> {}
319
320/** Raised when a queue identity depends on an unresolved resource output. */
321export class UnresolvedQueueIdentity extends Data.TaggedError(
322 "UnresolvedQueueIdentity",
323)<{ message: string }> {}
324
325const validateEncryption = (props: QueueProps) =>
326 props.kmsMasterKeyId !== undefined && props.sqsManagedSseEnabled
327 ? Effect.fail(
328 new SqsEncryptionConflict({
329 message:
330 "kmsMasterKeyId (SSE-KMS) and sqsManagedSseEnabled (SSE-SQS) are mutually exclusive — set only one.",
331 }),
332 )
333 : Effect.void;
334
335export const QueueProvider = () =>
336 Provider.effect(
337 Queue,
338 Effect.gen(function* () {
339 const createQueueName = Effect.fn(function* (
340 id: string,
341 props: {
342 queueName?: string | undefined;
343 fifo?: boolean;
344 },
345 ) {
346 if (props.queueName) {
347 const baseName = props.queueName.endsWith(".fifo")
348 ? props.queueName.slice(0, -".fifo".length)
349 : props.queueName;
350 return props.fifo ? `${baseName}.fifo` : baseName;
351 }
352 const baseName = yield* createPhysicalName({
353 id,
354 maxLength: props.fifo ? 80 - ".fifo".length : 80,
355 });
356 return props.fifo ? `${baseName}.fifo` : baseName;
357 });
358 const buildPolicy = (
359 props: QueueProps,
360 bindings: ResourceBinding<Queue["Binding"]>[],
361 ): string | undefined => {
362 const bindingStatements = bindings.flatMap(
363 (p) => p.data.policyStatements,
364 );
365 let userStatements: any[] = [];
366 if (props.policy !== undefined) {
367 const doc =
368 typeof props.policy === "string"
369 ? JSON.parse(props.policy)
370 : props.policy;
371 const stmt = doc?.Statement;
372 userStatements = Array.isArray(stmt) ? stmt : stmt ? [stmt] : [];
373 }
374 const statements = [...userStatements, ...bindingStatements];
375 if (statements.length === 0) return undefined;
376 return JSON.stringify({
377 Version: "2012-10-17",
378 Statement: statements,
379 });
380 };
381 // Build the desired attribute map. Keys present here are reconciled
382 // against observed cloud state. An empty-string value explicitly
383 // CLEARS an attribute (SQS interprets `""` as "remove"); `undefined`
384 // means "leave alone" and is filtered out before diffing.
385 const createAttributes = (
386 props: QueueProps,
387 bindings: ResourceBinding<Queue["Binding"]>[],
388 ) => {
389 // Removable attributes always emit a key: the desired value when set,
390 // or "" to clear when the prop is absent. This lets the delta loop
391 // converge when a user removes redrive/policy/kms props.
392 const redrivePolicy = props.redrivePolicy
393 ? JSON.stringify({
394 deadLetterTargetArn: props.redrivePolicy.deadLetterTargetArn,
395 maxReceiveCount: props.redrivePolicy.maxReceiveCount,
396 })
397 : "";
398 const redriveAllowPolicy = props.redriveAllowPolicy
399 ? JSON.stringify({
400 redrivePermission: props.redriveAllowPolicy.redrivePermission,
401 ...(props.redriveAllowPolicy.sourceQueueArns
402 ? { sourceQueueArns: props.redriveAllowPolicy.sourceQueueArns }
403 : {}),
404 })
405 : "";
406 const policy = buildPolicy(props, bindings) ?? "";
407
408 const baseAttributes: Record<string, string | undefined> = {
409 DelaySeconds: toWireSeconds(props.delay)?.toString(),
410 MaximumMessageSize: props.maximumMessageSize?.toString(),
411 MessageRetentionPeriod: toWireSeconds(
412 props.messageRetentionPeriod,
413 )?.toString(),
414 ReceiveMessageWaitTimeSeconds: toWireSeconds(
415 props.receiveMessageWaitTime,
416 )?.toString(),
417 VisibilityTimeout: toWireSeconds(props.visibilityTimeout)?.toString(),
418 RedrivePolicy: redrivePolicy,
419 RedriveAllowPolicy: redriveAllowPolicy,
420 Policy: policy,
421 KmsMasterKeyId: props.kmsMasterKeyId,
422 KmsDataKeyReusePeriodSeconds: toWireSeconds(
423 props.kmsDataKeyReusePeriod,
424 )?.toString(),
425 SqsManagedSseEnabled:
426 props.sqsManagedSseEnabled === undefined
427 ? undefined
428 : props.sqsManagedSseEnabled
429 ? "true"
430 : "false",
431 };
432
433 if (props.fifo) {
434 return {
435 ...baseAttributes,
436 FifoQueue: "true",
437 FifoThroughputLimit: props.fifoThroughputLimit,
438 ContentBasedDeduplication: props.contentBasedDeduplication
439 ? "true"
440 : "false",
441 DeduplicationScope: props.deduplicationScope,
442 };
443 }
444
445 return baseAttributes;
446 };
447 const ensureQueueExists = Effect.fn(function* ({
448 id,
449 news = {},
450 output,
451 bindings = [],
452 }: {
453 id: string;
454 news: QueueProps;
455 output?: Queue["Attributes"];
456 bindings?: ResourceBinding<Queue["Binding"]>[];
457 }) {
458 yield* validateEncryption(news);
459 const { accountId, region } = yield* AWSEnvironment.current;
460 const queueName =
461 output?.queueName ?? (yield* createQueueName(id, news));
462 const queueArn =
463 `arn:aws:sqs:${region}:${accountId}:${queueName}` as const;
464 let queueUrl = yield* sqs.getQueueUrl({ QueueName: queueName }).pipe(
465 Effect.map((r) => r.QueueUrl),
466 Effect.catchTag("QueueDoesNotExist", () => Effect.succeed(undefined)),
467 );
468
469 if (queueUrl === undefined) {
470 const internalTags = yield* createInternalTags(id);
471 const attributes: Record<string, string> = {};
472 // Empty strings clear existing attributes but are invalid on create.
473 for (const [key, value] of Object.entries(
474 createAttributes(news, bindings),
475 )) {
476 if (value !== undefined && value !== "") attributes[key] = value;
477 }
478 queueUrl = yield* sqs
479 .createQueue({
480 QueueName: queueName,
481 Attributes: attributes,
482 tags: { ...news.tags, ...internalTags },
483 })
484 .pipe(
485 Effect.catchTag("QueueNameExists", () =>
486 sqs.getQueueUrl({ QueueName: queueName }),
487 ),
488 Effect.retry({
489 while: (error) =>
490 error._tag === "QueueDeletedRecently" ||
491 error._tag === "QueueDoesNotExist" ||
492 (news.redrivePolicy !== undefined &&
493 error._tag === "InvalidParameterValueException" &&
494 error.message?.includes(
495 "Dead letter target does not exist",
496 ) === true),
497 // Allow propagation beyond SQS's 60-second name reuse cooldown.
498 schedule: Schedule.spaced("7 seconds"),
499 times: 10,
500 }),
501 Effect.map((r) => r.QueueUrl!),
502 );
503 }
504
505 return { queueName, queueUrl, queueArn };
506 });
507 return Queue.Provider.of({
508 stables: ["queueName", "queueUrl", "queueArn"],
509 // Enumerate every queue in the ambient account/region. `listQueues`
510 // returns queue URLs (paginated), which we hydrate into the exact `read`
511 // shape. A queue can vanish between enumeration and hydration, so we
512 // tolerate the typed `QueueDoesNotExist` per item and drop it.
513 list: () =>
514 Effect.gen(function* () {
515 const { accountId, region } = yield* AWSEnvironment.current;
516 const urls = yield* sqs.listQueues.pages({}).pipe(
517 Stream.runCollect,
518 Effect.map((chunk) =>
519 Array.from(chunk).flatMap((page) => page.QueueUrls ?? []),
520 ),
521 );
522 const items = yield* Effect.forEach(
523 urls,
524 (queueUrl) =>
525 sqs
526 .getQueueAttributes({
527 QueueUrl: queueUrl,
528 AttributeNames: ["QueueArn"],
529 })
530 .pipe(
531 Effect.map((r) => {
532 const queueName =
533 r.Attributes?.QueueArn?.split(":").pop() ??
534 queueUrl.split("/").pop()!;
535 const queueArn =
536 `arn:aws:sqs:${region}:${accountId}:${queueName}` as const;
537 return { queueName, queueUrl, queueArn };
538 }),
539 Effect.catchTag("QueueDoesNotExist", () =>
540 Effect.succeed(undefined),
541 ),
542 ),
543 { concurrency: 10 },
544 );
545 return items.filter(
546 (item): item is Queue["Attributes"] => item !== undefined,
547 );
548 }),
549 read: Effect.fn(function* ({ id, olds, output }) {
550 const { accountId, region } = yield* AWSEnvironment.current;
551 const queueName =
552 output?.queueName ?? (yield* createQueueName(id, olds ?? {}));
553 const url = yield* sqs.getQueueUrl({ QueueName: queueName }).pipe(
554 Effect.map((r) => r.QueueUrl),
555 Effect.catchTag("QueueDoesNotExist", () =>
556 Effect.succeed(undefined),
557 ),
558 );
559 if (!url) return undefined;
560 const queueArn =
561 `arn:aws:sqs:${region}:${accountId}:${queueName}` as const;
562 const tagsResp = yield* sqs.listQueueTags({ QueueUrl: url }).pipe(
563 Effect.map((r) => r.Tags ?? {}),
564 Effect.catchTag("QueueDoesNotExist", () =>
565 Effect.succeed(undefined),
566 ),
567 );
568 if (tagsResp === undefined) return undefined;
569 const attrs = {
570 queueName,
571 queueUrl: url,
572 queueArn,
573 };
574 return (yield* hasAlchemyTags(id, tagsResp)) ? attrs : Unowned(attrs);
575 }),
576 diff: Effect.fn(function* ({ id, news = {}, olds = {}, output }) {
577 if (isOutput(news) || Effect.isEffect(news) || Config.isConfig(news))
578 return undefined;
579 const identity = { queueName: news.queueName, fifo: news.fifo };
580 if (!isResolved<Pick<QueueProps, "queueName" | "fifo">>(identity))
581 return undefined;
582 const encryption = {
583 kmsMasterKeyId: news.kmsMasterKeyId,
584 sqsManagedSseEnabled: news.sqsManagedSseEnabled,
585 };
586 if (
587 isResolved<
588 Pick<QueueProps, "kmsMasterKeyId" | "sqsManagedSseEnabled">
589 >(encryption)
590 )
591 yield* validateEncryption(encryption);
592 const oldFifo = olds.fifo ?? false;
593 const newFifo = identity.fifo ?? false;
594 if (oldFifo !== newFifo) {
595 return { action: "replace" } as const;
596 }
597 const oldQueueName =
598 output?.queueName ?? (yield* createQueueName(id, olds));
599 const newQueueName = yield* createQueueName(id, identity);
600 if (oldQueueName !== newQueueName) {
601 return { action: "replace" } as const;
602 }
603 // Return undefined to allow update function to be called for other attribute changes
604 }),
605 // Mutable properties and bindings may reference identities that do not exist yet.
606 precreate: Effect.fn(function* ({ id, fqn, news = {} }) {
607 const identity = { queueName: news.queueName, fifo: news.fifo };
608 if (!isResolved(identity)) {
609 return yield* Effect.fail(
610 new UnresolvedQueueIdentity({
611 message:
612 "Queue queueName and fifo must be known before precreation; use literal identity properties and put circular references in bindings.",
613 }),
614 );
615 }
616 if (
617 isResolved({
618 kmsMasterKeyId: news.kmsMasterKeyId,
619 sqsManagedSseEnabled: news.sqsManagedSseEnabled,
620 })
621 ) {
622 yield* validateEncryption(news);
623 }
624 const output = yield* ensureQueueExists({ id, news: identity });
625 // Unresolved mutable props can defer the engine's ownership probe.
626 const tags = yield* sqs
627 .listQueueTags({ QueueUrl: output.queueUrl })
628 .pipe(
629 Effect.retry({
630 while: (error) => error._tag === "QueueDoesNotExist",
631 schedule: Schedule.spaced("2 seconds"),
632 times: 10,
633 }),
634 );
635 if (!(yield* hasAlchemyTags(id, tags.Tags ?? {}))) {
636 const stack = yield* Stack;
637 const adopt =
638 stack.resources[fqn]?.Adopt ??
639 Option.getOrUndefined(yield* Effect.serviceOption(AdoptPolicy)) ??
640 (yield* AlchemyContext).adopt;
641 if (!adopt) {
642 return yield* new OwnedBySomeoneElse({
643 message: `Cannot adopt SQS queue '${output.queueName}' without explicit adoption.`,
644 resourceType: Queue.Type,
645 logicalId: id,
646 physicalName: output.queueName,
647 });
648 }
649 }
650 return output;
651 }),
652 reconcile: Effect.fn(function* ({
653 id,
654 news = {},
655 output,
656 session,
657 bindings,
658 }) {
659 const { queueName, queueUrl, queueArn } = yield* ensureQueueExists({
660 id,
661 news,
662 output,
663 bindings,
664 });
665 const desiredAttributes = createAttributes(news, bindings);
666 const internalTags = yield* createInternalTags(id);
667
668 // Sync attributes — diff observed cloud state against desired and
669 // apply only the delta. SQS returns all attribute values as strings,
670 // and `desiredAttributes` is already string-shaped, so equality
671 // comparison is direct.
672 // SQS is eventually consistent: a freshly-created queue can return
673 // `QueueDoesNotExist` from `getQueueAttributes` for a few seconds
674 // even after `createQueue` succeeded. Retry briefly so the
675 // reconciler converges instead of failing the first deploy.
676 const currentAttributes = yield* sqs
677 .getQueueAttributes({
678 QueueUrl: queueUrl,
679 AttributeNames: ["All"],
680 })
681 .pipe(
682 Effect.retry({
683 while: (e) => e._tag === "QueueDoesNotExist",
684 schedule: Schedule.spaced("2 seconds"),
685 times: 10,
686 }),
687 Effect.map((r) => r.Attributes ?? {}),
688 );
689
690 const attributeDelta: Record<string, string> = {};
691 for (const [key, value] of Object.entries(desiredAttributes)) {
692 // FifoQueue is immutable and only valid on CreateQueue.
693 if (value === undefined || key === "FifoQueue") continue;
694 const current =
695 currentAttributes[key as keyof typeof currentAttributes];
696 // Desired-to-clear ("") only needs an API call when the attribute
697 // is actually present; SQS rejects clearing an already-absent attr.
698 if (value === "" && (current === undefined || current === "")) {
699 continue;
700 }
701 if (current !== value) {
702 attributeDelta[key] = value;
703 }
704 }
705 if (Object.keys(attributeDelta).length > 0) {
706 yield* sqs
707 .setQueueAttributes({
708 QueueUrl: queueUrl,
709 Attributes: attributeDelta,
710 })
711 .pipe(
712 Effect.retry({
713 while: (e) =>
714 e._tag === "QueueDoesNotExist" ||
715 (attributeDelta.RedrivePolicy !== undefined &&
716 e._tag === "InvalidParameterValueException" &&
717 e.message?.includes(
718 "Dead letter target does not exist",
719 ) === true),
720 schedule: Schedule.spaced("2 seconds"),
721 times: 10,
722 }),
723 );
724 }
725
726 // Sync alchemy-owned tags. The `tags` parameter on `createQueue`
727 // only applies on first create, so on adoption (or after a queue
728 // was created without our tags) we fix them up here.
729 const observedTags = yield* sqs
730 .listQueueTags({ QueueUrl: queueUrl })
731 .pipe(
732 Effect.retry({
733 while: (e) => e._tag === "QueueDoesNotExist",
734 schedule: Schedule.spaced("2 seconds"),
735 times: 10,
736 }),
737 Effect.map((r) => r.Tags ?? {}),
738 );
739 // Merge user tags with internal Alchemy tags and diff against the
740 // OBSERVED cloud tags (not olds) so adoption converges. User tags
741 // can be removed, so we untag removed keys; internal tags are never
742 // user-removable so they survive.
743 const desiredTags: Record<string, string> = {
744 ...(news.tags ?? {}),
745 ...internalTags,
746 };
747 const currentTags: Record<string, string> = {};
748 for (const [key, value] of Object.entries(observedTags)) {
749 if (value !== undefined) currentTags[key] = value;
750 }
751 const { upsert, removed } = diffTags(currentTags, desiredTags);
752 if (upsert.length > 0) {
753 yield* sqs
754 .tagQueue({
755 QueueUrl: queueUrl,
756 Tags: Object.fromEntries(upsert.map((t) => [t.Key, t.Value])),
757 })
758 .pipe(
759 Effect.retry({
760 while: (e) => e._tag === "QueueDoesNotExist",
761 schedule: Schedule.spaced("2 seconds"),
762 times: 10,
763 }),
764 );
765 }
766 if (removed.length > 0) {
767 yield* sqs
768 .untagQueue({ QueueUrl: queueUrl, TagKeys: removed })
769 .pipe(
770 Effect.retry({
771 while: (e) => e._tag === "QueueDoesNotExist",
772 schedule: Schedule.spaced("2 seconds"),
773 times: 10,
774 }),
775 );
776 }
777
778 yield* session.note(queueUrl);
779 return {
780 queueName,
781 queueUrl,
782 queueArn,
783 };
784 }),
785 delete: Effect.fn(function* (input) {
786 const queueUrl = input.output.queueUrl;
787 yield* sqs
788 .deleteQueue({
789 QueueUrl: queueUrl,
790 })
791 .pipe(
792 Effect.retry({
793 while: (error) => error._tag === "RequestThrottled",
794 schedule: Schedule.exponential("500 millis"),
795 times: 6,
796 }),
797 Effect.catchTag("QueueDoesNotExist", () => Effect.void),
798 );
799
800 // DeleteQueue is asynchronous and SQS can keep serving the queue
801 // for up to 60 seconds. GetQueueAttributes can report not-found
802 // before ListQueues stops returning the URL, and nuke discovers
803 // queues through ListQueues. Require both control-plane views to
804 // agree before state is removed or a same-name replacement starts.
805 // Typed read throttles mean "absence not confirmed yet" and retry;
806 // re-entering delete remains safe because QueueDoesNotExist is OK.
807 const queueName = queueUrl.slice(queueUrl.lastIndexOf("/") + 1);
808 yield* Effect.gen(function* () {
809 const attributesAbsent = yield* sqs
810 .getQueueAttributes({
811 QueueUrl: queueUrl,
812 AttributeNames: ["QueueArn"],
813 })
814 .pipe(
815 Effect.as(false),
816 Effect.catchTag("QueueDoesNotExist", () =>
817 Effect.succeed(true),
818 ),
819 Effect.catchTag("RequestThrottled", () =>
820 Effect.succeed(false),
821 ),
822 );
823 const absentFromList = yield* sqs
824 .listQueues({ QueueNamePrefix: queueName })
825 .pipe(
826 Effect.map(
827 (result) => !(result.QueueUrls ?? []).includes(queueUrl),
828 ),
829 Effect.catchTag("RequestThrottled", () =>
830 Effect.succeed(false),
831 ),
832 );
833 if (!attributesAbsent || !absentFromList) {
834 return yield* Effect.fail(new QueueStillExists({ queueUrl }));
835 }
836 }).pipe(
837 Effect.retry({
838 while: (error) => error._tag === "QueueStillExists",
839 schedule: Schedule.spaced("6 seconds"),
840 times: 10,
841 }),
842 );
843 }),
844 });
845 }),
846 );