serverless/lib/plugins/aws/package/compile/events/stream.js

'use strict';

const _ = require('lodash');
const resolveLambdaTarget = require('../../../utils/resolveLambdaTarget');

class AwsCompileStreamEvents {
  constructor(serverless) {
    this.serverless = serverless;
    this.provider = this.serverless.getProvider('aws');

    this.hooks = {
      'package:compileEvents': this.compileStreamEvents.bind(this),
    };

    this.serverless.configSchemaHandler.defineFunctionEvent('aws', 'stream', {
      anyOf: [
        { $ref: '#/definitions/awsArnString' },
        {
          type: 'object',
          properties: {
            // arn constraints are listed in oneOf property of this schema
            arn: {},
            type: { enum: ['dynamodb', 'kinesis'] },
            batchSize: { type: 'integer', minimum: 1, maximum: 10000 },
            parallelizationFactor: { type: 'integer', minimum: 1, maximum: 10 },
            startingPosition: { enum: ['LATEST', 'TRIM_HORIZON'] },
            enabled: { type: 'boolean' },
            consumer: { anyOf: [{ const: true }, { $ref: '#/definitions/awsArn' }] },
            batchWindow: { type: 'integer', minimum: 0, maximum: 300 },
            maximumRetryAttempts: { type: 'integer', minimum: -1, maximum: 10000 },
            bisectBatchOnFunctionError: { type: 'boolean' },
            maximumRecordAgeInSeconds: {
              anyOf: [{ const: -1 }, { type: 'integer', minimum: 60, maximum: 604800 }],
            },
            functionResponseType: { enum: ['ReportBatchItemFailures'] },
            destinations: {
              type: 'object',
              properties: {
                onFailure: {
                  anyOf: [
                    { $ref: '#/definitions/awsArnString' },
                    {
                      type: 'object',
                      properties: {
                        // arn constraints are listed in oneOf property of this schema
                        arn: {},
                        type: { enum: ['sns', 'sqs'] },
                      },
                      additionalProperties: false,
                      anyOf: [
                        {
                          properties: {
                            arn: { $ref: '#/definitions/awsCfFunction' },
                          },
                          required: ['arn', 'type'],
                        },
                        {
                          properties: {
                            arn: { $ref: '#/definitions/awsArnString' },
                          },
                          required: ['arn'],
                        },
                      ],
                    },
                  ],
                },
              },
              additionalProperties: false,
              required: ['onFailure'],
            },
          },
          additionalProperties: false,
          anyOf: [
            {
              properties: {
                arn: { $ref: '#/definitions/awsCfFunction' },
              },
              required: ['arn', 'type'],
            },
            {
              properties: {
                arn: { $ref: '#/definitions/awsArnString' },
              },
              required: ['arn'],
            },
          ],
        },
      ],
    });
  }

