SQS 内のメッセージを EventBridge Pipes 経由で Step Functions(Express) に同期処理させる構成を CDK で実装してみる

npx cdk init app --language typescript

npm install @aws-cdk/aws-pipes-alpha
npm install @aws-cdk/aws-pipes-sources-alpha
npm install @aws-cdk/aws-pipes-targets-alpha

import * as cdk from "aws-cdk-lib";
import { Construct } from "constructs";
import * as sqs from "aws-cdk-lib/aws-sqs";
import * as sfn from "aws-cdk-lib/aws-stepfunctions";
import * as logs from "aws-cdk-lib/aws-logs";
import * as dynamodb from "aws-cdk-lib/aws-dynamodb";
import * as pipes from "@aws-cdk/aws-pipes-alpha";
import * as pipes_source from "@aws-cdk/aws-pipes-sources-alpha";
import * as pipes_target from "@aws-cdk/aws-pipes-targets-alpha";
import * as apigateway from "aws-cdk-lib/aws-apigateway";
import * as iam from "aws-cdk-lib/aws-iam";
import * as sfn_tasks from "aws-cdk-lib/aws-stepfunctions-tasks";
import * as sns from "aws-cdk-lib/aws-sns";

export class CdkSqsPipeStepFunctionsStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // SQS Dead Letter Queue
    const dlQueue = new sqs.Queue(this, "DeadLetterQueue", {
      fifo: false,
      visibilityTimeout: cdk.Duration.seconds(300),
    });

    // SQS Queue
    const queue = new sqs.Queue(this, "SqsQueue", {
      fifo: false,
      visibilityTimeout: cdk.Duration.seconds(30),
      deadLetterQueue: {
        maxReceiveCount: 5,
        queue: dlQueue,
      },
    });

    // API Gateway
    const api = new apigateway.RestApi(this, "ApiGatewaySqsApi", {
      description: "API that sends requests to SQS",
      deployOptions: {
        stageName: "prod",
      },
    });
    const apiGatewayRole = new iam.Role(this, "ApiGatewaySqsRole", {
      assumedBy: new iam.ServicePrincipal("apigateway.amazonaws.com"),
    });
    queue.grantSendMessages(apiGatewayRole);

    // Integration with SQS
    const sqsIntegration = new apigateway.AwsIntegration({
      service: "sqs",
      path: `${cdk.Aws.ACCOUNT_ID}/${queue.queueName}`,
      integrationHttpMethod: "POST",
      options: {
        credentialsRole: apiGatewayRole,
        requestParameters: {
          "integration.request.header.Content-Type":
            "'application/x-www-form-urlencoded'",
        },
        requestTemplates: {
          "application/json":
            "Action=SendMessage&MessageBody=$util.urlEncode($input.body)",
        },
        integrationResponses: [
          {
            statusCode: "200",
            responseTemplates: {
              "application/json": JSON.stringify({
                success: true,
                message: "Message sent to SQS",
              }),
            },
          },
        ],
      },
    });

    // POST method for API Gateway
    api.root.addMethod("POST", sqsIntegration, {
      methodResponses: [
        {
          statusCode: "200",
          responseModels: {
            "application/json": apigateway.Model.EMPTY_MODEL,
          },
        },
      ],
    });

    // SNS Topic
    const topic = new sns.Topic(this, "SnsTopic", {
      displayName: "SQS Topic",
    });

    // SNS Notification Task
    const snsTask = sfn_tasks.SnsPublish.jsonata(this, "SnsTask", {
      topic,
      message: sfn.TaskInput.fromText("{% $message %}"),
    });

    // Dynamodb Table
    const table = new dynamodb.Table(this, "Table", {
      partitionKey: { name: "id", type: dynamodb.AttributeType.STRING },
      removalPolicy: cdk.RemovalPolicy.DESTROY,
      billingMode: dynamodb.BillingMode.PAY_PER_REQUEST,
    });

    // CloudWatch Log Group for Step Functions
    const sfnLogGroup = new logs.LogGroup(this, "SfnLogGroup", {
      retention: logs.RetentionDays.ONE_MONTH,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    // Set Variables for Step Functions
    const setVariables = sfn.Pass.jsonata(this, "SetVariables", {
      assign: {
        id: "{% $parse($states.input.body).id %}",
        message: "{% $parse($states.input.body).message %}",
        startTime: "{% $toMillis($states.context.Execution.StartTime) %}",
      },
    });

    // Step Functions Task to Get item in DynamoDB
    const getItemTask = sfn_tasks.DynamoGetItem.jsonata(this, "GetItemTask", {
      table,
      key: {
        id: sfn_tasks.DynamoAttributeValue.fromString("{% $id %}"),
      },
    });

    // Step Functions Task to put item in DynamoDB
    const putItemTask = sfn_tasks.DynamoPutItem.jsonata(this, "PutItemTask", {
      table,
      item: {
        id: sfn_tasks.DynamoAttributeValue.fromString("{% $id %}"),
        message: sfn_tasks.DynamoAttributeValue.fromString("{% $message %}"),
        time: sfn_tasks.DynamoAttributeValue.fromString(
          "{% $string($startTime) %}"
        ),
      },
    });

    // Pass Task to set the choice
    const passTask = new sfn.Pass(this, "PassTask", {});

    const choice = sfn.Choice.jsonata(this, "Choice")
      .when(
        sfn.Condition.jsonata(
          "{% ($startTime - $number($states.input.Item.time.S)) >= 60 * 1000 * 5 %}"
        ),
        putItemTask.next(snsTask)
      )
      .otherwise(passTask);

    // Step Functions State Machine
    const definitionBody = sfn.DefinitionBody.fromChainable(
      setVariables.next(getItemTask.next(choice))
    );
    const stateMachine = new sfn.StateMachine(this, "StateMachine", {
      stateMachineType: sfn.StateMachineType.EXPRESS,
      definitionBody: definitionBody,
      queryLanguage: sfn.QueryLanguage.JSONATA,
      timeout: cdk.Duration.minutes(5),
      logs: {
        destination: sfnLogGroup,
        level: sfn.LogLevel.ALL,
        includeExecutionData: true,
      },
    });

    // CloudWatch Log Group for EventBridge Pipes
    const pipeLogGroup = new logs.LogGroup(this, "PipeLogGroup", {
      retention: logs.RetentionDays.ONE_MONTH,
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });
    const cwlLogDestination = new pipes.CloudwatchLogsLogDestination(
      pipeLogGroup
    );

    // EventBridge Pipes
    const pipe = new pipes.Pipe(this, "Pipe", {
      source: new pipes_source.SqsSource(queue),
      target: new pipes_target.SfnStateMachine(stateMachine, {
        invocationType:
          pipes_target.StateMachineInvocationType.REQUEST_RESPONSE,
      }),
      logLevel: pipes.LogLevel.TRACE,
      logIncludeExecutionData: [pipes.IncludeExecutionData.ALL],
      logDestinations: [cwlLogDestination],
    });
  }
}

