ryan
dc313fda13
Motivation: crypto-strategy 交易信号采用 envelope 外壳与字符串化/嵌套的 payload，原有 trade_signal 格式化器无法解析，需要独立的转换逻辑以生成通知文本和事件数据。 Changes: * 新增 crypto_strategy 消息转换器，解析 envelope 及嵌套、字符串化、扁平三种 payload 形态 * 根据平仓/止盈/卖出等标志推断交易动作，生成对应事件类型与中文通知文本 * 提取订单号、策略代码、价格、杠杆等字段用于通知数据 * 将消息转换器抽象为接口，订阅器按 formatter 配置选择对应实现 * 更新配置校验以支持 crypto_strategy 格式化器，并补充单元测试
131 lines
3.2 KiB
Go
131 lines
3.2 KiB
Go
package subscriber
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"log/slog"
|
|
"strconv"
|
|
|
|
"aiaa-notification-service/internal/model"
|
|
"aiaa-notification-service/internal/notify"
|
|
)
|
|
|
|
const retryHeader = "x-retry-count"
|
|
|
|
type Disposition int
|
|
|
|
const (
|
|
DispositionAck Disposition = iota
|
|
DispositionRetry
|
|
DispositionDLQ
|
|
)
|
|
|
|
type SourceLookup func(ctx context.Context, name string) (*model.Source, error)
|
|
|
|
type ProcessFunc func(ctx context.Context, req notify.Request) (notify.Result, error)
|
|
|
|
type HandleInput struct {
|
|
Body []byte
|
|
Headers map[string]any
|
|
Name string
|
|
Queue string
|
|
SourceName string
|
|
MaxRetry int
|
|
Deduper Deduper
|
|
}
|
|
|
|
func DecideRetry(retryCount, maxRetry int) Disposition {
|
|
if retryCount+1 > maxRetry {
|
|
return DispositionDLQ
|
|
}
|
|
return DispositionRetry
|
|
}
|
|
|
|
func RetryCount(headers map[string]any) int {
|
|
if headers == nil {
|
|
return 0
|
|
}
|
|
v, ok := headers[retryHeader]
|
|
if !ok {
|
|
return 0
|
|
}
|
|
switch n := v.(type) {
|
|
case int:
|
|
return n
|
|
case int32:
|
|
return int(n)
|
|
case int64:
|
|
return int(n)
|
|
case float64:
|
|
return int(n)
|
|
case string:
|
|
i, _ := strconv.Atoi(n)
|
|
return i
|
|
default:
|
|
return 0
|
|
}
|
|
}
|
|
|
|
type MessageConverter interface {
|
|
Convert(body []byte) (event string, data map[string]interface{}, err error)
|
|
}
|
|
|
|
func HandleMessage(ctx context.Context, in HandleInput, conv MessageConverter, lookup SourceLookup, process ProcessFunc) Disposition {
|
|
owned := false
|
|
hash := ""
|
|
if in.Deduper != nil {
|
|
hash = MessageHash(in.Body)
|
|
ok, err := in.Deduper.Claim(ctx, hash)
|
|
if err != nil {
|
|
slog.Warn("dedup claim failed, processing anyway", "hash", hash, "error", err)
|
|
} else if !ok {
|
|
slog.Info("duplicate message, ack", "hash", hash)
|
|
return DispositionAck
|
|
} else {
|
|
owned = true
|
|
}
|
|
}
|
|
|
|
raw := string(in.Body)
|
|
if hash == "" {
|
|
hash = MessageHash(in.Body)
|
|
}
|
|
slog.Info("mq message", "name", in.Name, "queue", in.Queue, "hash", hash, "raw", raw)
|
|
|
|
event, data, err := conv.Convert(in.Body)
|
|
if err != nil {
|
|
slog.Warn("invalid signal, ack", "hash", hash, "raw", raw, "error", err)
|
|
return DispositionAck
|
|
}
|
|
|
|
src, err := lookup(ctx, in.SourceName)
|
|
if err != nil || src == nil || src.Status != 1 {
|
|
slog.Warn("source unavailable, ack", "source", in.SourceName, "hash", hash, "raw", raw, "error", err)
|
|
return DispositionAck
|
|
}
|
|
|
|
res, err := process(ctx, notify.Request{Source: src, Event: event, Data: data})
|
|
if err == nil {
|
|
if !res.Matched {
|
|
slog.Info("no matching rule", "source", src.Name, "event", event, "hash", hash, "raw", raw)
|
|
} else if res.Filtered {
|
|
slog.Info("rule filtered", "source", src.Name, "event", event, "reason", res.Reason, "hash", hash, "raw", raw)
|
|
} else {
|
|
slog.Info("mq message accepted", "source", src.Name, "event", event, "channels", res.Channels, "hash", hash, "raw", raw)
|
|
}
|
|
return DispositionAck
|
|
}
|
|
if errors.Is(err, notify.ErrUnprocessable) {
|
|
slog.Warn("unprocessable notify, ack", "source", src.Name, "event", event, "hash", hash, "raw", raw, "error", err)
|
|
return DispositionAck
|
|
}
|
|
|
|
disp := DecideRetry(RetryCount(in.Headers), in.MaxRetry)
|
|
if owned && in.Deduper != nil {
|
|
if relErr := in.Deduper.Release(ctx, hash); relErr != nil {
|
|
slog.Warn("dedup release failed", "hash", hash, "error", relErr)
|
|
}
|
|
}
|
|
return disp
|
|
}
|