fix(notifications): merge SNS/SQS policies for shared topics and queues

AWS::SNS::TopicPolicy and AWS::SQS::QueuePolicy are full replacements:
when two separate CloudFormation resources target the same topic/queue,
the second overwrites the first. When multiple state machines share a
notification topic or queue (fan-in), the second machine's policy
destroys the first machine's permissions — its notification rules stop
working silently.

Fix: after compiling all notification resources, group SNS topic
policies by topic and SQS queue policies by queue. Same-target policies
are merged into a single resource with all statements combined, so
CloudFormation sets the policy exactly once with all principals
authorised.

Adds three unit tests covering:
- Two state machines sharing the same SNS topic
- Two state machines sharing the same SQS queue
- One state machine with the same topic on multiple statuses

Updates the notifications integration fixture to include a second state
machine sharing the same SNS topic and SQS queue as the first, matching
the fan-in scenario described in #275.

Closes #275

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: VirtueMe

fixtures/notifications/serverless.yml

@@ -29,3 +29,22 @@ stepFunctions:
         SUCCEEDED:
           - sqs:
               Fn::GetAtt: [NotificationQueue, Arn]
+
+    # Second state machine sharing the same SNS topic and SQS queue (fan-in).
+    # Reproduces issue #275: without the fix, this machine's TopicPolicy and
+    # QueuePolicy overwrite the first machine's, leaving it unable to publish.
+    notificationMachine2:
+      name: integration-notifications-2-${opt:stage, 'test'}
+      definition:
+        StartAt: PassThrough
+        States:
+          PassThrough:
+            Type: Pass
+            End: true
+      notifications:
+        FAILED:
+          - sns:
+              Ref: NotificationTopic
+        SUCCEEDED:
+          - sqs:
+              Fn::GetAtt: [NotificationQueue, Arn]

lib/deploy/stepFunctions/compileNotifications.js

@@ -387,6 +387,54 @@ function validateConfig(serverless, stateMachineName, stateMachineObj, notificat
   return true;
 }
 
