dotnet/molilian.core/Core/taoke/TkEndpointManager.cs

using dodohold.core;
using Google.Protobuf.WellKnownTypes;
using molilian.core;
using System;
using System.Collections.Concurrent;
using System.Linq;

public partial class TkEndpointManager
{
    // 存储每个账号的端点状态（Key: accountId）
    private static readonly ConcurrentDictionary<int, AccountEndpointState> accountStates = new();

    // 记录每个账号最后一次挂起操作的时间
    private static readonly ConcurrentDictionary<int, DateTime> lastSuspendTimes = new();

    // 每个账号的挂起锁对象
    private static readonly ConcurrentDictionary<int, object> accountLocks = new();

    // 挂起冷却时间（3秒）
    private static readonly TimeSpan SuspendCooldown = TimeSpan.FromSeconds(3);

    // 替换原来的 ConcurrentDictionary<int, object>
    private static readonly ConcurrentDictionary<int, SemaphoreSlim> accountSemaphores = new();


    public static void Refresh()
    {
        accountStates.Clear();
        lastSuspendTimes.Clear();
        accountLocks.Clear();
        accountSemaphores.Clear();
    }


    // 从数据库获取所有可用的API端点
    private static async Task<Dictionary<int, string>> GetAllAvailableApisAsync(int accountId)
    {
        var endpoints = await TkEndpointCore.GetEndpointsByAccountAsync(accountId);
        return endpoints?
            .Where(e => e.status)
            .ToDictionary(e => e.ep_id, e => e.endpoint) ?? new Dictionary<int, string>();
    }

    // 从数据库获取端点的挂起时间配置
    private static async Task<ConcurrentDictionary<string, int>> GetEndpointHoldMinutesAsync(int accountId)
    {
        var endpoints = await TkEndpointCore.GetEndpointsByAccountAsync(accountId);
        var dict = new ConcurrentDictionary<string, int>();

        if (endpoints != null)
        {
            foreach (var endpoint in endpoints)
            {
                dict.TryAdd(endpoint.endpoint, endpoint.suspend_duration);
            }
        }

        return dict;
    }
    public static bool IsEndpointSuspended(int accountId, string endpoint)
    {
        if (!accountStates.TryGetValue(accountId, out var state))
            return false;

        return state.IsEndpointSuspended(endpoint);
    }

    // 新增：直接接收端点对象的版本
    public static bool IsEndpointSuspended(TkEndpointConfigDTO endpoint)
    {
        if (endpoint == null) return true;
        if (!accountStates.TryGetValue(endpoint.tk_pool_id, out var state))
            return false;

        return state.IsEndpointSuspended(endpoint.endpoint);
    }


    /// <summary>
    /// 检查指定账号是否至少拥有一个可用的endpoint
    /// </summary>
    /// <param name="accountId">账号ID</param>
    /// <returns>true如果至少有一个可用endpoint，否则false</returns>
    public static async Task<bool> HasAvailableEndpointAsync(int accountId)
    {
        var availableApis = await GetAllAvailableApisAsync(accountId);
        if (availableApis.Count == 0) return false;

        if (!accountStates.TryGetValue(accountId, out var state))
        {
            return true; // 如果账号没有状态记录，所有节点都是正常的
        }

        return availableApis.Any(api => !state.IsEndpointSuspended(api.Value));
    }

    /// <summary>
    /// 获取多个账号的节点状态信息
    /// </summary>
    public static async Task<Dictionary<int, Dictionary<string, string>>> GetStatus(IEnumerable<int> accountIds)
    {
        var result = new Dictionary<int, Dictionary<string, string>>();

        foreach (var accountId in accountIds.Distinct())
        {
            var availableApis = await GetAllAvailableApisAsync(accountId);
            result[accountId] = await GetStatus(accountId, availableApis);
        }

        return result;
    }

