Prisma 服务端订阅（Server-side Subscriptions）实战：用 Webhook 把数据库变更推送给 Serverless 应用-平芜编程栈

后端
数据库
GraphQL

【免费下载链接】prisma1

💾 Database Tools incl. ORM, Migrations and Admin UI (Postgres, MySQL & MongoDB) [deprecated]

项目地址：https://gitcode.com/gh_mirrors/pr/prisma1

导读

本文讲解 Prisma（当前仓库 prisma1，即 deprecated 的 Prisma 1.x 开源版本）中的**服务端订阅（Server-side Subscriptions）**机制：它拥有与普通 GraphQL Subscriptions 等价的过滤能力，但事件不是通过 WebSocket 推送给订阅客户端，而是由 Prisma 服务端在数据变更时执行订阅查询，并把查询结果以Webhook的形式通过 HTTP 推送到你指定的 URL。读完本文，你将掌握在prisma.yml中配置服务端订阅的完整写法（含内联 query、.graphql文件、自定义 headers 等变体），理解其触发与投递的底层执行链路，并能从源码与集成测试层面验证其行为边界。

服务端订阅：与 GraphQL 订阅等价、交付方式不同

能力等价：同样的过滤 API

原文档明确指出：服务端订阅在能力上与普通 GraphQL 订阅完全等价。这意味着你可以使用与客户端 GraphQL 订阅完全相同的过滤语法，例如：

mutation_in: [CREATED, UPDATED, DELETED]：只在感兴趣的事件类型上收到通知；
node内层过滤：对变更节点的字段值进行约束；
previousValues：读取变更前的旧值。

Prisma 会持续监控数据变更，当一次 mutation 发生且满足订阅查询中的过滤条件时，就执行该订阅查询——这与普通 GraphQL 订阅的触发逻辑一致，区别仅在于结果如何送达：

对比维度	普通 GraphQL 订阅	服务端订阅（Server-side Subscriptions）
过滤 API	相同（`where`等参数）	相同
触发时机	数据变更时由 Prisma 监控并执行查询	数据变更时由 Prisma 监控并执行查询
交付方式	通过 WebSocket 推送给在线订阅客户端	通过 HTTP Webhook 投递到配置的 URL
目标场景	浏览器 / 实时前端	无状态的后端服务、Serverless 函数

为 Serverless 而生

服务端订阅的设计初衷是与现代的Serverless 基础设施协同工作：Serverless 函数（如 AWS Lambda）是无状态的、无法维持长连接，因此基于 WebSocket 的订阅模型并不适合直接对接。Webhook 模型让 Prisma 可以在事件发生时发起一次普通的 HTTP 请求，Serverless 函数只需要暴露一个 HTTP 端点即可接收事件。

从本仓库的实现看，当前版本支持通过webhook交付事件；文档同时预告了未来会加入对 AWS Lambda 直接调用以及不同消息队列实现的支持（当前以 webhook 为主要交付通道，相关队列基础设施可参见 PrismaLocalDependencies.scala 与 PrismaProdDependencies.scala 中的webhookPublisher/webhooksConsumer定义）。

配置服务端订阅：prisma.yml 中的 subscriptions 属性

服务端订阅的配置入口是服务根目录下的prisma.yml文件，通过subscriptions属性声明。该属性在prisma.yml的 YAML 结构文档中被明确定义为**可选（optional）**对象，每个订阅至少需要两部分信息：

订阅查询（subscription query）：定义在何种事件上触发函数、以及回调负载（payload）长什么样；
webhook 的 URL：事件发生时通过 HTTP 调用的地址；
（可选）若干 HTTP headers：附加到发送到该 URL 的请求上。

完整配置示例（原文档）

以下配置声明了一个名为userChangedEmail的服务端订阅：当任意user节点发生UPDATED变更时，Prisma 执行订阅查询，并把查询结果 POST 到http://example.org/sendSlackMessage，同时携带Content-Type与Authorization两个请求头：

endpoint: ${env:PRISMA_ENDPOINT} secret: ${env:PRISMA_SECRET} datamodel: database/datamodel.graphql subscriptions: userChangedEmail: webhook: url: http://example.org/sendSlackMessage headers: Content-Type: application/json Authorization: Bearer cha2eiheiphesash3shoofo7eceexaequeebuyaequ1reishiujuu6weisao7ohc query: | subscription { user(where: { mutation_in: [UPDATED] }) { node { name email } } }

要点解读：

subscriptions的每个键（这里是userChangedEmail）是订阅的自定义名称，也是投递到 webhook 时标识函数身份的functionName；
query可以像上面这样内联写在prisma.yml中，也可以指向一个.graphql文件（见下文）；
webhook需要提供url与可选的headers；headers 常用于鉴权（如Authorization）与告知接收方内容类型。

