📣

Server-Sent Events(SSE)をLambdaから実行する

に公開

はじめに

最近、チャットアプリケーションを作成しており、ChatGPT のようなリアルタイムストリーミング応答機能を実装したいと技術調査をしていました。
この機能を実装するために、Server-Sent Events(SSE)を使うことにしました。

https://developer.mozilla.org/ja/docs/Web/API/Server-sent_events/Using_server-sent_events

そこで、AWS Lambda から SSE を実行する方法を調査しました。
本記事では、Hono ベースの SSE サーバを AWS Lambda Web Adapter を使い、 AWS Lambda から実行する方法を紹介します。

アーキテクチャ

今回、Lambda で SSE を実行するにあたり、以下の技術を使用しました。

Hono の SSE コードは以下です。

Hono の SSE コード
index.ts
app.post("/sse", (c) => {
  return streamSSE(c, async (stream) => {
    console.log("クライアントが接続しました");

    // ウェルカムメッセージ
    await stream.writeSSE({
      data: JSON.stringify({ message: "接続しました!" }),
    });

    // 10回データを送信
    for (let i = 0; i < 10; i++) {
      try {
        const data = {
          timestamp: new Date().toISOString(),
          randomValue: Math.floor(Math.random() * 100),
          message: `現在日時 ${new Date().toLocaleString()}`,
        };

        await stream.writeSSE({
          data: JSON.stringify(data),
        });

        // 2秒待機
        await stream.sleep(2000);
      } catch (error) {
        console.log("クライアントが切断しました");
        break;
      }
    }
  });
});

アーキテクチャ図

アーキテクチャ図

今回のアーキテクチャは CloudFront + Lambda@Edge + Lambda です。
Lambda の Function URL を CloudFront でプロキシし、CloudFront 経由でないと Lambda にアクセスできないようにしています。
また、 CloudFront のオリジンアクセスコントロール(OAC)Lambda@Edge を使用して、x-amz-content-sha256を付与しています。

AWS CDK のコードも以下に残しておきます。
※ Lambda、CloudFront、OAC の設定のみです。Lambda@Edge はコンソールから作りました。

AWS CDK のコード
lib.ts
import * as cdk from "aws-cdk-lib";
import * as cloudfront from "aws-cdk-lib/aws-cloudfront";
import * as origins from "aws-cdk-lib/aws-cloudfront-origins";
import * as ecr from "aws-cdk-lib/aws-ecr";
import * as lambda from "aws-cdk-lib/aws-lambda";
import type { Construct } from "constructs";

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

    // ECR Repository
    const repository = new ecr.Repository(this, "TestRepository", {
      repositoryName: "test-repository",
      removalPolicy: cdk.RemovalPolicy.DESTROY,
    });

    // Lambda Function
    const lambdaFunction = new lambda.Function(this, "TestLambda", {
      runtime: lambda.Runtime.FROM_IMAGE,
      code: lambda.Code.fromEcrImage(repository, {
        tagOrDigest: "latest",
      }),
      handler: lambda.Handler.FROM_IMAGE,
      timeout: cdk.Duration.minutes(5),
      memorySize: 150,
      environment: {
        AWS_LWA_INVOKE_MODE: "response_stream",
      },
    });

    repository.grantPull(lambdaFunction);

    // Lambda Function URL
    const lambdaFunctionUrl = lambdaFunction.addFunctionUrl({
      authType: lambda.FunctionUrlAuthType.AWS_IAM,
      cors: {
        allowedOrigins: ["*"],
        allowedMethods: [lambda.HttpMethod.ALL], // OPTIONS含め全許可
        allowedHeaders: ["*"],
      },
      invokeMode: lambda.InvokeMode.RESPONSE_STREAM,
    });

    // TODO: 既存のLambda@Edge関数を参照させる
    const edgeVersion = lambda.Version.fromVersionArn(
      this,
      `${prefixCapitalized}ExistingEdgeVersion`,
      `arn:aws:lambda:us-east-1:1234556789:function:lambda-edge-function:1`,
    );

    // CloudFront Distribution with OAC
    const oac = new cloudfront.CfnOriginAccessControl(
      this,
      "OriginAccessControl",
      {
        originAccessControlConfig: {
          name: "OAC for Lambda Function URLs",
          originAccessControlOriginType: "lambda",
          signingBehavior: "always",
          signingProtocol: "sigv4",
        },
      }
    );

    const distribution = new cloudfront.Distribution(this, "TestDistribution", {
      defaultBehavior: {
        origin:
          origins.FunctionUrlOrigin.withOriginAccessControl(lambdaFunctionUrl),
        allowedMethods: cloudfront.AllowedMethods.ALLOW_ALL,
        edgeLambdas: [
          {
            functionVersion: edgeVersion,
            eventType: cloudfront.LambdaEdgeEventType.ORIGIN_REQUEST,
            includeBody: true,
          },
        ],
        cachePolicy: cloudfront.CachePolicy.CACHING_DISABLED,
      },
    });

    // OACをCloudFront Distributionに関連付け
    const cfnDist = distribution.node
      .defaultChild as cloudfront.CfnDistribution;
    cfnDist.addPropertyOverride(
      "DistributionConfig.Origins.0.OriginAccessControlId",
      oac.attrId
    );

    new cdk.CfnOutput(this, "CloudFrontUrl", {
      value: `https://${distribution.domainName}`,
    });
  }
}

