serverless-express 自定义事件映射器实战：用 DynamoDB Streams 驱动 Express 应用

• 后端

serverless-express（即@codegenie/serverless-express）在 4.x 版本中除了原生支持 API Gateway、ALB、Lambda@Edge、VPC Lattice 等 HTTP 类事件源之外，还开放了eventSource高级接口，允许开发者把任意 AWS 服务事件（如 DynamoDB Streams、SNS、SQS、Kinesis 等）映射为 Express 能处理的 HTTP 请求。本文以仓库中的 custom-mapper-dynamodb 示例 为骨架，完整讲解自定义事件映射器的设计思路、getRequest/getResponse映射函数写法、本地开发验证方法，以及 SAM 部署配置，帮助你掌握"把非 HTTP 事件接入 Express 应用"的完整实战方案。

示例目录总览：一个完整的 DynamoDB 事件接入示例

custom-mapper-dynamodb是一个独立、可运行的示例应用，其目录结构如下：

examples/custom-mapper-dynamodb/ ├── scripts/ │ └── local.js # 本地模拟 Lambda 调用，喂入 DynamoDB 事件 ├── src/ │ ├── app.js # Express 应用（users CRUD） │ ├── lambda.js # Lambda 入口，装配自定义事件映射器 │ └── dynamodb-event-mappings.js # 核心：DynamoDB 事件 ⇄ HTTP 请求/响应的映射函数 ├── dynamodb-event.json # 模拟的 DynamoDB Streams 事件样本 ├── package.json └── template.yaml # SAM 部署模板（DynamoDB 表 + Stream 触发 Lambda）

该示例的业务场景是：一张 DynamoDB 表开启 Streams 后，每当表数据发生INSERT/MODIFY/REMOVE变更，Lambda 都会被触发，serverless-express 再将这条 Stream 记录翻译成对 Express 路由/users的POST、PUT、DELETE请求，从而用一个普通的 Express CRUD 应用直接消费数据库变更事件。

核心原理：事件源（Event Source）在 serverless-express 中的角色

理解自定义映射器之前，先看 serverless-express 的事件源抽象。在 4.x 中，每个事件源本质上由一对函数构成（见 src/configure.js 的配置入口，以及 src/event-sources/index.js 的getEventSource分发逻辑）：

• getRequest({ event, context, log })：把事件源原生事件（event）转换成 Node HTTP 请求所需的{ method, path, headers, body }；
• getResponse({ event, statusCode, body, headers, isBase64Encoded, response })：把 Express 处理完的响应再映射回事件源期望的返回结构。

serverless-express 内置了大量事件源映射器（API Gateway V1/V2、ALB、Lambda@Edge、SNS、SQS、Kinesis、EventBridge、Step Functions 等）。其中内置的 DynamoDB 映射器 src/event-sources/aws/dynamodb.js 非常简单——它把所有 Stream 记录原样作为POST请求体、host设为dynamodb.amazonaws.com转发给应用，并用emptyResponseMapper丢弃响应：

const getRequestValuesFromDynamoDB = ({ event }) => { const method = 'POST' const headers = { host: 'dynamodb.amazonaws.com' } const body = event return { method, headers, body } }

内置映射器只解决"事件能进来"的问题，并不做业务语义翻译。自定义事件映射器（eventSource配置）则允许你完全接管这层转换逻辑，把事件中的业务字段精准映射到 HTTP 方法、路径和响应上——这正是本示例的价值所在。

自定义映射器详解：dynamodb-event-mappings.js

示例的核心文件 src/dynamodb-event-mappings.js 实现了两个导出函数：mapDynamoDbEventToHttpRequest与mapResponseToDynamoDb。

请求方向：Stream 记录 → HTTP 请求

function getMethodBasedOnRecordEventName ({ record }) { const { eventName } = record switch (eventName) { case 'INSERT': return 'post' case 'MODIFY': return 'put' case 'REMOVE': return 'delete' } } function getPath ({ method, record }) { switch (method) { case 'post': return '/users' default: return `/users/${record.dynamodb.NewImage.id.S}` } } function mapDynamoDbEventToHttpRequest ({ event }) { const record = event.Records[0] const method = getMethodBasedOnRecordEventName({ event: record }) const path = getPath({ method, record }) return { method, path, headers: {} } }

这里建立了一张清晰的事件语义映射表：

几点实现细节值得注意：

• 字段路径：record.dynamodb.NewImage.id.S是 DynamoDB 流记录的标准结构，NewImage为变更后的数据（在StreamViewType为NEW_IMAGE时存在），S表示 String 类型属性；
• 取第一条记录：event.Records[0]只处理批次中的第一条记录。由于StreamViewType: NEW_IMAGE且BatchSize: 1（见下文 template.yaml），这种简化是安全的；
• 返回值结构：必须提供method、path、headers三个字段，与 README.md 中requestMapper的约定一致，body字段可省略（示例中的 CRUD 操作无需请求体）。

