fix(notify): keep notification limit window from resetting every hour (#7877)

* fix(notify): keep notification limit window from resetting every hour

The rate-limit key included the current clock hour, so the count reset on
every hour boundary and NOTIFICATION_LIMIT_DURATION_MINUTE above 60 had no
effect. Drop the hour suffix so the window runs for the configured duration
from the first notification, and make the Redis counter atomic with
SETNX+INCR in one transaction.

* fix(notify): fall back to default for nonpositive notification window

A zero or negative NOTIFICATION_LIMIT_DURATION_MINUTE left the Redis
counter without an expiry (or failed the SET), which blocked quota
notifications for good once the counter filled.

* test(notify): cover the Redis notification limit window

common/init.go

@@ -198,7 +198,7 @@ func initConstantEnv() {
 	constant.TaskPluginEnabled = GetEnvOrDefaultBool("TASK_PLUGIN_ENABLED", true)
 	constant.AzureDefaultAPIVersion = GetEnvOrDefaultString("AZURE_DEFAULT_API_VERSION", "2025-04-01-preview")
 	constant.NotifyLimitCount = GetEnvOrDefault("NOTIFY_LIMIT_COUNT", 2)
-	constant.NotificationLimitDurationMinute = GetEnvOrDefault("NOTIFICATION_LIMIT_DURATION_MINUTE", 10)
+	constant.NotificationLimitDurationMinute = positiveUserSessionEnv("NOTIFICATION_LIMIT_DURATION_MINUTE", 10)
 	// GenerateDefaultToken 是否生成初始令牌，默认关闭。
 	constant.GenerateDefaultToken = GetEnvOrDefaultBool("GENERATE_DEFAULT_TOKEN", false)
 	// 是否启用错误日志

service/notify-limit.go

@@ -1,8 +1,8 @@
 package service
 
 import (
+	"context"
 	"fmt"
-	"strconv"
 	"sync"
 	"time"
 
@@ -55,42 +55,26 @@ func CheckNotificationLimit(userId int, notifyType string) (bool, error) {
 }
 
 func checkRedisLimit(userId int, notifyType string) (bool, error) {
-	key := fmt.Sprintf("notify_limit:%d:%s:%s", userId, notifyType, time.Now().Format("2006010215"))
+	ctx := context.Background()
+	key := fmt.Sprintf("notify_limit:%d:%s", userId, notifyType)
 
-	// Get current count
-	count, err := common.RedisGet(key)
-	if err != nil && err.Error() != "redis: nil" {
-		return false, fmt.Errorf("failed to get notification count: %w", err)
-	}
-
-	// If key doesn't exist, initialize it
-	if count == "" {
-		err = common.RedisSet(key, "1", getDuration())
-		return true, err
-	}
-
-	currentCount, _ := strconv.Atoi(count)
-	limit := constant.NotifyLimitCount
-
-	// Check if limit is already reached
-	if currentCount >= limit {
-		return false, nil
-	}
-
-	// Only increment if under limit
-	err = common.RedisIncr(key, 1)
-	if err != nil {
+	// SETNX 只在窗口开始时写入过期时间，INCR 会保留原 TTL；放在同一事务里避免产生永不过期的计数
+	pipe := common.RDB.TxPipeline()
+	pipe.SetNX(ctx, key, 0, getDuration())
+	incr := pipe.Incr(ctx, key)
+	if _, err := pipe.Exec(ctx); err != nil {
 		return false, fmt.Errorf("failed to increment notification count: %w", err)
 	}
 
-	return true, nil
+	count := incr.Val()
+	return count <= int64(constant.NotifyLimitCount), nil
 }
 
 func checkMemoryLimit(userId int, notifyType string) (bool, error) {
 	// Ensure cleanup task is started
 	cleanupOnce.Do(startCleanupTask)
 
-	key := fmt.Sprintf("%d:%s:%s", userId, notifyType, time.Now().Format("2006010215"))
+	key := fmt.Sprintf("%d:%s", userId, notifyType)
 	now := time.Now()
 
 	// Get current limit count or initialize new one

service/notify_limit_test.go

@@ -0,0 +1,94 @@
+package service
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/QuantumNous/new-api/common"
+	"github.com/QuantumNous/new-api/constant"
+	"github.com/QuantumNous/new-api/relaykit/dto"
+	"github.com/alicebob/miniredis/v2"
+	"github.com/go-redis/redis/v8"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// 限流窗口必须由 NOTIFICATION_LIMIT_DURATION_MINUTE 决定，不能在整点被重置
+func TestCheckNotificationLimitWindowSpansClockHours(t *testing.T) {
+	originalRedis := common.RedisEnabled
+	originalCount := constant.NotifyLimitCount
+	originalMinute := constant.NotificationLimitDurationMinute
+	t.Cleanup(func() {
+		common.RedisEnabled = originalRedis
+		constant.NotifyLimitCount = originalCount
+		constant.NotificationLimitDurationMinute = originalMinute
+	})
+	common.RedisEnabled = false
+	constant.NotifyLimitCount = 2
+	constant.NotificationLimitDurationMinute = 720
+
+	tests := []struct {
+		name        string
+		windowStart time.Duration
+		expected    bool
+	}{
+		{name: "limit reached two hours ago still blocks", windowStart: 2 * time.Hour, expected: false},
+		{name: "limit reached after window expired allows", windowStart: 721 * time.Minute, expected: true},
+	}
+	for i, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			userId := 900001 + i
+			key := fmt.Sprintf("%d:%s", userId, dto.NotifyTypeQuotaExceed)
+			notifyLimitStore.Store(key, limitCount{Count: 2, Timestamp: time.Now().Add(-tt.windowStart)})
+			t.Cleanup(func() { notifyLimitStore.Delete(key) })
+
+			canSend, err := CheckNotificationLimit(userId, dto.NotifyTypeQuotaExceed)
+			require.NoError(t, err)
+			assert.Equal(t, tt.expected, canSend)
+		})
+	}
+}
+
+func TestCheckNotificationLimitRedisWindowSpansClockHours(t *testing.T) {
+	originalRedis := common.RedisEnabled
+	originalRDB := common.RDB
+	originalCount := constant.NotifyLimitCount
+	originalMinute := constant.NotificationLimitDurationMinute
+	server := miniredis.RunT(t)
+	common.RDB = redis.NewClient(&redis.Options{Addr: server.Addr(), MaxRetries: -1})
+	t.Cleanup(func() {
+		_ = common.RDB.Close()
+		common.RDB = originalRDB
+		common.RedisEnabled = originalRedis
+		constant.NotifyLimitCount = originalCount
+		constant.NotificationLimitDurationMinute = originalMinute
+	})
+	common.RedisEnabled = true
+	constant.NotifyLimitCount = 2
+	constant.NotificationLimitDurationMinute = 720
+
+	// 窗口在 2 小时前开始并已发满，剩余 10 小时
+	userId := 900010
+	key := fmt.Sprintf("notify_limit:%d:%s", userId, dto.NotifyTypeQuotaExceed)
+	require.NoError(t, server.Set(key, "2"))
+	server.SetTTL(key, 10*time.Hour)
+
+	steps := []struct {
+		name     string
+		elapse   time.Duration
+		expected bool
+	}{
+		{name: "limit reached two hours ago still blocks", expected: false},
+		{name: "new window opens after the old one ends", elapse: 10*time.Hour + time.Minute, expected: true},
+		{name: "second notification in the new window", expected: true},
+		{name: "third notification in the new window is blocked", expected: false},
+		{name: "new window ends after the configured duration", elapse: 720 * time.Minute, expected: true},
+	}
+	for _, step := range steps {
+		server.FastForward(step.elapse)
+		canSend, err := CheckNotificationLimit(userId, dto.NotifyTypeQuotaExceed)
+		require.NoError(t, err, step.name)
+		assert.Equal(t, step.expected, canSend, step.name)
+	}
+}