fix(aws-sqs): precreate self-referential queue identity

packages/alchemy/src/AWS/SQS/Queue.ts

@@ -20,6 +20,18 @@ export type QueueName = string;
 export type QueueArn = `arn:aws:sqs:${RegionID}:${AccountID}:${QueueName}`;
 export type QueueUrl = string;
 
+export class QueuePrecreateConflict extends Data.TaggedError(
+  "QueuePrecreateConflict",
+)<{
+  readonly id: string;
+  readonly queueName: string;
+  readonly reason: string;
+}> {
+  override get message() {
+    return `Cannot pre-create SQS queue '${this.id}' as '${this.queueName}': ${this.reason}`;
+  }
+}
+
 export type QueueProps = {
   /**
    * Name of the queue.
@@ -486,6 +498,108 @@ export const QueueProvider = () =>
           }
           // Return undefined to allow update function to be called for other attribute changes
         }),
+        precreate: Effect.fn(function* ({ id, news = {}, session }) {
+          // Queue policies commonly name the queue's own ARN. That makes the
+          // binding graph self-referential on a greenfield deploy: the policy
+          // cannot resolve until the queue has an identity, while reconcile
+          // cannot receive the resolved policy until that identity exists.
+          //
+          // Create only the stable physical shell here. Reconcile remains the
+          // single authority for mutable attributes, user tags, and binding-
+          // contributed policy once every dependency can be evaluated.
+          const identity = {
+            queueName: news.queueName,
+            fifo: news.fifo,
+          };
+          if (!isResolved(identity)) {
+            return yield* Effect.die(
+              new Error(
+                `SQS Queue '${id}' cannot be pre-created because queueName or fifo depends on an unresolved resource output`,
+              ),
+            );
+          }
+
+          const { accountId, region } = yield* AWSEnvironment.current;
+          const queueName = yield* createQueueName(id, identity);
+          const queueArn =
+            `arn:aws:sqs:${region}:${accountId}:${queueName}` as const;
+          const internalTags = yield* createInternalTags(id);
+          let queueUrl = yield* sqs.getQueueUrl({ QueueName: queueName }).pipe(
+            Effect.map((result) => result.QueueUrl),
+            Effect.catchTag("QueueDoesNotExist", () =>
+              Effect.succeed(undefined),
+            ),
+          );
+          let requiresOwnershipCheck = queueUrl !== undefined;
+
+          if (queueUrl === undefined) {
+            const observed = yield* sqs
+              .createQueue({
+                QueueName: queueName,
+                ...(identity.fifo ? { Attributes: { FifoQueue: "true" } } : {}),
+                tags: internalTags,
+              })
+              .pipe(
+                Effect.retry({
+                  while: (error) => error._tag === "QueueDeletedRecently",
+                  schedule: Schedule.max([
+                    Schedule.fixed(1000),
+                    Schedule.recurs(30),
+                  ]).pipe(
+                    Schedule.tap(({ attempt }) =>
+                      session.note(
+                        `Queue was deleted recently, retrying... ${attempt}s`,
+                      ),
+                    ),
+                  ),
+                }),
+                Effect.map((result) => ({
+                  queueUrl: result.QueueUrl!,
+                  raced: false,
+                })),
+                Effect.catchTag("QueueNameExists", () =>
+                  sqs.getQueueUrl({ QueueName: queueName }).pipe(
+                    Effect.map((result) => ({
+                      queueUrl: result.QueueUrl!,
+                      raced: true,
+                    })),
+                  ),
+                ),
+              );
+            queueUrl = observed.queueUrl;
+            requiresOwnershipCheck = observed.raced;
+          }
+
+          if (requiresOwnershipCheck) {
+            const [tags, attributes] = yield* Effect.all([
+              sqs
+                .listQueueTags({ QueueUrl: queueUrl })
+                .pipe(Effect.map((result) => result.Tags ?? {})),
+              sqs.getQueueAttributes({
+                QueueUrl: queueUrl,
+                AttributeNames: ["FifoQueue"],
+              }),
+            ]);
+            if (!(yield* hasAlchemyTags(id, tags))) {
+              return yield* new QueuePrecreateConflict({
+                id,
+                queueName,
+                reason:
+                  "an existing queue with that name is not owned by this stack",
+              });
+            }
+            const observedFifo = attributes.Attributes?.FifoQueue === "true";
+            if (observedFifo !== (identity.fifo ?? false)) {
+              return yield* new QueuePrecreateConflict({
+                id,
+                queueName,
+                reason: `the existing queue is ${observedFifo ? "FIFO" : "standard"}, but the desired queue is ${identity.fifo ? "FIFO" : "standard"}`,
+              });
+            }
+          }
+
+          return { queueName, queueUrl, queueArn };
+        }),
         reconcile: Effect.fn(function* ({
           id,
           news = {},

packages/alchemy/test/AWS/SQS/Queue.test.ts

@@ -54,6 +54,48 @@ provider("create and delete queue with default props", (stack) =>
   }),
 );
 
+provider(
+  "fresh create resolves a queue policy that references its own ARN",
+  (stack) =>
+    Effect.gen(function* () {
+      yield* stack.destroy();
+
+      const queue = yield* stack.deploy(
+        Effect.gen(function* () {
+          const queue = yield* Queue("SelfPolicyQueue");
+          yield* queue.bind`Allow(${queue}, AWS.SQS.SendMessage(${queue}))`({
+            policyStatements: [
+              {
+                Effect: "Allow",
+                Principal: { Service: "sns.amazonaws.com" },
+                Action: ["sqs:SendMessage"],
+                Resource: queue.queueArn,
+              },
+            ],
+          });
+          return queue;
+        }),
+      );
+
+      const attributes = yield* SQS.getQueueAttributes({
+        QueueUrl: queue.queueUrl,
+        AttributeNames: ["Policy"],
+      });
+      const policy = JSON.parse(attributes.Attributes?.Policy ?? "null") as {
+        readonly Statement?: ReadonlyArray<{ readonly Resource?: string }>;
+      } | null;
+
+      expect(
+        policy?.Statement?.some(
+          (statement) => statement.Resource === queue.queueArn,
+        ),
+      ).toBe(true);
+
+      yield* stack.destroy();
+      yield* assertQueueDeleted(queue.queueUrl);
+    }),
+);
+
 provider("create, update, delete standard queue", (stack) =>
   Effect.gen(function* () {
     yield* stack.destroy();