const notifyFailure = new sfn_tasks.SnsPublish(this, "NotifyFailure", {
  topic: snsTopic,
  subject: "Task failed",
  message: sfn.TaskInput.fromJsonPathAt("$"),
  resultPath: "$.Notify",
});

const snsTask = sfn_tasks.SnsPublish.jsonata(this, "SnsTask", {
  topic,
  message: sfn.TaskInput.fromText("{% $message %}"),
});

// Set Variables for Step Functions
const setVariables = sfn.Pass.jsonata(this, "SetVariables", {
  assign: {
    id: "{% $parse($states.input.body).id %}",
    message: "{% $parse($states.input.body).message %}",
    startTime: "{% $toMillis($states.context.Execution.StartTime) %}",
  },
});

npx cdk deploy CdkSqsPipeStepFunctionsStack

% curl -v -X POST \
  -H "Content-Type: application/json" \
  -d '{"id":"1", "message": "test"}' \
  https://uxgl2kuuv7.execute-api.ap-northeast-1.amazonaws.com/prod

% curl -v -X POST \
  -H "Content-Type: application/json" \
  -d '{"id":"1", "message": "test"}' \
  https://uxgl2kuuv7.execute-api.ap-northeast-1.amazonaws.com/prod

% curl -v -X POST \
  -H "Content-Type: application/json" \
  -d '{"id":"2", "message": "test"}' \
  https://uxgl2kuuv7.execute-api.ap-northeast-1.amazonaws.com/prod