+// Merge AWS::SNS::TopicPolicy and AWS::SQS::QueuePolicy resources that target
+// the same topic/queue into a single resource with combined statements.
+//
+// CloudFormation treats these resource types as full policy replacements: if two
+// separate resources target the same topic/queue, the second one overwrites the
+// first. When multiple state machines share a notification topic/queue (fan-in),
+// this destroys the first machine's permissions. Merging into one resource with
+// all statements ensures CloudFormation sets the policy exactly once, correctly.
+function mergeResourcePolicies(resourcePairs) {
+  const snsMap = new Map(); // JSON-serialised Topics array → { logicalId, resource }
+  const sqsMap = new Map(); // JSON-serialised Queues array → { logicalId, resource }
+  const result = {};
+
+  for (const [logicalId, resource] of resourcePairs) {
+    if (resource.Type === 'AWS::SNS::TopicPolicy') {
+      const key = JSON.stringify(resource.Properties.Topics);
+      if (snsMap.has(key)) {
+        const incoming = [].concat(resource.Properties.PolicyDocument.Statement);
+        snsMap.get(key).resource.Properties.PolicyDocument.Statement.push(...incoming);
+      } else {
+        const merged = _.cloneDeep(resource);
+        merged.Properties.PolicyDocument.Statement = [].concat(
+          merged.Properties.PolicyDocument.Statement,
+        );
+        snsMap.set(key, { logicalId, resource: merged });
+        result[logicalId] = merged;
+      }
+    } else if (resource.Type === 'AWS::SQS::QueuePolicy') {
+      const key = JSON.stringify(resource.Properties.Queues);
+      if (sqsMap.has(key)) {
+        const incoming = [].concat(resource.Properties.PolicyDocument.Statement);
+        sqsMap.get(key).resource.Properties.PolicyDocument.Statement.push(...incoming);
+      } else {
+        const merged = _.cloneDeep(resource);
+        merged.Properties.PolicyDocument.Statement = [].concat(
+          merged.Properties.PolicyDocument.Statement,
+        );
+        sqsMap.set(key, { logicalId, resource: merged });
+        result[logicalId] = merged;
+      }
+    } else {
+      result[logicalId] = resource;
+    }
+  }
+
+  return result;
+}
+
 module.exports = {
   compileNotifications() {
     logger.config(this.serverless, this.v3Api);
@@ -412,7 +460,7 @@ module.exports = {
 
       return Array.from(resourcesIterator);
     });
-    const newResources = _.fromPairs(newResourcePairs);
+    const newResources = mergeResourcePolicies(newResourcePairs);
 
     _.merge(
       this.serverless.service.provider.compiledCloudFormationTemplate.Resources,

lib/deploy/stepFunctions/compileNotifications.test.js

@@ -623,4 +623,79 @@ describe('#compileNotifications', () => {
 
     expect(resources.Beta1NotificationsIamRole.Properties.Path).to.equal('/teamA/');
   });
+
+  it('should produce a single SNS topic policy when multiple state machines share the same topic', () => {
+    // Reproduces issue #275: each state machine generates its own AWS::SNS::TopicPolicy
+    // for the shared topic. CloudFormation applies them sequentially — the second
+    // overwrites the first, leaving the first machine's notifications unauthorised.
+    // The fix merges policies for the same topic into a single resource with combined
+    // statements so CloudFormation only sets the policy once, correctly.
+    serverless.service.stepFunctions = {
+      stateMachines: {
+        machine1: genStateMachineWithTargets('Machine1', [{ sns: 'arn:aws:sns:us-east-1:123:shared-topic' }]),
+        machine2: genStateMachineWithTargets('Machine2', [{ sns: 'arn:aws:sns:us-east-1:123:shared-topic' }]),
+      },
+    };
+
+    serverlessStepFunctions.compileNotifications();
+    const resources = serverlessStepFunctions.serverless.service
+      .provider.compiledCloudFormationTemplate.Resources;
+
+    const snsPolicies = _.values(resources).filter((r) => r.Type === 'AWS::SNS::TopicPolicy');
+    const policiesForSharedTopic = snsPolicies.filter((p) => _.isEqual(
+      p.Properties.Topics[0],
+      'arn:aws:sns:us-east-1:123:shared-topic',
+    ));
+
+    expect(policiesForSharedTopic).to.have.lengthOf(1);
+
+    // Both machines × 5 statuses = 10 statements in the merged policy
+    const statements = [].concat(policiesForSharedTopic[0].Properties.PolicyDocument.Statement);
+    expect(statements.length).to.equal(10);
+  });
+
+  it('should produce a single SQS queue policy when multiple state machines share the same queue', () => {
+    serverless.service.stepFunctions = {
+      stateMachines: {
+        machine1: genStateMachineWithTargets('Machine1', [{ sqs: 'arn:aws:sqs:us-east-1:123:shared-queue' }]),
+        machine2: genStateMachineWithTargets('Machine2', [{ sqs: 'arn:aws:sqs:us-east-1:123:shared-queue' }]),
+      },
+    };
+
+    serverlessStepFunctions.compileNotifications();
+    const resources = serverlessStepFunctions.serverless.service
+      .provider.compiledCloudFormationTemplate.Resources;
+
+    const sqsPolicies = _.values(resources).filter((r) => r.Type === 'AWS::SQS::QueuePolicy');
+    expect(sqsPolicies).to.have.lengthOf(1);
+
+    const statements = [].concat(sqsPolicies[0].Properties.PolicyDocument.Statement);
+    expect(statements.length).to.equal(10);
+  });
+
+  it('should merge SNS policies when same topic is used for multiple statuses in one state machine', () => {
+    serverless.service.stepFunctions = {
+      stateMachines: {
+        machine1: {
+          id: 'Machine1',
+          name: 'Machine1',
+          definition: { StartAt: 'A', States: { A: { Type: 'Pass', End: true } } },
+          notifications: {
+            SUCCEEDED: [{ sns: 'arn:aws:sns:us-east-1:123:shared-topic' }],
+            FAILED: [{ sns: 'arn:aws:sns:us-east-1:123:shared-topic' }],
+          },
+        },
+      },
+    };
+
+    serverlessStepFunctions.compileNotifications();
+    const resources = serverlessStepFunctions.serverless.service
+      .provider.compiledCloudFormationTemplate.Resources;
+
+    const snsPolicies = _.values(resources).filter((r) => r.Type === 'AWS::SNS::TopicPolicy');
+    expect(snsPolicies).to.have.lengthOf(1);
+
+    const statements = [].concat(snsPolicies[0].Properties.PolicyDocument.Statement);
+    expect(statements.length).to.equal(2);
+  });
 });