Integrate SQS and Lambda

index.ts

// Consumer
import {
  SQSEvent,
  SQSHandler, SQSRecord,
} from "aws-lambda";
import * as Lambda from "aws-sdk/clients/lambda";
import { InvocationRequest, InvocationResponse } from "aws-sdk/clients/lambda";

async function invokeWorkerLambda(record: SQSRecord): Promise<InvocationResponse> {
  const lambda = new Lambda({region: process.env.AWS_REGION});
  const params: InvocationRequest = {
    FunctionName: process.env.WORKER_LAMBDA_NAME,
    InvocationType: "Event",
    Payload: JSON.stringify(record),
  };
  return lambda.invoke(params).promise();
}

export const handler: SQSHandler = async (event: SQSEvent): Promise<void> => {
  try {
    if (event.Records && event.Records.length > 0) {
      const invocations: Array<Promise<InvocationResponse>> = [];
      for (const record of event.Records) {
        invocations.push(invokeWorkerLambda(record));
      }
      await Promise.all(invocations);
    }
  } catch (err) {
    console.error(err, err.stack);
  }
};

index2.ts

// Worker
import {
  SQSRecord,
} from "aws-lambda";
import * as SQS from "aws-sdk/clients/sqs";

async function deleteMessage(receiptHandle: string): Promise<{}> {
  const sqs = new SQS({region: process.env.AWS_REGION});
  return await sqs.deleteMessage({
    ReceiptHandle: receiptHandle,
    QueueUrl: process.env.TASK_QUEUE_URL,
  }).promise();
}

async function work(task: SQSRecord) {
  console.log(task);
  // TODO implement
  return Promise.resolve();
}

export const handler = async (event: SQSRecord) => {
  try {
    await work(event);
    await deleteMessage(event.receiptHandle);
  } catch (err) {
    console.error(err, err.stack);
  }
};

template.yaml

MyTaskQueue: 
  Type: AWS::SQS::Queue
  Properties: 
    VisibilityTimeout: 90
    RedrivePolicy: 
      deadLetterTargetArn: !Sub ${DeadLetterQueue.Arn}
      maxReceiveCount: 10
DeadLetterQueue: 
  Type: AWS::SQS::Queue

ConsumerLambdaRole: 
  Type: AWS::IAM::Role
  Properties: 
    AssumeRolePolicyDocument: 
      Version: '2012-10-17'
      Statement: 
      - Effect: Allow
        Principal: 
          Service: lambda.amazonaws.com
        Action: 
        - sts:AssumeRole
    Path: '/'
    Policies: 
    - PolicyName: logs
      PolicyDocument: 
        Statement: 
        - Effect: Allow
          Action: 
          - logs:CreateLogGroup
          - logs:CreateLogStream
          - logs:PutLogEvents
          Resource: arn:aws:logs:*:*:*
    - PolicyName: lambda
      PolicyDocument: 
        Statement: 
        - Effect: Allow
          Action: 
          - lambda:InvokeFunction
          Resource: !Sub ${WorkerLambda.Arn}
ConsumerLambda:
  Type: AWS::Serverless::Function
  Properties:
    CodeUri: ./consumer
    Handler: index.handler
    MemorySize: 128
    Role: !Sub ${ConsumerLambdaRole.Arn}
    Runtime: nodejs8.10
    Timeout: 90
    Environment:
      Variables:
        TASK_QUEUE_URL: !Ref MyTaskQueue
        WORKER_LAMBDA_NAME: !Ref WorkerLambda
          
LambdaFunctionEventSourceMapping:
  Type: AWS::Lambda::EventSourceMapping
  Properties:
    BatchSize: 10
    Enabled: true
    EventSourceArn: !GetAtt MyTaskQueue.Arn
    FunctionName: !GetAtt ConsumerLambda.Arn


WorkerLambdaRole: 
  Type: AWS::IAM::Role
  Properties: 
    AssumeRolePolicyDocument: 
      Version: '2012-10-17'
      Statement: 
      - Effect: Allow
        Principal: 
          Service: lambda.amazonaws.com
        Action: 
        - sts:AssumeRole
    Path: '/'
    Policies: 
    - PolicyName: logs
      PolicyDocument: 
        Statement: 
        - Effect: Allow
          Action: 
          - logs:CreateLogGroup
          - logs:CreateLogStream
          - logs:PutLogEvents
          Resource: arn:aws:logs:*:*:*
    - PolicyName: sqs
      PolicyDocument: 
        Statement: 
        - Effect: Allow
          Action: 
          - sqs:DeleteMessage
          Resource: !Sub ${TaskQueue.Arn}
WorkerLambda:
  Type: AWS::Serverless::Function
  Properties:
    CodeUri: ./worker 
    Handler: index.handler
    MemorySize: 128
    Role: !Sub ${WorkerLambdaRole.Arn}
    Runtime: nodejs8.10
    Timeout: 90
    Environment:
      Variables:
        TASK_QUEUE_URL: !Ref TaskQueue