SNS Tag update

docker-compose.yml

@@ -15,7 +15,7 @@ services:
       - '127.0.0.1:4566:4566' # LocalStack Gateway
       - '127.0.0.1:4510-4559:4510-4559' # external services port range
     environment:
-      - SERVICES=sns,sqs,s3
+      - SERVICES=sns,sqs,s3,sts
       - DEBUG=0
       - DATA_DIR=${DATA_DIR-}
       - DOCKER_HOST=unix:///var/run/docker.sock

packages/sns/docker-compose.yml

@@ -7,7 +7,7 @@ services:
       - '127.0.0.1:4566:4566' # LocalStack Gateway
       - '127.0.0.1:4510-4559:4510-4559' # external services port range
     environment:
-      - SERVICES=sns,sqs,s3
+      - SERVICES=sns,sqs,s3,sts
       - DEBUG=0
       - DATA_DIR=${DATA_DIR-}
       - DOCKER_HOST=unix:///var/run/docker.sock

packages/sns/lib/sns/AbstractSnsService.ts

@@ -31,6 +31,7 @@ export type SNSTopicConfig = {
 export type ExtraSNSCreationParams = {
   queueUrlsWithSubscribePermissionsPrefix?: string | readonly string[]
   allowedSourceOwner?: string
+  forceTagUpdate?: boolean
 }
 
 export type SNSCreationConfig = {

packages/sns/lib/utils/snsInitter.ts

@@ -69,6 +69,7 @@ export async function initSnsSqs(
         allowedSourceOwner: creationConfig.allowedSourceOwner,
         topicArnsWithPublishPermissionsPrefix: creationConfig.topicArnsWithPublishPermissionsPrefix,
         logger: extraParams?.logger,
+        forceTagUpdate: creationConfig.forceTagUpdate,
       },
     )
     if (!subscriptionArn) {
@@ -229,6 +230,7 @@ export async function initSns(
   const topicArn = await assertTopic(snsClient, creationConfig.topic!, {
     queueUrlsWithSubscribePermissionsPrefix: creationConfig.queueUrlsWithSubscribePermissionsPrefix,
     allowedSourceOwner: creationConfig.allowedSourceOwner,
+    forceTagUpdate: creationConfig.forceTagUpdate,
   })
   return {
     topicArn,

packages/sns/lib/utils/snsSubscriber.ts

@@ -55,6 +55,7 @@ export async function subscribeToTopic(
   const { queueUrl, queueArn } = await assertQueue(sqsClient, queueConfiguration, {
     topicArnsWithPublishPermissionsPrefix: extraParams?.topicArnsWithPublishPermissionsPrefix,
     updateAttributesIfExists: extraParams?.updateAttributesIfExists,
+    forceTagUpdate: extraParams?.forceTagUpdate,
   })
 
   const subscribeCommand = new SubscribeCommand({

packages/sns/lib/utils/snsUtils.ts

@@ -1,6 +1,7 @@
 import {
   type CreateTopicCommandInput,
   type SNSClient,
+  TagResourceCommand,
   paginateListTopics,
 } from '@aws-sdk/client-sns'
 import {
@@ -12,11 +13,12 @@ import {
   SetTopicAttributesCommand,
   UnsubscribeCommand,
 } from '@aws-sdk/client-sns'
-import type { Either } from '@lokalise/node-core'
+import { type Either, isError } from '@lokalise/node-core'
 import { calculateOutgoingMessageSize as sqsCalculateOutgoingMessageSize } from '@message-queue-toolkit/sqs'
 
 import type { ExtraSNSCreationParams } from '../sns/AbstractSnsService'
 
+import { GetCallerIdentityCommand, STSClient } from '@aws-sdk/client-sts'
 import { generateTopicSubscriptionPolicy } from './snsAttributeUtils'
 
 type AttributesResult = {
@@ -82,13 +84,19 @@ export async function assertTopic(
   topicOptions: CreateTopicCommandInput,
   extraParams?: ExtraSNSCreationParams,
 ) {
-  const command = new CreateTopicCommand(topicOptions)
-  const response = await snsClient.send(command)
-
-  if (!response.TopicArn) {
-    throw new Error('No topic arn in response')
+  let topicArn: string
+  try {
+    const command = new CreateTopicCommand(topicOptions)
+    const response = await snsClient.send(command)
+    if (!response.TopicArn) throw new Error('No topic arn in response')
+    topicArn = response.TopicArn
+  } catch (err) {
+    // We only manually build ARN in case of tag update
+    if (!extraParams?.forceTagUpdate) throw err
+    // To build ARN we need topic name and error should be "topic already exist with different tags"
+    if (!topicOptions.Name || !isTopicAlreadyExistWithDifferentTagsError(err)) throw err
+    topicArn = await buildTopicArn(snsClient, topicOptions.Name)
   }
-  const topicArn = response.TopicArn
 
   if (extraParams?.queueUrlsWithSubscribePermissionsPrefix || extraParams?.allowedSourceOwner) {
     const setTopicAttributesCommand = new SetTopicAttributesCommand({
@@ -102,6 +110,13 @@ export async function assertTopic(
     })
     await snsClient.send(setTopicAttributesCommand)
   }
+  if (extraParams?.forceTagUpdate) {
+    const tagTopicCommand = new TagResourceCommand({
+      ResourceArn: topicArn,
+      Tags: topicOptions.Tags,
+    })
+    await snsClient.send(tagTopicCommand)
+  }
 
   return topicArn
 }
@@ -178,3 +193,35 @@ export async function getTopicArnByName(snsClient: SNSClient, topicName?: string
  */
 export const calculateOutgoingMessageSize = (message: unknown) =>
   sqsCalculateOutgoingMessageSize(message)
+
+const isTopicAlreadyExistWithDifferentTagsError = (error: unknown) =>
+  !!error &&
+  isError(error) &&
+  'Error' in error &&
+  !!error.Error &&
+  typeof error.Error === 'object' &&
+  'Code' in error.Error &&
+  'Message' in error.Error &&
+  typeof error.Error.Message === 'string' &&
+  error.Error.Code === 'InvalidParameter' &&
+  error.Error.Message.includes('already exists with different tags')
+
+/**
+ * Manually builds the ARN of a topic based on the current AWS account and the topic name.
+ * It follows the following pattern: arn:aws:sns:<region>:<account-id>:<topic-name>
+ * Doc -> https://docs.aws.amazon.com/IAM/latest/UserGuide/reference-arns.html
+ */
+const buildTopicArn = async (client: SNSClient, topicName: string) => {
+  const region =
+    typeof client.config.region === 'string' ? client.config.region : await client.config.region()
+
+  const stsClient = new STSClient({
+    endpoint: client.config.endpoint,
+    region,
+    credentials: client.config.credentials,
+    endpointProvider: client.config.endpointProvider,
+  })
+  const identityResponse = await stsClient.send(new GetCallerIdentityCommand({}))
+
+  return `arn:aws:sns:${region}:${identityResponse.Account}:${topicName}`
+}

packages/sns/package.json

@@ -32,6 +32,7 @@
   "peerDependencies": {
     "@aws-sdk/client-sns": "^3.632.0",
     "@aws-sdk/client-sqs": "^3.632.0",
+    "@aws-sdk/client-sts": "^3.632.0",
     "@message-queue-toolkit/core": ">=15.0.0",
     "@message-queue-toolkit/schemas": ">=2.0.0",
     "@message-queue-toolkit/sqs": "^17.0.0"
@@ -40,11 +41,11 @@
     "@aws-sdk/client-s3": "^3.670.0",
     "@aws-sdk/client-sns": "^3.670.0",
     "@aws-sdk/client-sqs": "^3.670.0",
+    "@biomejs/biome": "1.9.3",
+    "@kibertoad/biome-config": "^1.2.1",
     "@message-queue-toolkit/core": "*",
     "@message-queue-toolkit/s3-payload-store": "*",
     "@message-queue-toolkit/sqs": "*",
-    "@biomejs/biome": "1.9.3",
-    "@kibertoad/biome-config": "^1.2.1",
     "@types/node": "^22.7.5",
     "@vitest/coverage-v8": "^2.1.2",
     "awilix": "^12.0.1",