feat: engine — template renderer, rule matcher, channel router

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
+}

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
+}

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")
+	}
+}

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
+}