    /// <summary>
    /// 获取指定账号的节点状态信息
    /// </summary>
    public static async Task<Dictionary<string, string>> GetStatus(int accountId, Dictionary<int, string> availableApis)
    {
        var status = new Dictionary<string, string>();
        if (availableApis.Count == 0)
        {
            return status;
        }

        if (!accountStates.TryGetValue(accountId, out var state))
        {
            // 如果账号没有状态记录，所有节点都是正常的
            foreach ((var api_id, var api) in availableApis)
            {
                status[api] = "正常";
            }
            return status;
        }

        foreach ((var api_id, var api) in availableApis)
        {
            if (state.IsEndpointSuspended(api))
            {
                var suspendUntil = state.GetSuspendTime(api);
                var remainingTime = suspendUntil - DateTime.Now;
                status[api] = $"挂起 (剩余时间: {remainingTime:mm\\:ss})";
            }
            else
            {
                status[api] = "正常";
            }
        }
        return status;
    }

    public static async Task<(int, string)> GetConvertApiAsync(int accountId, string parseEndpoint)
    {
        var endpoints = await TkEndpointCore.GetEndpointsByAccountAsync(accountId);
        if (endpoints == null || !endpoints.Any(e => e.status)) return (0, string.Empty);

        // 1. 过滤可用端点（状态正常）
        var availableApis = endpoints.Where(e => e.status);

        // 2. 如果指定了 parseEndpoint，按 ID 进一步过滤
        if (!string.IsNullOrEmpty(parseEndpoint))
        {
            var endpointIds = parseEndpoint.Split(',')
                .Select(idStr => int.TryParse(idStr.Trim(), out var id) ? id : (int?)null)
                .Where(id => id.HasValue)
                .Select(id => id.Value)
                .ToHashSet();

            // 添加日志，看看过滤前后的端点
            var beforeCount = availableApis.Count();
            availableApis = availableApis.Where(e => endpointIds.Contains(e.ep_id));
            var afterCount = availableApis.Count();

            if (!availableApis.Any())
                return (0, string.Empty);
        }

        var state = accountStates.GetOrAdd(accountId, _ => new AccountEndpointState());
        var now = DateTime.Now;

        // 3. 先找出所有未被挂起、未达限制的端点
        var availableEndpoints = availableApis
            .Where(e => !state.IsEndpointSuspended(e.endpoint))
            .Where(e => e.hourly_calls_limit <= 0 || e.current_hourly_calls < e.hourly_calls_limit)
            .Where(e => e.daily_calls_limit <= 0 || e.current_daily_calls < e.daily_calls_limit)
            .ToArray();

        // 4. 如果存在可用端点，直接返回
        if (availableEndpoints.Length > 0)
        {
            // 生成唯一的轮询key，确保每种可用端点组合都有独立索引
            string roundRobinKey = $"{accountId}:{string.Join(",", availableEndpoints.Select(e => e.ep_id).OrderBy(x => x))}";
            var currentIndex = state.GetOrAddAccountIndex(roundRobinKey);
            var newIndex = (currentIndex + 1) % availableEndpoints.Length;
            state.UpdateAccountIndex(roundRobinKey, newIndex);
            var selectedEndpoint = availableEndpoints[newIndex];

            // 异步更新调用统计（不阻塞当前请求）
            _ = UpdateEndpointStatsAsync(accountId, selectedEndpoint, now);
            return (selectedEndpoint.ep_id, selectedEndpoint.endpoint);
        }

        // 5. 检查是否仅因时间间隔限制
        var intervalLimitedEndpoints = availableApis
            .Where(e => !state.IsEndpointSuspended(e.endpoint))
            .Where(e => e.hourly_calls_limit <= 0 || e.current_hourly_calls < e.hourly_calls_limit)
            .Where(e => e.daily_calls_limit <= 0 || e.current_daily_calls < e.daily_calls_limit)
            .Where(e => e.interval_seconds > 0 &&
                       e.last_call_time != null &&
                       (now - e.last_call_time).TotalSeconds < e.interval_seconds)
            .ToArray();

        if (intervalLimitedEndpoints.Length > 0)
        {
            // 如果是时间间隔限制，返回空字符串
            return (0, string.Empty);
        }

        // 6. 检查所有端点的状态
        var allEndpointsSuspended = availableApis.All(e => state.IsEndpointSuspended(e.endpoint));
        var allEndpointsLimited = availableApis.All(e => 
            (e.hourly_calls_limit > 0 && e.current_hourly_calls >= e.hourly_calls_limit) ||
            (e.daily_calls_limit > 0 && e.current_daily_calls >= e.daily_calls_limit));

        // 如果所有端点都被挂起或达到限制，返回 "ALL"
        if (allEndpointsSuspended || allEndpointsLimited)
        {
            return (0, "ALL");
        }

        // 7. 其他情况（理论上不应该到达这里）
        return (0, string.Empty);
    }