なぜ Lambda の Function URL を CloudFront でプロキシするのか?

Lambda でレスポンスストリーミングを使用するには、Function URL を使用する必要があります。

https://docs.aws.amazon.com/ja_jp/lambda/latest/dg/configuration-response-streaming.html

ちなみに、API Gateway はサポートされていないようです。

https://aws.amazon.com/jp/blogs/news/introducing-aws-lambda-response-streaming/

ただ、Function URL を直接叩くのはセキュリティ的に弱くなります。
そこで、CloudFront でプロキシすることで、CloudFront 経由でないと Lambda にアクセスできないようにします。
さらに、WAF を適用し、セキュリティを強化できることも、本構成のメリットです。

CloudFront OAC と Lambda@Edge を使用した x-amz-content-sha256ヘッダーの付与

本構成でPOSTメソッド、PUTメソッドを使用する場合、Lambda の Function URL にリクエストを送信する際に、x-amz-content-sha256ヘッダーを付与する必要があります。

https://docs.aws.amazon.com/ja_jp/AmazonCloudFront/latest/DeveloperGuide/private-content-restricting-access-to-lambda.html

そのため、以下の流れで CloudFront OAC と Lambda@Edge を使用して、x-amz-content-sha256ヘッダーを付与しています。

  1. CloudFront のオリジンアクセスコントロール (OAC)を使用した SigV4 署名プロトコルでリクエストを署名。
  2. CloudFront をトリガーとした Lambda@Edge でリクエストボディのハッシュ値を計算し、x-amz-content-sha256ヘッダーを付与。
  3. Lambda の Function URL にリクエストを送信。

ハッシュ値の結果に使用したコードは以下です。
Node.js 22 のランタイムを使用しています。

const crypto = require("crypto");

exports.handler = async (event) => {
  // event 構造チェック
  if (
    !event.Records ||
    !event.Records[0] ||
    !event.Records[0].cf ||
    !event.Records[0].cf.request
  ) {
    console.log("Invalid event structure:", JSON.stringify(event));
    return event; // 何もせず返す
  }

  const request = event.Records[0].cf.request;

  // body が無い場合 → Viewer Request や Include Body OFF
  if (!request.body || !request.body.data) {
    console.log("No body found in request. x-amz-content-sha256 not added.");
    return request;
  }

  let bodyData;

  if (request.body.encoding === "base64") {
    bodyData = Buffer.from(request.body.data, "base64");
  } else {
    bodyData = Buffer.from(request.body.data, "utf-8");
  }

  // SHA256 計算
  const hash = crypto.createHash("sha256").update(bodyData).digest("hex");

  // ヘッダ追加
  request.headers["x-amz-content-sha256"] = [
    {
      key: "x-amz-content-sha256",
      value: hash,
    },
  ];

  return request;
};

確認ができないと、403エラーが返ってくるので注意してください。

Lambda の呼び出しモードの設定

レスポンスストリーミングを使用するために以下を設定します。

  • Lambda の呼び出しモード(InvokeMode)をRESPONSE_STREAMに設定
  • 環境変数AWS_LWA_INVOKE_MODEresponse_streamに設定

https://docs.aws.amazon.com/ja_jp/lambda/latest/dg/config-rs-invoke-furls.html
https://github.com/awslabs/aws-lambda-web-adapter?tab=readme-ov-file#configurations

結果

実行した結果は以下になります。

実行

おわりに

以上、AWS Lambda から SSE を実行する方法を紹介しました。

参考記事

https://developer.mozilla.org/ja/docs/Web/API/Server-sent_events/Using_server-sent_events
https://docs.aws.amazon.com/ja_jp/lambda/latest/dg/configuration-response-streaming.html
https://aws.amazon.com/jp/blogs/news/introducing-aws-lambda-response-streaming/
https://docs.aws.amazon.com/ja_jp/lambda/latest/dg/config-rs-invoke-furls.html
https://dev.classmethod.jp/articles/cloudfront-lambda-url-with-post-put-request/
https://docs.aws.amazon.com/ja_jp/AmazonCloudFront/latest/DeveloperGuide/private-content-restricting-access-to-lambda.html
https://aws.amazon.com/jp/blogs/news/amazon-cloudfront-introduces-origin-access-control-oac/
https://dev.classmethod.jp/articles/axum-on-lambda-http-server-sent-events/#%25E3%2581%25AF%25E3%2581%2598%25E3%2582%2581%25E3%2581%25AB
https://github.com/awslabs/aws-lambda-web-adapter?tab=readme-ov-file#configurations

Discussion