触发示例

上面配置的userChangedEmail订阅，会在执行如下 mutation 时被触发：

mutation { updateUser( data: { email: "new@email.com" }, where: { id: "cjcgo976g5twb018740bzyy4q" } ) { id } }

该 mutation 把 id 为cjcgo976g5twb018740bzyy4q的user的email更新为new@email.com。因为订阅过滤条件mutation_in: [UPDATED]命中了UPDATED类型，Prisma 会执行订阅查询，webhook 负载中包含node（更新后的name与email）。

subscriptions 属性的完整语法与变体

结合 prisma.yml 的 YAML 结构参考，subscriptions的完整用法如下：

类型定义

query（必填）：订阅查询的文件路径，或内联的 GraphQL 订阅字符串；
webhook（必填）：要调用的 webhook 信息（URL 与可选 headers）。如果没有 headers，可以直接把 URL 字符串赋给webhook属性；如果带 headers，则webhook需是一个含url与headers的对象。

变体一：无 headers，直接使用 URL 字符串

subscriptions: sendWelcomeEmail: query: database/subscriptions/sendWelcomeEmail.graphql webhook: https://bcdeaxokbj.execute-api.eu-west-1.amazonaws.com/dev/sendWelcomeEmail

变体二：带多个 headers

subscriptions: sendWelcomeEmail: query: database/subscriptions/sendWelcomeEmail.graphql webhook: url: https://bcdeaxokbj.execute-api.eu-west-1.amazonaws.com/dev/sendWelcomeEmail headers: Authorization: ${env:MY_ENDPOINT_SECRET} Content-Type: application/json

变体三：与 custom 变量结合

subscriptions中的值同样支持变量引用（如${env:...}、${self:custom...}），便于复用端点地址与查询目录：

custom: serverlessEndpoint: https://bcdeaxokbj.execute-api.eu-west-1.amazonaws.com/dev subscriptionQueries: database/subscriptions/ subscriptions: sendWelcomeEmail: query: ${self:custom.subscriptionQueries}/sendWelcomeEmail.graphql webhook: https://${self:custom.serverlessEndpoint}/sendWelcomeEmail

源码级剖析：从 prisma.yml 到 webhook 投递的完整执行链路

1. CLI 侧：订阅配置的解析

Prisma CLI 在部署时读取prisma.yml并解析订阅定义，核心逻辑位于 cli/packages/prisma-yml/src/PrismaDefinition.ts 的getSubscriptions()方法（PrismaDefinition.ts 第 318-352 行）：

遍历subscriptions对象的每个键值对；
兼容webhook的两种写法：若webhook是字符串则直接作为 URL，若是对象则取.url与.headers（headers 会被transformHeaders归一化处理）；
对query做处理：如果query以.graphql结尾，则视为相对prisma.yml所在目录的文件路径，读取该文件内容作为查询（文件不存在会抛出明确错误：Subscription query ... provided in subscription ... in prisma.yml does not exist）；否则视为内联查询字符串；
最终产出{ name, query, headers, url }的订阅描述供部署使用。

2. 服务器侧：触发与执行

当一次 mutation 在数据库中产生变更后，API 服务器会生成副作用 mutaction（side-effect mutaction），由 SideEffectMutactionExecutor.scala 分派执行（第 22-25 行）：

def execute(mutaction: SideEffectMutaction): Future[Unit] = mutaction match { case mutaction: PublishSubscriptionEvent => PublishSubscriptionEventExecutor.execute(mutaction, apiDependencies.sssEventsPubSub) case mutaction: ExecuteServerSideSubscription => ServerSideSubscriptionExecutor.execute(mutaction) }

其中ServerSideSubscriptionExecutor.execute（第 37-75 行）是服务端订阅的核心：

根据function.delivery的形态判断交付方式，WebhookDelivery走 webhook 分支；
调用SubscriptionExecutor.execute在内存中执行订阅查询，注意两个关键参数：
- skipPermissionCheck = true：服务端订阅的查询不受普通 API 权限检查限制（由服务端自动执行）；
- alwaysQueryMasterDatabase = true：始终查询主数据库，保证读到最新数据；
只有当执行结果包含data键（即订阅查询有匹配结果）时，才构造Webhook并发布到webhookPublisher队列：
- url、headers直接来自prisma.yml中的配置；
- payload是订阅查询的 JSON 结果字符串；
- projectId、functionName（即prisma.yml中订阅的名字）、requestId用于追踪标识；
若查询结果为空（过滤器未匹配），则不发布任何 webhook。

