feat: redis cache layer with rule/channel caching and rate limiter

internal/cache/redis.go

@@ -0,0 +1,150 @@
+package cache
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strconv"
+	"time"
+
+	"aiaa-notification-service/internal/config"
+
+	"github.com/redis/go-redis/v9"
+)
+
+type Cache struct {
+	rdb *redis.Client
+}
+
+// CachedRule holds the minimal info needed after rule matching.
+type CachedRule struct {
+	RuleID     int    `json:"rule_id"`
+	TemplateID int    `json:"template_id"`
+	Content    string `json:"content"`
+	Conditions string `json:"conditions"` // JSON string, empty if null
+}
+
+type CachedChannel struct {
+	ID     int              `json:"id"`
+	Type   string           `json:"type"`
+	Config *json.RawMessage `json:"config"`
+}
+
+func NewCache(cfg config.RedisConfig) (*Cache, error) {
+	rdb := redis.NewClient(&redis.Options{
+		Addr:     cfg.Addr(),
+		Password: cfg.Password,
+		DB:       cfg.DB,
+	})
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	if err := rdb.Ping(ctx).Err(); err != nil {
+		return nil, fmt.Errorf("redis ping: %w", err)
+	}
+	return &Cache{rdb: rdb}, nil
+}
+
+// --- Rule cache ---
+
+func ruleKey(sourceID int, event string) string {
+	return fmt.Sprintf("notify:rule:%d:%s", sourceID, event)
+}
+
+func (c *Cache) GetRule(ctx context.Context, sourceID int, event string) (*CachedRule, error) {
+	data, err := c.rdb.Get(ctx, ruleKey(sourceID, event)).Bytes()
+	if err != nil {
+		return nil, err
+	}
+	var cr CachedRule
+	if err := json.Unmarshal(data, &cr); err != nil {
+		return nil, err
+	}
+	return &cr, nil
+}
+
+func (c *Cache) SetRule(ctx context.Context, sourceID int, event string, cr *CachedRule) error {
+	data, err := json.Marshal(cr)
+	if err != nil {
+		return err
+	}
+	return c.rdb.Set(ctx, ruleKey(sourceID, event), data, 5*time.Minute).Err()
+}
+
+func (c *Cache) InvalidateRule(ctx context.Context, sourceID int, event string) error {
+	return c.rdb.Del(ctx, ruleKey(sourceID, event)).Err()
+}
+
+// --- Channel cache ---
+
+func channelsKey(ruleID int) string {
+	return fmt.Sprintf("notify:channels:%d", ruleID)
+}
+
+func (c *Cache) GetChannels(ctx context.Context, ruleID int) ([]CachedChannel, error) {
+	data, err := c.rdb.Get(ctx, channelsKey(ruleID)).Bytes()
+	if err != nil {
+		return nil, err
+	}
+	var channels []CachedChannel
+	if err := json.Unmarshal(data, &channels); err != nil {
+		return nil, err
+	}
+	return channels, nil
+}
+
+func (c *Cache) SetChannels(ctx context.Context, ruleID int, channels []CachedChannel) error {
+	data, err := json.Marshal(channels)
+	if err != nil {
+		return err
+	}
+	return c.rdb.Set(ctx, channelsKey(ruleID), data, 5*time.Minute).Err()
+}
+
+func (c *Cache) InvalidateChannels(ctx context.Context, ruleID int) error {
+	return c.rdb.Del(ctx, channelsKey(ruleID)).Err()
+}
+
+// --- Bulk invalidation ---
+
+func (c *Cache) InvalidateBySource(ctx context.Context, sourceID int) error {
+	pattern := fmt.Sprintf("notify:rule:%d:*", sourceID)
+	iter := c.rdb.Scan(ctx, 0, pattern, 0).Iterator()
+	for iter.Next(ctx) {
+		c.rdb.Del(ctx, iter.Val())
+	}
+	return iter.Err()
+}
+
+func (c *Cache) InvalidateByTemplate(ctx context.Context, templateID int) error {
+	// Template changes mean all cached rule info is stale. Simplest: scan all rule keys.
+	// In production, you'd maintain a template to rule reverse index. For v1, scan is acceptable.
+	iter := c.rdb.Scan(ctx, 0, "notify:rule:*", 0).Iterator()
+	for iter.Next(ctx) {
+		c.rdb.Del(ctx, iter.Val())
+	}
+	return iter.Err()
+}
+
+// --- Rate limiter ---
+
+func (c *Cache) CheckRateLimit(ctx context.Context, sourceID int, limitPerSec int) (bool, int, error) {
+	key := fmt.Sprintf("ratelimit:%d", sourceID)
+	now := time.Now().Unix()
+	pipe := c.rdb.Pipeline()
+	pipe.ZRemRangeByScore(ctx, key, "0", strconv.FormatInt(now-1, 10))
+	countCmd := pipe.ZCard(ctx, key)
+	pipe.Exec(ctx)
+
+	count := countCmd.Val()
+	if count >= int64(limitPerSec) {
+		return false, 1, nil // rate limited, retry after 1s
+	}
+
+	c.rdb.ZAdd(ctx, key, redis.Z{Score: float64(now), Member: strconv.FormatInt(now*1000, 10)})
+	c.rdb.Expire(ctx, key, 2*time.Second)
+	return true, 0, nil
+}
+
+func (c *Cache) Close() error {
+	return c.rdb.Close()
+}