package cache import ( "context" "encoding/json" "fmt" "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() var errs []error for iter.Next(ctx) { if err := c.rdb.Del(ctx, iter.Val()).Err(); err != nil { errs = append(errs, err) } } if len(errs) > 0 { return fmt.Errorf("invalidate by source %d: %d deletions failed: %w", sourceID, len(errs), errs[0]) } 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() var errs []error for iter.Next(ctx) { if err := c.rdb.Del(ctx, iter.Val()).Err(); err != nil { errs = append(errs, err) } } if len(errs) > 0 { return fmt.Errorf("invalidate by template %d: %d deletions failed: %w", templateID, len(errs), errs[0]) } 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().UnixNano() / 1e6 // milliseconds for uniqueness windowStart := time.Now().Unix() - 1 script := redis.NewScript(` local key = KEYS[1] local now = tonumber(ARGV[1]) local window_start = tonumber(ARGV[2]) local limit = tonumber(ARGV[3]) redis.call('ZREMRANGEBYSCORE', key, '-inf', window_start) local count = redis.call('ZCARD', key) if count >= limit then return 0 end redis.call('ZADD', key, now, now) redis.call('EXPIRE', key, 2) return 1 `) result, err := script.Run(ctx, c.rdb, []string{key}, now, windowStart, limitPerSec).Int() if err != nil { return false, 1, fmt.Errorf("rate limit check: %w", err) } if result == 0 { return false, 1, nil // rate limited } return true, 0, nil // allowed } func (c *Cache) Close() error { return c.rdb.Close() }