import 'module-alias/register' ; import Serverless from 'serverless' ; import { config } from 'aws-sdk' ; import { DynamoService } from './dynamo/dynamoService' ; import { LambdaService } from './lambda/lambdaService' ; import { IamService } from './iam/iamService' ; import { DynamoStream , DynamoStreamType , IDynamoStream , StreamSpecification , Table } from './dynamo/types' ; import { RolePolicy } from './iam/types' ; import { EventSourcePosition } from './_base/EventSourcePosition' ; import { EventSourceMappingState } from './lambda/types' ; export class ServerlessDynamoStreamPlugin { readonly dynamoService : DynamoService ; readonly lambdaService : LambdaService ; readonly iamService : IamService ; readonly serverless : Serverless ; readonly options : Serverless . Options ; description readonly existingStreamKey : string = 'existingDynamoStream' ; readonly requiredDynamoStreamPermissions = [ 'dynamodb:GetRecords' , 'dynamodb:GetShardIterator' , 'dynamodb:DescribeStream' , 'dynamodb:ListStreams' ] ; description readonly hooks : { [ key ] : any ; } ; description readonly commands : { [ key ] : any ; } ; description readonly finalWaitPeriod : number = 60000 ; private readonly maxIamWaitAttempts : number = 3 ; private readonly iamWaitIntervalMillis : number = 5000 ; constructor ( serverless , options ) { this . serverless = serverless ; this . options = options ; this . commands = { 'dynamo-stream' : { usage : 'Creates and connects DynamoDB streams for pre-existing tables with AWS Lambdas' , lifecycleEvents : [ 'connect' ] , } , } ; this . hooks = { 'dynamo-stream:connect' : this . run . bind ( this ) , 'after:deploy:deploy' : this . run . bind ( this ) , } ; if ( ( serverless as any ) . configSchemaHandler ) { ( serverless as any ) . configSchemaHandler . defineFunctionEvent ( 'aws' , 'existingDynamoStream' , { type : 'object' , properties : { tableName : { type : 'string' } , streamType : { type : 'string' } , startingPosition : { type : 'string' } , } , required : [ 'tableName' , 'streamType' , 'startingPosition' ] , additionalProperties : false , } ) ; } config . update ( { region : serverless . service . provider . region , apiVersions : { dynamodb : '2012-08-10' , } } ) ; this . dynamoService = new DynamoService ( ) ; this . lambdaService = new LambdaService ( ) ; this . iamService = new IamService ( ) ; } async run ( ) { this . serverless . cli . log ( 'dynamoStream:connect - connecting functions to DynamoDB streams' ) ; const lambdaToStreams = this . getLambdaStreams ( ) ; if ( Object . keys ( lambdaToStreams ) . length <= 0 ) { return true ; } this . serverless . cli . log ( `The following functions need some connections: ${ Object . keys ( lambdaToStreams ) }` ) ; for ( const [ functionName , events ] of Object . entries ( lambdaToStreams ) ) { const functionObject = this . serverless . service . getFunction ( functionName ) ; const fullFunctionName = functionObject . name ; for ( const event of events ) { try { let table = await this . getTable ( event . tableName ) ; const updatedTable = await this . enableStreamIfNeeded ( event . tableName , event . streamType , table . streamSpecification ) ; if ( updatedTable ) { table = updatedTable ; this . serverless . cli . log ( `Updated the stream for table ${ event . tableName } to ${ event . streamType }` ) ; } const streamArn = table ?. latestStreamArn ; if ( ! streamArn ) { throw new Error ( 'Failed to find the stream arn to connect the function to.' ) ; } const lambdaRoleName = await this . getLambdaRoleName ( fullFunctionName ) ; this . serverless . cli . log ( `Fetched role name ${ lambdaRoleName } for lambda ${ fullFunctionName }` ) ; const rolePolicy = await this . getRolePolicy ( lambdaRoleName ) ; this . serverless . cli . log ( `Found policy name ${ rolePolicy . policyName } for lambda ${ fullFunctionName }` ) ; const newPermissionsAdded = await this . addPermissionsIfNeeded ( rolePolicy , streamArn , lambdaRoleName ) ; newPermissionsAdded ? this . serverless . cli . log ( `Created a new policy for Role ${ lambdaRoleName } to access the stream ${ streamArn }` ) : this . serverless . cli . log ( `Role ${ lambdaRoleName } already has a policy for ${ streamArn }` ) ; if ( newPermissionsAdded ) { let attempts = 0 ; while ( attempts <= this . maxIamWaitAttempts ) { attempts ? this . serverless . cli . log ( '.' ) : this . serverless . cli . log ( 'Hang in there, IAM takes some time to process new permissions.' ) ; attempts += 1 ; await this . sleep ( this . iamWaitIntervalMillis ) ; } } const mappingCreatedOrActivated = await this . createOrActivateMappingIfNeeded ( streamArn , fullFunctionName , event . startingPosition ) ; mappingCreatedOrActivated ? this . serverless . cli . log ( `Mapping of lambda ${ fullFunctionName } to table ${ event . tableName } stream is now active.` ) : this . serverless . cli . log ( `Active mapping lambda ${ fullFunctionName } to table ${ event . tableName } stream already exists.` ) ; } catch ( error ) { this . serverless . cli . log ( `Failed to connect lambda ${ fullFunctionName } to table ${ event . tableName }.` + ` Error: ${ error . message ? error . message : error }` ) ; } } } return true ; } description param rolePolicy param streamArn param lambdaRoleName private async addPermissionsIfNeeded ( rolePolicy , streamArn , lambdaRoleName ) { const policyDocument = JSON . parse ( decodeURIComponent ( rolePolicy . policyDocument ) ) ; const alreadyAllowsDynamoStream = policyDocument . Statement . some ( ( statement ) => statement . Resource . includes ( streamArn ) && this . requiredDynamoStreamPermissions . every ( requiredPermission => statement . Action . includes ( requiredPermission ) ) && statement . Effect === 'Allow' ) ; let newPermissionsAdded = false ; if ( ! alreadyAllowsDynamoStream ) { policyDocument . Statement . push ( { Action : this . requiredDynamoStreamPermissions , Resource : [ streamArn ] , Effect : 'Allow' , } ) ; rolePolicy . policyDocument = JSON . stringify ( policyDocument ) ; await this . iamService . putRolePolicy ( rolePolicy ) ; newPermissionsAdded = true ; } return newPermissionsAdded ; } description param lambdaRoleName private async getRolePolicy ( lambdaRoleName ) { const policyNames = await this . iamService . listRolePolicies ( lambdaRoleName ) ; const policyName = policyNames . find ( name => name . includes ( this . serverless . service . getServiceName ( ) ) ) ; if ( ! policyName ) { throw new Error ( 'Couldn\'t find Serverless policy, please make sure Serverless deploy succeeds first.' ) ; } return this . iamService . getRolePolicy ( lambdaRoleName , policyName ) ; } description param fullFunctionName private async getLambdaRoleName ( fullFunctionName ) { const configuration = await this . lambdaService . getFunctionConfiguration ( fullFunctionName ) ; const functionRoleArn = configuration ?. roleArn ; if ( ! functionRoleArn ) { throw new Error ( 'Function configuration doesn\'t have a role.' + ' Please make sure Serverless deploy succeeds to deploy the lambdas first.' ) ; } const fullLambdaRoleName = functionRoleArn ?. split ( ':' ) . pop ( ) ; const lambdaRoleName = fullLambdaRoleName ?. split ( '/' ) . pop ( ) ; if ( ! lambdaRoleName ) { throw new Error ( `Lambda role arn ${ functionRoleArn } doesn\'t have a name.` + ` Make sure the right role is set on lambda ${ fullFunctionName }.` ) ; } return lambdaRoleName ; } description getLambdaStreams ( ) { const lambdaToStreams = { } ; for ( const functionName of this . serverless . service . getAllFunctions ( ) ) { const events = this . serverless . service . getAllEventsInFunction ( functionName ) ; const existingStreamEvents = events . filter ( event => ( event as any ) [ this . existingStreamKey ] ) ; if ( existingStreamEvents . length > 0 ) { lambdaToStreams [ functionName ] = existingStreamEvents . map ( event => new DynamoStream ( ( event as any ) [ this . existingStreamKey ] as IDynamoStream ) ) ; } } return lambdaToStreams ; } description param currentStream param tableName param targetStreamType private async enableStreamIfNeeded ( tableName , targetStreamType , currentStream ? ) { if ( ! currentStream || ! currentStream . isEnabled || ( currentStream . isEnabled && currentStream . viewType !== targetStreamType ) ) { let updateStream = new StreamSpecification ( { isEnabled : true , viewType : targetStreamType , } ) ; updateStream . isEnabled = true ; return this . dynamoService . updateTable ( tableName , updateStream ) ; } else { return undefined ; } } description param tableName throws private async getTable ( tableName ) { const table = await this . dynamoService . describeTable ( tableName ) ; if ( ! table ) { throw new Error ( `Table ${ tableName } is not found.` ) ; } return table ; } description param streamArn param fullFunctionName param startingPosition private async createOrActivateMappingIfNeeded ( streamArn , fullFunctionName , startingPosition ) { const existingMappings = await this . lambdaService . listEventSourceMappings ( fullFunctionName ) ; let existingMapping = existingMappings . find ( mapping => mapping . eventSourceArn === streamArn ) ; let createdOrActivated = false ; if ( ! existingMapping ) { await this . lambdaService . createEventSourceMapping ( streamArn , fullFunctionName , startingPosition ) ; createdOrActivated = true ; } else if ( existingMapping && existingMapping . state !== EventSourceMappingState . ENABLED ) { if ( ! existingMapping . id ) { throw new Error ( 'Updating an event source mapping requires an id' ) ; } await this . lambdaService . enableEventSourceMapping ( existingMapping . id ) ; createdOrActivated = true ; } return createdOrActivated ; } private sleep ( ms ) { return new Promise ( ( resolve ) => { setTimeout ( resolve , ms ) ; } ) ; } } module . exports = ServerlessDynamoStreamPlugin ;