    // 辅助方法：异步更新调用统计（原逻辑，仅拆分以保持清晰）
    private static async Task UpdateEndpointStatsAsync(int accountId, TkEndpointConfigDTO selectedEndpoint, DateTime now)
    {
        var currentHour = now.ToString("yyyyMMddHH");
        var currentDay = now.ToString("yyyyMMdd");

        // 更新调用计数
        selectedEndpoint.current_hourly_calls++;
        selectedEndpoint.current_daily_calls++;
        selectedEndpoint.last_call_time = now;

        // 保存到 Redis
        await RiskControlCore.SetTkEndpointCallsAsync(accountId, selectedEndpoint.id, currentHour, selectedEndpoint.current_hourly_calls);
        await RiskControlCore.SetTkEndpointCallsAsync(accountId, selectedEndpoint.id, currentDay, selectedEndpoint.current_daily_calls);

        // 检查是否需要挂起（原逻辑）
        if (selectedEndpoint.hourly_calls_limit > 0 && selectedEndpoint.current_hourly_calls >= selectedEndpoint.hourly_calls_limit)
        {
            _ = SuspendAsync(accountId, selectedEndpoint.endpoint, SuspendReason.HourlyLimit, selectedEndpoint.suspend_duration);
        }
        if (selectedEndpoint.daily_calls_limit > 0 && selectedEndpoint.current_daily_calls >= selectedEndpoint.daily_calls_limit)
        {
            _ = SuspendAsync(accountId, selectedEndpoint.endpoint, SuspendReason.DailyLimit, selectedEndpoint.suspend_duration);
        }
    }

    public static async Task SuspendAsync(int accountId, string endpoint, SuspendReason reason = SuspendReason.Passive, int? customHoldMinutes = null)
    {
        // 获取或创建账号特定的信号量
        var semaphore = accountSemaphores.GetOrAdd(accountId, _ => new SemaphoreSlim(1, 1));

        await semaphore.WaitAsync();
        try
        {
            // 在锁外部获取所需数据（避免在锁内await）
            var holdMinutes = await GetEndpointHoldMinutesAsync(accountId);

            // 如果是被动暂停，检查冷却时间
            if (lastSuspendTimes.TryGetValue(accountId, out var lastSuspendTime) &&
                DateTime.UtcNow - lastSuspendTime < SuspendCooldown)
            {
                return;
            }

            // 确定挂起时长
            TimeSpan suspendDuration;
            if (customHoldMinutes.HasValue)
            {
                suspendDuration = TimeSpan.FromMinutes(customHoldMinutes.Value);
            }
            else
            {
                suspendDuration = reason switch
                {
                    SuspendReason.HourlyLimit => CalculateHourlySuspendDuration(),
                    SuspendReason.DailyLimit => CalculateDailySuspendDuration(),
                    _ => TimeSpan.FromMinutes(holdMinutes.GetValueOrDefault(endpoint, 240))
                };
            }

            // 执行挂起操作
            var state = accountStates.GetOrAdd(accountId, _ => new AccountEndpointState());
            state.SuspendEndpoint(endpoint, suspendDuration);

            lastSuspendTimes[accountId] = DateTime.UtcNow;

            _ = new LoggerLibrary("转链接口风控", $"{accountId}").Info($"{accountId}_{endpoint}", reason.ToString()).SaveAsync();
            _ = NotifyCore.NotifyAsync($"【转链接口风控】{accountId}_{endpoint} ({reason})");
        }
        finally
        {
            semaphore.Release();
        }
    }


