Hasura Event Triggers 实战用 Go 编写 AWS Lambda Webhook在数据库事件上执行 GraphQL Mutation【免费下载链接】graphql-engineBlazing fast, instant realtime GraphQL APIs on all your data with fine grained access control, also trigger webhooks on database events.项目地址: https://gitcode.com/gh_mirrors/gr/graphql-engine本指南基于 graphql-engine 仓库 中的 Go 语言 AWS Lambda 事件触发器模板mutation 示例讲解如何编写一个部署在 AWS Lambda 上的 Webhook当note表发生更新等数据库事件时由 Hasura 事件触发器Event Trigger把事件载荷 POST 到该 WebhookWebhook 再调用 Hasura GraphQL Engine 的 GraphQL Mutation 接口向note_revision表写入一条修订记录。读完本文你将掌握事件触发器的端到端搭建流程、Go 侧对事件载荷的解析方式、GraphQL Mutation 的构造与调用方法以及整个链路的验证手段。一、示例要解决的问题数据库事件驱动的数据联动在典型的异步后端架构中业务系统往往需要在「数据发生变化」时触发一系列后续动作例如写入审计日志、生成修订记录、发送通知、同步到外部索引等。Hasura 的事件触发器正是为此设计的它基于 Postgres 原生触发器捕获INSERT/UPDATE/DELETE操作并将事件以 HTTP 请求形式投递到开发者指定的 Webhook 端点见仓库根目录的 event-triggers.md。本示例的「mutation」场景非常直观note表保存便签本体note_revision表保存便签的每一次修订快照当note表的某行被更新时事件触发器把事件投递给 LambdaLambda 解析事件中的新旧数据取出note_id与新的note文本Lambda 调用 HGE 的 GraphQLinsert_note_revisionMutation插入一条修订记录从而自动实现「便签每次更新都留下历史版本」。该目录所在的 aws-lambda 父级 README 给出了此类模板的用途矩阵echo回显载荷、帮助理解事件结构与mutation在数据库事件上执行 GraphQL 变更两种用例在 Go、Python、Node.js、Ruby 等语言中均已实现本指南聚焦 Go 语言版本。二、数据表设计note 与 note_revision在 Hasura Console 中按如下结构创建两张表完整 SQL 结构见 README第一张表note保存便签内容表名: note 列: id Integer 自增 note Text 主键: id第二张表note_revision保存每次修订的历史记录表名: note_revision 列: id Integer 自增 note Text note_id Integer update_at Timestamp, 默认值: now() 主键: id其中note_id关联到note.idupdate_at默认取当前时间戳用于标记本次修订的生成时间。这两张表建好后Hasura 会自动为它们生成 GraphQL API包括根查询字段、以及本文会用到的插入根字段insert_note_revision和输入类型note_revision_insert_input——这正是后面 Lambda 中 GraphQL Mutation 语法的基础。三、Lambda 函数源码解析mutation.go示例的核心代码位于 mutation.go。下面按职责拆解其实现。3.1 事件载荷的结构化定义Hasura 投递到 Webhook 的请求体是 JSONGo 侧先定义与之对应的结构体把id、event、table、trigger四个顶层字段映射出来type HasuraEvent struct { ID string json:id Event json:event Table json:table Trigger json:trigger } type Event struct { Op string json:op Data json:data } type Data struct { Old map[string]interface{} json:old New map[string]interface{} json:new } type Table struct { Name string json:name Schema string json:schema } type Trigger struct { ID string json:id Name string json:name }其中event.op对应操作类型INSERT/UPDATE/DELETEevent.data.old与event.data.new分别保存变更前后的行数据。由于列名与取值在运行时才知道Old/New被声明为map[string]interface{}这是处理任意列集合的常见做法。源码细节提醒仓库中Data.Old的 JSON tag 写作json:old 含两个空格与事件载荷实际的old键不一致这会导致 UPDATE/DELETE 事件中旧值解析为空。实际部署时建议将其修正为json:old。同理解析请求体的一行json.Unmarshal([]byte(request.Body), body)传入的是值而非指针bodyencoding/json要求目标必须为指针否则会返回InvalidUnmarshalError并走入「无法解析事件」的错误分支——这两处都属于从源码结构可以观察到的细节部署前需要留意修正。3.2 GraphQL Mutation 的定义Webhook 要执行的变更被定义为一个字符串常量它直接对应 Hasura 自动生成的 Mutation APIconst MUTATION_UPDATE_NOTE_REVISION mutation updateNoteRevision ($object: note_revision_insert_input!) { insert_note_revision (objects: [$object]) { affected_rows returning { id } } } 要点说明$object: note_revision_insert_input!变量类型为 Hasura 为note_revision表自动生成的插入输入类型!表示必填insert_note_revision(objects: [$object])插入根字段objects参数接收一个对象数组返回选择集affected_rows与returning { id }前者表示受影响行数后者回传新插入行的主键。3.3 读取环境变量与校验输入Lambda 通过环境变量获得 HGE 端点var HGE_ENDPOINT os.Getenv(HGE_ENDPOINT)Handler中先检查端点是否配置再解析请求体、校验关键字段是否存在func Handler(request events.APIGatewayProxyRequest) (events.APIGatewayProxyResponse, error) { log.Printf(Processing Lambda request %v\n, request.Body) if len(HGE_ENDPOINT) 0 { return constructErrorResponse(map[string]string{ message: HGE Endpoint not defined in environment variable, }) } var body HasuraEvent err : json.Unmarshal([]byte(request.Body), body) event : body.Event if err ! nil { return constructErrorResponse(map[string]string{ message: Unable to parse Hasura Event, }) } note_id, ok : event.Data.Old[id] if !ok { return constructErrorResponse(map[string]string{ message: invalid payload: note id not found, }) } note, ok : event.Data.New[note] if !ok { return constructErrorResponse(map[string]string{ message: invalid payload: note not found, }) } // ... }校验逻辑从event.Data.Old中取便签主键id用于关联修订记录从event.Data.New中取更新后的note文本用于保存修订内容任一缺失都会返回明确的错误提示。3.4 构造 GraphQL 请求并调用 HGE校验通过后把 Mutation 语句与变量打包成 GraphQL 请求体通过标准库net/http以POST方式发送到 HGEpayload : map[string]interface{}{ query: MUTATION_UPDATE_NOTE_REVISION, variables: map[string]interface{}{ object: map[string]interface{}{ note_id: note_id.(float64), note: note.(string), }, }, } b : new(bytes.Buffer) json.NewEncoder(b).Encode(payload) res, err : http.Post(HGE_ENDPOINT, application/json; charsetutf-8, b) // ... var response map[string]interface{} err json.NewDecoder(res.Body).Decode(response) // ... return events.APIGatewayProxyResponse{ Body: string(responseBody), StatusCode: 200, }, nil这里有几个值得注意的实现细节note_id.(float64)JSON 中的数字经encoding/json解码后统一是float64类型因此做一次类型断言后再作为变量传入http.Post(HGE_ENDPOINT, ...)直接把 GraphQL 请求发往 HGE 的/v1/graphql端点当HGE_ENDPOINT配置为该路径时Content-Type为application/json; charsetutf-8若你的 HGE 开启了 Admin Secret还需在请求头中携带x-hasura-admin-secretPython 版本的 mutation.py 就通过ADMIN_SECRET环境变量做了示范Go 版本可自行扩展最终以events.APIGatewayProxyResponse返回 HGE 的响应体供 API Gateway 转发给事件触发器的投递机制记录。3.5 统一的错误响应构造constructErrorResponse把错误信息序列化成 JSON 返回并在序列化失败时回退为固定提示func constructErrorResponse(responsePayload map[string]string) (events.APIGatewayProxyResponse, error) { var responseBody []byte responseBody, err : json.Marshal(responsePayload) var statusCode int 200 if err ! nil { responseBody, _ json.Marshal(map[string]string{ message: Internal error ocurred while constructing error response, }) statusCode 500 } return events.APIGatewayProxyResponse{ Body: string(responseBody), StatusCode: statusCode, }, nil }入口处通过lambda.Start(Handler)将Handler注册为 AWS Lambda 的 Go 运行时入口函数。四、构建并打包为 Lambda 部署包目录下提供了 build.sh 一键构建脚本#!/bin/bash set -e echo Fetching dependencies go get . echo Building binary env GOOSlinux GOARCHamd64 go build echo Binary build complete. Zipping output. zip -j ./mutation.zip mutation echo Zip complete. You can now upload the zip file脚本做的事情依次是go get .拉取依赖主要是github.com/aws/aws-lambda-go的events与lambda包以GOOSlinux GOARCHamd64交叉编译出 Linux 可执行文件——这是 AWS Lambda 的 Go 运行时要求的目标平台与架构zip -j ./mutation.zip mutation把二进制压成mutation.zip-j表示剥离目录层级只打包文件本身供后续上传。五、在 AWS 控制台创建 Lambda 函数按 README 的步骤在 AWS 中创建 Webhook 函数登录 AWS Management Console打开 AWS Lambda 控制台在Get Started区域选择Create a function选择Author from scratch从头编写为 Lambda 指定Name将Runtime选为go 1.x在Role中选择Create new role from template(s)在Role name中为角色命名Policy Templates留空构建并上传代码i. 执行提供的 bash 脚本bash build.shii. 在 Lambda 控制台上传生成的mutation.zip在 Designer 面板中添加环境变量HGE_ENDPOINT值为 Hasura GraphQL Engine 的端点指向/v1/graphql点击save使配置生效创建一个 API并配置 Lambda 代理集成Lambda proxy integration使 API Gateway 能把事件触发器的请求透传给 Lambda、并把响应原样返回。依赖说明本示例需要 AWS 账号并开启计费同时需要一台已运行、可被 AWS 访问的 Hasura GraphQL Engine 实例前置条件详见 aws-lambda/README.md。六、在 Hasura 中创建事件触发器回到 Hasura Console 完成触发器的注册打开Events事件页签点击Create trigger新建触发器选择全部操作类型insert、update、delete将上一步 API Gateway 暴露的端点粘贴为Webhook URL。创建完成后任何对note表执行的操作都会触发一次事件投递Hasura 捕获数据库变更、生成事件载荷通过 HTTP POST 发送到你配置的 Webhook即 Lambda 的 API Gateway 端点Lambda 再反向调用 HGE 执行 GraphQL Mutation。七、测试触发器与验证结果在 Console 的Data页签中浏览note表Browse rows编辑一条既有记录触发UPDATE事件然后检查note_revision表是否新增了对应修订条目同时可在事件触发器详情页查看本次投递的请求与响应。7.1 触发载荷请求体README 中给出的更新事件载荷如下这是事件载荷的核心字段完整字段说明见下文第八节{ event: { op: UPDATE, data: { old: { note: note1, id: 1 }, new: { note: note1 updated, id: 1 } } }, created_at: 2018-10-02T06:38:22.67311Z, id: f57a1c79-72ba-4c19-8791-37d1b9616bcf, trigger: { name: note_revision_trigger, id: 5d85cbd1-c134-45ce-810c-7ecd3b4fc1ee }, table: { schema: public, name: note } }对照mutation.go的解析逻辑可以看到event.data.old.id提供note_idevent.data.new.note提供修订文本二者正是插入note_revision所需的数据来源。7.2 Webhook 响应Lambda 调用 HGE 的 Mutation 后把 HGE 的响应原样返回形如{ data: { insert_note_revision: { affected_rows: 1, returning: [ { id: 2 } ] } } }affected_rows: 1表明成功插入一行returning[0].id是新修订记录的自增主键与 Mutation 语句中声明的返回选择集一一对应。八、事件触发器载荷结构深入仓库的官方文档 docs/docs/event-triggers/payload.mdx 对事件载荷有更完整的字段说明。投递到 Webhook 的请求为Content-Type: application/json的 POST 请求载荷中的关键字段如下字段类型说明created_atString触发器调用的时间戳delivery_infoObject投递重试信息含current_retry与max_retrieseventObject事件本身及其相关数据event.data.newObject变更后的数据列为键、值为键值event.data.oldObject /null变更前的数据不适用时为nullevent.opString操作类型取值INSERT/UPDATE/DELETE/MANUALevent.session_variablesObject触发事件的用户的会话变量如x-hasura-role仅 Postgres 提供event.trace_contextObject追踪上下文含span_id与trace_ididString本次调用的 UUIDtableObject表信息含name与schematriggerObject触发器信息含name不同操作类型下old/new的取值规则INSERT时old为null、new为新行UPDATE时old为更新前、new为更新后DELETE时old为被删行、new为nullMANUAL手动触发时old为null、new为当前行。README 示例中的载荷即UPDATE场景。需要说明的是README 展示的是事件载荷的核心字段实际投递请求体的完整包装结构与 Hasura 版本有关字段定义以 payload.mdx 为准。九、延伸阅读与同类模板想先理解事件载荷本身可参考同目录的 echo 示例README它只做「回显载荷」一件事非常适合验证链路与调试Go 目录下的 README.md 总览了 echo 与 mutation 两个用例的定位差异同类场景在 Pythonmutation.py、Node.jsindex.js中也有实现可作为对照参考事件触发器的创建、投递与监控细节可继续阅读 create-trigger.mdx 与仓库根目录的 event-triggers.md。至此一条「数据库事件 → 事件触发器 → AWS LambdaGo→ GraphQL Mutation → 写入关联表」的完整链路已经打通。这套模式可以泛化到审计日志、修订历史、通知推送、数据同步等各类事件驱动场景只需要替换表结构、Mutation 语句与字段映射即可。【免费下载链接】graphql-engineBlazing fast, instant realtime GraphQL APIs on all your data with fine grained access control, also trigger webhooks on database events.项目地址: https://gitcode.com/gh_mirrors/gr/graphql-engine创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考