响应方向：Express 响应 → Lambda 返回值

function mapResponseToDynamoDb ({ statusCode, body, headers, isBase64Encoded }) { return { statusCode, body, headers, isBase64Encoded } }

响应映射器直接把 Express 响应的四个核心要素原样透传。在 Lambda 消费 DynamoDB Streams 的场景下，该返回值会被 Lambda 服务记录到 CloudWatch 日志，用于排查每次事件处理的结果。这个函数也印证了 README.md 中responseMapper的标准签名：接收statusCode/body/headers/isBase64Encoded，返回最终结果对象。

装配入口：lambda.js 中如何接入自定义映射器

src/lambda.js 展示了自定义映射器的装配方式：

require('source-map-support/register') const serverlessExpress = require('@codegenie/serverless-express') const app = require('./app') const { mapDynamoDbEventToHttpRequest, mapResponseToDynamoDb } = require('./dynamodb-event-mappings') exports.handler = serverlessExpress({ app, eventSource: { getRequest: mapDynamoDbEventToHttpRequest, getResponse: mapResponseToDynamoDb } })

serverlessExpress({...})返回的正是 src/configure.js 中定义的handler(event, context)异步函数。配置项eventSource覆盖了默认的事件源解析逻辑（默认逻辑见 src/event-sources/utils.js 的getEventSourceNameBasedOnEvent），从而完全由你的两个映射函数接管转换。

底层调用链：事件如何流经 Express

从源码结构可以还原这条完整的调用链（参考 src/transport.js 的forwardRequestToNodeServer）：

1. handler(event, context)被调用，src/configure.js 通过setCurrentInvoke缓存当前事件与上下文；
2. forwardRequestToNodeServer调用eventSource.getRequest({ event, context, log })得到{ method, headers, body, path }；
3. 这些值被包装成ServerlessRequest对象（src/request.js），通过 Express 框架适配器 src/frameworks/express.js 的app.handle(request, response)注入 Express 应用；
4. Express 路由处理完成后，forwardResponse（src/transport.js）调用eventSource.getResponse({...})得到最终返回对象并 resolve。

也就是说，只要你的getRequest返回结构合法，Express 应用的任何中间件、路由、错误处理机制都能照常工作，这正是该抽象设计的关键价值。

Express 应用：被 Streams 驱动的 users CRUD

src/app.js 是一个标准的 Express 4 应用，通过express.Router()暴露/users的完整 CRUD：

• GET /users：返回内存中的用户列表；
• POST /users：读取req.body.name创建用户，返回201；
• PUT /users/:userId：按 id 更新用户 name，找不到时返回404；
• DELETE /users/:userId：按 id 删除用户，找不到时返回404。

几个值得注意的工程细节：

• 无需app.listen：文件中注释明确说明"serverless-express 会为你创建服务器并监听 Unix Domain Socket（4.x 则使用 mock Request/Response 对象），所以可以移除常规的app.listen(3000)调用"；
• 临时内存存储：users数组与userIdCounter都位于模块作用域，注释标明是 "Ephemeral in-memory data store"，即冷启动后数据会重置，真实生产环境应替换为持久化存储；
• ID 类型转换：getUser与getUserIndex通过parseInt(userId)把路径参数转为数字，而映射器从 DynamoDB 取出的 id 是字符串，二者需要保持类型一致（示例中 DynamoDB 的 id 值为"b"这样的字符串，实际使用时需结合业务统一 id 的生成与存储格式）。

本地开发与验证：从 README 的开发命令说起

examples/custom-mapper-dynamodb/README.md 给出了示例应用的本地开发四步流程：

npm pack ../.. npm install ./codegenie-serverless-express-4.0.0-rc.1.tgz npm install --prefix ./src ./ node local.js

逐步说明：

1. npm pack ../..：在示例目录中对仓库根目录执行打包，生成codegenie-serverless-express-4.0.0-rc.1.tgz（版本号以实际打包产物为准）。由于示例依赖的是尚在发布流程中的 rc 版本包，通过npm pack本地打包是为了脱离 npm registry 验证最新源码；
2. npm install ./codegenie-serverless-express-4.0.0-rc.1.tgz：把上一步的 tgz 作为本地依赖安装进示例项目，等价于用最新代码替换package.json中声明的@codegenie/serverless-express版本；
3. npm install --prefix ./src ./：在src子目录下安装依赖（示例的package.json中#postinstall脚本注释也暗示了依赖安装在src/node_modules下的布局方式）；
4. node local.js：运行本地模拟脚本。

local.js 如何模拟 Lambda 调用

scripts/local.js 的机制非常直接：加载../src/lambda.js导出的 handler，把 dynamodb-event.json 作为event喂入：

