feat: fixed and made notifications work

Author: theburningmonk

lib/deploy/stepFunctions/compileNotifications.js

@@ -33,7 +33,23 @@ function randomTargetId(stateMachineName, status) {
   return `${stateMachineName}-${status}-${suffix}`;
 }
 
-function compileTarget(stateMachineName, status, targetObj) {
+function randomLogicalId(prefix) {
+  const suffix = chance.string({
+    length: 5,
+    pool: 'ABCDEFGHIJKLMNOPQRSTUFWXYZ',
+  });
+  return `${prefix}${suffix}`;
+}
+
+function randomPolicyName(status, targetType) {
+  const suffix = chance.string({
+    length: 5,
+    pool: 'abcdefghijklmnopqrstufwxyzABCDEFGHIJKLMNOPQRSTUFWXYZ',
+  });
+  return `${status}-${targetType}-${suffix}`;
+}
+
+function compileTarget(stateMachineName, status, targetObj, iamRoleLogicalId) {
   // SQS and Kinesis are special cases as they can have additional props
   if (_.has(targetObj, 'sqs.arn')) {
     return {
@@ -51,7 +67,16 @@ function compileTarget(stateMachineName, status, targetObj) {
         PartitionKeyPath: targetObj.kinesis.partitionKeyPath,
       },
     };
+  } else if (_.has(targetObj, 'stepFunctions')) {
+    return {
+      Arn: targetObj.stepFunctions,
+      Id: randomTargetId(stateMachineName, status),
+      RoleArn: {
+        'Fn::GetAtt': [iamRoleLogicalId, 'Arn'],
+      },
+    };
   }
+
   const targetType = supportedTargets.find(t => _.has(targetObj, t));
   const arn = _.get(targetObj, targetType);
   return {
@@ -60,24 +85,110 @@ function compileTarget(stateMachineName, status, targetObj) {
   };
 }
 
-function compileIamPermission(targetObj) {
-  const targetType = supportedTargets.find(t => _.has(targetObj, t));
-  const action = targetPermissions[targetType];
+function compileSnsPolicy(status, snsTarget) {
+  return {
+    Type: 'AWS::SNS::TopicPolicy',
+    Properties: {
+      PolicyDocument: {
+        Version: '2012-10-17',
+        Statement: {
+          Sid: randomPolicyName(status, 'sns'),
+          Principal: {
+            Service: 'events.amazonaws.com',
+          },
+          Effect: 'Allow',
+          Action: 'sns:Publish',
+          Resource: snsTarget,
+        },
+      },
+      Topics: [snsTarget],
+    },
+  };
+}
 
-  // SQS and Kinesis are special cases as they can have additional props
-  if (_.has(targetObj, 'sqs.arn')) {
+function convertToQueueUrl(sqsArn) {
+  if (_.isString(sqsArn)) {
+    const segments = sqsArn.split(':');
+    const queueName = _.last(segments);
     return {
-      action,
-      resource: _.get(targetObj, 'sqs.arn'),
+      'Fn::Sub': [
+        'https://sqs.${AWS::Region}.amazonaws.com/${AWS::AccountId}/${QueueName}',
+        { QueueName: queueName },
+      ],
     };
-  } else if (_.has(targetObj, 'kinesis.arn')) {
+  } else if (sqsArn['Fn::GetAtt']) {
+    const logicalId = sqsArn['Fn::GetAtt'][0];
+    return { Ref: logicalId };
+  }
+  throw new Error(
+    `Unable to convert SQS ARN [${sqsArn}] to SQS Url. ` +
+    'This is required for setting up Step Functions notifications to SQS. ' +
+    'Try using Fn::GetAtt when setting the SQS arn.');
+}
+
+function compileSqsPolicy(status, sqsTarget) {
+  return {
+    Type: 'AWS::SQS::QueuePolicy',
+    Properties: {
+      PolicyDocument: {
+        Version: '2012-10-17',
+        Statement: {
+          Sid: randomPolicyName(status, 'sqs'),
+          Principal: {
+            Service: 'events.amazonaws.com',
+          },
+          Effect: 'Allow',
+          Action: 'sqs:SendMessage',
+          Resource: sqsTarget,
+        },
+      },
+      Queues: [convertToQueueUrl(sqsTarget)],
+    },
+  };
+}
+
+function compileLambdaPermission(lambdaTarget) {
+  return {
+    Type: 'AWS::Lambda::Permission',
+    Properties: {
+      Action: 'lambda:InvokeFunction',
+      FunctionName: lambdaTarget,
+      Principal: 'events.amazonaws.com',
+    },
+  };
+}
+
+function compilePermissionForTarget(status, targetObj) {
+  if (targetObj.sns) {
+    return {
+      type: 'policy',
+      resource: compileSnsPolicy(status, targetObj.sns),
+    };
+  } else if (targetObj.sqs) {
+    const arn = _.get(targetObj, 'sqs.arn', targetObj.sqs);
+    return {
+      type: 'policy',
+      resource: compileSqsPolicy(status, arn),
+    };
+  } else if (targetObj.kinesis) {
+    const arn = _.get(targetObj, 'kinesis.arn', targetObj.kinesis);
+    return {
+      type: 'iam',
+      action: 'kinesis:PutRecord',
+      resource: arn,
+    };
+  } else if (targetObj.lambda) {
     return {
-      action,
-      resource: _.get(targetObj, 'kinesis.arn'),
+      type: 'policy',
+      resource: compileLambdaPermission(targetObj.lambda),
     };
   }
 
+  const targetType = supportedTargets.find(t => _.has(targetObj, t));
+  const action = targetPermissions[targetType];
+
   return {
+    type: 'iam',
     action,
     resource: targetObj[targetType],
   };
@@ -96,42 +207,74 @@ function bootstrapIamRole() {
           },
         },
       },
-      Policies: [
-        {
-          PolicyName: 'root',
-          PolicyDocument: {
-            Version: '2012-10-17',
-            Statement: [],
-          },
-        },
-      ],
+      Policies: [],
     },
   };
-  const addPermission = (action, resource) => {
-    iamRole.Properties.Policies[0].PolicyDocument.Statement.push({
-      Effect: 'Allow',
-      Action: action,
-      Resource: resource,
+  const addPolicy = (name, action, resource) => {
+    iamRole.Properties.Policies.push({
+      PolicyName: name,
+      PolicyDocument: {
+        Version: '2012-10-17',
+        Statement: [{
+          Effect: 'Allow',
+          Action: action,
+          Resource: resource,
+        }],
+      },
     });
   };
 
-  return { iamRole, addPermission };
+  return { iamRole, addPolicy };
+}
+
+function* compilePermissionResources(stateMachineLogicalId, iamRoleLogicalId, targets) {
+  const { iamRole, addPolicy } = bootstrapIamRole();
+
+  for (const { status, target } of targets) {
+    const perm = compilePermissionForTarget(status, target);
+    if (perm.type === 'iam') {
+      const targetType = _.keys(target)[0];
+      addPolicy(
+        randomPolicyName(status, targetType),
+        perm.action,
+        perm.resource);
+    } else if (perm.type === 'policy') {
+      yield {
+        logicalId: randomLogicalId(`${stateMachineLogicalId}ResourcePolicy`),
+        resource: perm.resource,
+      };
+    }
+  }
+
+  if (!_.isEmpty(iamRole.Properties.Policies)) {
+    yield {
+      logicalId: iamRoleLogicalId,
+      resource: iamRole,
+    };
+  }
 }
 
 function* compileResources(stateMachineLogicalId, stateMachineName, notificationsObj) {
   const iamRoleLogicalId = `${stateMachineLogicalId}NotificationsIamRole`;
-  const { iamRole, addPermission } = bootstrapIamRole();
+  const allTargets = _.flatMap(executionStatuses, status =>
+    _.get(notificationsObj, status, []).map(target => ({ status, target })));
+  const permissions = compilePermissionResources(
+    stateMachineLogicalId, iamRoleLogicalId, allTargets);
+  const permissionResources = Array.from(permissions);
+  for (const { logicalId, resource } of permissionResources) {
+    yield [logicalId, resource];
+  }
+
+  const needRoleArn = permissionResources.some(({ logicalId }) => logicalId === iamRoleLogicalId);
+  const roleArn = needRoleArn
+    ? { 'Fn::GetAtt': [iamRoleLogicalId, 'Arn'] }
+    : undefined;
 
   for (const status of executionStatuses) {
     const targets = notificationsObj[status];
     if (!_.isEmpty(targets)) {
-      const cfnTargets = targets
-        .map(t => compileTarget(stateMachineName, status, t))
-        .filter(_.isObjectLike);
-      targets
-        .map(compileIamPermission)
-        .filter(_.isObjectLike)
-        .forEach(({ action, resource }) => addPermission(action, resource));
+      const cfnTargets = targets.map(t =>
+        compileTarget(stateMachineName, status, t, iamRoleLogicalId));
 
       const eventRuleLogicalId =
         `${stateMachineLogicalId}Notifications${status.replace('_', '')}EventRule`;
@@ -144,22 +287,19 @@ function* compileResources(stateMachineLogicalId, stateMachineName, notification
             'detail-type': ['Step Functions Execution Status Change'],
             detail: {
               status: [status],
+              stateMachineArn: [{
+                Ref: stateMachineLogicalId,
+              }],
             },
           },
           Name: `${stateMachineName}-${status}-notification`,
-          RoleArn: {
-            'Fn::GetAtt': [iamRoleLogicalId, 'Arn'],
-          },
+          RoleArn: roleArn,
           Targets: cfnTargets,
         },
       };
       yield [eventRuleLogicalId, eventRule];
     }
   }
-
-  if (!_.isEmpty(iamRole.Properties.Policies[0].PolicyDocument.Statement)) {
-    yield [iamRoleLogicalId, iamRole];
-  }
 }
 
 function validateConfig(serverless, stateMachineName, notificationsObj) {