From d2e8476398e476ba492673a1972dc64fd67439f9 Mon Sep 17 00:00:00 2001 From: ryan Date: Mon, 17 Aug 2026 01:20:49 +0800 Subject: [PATCH] =?UTF-8?q?feat(=E8=A7=84=E5=88=99):=20=E6=94=AF=E6=8C=81?= =?UTF-8?q?=E5=90=8C=E4=B8=80=E4=BA=8B=E4=BB=B6=E5=A4=9A=E8=A7=84=E5=88=99?= =?UTF-8?q?=E6=8C=89=E6=9D=A1=E4=BB=B6=E5=8C=BA=E5=88=86=E5=B9=B6=E5=85=A8?= =?UTF-8?q?=E9=83=A8=E5=8F=91=E9=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Motivation: 止盈、止损等不同策略信号会映射到同一事件（如 trade.close），但原有唯一约束要求每个事件只能有一条规则，无法按策略区分处理。放开该约束后，同一事件可配置多条规则，通过规则条件与精确/通配优先级区分，命中条件的规则全部发送；同时将去重键从消息原文改为信号维度，避免同一信号因时间戳等无关字段差异被误判为重复。 Changes: * 移除规则 source_id+event 的唯一约束，改为普通索引 * 事件匹配改为返回命中优先级内所有启用规则，并按 ID 逐条派发 * 通知服务遍历多条规则，按条件过滤后聚合发送渠道 * 去重键由消息 body 哈希改为策略/币种/周期/方向/价格信号维度哈希 --- README.md | 6 +- config/config.yaml | 2 +- internal/engine/event_match.go | 39 +++++--- internal/engine/event_match_test.go | 29 ++++++ internal/engine/matcher.go | 60 +++---------- internal/notify/service.go | 89 ++++++++++++------- internal/notify/service_test.go | 81 +++++++++++++++-- internal/store/rule.go | 2 +- internal/subscriber/dedup.go | 50 +++++++++++ internal/subscriber/dedup_test.go | 79 ++++++++++++++++ internal/subscriber/handle.go | 21 ++--- migrations/003_rule_event_not_unique.down.sql | 3 + migrations/003_rule_event_not_unique.up.sql | 3 + 13 files changed, 349 insertions(+), 115 deletions(-) create mode 100644 migrations/003_rule_event_not_unique.down.sql create mode 100644 migrations/003_rule_event_not_unique.up.sql diff --git a/README.md b/README.md index 01cf38a..4953ab0 100644 --- a/README.md +++ b/README.md @@ -95,14 +95,14 @@ make build && ./bin/server | `smtp.*` | 邮件发送（email 渠道） | — | | `rate_limit.default` | 每 source 每秒请求上限 | `100` | | `rate_limit.dingtalk_per_min` | 同一钉钉机器人（access_token）每分钟发送上限；超限排队到下一分钟 | `18`（官方 20，留余量） | -| `subscription_dedup_ttl` | 多队列重复消息（body SHA-256）去重窗口 | `1h` | +| `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`。 -`crypto-strategy` 开仓（多/空，含原来的 `isSale` 空单）都映射为 `trade.open`，不再发 `trade.sell`。止盈（`isGain`）为 `trade.gain`，止损（`isClose` 且非 `isGain`）为 `trade.close`。高低分 `HLSS` 仍用 `HLSS.open` / `HLSS.sell` / `HLSS.close`。同一 Source 每个 event 只能有一条规则，所以止盈不能再和止损共用 `trade.close`。AI crypto signals 的现成模板与规则见 `docs/httpie/curls.md`。 +`crypto-strategy` 开仓（多/空，含原来的 `isSale` 空单）都映射为 `trade.open`，不再发 `trade.sell`。止盈（`isGain`）为 `trade.gain`，止损（`isClose` 且非 `isGain`）为 `trade.close`。高低分 `HLSS` 仍用 `HLSS.open` / `HLSS.sell` / `HLSS.close`。同一 Source 允许多条相同 event 的规则（用条件区分）；精确 event 优先于通配，条件通过的规则都会发送。AI crypto signals 的现成模板与规则见 `docs/httpie/curls.md`。 健康检查：`GET /health` → `{"status":"ok"}` @@ -482,7 +482,7 @@ Query：`page`、`page_size`。**200：** `{ "data": Channel[], "total", "page" 前缀：`/api/v1/rules` **鉴权：** Admin Key -同一 Source 下 `event` 唯一。 +同一 Source 下允许重复 `event`；用规则条件和精确/通配优先级区分。条件通过的规则都会发送。 #### `POST /api/v1/rules` — 创建 diff --git a/config/config.yaml b/config/config.yaml index 44fb07c..a008651 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -34,7 +34,7 @@ logbull: api_key: "lb_60701971723797ed0374aa3896078fe5" log_level: "INFO" -# 多队列重复消息按 body SHA-256 去重；有 Redis 时跨进程共享 +# 多队列重复消息按策略/币种/周期/方向/价格去重；有 Redis 时跨进程共享 subscription_dedup_ttl: 1h subscriptions: diff --git a/internal/engine/event_match.go b/internal/engine/event_match.go index 85bbe37..ef4877f 100644 --- a/internal/engine/event_match.go +++ b/internal/engine/event_match.go @@ -11,18 +11,33 @@ import ( // Exact event wins; otherwise glob patterns (* and ?) via path.Match. // Among globs, more literal characters win; ties go to the smaller ID. func PickEventRule(event string, rules []model.Rule) *model.Rule { - var exact *model.Rule - var best *model.Rule + picked := PickEventRules(event, rules) + if len(picked) == 0 { + return nil + } + best := &picked[0] + for i := range picked[1:] { + if picked[i+1].ID < best.ID { + best = &picked[i+1] + } + } + return best +} + +// PickEventRules returns every enabled rule at the winning specificity. +// All exact event matches win as a group; otherwise all globs that share +// the highest literal-character score. +func PickEventRules(event string, rules []model.Rule) []model.Rule { + var exact []model.Rule + var globs []model.Rule bestScore := -1 for i := range rules { - r := &rules[i] + r := rules[i] if r.Enabled == 0 { continue } if r.Event == event { - if exact == nil || r.ID < exact.ID { - exact = r - } + exact = append(exact, r) continue } if !strings.ContainsAny(r.Event, "*?") { @@ -33,15 +48,19 @@ func PickEventRule(event string, rules []model.Rule) *model.Rule { continue } score := globSpecificity(r.Event) - if score > bestScore || (score == bestScore && (best == nil || r.ID < best.ID)) { + if score > bestScore { bestScore = score - best = r + globs = []model.Rule{r} + continue + } + if score == bestScore { + globs = append(globs, r) } } - if exact != nil { + if len(exact) > 0 { return exact } - return best + return globs } func globSpecificity(pattern string) int { diff --git a/internal/engine/event_match_test.go b/internal/engine/event_match_test.go index 902f03b..29d5359 100644 --- a/internal/engine/event_match_test.go +++ b/internal/engine/event_match_test.go @@ -60,3 +60,32 @@ func TestPickEventRuleNoMatch(t *testing.T) { t.Fatal("expected no match") } } + +func TestPickEventRulesAllExactMatches(t *testing.T) { + rules := []model.Rule{ + {ID: 12, Event: "trade.*", Enabled: 1}, + {ID: 17, Event: "trade.close", Enabled: 1}, + {ID: 18, Event: "trade.close", Enabled: 1}, + {ID: 19, Event: "trade.close", Enabled: 0}, + } + got := PickEventRules("trade.close", rules) + if len(got) != 2 { + t.Fatalf("want 2 exact trade.close, got %#v", got) + } + ids := []int{got[0].ID, got[1].ID} + if ids[0] != 17 || ids[1] != 18 { + t.Fatalf("ids=%v", ids) + } +} + +func TestPickEventRulesSameGlobBoth(t *testing.T) { + rules := []model.Rule{ + {ID: 1, Event: "trade.*", Enabled: 1}, + {ID: 2, Event: "trade.*", Enabled: 1}, + {ID: 3, Event: "*", Enabled: 1}, + } + got := PickEventRules("trade.open", rules) + if len(got) != 2 { + t.Fatalf("want both trade.*, got %#v", got) + } +} diff --git a/internal/engine/matcher.go b/internal/engine/matcher.go index 9fc8c90..e264dd7 100644 --- a/internal/engine/matcher.go +++ b/internal/engine/matcher.go @@ -2,7 +2,6 @@ package engine import ( "context" - "encoding/json" "fmt" "aiaa-notification-service/internal/cache" @@ -12,59 +11,20 @@ import ( type Matcher struct { store *store.Store - cache *cache.Cache } -func NewMatcher(s *store.Store, c *cache.Cache) *Matcher { - return &Matcher{store: s, cache: c} +func NewMatcher(s *store.Store, _ *cache.Cache) *Matcher { + return &Matcher{store: s} } -func (m *Matcher) Match(ctx context.Context, sourceID int, event string) (*model.Rule, error) { - // Try cache first - if m.cache != nil { - cr, err := m.cache.GetRule(ctx, sourceID, event) - if err == nil { - rule := &model.Rule{ID: cr.RuleID, TemplateID: cr.TemplateID, SourceID: sourceID, Event: event} - if cr.Conditions != "" && cr.Conditions != "null" { - raw := json.RawMessage(cr.Conditions) - rule.Conditions = &raw - } - return rule, nil - } - } - - rule, err := m.store.GetRuleBySourceEvent(ctx, sourceID, event) +func (m *Matcher) Match(ctx context.Context, sourceID int, event string) ([]model.Rule, error) { + rules, err := m.store.ListEnabledRulesBySource(ctx, sourceID) if err != nil { - rules, listErr := m.store.ListEnabledRulesBySource(ctx, sourceID) - if listErr != nil { - return nil, fmt.Errorf("match rule: %w", listErr) - } - picked := PickEventRule(event, rules) - if picked == nil { - return nil, fmt.Errorf("match rule: %w", err) - } - return picked, nil + return nil, fmt.Errorf("match rule: %w", err) } - - m.warmRuleCache(ctx, sourceID, event, rule) - return rule, nil -} - -func (m *Matcher) warmRuleCache(ctx context.Context, sourceID int, event string, rule *model.Rule) { - if m.cache == nil { - return - } - tmpl, err := m.store.GetTemplate(ctx, rule.TemplateID) - if err != nil { - return - } - cr := &cache.CachedRule{ - RuleID: rule.ID, - TemplateID: rule.TemplateID, - Content: tmpl.Content, - } - if rule.Conditions != nil { - cr.Conditions = string(*rule.Conditions) - } - _ = m.cache.SetRule(ctx, sourceID, event, cr) + picked := PickEventRules(event, rules) + if len(picked) == 0 { + return nil, fmt.Errorf("match rule: no rule for source %d event %s", sourceID, event) + } + return picked, nil } diff --git a/internal/notify/service.go b/internal/notify/service.go index ccb7d7c..c1139e4 100644 --- a/internal/notify/service.go +++ b/internal/notify/service.go @@ -31,7 +31,7 @@ type Result struct { } type RuleMatcher interface { - Match(ctx context.Context, sourceID int, event string) (*model.Rule, error) + Match(ctx context.Context, sourceID int, event string) ([]model.Rule, error) } type TemplateStore interface { @@ -59,27 +59,11 @@ func NewService(m RuleMatcher, t TemplateStore, r *engine.Renderer, rt ChannelRo } func (s *Service) Process(ctx context.Context, req Request) (Result, error) { - rule, err := s.matcher.Match(ctx, req.Source.ID, req.Event) - if err != nil { + rules, err := s.matcher.Match(ctx, req.Source.ID, req.Event) + if err != nil || len(rules) == 0 { return Result{Matched: false}, nil } - if rule.Conditions != nil { - var conds []model.Condition - if err := json.Unmarshal(*rule.Conditions, &conds); err != nil { - slog.Error("failed to unmarshal rule conditions", "rule_id", rule.ID, "error", err) - return Result{}, fmt.Errorf("%w: invalid rule conditions", ErrUnprocessable) - } - if !condition.Evaluate(conds, req.Data) { - return Result{Matched: true, Filtered: true, Reason: "condition not met"}, nil - } - } - - tmpl, err := s.templates.GetTemplate(ctx, rule.TemplateID) - if err != nil { - return Result{}, fmt.Errorf("template not found") - } - if req.Data == nil { req.Data = map[string]interface{}{} } @@ -90,24 +74,70 @@ func (s *Service) Process(ctx context.Context, req Request) (Result, error) { req.Data["pushedAt"] = time.Now().In(time.Local).Format("2006.01.02 15:04:05") } + var channels []string + accepted := 0 + for i := range rules { + chs, filtered, err := s.dispatch(ctx, req, &rules[i]) + if err != nil { + return Result{}, err + } + if filtered { + continue + } + accepted++ + channels = append(channels, chs...) + } + if accepted == 0 { + return Result{Matched: true, Filtered: true, Reason: "condition not met"}, nil + } + + slog.Info("notification accepted", + "source", req.Source.Name, + "event", req.Event, + "channels", channels, + ) + return Result{Matched: true, Channels: channels}, nil +} + +func (s *Service) dispatch(ctx context.Context, req Request, rule *model.Rule) ([]string, bool, error) { + if rule.Conditions != nil { + var conds []model.Condition + if err := json.Unmarshal(*rule.Conditions, &conds); err != nil { + slog.Error("failed to unmarshal rule conditions", "rule_id", rule.ID, "error", err) + return nil, false, fmt.Errorf("%w: invalid rule conditions", ErrUnprocessable) + } + if !condition.Evaluate(conds, req.Data) { + return nil, true, nil + } + } + + tmpl, err := s.templates.GetTemplate(ctx, rule.TemplateID) + if err != nil { + return nil, false, fmt.Errorf("template not found") + } + content, err := s.renderer.Render(tmpl.Content, req.Data) if err != nil { - return Result{}, fmt.Errorf("%w: template render failed: %s", ErrUnprocessable, err.Error()) + return nil, false, fmt.Errorf("%w: template render failed: %s", ErrUnprocessable, err.Error()) } title := req.Source.Name + ": " + req.Event channels := s.router.Route(ctx, rule, title, content) if s.logs != nil { + ruleID := rule.ID + srcName := req.Source.Name + event := req.Event + payloadJSON, _ := json.Marshal(req.Data) + chs := append([]string(nil), channels...) go func() { - payloadJSON, _ := json.Marshal(req.Data) logCtx := context.Background() - for _, chName := range channels { + for _, chName := range chs { ml := &model.MessageLog{ - RuleID: rule.ID, + RuleID: ruleID, ChannelID: parseChannelID(chName), - Source: req.Source.Name, - Event: req.Event, + Source: srcName, + Event: event, Payload: payloadJSON, Content: content, Status: "pending", @@ -118,14 +148,7 @@ func (s *Service) Process(ctx context.Context, req Request) (Result, error) { } }() } - - slog.Info("notification accepted", - "source", req.Source.Name, - "event", req.Event, - "channels", channels, - ) - - return Result{Matched: true, Channels: channels}, nil + return channels, false, nil } func parseChannelID(chName string) int { diff --git a/internal/notify/service_test.go b/internal/notify/service_test.go index a185111..9e42d5f 100644 --- a/internal/notify/service_test.go +++ b/internal/notify/service_test.go @@ -12,12 +12,22 @@ import ( ) type fakeMatcher struct { - rule *model.Rule - err error + rule *model.Rule + rules []model.Rule + err error } -func (f *fakeMatcher) Match(context.Context, int, string) (*model.Rule, error) { - return f.rule, f.err +func (f *fakeMatcher) Match(context.Context, int, string) ([]model.Rule, error) { + if f.err != nil { + return nil, f.err + } + if len(f.rules) > 0 { + return f.rules, nil + } + if f.rule != nil { + return []model.Rule{*f.rule}, nil + } + return nil, nil } type fakeTemplates struct { @@ -33,11 +43,15 @@ type fakeRouter struct { channels []string title string content string + ruleIDs []int } -func (f *fakeRouter) Route(_ context.Context, _ *model.Rule, title, content string) []string { +func (f *fakeRouter) Route(_ context.Context, rule *model.Rule, title, content string) []string { f.title = title f.content = content + if rule != nil { + f.ruleIDs = append(f.ruleIDs, rule.ID) + } return f.channels } @@ -165,3 +179,60 @@ func TestProcessTemplateMissing(t *testing.T) { t.Fatalf("want retryable error, got %v", err) } } + +func TestProcessMultipleRulesSameEvent(t *testing.T) { + rt := &fakeRouter{channels: []string{"safew:1"}} + svc := newSvc( + &fakeMatcher{rules: []model.Rule{ + {ID: 17, TemplateID: 1, Event: "trade.close"}, + {ID: 18, TemplateID: 1, Event: "trade.close"}, + }}, + &fakeTemplates{tmpl: &model.Template{ID: 1, Content: "ok"}}, + rt, + ) + res, err := svc.Process(context.Background(), Request{ + Source: &model.Source{ID: 11, Name: "crypto-strategy"}, + Event: "trade.close", + Data: map[string]interface{}{"symbol": "CRV"}, + }) + if err != nil { + t.Fatal(err) + } + if !res.Matched || res.Filtered { + t.Fatalf("%+v", res) + } + if len(rt.ruleIDs) != 2 || rt.ruleIDs[0] != 17 || rt.ruleIDs[1] != 18 { + t.Fatalf("routed=%v", rt.ruleIDs) + } + if len(res.Channels) != 2 { + t.Fatalf("channels=%v", res.Channels) + } +} + +func TestProcessSkipsFilteredSiblingRule(t *testing.T) { + hlss := json.RawMessage(`[{"field":"strategyCode","op":"eq","value":"HLSS"}]`) + ai := json.RawMessage(`[{"field":"strategyCode","op":"eq","value":"ai-crypto-signals"}]`) + rt := &fakeRouter{channels: []string{"safew:1"}} + svc := newSvc( + &fakeMatcher{rules: []model.Rule{ + {ID: 1, TemplateID: 1, Event: "trade.close", Conditions: &hlss}, + {ID: 2, TemplateID: 1, Event: "trade.close", Conditions: &ai}, + }}, + &fakeTemplates{tmpl: &model.Template{ID: 1, Content: "ok"}}, + rt, + ) + res, err := svc.Process(context.Background(), Request{ + Source: &model.Source{ID: 11, Name: "crypto-strategy"}, + Event: "trade.close", + Data: map[string]interface{}{"strategyCode": "ai-crypto-signals"}, + }) + if err != nil { + t.Fatal(err) + } + if !res.Matched || res.Filtered { + t.Fatalf("%+v", res) + } + if len(rt.ruleIDs) != 1 || rt.ruleIDs[0] != 2 { + t.Fatalf("routed=%v want only rule 2", rt.ruleIDs) + } +} diff --git a/internal/store/rule.go b/internal/store/rule.go index 50d4863..9793b6a 100644 --- a/internal/store/rule.go +++ b/internal/store/rule.go @@ -54,7 +54,7 @@ func (s *Store) GetRule(ctx context.Context, id int) (*model.Rule, error) { func (s *Store) GetRuleBySourceEvent(ctx context.Context, sourceID int, event string) (*model.Rule, error) { var r model.Rule var condsBytes []byte - query := `SELECT id, name, source_id, event, template_id, conditions, enabled, created_at, updated_at FROM notification_rule WHERE source_id = ? AND event = ? AND enabled = 1` + query := `SELECT id, name, source_id, event, template_id, conditions, enabled, created_at, updated_at FROM notification_rule WHERE source_id = ? AND event = ? AND enabled = 1 ORDER BY id LIMIT 1` row := s.DB.QueryRowContext(ctx, query, sourceID, event) if err := row.Scan(&r.ID, &r.Name, &r.SourceID, &r.Event, &r.TemplateID, &condsBytes, &r.Enabled, &r.CreatedAt, &r.UpdatedAt); err != nil { return nil, fmt.Errorf("get rule by source+event: %w", err) diff --git a/internal/subscriber/dedup.go b/internal/subscriber/dedup.go index 06e9eae..5189678 100644 --- a/internal/subscriber/dedup.go +++ b/internal/subscriber/dedup.go @@ -4,6 +4,10 @@ import ( "context" "crypto/sha256" "encoding/hex" + "encoding/json" + "fmt" + "strconv" + "strings" "sync" "time" @@ -20,6 +24,52 @@ func MessageHash(body []byte) string { return hex.EncodeToString(sum[:]) } +func SignalHash(data map[string]interface{}) string { + sum := sha256.Sum256([]byte(signalKey(data))) + return hex.EncodeToString(sum[:]) +} + +func signalKey(data map[string]interface{}) string { + if data == nil { + data = map[string]interface{}{} + } + strategy := fieldString(data["strategyCode"]) + symbol := firstNonEmptyField(fieldString(data["symbol"]), fieldString(data["currency"])) + period := fieldString(data["period"]) + direction := strings.ToUpper(firstNonEmptyField(fieldString(data["direction"]), fieldString(data["side"]))) + price := fieldString(data["price"]) + return strings.Join([]string{strategy, symbol, period, direction, price}, "\x1f") +} + +func firstNonEmptyField(a, b string) string { + if a != "" { + return a + } + return b +} + +func fieldString(v any) string { + if v == nil { + return "" + } + switch n := v.(type) { + case string: + return strings.TrimSpace(n) + case float64: + return strconv.FormatFloat(n, 'f', -1, 64) + case float32: + return strconv.FormatFloat(float64(n), 'f', -1, 64) + case int: + return strconv.Itoa(n) + case int64: + return strconv.FormatInt(n, 10) + case json.Number: + return n.String() + default: + return strings.TrimSpace(fmt.Sprint(v)) + } +} + type MemoryDeduper struct { mu sync.Mutex seen map[string]struct{} diff --git a/internal/subscriber/dedup_test.go b/internal/subscriber/dedup_test.go index f3faa08..95b1dda 100644 --- a/internal/subscriber/dedup_test.go +++ b/internal/subscriber/dedup_test.go @@ -3,11 +3,13 @@ package subscriber import ( "context" "errors" + "fmt" "sync/atomic" "testing" "aiaa-notification-service/internal/model" "aiaa-notification-service/internal/notify" + "aiaa-notification-service/internal/subscriber/cryptostrategy" "aiaa-notification-service/internal/subscriber/tradesignal" ) @@ -23,6 +25,38 @@ func TestMessageHashStable(t *testing.T) { } } +func TestSignalHashIgnoresUnrelatedFields(t *testing.T) { + a := SignalHash(map[string]interface{}{ + "strategyCode": "ai-crypto-signals", + "symbol": "CRV", + "period": "1h", + "direction": "LONG", + "price": 0.2528, + "eventTime": int64(1), + }) + b := SignalHash(map[string]interface{}{ + "strategyCode": "ai-crypto-signals", + "currency": "CRV", + "period": "1h", + "side": "long", + "price": 0.2528, + "eventTime": int64(2), + }) + if a == "" || a != b { + t.Fatalf("same signal fields should hash equal, a=%q b=%q", a, b) + } + c := SignalHash(map[string]interface{}{ + "strategyCode": "ai-crypto-signals", + "symbol": "CRV", + "period": "1h", + "direction": "LONG", + "price": 0.26, + }) + if a == c { + t.Fatal("different price should hash differently") + } +} + func TestMemoryDeduperClaimOnce(t *testing.T) { d := NewMemoryDeduper() ok, err := d.Claim(context.Background(), "abc") @@ -42,6 +76,51 @@ func TestMemoryDeduperClaimOnce(t *testing.T) { } } +func TestHandleDedupByStrategySymbolPeriodDirectionPrice(t *testing.T) { + dedup := NewMemoryDeduper() + var n atomic.Int32 + process := func(context.Context, notify.Request) (notify.Result, error) { + n.Add(1) + return notify.Result{Matched: true}, nil + } + lookup := func(context.Context, string) (*model.Source, error) { + return &model.Source{ID: 1, Name: "crypto-strategy", Status: 1}, nil + } + conv := cryptostrategy.NewConverter() + in := func(eventTime int64) HandleInput { + body := []byte(fmt.Sprintf(`{ + "eventType":"SIGNAL_RECEIVED","symbol":"CRV","direction":"LONG", + "payload":"{\"strategyCode\":\"ai-crypto-signals\",\"period\":\"1h\",\"currency\":\"CRV\",\"isClose\":true,\"isGain\":true,\"price\":0.2528}", + "eventTime":%d + }`, eventTime)) + return HandleInput{Body: body, SourceName: "crypto-strategy", MaxRetry: 3, Deduper: dedup} + } + + if d := HandleMessage(context.Background(), in(1786899538978), conv, lookup, process); d != DispositionAck { + t.Fatalf("first=%v", d) + } + if d := HandleMessage(context.Background(), in(1786899539730), conv, lookup, process); d != DispositionAck { + t.Fatalf("dup=%v", d) + } + if n.Load() != 1 { + t.Fatalf("same strategy/symbol/period/direction/price should process once, got %d", n.Load()) + } + + bodyDiffPrice := []byte(`{ + "eventType":"SIGNAL_RECEIVED","symbol":"CRV","direction":"LONG", + "payload":"{\"strategyCode\":\"ai-crypto-signals\",\"period\":\"1h\",\"currency\":\"CRV\",\"isClose\":true,\"isGain\":true,\"price\":0.26}", + "eventTime":1786899539731 + }`) + if d := HandleMessage(context.Background(), HandleInput{ + Body: bodyDiffPrice, SourceName: "crypto-strategy", MaxRetry: 3, Deduper: dedup, + }, conv, lookup, process); d != DispositionAck { + t.Fatalf("diff price=%v", d) + } + if n.Load() != 2 { + t.Fatalf("different price should process again, got %d", n.Load()) + } +} + func TestHandleDuplicateAckSkipsProcess(t *testing.T) { dedup := NewMemoryDeduper() var n atomic.Int32 diff --git a/internal/subscriber/handle.go b/internal/subscriber/handle.go index ac34b3d..bfabab5 100644 --- a/internal/subscriber/handle.go +++ b/internal/subscriber/handle.go @@ -78,10 +78,17 @@ func errText(err error) string { } 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 := "" + hash := SignalHash(data) 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", errText(err)) @@ -93,18 +100,8 @@ func HandleMessage(ctx context.Context, in HandleInput, conv MessageConverter, l } } - 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", errText(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", errText(err)) diff --git a/migrations/003_rule_event_not_unique.down.sql b/migrations/003_rule_event_not_unique.down.sql new file mode 100644 index 0000000..0f097ab --- /dev/null +++ b/migrations/003_rule_event_not_unique.down.sql @@ -0,0 +1,3 @@ +ALTER TABLE notification_rule + DROP INDEX idx_source_event, + ADD UNIQUE KEY uk_source_event (source_id, event); diff --git a/migrations/003_rule_event_not_unique.up.sql b/migrations/003_rule_event_not_unique.up.sql new file mode 100644 index 0000000..ef3caa0 --- /dev/null +++ b/migrations/003_rule_event_not_unique.up.sql @@ -0,0 +1,3 @@ +ALTER TABLE notification_rule + DROP INDEX uk_source_event, + ADD INDEX idx_source_event (source_id, event);