Webhook的数据结构定义在 server/servers/api/src/main/scala/com/prisma/subscriptions/Webhook.scala：包含projectId、functionName、requestId、url、payload、id、headers七个字段。

3. 过滤器评估与结果判定

server/servers/api/src/main/scala/com/prisma/subscriptions/SubscriptionExecutor.scala 揭示了“订阅是否触发”的判定逻辑（第 72-104 行）：

用本次 mutation 的类型、updatedFields、previousValues构建内部订阅 schema（SubscriptionSchema）；
通过QueryTransformer.evaluateInMemoryFilters与VariablesTransformer.evaluateInMemoryFilters先在内存中评估过滤条件（如mutation_in、node字段约束）；
只有filtersMatch && variablesMatch都成立时，才用 Sangria Executor 真正执行转换后的订阅查询；
执行完毕后，在结果中查找与模型名对应的顶层字段（如user），若该字段值为null（例如node: null），则整体视为未匹配、不投递 webhook（第 127-130 行）。

测试验证：webhook 在各种变更类型下的行为

仓库中的集成测试 server/servers/api/src/test/scala/com/prisma/subscriptions/EmbeddedServerSideSubscriptionSpec.scala 通过内存 webhook 队列（webhookTestKit）完整验证了服务端订阅的行为，这些用例可以帮助你理解并预测生产环境中的实际表现：

场景	行为
`createTodo`创建节点，命中`CREATED`订阅	发布 1 个 webhook，payload 中`node`为新建节点数据、`previousValues`为`null`
`updateTodo`更新节点，命中`UPDATED`订阅	发布 1 个 webhook，payload 同时包含更新后的`node`与`previousValues`（旧值）
`deleteTodo`删除节点，命中`DELETED`订阅	发布 1 个 webhook，payload 中`node`为`null`、`previousValues`为删除前的值
嵌套变更（如`createTodo`内嵌`create`comment，命中`comment`订阅）	同样发布 1 个 webhook，`node.comments`包含嵌套创建的节点
变更不满足过滤条件（如`status: DONE`不匹配`node.status: ACTIVE`过滤）	不发布任何 webhook

从测试断言还可以看到两个重要事实：

webhook 的 headers 会原样传递：测试配置headers = Vector("header" -> "value")，最终断言webhook.headers == Map("header" -> "value")；
payload 即为订阅查询的 JSON 结果：如更新场景下 payload 形如{"data": {"todo": {"node": {...}, "previousValues": {...}}}}，与你在订阅查询中声明的字段选择完全一致。

部署与使用注意事项

webhook 端点必须是可公网访问的 HTTP(S) 地址：Prisma 服务器会主动向该地址发起请求，本地localhost仅在 Prisma 与接收方同机部署时可用；
secret 与鉴权：推荐在prisma.yml顶层配置secret（如secret: ${env:PRISMA_SECRET}）保护 API 与订阅配置，并通过 webhook 的headers携带接收方所需的鉴权凭证（如Authorization: Bearer ...）；
payload 形状与字段选择一致：订阅查询中声明的字段决定了 webhook 负载内容；如需previousValues，必须在查询中显式声明该字段；
匹配失败不投递：订阅查询过滤条件未命中时不会产生 webhook 请求（源码层面由filtersMatch/variablesMatch与结果判空共同保证），因此不用担心无关变更带来的噪声流量；
交付通道现状：本仓库版本以 webhook 为唯一交付实现（WebhookDelivery分支），源码中保留了其他交付类型的兜底分支（case _ => Future.unit），为后续扩展 AWS Lambda 直接调用与队列实现预留了位置。

参考文档与源码索引

本文主体文档：docs/1.12/04-Reference/04-Server_side-Subscriptions/01-Overview.md
prisma.yml完整结构与subscriptions属性说明：docs/1.12/04-Reference/02-Service-Configuration/02-prisma.yml/02-YAML-Structure.md
prisma.yml概览与示例（含订阅与目录结构）：docs/1.12/04-Reference/02-Service-Configuration/02-prisma.yml/01-Overview-&-Example.md
CLI 订阅配置解析：cli/packages/prisma-yml/src/PrismaDefinition.ts
服务器侧执行与投递：server/servers/api/src/main/scala/com/prisma/api/mutactions/SideEffectMutactionExecutor.scala
订阅过滤器评估与执行：server/servers/api/src/main/scala/com/prisma/subscriptions/SubscriptionExecutor.scala
Webhook 数据结构：server/servers/api/src/main/scala/com/prisma/subscriptions/Webhook.scala
集成测试（行为验证）：server/servers/api/src/test/scala/com/prisma/subscriptions/EmbeddedServerSideSubscriptionSpec.scala