claude-relay-service/src/routes/admin/concurrency.js

/**
 * 并发管理 API 路由
 * 提供并发状态查看和手动清理功能
 */

const express = require('express')
const router = express.Router()
const redis = require('../../models/redis')
const logger = require('../../utils/logger')
const { authenticateAdmin } = require('../../middleware/auth')
const { calculateWaitTimeStats } = require('../../utils/statsHelper')

/**
 * GET /admin/concurrency
 * 获取所有并发状态
 */
router.get('/concurrency', authenticateAdmin, async (req, res) => {
  try {
    const status = await redis.getAllConcurrencyStatus()

    // 为每个 API Key 获取排队计数
    const statusWithQueue = await Promise.all(
      status.map(async (s) => {
        const queueCount = await redis.getConcurrencyQueueCount(s.apiKeyId)
        return {
          ...s,
          queueCount
        }
      })
    )

    // 计算汇总统计
    const summary = {
      totalKeys: statusWithQueue.length,
      totalActiveRequests: statusWithQueue.reduce((sum, s) => sum + s.activeCount, 0),
      totalExpiredRequests: statusWithQueue.reduce((sum, s) => sum + s.expiredCount, 0),
      totalQueuedRequests: statusWithQueue.reduce((sum, s) => sum + s.queueCount, 0)
    }

    res.json({
      success: true,
      summary,
      concurrencyStatus: statusWithQueue
    })
  } catch (error) {
    logger.error('❌ Failed to get concurrency status:', error)
    res.status(500).json({
      success: false,
      error: 'Failed to get concurrency status',
      message: error.message
    })
  }
})

/**
 * GET /admin/concurrency-queue/stats
 * 获取排队统计信息
 */
router.get('/concurrency-queue/stats', authenticateAdmin, async (req, res) => {
  try {
    // 获取所有有统计数据的 API Key
    const statsKeys = await redis.scanConcurrencyQueueStatsKeys()
    const queueKeys = await redis.scanConcurrencyQueueKeys()

    // 合并所有相关的 API Key
    const allApiKeyIds = [...new Set([...statsKeys, ...queueKeys])]

    // 获取各 API Key 的详细统计
    const perKeyStats = await Promise.all(
      allApiKeyIds.map(async (apiKeyId) => {
        const [queueCount, stats, waitTimes] = await Promise.all([
          redis.getConcurrencyQueueCount(apiKeyId),
          redis.getConcurrencyQueueStats(apiKeyId),
          redis.getQueueWaitTimes(apiKeyId)
        ])

        return {
          apiKeyId,
          currentQueueCount: queueCount,
          stats,
          waitTimeStats: calculateWaitTimeStats(waitTimes)
        }
      })
    )

    // 获取全局等待时间统计
    const globalWaitTimes = await redis.getGlobalQueueWaitTimes()
    const globalWaitTimeStats = calculateWaitTimeStats(globalWaitTimes)

    // 计算全局汇总
    const globalStats = {
      totalEntered: perKeyStats.reduce((sum, s) => sum + s.stats.entered, 0),
      totalSuccess: perKeyStats.reduce((sum, s) => sum + s.stats.success, 0),
      totalTimeout: perKeyStats.reduce((sum, s) => sum + s.stats.timeout, 0),
      totalCancelled: perKeyStats.reduce((sum, s) => sum + s.stats.cancelled, 0),
      totalSocketChanged: perKeyStats.reduce((sum, s) => sum + (s.stats.socket_changed || 0), 0),
      totalRejectedOverload: perKeyStats.reduce(
        (sum, s) => sum + (s.stats.rejected_overload || 0),
        0
      ),
      currentTotalQueued: perKeyStats.reduce((sum, s) => sum + s.currentQueueCount, 0),
      // 队列资源利用率指标
      peakQueueSize:
        perKeyStats.length > 0 ? Math.max(...perKeyStats.map((s) => s.currentQueueCount)) : 0,
      avgQueueSize:
        perKeyStats.length > 0
          ? Math.round(
              perKeyStats.reduce((sum, s) => sum + s.currentQueueCount, 0) / perKeyStats.length
            )
          : 0,
      activeApiKeys: perKeyStats.filter((s) => s.currentQueueCount > 0).length
    }

    // 计算成功率
    if (globalStats.totalEntered > 0) {
      globalStats.successRate = Math.round(
        (globalStats.totalSuccess / globalStats.totalEntered) * 100
      )
      globalStats.timeoutRate = Math.round(
        (globalStats.totalTimeout / globalStats.totalEntered) * 100
      )
      globalStats.cancelledRate = Math.round(
        (globalStats.totalCancelled / globalStats.totalEntered) * 100
      )
    }

    // 从全局等待时间统计中提取关键指标
    if (globalWaitTimeStats) {
      globalStats.avgWaitTimeMs = globalWaitTimeStats.avg
      globalStats.p50WaitTimeMs = globalWaitTimeStats.p50
      globalStats.p90WaitTimeMs = globalWaitTimeStats.p90
      globalStats.p99WaitTimeMs = globalWaitTimeStats.p99
      // 多实例采样策略标记（详见 design.md Decision 9）
      // 全局 P90 仅用于可视化和监控，不用于系统决策
      // 健康检查使用 API Key 级别的 P90（每 Key 独立采样）
      globalWaitTimeStats.globalP90ForVisualizationOnly = true
    }

    res.json({
      success: true,
      globalStats,
      globalWaitTimeStats,
      perKeyStats
    })
  } catch (error) {
    logger.error('❌ Failed to get queue stats:', error)
    res.status(500).json({
      success: false,
      error: 'Failed to get queue stats',
      message: error.message
    })
  }
})

