ryan
33b2a96bcf
Motivation: 不同来源（如 crypto-strategy 与 trade-signal）可能产生策略、币种、周期、方向、价格完全一致的信号，旧去重键仅按这五维计算，导致后到的不同来源信号被误判为重复而漏推，无法支撑跟单策略等新增来源并行推送。 Changes: * 信号去重键新增来源（source）维度，不同来源的相同信号不再互相去重 * 信号哈希计算时透传来源名称，同一来源内的重复消息仍正常去重 * 补充跟单策略文案模板的渲染测试与接口创建文档
135 lines
3.3 KiB
Go
135 lines
3.3 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 errText(err error) string {
|
|
if err == nil {
|
|
return ""
|
|
}
|
|
return err.Error()
|
|
}
|
|
|
|
func HandleMessage(ctx context.Context, in HandleInput, conv MessageConverter, lookup SourceLookup, process ProcessFunc) Disposition {
|
|
raw := string(in.Body)
|
|
event, data, err := conv.Convert(in.Body)
|
|
if err != nil {
|
|
hash := MessageHash(in.Body)
|
|
slog.Warn("invalid signal, ack", "hash", hash, "raw", raw, "error", errText(err))
|
|
return DispositionAck
|
|
}
|
|
|
|
owned := false
|
|
hash := SignalHash(in.SourceName, data)
|
|
if in.Deduper != nil {
|
|
ok, err := in.Deduper.Claim(ctx, hash)
|
|
if err != nil {
|
|
slog.Warn("dedup claim failed, processing anyway", "hash", hash, "error", errText(err))
|
|
} else if !ok {
|
|
slog.Info("duplicate message, ack", "hash", hash)
|
|
return DispositionAck
|
|
} else {
|
|
owned = true
|
|
}
|
|
}
|
|
|
|
slog.Info("mq message", "name", in.Name, "queue", in.Queue, "hash", hash, "raw", raw)
|
|
|
|
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", errText(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", errText(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", errText(relErr))
|
|
}
|
|
}
|
|
return disp
|
|
}
|