diff --git a/src/providers/claude/claude-kiro.js b/src/providers/claude/claude-kiro.js index 8b0adc0..9bf892d 100644 --- a/src/providers/claude/claude-kiro.js +++ b/src/providers/claude/claude-kiro.js @@ -11,7 +11,6 @@ import { countTokens } from '@anthropic-ai/tokenizer'; import { configureAxiosProxy } from '../../utils/proxy-utils.js'; import { isRetryableNetworkError, MODEL_PROVIDER } from '../../utils/common.js'; import { getProviderPoolManager } from '../../services/service-manager.js'; -import { CredentialCacheManager } from '../../utils/credential-cache-manager.js'; const KIRO_THINKING = { MAX_BUDGET_TOKENS: 24576, @@ -465,14 +464,10 @@ async initializeAuth(forceRefresh = false) { return; } - // 获取凭证文件路径，用于去重锁的 key + // 获取凭证文件路径 const tokenFilePath = this.credsFilePath || path.join(this.credPath, KIRO_AUTH_TOKEN_FILE); - // 获取凭证缓存管理器 - const credentialCache = CredentialCacheManager.getInstance(); - const providerType = 'claude-kiro-oauth'; - - // Helper to load credentials from a file (fallback when cache miss) + // Helper to load credentials from a file const loadCredentialsFromFile = async (filePath) => { try { const fileContent = await fs.readFile(filePath, 'utf8'); @@ -507,55 +502,41 @@ async initializeAuth(forceRefresh = false) { } }; - // Helper to save credentials - 使用内存缓存替代直接文件写入 + // Helper to save credentials const saveCredentialsToFile = async (filePath, newData) => { - // 优先更新内存缓存 - if (this.uuid && credentialCache.hasCredentials(providerType, this.uuid)) { - const entry = credentialCache.getCredentials(providerType, this.uuid); - if (entry) { - const mergedData = { ...entry.credentials, ...newData }; - credentialCache.updateCredentials(providerType, this.uuid, mergedData, filePath); - console.info(`[Kiro Auth] Updated credentials in memory cache: ${this.uuid}`); - return; - } - } - - // 如果没有缓存条目，使用内存锁方式写入 - await credentialCache.withMemoryLock(`kiro-save:${filePath}`, async () => { - let existingData = {}; + let existingData = {}; + try { + const fileContent = await fs.readFile(filePath, 'utf8'); try { - const fileContent = await fs.readFile(filePath, 'utf8'); + existingData = JSON.parse(fileContent); + } catch (parseError) { + console.warn('[Kiro Auth] JSON parse failed, attempting repair...'); try { - existingData = JSON.parse(fileContent); - } catch (parseError) { - console.warn('[Kiro Auth] JSON parse failed, attempting repair...'); - try { - const repaired = repairJson(fileContent); - existingData = JSON.parse(repaired); - console.info('[Kiro Auth] JSON repair successful'); - } catch (repairError) { - console.warn('[Kiro Auth] JSON repair failed, attempting field extraction...'); - const extracted = extractCredentialsFromCorruptedJson(fileContent); - if (extracted) { - existingData = extracted; - console.info('[Kiro Auth] Field extraction successful'); - } else { - console.error('[Kiro Auth] All recovery methods failed:', repairError.message); - existingData = {}; - } + const repaired = repairJson(fileContent); + existingData = JSON.parse(repaired); + console.info('[Kiro Auth] JSON repair successful'); + } catch (repairError) { + console.warn('[Kiro Auth] JSON repair failed, attempting field extraction...'); + const extracted = extractCredentialsFromCorruptedJson(fileContent); + if (extracted) { + existingData = extracted; + console.info('[Kiro Auth] Field extraction successful'); + } else { + console.error('[Kiro Auth] All recovery methods failed:', repairError.message); + existingData = {}; } } - } catch (readError) { - if (readError.code === 'ENOENT') { - console.debug(`[Kiro Auth] Token file not found, creating new one: ${filePath}`); - } else { - console.warn(`[Kiro Auth] Could not read existing token file ${filePath}: ${readError.message}`); - } } - const mergedData = { ...existingData, ...newData }; - await fs.writeFile(filePath, JSON.stringify(mergedData, null, 2), 'utf8'); - console.info(`[Kiro Auth] Updated token file: ${filePath}`); - }); + } catch (readError) { + if (readError.code === 'ENOENT') { + console.debug(`[Kiro Auth] Token file not found, creating new one: ${filePath}`); + } else { + console.warn(`[Kiro Auth] Could not read existing token file ${filePath}: ${readError.message}`); + } + } + const mergedData = { ...existingData, ...newData }; + await fs.writeFile(filePath, JSON.stringify(mergedData, null, 2), 'utf8'); + console.info(`[Kiro Auth] Updated token file: ${filePath}`); }; try { @@ -568,43 +549,34 @@ async initializeAuth(forceRefresh = false) { this.base64Creds = null; } - // Priority 2: 尝试从内存缓存加载凭证 - if (this.uuid && credentialCache.hasCredentials(providerType, this.uuid)) { - const cachedEntry = credentialCache.getCredentials(providerType, this.uuid); - if (cachedEntry && cachedEntry.credentials) { - Object.assign(mergedCredentials, cachedEntry.credentials); - console.info(`[Kiro Auth] Successfully loaded credentials from memory cache: ${this.uuid}`); + // 从文件加载 + const targetFilePath = this.credsFilePath || path.join(this.credPath, KIRO_AUTH_TOKEN_FILE); + const dirPath = path.dirname(targetFilePath); + const targetFileName = path.basename(targetFilePath); + + console.debug(`[Kiro Auth] Loading credentials from directory: ${dirPath}`); + + try { + const targetCredentials = await loadCredentialsFromFile(targetFilePath); + if (targetCredentials) { + Object.assign(mergedCredentials, targetCredentials); + console.info(`[Kiro Auth] Successfully loaded OAuth credentials from ${targetFilePath}`); } - } else { - // Priority 3: 从文件加载（缓存未命中时的回退） - const targetFilePath = this.credsFilePath || path.join(this.credPath, KIRO_AUTH_TOKEN_FILE); - const dirPath = path.dirname(targetFilePath); - const targetFileName = path.basename(targetFilePath); - console.debug(`[Kiro Auth] Cache miss, loading credentials from directory: ${dirPath}`); - - try { - const targetCredentials = await loadCredentialsFromFile(targetFilePath); - if (targetCredentials) { - Object.assign(mergedCredentials, targetCredentials); - console.info(`[Kiro Auth] Successfully loaded OAuth credentials from ${targetFilePath}`); - } - - const files = await fs.readdir(dirPath); - for (const file of files) { - if (file.endsWith('.json') && file !== targetFileName) { - const filePath = path.join(dirPath, file); - const credentials = await loadCredentialsFromFile(filePath); - if (credentials) { - credentials.expiresAt = mergedCredentials.expiresAt; - Object.assign(mergedCredentials, credentials); - console.debug(`[Kiro Auth] Loaded Client credentials from ${file}`); - } + const files = await fs.readdir(dirPath); + for (const file of files) { + if (file.endsWith('.json') && file !== targetFileName) { + const filePath = path.join(dirPath, file); + const credentials = await loadCredentialsFromFile(filePath); + if (credentials) { + credentials.expiresAt = mergedCredentials.expiresAt; + Object.assign(mergedCredentials, credentials); + console.debug(`[Kiro Auth] Loaded Client credentials from ${file}`); } } - } catch (error) { - console.warn(`[Kiro Auth] Error loading credentials from directory ${dirPath}: ${error.message}`); } + } catch (error) { + console.warn(`[Kiro Auth] Error loading credentials from directory ${dirPath}: ${error.message}`); } // Apply loaded credentials @@ -636,15 +608,7 @@ async initializeAuth(forceRefresh = false) { throw new Error('No refresh token available to refresh access token.'); } - // 使用内存锁替代文件锁进行去重 - const dedupeKey = `kiro-token-refresh:${tokenFilePath}`; - await credentialCache.withDeduplication(dedupeKey, async () => { - await this._doTokenRefresh(saveCredentialsToFile, tokenFilePath); - }); - - // 刷新完成后（无论是自己执行还是等待其他请求），都从缓存重新加载凭证 - // 确保所有并发请求都能获取到最新的 token - await this._reloadCredentialsAfterRefresh(tokenFilePath); + await this._doTokenRefresh(saveCredentialsToFile, tokenFilePath); } if (!this.accessToken) { @@ -709,58 +673,6 @@ async initializeAuth(forceRefresh = false) { } } - /** - * 在并发刷新完成后重新加载凭证（内部方法） - * @param {string} tokenFilePath - 凭证文件路径 - */ - async _reloadCredentialsAfterRefresh(tokenFilePath) { - // 优先从内存缓存加载 - const credentialCache = CredentialCacheManager.getInstance(); - const providerType = 'claude-kiro-oauth'; - - if (this.uuid && credentialCache.hasCredentials(providerType, this.uuid)) { - const cachedEntry = credentialCache.getCredentials(providerType, this.uuid); - if (cachedEntry && cachedEntry.credentials) { - this.accessToken = cachedEntry.credentials.accessToken; - this.refreshToken = cachedEntry.credentials.refreshToken; - this.expiresAt = cachedEntry.credentials.expiresAt; - if (cachedEntry.credentials.profileArn) { - this.profileArn = cachedEntry.credentials.profileArn; - } - console.debug('[Kiro Auth] Credentials reloaded from memory cache after concurrent refresh'); - return; - } - } - - // 回退到文件加载 - try { - const fileContent = await fs.readFile(tokenFilePath, 'utf8'); - let credentials; - try { - credentials = JSON.parse(fileContent); - } catch (parseError) { - console.warn('[Kiro Auth] JSON parse failed, attempting repair...'); - try { - const repaired = repairJson(fileContent); - credentials = JSON.parse(repaired); - console.info('[Kiro Auth] JSON repair successful'); - } catch (repairError) { - console.error('[Kiro Auth] JSON repair failed:', repairError.message); - throw new Error(`Failed to parse credentials file after repair attempt: ${repairError.message}`); - } - } - this.accessToken = credentials.accessToken; - this.refreshToken = credentials.refreshToken; - this.expiresAt = credentials.expiresAt; - if (credentials.profileArn) { - this.profileArn = credentials.profileArn; - } - console.debug('[Kiro Auth] Credentials reloaded from file after concurrent refresh'); - } catch (error) { - console.warn(`[Kiro Auth] Failed to reload credentials after refresh: ${error.message}`); - throw error; - } - } /** * Extract text content from OpenAI message format @@ -2657,22 +2569,12 @@ async initializeAuth(forceRefresh = false) { * 后台异步刷新 token（不阻塞当前请求） */ triggerBackgroundRefresh() { - const tokenFilePath = this.credsFilePath || path.join(this.credPath, KIRO_AH_TOKEN_FILE); - const dedupeKey = `kiro-token-refresh:${tokenFilePath}`; - const credentialCache = CredentialCacheManager.getInstance(); - - // 使用 withDeduplication 确保只有一个刷新任务在执行 - credentialCache.withDeduplication(dedupeKey, async () => { - console.log('[Kiro] Background token refresh started...'); - try { - await this.initializeAuth(true); - console.log('[Kiro] Background token refresh completed successfully'); - } catch (error) { - console.error('[Kiro] Background token refresh failed:', error.message); - // 后台刷新失败不抛出错误，下次请求会重试 - } - }).catch(() => { - // 忽略后台刷新错误，不影响当前请求 + console.log('[Kiro] Background token refresh started...'); + this.initializeAuth(true).then(() => { + console.log('[Kiro] Background token refresh completed successfully'); + }).catch((error) => { + console.error('[Kiro] Background token refresh failed:', error.message); + // 后台刷新失败不抛出错误，下次请求会重试 }); } diff --git a/src/providers/gemini/antigravity-core.js b/src/providers/gemini/antigravity-core.js index 4c7c2cc..c2e6412 100644 --- a/src/providers/gemini/antigravity-core.js +++ b/src/providers/gemini/antigravity-core.js @@ -14,7 +14,6 @@ import { getProviderModels } from '../provider-models.js'; import { handleGeminiAntigravityOAuth } from '../../auth/oauth-handlers.js'; import { getProxyConfigForProvider, getGoogleAuthProxyConfig } from '../../utils/proxy-utils.js'; import { cleanJsonSchemaProperties } from '../../converters/utils.js'; -import { CredentialCacheManager } from '../../utils/credential-cache-manager.js'; // 配置 HTTP/HTTPS agent 限制连接池大小，避免资源泄漏 const httpAgent = new http.Agent({ @@ -777,9 +776,6 @@ export class AntigravityApiService { } async initializeAuth(forceRefresh = false) { - const credentialCache = CredentialCacheManager.getInstance(); - const providerType = 'gemini-antigravity'; - // 检查是否需要刷新 Token const needsRefresh = forceRefresh || this.isTokenExpiringSoon(); @@ -788,64 +784,19 @@ export class AntigravityApiService { return; } - // Antigravity 不支持 base64 配置，直接使用文件路径 - const credPath = this.oauthCredsFilePath || path.join(os.homedir(), CREDENTIALS_DIR, CREDENTIALS_FILE); try { - // 优先从内存缓存加载 - let credentials = null; - if (this.uuid && credentialCache.hasCredentials(providerType, this.uuid)) { - const cachedEntry = credentialCache.getCredentials(providerType, this.uuid); - if (cachedEntry && cachedEntry.credentials) { - credentials = cachedEntry.credentials; - console.log('[Antigravity Auth] Loaded credentials from memory cache'); - } - } - - // Fallback: 从文件加载 - if (!credentials) { - const data = await fs.readFile(credPath, "utf8"); - credentials = JSON.parse(data); - console.log('[Antigravity Auth] Loaded credentials from file'); - } - + const data = await fs.readFile(credPath, "utf8"); + const credentials = JSON.parse(data); this.authClient.setCredentials(credentials); console.log('[Antigravity Auth] Authentication configured successfully from file.'); if (needsRefresh) { - // 使用去重锁：多个并发刷新请求只执行一次，共享结果 - const dedupeKey = `antigravity-token-refresh:${credPath}`; - await credentialCache.withDeduplication(dedupeKey, async () => { - console.log('[Antigravity Auth] Token expiring soon or force refresh requested. Refreshing token...'); - const { credentials: newCredentials } = await this.authClient.refreshAccessToken(); - this.authClient.setCredentials(newCredentials); - // 保存刷新后的凭证（优先保存到内存缓存） - if (this.uuid && credentialCache.hasCredentials(providerType, this.uuid)) { - credentialCache.updateCredentials(providerType, this.uuid, newCredentials, credPath); - console.log(`[Antigravity Auth] Token refreshed and saved to memory cache: ${this.uuid}`); - } else { - await this._saveCredentialsToFile(credPath, newCredentials); - } - console.log(`[Antigravity Auth] Token refreshed and saved to ${credPath} successfully.`); - }); - - // 如果是等待其他请求完成的刷新，需要重新加载凭证 - // 因为 withDeduplication 只让第一个调用者执行刷新并更新自己的内存状态 - // 其他等待者需要从缓存重新加载 - if (this.isTokenExpiringSoon()) { - if (this.uuid && credentialCache.hasCredentials(providerType, this.uuid)) { - const cachedEntry = credentialCache.getCredentials(providerType, this.uuid); - if (cachedEntry && cachedEntry.credentials) { - this.authClient.setCredentials(cachedEntry.credentials); - console.log('[Antigravity Auth] Credentials reloaded from memory cache after concurrent refresh'); - } - } else { - const refreshedData = await fs.readFile(credPath, "utf8"); - const refreshedCredentials = JSON.parse(refreshedData); - this.authClient.setCredentials(refreshedCredentials); - console.log('[Antigravity Auth] Credentials reloaded from file after concurrent refresh'); - } - } + console.log('[Antigravity Auth] Token expiring soon or force refresh requested. Refreshing token...'); + const { credentials: newCredentials } = await this.authClient.refreshAccessToken(); + this.authClient.setCredentials(newCredentials); + await this._saveCredentialsToFile(credPath, newCredentials); + console.log(`[Antigravity Auth] Token refreshed and saved to ${credPath} successfully.`); } } catch (error) { console.error('[Antigravity Auth] Error initializing authentication:', error.code); @@ -921,30 +872,18 @@ export class AntigravityApiService { } /** - * 保存凭证到文件（优先使用内存缓存） + * 保存凭证到文件 * @param {string} filePath - 凭证文件路径 * @param {Object} credentials - 凭证数据 */ async _saveCredentialsToFile(filePath, credentials) { - const credentialCache = CredentialCacheManager.getInstance(); - const providerType = 'gemini-antigravity'; - - // 优先保存到内存缓存 - if (this.uuid && credentialCache.hasCredentials(providerType, this.uuid)) { - credentialCache.updateCredentials(providerType, this.uuid, credentials, filePath); - console.log(`[Antigravity Auth] Credentials saved to memory cache: ${this.uuid}`); - return; + try { + await fs.writeFile(filePath, JSON.stringify(credentials, null, 2)); + console.log(`[Antigravity Auth] Credentials saved to ${filePath}`); + } catch (error) { + console.error(`[Antigravity Auth] Failed to save credentials to ${filePath}: ${error.message}`); + throw error; } - - // Fallback: 使用内存锁的文件写入 - await credentialCache.withMemoryLock(`antigravity-save:${filePath}`, async () => { - try { - await fs.writeFile(filePath, JSON.stringify(credentials, null, 2)); - console.log(`[Antigravity Auth] Credentials saved to ${filePath}`); - } catch (error) { - console.error(`[Antigravity Auth] Failed to save credentials to ${filePath}: ${error.message}`); - } - }); } async discoverProjectAndModels() { diff --git a/src/providers/gemini/gemini-core.js b/src/providers/gemini/gemini-core.js index 5f65848..bc4d913 100644 --- a/src/providers/gemini/gemini-core.js +++ b/src/providers/gemini/gemini-core.js @@ -10,7 +10,6 @@ import { API_ACTIONS, formatExpiryTime, isRetryableNetworkError } from '../../ut import { getProviderModels } from '../provider-models.js'; import { handleGeminiCliOAuth } from '../../auth/oauth-handlers.js'; import { getProxyConfigForProvider, getGoogleAuthProxyConfig } from '../../utils/proxy-utils.js'; -import { CredentialCacheManager } from '../../utils/credential-cache-manager.js'; // 配置 HTTP/HTTPS agent 限制连接池大小，避免资源泄漏 const httpAgent = new http.Agent({ @@ -273,25 +272,11 @@ export class GeminiApiService { if (forceRefresh) { console.log('[Gemini Auth] Forcing token refresh...'); - - // 使用去重锁：多个并发刷新请求只执行一次，共享结果 - const dedupeKey = `gemini-token-refresh:${credPath}`; - const credentialCache = CredentialCacheManager.getInstance(); - await credentialCache.withDeduplication(dedupeKey, async () => { - const { credentials: newCredentials } = await this.authClient.refreshAccessToken(); - this.authClient.setCredentials(newCredentials); - // Save refreshed credentials back to file (with file locking) - await this._saveCredentialsToFile(credPath, newCredentials); - console.log('[Gemini Auth] Token refreshed and saved successfully.'); - }); - - // 如果是等待其他请求完成的刷新，需要重新加载凭证 - if (this.isExpiryDateNear()) { - const refreshedData = await fs.readFile(credPath, "utf8"); - const refreshedCredentials = JSON.parse(refreshedData); - this.authClient.setCredentials(refreshedCredentials); - console.log('[Gemini Auth] Credentials reloaded after concurrent refresh'); - } + const { credentials: newCredentials } = await this.authClient.refreshAccessToken(); + this.authClient.setCredentials(newCredentials); + // Save refreshed credentials back to file + await this._saveCredentialsToFile(credPath, newCredentials); + console.log('[Gemini Auth] Token refreshed and saved successfully.'); } } catch (error) { console.error('[Gemini Auth] Error initializing authentication:', error.code); @@ -637,30 +622,18 @@ export class GeminiApiService { } /** - * 保存凭证到文件（使用内存缓存优先） + * 保存凭证到文件 * @param {string} filePath - 凭证文件路径 * @param {Object} credentials - 凭证数据 */ async _saveCredentialsToFile(filePath, credentials) { - // 优先更新内存缓存 - const credentialCache = CredentialCacheManager.getInstance(); - const providerType = 'gemini-cli-oauth'; - - if (this.uuid && credentialCache.hasCredentials(providerType, this.uuid)) { - credentialCache.updateCredentials(providerType, this.uuid, credentials, filePath); - console.log(`[Gemini Auth] Credentials saved to memory cache: ${this.uuid}`); - return; + try { + await fs.writeFile(filePath, JSON.stringify(credentials, null, 2)); + console.log(`[Gemini Auth] Credentials saved to ${filePath}`); + } catch (error) { + console.error(`[Gemini Auth] Failed to save credentials to ${filePath}: ${error.message}`); + throw error; } - - // 回退到使用内存锁写入文件 - await credentialCache.withMemoryLock(`gemini-save:${filePath}`, async () => { - try { - await fs.writeFile(filePath, JSON.stringify(credentials, null, 2)); - console.log(`[Gemini Auth] Credentials saved to ${filePath}`); - } catch (error) { - console.error(`[Gemini Auth] Failed to save credentials to ${filePath}: ${error.message}`); - } - }); } /** diff --git a/src/providers/openai/iflow-core.js b/src/providers/openai/iflow-core.js index 6b75305..6272d4e 100644 --- a/src/providers/openai/iflow-core.js +++ b/src/providers/openai/iflow-core.js @@ -24,7 +24,6 @@ import * as path from 'path'; import * as os from 'os'; import { configureAxiosProxy } from '../../utils/proxy-utils.js'; import { isRetryableNetworkError } from '../../utils/common.js'; -import { CredentialCacheManager } from '../../utils/credential-cache-manager.js'; // iFlow API 端点 const IFLOW_API_BASE_URL = 'https://apis.iflow.cn/v1'; @@ -137,51 +136,28 @@ async function saveTokenToFile(filePath, tokenStorage, uuid = null) { ? filePath : path.join(process.cwd(), filePath); - const credentialCache = CredentialCacheManager.getInstance(); - const providerType = 'openai-iflow'; + try { + // 确保目录存在 + const dir = path.dirname(absolutePath); + await fs.mkdir(dir, { recursive: true }); - // 优先保存到内存缓存 - if (uuid && credentialCache.hasCredentials(providerType, uuid)) { + // 写入文件 const json = tokenStorage.toJSON(); // 验证关键字段是否存在 if (!json.refresh_token || json.refresh_token.trim() === '') { - console.error('[iFlow] WARNING: Attempting to save token with empty refresh_token!'); + console.error('[iFlow] WARNING: Attempting to save token file with empty refresh_token!'); } if (!json.apiKey || json.apiKey.trim() === '') { - console.error('[iFlow] WARNING: Attempting to save token with empty apiKey!'); + console.error('[iFlow] WARNING: Attempting to save token file with empty apiKey!'); } - credentialCache.updateCredentials(providerType, uuid, json, absolutePath); - console.log(`[iFlow] Token saved to memory cache: ${uuid} (refresh_token: ${json.refresh_token ? json.refresh_token.substring(0, 8) + '...' : 'EMPTY'})`); - return; + await fs.writeFile(absolutePath, JSON.stringify(json, null, 2), 'utf-8'); + + console.log(`[iFlow] Token saved to: ${filePath} (refresh_token: ${json.refresh_token ? json.refresh_token.substring(0, 8) + '...' : 'EMPTY'})`); + } catch (error) { + throw new Error(`[iFlow] Failed to save token to file: ${error.message}`); } - - // Fallback: 使用内存锁的文件写入 - await credentialCache.withMemoryLock(`iflow-save:${absolutePath}`, async () => { - try { - // 确保目录存在 - const dir = path.dirname(absolutePath); - await fs.mkdir(dir, { recursive: true }); - - // 写入文件 - const json = tokenStorage.toJSON(); - - // 验证关键字段是否存在 - if (!json.refresh_token || json.refresh_token.trim() === '') { - console.error('[iFlow] WARNING: Attempting to save token file with empty refresh_token!'); - } - if (!json.apiKey || json.apiKey.trim() === '') { - console.error('[iFlow] WARNING: Attempting to save token file with empty apiKey!'); - } - - await fs.writeFile(absolutePath, JSON.stringify(json, null, 2), 'utf-8'); - - console.log(`[iFlow] Token saved to: ${filePath} (refresh_token: ${json.refresh_token ? json.refresh_token.substring(0, 8) + '...' : 'EMPTY'})`); - } catch (error) { - throw new Error(`[iFlow] Failed to save token to file: ${error.message}`); - } - }); } // ==================== Token 刷新逻辑 ==================== @@ -528,9 +504,6 @@ export class IFlowApiService { * @param {boolean} forceRefresh - 是否强制刷新 Token */ async initializeAuth(forceRefresh = false) { - const credentialCache = CredentialCacheManager.getInstance(); - const providerType = 'openai-iflow'; - // 如果已有 API Key 且不强制刷新，直接返回 if (this.apiKey && !forceRefresh) return; @@ -540,18 +513,7 @@ export class IFlowApiService { } try { - // 优先从内存缓存加载 - let tokenData = null; - if (this.uuid && credentialCache.hasCredentials(providerType, this.uuid)) { - const cachedEntry = credentialCache.getCredentials(providerType, this.uuid); - if (cachedEntry && cachedEntry.credentials) { - tokenData = cachedEntry.credentials; - this.tokenStorage = IFlowTokenStorage.fromJSON(tokenData); - console.log('[iFlow Auth] Loaded credentials from memory cache'); - } - } - - // Fallback: 从文件加载 + // 从文件加载 if (!this.tokenStorage) { this.tokenStorage = await loadTokenFromFile(this.tokenFilePath); console.log('[iFlow Auth] Loaded credentials from file'); @@ -589,9 +551,6 @@ export class IFlowApiService { * @returns {Promise} - 是否执行了刷新 */ async _checkAndRefreshTokenIfNeeded() { - const credentialCache = CredentialCacheManager.getInstance(); - const providerType = 'openai-iflow'; - if (!this.tokenStorage) { return false; } @@ -611,33 +570,7 @@ export class IFlowApiService { console.log('[iFlow] Token is expiring soon, attempting refresh...'); try { - // 使用去重锁：多个并发刷新请求只执行一次,共享结果 - const dedupeKey = `iflow-token-refresh:${this.tokenFilePath}`; - await credentialCache.withDeduplication(dedupeKey, async () => { - await this._refreshOAuthTokens(); - }); - - // 如果是等待其他请求完成的刷新，需要重新加载凭证 - if (this.isExpiryDateNear()) { - if (this.uuid && credentialCache.hasCredentials(providerType, this.uuid)) { - const cachedEntry = credentialCache.getCredentials(providerType, this.uuid); - if (cachedEntry && cachedEntry.credentials) { - this.tokenStorage = IFlowTokenStorage.fromJSON(cachedEntry.credentials); - this.apiKey = this.tokenStorage.apiKey; - this.axiosInstance.defaults.headers['Authorization'] = `Bearer ${this.apiKey}`; - console.log('[iFlow] Credentials reloaded from memory cache after concurrent refresh'); - } - } else { - const refreshedStorage = await loadTokenFromFile(this.tokenFilePath); - if (refreshedStorage && refreshedStorage.apiKey) { - this.tokenStorage = refreshedStorage; - this.apiKey = refreshedStorage.apiKey; - this.axiosInstance.defaults.headers['Authorization'] = `Bearer ${this.apiKey}`; - console.log('[iFlow] Credentials reloaded from file after concurrent refresh'); - } - } - } - + await this._refreshOAuthTokens(); return true; } catch (error) { console.error('[iFlow] Token refresh failed:', error.message); diff --git a/src/providers/openai/qwen-core.js b/src/providers/openai/qwen-core.js index 08d81aa..cab299f 100644 --- a/src/providers/openai/qwen-core.js +++ b/src/providers/openai/qwen-core.js @@ -12,7 +12,6 @@ import { getProviderModels } from '../provider-models.js'; import { handleQwenOAuth } from '../../auth/oauth-handlers.js'; import { configureAxiosProxy } from '../../utils/proxy-utils.js'; import { isRetryableNetworkError } from '../../utils/common.js'; -import { CredentialCacheManager } from '../../utils/credential-cache-manager.js'; // --- Constants --- const QWEN_DIR = '.qwen'; @@ -389,18 +388,14 @@ export class QwenApiService { async _cacheQwenCredentials(credentials) { const filePath = this._getQwenCachedCredentialPath(); - const credentialCache = CredentialCacheManager.getInstance(); - // 使用内存锁替代文件锁，防止并发写入 - await credentialCache.withMemoryLock(`qwen-cache:${filePath}`, async () => { - try { - await fs.mkdir(path.dirname(filePath), { recursive: true }); - const credString = JSON.stringify(credentials, null, 2); - await fs.writeFile(filePath, credString); - console.log(`[Qwen Auth] Credentials cached to ${filePath}`); - } catch (error) { - console.error(`[Qwen Auth] Failed to cache credentials to ${filePath}: ${error.message}`); - } - }); + try { + await fs.mkdir(path.dirname(filePath), { recursive: true }); + const credString = JSON.stringify(credentials, null, 2); + await fs.writeFile(filePath, credString); + console.log(`[Qwen Auth] Credentials cached to ${filePath}`); + } catch (error) { + console.error(`[Qwen Auth] Failed to cache credentials to ${filePath}: ${error.message}`); + } } getCurrentEndpoint(resourceUrl) { @@ -794,77 +789,39 @@ class SharedTokenManager { throw new TokenManagerError(TokenError.NO_REFRESH_TOKEN, 'No refresh token available'); } - // 获取凭证缓存管理器 - const credentialCache = CredentialCacheManager.getInstance(); - const providerType = 'openai-qwen-oauth'; - - // 使用内存锁进行去重 - const dedupeKey = `qwen-token-refresh:${context.credentialFilePath}`; - try { - const credentials = await credentialCache.withDeduplication(dedupeKey, async () => { - await this.acquireLock(context); - try { - // 优先从全局缓存检查 - if (context.uuid && credentialCache.hasCredentials(providerType, context.uuid)) { - const cachedEntry = credentialCache.getCredentials(providerType, context.uuid); - if (cachedEntry && cachedEntry.credentials && !forceRefresh && this.isTokenValid(cachedEntry.credentials)) { - context.memoryCache.credentials = cachedEntry.credentials; - qwenClient.setCredentials(cachedEntry.credentials); - return cachedEntry.credentials; - } - } + await this.acquireLock(context); + try { + await this.checkAndReloadIfNeeded(context); - await this.checkAndReloadIfNeeded(context); - - if (!forceRefresh && context.memoryCache.credentials && this.isTokenValid(context.memoryCache.credentials)) { - qwenClient.setCredentials(context.memoryCache.credentials); - return context.memoryCache.credentials; - } - - const response = await qwenClient.refreshAccessToken(); - if (!response || isErrorResponse(response)) { - throw new TokenManagerError(TokenError.REFRESH_FAILED, `Token refresh failed: ${response?.error}`); - } - if (!response.access_token) { - throw new TokenManagerError(TokenError.REFRESH_FAILED, 'No access token in refresh response'); - } - const newCredentials = { - access_token: response.access_token, - token_type: response.token_type, - refresh_token: response.refresh_token || currentCredentials.refresh_token, - resource_url: response.resource_url, - expiry_date: Date.now() + response.expires_in * 1000, - }; - - context.memoryCache.credentials = newCredentials; - qwenClient.setCredentials(newCredentials); - await this.saveCredentialsToFile(context, newCredentials); - console.log('[Qwen Auth] Token refresh response: ok'); - return newCredentials; - } finally { - await this.releaseLock(context); + if (!forceRefresh && context.memoryCache.credentials && this.isTokenValid(context.memoryCache.credentials)) { + qwenClient.setCredentials(context.memoryCache.credentials); + return context.memoryCache.credentials; } - }); - // 如果是等待其他请求完成的刷新，需要重新加载凭证 - if (!context.memoryCache.credentials || !this.isTokenValid(context.memoryCache.credentials)) { - // 优先从全局缓存加载 - if (context.uuid && credentialCache.hasCredentials(providerType, context.uuid)) { - const cachedEntry = credentialCache.getCredentials(providerType, context.uuid); - if (cachedEntry && cachedEntry.credentials) { - context.memoryCache.credentials = cachedEntry.credentials; - qwenClient.setCredentials(cachedEntry.credentials); - } - } else { - await this.reloadCredentialsFromFile(context); - if (context.memoryCache.credentials) { - qwenClient.setCredentials(context.memoryCache.credentials); - } + const response = await qwenClient.refreshAccessToken(); + if (!response || isErrorResponse(response)) { + throw new TokenManagerError(TokenError.REFRESH_FAILED, `Token refresh failed: ${response?.error}`); } + if (!response.access_token) { + throw new TokenManagerError(TokenError.REFRESH_FAILED, 'No access token in refresh response'); + } + const newCredentials = { + access_token: response.access_token, + token_type: response.token_type, + refresh_token: response.refresh_token || currentCredentials.refresh_token, + resource_url: response.resource_url, + expiry_date: Date.now() + response.expires_in * 1000, + }; + + context.memoryCache.credentials = newCredentials; + qwenClient.setCredentials(newCredentials); + await this.saveCredentialsToFile(context, newCredentials); + console.log('[Qwen Auth] Token refresh response: ok'); + return newCredentials; + } finally { + await this.releaseLock(context); } - - return credentials; } catch (error) { if (error instanceof TokenManagerError) throw error; @@ -890,26 +847,14 @@ class SharedTokenManager { } async saveCredentialsToFile(context, credentials) { - // 优先更新内存缓存 - const credentialCache = CredentialCacheManager.getInstance(); - const providerType = 'openai-qwen-oauth'; - - if (context.uuid && credentialCache.hasCredentials(providerType, context.uuid)) { - credentialCache.updateCredentials(providerType, context.uuid, credentials, context.credentialFilePath); - console.info(`[Qwen Auth] Updated credentials in memory cache: ${context.uuid}`); - // 同时更新本地内存缓存 - context.memoryCache.credentials = credentials; - context.memoryCache.fileModTime = Date.now(); - return; - } - - // 回退到使用内存锁写入文件 - await credentialCache.withMemoryLock(`qwen-save:${context.credentialFilePath}`, async () => { + try { await fs.mkdir(path.dirname(context.credentialFilePath), { recursive: true, mode: 0o700 }); await fs.writeFile(context.credentialFilePath, JSON.stringify(credentials, null, 2), { mode: 0o600 }); const stats = await fs.stat(context.credentialFilePath); context.memoryCache.fileModTime = stats.mtimeMs; - }); + } catch (error) { + console.error(`[Qwen Auth] Failed to save credentials to ${context.credentialFilePath}: ${error.message}`); + } } isTokenValid(credentials) { diff --git a/src/providers/provider-pool-manager.js b/src/providers/provider-pool-manager.js index 469df17..d99f9e2 100644 --- a/src/providers/provider-pool-manager.js +++ b/src/providers/provider-pool-manager.js @@ -2,7 +2,6 @@ import * as fs from 'fs'; // Import fs module import { getServiceAdapter } from './adapter.js'; import { MODEL_PROVIDER, getProtocolPrefix } from '../utils/common.js'; import { getProviderModels } from './provider-models.js'; -import { CredentialCacheManager } from '../utils/credential-cache-manager.js'; import axios from 'axios'; /** @@ -884,45 +883,6 @@ export class ProviderPoolManager { }; } - // ========== 快速预检：从内存缓存检查 OAuth 凭证状态 ========== - // 对于 OAuth 类型的 provider，先检查 token 是否过期，避免阻塞 - // 注意：只有在开启自动刷新 token (CRON_REFRESH_TOKEN) 时才执行快速预检 - // 因为如果没有自动刷新机制，缓存中的 token 状态可能不准确 - const oauthProviderTypes = ['claude-kiro-oauth', 'gemini-cli-oauth', 'gemini-antigravity', 'openai-qwen-oauth', 'openai-iflow-oauth', 'claude-orchids-oauth']; - const cronRefreshTokenEnabled = this.globalConfig?.CRON_REFRESH_TOKEN === true; - - if (cronRefreshTokenEnabled && oauthProviderTypes.includes(providerType) && providerConfig.uuid) { - const credentialCache = CredentialCacheManager.getInstance(); - const cachedEntry = credentialCache.getCredentials(providerType, providerConfig.uuid); - - if (cachedEntry && cachedEntry.credentials) { - const { expiresAt, accessToken } = cachedEntry.credentials; - - // 检查 token 是否存在 - if (!accessToken) { - this._log('warn', `Health check fast-fail for ${providerConfig.uuid}: No access token in cache`); - return { success: false, modelName, errorMessage: 'No access token available' }; - } - - // 检查 token 是否已过期（30秒缓冲） - if (expiresAt) { - const expirationTime = new Date(expiresAt).getTime(); - const now = Date.now(); - const bufferMs = 30 * 1000; - - if (expirationTime <= now + bufferMs) { - this._log('warn', `Health check fast-fail for ${providerConfig.uuid}: Token expired`); - return { success: false, modelName, errorMessage: 'Token expired' }; - } - } - - this._log('debug', `Health check pre-check passed for ${providerConfig.uuid}: Token valid in cache`); - } - // 注意：如果缓存中没有凭证，不要直接返回失败 - // 让后续的实际健康检查去尝试初始化和加载凭证 - // 这样可以支持刚重置健康状态或新添加的 provider - } - // ========== 实际 API 健康检查（带超时保护）========== const tempConfig = { ...providerConfig, diff --git a/src/services/service-manager.js b/src/services/service-manager.js index 3a50a16..dbf86f6 100644 --- a/src/services/service-manager.js +++ b/src/services/service-manager.js @@ -1,6 +1,5 @@ import { getServiceAdapter, serviceInstances } from '../providers/adapter.js'; import { ProviderPoolManager } from '../providers/provider-pool-manager.js'; -import { CredentialCacheManager } from '../utils/credential-cache-manager.js'; import deepmerge from 'deepmerge'; import * as fs from 'fs'; import { promises as pfs } from 'fs'; @@ -176,62 +175,6 @@ export async function initApiService(config) { }); console.log('[Initialization] ProviderPoolManager initialized with configured pools.'); // 健康检查将在服务器完全启动后执行 - - // 初始化凭证缓存管理器 - const credentialCache = CredentialCacheManager.getInstance(); - - // 获取单实例锁,防止多实例并发运行 - try { - await credentialCache.acquireInstanceLock(); - } catch (lockError) { - console.error('[Initialization] Failed to acquire instance lock:', lockError.message); - console.error('[Initialization] Please ensure no other instance is running.'); - process.exit(1); - } - - await credentialCache.preloadAllCredentials(config.providerPools); - credentialCache.startPeriodicSync(config.CREDENTIAL_SYNC_INTERVAL || 5000); - - // 注册进程退出钩子 - const shutdownHandler = async () => { - await credentialCache.shutdown(); - }; - process.on('beforeExit', shutdownHandler); - process.on('SIGINT', async () => { - await shutdownHandler(); - process.exit(0); - }); - process.on('SIGTERM', async () => { - await shutdownHandler(); - process.exit(0); - }); - - // 处理未捕获的异常,进行紧急同步 - process.on('uncaughtException', async (error) => { - console.error('[FATAL] Uncaught exception:', error); - console.log('[CredentialCache] Attempting emergency sync before crash...'); - try { - await credentialCache.syncToFile(); - console.log('[CredentialCache] Emergency sync completed'); - } catch (syncError) { - console.error('[CredentialCache] Emergency sync failed:', syncError.message); - } - await credentialCache.shutdown(); - process.exit(1); - }); - - process.on('unhandledRejection', async (reason, promise) => { - console.error('[FATAL] Unhandled rejection at:', promise, 'reason:', reason); - console.log('[CredentialCache] Attempting emergency sync...'); - try { - await credentialCache.syncToFile(); - console.log('[CredentialCache] Emergency sync completed'); - } catch (syncError) { - console.error('[CredentialCache] Emergency sync failed:', syncError.message); - } - }); - - console.log('[Initialization] CredentialCacheManager initialized.'); } else { console.log('[Initialization] No provider pools configured. Using single provider mode.'); } diff --git a/src/utils/credential-cache-manager.js b/src/utils/credential-cache-manager.js deleted file mode 100644 index 2957bce..0000000 --- a/src/utils/credential-cache-manager.js +++ /dev/null @@ -1,925 +0,0 @@ -/** - * 凭证缓存管理器 - * 纯内存操作,使用内存锁保证并发安全,定时同步到文件 - * 优先考虑并发性能,支持单实例部署 - */ - -import { promises as fs } from 'fs'; -import * as path from 'path'; -import { PROVIDER_MAPPINGS } from './provider-utils.js'; -import * as os from 'os'; - -/** - * 凭证条目结构 - * @typedef {Object} CredentialEntry - * @property {string} providerType - 提供商类型 - * @property {string} uuid - 节点唯一标识 - * @property {string} credPath - 凭证文件路径 - * @property {Object} credentials - 凭证数据 - * @property {number} lastModified - 最后修改时间戳 - * @property {number} lastAccessed - 最后访问时间戳 - * @property {boolean} isDirty - 是否有未同步的修改 - * @property {number} retryCount - 同步失败重试次数 - */ - -export class CredentialCacheManager { - static instance = null; - - constructor() { - // 凭证缓存: Map - // cacheKey = `${providerType}:${uuid}` - this.credentialCache = new Map(); - - // Promise链锁: Map - this.lockChains = new Map(); - - // 脏数据标记（需要同步到文件的 cacheKey） - this.dirtyKeys = new Set(); - - // 同步定时器 - this.syncTimer = null; - this.syncIntervalMs = 5000; // 默认5秒 - - // 是否已初始化 - this.isInitialized = false; - - // 是否正在同步 - this.isSyncing = false; - - // 最大重试次数 - this.maxRetries = 5; - - // 最大脏数据条目数 - this.maxDirtyKeys = 1000; - - // 死信队列 - 存储同步失败的凭证 - this.deadLetterQueue = new Map(); - - // 实例锁文件句柄 - this.instanceLockRelease = null; - - // 提供商类型到凭证路径键的映射 - this.providerCredPathKeys = {}; - for (const mapping of PROVIDER_MAPPINGS) { - this.providerCredPathKeys[mapping.providerType] = mapping.credPathKey; - } - } - - /** - * 获取单例实例 - * @returns {CredentialCacheManager} - */ - static getInstance() { - if (!CredentialCacheManager.instance) { - CredentialCacheManager.instance = new CredentialCacheManager(); - } - return CredentialCacheManager.instance; - } - - /** - * 生成缓存键 - * @param {string} providerType - 提供商类型 - * @param {string} uuid - 节点UUID - * @returns {string} - */ - _getCacheKey(providerType, uuid) { - return `${providerType}:${uuid}`; - } - - /** - * 获取单实例锁,防止多实例并发运行 - * @throws {Error} 如果已有实例在运行 - */ - async acquireInstanceLock() { - const lockPath = path.join(os.tmpdir(), 'credential-cache.lock'); - const pidPath = path.join(os.tmpdir(), 'credential-cache.pid'); - - try { - // 尝试读取已存在的 PID 文件 - try { - const existingPid = await fs.readFile(pidPath, 'utf8'); - const pid = parseInt(existingPid.trim(), 10); - - // 如果是当前进程的 PID，说明是配置重载，直接返回（已持有锁） - if (pid === process.pid) { - console.log(`[CredentialCache] Instance lock already held by current process (PID: ${pid})`); - return; - } - - // 检查进程是否还在运行 - try { - process.kill(pid, 0); // 0 信号仅检查进程存在性 - // 如果进程存在，检查是否是当前进程 - throw new Error(`[CredentialCache] Another instance is running (PID: ${pid}). Please stop it first.`); - } catch (killError) { - if (killError.code === 'ESRCH') { - // 进程已死亡,可以继续 - console.log(`[CredentialCache] Stale lock detected (PID: ${pid}), cleaning up...`); - } else { - throw killError; - } - } - } catch (readError) { - if (readError.code !== 'ENOENT') { - throw readError; - } - // PID 文件不存在,可以继续 - } - - // 写入当前进程 PID - await fs.writeFile(pidPath, String(process.pid), 'utf8'); - console.log(`[CredentialCache] Instance lock acquired (PID: ${process.pid})`); - - // 注册清理钩子 - this.instanceLockRelease = async () => { - try { - await fs.unlink(pidPath); - console.log('[CredentialCache] Instance lock released'); - } catch (error) { - if (error.code !== 'ENOENT') { - console.warn(`[CredentialCache] Failed to release lock: ${error.message}`); - } - } - }; - } catch (error) { - console.error(`[CredentialCache] Failed to acquire instance lock: ${error.message}`); - throw error; - } - } - - /** - * 预加载所有凭证到内存 - * @param {Object} providerPools - 提供商池配置 - */ - async preloadAllCredentials(providerPools) { - if (!providerPools || typeof providerPools !== 'object') { - console.log('[CredentialCache] No provider pools to preload'); - return; - } - - let loadedCount = 0; - let failedCount = 0; - let skippedCount = 0; - - for (const [providerType, providers] of Object.entries(providerPools)) { - if (!Array.isArray(providers)) continue; - - const credPathKey = this.providerCredPathKeys[providerType]; - if (!credPathKey) { - console.warn(`[CredentialCache] Unknown provider type: ${providerType}`); - continue; - } - - for (const providerConfig of providers) { - const uuid = providerConfig.uuid; - const credPath = providerConfig[credPathKey]; - const customName = providerConfig.customName || uuid; - - if (!uuid || !credPath) { - continue; - } - - try { - const credentials = await this._loadCredentialsFromFile(credPath); - if (credentials) { - // 预检查：验证凭证是否有效（至少有 token） - const hasValidToken = this._validateCredentials(providerType, credentials); - if (!hasValidToken) { - skippedCount++; - console.warn(`[CredentialCache] Skipping ${providerType}:${customName} - no valid token found in credentials`); - continue; - } - - const cacheKey = this._getCacheKey(providerType, uuid); - this.credentialCache.set(cacheKey, { - providerType, - uuid, - credPath, - credentials, - lastModified: Date.now(), - lastAccessed: Date.now(), - isDirty: false, - retryCount: 0 - }); - loadedCount++; - } else { - skippedCount++; - console.warn(`[CredentialCache] Skipping ${providerType}:${customName} - credentials file empty or unreadable`); - } - } catch (error) { - failedCount++; - console.warn(`[CredentialCache] Failed to load credentials for ${providerType}:${customName}: ${error.message}`); - } - } - } - - this.isInitialized = true; - console.log(`[CredentialCache] Preloaded ${loadedCount} credentials (${failedCount} failed, ${skippedCount} skipped)`); - } - - /** - * 验证凭证是否有效 - * @param {string} providerType - 提供商类型 - * @param {Object} credentials - 凭证对象 - * @returns {boolean} - */ - _validateCredentials(providerType, credentials) { - if (!credentials || typeof credentials !== 'object') { - return false; - } - - // 根据提供商类型检查必要的 token 字段 - switch (providerType) { - case 'claude-kiro-oauth': - // Kiro 需要 refreshToken 或 accessToken - return !!(credentials.refreshToken || credentials.accessToken); - - case 'gemini-cli-oauth': - case 'gemini-antigravity': - // Gemini 需要 refresh_token 或 access_token - return !!(credentials.refresh_token || credentials.access_token); - - case 'openai-qwen-oauth': - // Qwen 需要 refresh_token 或 access_token - return !!(credentials.refresh_token || credentials.access_token); - - case 'openai-iflow': - // iFlow 需要 refresh_token 或 access_token - return !!(credentials.refresh_token || credentials.access_token); - - case 'claude-orchids-oauth': - // Orchids 需要 sessionKey 或 accessToken - return !!(credentials.sessionKey || credentials.accessToken); - - default: - // 默认：只要有任何 token 相关字段就认为有效 - return !!( - credentials.refreshToken || credentials.accessToken || - credentials.refresh_token || credentials.access_token || - credentials.sessionKey || credentials.apiKey - ); - } - } - - /** - * 原子写入文件 (temp + rename 模式) - * @param {string} filePath - 目标文件路径 - * @param {string} data - 要写入的数据 - */ - async _atomicWriteFile(filePath, data) { - const tmpPath = `${filePath}.tmp.${Date.now()}.${process.pid}`; - - try { - // 1. 写入临时文件 - await fs.writeFile(tmpPath, data, 'utf8'); - - // 2. fsync 确保落盘 (需要文件句柄) - const fileHandle = await fs.open(tmpPath, 'r+'); - try { - await fileHandle.sync(); - } finally { - await fileHandle.close(); - } - - // 3. 原子重命名 - await fs.rename(tmpPath, filePath); - - return true; - } catch (error) { - // 清理临时文件 - try { - await fs.unlink(tmpPath); - } catch (unlinkError) { - // 忽略清理失败 - } - throw error; - } - } - /** - * 从文件加载凭证（仅用于初始导入和手动重载） - * 支持从损坏文件的备份恢复 - * @param {string} filePath - 文件路径 - * @returns {Promise} - */ - async _loadCredentialsFromFile(filePath) { - try { - // 处理相对路径 - const absolutePath = path.isAbsolute(filePath) - ? filePath - : path.join(process.cwd(), filePath); - - const content = await fs.readFile(absolutePath, 'utf8'); - try { - return JSON.parse(content); - } catch (parseError) { - console.warn(`[CredentialCache] JSON parse failed for ${filePath}, attempting field extraction...`); - // 尝试从损坏的 JSON 中提取关键字段 - const extracted = this._extractCredentialsFromCorruptedJson(content); - if (extracted) { - console.info(`[CredentialCache] Field extraction successful for ${filePath}`); - return extracted; - } - console.error(`[CredentialCache] All recovery methods failed for ${filePath}: ${parseError.message}`); - return null; - } - } catch (error) { - if (error.code === 'ENOENT') { - return null; - } - throw error; - } - } - - /** - * 从损坏的 JSON 中提取关键凭证字段 - * @param {string} content - 文件内容 - * @returns {Object|null} - */ - _extractCredentialsFromCorruptedJson(content) { - const extracted = {}; - - // 定义需要提取的关键字段及其正则模式 - const fieldPatterns = { - refreshToken: /"refreshToken"\s*:\s*"([^"]+)"/, - accessToken: /"accessToken"\s*:\s*"([^"]+)"/, - clientId: /"clientId"\s*:\s*"([^"]+)"/, - clientSecret: /"clientSecret"\s*:\s*"([^"]+)"/, - profileArn: /"profileArn"\s*:\s*"([^"]+)"/, - region: /"region"\s*:\s*"([^"]+)"/, - authMethod: /"authMethod"\s*:\s*"([^"]+)"/, - expiresAt: /"expiresAt"\s*:\s*"([^"]+)"/, - startUrl: /"startUrl"\s*:\s*"([^"]+)"/, - // Gemini/Qwen 相关字段 - refresh_token: /"refresh_token"\s*:\s*"([^"]+)"/, - access_token: /"access_token"\s*:\s*"([^"]+)"/, - client_id: /"client_id"\s*:\s*"([^"]+)"/, - client_secret: /"client_secret"\s*:\s*"([^"]+)"/, - }; - - for (const [field, pattern] of Object.entries(fieldPatterns)) { - const match = content.match(pattern); - if (match && match[1]) { - extracted[field] = match[1]; - } - } - - // 至少需要某种 token 才算有效 - if (extracted.refreshToken || extracted.accessToken || - extracted.refresh_token || extracted.access_token) { - console.info(`[CredentialCache] Extracted ${Object.keys(extracted).length} fields: ${Object.keys(extracted).join(', ')}`); - return extracted; - } - - return null; - } - - /** - * 获取凭证（从内存） - * @param {string} providerType - 提供商类型 - * @param {string} uuid - 节点UUID - * @returns {CredentialEntry|null} - */ - getCredentials(providerType, uuid) { - const cacheKey = this._getCacheKey(providerType, uuid); - const entry = this.credentialCache.get(cacheKey); - - if (entry) { - entry.lastAccessed = Date.now(); - return entry; - } - - return null; - } - - /** - * 检查凭证是否存在于缓存中 - * @param {string} providerType - 提供商类型 - * @param {string} uuid - 节点UUID - * @returns {boolean} - */ - hasCredentials(providerType, uuid) { - const cacheKey = this._getCacheKey(providerType, uuid); - return this.credentialCache.has(cacheKey); - } - - /** - * 更新凭证（仅更新内存，标记为dirty） - * @param {string} providerType - 提供商类型 - * @param {string} uuid - 节点UUID - * @param {Object} newCredentials - 新凭证数据 - * @param {string} [credPath] - 凭证文件路径（可选，用于新建条目） - */ - updateCredentials(providerType, uuid, newCredentials, credPath = null) { - const cacheKey = this._getCacheKey(providerType, uuid); - let entry = this.credentialCache.get(cacheKey); - - if (entry) { - // 更新现有条目 - entry.credentials = newCredentials; - entry.lastModified = Date.now(); - entry.isDirty = true; - entry.retryCount = 0; // 重置重试计数 - } else if (credPath) { - // 创建新条目 - entry = { - providerType, - uuid, - credPath, - credentials: newCredentials, - lastModified: Date.now(), - lastAccessed: Date.now(), - isDirty: true, - retryCount: 0 - }; - this.credentialCache.set(cacheKey, entry); - } else { - console.warn(`[CredentialCache] Cannot update non-existent entry without credPath: ${cacheKey}`); - return; - } - - // 标记为需要同步 - this.dirtyKeys.add(cacheKey); - - // 检查脏数据是否过多 - if (this.dirtyKeys.size > this.maxDirtyKeys) { - console.warn(`[CredentialCache] Dirty keys exceeded ${this.maxDirtyKeys}, triggering immediate sync`); - this.syncToFile().catch(err => console.error('[CredentialCache] Emergency sync failed:', err.message)); - } - } - - /** - * 删除凭证（从内存中删除，同时删除文件） - * @param {string} providerType - 提供商类型 - * @param {string} uuid - 节点UUID - */ - async deleteCredentials(providerType, uuid) { - const cacheKey = this._getCacheKey(providerType, uuid); - const entry = this.credentialCache.get(cacheKey); - - if (!entry) { - return; - } - - // 从内存中删除 - this.credentialCache.delete(cacheKey); - this.dirtyKeys.delete(cacheKey); - - // 删除文件 - if (entry.credPath) { - try { - const absolutePath = path.isAbsolute(entry.credPath) - ? entry.credPath - : path.join(process.cwd(), entry.credPath); - await fs.unlink(absolutePath); - console.log(`[CredentialCache] Deleted credential file: ${entry.credPath}`); - } catch (error) { - if (error.code !== 'ENOENT') { - console.error(`[CredentialCache] Failed to delete ${entry.credPath}: ${error.message}`); - } - } - } - } - - /** - * Promise链式内存锁 - 保证串行执行 - * 使用 Promise 构造器模式避免 Read-Check-Write 竞态 - * @param {string} key - 锁的唯一标识 - * @param {Function} operation - 要执行的操作 - * @returns {Promise} - */ - async withMemoryLock(key, operation) { - // 创建新的 Promise 用于链接后续操作 - let resolveNext, rejectNext; - const nextPromise = new Promise((resolve, reject) => { - resolveNext = resolve; - rejectNext = reject; - }); - - // 原子化获取前序链并立即更新 Map (避免竞态) - const prevChain = this.lockChains.get(key); - this.lockChains.set(key, nextPromise); - - // 链接到前序 Promise 执行操作 - const executeOperation = async () => { - try { - // 等待前序操作完成（忽略前序的错误，继续执行当前操作） - if (prevChain) { - await prevChain.catch(() => {}); - } - const result = await operation(); - resolveNext(result); - } catch (error) { - rejectNext(error); - } finally { - // 只有当前 Promise 还在 Map 中时才删除 - if (this.lockChains.get(key) === nextPromise) { - this.lockChains.delete(key); - } - } - }; - - // 立即开始执行（不阻塞） - executeOperation(); - - return nextPromise; - } - - /** - * 去重执行 - 多个并发请求共享同一个操作结果 - * @param {string} key - 去重键 - * @param {Function} operation - 要执行的操作 - * @returns {Promise} - */ - async withDeduplication(key, operation) { - const dedupeKey = `dedupe:${key}`; - - // 检查是否已有正在执行的操作 - const existingPromise = this.lockChains.get(dedupeKey); - if (existingPromise) { - // 直接等待现有操作完成并返回结果 - return existingPromise; - } - - // 创建新操作 Promise 并立即存储（避免竞态） - const operationPromise = (async () => { - try { - return await operation(); - } finally { - // 操作完成后清理 - if (this.lockChains.get(dedupeKey) === operationPromise) { - this.lockChains.delete(dedupeKey); - } - } - })(); - - // 立即存储，确保后续请求能看到 - this.lockChains.set(dedupeKey, operationPromise); - - return operationPromise; - } - - /** - * 启动定时同步 - * @param {number} intervalMs - 同步间隔（毫秒） - */ - startPeriodicSync(intervalMs = 5000) { - this.syncIntervalMs = intervalMs; - - if (this.syncTimer) { - clearInterval(this.syncTimer); - } - - this.syncTimer = setInterval(() => { - this.syncToFile().catch(error => { - console.error('[CredentialCache] Periodic sync failed:', error.message); - }); - }, this.syncIntervalMs); - - // 确保定时器不阻止进程退出 - if (this.syncTimer.unref) { - this.syncTimer.unref(); - } - - console.log(`[CredentialCache] Started periodic sync (interval: ${intervalMs}ms)`); - } - - /** - * 停止定时同步 - */ - stopPeriodicSync() { - if (this.syncTimer) { - clearInterval(this.syncTimer); - this.syncTimer = null; - console.log('[CredentialCache] Stopped periodic sync'); - } - } - - /** - * 同步脏数据到文件（纯内存操作，不使用文件锁） - */ - async syncToFile() { - if (this.dirtyKeys.size === 0) { - return; - } - - if (this.isSyncing) { - return; // 防止重入 - } - - this.isSyncing = true; - - // 复制脏数据集合（不立即清空，同步成功后再清理） - const keysToSync = Array.from(this.dirtyKeys); - - console.log(`[CredentialCache] Syncing ${keysToSync.length} credential(s) to file...`); - - let successCount = 0; - let failedCount = 0; - const successKeys = new Set(); - - // 并发写入所有文件（提高性能） - await Promise.allSettled( - keysToSync.map(async (cacheKey) => { - const entry = this.credentialCache.get(cacheKey); - if (!entry || !entry.credPath) { - return; - } - - // 检查重试次数 - if (entry.retryCount >= this.maxRetries) { - // 达到最大重试次数,移入死信队列 - console.error(`[CredentialCache] Max retries exceeded for ${cacheKey}, moving to dead letter queue`); - - this.deadLetterQueue.set(cacheKey, { - entry: JSON.parse(JSON.stringify(entry)), // 深拷贝 - failureReason: entry.lastError || 'Max retries exceeded', - firstFailureTime: entry.firstFailureTime || Date.now(), - timestamp: Date.now() - }); - - // 尝试导出到紧急备份 - try { - const emergencyDir = path.join(process.cwd(), 'emergency_backup'); - await fs.mkdir(emergencyDir, { recursive: true }); - const emergencyPath = path.join(emergencyDir, `${cacheKey.replace(/:/g, '_')}.json`); - await fs.writeFile(emergencyPath, JSON.stringify(entry.credentials, null, 2), 'utf8'); - console.log(`[CredentialCache] Credential backed up to: ${emergencyPath}`); - } catch (backupError) { - console.error(`[CredentialCache] Failed to backup credential: ${backupError.message}`); - } - - this.dirtyKeys.delete(cacheKey); - entry.isDirty = false; - return; - } - - try { - // 处理相对路径 - const absolutePath = path.isAbsolute(entry.credPath) - ? entry.credPath - : path.join(process.cwd(), entry.credPath); - - // 确保目录存在 - const dir = path.dirname(absolutePath); - await fs.mkdir(dir, { recursive: true }); - - // 使用原子写入 - await this._atomicWriteFile( - absolutePath, - JSON.stringify(entry.credentials, null, 2) - ); - - entry.isDirty = false; - entry.retryCount = 0; - entry.lastError = null; - entry.firstFailureTime = null; - successKeys.add(cacheKey); - successCount++; - } catch (error) { - // 同步失败，增加重试计数并记录错误 - entry.retryCount++; - entry.lastError = error.message; - if (!entry.firstFailureTime) { - entry.firstFailureTime = Date.now(); - } - failedCount++; - console.error(`[CredentialCache] Failed to sync ${entry.credPath} (retry ${entry.retryCount}/${this.maxRetries}): ${error.message}`); - } - }) - ); - - // 从脏数据集合中移除成功同步的键 - for (const key of successKeys) { - this.dirtyKeys.delete(key); - } - - this.isSyncing = false; - - if (successCount > 0 || failedCount > 0) { - console.log(`[CredentialCache] Sync completed: ${successCount} success, ${failedCount} failed, ${this.dirtyKeys.size} pending`); - } - } - - /** - * 立即同步指定凭证到文件 - * @param {string} providerType - 提供商类型 - * @param {string} uuid - 节点UUID - */ - async syncCredentialToFile(providerType, uuid) { - const cacheKey = this._getCacheKey(providerType, uuid); - const entry = this.credentialCache.get(cacheKey); - - if (!entry || !entry.credPath) { - return; - } - - try { - const absolutePath = path.isAbsolute(entry.credPath) - ? entry.credPath - : path.join(process.cwd(), entry.credPath); - - const dir = path.dirname(absolutePath); - await fs.mkdir(dir, { recursive: true }); - - // 使用原子写入 - await this._atomicWriteFile( - absolutePath, - JSON.stringify(entry.credentials, null, 2) - ); - - entry.isDirty = false; - entry.retryCount = 0; - entry.lastError = null; - entry.firstFailureTime = null; - this.dirtyKeys.delete(cacheKey); - } catch (error) { - console.error(`[CredentialCache] Failed to sync ${entry.credPath}: ${error.message}`); - throw error; - } - } - - /** - * 关闭时同步所有数据（带超时） - */ - async shutdown() { - console.log('[CredentialCache] Shutting down...'); - - // 停止定时同步 - this.stopPeriodicSync(); - - // 同步所有脏数据（带动态超时） - if (this.dirtyKeys.size > 0) { - console.log(`[CredentialCache] Syncing ${this.dirtyKeys.size} dirty credential(s) before shutdown...`); - - // 根据脏数据量动态调整超时时间 - const timeoutMs = Math.max(10000, this.dirtyKeys.size * 50 + 5000); - const timeoutPromise = new Promise((_, reject) => - setTimeout(() => reject(new Error('Shutdown sync timeout')), timeoutMs) - ); - - try { - await Promise.race([ - this.syncToFile(), - timeoutPromise - ]); - } catch (error) { - console.error(`[CredentialCache] Shutdown sync failed: ${error.message}`); - console.error(`[CredentialCache] ${this.dirtyKeys.size} credentials NOT saved`); - - // 尝试紧急备份未保存的凭证 - if (this.dirtyKeys.size > 0) { - console.log('[CredentialCache] Attempting emergency backup...'); - try { - const emergencyDir = path.join(process.cwd(), 'emergency_backup'); - await fs.mkdir(emergencyDir, { recursive: true }); - for (const cacheKey of this.dirtyKeys) { - const entry = this.credentialCache.get(cacheKey); - if (entry && entry.credentials) { - const emergencyPath = path.join(emergencyDir, `${cacheKey.replace(/:/g, '_')}.json`); - await fs.writeFile(emergencyPath, JSON.stringify(entry.credentials, null, 2), 'utf8'); - } - } - console.log(`[CredentialCache] Emergency backup completed to: ${emergencyDir}`); - } catch (backupError) { - console.error(`[CredentialCache] Emergency backup failed: ${backupError.message}`); - } - } - } - } - - // 释放实例锁 - if (this.instanceLockRelease) { - await this.instanceLockRelease(); - } - - // 输出死信队列状态 - if (this.deadLetterQueue.size > 0) { - console.warn(`[CredentialCache] ${this.deadLetterQueue.size} credential(s) in dead letter queue`); - } - - console.log('[CredentialCache] Shutdown complete'); - } - - /** - * 获取缓存统计信息 - * @returns {Object} - */ - getStats() { - const stats = { - totalEntries: this.credentialCache.size, - dirtyEntries: this.dirtyKeys.size, - activeLocks: this.lockChains.size, - deadLetterQueueSize: this.deadLetterQueue.size, - isInitialized: this.isInitialized, - isSyncing: this.isSyncing, - syncInterval: this.syncIntervalMs, - byProvider: {} - }; - - for (const [cacheKey, entry] of this.credentialCache) { - const providerType = entry.providerType; - if (!stats.byProvider[providerType]) { - stats.byProvider[providerType] = { - count: 0, - dirty: 0, - maxRetries: 0 - }; - } - stats.byProvider[providerType].count++; - if (entry.isDirty) { - stats.byProvider[providerType].dirty++; - stats.byProvider[providerType].maxRetries = Math.max( - stats.byProvider[providerType].maxRetries, - entry.retryCount - ); - } - } - - return stats; - } - - /** - * 获取死信队列内容 - * @returns {Array} - */ - getDeadLetterQueue() { - return Array.from(this.deadLetterQueue.entries()).map(([key, value]) => ({ - cacheKey: key, - ...value - })); - } - - /** - * 从死信队列恢复凭证 - * @param {string} cacheKey - 缓存键 - * @returns {boolean} 是否恢复成功 - */ - recoverFromDeadLetter(cacheKey) { - const deadEntry = this.deadLetterQueue.get(cacheKey); - if (!deadEntry) { - return false; - } - - // 恢复到缓存 - const entry = deadEntry.entry; - entry.retryCount = 0; - entry.isDirty = true; - entry.lastError = null; - entry.firstFailureTime = null; - this.credentialCache.set(cacheKey, entry); - this.dirtyKeys.add(cacheKey); - - // 从死信队列移除 - this.deadLetterQueue.delete(cacheKey); - - console.log(`[CredentialCache] Recovered credential from dead letter queue: ${cacheKey}`); - return true; - } - - /** - * 清除所有缓存（用于测试） - */ - clear() { - this.credentialCache.clear(); - this.lockChains.clear(); - this.dirtyKeys.clear(); - this.deadLetterQueue.clear(); - this.isInitialized = false; - } - - /** - * 重新加载指定凭证（从文件） - * @param {string} providerType - 提供商类型 - * @param {string} uuid - 节点UUID - * @returns {Promise} - */ - async reloadCredentials(providerType, uuid) { - const cacheKey = this._getCacheKey(providerType, uuid); - const entry = this.credentialCache.get(cacheKey); - - if (!entry || !entry.credPath) { - return null; - } - - try { - const credentials = await this._loadCredentialsFromFile(entry.credPath); - if (credentials) { - entry.credentials = credentials; - entry.lastModified = Date.now(); - entry.isDirty = false; - entry.retryCount = 0; - this.dirtyKeys.delete(cacheKey); - return entry; - } - } catch (error) { - console.error(`[CredentialCache] Failed to reload ${entry.credPath}: ${error.message}`); - } - - return null; - } -} - -// 导出单例获取函数 -export function getCredentialCacheManager() { - return CredentialCacheManager.getInstance(); -}