/**
 * DELETE /admin/concurrency-queue/:apiKeyId
 * 清理特定 API Key 的排队计数
 */
router.delete('/concurrency-queue/:apiKeyId', authenticateAdmin, async (req, res) => {
  try {
    const { apiKeyId } = req.params
    await redis.clearConcurrencyQueue(apiKeyId)

    logger.warn(`🧹 Admin ${req.admin?.username || 'unknown'} cleared queue for key ${apiKeyId}`)

    res.json({
      success: true,
      message: `Successfully cleared queue for API key ${apiKeyId}`
    })
  } catch (error) {
    logger.error(`❌ Failed to clear queue for ${req.params.apiKeyId}:`, error)
    res.status(500).json({
      success: false,
      error: 'Failed to clear queue',
      message: error.message
    })
  }
})

/**
 * DELETE /admin/concurrency-queue
 * 清理所有排队计数
 */
router.delete('/concurrency-queue', authenticateAdmin, async (req, res) => {
  try {
    const cleared = await redis.clearAllConcurrencyQueues()

    logger.warn(`🧹 Admin ${req.admin?.username || 'unknown'} cleared ALL queues`)

    res.json({
      success: true,
      message: 'Successfully cleared all queues',
      cleared
    })
  } catch (error) {
    logger.error('❌ Failed to clear all queues:', error)
    res.status(500).json({
      success: false,
      error: 'Failed to clear all queues',
      message: error.message
    })
  }
})

/**
 * GET /admin/concurrency/:apiKeyId
 * 获取特定 API Key 的并发状态详情
 */
router.get('/concurrency/:apiKeyId', authenticateAdmin, async (req, res) => {
  try {
    const { apiKeyId } = req.params
    const status = await redis.getConcurrencyStatus(apiKeyId)
    const queueCount = await redis.getConcurrencyQueueCount(apiKeyId)

    res.json({
      success: true,
      concurrencyStatus: {
        ...status,
        queueCount
      }
    })
  } catch (error) {
    logger.error(`❌ Failed to get concurrency status for ${req.params.apiKeyId}:`, error)
    res.status(500).json({
      success: false,
      error: 'Failed to get concurrency status',
      message: error.message
    })
  }
})

/**
 * DELETE /admin/concurrency/:apiKeyId
 * 强制清理特定 API Key 的并发计数
 */
router.delete('/concurrency/:apiKeyId', authenticateAdmin, async (req, res) => {
  try {
    const { apiKeyId } = req.params
    const result = await redis.forceClearConcurrency(apiKeyId)

    logger.warn(
      `🧹 Admin ${req.admin?.username || 'unknown'} force cleared concurrency for key ${apiKeyId}`
    )

    res.json({
      success: true,
      message: `Successfully cleared concurrency for API key ${apiKeyId}`,
      result
    })
  } catch (error) {
    logger.error(`❌ Failed to clear concurrency for ${req.params.apiKeyId}:`, error)
    res.status(500).json({
      success: false,
      error: 'Failed to clear concurrency',
      message: error.message
    })
  }
})

/**
 * DELETE /admin/concurrency
 * 强制清理所有并发计数
 */
router.delete('/concurrency', authenticateAdmin, async (req, res) => {
  try {
    const result = await redis.forceClearAllConcurrency()

    logger.warn(`🧹 Admin ${req.admin?.username || 'unknown'} force cleared ALL concurrency`)

    res.json({
      success: true,
      message: 'Successfully cleared all concurrency',
      result
    })
  } catch (error) {
    logger.error('❌ Failed to clear all concurrency:', error)
    res.status(500).json({
      success: false,
      error: 'Failed to clear all concurrency',
      message: error.message
    })
  }
})

/**
 * POST /admin/concurrency/cleanup
 * 清理过期的并发条目（不影响活跃请求）
 */
router.post('/concurrency/cleanup', authenticateAdmin, async (req, res) => {
  try {
    const { apiKeyId } = req.body
    const result = await redis.cleanupExpiredConcurrency(apiKeyId || null)

    logger.info(`🧹 Admin ${req.admin?.username || 'unknown'} cleaned up expired concurrency`)

    res.json({
      success: true,
      message: apiKeyId
        ? `Successfully cleaned up expired concurrency for API key ${apiKeyId}`
        : 'Successfully cleaned up all expired concurrency',
      result
    })
  } catch (error) {
    logger.error('❌ Failed to cleanup expired concurrency:', error)
    res.status(500).json({
      success: false,
      error: 'Failed to cleanup expired concurrency',
      message: error.message
    })
  }
})

module.exports = router