diff --git a/docs/superpowers/specs/2026-08-15-rabbitmq-subscriber-design.md b/docs/superpowers/specs/2026-08-15-rabbitmq-subscriber-design.md new file mode 100644 index 0000000..016f8e1 --- /dev/null +++ b/docs/superpowers/specs/2026-08-15-rabbitmq-subscriber-design.md @@ -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.`，如 `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 热更新订阅