    /// <summary>
    /// 主动释放指定账号的挂起状态
    /// </summary>
    /// <param name="accountId">账号ID</param>
    /// <param name="endpoint">要释放的端点名称，如果为null则释放所有端点的挂起状态</param>
    public static void ReleaseSuspend(int accountId, string endpoint = null)
    {
        // 获取或创建账号特定的锁对象
        var accountLock = accountLocks.GetOrAdd(accountId, _ => new object());

        lock (accountLock)
        {
            if (accountStates.TryGetValue(accountId, out var state))
            {
                if (endpoint == null)
                {
                    // 释放所有端点的挂起状态
                    state.suspendedEndpoints.Clear();
                }
                else
                {
                    // 释放指定端点的挂起状态
                    state.suspendedEndpoints.TryRemove(endpoint, out _);
                }

                // 记录日志
                var action = endpoint == null ? "释放所有挂起" : $"释放挂起({endpoint})";
                _ = new LoggerLibrary("转链接口风控", $"{accountId}").Info($"{accountId}", action).SaveAsync();
                _ = NotifyCore.NotifyAsync($"【转链接口风控】{accountId} {action}");
            }
        }
    }


    private static TimeSpan CalculateHourlySuspendDuration()
    {
        var now = DateTime.UtcNow;
        var nextHour = now.AddHours(1).Date.AddHours(now.Hour + 1); // 下一个整点（如 14:30 → 15:00）
        return nextHour - now;
    }

    private static TimeSpan CalculateDailySuspendDuration()
    {
        var now = DateTime.UtcNow;
        var tomorrow = now.Date.AddDays(1); // 次日零点
        return tomorrow - now;
    }

    public enum SuspendReason
    {
        /// <summary>
        /// 主动暂停（手动操作）
        /// </summary>
        Active,

        /// <summary>
        /// 风控导致的被动暂停
        /// </summary>
        Passive,

        /// <summary>
        /// 每小时总量限制触发的暂停
        /// </summary>
        HourlyLimit,

        /// <summary>
        /// 每日总量限制触发的暂停
        /// </summary>
        DailyLimit
    }

    private class AccountEndpointState
    {
        internal readonly ConcurrentDictionary<string, DateTime> suspendedEndpoints = new();

        // 为每个账号和 parseEndpoint 组合维护独立的索引
        private readonly ConcurrentDictionary<string, int> roundRobinIndices = new();

        public bool IsEndpointSuspended(string endpoint)
        {
            if (!suspendedEndpoints.TryGetValue(endpoint, out var suspendUntil))
                return false;

            if (DateTime.Now >= suspendUntil)
            {
                suspendedEndpoints.TryRemove(endpoint, out _);
                return false;
            }
            return true;
        }

        public DateTime? GetSuspendTime(string endpoint)
        {
            if (suspendedEndpoints.TryGetValue(endpoint, out var suspendUntil))
            {
                return suspendUntil;
            }
            return null;
        }

        public void SuspendEndpoint(string endpoint, TimeSpan duration)
        {
            suspendedEndpoints[endpoint] = DateTime.Now.Add(duration);
        }

        public int GetOrAddAccountIndex(string key) => roundRobinIndices.GetOrAdd(key, -1);
        public void UpdateAccountIndex(string key, int newIndex) => roundRobinIndices[key] = newIndex;
    }


}