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 accountStates = new(); // 记录每个账号最后一次挂起操作的时间 private static readonly ConcurrentDictionary lastSuspendTimes = new(); // 短时缓存挂起状态,削峰同一波请求对 Redis Exists 的放大访问 private static readonly ConcurrentDictionary suspendStateCache = new(); private static readonly ConcurrentDictionary 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> 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(); } // 从数据库获取端点的挂起时间配置 private static async Task> GetEndpointHoldMinutesAsync(int accountId, bool? isTaobaoUrl = null, string parseEndpoint = null) { var endpoints = await TkEndpointCore.GetEndpointsByAccountAsync(accountId, isTaobaoUrl, parseEndpoint); var dict = new ConcurrentDictionary(); 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; } /// /// 检查指定账号是否至少拥有一个可用的endpoint /// /// 账号ID /// true如果至少有一个可用endpoint,否则false public static async Task 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)); } /// /// 获取多个账号的节点状态信息 /// public static async Task>> GetStatus(IEnumerable accountIds) { var result = new Dictionary>(); foreach (var accountId in accountIds.Distinct()) { var availableApis = await GetAllAvailableApisAsync(accountId); result[accountId] = await GetStatus(accountId, availableApis); } return result; } /// /// 获取指定账号的节点状态信息 /// public static async Task> GetStatus(int accountId, Dictionary availableApis) { var status = new Dictionary(); 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})"); } } /// /// 主动释放指定账号的挂起状态 /// /// 账号ID /// 要释放的端点名称,如果为null则释放所有端点的挂起状态 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 { /// /// 主动暂停(手动操作) /// Active, /// /// 风控导致的被动暂停 /// Passive, /// /// 每小时总量限制触发的暂停 /// HourlyLimit, /// /// 每日总量限制触发的暂停 /// DailyLimit, /// /// 上游明确返回当日调用达到上限 /// RemoteDailyLimit } private class AccountEndpointState { // 只保留轮询索引 private readonly ConcurrentDictionary 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 GetSuspendStates(int accountId, IEnumerable endpoints) { var result = new Dictionary(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); }