  compileStreamEvents() {
    this.serverless.service.getAllFunctions().forEach((functionName) => {
      const functionObj = this.serverless.service.getFunction(functionName);

      if (functionObj.events) {
        const dynamodbStreamStatement = {
          Effect: 'Allow',
          Action: [
            'dynamodb:GetRecords',
            'dynamodb:GetShardIterator',
            'dynamodb:DescribeStream',
            'dynamodb:ListStreams',
          ],
          Resource: [],
        };
        const kinesisStreamStatement = {
          Effect: 'Allow',
          Action: [
            'kinesis:GetRecords',
            'kinesis:GetShardIterator',
            'kinesis:DescribeStream',
            'kinesis:ListStreams',
          ],
          Resource: [],
        };
        const kinesisStreamWithConsumerStatement = {
          Effect: 'Allow',
          Action: [
            'kinesis:GetRecords',
            'kinesis:GetShardIterator',
            'kinesis:DescribeStreamSummary',
            'kinesis:ListShards',
          ],
          Resource: [],
        };
        const kinesisConsumerStatement = {
          Effect: 'Allow',
          Action: ['kinesis:SubscribeToShard'],
          Resource: [],
        };
        const onFailureSnsStatement = {
          Effect: 'Allow',
          Action: ['sns:Publish'],
          Resource: [],
        };
        const onFailureSqsStatement = {
          Effect: 'Allow',
          Action: ['sqs:ListQueues', 'sqs:SendMessage'],
          Resource: [],
        };

        functionObj.events.forEach((event) => {
          if (event.stream) {
            let EventSourceArn;
            let BatchSize = 10;
            let ParallelizationFactor;
            let StartingPosition = 'TRIM_HORIZON';
            let Enabled = true;

            if (typeof event.stream === 'object') {
              EventSourceArn = event.stream.arn;
              BatchSize = event.stream.batchSize || BatchSize;
              if (event.stream.parallelizationFactor) {
                ParallelizationFactor = event.stream.parallelizationFactor;
              }
              StartingPosition = event.stream.startingPosition || StartingPosition;
              if (typeof event.stream.enabled !== 'undefined') {
                Enabled = event.stream.enabled;
              }
            } else {
              EventSourceArn = event.stream;
            }

            const streamType = event.stream.type || EventSourceArn.split(':')[2];
            const streamName = (function () {
              if (EventSourceArn['Fn::GetAtt']) {
                return EventSourceArn['Fn::GetAtt'][0];
              } else if (EventSourceArn['Fn::ImportValue']) {
                return EventSourceArn['Fn::ImportValue'];
              } else if (EventSourceArn.Ref) {
                return EventSourceArn.Ref;
              } else if (EventSourceArn['Fn::Join']) {
                // [0] is the used delimiter, [1] is the array with values
                const name = EventSourceArn['Fn::Join'][1].slice(-1).pop();
                if (name.split('/').length) {
                  return name.split('/').pop();
                }
                return name;
              }
              return EventSourceArn.split('/')[1];
            })();

            const streamLogicalId = this.provider.naming.getStreamLogicalId(
              functionName,
              streamType,
              streamName
            );

            const dependsOn = [];
            const functionIamRoleResourceName =
              this.provider.resolveFunctionIamRoleResourceName(functionObj);
            if (functionIamRoleResourceName) {
              dependsOn.push(functionIamRoleResourceName);
            }
            const { targetAlias } = this.serverless.service.functions[functionName];
            if (targetAlias) {
              dependsOn.push(targetAlias.logicalId);
            }

            const streamResource = {
              Type: 'AWS::Lambda::EventSourceMapping',
              DependsOn: dependsOn,
              Properties: {
                BatchSize,
                ParallelizationFactor,
                EventSourceArn,
                FunctionName: resolveLambdaTarget(functionName, functionObj),
                StartingPosition,
                Enabled,
              },
            };

            // add event source ARNs to PolicyDocument statements
            if (streamType === 'dynamodb') {
              dynamodbStreamStatement.Resource.push(EventSourceArn);
            } else if (event.stream.consumer) {
              kinesisStreamWithConsumerStatement.Resource.push(EventSourceArn);
            } else {
              kinesisStreamStatement.Resource.push(EventSourceArn);
            }

            if (event.stream.batchWindow != null) {
              streamResource.Properties.MaximumBatchingWindowInSeconds = event.stream.batchWindow;
            }

            if (event.stream.maximumRetryAttempts != null) {
              streamResource.Properties.MaximumRetryAttempts = event.stream.maximumRetryAttempts;
            }

            if (event.stream.bisectBatchOnFunctionError != null) {
              streamResource.Properties.BisectBatchOnFunctionError =
                event.stream.bisectBatchOnFunctionError;
            }

            if (event.stream.functionResponseType != null) {
              streamResource.Properties.FunctionResponseTypes = [event.stream.functionResponseType];
            }

            if (event.stream.maximumRecordAgeInSeconds) {
              streamResource.Properties.MaximumRecordAgeInSeconds =
                event.stream.maximumRecordAgeInSeconds;
            }

            if (event.stream.destinations) {
              let OnFailureDestinationArn;

              if (typeof event.stream.destinations.onFailure === 'object') {
                OnFailureDestinationArn = event.stream.destinations.onFailure.arn;
              } else {
                OnFailureDestinationArn = event.stream.destinations.onFailure;
              }

              const destinationType =
                event.stream.destinations.onFailure.type || OnFailureDestinationArn.split(':')[2];
              // add on failure destination ARNs to PolicyDocument statements
              if (destinationType === 'sns') {
                onFailureSnsStatement.Resource.push(OnFailureDestinationArn);
              } else {
                onFailureSqsStatement.Resource.push(OnFailureDestinationArn);
              }

              streamResource.Properties.DestinationConfig = {
                OnFailure: {
                  Destination: OnFailureDestinationArn,
                },
              };
            }

            const newStreamObject = {
              [streamLogicalId]: streamResource,
            };

            if (event.stream.consumer && streamType === 'kinesis') {
              if (event.stream.consumer === true) {
                const consumerName = this.provider.naming.getStreamConsumerName(
                  functionName,
                  streamName
                );
                const consumerResource = {
                  Type: 'AWS::Kinesis::StreamConsumer',
                  Properties: {
                    StreamARN: EventSourceArn,
                    ConsumerName: consumerName,
                  },
                };
                const consumerLogicalId =
                  this.provider.naming.getStreamConsumerLogicalId(consumerName);
                newStreamObject[consumerLogicalId] = consumerResource;
                if (Array.isArray(streamResource.DependsOn)) {
                  streamResource.DependsOn.push(consumerLogicalId);
                } else {
                  streamResource.DependsOn = [streamResource.DependsOn, consumerLogicalId];
                }
                const consumerArnRef = {
                  Ref: consumerLogicalId,
                };
                streamResource.Properties.EventSourceArn = consumerArnRef;
                kinesisConsumerStatement.Resource.push(consumerArnRef);
              } else {
                const consumerArn = event.stream.consumer;
                streamResource.Properties.EventSourceArn = consumerArn;
                kinesisConsumerStatement.Resource.push(consumerArn);
              }
            }

            _.merge(
              this.serverless.service.provider.compiledCloudFormationTemplate.Resources,
              newStreamObject
            );
          }
        });

        // update the PolicyDocument statements (if default policy is used)
        if (
          this.serverless.service.provider.compiledCloudFormationTemplate.Resources
            .IamRoleLambdaExecution
        ) {
          const statement =
            this.serverless.service.provider.compiledCloudFormationTemplate.Resources
              .IamRoleLambdaExecution.Properties.Policies[0].PolicyDocument.Statement;
          if (dynamodbStreamStatement.Resource.length) {
            statement.push(dynamodbStreamStatement);
          }
          if (kinesisStreamStatement.Resource.length) {
            statement.push(kinesisStreamStatement);
          }
          if (kinesisStreamWithConsumerStatement.Resource.length) {
            statement.push(kinesisStreamWithConsumerStatement);
          }
          if (kinesisConsumerStatement.Resource.length) {
            statement.push(kinesisConsumerStatement);
          }
          if (onFailureSnsStatement.Resource.length) {
            statement.push(onFailureSnsStatement);
          }
          if (onFailureSqsStatement.Resource.length) {
            statement.push(onFailureSqsStatement);
          }
        }
      }
    });
  }
}

module.exports = AwsCompileStreamEvents;