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