const lambdaFunction = require('../src/lambda.js') const albEvent = require('../dynamodb-event.json') const server = lambdaFunction.handler(albEvent, {}) .then(v => { console.info(v) process.exit(0) }) .catch(v => { console.error(v) process.exit(1) })

同时脚本对exit、SIGINT（Ctrl+C）、SIGTSTP（Ctrl+Z）和uncaughtException都挂了清理回调，避免进程悬挂。运行时它会打印出响应映射器的返回值（即mapResponseToDynamoDb的输出），由此即可验证整条链路是否正确。

测试事件的构造要点

dynamodb-event.json 是一份典型的 DynamoDB Streams 事件样本，关键字段：

• eventName: "INSERT"—— 触发getMethodBasedOnRecordEventName返回post，从而路由到POST /users；
• dynamodb.NewImage.id.S: "b"—— 新记录的 id 属性；
• dynamodb.StreamViewType: "NEW_IMAGE"—— 与下方 SAM 模板中表的StreamSpecification保持一致；
• eventSourceARN—— 指向serverless-express-dynamodb-example-Table表的流 ARN。

你可以把eventName改成MODIFY/REMOVE并补上OldImage字段，再运行node local.js来观察映射器产生的PUT/DELETE请求行为。

部署到 AWS：SAM 模板解读

template.yaml 使用 SAM 模板一次性声明了 Lambda 函数与 DynamoDB 表：

ServerlessExpressFunction: Type: AWS::Serverless::Function Properties: CodeUri: ./src Handler: lambda.handler MemorySize: 1024 Runtime: nodejs20.x Timeout: 3 Events: DynamoDB1: Type: DynamoDB Properties: Stream: !GetAtt Table.StreamArn StartingPosition: TRIM_HORIZON BatchSize: 1

参数要点：

• Runtime: nodejs20.x：当前示例声明的 Lambda 运行时，Node.js 版本需与本地开发环境保持一致；
• BatchSize: 1：每次只向 Lambda 投递 1 条 Stream 记录，配合映射器只读取Records[0]的实现，保证每条变更都被独立处理；
• StartingPosition: TRIM_HORIZON：从流中可读取的最早记录开始消费，适合测试期观察全量事件；
• Timeout: 3：3 秒超时，足够处理单条记录，生产环境可按业务耗时调整。

表的定义同样关键，StreamSpecification.StreamViewType: NEW_IMAGE决定了record.dynamodb.NewImage一定存在，这是映射器读取 id 的前置条件；KeySchema与AttributeDefinitions将id定义为 String 类型的哈希键（与NewImage.id.S的类型匹配）。

部署命令见 package.json 的 scripts：

npm run package # aws cloudformation package，先上传产物到 S3 npm run deploy # aws cloudformation deploy，创建 ServerlessExpressDynamoDbExample 栈 npm run package-deploy # 两步合并执行

部署完成后，向表中写入/修改/删除数据即可触发 Lambda，并在 CloudWatch 中看到映射后的 HTTP 方法与路径对应的响应输出。

进阶对比：eventSourceRoutes 与内置事件源

如果你并不需要精细的语义映射，只是想"让 DynamoDB Streams 等事件也能触发同一个 Express 应用"，serverless-express 还提供了更轻量的eventSourceRoutes配置（见 README.md）：

serverlessExpress({ app, eventSourceRoutes: { 'AWS_DYNAMODB': '/dynamodb', 'AWS_SNS': '/sns', 'AWS_SQS': '/sqs', 'AWS_EVENTBRIDGE': '/eventbridge' // ... 其他事件源同理 } })

事件将以POST请求投递到对应路由，且内置映射器会把Host头设为对应服务域名（如 DynamoDB 为dynamodb.amazonaws.com），README 建议据此校验请求来源，避免内部事件被外部伪造。

两种方案的取舍一目了然：

本示例选择自定义eventSource，正是为了演示如何把INSERT/MODIFY/REMOVE三种数据库事件翻译成 Express 应用语义完整的 RESTful 请求，这是默认映射器无法直接做到的。

小结与扩展方向

通过custom-mapper-dynamodb示例，你已经掌握了 serverless-express 自定义事件映射器的完整套路：

1. 实现getRequest：把任意事件转换成{ method, path, headers, body }；
2. 实现getResponse：把 Express 响应映射回事件源期望的返回值；
3. 装配：serverlessExpress({ app, eventSource: { getRequest, getResponse } })；
4. 本地验证：用node local.js配合模拟事件文件跑通 handler；
5. 部署：用 SAM 模板声明事件源（本示例为 DynamoDB Streams）。

这套机制同样适用于 SNS、SQS、Kinesis、EventBridge、Step Functions 等其他非 HTTP 事件源——只需替换映射函数中的事件解析逻辑即可。示例的users存储仍是内存实现，若要投入生产，可结合getCurrentInvoke()（从 src/current-invoke.js 导出，README 在 README.md 有用法说明）读取原始事件中的业务数据，再接入真实数据库完成持久化。

• 后端