fix: atomic rate limiter and error handling in cache invalidation

.superpowers/sdd/task-6-report.md

@@ -0,0 +1,47 @@
+# Task 6: Redis Cache Layer -- Complete
+
+## Summary
+
+Created the Redis cache layer in `internal/cache/redis.go` with rule caching, channel caching, bulk invalidation, and rate limiting.
+
+## Files Changed
+
+- **Created:** `internal/cache/redis.go` -- Cache struct with full implementation
+- **Modified:** `go.mod` -- Added `github.com/redis/go-redis/v9` dependency
+- **Modified:** `go.sum` -- Updated checksums
+
+## Implementation Details
+
+- `NewCache(cfg config.RedisConfig)` -- Connects to Redis using `config.RedisConfig` (fixed from the brief's typo `model.config.RedisConfig`)
+- `CachedRule` struct -- RuleID, TemplateID, Content, Conditions (JSON string)
+- `CachedChannel` struct -- ID, Type, Config (*json.RawMessage)
+- `GetRule` / `SetRule` / `InvalidateRule` -- Keys: `notify:rule:{sourceID}:{event}`, TTL 5 min
+- `GetChannels` / `SetChannels` / `InvalidateChannels` -- Keys: `notify:channels:{ruleID}`, TTL 5 min
+- `InvalidateBySource` -- Scans `notify:rule:{sourceID}:*` and deletes matches
+- `InvalidateByTemplate` -- Scans all `notify:rule:*` keys and deletes them
+- `CheckRateLimit` -- Sliding window via sorted set (ZRemRangeByScore + ZCard), key `ratelimit:{sourceID}`
+- `Close` -- Closes the Redis client
+
+## Verification
+
+- `go build ./internal/cache/...` completed with no errors
+- Commit: `3f75699` with message "feat: redis cache layer with rule/channel caching and rate limiter"
+
+## Fix Section -- Race Condition and Error Handling
+
+### Issue 1: TOCTOU race condition in CheckRateLimit
+
+The original implementation used a pipeline where `ZRemRangeByScore` (cleanup) and `ZCard` (count check) ran together, but the subsequent `ZAdd` (add request) ran separately in Go code. This created a time-of-check-to-time-of-use window where concurrent requests could all pass the limit check before any of them recorded the add.
+
+**Fix:** Replaced the pipeline approach with an atomic Lua script. The entire operation — cleanup, count check, and conditional add — now runs as a single atomic unit inside Redis, eliminating the race window. The script also switched from second-precision timestamps to millisecond-precision (`time.Now().UnixNano() / 1e6`) for unique sorted set members.
+
+### Issue 2: Discarded Del errors in InvalidateBySource and InvalidateByTemplate
+
+Both invalidate methods called `c.rdb.Del()` inside a scan loop but did not check the returned error, silently swallowing deletion failures.
+
+**Fix:** Both methods now collect errors from each failed `Del` call into an `[]error` slice. If any deletions failed, an aggregated error is returned quoting the count and the first error. The scan iterator error (`iter.Err()`) is still returned when no deletion errors occurred.
+
+### Verification
+
+- `go build ./internal/cache/...` completed with no errors
+- `go vet ./internal/cache/...` completed with no warnings

internal/cache/redis.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"strconv"
 	"time"
 
 	"aiaa-notification-service/internal/config"
@@ -109,8 +108,14 @@ func (c *Cache) InvalidateChannels(ctx context.Context, ruleID int) error {
 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) {
-		c.rdb.Del(ctx, iter.Val())
+		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()
 }
@@ -119,8 +124,14 @@ 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) {
-		c.rdb.Del(ctx, iter.Val())
+		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()
 }
@@ -129,20 +140,36 @@ func (c *Cache) InvalidateByTemplate(ctx context.Context, templateID int) error
 
 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)
+	now := time.Now().UnixNano() / 1e6 // milliseconds for uniqueness
+	windowStart := time.Now().Unix() - 1
 
-	count := countCmd.Val()
-	if count >= int64(limitPerSec) {
-		return false, 1, nil // rate limited, retry after 1s
+	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)
 	}
 
-	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
+	if result == 0 {
+		return false, 1, nil // rate limited
+	}
+	return true, 0, nil // allowed
 }
 
 func (c *Cache) Close() error {