docs: add RabbitMQ subscriber design spec
Lock the notify-service MQ module: configurable subscriptions, trade-signal formatting, and shared NotifyService.
This commit is contained in:
@@ -0,0 +1,208 @@
|
||||
# RabbitMQ 订阅模块设计
|
||||
|
||||
**Date:** 2026-08-15
|
||||
**Status:** Approved
|
||||
|
||||
## Goal
|
||||
|
||||
在通知服务内增加可配置的 RabbitMQ 订阅：消费交易信号、格式化文案、再走现有规则/模板/渠道发送。第一路订阅 fanout exchange `trade.signal.executor.queue`，与 executor 队列独立消费、互不争抢。配置做成订阅列表，后续加 exchange 只加配置、不改代码。
|
||||
|
||||
## Non-goals
|
||||
|
||||
- 移植参考项目的钉钉直发、按目标过滤、`x-dingtalk-sent` / `x-dingtalk-abandoned`
|
||||
- 在 MQ 层等待渠道发送结果
|
||||
- 均价持久化（进程内即可，重启丢失）
|
||||
- 连真实 CloudAMQP 的集成测试
|
||||
- 把 AMQP 账号写进仓库
|
||||
- 新增管理 API 或数据库表
|
||||
- 订阅侧解析 Source 的 `parse_mode`（MQ 路径自带 event + data）
|
||||
|
||||
## Decisions
|
||||
|
||||
| 项 | 选择 |
|
||||
|----|------|
|
||||
| 进 notify 的方式 | 抽出内部 `NotifyService`，MQ 与 HTTP 共用 |
|
||||
| 订阅 ↔ Source | 每条订阅绑定一个 `source` 名 |
|
||||
| event | `trade.<action小写>`，如 `OPEN` → `trade.open` |
|
||||
| 格式化 | 移植交易信号格式化器；`data.formatted` + 原始字段 |
|
||||
| 均价 / 策略覆盖 | 都移植；均价按 `strategyCode + symbol + side` 进程内追踪 |
|
||||
| 周期 | MQ `period` 原样使用（如 `1h`），不换算 |
|
||||
| 过滤 | 交给 Rule 条件（`strategyCode` / `symbol` 等） |
|
||||
| 凭据 | `url: "${RABBITMQ_URL}"` |
|
||||
|
||||
## Architecture
|
||||
|
||||
进程内 HTTP 与 MQ 消费并行。两者都进入同一个内部入口，规则、模板、渠道只维护一份。
|
||||
|
||||
```
|
||||
RabbitMQ fanout exchange
|
||||
│
|
||||
├─ executor 队列（已有，不改）
|
||||
└─ trade.signal.notify.queue（本服务独占）
|
||||
│
|
||||
▼
|
||||
Subscriber（声明/绑定/重连/消费/重试/DLQ）
|
||||
│
|
||||
▼
|
||||
trade_signal：解析 → 策略覆盖 → 均价 → 格式化
|
||||
│
|
||||
▼
|
||||
NotifyService
|
||||
匹配规则 → 条件过滤 → 渲染模板 → 异步发渠道
|
||||
```
|
||||
|
||||
订阅列表为空，或某条订阅的 `url` 未展开到非空值：该条不启动。全部未启动时 HTTP 通知不受影响。
|
||||
|
||||
## Config
|
||||
|
||||
`config/config.yaml` 增加 `subscriptions` 列表。第一路示例（URL 必须走环境变量）：
|
||||
|
||||
```yaml
|
||||
subscriptions:
|
||||
- name: trade-signal
|
||||
url: "${RABBITMQ_URL}"
|
||||
queue: trade.signal.notify.queue
|
||||
dead_letter_queue: trade.signal.notify.dlq
|
||||
exchange: trade.signal.executor.queue
|
||||
exchange_type: fanout
|
||||
routing_key: ""
|
||||
max_retry: 3
|
||||
source: trade-signal
|
||||
formatter: trade_signal
|
||||
strategy_overrides:
|
||||
BLONG:
|
||||
quantity_multipliers:
|
||||
open: 100
|
||||
add: 100
|
||||
reduce: 100
|
||||
close: 100
|
||||
leverage: 100
|
||||
```
|
||||
|
||||
| 字段 | 说明 |
|
||||
|------|------|
|
||||
| `name` | 订阅标识，日志用；缺省用 `queue` |
|
||||
| `url` | AMQP(S) URL，支持 `${ENV}` |
|
||||
| `queue` | 本服务独占队列 |
|
||||
| `dead_letter_queue` | 超过 `max_retry` 后投入；空则丢弃并 Ack |
|
||||
| `exchange` / `exchange_type` / `routing_key` | 声明并绑定；`exchange` 为空则只声明队列 |
|
||||
| `max_retry` | `Process` 内部错误的重投次数；`<=0` 时默认 3 |
|
||||
| `source` | 已有 `notification_source.name` |
|
||||
| `formatter` | 空则默认 `trade_signal`；未知值启动时报错 |
|
||||
| `strategy_overrides` | 按 `strategyCode`；数量倍数 `<=0` 视为 1 |
|
||||
|
||||
`exchange_type` 为空时默认 `fanout`。
|
||||
|
||||
## Modules
|
||||
|
||||
| 模块 | 职责 | 接口 |
|
||||
|------|------|------|
|
||||
| `internal/notify` | 匹配规则、条件、渲染、路由、记 message_log | `Process(ctx, Request) (Result, error)` |
|
||||
| `internal/subscriber` | 连 MQ、声明/绑定、重连、重试/DLQ；按订阅启动 goroutine | `Run(ctx)`，依赖 `Process` |
|
||||
| `internal/subscriber/tradesignal` | 解析信号、策略覆盖、均价、格式化 | 输入 JSON body，输出 `event` + `data` |
|
||||
|
||||
`POST /api/v1/notify` 按 Source `parse_mode` 解析 body 后调用 `notify.Process`，不再内嵌匹配/渲染/路由。
|
||||
|
||||
`subscriber` 不碰渠道，不读 `parse_mode`。
|
||||
|
||||
### NotifyService
|
||||
|
||||
```go
|
||||
type Request struct {
|
||||
Source *model.Source
|
||||
Event string
|
||||
Data map[string]interface{}
|
||||
}
|
||||
|
||||
type Result struct {
|
||||
Matched bool
|
||||
Filtered bool
|
||||
Channels []string
|
||||
Reason string
|
||||
}
|
||||
```
|
||||
|
||||
行为与现有 Handler 一致：无规则 → `Matched=false`；条件未过 → `Matched=true, Filtered=true`；命中则渲染、`Route`、写 log。标题仍为 `{source.name}: {event}`。`Process` 只在查库/渲染等内部失败时返回 `error`。
|
||||
|
||||
### Source 查找
|
||||
|
||||
订阅用 `source` 名调用已有 `GetSourceByName`。`status != 1` 或未找到：Ack，记日志，不进 DLQ。
|
||||
|
||||
## Data flow（trade_signal）
|
||||
|
||||
1. 消费 delivery，JSON 反序列化为信号。
|
||||
2. 按 `strategyCode` 套 `strategy_overrides`（数量倍数、杠杆覆盖）。
|
||||
3. 按 `strategyCode + symbol + side` 更新进程内均价；同一 `signalId` 只应用一次。
|
||||
4. 生成 `formatted` 文本。
|
||||
5. `event = "trade." + strings.ToLower(action)`。`action` 为空视为无效消息。
|
||||
6. 用订阅的 `source` 名取启用 Source，调用 `Process`。
|
||||
|
||||
### `data` 字段
|
||||
|
||||
`data` 来自**覆盖后**的信号（与 `formatted` 数字一致），camelCase，并附加：
|
||||
|
||||
| 键 | 来源 |
|
||||
|----|------|
|
||||
| `signalId` / `sourcePosId` / `strategyCode` / `symbol` / `side` / `action` | 信号 |
|
||||
| `quantity` / `amountMarginRatio` / `posMarginRatio` | 信号（quantity 可能被倍数改写） |
|
||||
| `price` / `leverage` / `period` / `eventTime` | 信号（leverage 可能被覆盖） |
|
||||
| `takeProfitPrice` / `stopLossPrice` / `takeProfitRatio` / `stopLossRatio` | 信号 |
|
||||
| `pnl` / `accountBalance` | 信号（可选） |
|
||||
| `formatted` | 格式化全文 |
|
||||
| `avgPrice` | 有均价才写入 |
|
||||
|
||||
规则条件可写 `strategyCode`、`symbol`、`period` 等。模板可用 `{{.formatted}}`，也可自己拼字段。
|
||||
|
||||
### 格式化文案
|
||||
|
||||
移植参考项目 `test-mq-to-ali` 的格式化器，并在「交易品种」后增加周期（`period` 非空才输出）：
|
||||
|
||||
```
|
||||
多单开仓
|
||||
交易品种: BTC
|
||||
周期: 1h
|
||||
开仓价格: 65000.00
|
||||
...
|
||||
策略: BLONG
|
||||
Time: 2026.08.15 16:39:00
|
||||
```
|
||||
|
||||
其余规则与参考项目一致：品种去掉报价后缀（`BTCUSDT` → `BTC`）；开/加/减/平仓标题与数量或比例行；可选均价、杠杆、止盈止损、盈亏、余额；时间为本地时区 `2006.01.02 15:04:05`。
|
||||
|
||||
## Error handling
|
||||
|
||||
| 情况 | 处理 |
|
||||
|------|------|
|
||||
| JSON 无效、缺 `action` | Ack 丢掉，不重试 |
|
||||
| Source 不存在/禁用、无匹配规则、条件未过 | Ack，记日志，不进 DLQ |
|
||||
| `Process` 返回 error（查库失败等） | `x-retry-count` +1 后重新投递本队列；超过 `max_retry` 则投入 `dead_letter_queue`（未配置则丢弃并 Ack） |
|
||||
| 连接/channel 断开 | 5s 后重连，重新 `consumeOnce` |
|
||||
|
||||
启动时声明 durable exchange（若配置了）、durable 队列、绑定、durable DLQ。消费 `autoAck=false`，QoS prefetch=1。consumer tag 用订阅 `name`。
|
||||
|
||||
渠道发送仍由 `engine.Router` 异步重试（1s / 5s / 30s）。MQ 在 `Process` 接受路由后即 Ack，不等待渠道结果。
|
||||
|
||||
不移植参考项目按钉钉目标的部分成功重试。
|
||||
|
||||
## Testing
|
||||
|
||||
- **格式化**：开/平/加/减仓文案；`period` 有则出现、空则省略；均价行；策略倍数后的数量
|
||||
- **均价追踪**：OPEN/ADD 更新，REDUCE/CLOSE 展示原均价；同 `signalId` 不重复应用
|
||||
- **策略覆盖**：数量倍数、杠杆覆盖；`<=0` 的倍数视为 1
|
||||
- **订阅 → Notify**：假 `Process`，断言 `event`、`data.formatted`、原始字段与 `period`
|
||||
- **重试/DLQ**：超过 `max_retry` 才进 DLQ；坏 JSON 直接 Ack
|
||||
- **HTTP `/notify`**：抽 `Process` 后现有解析与匹配行为不变
|
||||
|
||||
不写连真实 broker 的测试。声明/绑定逻辑用可注入的 channel 假对象，或抽纯函数测重试计数与 DLQ 判定。
|
||||
|
||||
## Boot
|
||||
|
||||
`cmd/server/main.go` 在 HTTP server 启动后、等信号退出前，对每条有效订阅 `go subscriber.Run(ctx)`。收到 SIGINT/SIGTERM 时 cancel 该 ctx，再 `Shutdown` HTTP。
|
||||
|
||||
运营侧需事先创建 Source（`name` 与订阅 `source` 一致）、Template、Rule（`event` 为 `trade.open` 等）、Channel。本模块不自动建这些记录。
|
||||
|
||||
## Out of scope later
|
||||
|
||||
- 第二种 `formatter`（有新 exchange 再加）
|
||||
- 均价写入 Redis
|
||||
- 管理 API 热更新订阅
|
||||
Reference in New Issue
Block a user