From 83c4166f0cef22759f418398636535f911fc3901 Mon Sep 17 00:00:00 2001 From: ryan Date: Sat, 27 Jun 2026 13:19:15 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20engine=20=E2=80=94=20template=20rendere?= =?UTF-8?q?r,=20rule=20matcher,=20channel=20router?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/engine/matcher.go | 60 +++++++++++++++++++++ internal/engine/renderer.go | 25 +++++++++ internal/engine/renderer_test.go | 27 ++++++++++ internal/engine/router.go | 92 ++++++++++++++++++++++++++++++++ 4 files changed, 204 insertions(+) create mode 100644 internal/engine/matcher.go create mode 100644 internal/engine/renderer.go create mode 100644 internal/engine/renderer_test.go create mode 100644 internal/engine/router.go diff --git a/internal/engine/matcher.go b/internal/engine/matcher.go new file mode 100644 index 0000000..91c8b0b --- /dev/null +++ b/internal/engine/matcher.go @@ -0,0 +1,60 @@ +package engine + +import ( + "context" + "encoding/json" + "fmt" + + "aiaa-notification-service/internal/cache" + "aiaa-notification-service/internal/model" + "aiaa-notification-service/internal/store" +) + +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 (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 + } + } + + // Fall back to DB + rule, err := m.store.GetRuleBySourceEvent(ctx, sourceID, event) + if err != nil { + return nil, fmt.Errorf("match rule: %w", err) + } + + // Warm cache + if m.cache != nil { + tmpl, err := m.store.GetTemplate(ctx, rule.TemplateID) + if err != nil { + return rule, nil // rule found but template fetch failed — still return rule + } + 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) + } + + return rule, nil +} diff --git a/internal/engine/renderer.go b/internal/engine/renderer.go new file mode 100644 index 0000000..15dde2b --- /dev/null +++ b/internal/engine/renderer.go @@ -0,0 +1,25 @@ +package engine + +import ( + "bytes" + "fmt" + "text/template" +) + +type Renderer struct{} + +func NewRenderer() *Renderer { + return &Renderer{} +} + +func (r *Renderer) Render(tmplContent string, data map[string]interface{}) (string, error) { + tmpl, err := template.New("notify").Option("missingkey=error").Parse(tmplContent) + if err != nil { + return "", fmt.Errorf("parse template: %w", err) + } + var buf bytes.Buffer + if err := tmpl.Execute(&buf, data); err != nil { + return "", fmt.Errorf("execute template: %w", err) + } + return buf.String(), nil +} diff --git a/internal/engine/renderer_test.go b/internal/engine/renderer_test.go new file mode 100644 index 0000000..ab70d75 --- /dev/null +++ b/internal/engine/renderer_test.go @@ -0,0 +1,27 @@ +package engine + +import ( + "strings" + "testing" +) + +func TestRenderer(t *testing.T) { + r := NewRenderer() + tmpl := "🚀 {{.symbol}} 开仓通知\n价格: {{.price}}" + data := map[string]interface{}{"symbol": "BTC", "price": 65000} + result, err := r.Render(tmpl, data) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !strings.Contains(result, "BTC") || !strings.Contains(result, "65000") { + t.Errorf("unexpected output: %s", result) + } +} + +func TestRenderer_Error(t *testing.T) { + r := NewRenderer() + _, err := r.Render("{{.nonexistent}}", map[string]interface{}{}) + if err == nil { + t.Error("expected error for missing field, got nil") + } +} diff --git a/internal/engine/router.go b/internal/engine/router.go new file mode 100644 index 0000000..98890c0 --- /dev/null +++ b/internal/engine/router.go @@ -0,0 +1,92 @@ +package engine + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + + "aiaa-notification-service/internal/adapter" + "aiaa-notification-service/internal/cache" + "aiaa-notification-service/internal/model" + "aiaa-notification-service/internal/store" +) + +type SendResult struct { + ChannelType string `json:"channel_type"` + ChannelID int `json:"channel_id"` + Success bool `json:"success"` + Error string `json:"error,omitempty"` +} + +type Router struct { + store *store.Store + cache *cache.Cache + senderFactory func(channelType string) (adapter.ChannelSender, error) +} + +func NewRouter(s *store.Store, c *cache.Cache, sf func(channelType string) (adapter.ChannelSender, error)) *Router { + return &Router{store: s, cache: c, senderFactory: sf} +} + +// Route sends content to all enabled channels for the rule. Returns immediately, sends async. +func (r *Router) Route(ctx context.Context, rule *model.Rule, title, content string) []string { + channels, err := r.getChannels(ctx, rule.ID) + if err != nil { + slog.Error("get channels for rule", "rule_id", rule.ID, "error", err) + return nil + } + + channelNames := make([]string, 0, len(channels)) + for _, ch := range channels { + channelNames = append(channelNames, fmt.Sprintf("%s:%d", ch.Type, ch.ID)) + sender, err := r.senderFactory(ch.Type) + if err != nil { + slog.Error("create sender", "type", ch.Type, "error", err) + continue + } + + go func(ch cache.CachedChannel, s adapter.ChannelSender) { + var cfg json.RawMessage + if ch.Config != nil { + cfg = *ch.Config + } + if err := s.Send(title, content, cfg); err != nil { + slog.Error("send failed", "channel_type", ch.Type, "channel_id", ch.ID, "error", err) + } else { + slog.Info("sent", "channel_type", ch.Type, "channel_id", ch.ID) + } + }(ch, sender) + } + + return channelNames +} + +func (r *Router) getChannels(ctx context.Context, ruleID int) ([]cache.CachedChannel, error) { + if r.cache != nil { + chs, err := r.cache.GetChannels(ctx, ruleID) + if err == nil { + return chs, nil + } + } + + rcs, err := r.store.GetRuleChannels(ctx, ruleID) + if err != nil { + return nil, err + } + + var cached []cache.CachedChannel + for _, rc := range rcs { + ch, err := r.store.GetChannel(ctx, rc.ChannelID) + if err != nil { + continue + } + cc := cache.CachedChannel{ID: ch.ID, Type: ch.Type, Config: ch.Config} + cached = append(cached, cc) + } + + if r.cache != nil { + _ = r.cache.SetChannels(ctx, ruleID, cached) + } + return cached, nil +}