feat: add concurrency ops snapshot API

Expose admin Redis snapshot for user and channel concurrency plus current user RPM usage.

controller/ops_concurrency.go

@@ -0,0 +1,11 @@
+package controller
+
+import (
+	"github.com/QuantumNous/new-api/common"
+	"github.com/QuantumNous/new-api/service"
+	"github.com/gin-gonic/gin"
+)
+
+func GetOpsConcurrencySnapshot(c *gin.Context) {
+	common.ApiSuccess(c, service.GetOpsConcurrencySnapshot())
+}

router/api-router.go

@@ -329,6 +329,11 @@ func SetApiRouter(router *gin.Engine) {
 			affiliateAdminRoute.DELETE("/users/:user_id", controller.AdminClearAffiliateUserSettings)
 			affiliateAdminRoute.PUT("/config", controller.AdminUpdateAffiliateConfig)
 		}
+		opsAdminRoute := apiRouter.Group("/admin/ops")
+		opsAdminRoute.Use(middleware.AdminAuth())
+		{
+			opsAdminRoute.GET("/concurrency", controller.GetOpsConcurrencySnapshot)
+		}
 		channelMonitorAdminRoute := apiRouter.Group("/admin/channel-monitors")
 		channelMonitorAdminRoute.Use(middleware.AdminAuth())
 		{

service/concurrency.go

@@ -2,7 +2,9 @@ package service
 
 import (
 	"context"
+	"sort"
 	"strconv"
+	"strings"
 	"time"
 
 	"github.com/QuantumNous/new-api/common"
@@ -17,6 +19,24 @@ const (
 	rpmTTL         = 2 * time.Minute
 )
 
+type OpsConcurrencySnapshot struct {
+	RedisEnabled bool                            `json:"redis_enabled"`
+	Users        []OpsUserConcurrencySnapshot    `json:"users"`
+	Channels     []OpsChannelConcurrencySnapshot `json:"channels"`
+	GeneratedAt  int64                           `json:"generated_at"`
+}
+
+type OpsUserConcurrencySnapshot struct {
+	UserID             int `json:"user_id"`
+	CurrentConcurrency int `json:"current_concurrency"`
+	CurrentRPM         int `json:"current_rpm"`
+}
+
+type OpsChannelConcurrencySnapshot struct {
+	ChannelID          int `json:"channel_id"`
+	CurrentConcurrency int `json:"current_concurrency"`
+}
+
 func AcquireUserConcurrency(userId, limit int) (bool, func()) {
 	return acquireConcurrency(userConcurrencyKeyPrefix+strconv.Itoa(userId), limit)
 }
@@ -43,6 +63,70 @@ func CheckUserRPM(userId, limit int) bool {
 	return count <= int64(limit)
 }
 
+func GetOpsConcurrencySnapshot() OpsConcurrencySnapshot {
+	now := time.Now()
+	snapshot := OpsConcurrencySnapshot{
+		RedisEnabled: redisAvailable(),
+		Users:        []OpsUserConcurrencySnapshot{},
+		Channels:     []OpsChannelConcurrencySnapshot{},
+		GeneratedAt:  now.Unix(),
+	}
+	if !snapshot.RedisEnabled {
+		return snapshot
+	}
+
+	ctx := context.Background()
+	userConcurrency, ok := scanIntValues(ctx, userConcurrencyKeyPrefix+"*", parseUserConcurrencyKey)
+	if !ok {
+		snapshot.RedisEnabled = false
+		return emptyOpsConcurrencySnapshot(snapshot)
+	}
+
+	currentRPMMinute := now.Format("200601021504")
+	userRPM, ok := scanIntValues(ctx, userRPMKeyPrefix+"*:"+currentRPMMinute, func(key string) (int, bool) {
+		return parseUserRPMKey(key, currentRPMMinute)
+	})
+	if !ok {
+		snapshot.RedisEnabled = false
+		return emptyOpsConcurrencySnapshot(snapshot)
+	}
+
+	channelConcurrency, ok := scanIntValues(ctx, channelConcurrencyKeyPrefix+"*", parseChannelConcurrencyKey)
+	if !ok {
+		snapshot.RedisEnabled = false
+		return emptyOpsConcurrencySnapshot(snapshot)
+	}
+
+	for userID, concurrency := range userConcurrency {
+		snapshot.Users = append(snapshot.Users, OpsUserConcurrencySnapshot{
+			UserID:             userID,
+			CurrentConcurrency: concurrency,
+			CurrentRPM:         userRPM[userID],
+		})
+		delete(userRPM, userID)
+	}
+	for userID, rpm := range userRPM {
+		snapshot.Users = append(snapshot.Users, OpsUserConcurrencySnapshot{
+			UserID:     userID,
+			CurrentRPM: rpm,
+		})
+	}
+	for channelID, concurrency := range channelConcurrency {
+		snapshot.Channels = append(snapshot.Channels, OpsChannelConcurrencySnapshot{
+			ChannelID:          channelID,
+			CurrentConcurrency: concurrency,
+		})
+	}
+	sort.Slice(snapshot.Users, func(i, j int) bool {
+		return snapshot.Users[i].UserID < snapshot.Users[j].UserID
+	})
+	sort.Slice(snapshot.Channels, func(i, j int) bool {
+		return snapshot.Channels[i].ChannelID < snapshot.Channels[j].ChannelID
+	})
+
+	return snapshot
+}
+
 func acquireConcurrency(key string, limit int) (bool, func()) {
 	release := func() {}
 	if limit <= 0 || !redisAvailable() {
@@ -75,3 +159,70 @@ func acquireConcurrency(key string, limit int) (bool, func()) {
 func redisAvailable() bool {
 	return common.RedisEnabled && common.RDB != nil
 }
+
+func emptyOpsConcurrencySnapshot(snapshot OpsConcurrencySnapshot) OpsConcurrencySnapshot {
+	snapshot.Users = []OpsUserConcurrencySnapshot{}
+	snapshot.Channels = []OpsChannelConcurrencySnapshot{}
+	return snapshot
+}
+
+func scanIntValues(ctx context.Context, pattern string, parseKey func(string) (int, bool)) (map[int]int, bool) {
+	values := make(map[int]int)
+	var cursor uint64
+	for {
+		keys, nextCursor, err := common.RDB.Scan(ctx, cursor, pattern, 100).Result()
+		if err != nil {
+			return nil, false
+		}
+		for _, key := range keys {
+			id, ok := parseKey(key)
+			if !ok {
+				continue
+			}
+			value, err := common.RDB.Get(ctx, key).Int()
+			if err != nil {
+				continue
+			}
+			values[id] = value
+		}
+		if nextCursor == 0 {
+			break
+		}
+		cursor = nextCursor
+	}
+	return values, true
+}
+
+func parseUserConcurrencyKey(key string) (int, bool) {
+	if !strings.HasPrefix(key, userConcurrencyKeyPrefix) {
+		return 0, false
+	}
+	return parsePositiveID(strings.TrimPrefix(key, userConcurrencyKeyPrefix))
+}
+
+func parseChannelConcurrencyKey(key string) (int, bool) {
+	if !strings.HasPrefix(key, channelConcurrencyKeyPrefix) {
+		return 0, false
+	}
+	return parsePositiveID(strings.TrimPrefix(key, channelConcurrencyKeyPrefix))
+}
+
+func parseUserRPMKey(key, minute string) (int, bool) {
+	prefix := userRPMKeyPrefix
+	suffix := ":" + minute
+	if !strings.HasPrefix(key, prefix) || !strings.HasSuffix(key, suffix) {
+		return 0, false
+	}
+	return parsePositiveID(strings.TrimSuffix(strings.TrimPrefix(key, prefix), suffix))
+}
+
+func parsePositiveID(value string) (int, bool) {
+	if value == "" {
+		return 0, false
+	}
+	id, err := strconv.Atoi(value)
+	if err != nil || id <= 0 {
+		return 0, false
+	}
+	return id, true
+}

service/concurrency_test.go

@@ -57,3 +57,60 @@ func TestAcquireConcurrencyAllowsWhenRedisUnavailable(t *testing.T) {
 		})
 	}
 }
+
+func TestGetOpsConcurrencySnapshotReturnsEmptyWhenRedisDisabled(t *testing.T) {
+	previousRedisEnabled := common.RedisEnabled
+	previousRDB := common.RDB
+	common.RedisEnabled = false
+	common.RDB = nil
+	t.Cleanup(func() {
+		common.RedisEnabled = previousRedisEnabled
+		common.RDB = previousRDB
+	})
+
+	snapshot := GetOpsConcurrencySnapshot()
+
+	require.False(t, snapshot.RedisEnabled)
+	require.Empty(t, snapshot.Users)
+	require.Empty(t, snapshot.Channels)
+	require.NotZero(t, snapshot.GeneratedAt)
+}
+
+func TestParseOpsConcurrencyKeys(t *testing.T) {
+	userID, ok := parseUserConcurrencyKey("user:concurrency:123")
+	require.True(t, ok)
+	require.Equal(t, 123, userID)
+
+	channelID, ok := parseChannelConcurrencyKey("channel:concurrency:456")
+	require.True(t, ok)
+	require.Equal(t, 456, channelID)
+
+	rpmUserID, ok := parseUserRPMKey("user:rpm:789:202605201430", "202605201430")
+	require.True(t, ok)
+	require.Equal(t, 789, rpmUserID)
+}
+
+func TestParseOpsConcurrencyKeysRejectsInvalidKeys(t *testing.T) {
+	invalidUserKeys := []string{
+		"user:concurrency:",
+		"user:concurrency:0",
+		"user:concurrency:abc",
+		"channel:concurrency:1",
+	}
+	for _, key := range invalidUserKeys {
+		_, ok := parseUserConcurrencyKey(key)
+		require.False(t, ok)
+	}
+
+	invalidRPMKeys := []string{
+		"user:rpm:",
+		"user:rpm:0:202605201430",
+		"user:rpm:abc:202605201430",
+		"user:rpm:123:202605201429",
+		"user:rpm:123",
+	}
+	for _, key := range invalidRPMKeys {
+		_, ok := parseUserRPMKey(key, "202605201430")
+		require.False(t, ok)
+	}
+}