Merge pull request #783 from j5ik2o/fix/lambda-rmu-on-localstack

j5ik2o · web-flow · commit 86b4c1e426d4 · 2026-03-31T22:53:31.000+09:00
fix: Lambda RMU on LocalStack with DynamoDB Streams
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -58,6 +58,9 @@ jobs:
       - uses: pnpm/action-setup@v3
         with:
           version: 8
+      - uses: unfor19/install-aws-cli-action@v1
+        with:
+          version: "2"
       - name: Use Node.js ${{ matrix.node-version }}
         uses: actions/setup-node@v4
         with:
diff --git a/.gitignore b/.gitignore
@@ -19,3 +19,6 @@ yarn-error.log
 .turbo/
 .turbo/*
 
+# Local developer overrides (see common.env.default)
+common.env
+
diff --git a/.gitmodules b/.gitmodules
@@ -0,0 +1,3 @@
+[submodule "references/okite-ai"]
+	path = references/okite-ai
+	url = git@github.com:j5ik2o/okite-ai.git
diff --git a/README.ja.md b/README.ja.md
@@ -19,7 +19,7 @@
 - [x] Read API Server(GraphQL)
 - [x] Read Model Updater on Local
 - [x] Docker Compose Support
-- [ ] Read Model Updater on AWS Lambda
+- [x] Read Model Updater on AWS Lambda（LocalStack 向けビルド／デプロイと `common.env`。詳細は [ビルドとテスト](./docs/BUILD_AND_TEST.ja.md)）
 - [ ] Deployment to AWS
 
 ## 概要
diff --git a/README.md b/README.md
@@ -22,7 +22,7 @@ Please refer to [here](https://github.com/j5ik2o/cqrs-es-example) for implementa
 - [x] Read API Server(GraphQL)
 - [x] Read Model Updater on Local
 - [x] Docker Compose Support
-- [ ] Read Model Updater on AWS Lambda
+- [x] Read Model Updater on AWS Lambda (LocalStack: build/deploy scripts and `common.env`; see [Build and Test](./docs/BUILD_AND_TEST.md))
 - [ ] Deployment to AWS
 
 ## Component Composition
diff --git a/common.env.default b/common.env.default
@@ -0,0 +1,29 @@
+# Copy to common.env and adjust. common.env is gitignored.
+# Host-side AWS CLI uses this URL (Compose maps LocalStack to host port 34566).
+LOCALSTACK_ENDPOINT_URL=http://localhost:34566
+
+AWS_REGION=ap-northeast-1
+AWS_DEFAULT_REGION=ap-northeast-1
+AWS_ACCESS_KEY_ID=x
+AWS_SECRET_ACCESS_KEY=x
+
+STREAM_JOURNAL_TABLE_NAME=journal
+STREAM_MAX_ITEM_COUNT=32
+
+# Lambda container reaches MySQL on the Compose network (see docker-compose-databases.yml).
+DATABASE_URL_LAMBDA=mysql://ceer:ceer@mysql-local:3306/ceer
+
+FUNCTION_NAME=read-model-updater-rmu
+# LocalStack 2.x では nodejs20.x が無い場合があるため既定は nodejs18.x
+LAMBDA_RUNTIME=nodejs18.x
+COMPOSE_PROJECT=cqrs-es-example-js
+
+# 1 = after deploy, stop container read-model-updater-1 to avoid double stream processing
+STOP_LOCAL_RMU_AFTER_LAMBDA_DEPLOY=1
+
+# docker-compose-up.sh: build/deploy Lambda zip when non-zero
+DOCKER_COMPOSE_UP_BUILD_LAMBDA=1
+DOCKER_COMPOSE_UP_DEPLOY_LAMBDA=1
+
+# Override if your compose project name differs (affects LAMBDA_DOCKER_NETWORK default in compose).
+# LAMBDA_DOCKER_NETWORK=cqrs-es-example-js_default
diff --git a/docs/BUILD_AND_TEST.ja.md b/docs/BUILD_AND_TEST.ja.md
@@ -26,4 +26,25 @@ docker composeで起動したアプリケーションに対してE2Eテストを
 $ pnpm docker-build
 $ pnpm docker-compose-up
 $ pnpm verify-group-chat # E2E
-```
+```
+
+## Read Model Updater Lambda（LocalStack）
+
+Rust 版と同様に、Read Model Updater を LocalStack 上の Lambda（DynamoDB ストリーム → Lambda → MySQL）として動かす手順です。
+
+1. リポジトリルートに `common.env.default` を `common.env` としてコピーするか、`docker-compose-up.sh` / `docker-compose-e2e-test.sh` に任せて自動生成する。`common.env` は `.gitignore` されます。
+2. ホストから AWS CLI で LocalStack に接続するときは **`http://localhost:34566`** を使います（Compose がコンテナの `4566` をホストの `34566` に公開）。Compose 内のサービスからは `http://localstack:4566` です。
+3. デプロイ用 zip（Docker `linux/amd64`、Prisma の Lambda 用バイナリを含む）のビルド:
+
+   ```shell
+   $ pnpm build-read-model-updater-lambda
+   ```
+
+   成果物は `dist/lambda/read-model-updater/function.zip` です。
+
+4. `pnpm docker-compose-up` および `pnpm docker-compose-e2e-test` は、必要に応じて zip をビルドし、LocalStack 待機後に `deploy-read-model-updater-localstack.sh` を実行し、既定でコンテナ `read-model-updater-1` を停止します（ストリームの二重処理を防ぐため）。`DOCKER_COMPOSE_UP_DEPLOY_LAMBDA=0` や `DOCKER_COMPOSE_UP_BUILD_LAMBDA=0` でスキップできます。E2E 用コンテナは `profile: e2e` とし、スタック起動と Lambda デプロイの後に `verify-group-chat.sh` が走るようにしています。
+5. Compose 起動後に手動デプロイする場合:
+
+   ```shell
+   $ pnpm deploy-read-model-updater-localstack
+   ```
diff --git a/docs/BUILD_AND_TEST.md b/docs/BUILD_AND_TEST.md
@@ -27,3 +27,24 @@ $ pnpm docker-build
 $ pnpm docker-compose-up
 $ pnpm verify-group-chat # E2E
 ```
+
+## Read Model Updater Lambda (LocalStack)
+
+This mirrors the Rust example: run the Read Model Updater as an AWS Lambda on LocalStack (DynamoDB stream → Lambda → MySQL).
+
+1. Copy `common.env.default` to `common.env` at the repo root, or let `docker-compose-up.sh` / `docker-compose-e2e-test.sh` create it. `common.env` is gitignored.
+2. From the host, point the AWS CLI at **`http://localhost:34566`** (Compose maps container `4566` to host `34566`). Inside Compose, services use `http://localstack:4566`.
+3. Build the deployment zip (Docker `linux/amd64`, Prisma engines for Lambda):
+
+   ```shell
+   $ pnpm build-read-model-updater-lambda
+   ```
+
+   Output: `dist/lambda/read-model-updater/function.zip`.
+
+4. `pnpm docker-compose-up` and `pnpm docker-compose-e2e-test` can build the zip (unless `DOCKER_COMPOSE_UP_BUILD_LAMBDA=0`), wait for LocalStack, run `deploy-read-model-updater-localstack.sh`, and by default stop the `read-model-updater-1` container to avoid double-processing the stream. Set `DOCKER_COMPOSE_UP_DEPLOY_LAMBDA=0` to skip deploy. The e2e script brings stacks up, deploys Lambda, then runs the `e2e-test` container (`profile: e2e`) so GraphQL is ready before `verify-group-chat.sh` runs.
+5. To deploy manually after Compose is up:
+
+   ```shell
+   $ pnpm deploy-read-model-updater-localstack
+   ```
diff --git a/mise.toml b/mise.toml
@@ -0,0 +1,5 @@
+[tools]
+node = "20"
+python = "3"
+"npm:@fission-ai/openspec" = "1.2.0"
+"npm:happy-coder" = "0.13.0"
diff --git a/package.json b/package.json
@@ -11,6 +11,8 @@
     "docker-compose-up": "./tools/scripts/docker-compose-up.sh",
     "docker-compose-up-db": "./tools/scripts/docker-compose-up.sh -d",
     "docker-compose-down": "./tools/scripts/docker-compose-down.sh",
+    "build-read-model-updater-lambda": "./tools/scripts/build-read-model-updater-lambda.sh",
+    "deploy-read-model-updater-localstack": "./tools/scripts/deploy-read-model-updater-localstack.sh",
     "create-group-chat": "./tools/scripts/curl-create-group-chat.sh",
     "verify-group-chat": "./tools/e2e-test/verify-group-chat.sh",
     "build": "turbo build",
diff --git a/packages/bootstrap/package.json b/packages/bootstrap/package.json
@@ -2,7 +2,8 @@
   "name": "cqrs-es-example-js-bootstrap",
   "version": "1.0.0",
   "description": "",
-  "main": "src/index.ts",
+  "main": "dist/index.js",
+  "types": "dist/index.d.ts",
   "scripts": {
     "start": "ts-node src/index.ts",
     "start-write-api": "node dist/index.js writeApi",
diff --git a/packages/bootstrap/src/lambda-rmu-handler.ts b/packages/bootstrap/src/lambda-rmu-handler.ts
@@ -0,0 +1,29 @@
+import { PrismaClient } from "@prisma/client";
+import type { DynamoDBStreamEvent, Handler } from "aws-lambda";
+import { GroupChatDao, ReadModelUpdater } from "cqrs-es-example-js-rmu";
+
+let prisma: PrismaClient | undefined;
+let readModelUpdater: ReadModelUpdater | undefined;
+
+function getReadModelUpdater(): ReadModelUpdater {
+  if (readModelUpdater) {
+    return readModelUpdater;
+  }
+  const databaseUrl = process.env.DATABASE_URL;
+  if (!databaseUrl) {
+    throw new Error("DATABASE_URL is not set");
+  }
+  prisma = new PrismaClient({
+    datasources: {
+      db: {
+        url: databaseUrl,
+      },
+    },
+  });
+  readModelUpdater = ReadModelUpdater.of(GroupChatDao.of(prisma));
+  return readModelUpdater;
+}
+
+export const handler: Handler<DynamoDBStreamEvent, void> = async (event) => {
+  await getReadModelUpdater().updateReadModel(event);
+};
diff --git a/packages/command/domain/package.json b/packages/command/domain/package.json
@@ -2,7 +2,8 @@
   "name": "cqrs-es-example-js-command-domain",
   "version": "1.0.0",
   "description": "",
-  "main": "src/index.ts",
+  "main": "dist/index.js",
+  "types": "dist/index.d.ts",
   "scripts": {
     "build": "tsc && rimraf dist/internal/*.test.d.ts && rimraf dist/internal/*.test.js && rimraf dist/internal/test && rimraf dist/group-chat/*.test.*",
     "test": "NODE_OPTIONS=--experimental-vm-modules jest --no-cache",
diff --git a/packages/command/interface-adaptor-if/package.json b/packages/command/interface-adaptor-if/package.json
@@ -2,7 +2,8 @@
   "name": "cqrs-es-example-js-command-interface-adaptor-if",
   "version": "1.0.0",
   "description": "",
-  "main": "src/index.ts",
+  "main": "dist/index.js",
+  "types": "dist/index.d.ts",
   "scripts": {
     "build": "tsc && rm -f dist/internal/*.test.d.ts && rm -f dist/internal/*.test.js && rm -fr dist/internal/test",
     "test": "NODE_OPTIONS=--experimental-vm-modules jest --no-cache",
diff --git a/packages/command/interface-adaptor-impl/package.json b/packages/command/interface-adaptor-impl/package.json
@@ -2,7 +2,8 @@
   "name": "cqrs-es-example-js-command-interface-adaptor-impl",
   "version": "1.0.0",
   "description": "",
-  "main": "src/index.ts",
+  "main": "dist/index.js",
+  "types": "dist/index.d.ts",
   "scripts": {
     "build": "tsc && rimraf dist/internal/*.test.d.ts && rimraf dist/internal/*.test.js && rimraf dist/internal/test && rimraf dist/repository/group-chat/*.test.*",
     "test": "NODE_OPTIONS=--experimental-vm-modules jest --no-cache",
diff --git a/packages/command/interface-adaptor-impl/src/graphql/resolvers.ts b/packages/command/interface-adaptor-impl/src/graphql/resolvers.ts
@@ -20,7 +20,7 @@ import type { TaskEither } from "fp-ts/TaskEither";
 import { pipe } from "fp-ts/function";
 import { GraphQLError } from "graphql/error";
 import { Arg, Ctx, Mutation, Query, Resolver } from "type-graphql";
-import type {
+import {
   AddMemberInput,
   CreateGroupChatInput,
   DeleteGroupChatInput,
@@ -45,7 +45,7 @@ class GroupChatCommandResolver {
   @Mutation(() => GroupChatOutput)
   async createGroupChat(
     @Ctx() { groupChatCommandProcessor }: CommandContext,
-    @Arg("input") input: CreateGroupChatInput,
+    @Arg("input", () => CreateGroupChatInput) input: CreateGroupChatInput,
   ): Promise<GroupChatOutput> {
     return pipe(
       this.validateGroupChatName(input.name),
@@ -75,7 +75,7 @@ class GroupChatCommandResolver {
   @Mutation(() => GroupChatOutput)
   async deleteGroupChat(
     @Ctx() { groupChatCommandProcessor }: CommandContext,
-    @Arg("input") input: DeleteGroupChatInput,
+    @Arg("input", () => DeleteGroupChatInput) input: DeleteGroupChatInput,
   ): Promise<GroupChatOutput> {
     return pipe(
       this.validateGroupChatId(input.groupChatId),
@@ -105,7 +105,7 @@ class GroupChatCommandResolver {
   @Mutation(() => GroupChatOutput)
   async renameGroupChat(
     @Ctx() { groupChatCommandProcessor }: CommandContext,
-    @Arg("input") input: RenameGroupChatInput,
+    @Arg("input", () => RenameGroupChatInput) input: RenameGroupChatInput,
   ): Promise<GroupChatOutput> {
     return pipe(
       this.validateGroupChatId(input.groupChatId),
@@ -151,7 +151,7 @@ class GroupChatCommandResolver {
   @Mutation(() => GroupChatOutput)
   async addMember(
     @Ctx() { groupChatCommandProcessor }: CommandContext,
-    @Arg("input") input: AddMemberInput,
+    @Arg("input", () => AddMemberInput) input: AddMemberInput,
   ): Promise<GroupChatOutput> {
     return pipe(
       this.validateGroupChatId(input.groupChatId),
@@ -211,7 +211,7 @@ class GroupChatCommandResolver {
   @Mutation(() => GroupChatOutput)
   async removeMember(
     @Ctx() { groupChatCommandProcessor }: CommandContext,
-    @Arg("input") input: RemoveMemberInput,
+    @Arg("input", () => RemoveMemberInput) input: RemoveMemberInput,
   ): Promise<GroupChatOutput> {
     return pipe(
       this.validateGroupChatId(input.groupChatId),
@@ -257,7 +257,7 @@ class GroupChatCommandResolver {
   @Mutation(() => MessageOutput)
   async postMessage(
     @Ctx() { groupChatCommandProcessor }: CommandContext,
-    @Arg("input") input: PostMessageInput,
+    @Arg("input", () => PostMessageInput) input: PostMessageInput,
   ): Promise<MessageOutput> {
     return pipe(
       this.validateGroupChatId(input.groupChatId),
@@ -307,7 +307,7 @@ class GroupChatCommandResolver {
   @Mutation(() => GroupChatOutput)
   async deleteMessage(
     @Ctx() { groupChatCommandProcessor }: CommandContext,
-    @Arg("input") input: DeleteMessageInput,
+    @Arg("input", () => DeleteMessageInput) input: DeleteMessageInput,
   ): Promise<GroupChatOutput> {
     return pipe(
       this.validateGroupChatId(input.groupChatId),
diff --git a/packages/command/processor/package.json b/packages/command/processor/package.json
@@ -2,7 +2,8 @@
   "name": "cqrs-es-example-js-command-processor",
   "version": "1.0.0",
   "description": "",
-  "main": "src/index.ts",
+  "main": "dist/index.js",
+  "types": "dist/index.d.ts",
   "scripts": {
     "build": "tsc && rimraf dist/internal/*.test.d.ts && rimraf dist/internal/*.test.js && rimraf dist/internal/test && rimraf dist/group-chat/*.test.*",
     "test": "NODE_OPTIONS=--experimental-vm-modules jest --no-cache",
diff --git a/packages/infrastructure/package.json b/packages/infrastructure/package.json
@@ -2,7 +2,8 @@
   "name": "cqrs-es-example-js-infrastructure",
   "version": "1.0.0",
   "description": "",
-  "main": "src/index.ts",
+  "main": "dist/index.js",
+  "types": "dist/index.d.ts",
   "scripts": {
     "start": "ts-node src/index.ts",
     "start-write-api": "node dist/index.js writeApi",
diff --git a/packages/query/interface-adaptor/package.json b/packages/query/interface-adaptor/package.json
@@ -2,7 +2,8 @@
   "name": "cqrs-es-example-js-query-interface-adaptor",
   "version": "1.0.0",
   "description": "",
-  "main": "src/index.ts",
+  "main": "dist/index.js",
+  "types": "dist/index.d.ts",
   "scripts": {
     "prisma:generate": "prisma generate",
     "build": "tsc && rimraf dist/internal/*.test.d.ts && rimraf dist/internal/*.test.js && rimraf dist/internal/test && rimraf dist/*.test.*",
diff --git a/packages/rmu/package.json b/packages/rmu/package.json
@@ -2,7 +2,8 @@
   "name": "cqrs-es-example-js-rmu",
   "version": "1.0.0",
   "description": "",
-  "main": "src/index.ts",
+  "main": "dist/index.js",
+  "types": "dist/index.d.ts",
   "scripts": {
     "prisma:generate": "prisma generate",
     "build": "tsc && rimraf dist/internal/*.test.d.ts && rimraf dist/internal/*.test.js && rimraf dist/internal/test && rimraf dist/*.test.*",
diff --git a/packages/rmu/prisma/schema.prisma b/packages/rmu/prisma/schema.prisma
@@ -1,5 +1,6 @@
 generator client {
-  provider = "prisma-client-js"
+  provider      = "prisma-client-js"
+  binaryTargets = ["native", "rhel-openssl-3.0.x", "rhel-openssl-1.0.x"]
 }
 
 datasource db {
diff --git a/packages/rmu/src/update-read-model.ts b/packages/rmu/src/update-read-model.ts
@@ -44,15 +44,22 @@ class ReadModelUpdater {
         this.logger.warn("No NewImage");
         return;
       }
-      const base64EncodedPayload = attributeValues.payload.B;
-      if (!base64EncodedPayload) {
+      const rawPayload = attributeValues.payload.B;
+      if (!rawPayload) {
         this.logger.warn("No payload");
         return;
       }
-      const payload = this.decoder.decode(
-        new Uint8Array(base64EncodedPayload.split(",").map(Number)),
-      );
-      const payloadJson = JSON.parse(payload);
+      let payloadJson: unknown;
+      try {
+        // LocalStack: B フィールドに生 JSON 文字列が入る
+        payloadJson = JSON.parse(rawPayload);
+      } catch {
+        // AWS Lambda: Base64, ローカル RMU: カンマ区切り数値文字列
+        const payloadBytes = /^\d+(,\d+)*$/.test(rawPayload)
+          ? new Uint8Array(rawPayload.split(",").map(Number))
+          : new Uint8Array(Buffer.from(rawPayload, "base64"));
+        payloadJson = JSON.parse(this.decoder.decode(payloadBytes));
+      }
       const groupChatEvent = convertJSONToGroupChatEvent(payloadJson);
       switch (groupChatEvent.symbol) {
         case GroupChatCreatedTypeSymbol: {
diff --git a/references/okite-ai b/references/okite-ai
@@ -0,0 +1 @@
+Subproject commit 69fa7cb4761b8767d29310a1115a1e511daf5fcf
diff --git a/tools/docker-compose/docker-compose-databases.yml b/tools/docker-compose/docker-compose-databases.yml
diff --git a/tools/docker-compose/docker-compose-e2e-test.yml b/tools/docker-compose/docker-compose-e2e-test.yml
diff --git a/tools/e2e-test/verify-group-chat.sh b/tools/e2e-test/verify-group-chat.sh
diff --git a/tools/scripts/build-read-model-updater-lambda.docker-inner.sh b/tools/scripts/build-read-model-updater-lambda.docker-inner.sh
diff --git a/tools/scripts/build-read-model-updater-lambda.sh b/tools/scripts/build-read-model-updater-lambda.sh
diff --git a/tools/scripts/deploy-read-model-updater-localstack.sh b/tools/scripts/deploy-read-model-updater-localstack.sh
diff --git a/tools/scripts/docker-compose-e2e-test.sh b/tools/scripts/docker-compose-e2e-test.sh
diff --git a/tools/scripts/docker-compose-up.sh b/tools/scripts/docker-compose-up.sh

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+[submodule "references/okite-ai"]`
	`2`	`+ path = references/okite-ai`
	`3`	`+ url = git@github.com:j5ik2o/okite-ai.git`