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
{
    // 只保留轮询索引
    private static readonly ConcurrentDictionary<int, AccountEndpointState> accountStates = new();

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

    // 短时缓存挂起状态，削峰同一波请求对 Redis Exists 的放大访问
    private static readonly ConcurrentDictionary<string, SuspendCacheEntry> suspendStateCache = new();
    private static readonly ConcurrentDictionary<string, byte> suspendStateRefreshInFlight = new();
    private static readonly TimeSpan SuspendStateCacheDuration = TimeSpan.FromSeconds(10);
    private static readonly TimeSpan SuspendStateFailureBackoffDuration = TimeSpan.FromSeconds(3);


    public static void Refresh()
    {
        accountStates.Clear();
        lastSuspendTimes.Clear();
        suspendStateCache.Clear();
        suspendStateRefreshInFlight.Clear();
    }


    // 从数据库获取所有可用的API端点
    private static async Task<Dictionary<int, string>> GetAllAvailableApisAsync(int accountId, bool? isTaobaoUrl = null, string parseEndpoint = null)
    {
        var endpoints = await TkEndpointCore.GetEndpointsByAccountAsync(accountId, isTaobaoUrl, parseEndpoint);
        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, bool? isTaobaoUrl = null, string parseEndpoint = null)
    {
        var endpoints = await TkEndpointCore.GetEndpointsByAccountAsync(accountId, isTaobaoUrl, parseEndpoint);
        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 (string.IsNullOrWhiteSpace(endpoint)) return false;

        string key = BuildSuspendKey(accountId, endpoint);
        if (TryGetCachedSuspendState(key, out var isSuspended))
            return isSuspended;

        // A daily-limit account must remain unavailable after a process restart. Perform one
        // synchronous Redis check for AI Open on cache miss; subsequent checks use the cache.
        if (string.Equals(endpoint, AiOpenRiskControlCore.EndpointName, StringComparison.Ordinal))
        {
            try
            {
                isSuspended = RedisHelper.Exists(key);
                CacheSuspendState(
                    key,
                    isSuspended,
                    isSuspended ? SuspendStateCacheDuration : SuspendStateFailureBackoffDuration);
                return isSuspended;
            }
            catch (Exception)
            {
                CacheSuspendState(key, false, SuspendStateFailureBackoffDuration);
                return false;
            }
        }

        // 热路径不再同步访问 Redis。缓存未命中时先降级放行，再异步探测 Redis 状态。
        _ = RefreshSuspendStateAsync(key);
        CacheSuspendState(key, false, SuspendStateFailureBackoffDuration);
        return false;
    }

    /// <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; // 如果账号没有状态记录，所有节点都是正常的
        }

        var suspendStates = GetSuspendStates(accountId, availableApis.Values);
        return availableApis.Any(api => !suspendStates.GetValueOrDefault(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;
        }

        var suspendStates = GetSuspendStates(accountId, availableApis.Values);
        foreach ((var api_id, var api) in availableApis)
        {
            if (suspendStates.GetValueOrDefault(api))
            {
                // Redis没有挂起到期时间，展示"挂起"即可
                status[api] = $"挂起";
            }
            else
            {
                status[api] = "正常";
            }
        }
        return status;
    }

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

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

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

        // 2. 先找出所有未被挂起、未达限制的端点
        var availableEndpoints = availableApis
            .Where(e => !suspendStates.GetValueOrDefault(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();

        // 3. 如果存在可用端点，直接返回
        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];

            // 异步更新调用统计（不阻塞当前请求）
            if (selectedEndpoint.ep_id == AiOpenRiskControlCore.EndpointId)
            {
                // The returned daily counter is used as the observed AI Open trigger point,
                // so endpoint 8 must finish its atomic Redis increment before the HTTP call.
                await UpdateEndpointStatsAsync(accountId, selectedEndpoint, now);
            }
            else
            {
                _ = UpdateEndpointStatsAsync(accountId, selectedEndpoint, now);
            }
            return (selectedEndpoint.ep_id, selectedEndpoint.endpoint);
        }

        // 4. 检查是否仅因时间间隔限制
        var intervalLimitedEndpoints = availableApis
            .Where(e => !suspendStates.GetValueOrDefault(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);
        }

        // 5. 检查所有端点的状态
        var allEndpointsSuspended = availableApis.All(e => suspendStates.GetValueOrDefault(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");
        }

        // 6. 其他情况（理论上不应该到达这里）
        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, bool notifyOtherNodes = true)
    {
        var holdMinutes = await GetEndpointHoldMinutesAsync(accountId);
        TimeSpan suspendDuration;
        if (customHoldMinutes.HasValue)
        {
            suspendDuration = TimeSpan.FromMinutes(customHoldMinutes.Value);
        }
        else
        {
            suspendDuration = reason switch
            {
                SuspendReason.HourlyLimit => CalculateHourlySuspendDuration(),
                SuspendReason.DailyLimit or SuspendReason.RemoteDailyLimit => CalculateDailySuspendDuration(),
                _ => TimeSpan.FromMinutes(holdMinutes.GetValueOrDefault(endpoint, 240))
            };
        }

        await SuspendForDurationAsync(accountId, endpoint, suspendDuration, reason, notifyOtherNodes);
    }

    public static async Task SuspendForDurationAsync(
        int accountId,
        string endpoint,
        TimeSpan suspendDuration,
        SuspendReason reason = SuspendReason.Passive,
        bool notifyOtherNodes = true,
        bool emitNotification = true)
    {
        if (string.IsNullOrWhiteSpace(endpoint)) return;

        string key = BuildSuspendKey(accountId, endpoint);
        CacheSuspendState(key, true, suspendDuration);

        try
        {
            RedisHelper.Set(key, 1, (int)Math.Ceiling(suspendDuration.TotalSeconds));
        }
        catch (Exception)
        {
            // 本地状态已写入，Redis 持久化失败时只降级，不阻塞主流程
        }

        if (notifyOtherNodes)
        {
            _ = EndPointCore.NotifyChangeSuspend(accountId, endpoint, release: false, durationSeconds: (int)Math.Ceiling(suspendDuration.TotalSeconds));
        }

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


    /// <summary>
    /// 主动释放指定账号的挂起状态
    /// </summary>
    /// <param name="accountId">账号ID</param>
    /// <param name="endpoint">要释放的端点名称，如果为null则释放所有端点的挂起状态</param>
    public static void ReleaseSuspend(int accountId, string endpoint = null, bool notifyOtherNodes = true)
    {
        if (endpoint == null)
        {
            foreach (var key in suspendStateCache.Keys.Where(key => key.StartsWith($"tk_suspend:{accountId}:", StringComparison.Ordinal)))
            {
                suspendStateCache.TryRemove(key, out _);
            }

            try
            {
                var pattern = $"tk_suspend:{accountId}:*";
                var keys = RedisHelper.Keys(pattern);
                foreach (var key in keys)
                {
                    RedisHelper.Del(key);
                }
            }
            catch (Exception)
            {
                // Redis 不可用时只清理本地状态
            }
        }
        else
        {
            string key = BuildSuspendKey(accountId, endpoint);
            CacheSuspendState(key, false, SuspendStateCacheDuration);

            try
            {
                RedisHelper.Del(key);
            }
            catch (Exception)
            {
                // Redis 不可用时只清理本地状态
            }
        }

        if (notifyOtherNodes)
        {
            _ = EndPointCore.NotifyChangeSuspend(accountId, endpoint, release: true);
        }

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


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

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

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

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

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

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

        /// <summary>
        /// 上游明确返回当日调用达到上限
        /// </summary>
        RemoteDailyLimit
    }

    private class AccountEndpointState
    {
        // 只保留轮询索引
        private readonly ConcurrentDictionary<string, int> roundRobinIndices = new();
        public int GetOrAddAccountIndex(string key) => roundRobinIndices.GetOrAdd(key, -1);
        public void UpdateAccountIndex(string key, int newIndex) => roundRobinIndices[key] = newIndex;
    }

    private static string BuildSuspendKey(int accountId, string endpoint) => $"tk_suspend:{accountId}:{endpoint}";

    private static Dictionary<string, bool> GetSuspendStates(int accountId, IEnumerable<string> endpoints)
    {
        var result = new Dictionary<string, bool>(StringComparer.Ordinal);
        foreach (var endpoint in endpoints.Where(e => !string.IsNullOrWhiteSpace(e)).Distinct(StringComparer.Ordinal))
        {
            result[endpoint] = IsEndpointSuspended(accountId, endpoint);
        }
        return result;
    }

    private static bool TryGetCachedSuspendState(string key, out bool isSuspended)
    {
        if (suspendStateCache.TryGetValue(key, out var entry))
        {
            if (entry.ExpiresAtUtc > DateTime.UtcNow)
            {
                isSuspended = entry.IsSuspended;
                return true;
            }

            suspendStateCache.TryRemove(key, out _);
        }

        isSuspended = false;
        return false;
    }

    private static void CacheSuspendState(string key, bool isSuspended, TimeSpan duration)
    {
        if (duration <= TimeSpan.Zero)
        {
            suspendStateCache.TryRemove(key, out _);
            return;
        }

        suspendStateCache[key] = new SuspendCacheEntry(isSuspended, DateTime.UtcNow.Add(duration));
    }

    private static Task RefreshSuspendStateAsync(string key)
    {
        if (!suspendStateRefreshInFlight.TryAdd(key, 0))
            return Task.CompletedTask;

        return Task.Run(() =>
        {
            try
            {
                bool isSuspended = RedisHelper.Exists(key);
                CacheSuspendState(
                    key,
                    isSuspended,
                    isSuspended ? SuspendStateCacheDuration : SuspendStateFailureBackoffDuration);
            }
            catch (Exception)
            {
                CacheSuspendState(key, false, SuspendStateFailureBackoffDuration);
            }
            finally
            {
                suspendStateRefreshInFlight.TryRemove(key, out _);
            }
        });
    }

    private readonly record struct SuspendCacheEntry(bool IsSuspended, DateTime ExpiresAtUtc);

}