# AIAA Notification Service 纯 Go 实现的多渠道通知服务。上游系统通过 Webhook 推送消息,服务按来源解析、匹配规则、条件过滤、渲染模板后,异步分发到钉钉 / 企业微信 / 邮件 / Bark / SafeW。 ## 核心概念 ``` source 1──N rule template 1──N rule rule N──M channel (via rule_channel,可独立开关) ``` | 实体 | 说明 | |------|------| | **Source** | 来源系统,持有独立 `api_key`,定义 body 解析方式(json / regex / text) | | **Template** | Go `text/template` 模板 | | **Channel** | 发送渠道配置(钉钉 / 企微 / 邮件 / Bark / SafeW) | | **Rule** | 绑定 `source + event → template + channels`,可选条件过滤 | | **MessageLog** | 发送记录,便于排查 | **处理流程:** ``` POST /api/v1/notify → 鉴权(source api_key)+ 限流 → 按 parse_mode 解析 body → 匹配 enabled 规则 → 条件过滤(全部 AND) → 渲染模板 → 异步发送到启用渠道(失败重试 3 次:1s / 5s / 30s) → 立即返回 accepted ``` ## 技术栈 - Go 1.22+、Gin、sqlx、go-redis、Viper、slog - MySQL 8.0、Redis 7 - Docker 多阶段构建 ## 快速开始 ### 依赖 - Go 1.22+ - MySQL 8.0 - Redis 7 - [golang-migrate](https://github.com/golang-migrate/migrate)(本地迁移时) ### Docker Compose(推荐) ```bash docker-compose up -d ``` 会启动 MySQL、Redis 与 API(默认 `http://localhost:8080`)。 ### 本地开发 ```bash # 1. 启动依赖 docker-compose up -d mysql redis # 2. 迁移数据库 make migrate-up # 3. 修改配置(务必更换 admin_key) # config/config.yaml # 4. 运行 make run # 或 make build && ./bin/server ``` ### 常用命令 | 命令 | 说明 | |------|------| | `make run` | 本地启动 | | `make build` | 编译到 `bin/server` | | `make test` | 运行单元测试 | | `make migrate-up` / `migrate-down` | 数据库迁移 | | `make docker-up` / `docker-down` | Compose 启停 | ### 配置 主配置文件:`config/config.yaml`。支持 `${ENV_VAR}` / `${ENV_VAR:-default}` 插值,也可用 `NOTIFY_` 前缀环境变量覆盖(`.` → `_`,例如 `NOTIFY_DATABASE_HOST`)。 | 配置项 | 说明 | 默认 | |--------|------|------| | `server.port` | HTTP 端口 | `8080` | | `server.admin_key` | 管理 API Bearer Token | 须修改 | | `database.*` | MySQL 连接 | `notify` / `notification` | | `redis.*` | Redis(缓存 + 限流) | `127.0.0.1:6379` | | `smtp.*` | 邮件发送(email 渠道) | — | | `rate_limit.default` | 每 source 每秒请求上限 | `100` | | `rate_limit.dingtalk_per_min` | 同一钉钉机器人(access_token)每分钟发送上限;超限排队到下一分钟 | `18`(官方 20,留余量) | | `subscription_dedup_ttl` | 多队列重复消息(来源/策略/币种/周期/方向/价格)去重窗口 | `1h` | | `subscriptions` | RabbitMQ 订阅列表;某条 `url` 为空则跳过 | 空 | | `subscriptions[].source` | 对应已有 Source.name | 有 url 时必填 | | `subscriptions[].formatter` | 目前仅 `trade_signal` | `trade_signal` | 环境变量 `RABBITMQ_URL` 未设置时不启动消费,HTTP 通知不受影响。交易信号订阅需事先创建 Source(如 `trade-signal`)、模板(可用 `{{.formatted}}`)、规则 `trade.open` / `trade.add` / `trade.close` / `trade.reduce`、以及渠道。规则条件可用 `strategyCode` / `symbol` / `period`。跟单策略(B龙 `BLONG` 及后续)复用模板 `跟单策略`,规则 `trade.*` + `strategyCode`。巴菲特激进 `PUTEJJ` / 稳健 `PUTEWJ` 用模板 `巴菲特策略`(`rawMessage`),发到巴菲特三群,并过滤含「启动」的文案。 `crypto-strategy` 开仓(多/空,含原来的 `isSale` 空单)都映射为 `trade.open`,不再发 `trade.sell`。止盈(`isGain`)为 `trade.gain`,止损(`isClose` 且非 `isGain`)为 `trade.close`。高低分 `HLSS`、异动 `AMA`、波段 `BTS`、AG 趋势 `AGTS` 用各自前缀:`HLSS.open` / `AMA.open` / `BTS.close` / `AGTS.open` 等。同一 Source 允许多条相同 event 的规则(用条件区分);精确 event 优先于通配,条件通过的规则都会发送。现成模板与规则见 `docs/httpie/curls.md`。 健康检查:`GET /health` → `{"status":"ok"}` --- ## 认证 | 接口类型 | Header | 说明 | |----------|--------|------| | 通知接口 | `Authorization: Bearer ` | 创建 Source 时自动生成 | | 管理接口 | `Authorization: Bearer ` | 来自 `config.yaml` 的 `server.admin_key` | 未授权返回 `401`:`{"error":"unauthorized"}` 或 `{"error":"missing api key"}` / `{"error":"invalid api key"}`。 通知接口超限返回 `429`: ```json { "error": "rate_limit_exceeded", "message": "...", "retry_after": 1 } ``` --- ## 接口文档 Base URL:`http://localhost:8080` 通用错误体:`{"error": ""}` --- ### 1. 通知接口 #### `POST /api/v1/notify` 上游业务推送入口。Body 解析方式由 Source 的 `parse_mode` 决定。 **鉴权:** Source API Key **限流:** 按 source,默认 100 req/s(需 Redis) ##### JSON 模式(`parse_mode=json`) ```http POST /api/v1/notify Authorization: Bearer Content-Type: application/json { "event": "trade.open", "data": { "symbol": "BTC", "price": 65000 } } ``` - 必须包含 `event`(string) - `data` 为模板变量;若省略 `data`,则除 `event` 外的顶层字段作为 data ##### Regex 模式(`parse_mode=regex`) ```http POST /api/v1/notify Authorization: Bearer Content-Type: text/plain trade.open BTC 开仓 价格:65000 ``` - 使用 Source 的 `parse_pattern`(需含命名分组) - 命名分组 `event` 作为事件名,其余命名分组进入模板 data 示例 pattern:`(?P\w+)\s+(?P\w+)\s+(?P.+)` ##### Text 模式(`parse_mode=text`) ```http POST /api/v1/notify Authorization: Bearer Content-Type: text/plain BTC 突破 68000,请注意风险 ``` - body 原样透传,模板变量为 `.Body` - **固定 event 为 `default`**,规则需按 `event=default` 配置 ##### 成功响应 匹配并接受发送: ```json { "matched": true, "channels": ["dingtalk:1", "email:2"], "accepted": true } ``` 匹配但条件未通过: ```json { "matched": true, "filtered": true, "reason": "condition not met" } ``` 无匹配规则: ```json { "matched": false } ``` ##### 错误码 | HTTP | 场景 | |------|------| | 400 | 读取 body 失败 | | 401 | 鉴权失败 | | 422 | 解析失败 / 模板渲染失败 / 规则 conditions 非法 | | 429 | 限流 | | 500 | 模板缺失等内部错误 | 发送为异步:接口返回 `accepted` 不代表渠道侧已投递成功,请查 Message Log。 --- ### 2. Source(来源) 前缀:`/api/v1/sources` **鉴权:** Admin Key #### `POST /api/v1/sources` — 创建 ```json { "name": "trading-system", "parse_mode": "json", "parse_pattern": "", "status": 1 } ``` | 字段 | 类型 | 必填 | 说明 | |------|------|------|------| | `name` | string | 是 | 唯一名称 | | `parse_mode` | string | 否 | `json`(默认)/ `regex` / `text` | | `parse_pattern` | string | regex 时建议填 | 正则(命名分组) | | `status` | int | 否 | `1` 启用(默认)/ `0` 禁用 | **201** 返回完整 Source(含自动生成的 `api_key`)。名称冲突 → **409**。 Regex 示例: ```json { "name": "legacy-monitor", "parse_mode": "regex", "parse_pattern": "(?P\\w+)\\s+(?P.+)" } ``` #### `GET /api/v1/sources` — 列表 Query:`page`(默认 1)、`page_size`(默认 20) **200:** ```json { "data": [ /* Source[] */ ], "total": 10, "page": 1 } ``` #### `GET /api/v1/sources/:id` — 详情 **200** → Source;不存在 → **404** #### `PUT /api/v1/sources/:id` — 更新 Body 同创建。成功:`{"ok": true}` #### `DELETE /api/v1/sources/:id` — 删除 成功:`{"ok": true}` --- ### 3. Template(模板) 前缀:`/api/v1/templates` **鉴权:** Admin Key 模板语法为 Go `text/template`,变量来自解析后的 data。缺字段不再报错,按空值处理。 可用 `line` 把前缀和值包在一起:值为空则整行不输出(含前缀和换行)。 `case` 类似 switch:按值匹配成对的 key/文案,最后一个奇数参数是默认值。`trade.close` 能匹配 `.close` / `close` / `CLOSE`。 `replace` 做全文替换:`{{replace .rawMessage "Time:" "推送时间:"}}`。 ``` ### {{.symbol}} {{case .action "OPEN" "开仓" "CLOSE" "平仓" "GAIN" "止盈" "SELL" "卖出" "ADD" "加仓" "REDUCE" "减仓"}} {{line "币种" .symbol}}{{line "周期" .period}}{{line "方向" .side}}{{line "价格" .price}}{{line "平均价" .totalAvgPx}}{{line "止盈价" .takeProfitPrice}}{{line "止损价" .stopLossPrice}}{{line "推送时间" .pushedAt}} ``` 渲染时会自动注入 `event`、`pushedAt`(UTC+8,`2006.01.02 15:04:05`)。 也可用事件名:`{{case .event ".open" "开仓" ".close" "平仓"}}` 等价写法:`{{with .totalAvgPx}}平均价:{{.}}{{end}}` 示例: ``` ### {{.symbol}} 开仓 {{line "价格" .price}} ``` Text 模式示例:`{{.Body}}` #### `POST /api/v1/templates` — 创建 ```json { "name": "trade_open", "content": "### {{.symbol}} 开仓\n价格: {{.price}}" } ``` | 字段 | 类型 | 必填 | |------|------|------| | `name` | string | 是 | | `content` | string | 是 | **201** → Template;冲突 → **409** #### `GET /api/v1/templates` — 列表 Query:`page`、`page_size`(默认同 sources)。**200:** `{ "data": Template[], "total", "page" }` #### `GET /api/v1/templates/:id` #### `PUT /api/v1/templates/:id` / `DELETE /api/v1/templates/:id` 更新/删除成功返回 `{"ok": true}` --- ### 4. Channel(渠道) 前缀:`/api/v1/channels` **鉴权:** Admin Key #### `POST /api/v1/channels` — 创建 ```json { "name": "dingtalk-prod", "type": "dingtalk", "config": { ... }, "status": 1 } ``` | 字段 | 类型 | 必填 | 说明 | |------|------|------|------| | `name` | string | 是 | 唯一名称(规则里用此名引用) | | `type` | string | 是 | `dingtalk` / `wecom` / `email` / `bark` / `safew` | | `config` | object | 是 | 见下方各渠道配置 | | `status` | int | 否 | 默认 `1` | **201** → Channel;冲突 → **409** ##### 渠道 config **钉钉 `dingtalk`** ```json { "webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=xxx", "secret": "SEC..." } ``` `secret` 可选(加签机器人时填写)。消息类型:markdown。 **企业微信 `wecom`** ```json { "webhook_url": "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx" } ``` 消息类型:markdown。 **邮件 `email`** ```json { "to": ["ops@example.com", "dev@example.com"] } ``` SMTP 使用全局 `config.yaml` 的 `smtp` 段;支持 587 STARTTLS / 465 TLS。 **Bark `bark`** ```json { "url": "https://api.day.app/" } ``` **SafeW `safew`** ```json { "token": "", "chat_id": "123456789" } ``` `chat_id` 可为数字或字符串(含 `@username`)。消息以纯文本发送正文,不附带 `{source}: {event}` 标题,也不做 Markdown 转义。 #### `POST /api/v1/channels/safew/chats` — 列出已监控的 SafeW 群 创建渠道时选群。Admin Key。用 `getUpdates` 攒到 Redis 里的群;bot 尚未在群里收到过更新时列表为空。 ```json { "token": "", "q": "测试AI" } ``` `q` 可选,按群名 / username / chat_id 子串过滤(不区分大小写)。 **200:** ```json { "data": [ { "id": "10000778141", "type": "group", "title": "测试AI", "username": null } ], "total": 1 } ``` `id` 为字符串。Token 无效 → **401**。 #### `GET /api/v1/channels/:id/chats?q=` — 用已存 token 列群 编辑已有 safew 渠道。非 safew / 无 token → **400**;渠道不存在 → **404**。响应同上。 #### `GET /api/v1/channels` — 列表 Query:`page`、`page_size`。**200:** `{ "data": Channel[], "total", "page" }` #### `GET /api/v1/channels/:id` #### `PUT /api/v1/channels/:id` / `DELETE /api/v1/channels/:id` --- ### 5. Rule(规则) 前缀:`/api/v1/rules` **鉴权:** Admin Key 同一 Source 下允许重复 `event`;用规则条件和精确/通配优先级区分。条件通过的规则都会发送。 #### `POST /api/v1/rules` — 创建 ```json { "name": "trade-open-alert", "source_name": "trading-system", "event": "trade.open", "template_name": "trade_open", "channels": ["dingtalk-prod", "email-ops"], "conditions": [ {"field": "symbol", "op": "exists"}, {"field": "price", "op": "gt", "value": "0"} ], "enabled": 1 } ``` | 字段 | 类型 | 必填 | 说明 | |------|------|------|------| | `name` | string | 是 | 规则名称(代号),全局唯一 | | `source_name` | string | 是 | Source 名称 | | `event` | string | 是 | 事件名;支持 `*` / `?` 通配(如 `trade.*`)。精确匹配优先于通配,更具体的通配优先于 `*`。text 模式请用 `default` | | `template_name` | string | 是 | Template 名称 | | `channels` | string[] | 否 | Channel 名称列表 | | `conditions` | object[] | 否 | 过滤条件,全部 AND;省略则不过滤 | | `enabled` | int | 否 | 默认 `1` | **条件操作符:** | op | 说明 | |----|------| | `eq` / `ne` | 等于 / 不等于(字符串比较) | | `gt` / `gte` / `lt` / `lte` | 数值比较 | | `exists` / `not_exists` | 字段是否存在 | | `contains` / `not_contains` | 字符串包含 / 不包含 | 条件结构:`{"field":"<字段>","op":"<操作符>","value":"<可选>"}` **201** → Rule;冲突 → **409** #### `GET /api/v1/rules` — 列表 Query:`page`、`page_size`。**200:** `{ "data": Rule[], "total", "page" }` #### `GET /api/v1/rules/:id` #### `PUT /api/v1/rules/:id` — 更新 Body 同创建。成功:`{"ok": true}` #### `DELETE /api/v1/rules/:id` 成功:`{"ok": true}`(关联 `rule_channel` 级联删除) #### 开关 | 方法 | 路径 | 说明 | |------|------|------| | `PATCH` | `/api/v1/rules/:id/enable` | 启用规则 | | `PATCH` | `/api/v1/rules/:id/disable` | 禁用规则 | | `PATCH` | `/api/v1/rules/:id/channels/:channel_id/enable` | 启用该规则下某渠道 | | `PATCH` | `/api/v1/rules/:id/channels/:channel_id/disable` | 禁用该规则下某渠道 | 均返回 `{"ok": true}`。`:channel_id` 为渠道数字 ID。 --- ### 6. Message Log(消息记录) #### `GET /api/v1/message-logs` **鉴权:** Admin Key Query 参数: | 参数 | 说明 | |------|------| | `source` | 按来源名过滤 | | `event` | 按事件过滤 | | `status` | `pending` / `success` / `failed` / `retrying` | | `page` | 页码(从 1) | | `page_size` | 每页条数 | **200:** ```json { "data": [ /* MessageLog[] */ ], "total": 100, "page": 1 } ``` MessageLog 主要字段:`rule_id`、`channel_id`、`source`、`event`、`payload`、`content`、`status`、`retry_count`、`response`、`error_msg`、`created_at`。 --- ## 端到端示例 ```bash ADMIN="Authorization: Bearer admin-sk-change-me" # 1. 创建来源 curl -s -X POST http://localhost:8080/api/v1/sources \ -H "$ADMIN" -H "Content-Type: application/json" \ -d '{"name":"trading-system","parse_mode":"json"}' # 记下返回的 api_key # 2. 创建模板 curl -s -X POST http://localhost:8080/api/v1/templates \ -H "$ADMIN" -H "Content-Type: application/json" \ -d '{"name":"trade_open","content":"### {{.symbol}} 开仓\n价格: {{.price}}"}' # 3. 创建渠道 curl -s -X POST http://localhost:8080/api/v1/channels \ -H "$ADMIN" -H "Content-Type: application/json" \ -d '{ "name":"dingtalk-prod", "type":"dingtalk", "config":{"webhook_url":"https://oapi.dingtalk.com/robot/send?access_token=xxx","secret":"SEC..."} }' # 4. 创建规则 curl -s -X POST http://localhost:8080/api/v1/rules \ -H "$ADMIN" -H "Content-Type: application/json" \ -d '{ "name":"trade-open-alert", "source_name":"trading-system", "event":"trade.open", "template_name":"trade_open", "channels":["dingtalk-prod"], "conditions":[{"field":"symbol","op":"exists"}] }' # 5. 发送通知 curl -s -X POST http://localhost:8080/api/v1/notify \ -H "Authorization: Bearer " \ -H "Content-Type: application/json" \ -d '{"event":"trade.open","data":{"symbol":"BTC","price":65000}}' ``` --- ## 项目结构 ``` cmd/server/ 入口 config/ 配置文件 migrations/ MySQL 迁移 internal/ handler/ HTTP(鉴权、限流、CRUD、notify) engine/ 匹配 / 渲染 / 路由 adapter/ 渠道适配器(ChannelSender) parser/ json / regex / text condition/ 条件求值 store/ MySQL cache/ Redis model/ 数据模型 retry/ 重试 config/ 配置加载 ``` ## 扩展渠道 实现 `adapter.ChannelSender` 接口并在工厂中注册即可: ```go type ChannelSender interface { Type() string Send(title, content string, config json.RawMessage) error } ``` ## License 内部项目,按团队约定使用。