/** * 云业务数据库连接模块(MSSQL) * 支持本地 / 公网双配置与自动回退。 */ const os = require('os'); const sql = require('mssql'); const mysql = require('mysql2/promise'); const { getSystemDbRuntimeConfig } = require('../services/systemDbConfig'); const NOT_CONFIGURED_MESSAGE = '业务数据库尚未配置,请先在系统设置 -> 业务数据库中完成配置'; const TRANSIENT_ERROR_CODES = new Set(['ESOCKET', 'ECONNRESET', 'ETIMEOUT', 'ENOTOPEN', 'ECONNCLOSED']); const CLOUD_DATABASE_CONFIG_NAME = 'cloud_database'; const CLOUD_DATABASE_PUBLIC_CONFIG_NAME = 'cloud_database_public'; const DEFAULT_CLOUD_LOCAL_SERVER_IP = '192.168.150.234'; const DEFAULT_REPORT_REPLICA_DATABASE = 'JSReportReplica'; const DEFAULT_REPORT_REPLICA_SCHEMA = 'rpt'; const DEFAULT_REPORT_REPLICA_APP_SCHEMA = 'app'; let pool = null; let activeCloudConfig = null; let cachedCloudConfigCandidates = null; // 负缓存(失败态)的过期时间戳。0 表示没有失败态缓存。 let cachedCloudConfigExpiresAt = 0; function isPoolUsable(activePool) { return Boolean(activePool && activePool.connected && activePool.healthy !== false); } const parsePositiveInt = (value, fallback) => { const parsed = Number.parseInt(value, 10); return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback; }; const DEFAULT_CONNECT_TIMEOUT_MS = parsePositiveInt(process.env.CLOUD_DB_CONNECT_TIMEOUT_MS, 30000); // 原先默认 600000ms(10 分钟),配合 maxAttempts=2 最坏可挂 20 分钟,用户体感就是「卡死」。 // 收敛到 60s:本页最重的冷查实测 16~42s,仍有充足余量。 const DEFAULT_REQUEST_TIMEOUT_MS = parsePositiveInt(process.env.CLOUD_DB_REQUEST_TIMEOUT_MS, 60000); // 解析「系统库配置」失败时的负缓存存活时间。到期后自动重试,使隧道恢复后无需重启后端。 const NEGATIVE_CONFIG_CACHE_TTL_MS = parsePositiveInt(process.env.CLOUD_DB_CONFIG_NEGATIVE_TTL_MS, 30000); const CLOUD_DB_POOL_MAX = parsePositiveInt(process.env.CLOUD_DB_POOL_MAX, 10); const CLOUD_DB_POOL_IDLE_TIMEOUT_MS = parsePositiveInt(process.env.CLOUD_DB_POOL_IDLE_TIMEOUT_MS, 30000); const LOCAL_CONFIG_DB_CONNECT_TIMEOUT_MS = parsePositiveInt(process.env.LOCAL_DB_CONNECT_TIMEOUT_MS, 5000); const LOCAL_CONFIG_DB_CONNECTION_LIMIT = parsePositiveInt(process.env.LOCAL_CONFIG_DB_CONNECTION_LIMIT, 2); const DEFAULT_ENV_KEYS = Object.freeze({ driver: ['CLOUD_DB_TYPE', 'APP_DB_TYPE', 'SQLSERVER_DRIVER'], host: ['CLOUD_DB_HOST', 'APP_DB_HOST', 'SQLSERVER_HOST'], port: ['CLOUD_DB_PORT', 'APP_DB_PORT', 'SQLSERVER_PORT'], database: ['CLOUD_DB_NAME', 'APP_DB_NAME', 'SQLSERVER_DATABASE'], user: ['CLOUD_DB_USER', 'APP_DB_USER', 'SQLSERVER_USERNAME'], password: ['CLOUD_DB_PASSWORD', 'APP_DB_PASSWORD', 'SQLSERVER_PASSWORD'], ssl: ['CLOUD_DB_SSL', 'SQLSERVER_ENCRYPT'], }); const PUBLIC_ENV_KEYS = Object.freeze({ driver: ['CLOUD_DB_PUBLIC_TYPE', 'CLOUD_DB_TYPE', 'APP_DB_TYPE', 'SQLSERVER_DRIVER'], host: ['CLOUD_DB_PUBLIC_HOST', 'CLOUD_DB_HOST', 'APP_DB_HOST', 'SQLSERVER_HOST'], port: ['CLOUD_DB_PUBLIC_PORT', 'CLOUD_DB_PORT', 'APP_DB_PORT', 'SQLSERVER_PORT'], database: ['CLOUD_DB_PUBLIC_NAME', 'CLOUD_DB_PUBLIC_DATABASE', 'CLOUD_DB_NAME', 'APP_DB_NAME', 'SQLSERVER_DATABASE'], user: ['CLOUD_DB_PUBLIC_USER', 'CLOUD_DB_USER', 'APP_DB_USER', 'SQLSERVER_USERNAME'], password: ['CLOUD_DB_PUBLIC_PASSWORD', 'CLOUD_DB_PASSWORD', 'APP_DB_PASSWORD', 'SQLSERVER_PASSWORD'], ssl: ['CLOUD_DB_PUBLIC_SSL', 'CLOUD_DB_SSL', 'SQLSERVER_ENCRYPT'], }); function isTransientError(error) { if (!error) return false; const message = String(error.message || error.originalError?.message || '').toLowerCase(); const code = error.code || error.originalError?.code; return ( TRANSIENT_ERROR_CODES.has(code) || message.includes('connection lost') || message.includes('socket hang up') || message.includes('connection is closed') || message.includes('unable to process incoming packet') || message.includes('timed out') ); } function isTruthy(value) { return value === true || value === 1 || String(value || '').toLowerCase() === 'true' || String(value || '') === '1'; } function parseBoolean(value, fallback = false) { if (value === undefined || value === null || value === '') { return fallback; } const normalized = String(value).trim().toLowerCase(); if (['1', 'true', 'yes', 'on'].includes(normalized)) { return true; } if (['0', 'false', 'no', 'off'].includes(normalized)) { return false; } return fallback; } function getFirstEnvValue(source, keys = [], fallback = '') { for (const key of keys) { const value = source[key]; if (value !== undefined && value !== null && String(value).trim() !== '') { return String(value).trim(); } } return fallback; } function normalizeIdentifierText(value) { return String(value || '').replace(/[\[\]]/g, '').trim(); } function quoteMssqlIdentifier(value) { return `[${normalizeIdentifierText(value).replace(/]/g, ']]')}]`; } function escapeRegExp(value) { return String(value || '').replace(/[.*+?^${}()|[\]\\]/g, '\\$&'); } function getReportReplicaDatabaseName() { return normalizeIdentifierText(process.env.CLOUD_DB_REPORT_DATABASE || DEFAULT_REPORT_REPLICA_DATABASE); } function getReportReplicaSchemaName() { return normalizeIdentifierText(process.env.CLOUD_DB_REPORT_SCHEMA || DEFAULT_REPORT_REPLICA_SCHEMA) || DEFAULT_REPORT_REPLICA_SCHEMA; } function getReportReplicaAppSchemaName() { return normalizeIdentifierText(process.env.CLOUD_DB_APP_SCHEMA || DEFAULT_REPORT_REPLICA_APP_SCHEMA) || DEFAULT_REPORT_REPLICA_APP_SCHEMA; } function isReportReplicaRuntime(runtimeInfo = {}) { const mode = String(process.env.CLOUD_DB_REPORT_REWRITE || 'auto').trim().toLowerCase(); if (['0', 'false', 'off', 'disabled'].includes(mode)) { return false; } if (['1', 'true', 'on', 'enabled', 'force'].includes(mode)) { return true; } return normalizeIdentifierText(runtimeInfo.database).toLowerCase() === getReportReplicaDatabaseName().toLowerCase(); } function rewriteReportReplicaObjectName(match, tableName) { return `${quoteMssqlIdentifier(getReportReplicaSchemaName())}.${tableName.startsWith('[') ? tableName : quoteMssqlIdentifier(tableName)}`; } const REPORT_REPLICA_UNQUALIFIED_SOURCE_TABLES = Object.freeze([ 'customer_service_evaluation', 'fys_shopInfo', 'fys_SkuDataLibrary', 'fys_SkuDataLibraryNew', 'fys_SkuDataLibrary_SpuConfig', 'jst_goods_profit_kpi_items_afs', 'jst_inventoryInfo', 'jst_order_express_details', 'jst_qmcust_refundlist', 'jst_refund_pdd_fz_view', 'jst_returnAndRefundInfo', 'jst_sku', 'jst_skumap', 'jst_skumap_pdd', 'jst_skumap_qmcust_add', 'jst_warehouseInfo', 'op_GoodsIdBarcodeAnalysisDaily', 'op_LeafSkuFreight30dDaily', 'op_TargetManagement', 'pdd_DataGoods', 'pdd_Evaluation', 'pdd_ReportGoodsSpread', 'pc_kfChatLog', 'TmallImportBatch', 'TmallSrcBundle', 'TmallSrcBusinessRoi', 'TmallSrcOverall', ]); function buildReportReplicaProductCodeExpression(alias) { return ` CASE WHEN ${alias}.other_1 IS NULL OR LTRIM(RTRIM(${alias}.other_1)) = N'' THEN NULL WHEN CHARINDEX(N'K', ${alias}.other_1) > 0 THEN LEFT(${alias}.other_1, CHARINDEX(N'K', ${alias}.other_1) - 1) ELSE ${alias}.other_1 END`; } function buildReportReplicaSaleStatusExpression(alias, costAlias = null) { const fallbackStatus = ` CASE WHEN ISNULL(${alias}.enabled, 0) = 1 AND ISNULL(${alias}.stock_disabled, 0) = 0 THEN N'销售' WHEN ISNULL(${alias}.enabled, 0) = -1 OR ISNULL(${alias}.stock_disabled, 0) = 1 THEN N'停用' ELSE N'未配置' END`; if (!costAlias) { return fallbackStatus; } return ` COALESCE( NULLIF(LTRIM(RTRIM(TRY_CONVERT(NVARCHAR(100), ${costAlias}.skuStatus))), N''), ${fallbackStatus} )`; } function buildSkuDataLibraryCompatSql() { const schema = quoteMssqlIdentifier(getReportReplicaSchemaName()); const productCodeSql = buildReportReplicaProductCodeExpression('s'); const saleStatusSql = buildReportReplicaSaleStatusExpression('s', 'c'); return `( SELECT ROW_NUMBER() OVER (ORDER BY s.sku_id) AS Id, s.sku_id AS SkuCode, s.name AS SkuName, s.supplier_name AS Supplier, ${productCodeSql} AS ItemNum, s.created AS CreationTime, s.created AS created, ${saleStatusSql} AS ProStatus, ${saleStatusSql} AS Status, COALESCE(NULLIF(LTRIM(RTRIM(s.pic_big)), N''), NULLIF(LTRIM(RTRIM(s.pic)), N''), NULLIF(LTRIM(RTRIM(TRY_CONVERT(NVARCHAR(500), c.pictureSkucode))), N'')) AS Picture, s.category AS CategoryOne, s.category AS CategoryTwo, COALESCE(NULLIF(LTRIM(RTRIM(s.other_10)), N''), NULLIF(LTRIM(RTRIM(s.other_2)), N'')) AS materialcode, COALESCE( NULLIF(TRY_CONVERT(DECIMAL(18, 4), c.cleanCost), 0), NULLIF(TRY_CONVERT(DECIMAL(18, 4), c.costPrice), 0), NULLIF(TRY_CONVERT(DECIMAL(18, 4), s.cost_price), 0), NULLIF(TRY_CONVERT(DECIMAL(18, 4), s.purchase_price), 0) ) AS CostPrice, COALESCE(NULLIF(TRY_CONVERT(DECIMAL(18, 4), s.purchase_price), 0), NULLIF(TRY_CONVERT(DECIMAL(18, 4), s.cost_price), 0), NULLIF(TRY_CONVERT(DECIMAL(18, 4), c.costPrice), 0)) AS ErpPrice, COALESCE(NULLIF(TRY_CONVERT(DECIMAL(18, 4), s.sale_price), 0), NULLIF(TRY_CONVERT(DECIMAL(18, 4), s.market_price), 0)) AS ActualPrice, s.color AS Color, s.properties_value AS SkuSize, CAST(NULL AS NVARCHAR(255)) AS Model, s.unit AS PackSize, s.l AS Length, s.w AS Width, s.h AS Height, COALESCE(NULLIF(TRY_CONVERT(DECIMAL(18, 4), c.volume), 0), TRY_CONVERT(DECIMAL(18, 4), s.pack_volume)) AS Volume, COALESCE(NULLIF(TRY_CONVERT(DECIMAL(18, 4), c.weight), 0), TRY_CONVERT(DECIMAL(18, 4), s.weight)) AS Weight, s.other_3 AS PurchaseType FROM ${schema}.${quoteMssqlIdentifier('jst_sku')} s OUTER APPLY ( SELECT TOP 1 costPrice, volume, weight, skuStatus, createTime, cleanCost, pictureSkucode FROM ${schema}.${quoteMssqlIdentifier('fys_skuCostPriceUpdate')} c WHERE UPPER(LTRIM(RTRIM(TRY_CONVERT(NVARCHAR(100), c.sku)))) = UPPER(LTRIM(RTRIM(TRY_CONVERT(NVARCHAR(100), s.sku_id)))) ORDER BY c.createTime DESC ) c )`; } function buildSkuDimensionCompatSql() { const schema = quoteMssqlIdentifier(getReportReplicaSchemaName()); const productCodeSql = buildReportReplicaProductCodeExpression('s'); const saleStatusSql = buildReportReplicaSaleStatusExpression('s'); return `( SELECT s.sku_id AS skucode, ${productCodeSql} AS procode, s.name AS skuname, COALESCE(NULLIF(LTRIM(RTRIM(s.pic_big)), N''), NULLIF(LTRIM(RTRIM(s.pic)), N'')) AS pic_url, ${saleStatusSql} AS pro_status, COALESCE(NULLIF(TRY_CONVERT(DECIMAL(18, 4), s.cost_price), 0), NULLIF(TRY_CONVERT(DECIMAL(18, 4), s.purchase_price), 0)) AS cost_price, s.created AS created, s.supplier_name AS supplier, ${productCodeSql} AS spu, s.category AS category, 1 AS status FROM ${schema}.${quoteMssqlIdentifier('jst_sku')} s )`; } function buildSpuPrincipalCompatSql() { const schema = quoteMssqlIdentifier(getReportReplicaSchemaName()); return `( SELECT v.spu, v.designer, v.purchaser, v.oper, CAST(NULL AS DATETIME) AS created FROM ${schema}.${quoteMssqlIdentifier('V_caterAndOper')} v )`; } function buildEmptyBasicIdRankCompatSql() { return `( SELECT CAST(NULL AS DATE) AS stat_date, CAST(NULL AS NVARCHAR(255)) AS shop, CAST(NULL AS NVARCHAR(255)) AS item_id, CAST(NULL AS DECIMAL(38, 6)) AS payment, CAST(NULL AS DECIMAL(38, 6)) AS pay_buyer_num, CAST(NULL AS DECIMAL(38, 6)) AS item_visitor_num, CAST(NULL AS DECIMAL(38, 6)) AS add_purchase_num, CAST(NULL AS DECIMAL(38, 6)) AS search_guide_views_num WHERE 1 = 0 )`; } function buildEmptyBasicIdPromoteCompatSql() { return `( SELECT CAST(NULL AS DATE) AS stat_date, CAST(NULL AS NVARCHAR(255)) AS shopname, CAST(NULL AS NVARCHAR(255)) AS item_id, CAST(NULL AS DECIMAL(38, 6)) AS cost_charge, CAST(NULL AS DECIMAL(38, 6)) AS sup_charge, CAST(NULL AS DECIMAL(38, 6)) AS strategy_ids, CAST(NULL AS DECIMAL(38, 6)) AS strategy_keywords, CAST(NULL AS DECIMAL(38, 6)) AS strategy_person, CAST(NULL AS DECIMAL(38, 6)) AS strategy_scene WHERE 1 = 0 )`; } function rewriteSourceObject(sqlText, objectNames, replacementSql, defaultAlias) { const names = Array.isArray(objectNames) ? objectNames : [objectNames]; const tablePattern = names.map((name) => escapeRegExp(name)).join('|'); const schemaPattern = escapeRegExp(quoteMssqlIdentifier(getReportReplicaSchemaName())); const sourcePattern = new RegExp( `\\b(FROM|JOIN)\\s+(?:${schemaPattern}\\s*\\.\\s*)?\\[(?:${tablePattern})\\](\\s+(?!(?:WITH|ON|WHERE|INNER|LEFT|RIGHT|FULL|CROSS|JOIN|GROUP|ORDER|UNION|HAVING|OUTER|APPLY)\\b)(?:AS\\s+)?[A-Za-z_][\\w$#]*)?(\\s+WITH\\s*\\([^)]*\\))?`, 'giu' ); const bareSourcePattern = new RegExp( `\\b(FROM|JOIN)\\s+(?:${tablePattern})(\\s+(?!(?:WITH|ON|WHERE|INNER|LEFT|RIGHT|FULL|CROSS|JOIN|GROUP|ORDER|UNION|HAVING|OUTER|APPLY)\\b)(?:AS\\s+)?[A-Za-z_][\\w$#]*)?(\\s+WITH\\s*\\([^)]*\\))?`, 'giu' ); const replaceSource = (_match, keyword, aliasText = '') => `${keyword} ${replacementSql}${aliasText || ` AS ${defaultAlias}`}`; return sqlText.replace(sourcePattern, replaceSource).replace(bareSourcePattern, replaceSource); } function rewriteUnqualifiedReportReplicaSourceObjects(sqlText) { const tablePattern = REPORT_REPLICA_UNQUALIFIED_SOURCE_TABLES .map((name) => escapeRegExp(name)) .join('|'); if (!tablePattern) { return sqlText; } const sourcePattern = new RegExp( `\\b(FROM|JOIN)\\s+(${tablePattern})(?=\\s|$|;|,)`, 'giu' ); return sqlText.replace(sourcePattern, (_match, keyword, tableName) => ( `${keyword} ${quoteMssqlIdentifier(getReportReplicaSchemaName())}.${quoteMssqlIdentifier(tableName)}` )); } function rewriteReportReplicaWarehouseSchemaSources(sqlText) { const sourcePattern = /\b(FROM|JOIN)\s+\[?(?:dws|ads|dim|ods|dwm|dwd)\]?\s*\.\s*(\[[^\]]+\]|[A-Za-z_][\w$#]*)(?=\s|$|;|,)/giu; return sqlText.replace(sourcePattern, (_match, keyword, tableName) => ( `${keyword} ${rewriteReportReplicaObjectName(null, tableName)}` )); } function applyReportReplicaCompatibilitySources(sqlText) { let rewritten = sqlText; rewritten = rewriteSourceObject( rewritten, ['fys_SkuDataLibraryNew', 'fys_SkuDataLibrary'], buildSkuDataLibraryCompatSql(), 'skuLibraryCompat' ); rewritten = rewriteSourceObject( rewritten, 'dim_ope_sku_data_df', buildSkuDimensionCompatSql(), 'skuDimensionCompat' ); rewritten = rewriteSourceObject( rewritten, ['fys_SkuDataLibrary_SpuConfig', 'dim_ope_spu_data_df', 'op_spuPrincipal_log'], buildSpuPrincipalCompatSql(), 'spuPrincipalCompat' ); rewritten = rewriteSourceObject( rewritten, 'dwd_ope_basic_id_rank_di', buildEmptyBasicIdRankCompatSql(), 'basicIdRankCompat' ); rewritten = rewriteSourceObject( rewritten, 'dwd_ope_basic_id_promote_di', buildEmptyBasicIdPromoteCompatSql(), 'basicIdPromoteCompat' ); return rewritten; } function rewriteReportReplicaSql(sqlText, runtimeInfo = {}) { if (!isReportReplicaRuntime(runtimeInfo) || typeof sqlText !== 'string') { return sqlText; } let rewritten = sqlText; rewritten = rewriteReportReplicaWarehouseSchemaSources(rewritten); rewritten = rewriteUnqualifiedReportReplicaSourceObjects(rewritten); rewritten = rewritten.replace( /\[(?:JianSun|JSDataMiddlePlatform|JSDataWarehouse)\]\s*\.\s*\[(?:dbo|dws|ads|dim|ods|dwm|dwd)\]\s*\.\s*(\[[^\]]+\])/giu, rewriteReportReplicaObjectName ); rewritten = rewritten.replace( /\b(?:JianSun|JSDataMiddlePlatform|JSDataWarehouse)\s*\.\s*(?:dbo|dws|ads|dim|ods|dwm|dwd)\s*\.\s*(\[[^\]]+\]|[A-Za-z_][\w$#]*)/giu, rewriteReportReplicaObjectName ); rewritten = rewritten.replace( /\[(?:JianSun|JSDataMiddlePlatform|JSDataWarehouse)\]\s*\.\s*INFORMATION_SCHEMA\s*\.\s*COLUMNS/giu, 'INFORMATION_SCHEMA.COLUMNS' ); rewritten = rewritten.replace( /\b(?:JianSun|JSDataMiddlePlatform|JSDataWarehouse)\s*\.\s*INFORMATION_SCHEMA\s*\.\s*COLUMNS/giu, 'INFORMATION_SCHEMA.COLUMNS' ); rewritten = rewritten.replace( /\[(?:dbo)\]\s*\.\s*(\[[^\]]+\])/giu, rewriteReportReplicaObjectName ); rewritten = rewritten.replace( /\bdbo\s*\.\s*(\[[^\]]+\]|[A-Za-z_][\w$#]*)/giu, rewriteReportReplicaObjectName ); rewritten = rewritten.replace( /\[(?:dws|ads|dim|ods)\]\s*\.\s*(\[(?:dws|ads|dim|ods)_[^\]]+\])/giu, rewriteReportReplicaObjectName ); rewritten = rewritten.replace( /\b(?:dws|ads|dim|ods)\s*\.\s*((?:dws|ads|dim|ods)_[A-Za-z_][\w$#]*|\[(?:dws|ads|dim|ods)_[^\]]+\])/giu, rewriteReportReplicaObjectName ); rewritten = rewritten.replace( /(TABLE_SCHEMA\s*=\s*)N?(['"])(?:dbo|dws|ads|dim|ods|dwm|dwd)\2/giu, (_match, prefix, quote) => `${prefix}${quote}${getReportReplicaSchemaName()}${quote}` ); return applyReportReplicaCompatibilitySources(rewritten); } function normalizeRuntimeProfile(value) { const profile = String(value || process.env.CLOUD_DB_RUNTIME_PROFILE || 'auto').trim().toLowerCase(); if (['local', 'public', 'auto'].includes(profile)) { return profile; } return 'auto'; } function getConfiguredLocalServerIps() { return [ process.env.CLOUD_DB_LOCAL_SERVER_IPS, process.env.CLOUD_DB_LOCAL_SERVER_IP, DEFAULT_CLOUD_LOCAL_SERVER_IP, ] .filter(Boolean) .join(',') .split(',') .map((item) => item.trim()) .filter(Boolean); } function getLocalIpAddresses() { const manualIps = [ process.env.CLOUD_DB_SERVER_IP, process.env.SERVER_PRIVATE_IP, process.env.SERVER_IP, process.env.HOST_IP, ] .filter(Boolean) .flatMap((item) => String(item).split(',')) .map((item) => item.trim()) .filter(Boolean); const networkIps = Object.values(os.networkInterfaces() || {}) .flat() .filter((item) => item && item.family === 'IPv4') .map((item) => item.address) .filter(Boolean); return Array.from(new Set([...manualIps, ...networkIps])); } function isCloudLocalServer() { const localServerIps = new Set(getConfiguredLocalServerIps()); return getLocalIpAddresses().some((ip) => localServerIps.has(ip)); } function getCloudProfileOrder() { const runtimeProfile = normalizeRuntimeProfile(); if (runtimeProfile === 'local') { return ['local', 'public']; } if (runtimeProfile === 'public') { return ['public', 'local']; } return isCloudLocalServer() ? ['local', 'public'] : ['public', 'local']; } function normalizeMssqlConfig(rawConfig, meta = {}) { if (!rawConfig) { return null; } const driver = String(rawConfig.driver || 'mssql').trim().toLowerCase(); if (driver && driver !== 'mssql') { throw new Error('业务数据库当前仅支持 SQL Server'); } const host = String(rawConfig.host || '').trim(); const databaseName = String(rawConfig.database || rawConfig.database_name || '').trim(); const user = String(rawConfig.user || rawConfig.username || '').trim(); if (!host || !databaseName || !user) { return null; } const port = Number.parseInt(rawConfig.port || 1433, 10) || 1433; const ssl = parseBoolean(rawConfig.ssl, false); return { mssqlConfig: { server: host, port, database: databaseName, user, password: rawConfig.password || '', options: { encrypt: ssl, trustServerCertificate: true, connectTimeout: DEFAULT_CONNECT_TIMEOUT_MS, requestTimeout: DEFAULT_REQUEST_TIMEOUT_MS, // 业务库的 DATETIME 存的是北京时间(无时区标记)。mssql 驱动默认 useUTC=true 会把它 // 当成 UTC 解析,而本项目所有时间格式化都是用本地 getter 还原墙上时间,两者叠加会让 // 显示值整体 +8 小时(例如库里 16:56:07 被显示成次日 00:56:07)。按本地时间解析才对得上。 useUTC: false, }, pool: { max: CLOUD_DB_POOL_MAX, min: 0, idleTimeoutMillis: CLOUD_DB_POOL_IDLE_TIMEOUT_MS, }, }, runtimeInfo: { ...meta, driver, host, port, database: databaseName, username: user, ssl, }, }; } function getEnvBusinessDatabaseConfig(source = process.env, profile = 'default') { const envKeys = profile === 'public' ? PUBLIC_ENV_KEYS : DEFAULT_ENV_KEYS; return { driver: getFirstEnvValue(source, envKeys.driver, 'mssql'), host: getFirstEnvValue(source, envKeys.host), port: Number.parseInt(getFirstEnvValue(source, envKeys.port, '1433'), 10) || 1433, database: getFirstEnvValue(source, envKeys.database), user: getFirstEnvValue(source, envKeys.user), password: getFirstEnvValue(source, envKeys.password), ssl: parseBoolean(getFirstEnvValue(source, envKeys.ssl), false), }; } function hasCompleteEnvConfig(source = process.env, profile = 'default') { const config = getEnvBusinessDatabaseConfig(source, profile); return Boolean(config.host && config.database && config.user); } function getEnvBusinessDatabaseSummary(source = process.env, profile = 'default') { const config = getEnvBusinessDatabaseConfig(source, profile); return { available: hasCompleteEnvConfig(source, profile), driver: config.driver, host: config.host || '', port: config.port || 1433, database: config.database || '', username: config.user || '', ssl: Boolean(config.ssl), }; } function buildMssqlPoolConfigFromEnv(source = process.env, profile = 'default') { return normalizeMssqlConfig(getEnvBusinessDatabaseConfig(source, profile), { source: 'env', configName: profile === 'public' ? CLOUD_DATABASE_PUBLIC_CONFIG_NAME : CLOUD_DATABASE_CONFIG_NAME, selectedProfile: profile === 'public' ? 'public' : 'local', }); } async function createSystemConfigPool() { const systemDbConfig = getSystemDbRuntimeConfig(); return mysql.createPool({ host: systemDbConfig.host || 'localhost', port: parseInt(systemDbConfig.port || '3306', 10), user: systemDbConfig.username || 'root', password: systemDbConfig.password, database: systemDbConfig.database || 'shujuku2_system', charset: 'utf8mb4', waitForConnections: true, connectionLimit: LOCAL_CONFIG_DB_CONNECTION_LIMIT, maxIdle: LOCAL_CONFIG_DB_CONNECTION_LIMIT, idleTimeout: 60000, queueLimit: 10, connectTimeout: LOCAL_CONFIG_DB_CONNECT_TIMEOUT_MS, enableKeepAlive: true, keepAliveInitialDelay: 0, }); } async function loadStoredConfigFromDb(localPool, configName) { let rows = []; if (configName === CLOUD_DATABASE_CONFIG_NAME) { [rows] = await localPool.query( 'SELECT name, host, port, database_name, username, password, `ssl`, driver, use_env_config, is_default FROM database_configs WHERE name = ? AND status = ? LIMIT 1', [configName, 'active'] ); if (rows.length === 0) { [rows] = await localPool.query( 'SELECT name, host, port, database_name, username, password, `ssl`, driver, use_env_config, is_default FROM database_configs WHERE is_default = 1 AND status = ? LIMIT 1', ['active'] ); } } else { [rows] = await localPool.query( 'SELECT name, host, port, database_name, username, password, `ssl`, driver, use_env_config, is_default FROM database_configs WHERE name = ? AND status = ? LIMIT 1', [configName, 'active'] ); } const config = rows[0]; if (!config) { return null; } return normalizeMssqlConfig(config, { source: 'system_database', configName: config.name || configName, selectedProfile: configName === CLOUD_DATABASE_PUBLIC_CONFIG_NAME ? 'public' : 'local', useEnvConfig: isTruthy(config.use_env_config), isDefault: isTruthy(config.is_default), }); } async function loadCandidateConfig(localPool, profile) { if (profile === 'local') { const stored = await loadStoredConfigFromDb(localPool, CLOUD_DATABASE_CONFIG_NAME); if (stored?.runtimeInfo?.useEnvConfig) { if (hasCompleteEnvConfig(process.env, 'default')) { console.log('[CloudDB] 使用 .env 业务数据库配置(CLOUD_DB_*)'); return buildMssqlPoolConfigFromEnv(process.env, 'default'); } console.warn('[CloudDB] 已开启 .env 业务数据库配置,但 CLOUD_DB_* 不完整,继续使用 database_configs 表配置'); } return stored || buildMssqlPoolConfigFromEnv(process.env, 'default'); } const stored = await loadStoredConfigFromDb(localPool, CLOUD_DATABASE_PUBLIC_CONFIG_NAME); if (stored?.runtimeInfo?.useEnvConfig) { if (hasCompleteEnvConfig(process.env, 'public')) { console.log('[CloudDB] 使用 .env 公网业务数据库配置(CLOUD_DB_PUBLIC_*)'); return buildMssqlPoolConfigFromEnv(process.env, 'public'); } console.warn('[CloudDB] 已开启 .env 公网业务数据库配置,但 CLOUD_DB_PUBLIC_* 不完整,继续使用 database_configs 表配置'); } return stored || buildMssqlPoolConfigFromEnv(process.env, 'public'); } async function resolveCloudConfigCandidates() { // 缓存有效判据:**非 null** 且在有效期内。 // // 旧实现是 `if (cachedCloudConfigCandidates)`,而 [] 在 JS 里是 truthy —— 一旦系统库 // 读取失败一次(隧道抖动),就会把空数组当作合法缓存永久返回,之后永不重试, // 直接抛「业务数据库尚未配置」500,直到重启后端。这是典型的「永久负缓存」反模式。 // // 现在:成功态 expiresAt = Infinity(靠 resetPool 显式失效);失败态带 30s TTL, // 到期自动重新解析,隧道恢复后无需人工干预即可自愈。 if (cachedCloudConfigCandidates !== null && Date.now() < cachedCloudConfigExpiresAt) { return cachedCloudConfigCandidates; } let localPool; try { localPool = await createSystemConfigPool(); const profileOrder = getCloudProfileOrder(); const candidates = []; const seen = new Set(); for (const profile of profileOrder) { const candidate = await loadCandidateConfig(localPool, profile); if (!candidate?.mssqlConfig) { continue; } const dedupeKey = [ candidate.runtimeInfo.selectedProfile, candidate.runtimeInfo.source, candidate.runtimeInfo.host, candidate.runtimeInfo.port, candidate.runtimeInfo.database, candidate.runtimeInfo.username, ].join('::'); if (seen.has(dedupeKey)) { continue; } seen.add(dedupeKey); candidates.push(candidate); } cachedCloudConfigCandidates = candidates; cachedCloudConfigExpiresAt = Number.POSITIVE_INFINITY; return cachedCloudConfigCandidates; } catch (error) { console.warn('[CloudDB] 无法从数据库读取配置:', error.message); } finally { if (localPool) { await localPool.end().catch((error) => { console.warn('[CloudDB] 关闭本地配置库临时连接池失败:', error.message); }); } } // 失败态:按需回退到环境变量配置,并把结果记为「带 TTL 的负缓存」。 const fallbacks = []; if (hasCompleteEnvConfig(process.env, 'default')) { console.log('[CloudDB] 使用环境变量配置(CLOUD_DB_*)'); fallbacks.push(buildMssqlPoolConfigFromEnv(process.env, 'default')); } if (hasCompleteEnvConfig(process.env, 'public')) { console.log('[CloudDB] 使用环境变量配置(CLOUD_DB_PUBLIC_*)'); fallbacks.push(buildMssqlPoolConfigFromEnv(process.env, 'public')); } cachedCloudConfigCandidates = fallbacks; cachedCloudConfigExpiresAt = Date.now() + NEGATIVE_CONFIG_CACHE_TTL_MS; console.warn( `[CloudDB] 系统库配置暂不可用(${fallbacks.length > 0 ? '暂用环境变量兜底' : '且无环境变量兜底'}),` + `${NEGATIVE_CONFIG_CACHE_TTL_MS}ms 后自动重试` ); return cachedCloudConfigCandidates; } async function connectWithCandidate(candidate) { const nextPool = new sql.ConnectionPool(candidate.mssqlConfig); await nextPool.connect(); pool = nextPool; activeCloudConfig = candidate; return pool; } async function queryWithCandidateConnection(candidate, sqlText, params = {}, options = {}) { const nextPool = new sql.ConnectionPool(candidate.mssqlConfig); try { await nextPool.connect(); const resolvedSqlText = rewriteReportReplicaSql(sqlText, candidate.runtimeInfo); const request = nextPool.request(); const requestTimeout = parsePositiveInt(options.requestTimeout, 0); if (requestTimeout > 0) { request.timeout = requestTimeout; } if (params && typeof params === 'object') { Object.entries(params).forEach(([key, value]) => { request.input(key, value); }); } const result = await request.query(resolvedSqlText); return result.recordset; } finally { await nextPool.close().catch((error) => { console.warn('[CloudDB] 关闭临时连接失败:', error.message); }); } } function prioritizeCandidates(candidates = [], preferredProfiles = []) { if (!Array.isArray(candidates) || candidates.length <= 1 || !preferredProfiles.length) { return candidates; } const preferred = []; const fallback = []; const normalizedProfiles = preferredProfiles .map((profile) => String(profile || '').trim().toLowerCase()) .filter(Boolean); for (const candidate of candidates) { const profile = String(candidate?.runtimeInfo?.selectedProfile || '').trim().toLowerCase(); if (normalizedProfiles.includes(profile)) { preferred.push(candidate); } else { fallback.push(candidate); } } return [...preferred, ...fallback]; } async function verifyCandidate(candidate, options = {}) { if (!options.verifySql) { return; } const rows = await queryWithCandidateConnection( candidate, options.verifySql, options.verifyParams || {}, options.verifyOptions || {} ); if (typeof options.verifyResult === 'function') { const verification = await options.verifyResult(rows, candidate.runtimeInfo); if (verification === false) { throw new Error(options.verifyFailureMessage || '业务数据库候选验证失败'); } if (typeof verification === 'string' && verification.trim()) { throw new Error(verification.trim()); } } } function createQueryExecutorForCandidate(candidate) { let candidatePool = null; const ensureCandidatePool = async () => { if (isPoolUsable(candidatePool)) { return candidatePool; } if (candidatePool) { try { await candidatePool.close(); } catch (error) { console.warn('[CloudDB] 关闭固定候选连接池失败:', error.message); } finally { candidatePool = null; } } candidatePool = new sql.ConnectionPool(candidate.mssqlConfig); await candidatePool.connect(); return candidatePool; }; const closeCandidatePool = async () => { if (!candidatePool) { return; } try { await candidatePool.close(); } catch (error) { console.warn('[CloudDB] 关闭固定候选连接池失败:', error.message); } finally { candidatePool = null; } }; return { runtimeInfo: candidate.runtimeInfo, async query(sqlText, params = {}, options = {}) { const maxAttempts = parsePositiveInt(options.maxAttempts, 2); for (let attempt = 1; attempt <= maxAttempts; attempt += 1) { try { const activePool = await ensureCandidatePool(); const resolvedSqlText = rewriteReportReplicaSql(sqlText, candidate.runtimeInfo); const request = activePool.request(); const requestTimeout = parsePositiveInt(options.requestTimeout, 0); if (requestTimeout > 0) { request.timeout = requestTimeout; } if (params && typeof params === 'object') { Object.entries(params).forEach(([key, value]) => { request.input(key, value); }); } const result = await request.query(resolvedSqlText); return result.recordset; } catch (error) { const shouldRetry = attempt < maxAttempts && isTransientError(error); if (!shouldRetry) { error.cloudDbRuntime = candidate.runtimeInfo; throw error; } console.warn( `[CloudDB] 固定候选查询失败,准备重连后重试 (${attempt}/${maxAttempts}): ${error.message}` ); await closeCandidatePool(); } } return []; }, async close() { await closeCandidatePool(); }, }; } async function createPinnedQueryClient(options = {}) { const candidates = prioritizeCandidates( await resolveCloudConfigCandidates(), [ options.preferredProfile, ...(Array.isArray(options.preferredProfiles) ? options.preferredProfiles : []), ] ); if (!Array.isArray(candidates) || candidates.length === 0) { throw new Error(NOT_CONFIGURED_MESSAGE); } const errors = []; for (const candidate of candidates) { try { await verifyCandidate(candidate, options); return createQueryExecutorForCandidate(candidate); } catch (error) { errors.push(error); console.warn( `[CloudDB] 固定候选 ${candidate.runtimeInfo.selectedProfile} 验证失败: ${error.message}` ); } } const lastError = errors[errors.length - 1]; throw lastError || new Error(NOT_CONFIGURED_MESSAGE); } /** * 获取连接池 */ const getPool = async () => { if (isPoolUsable(pool)) { return pool; } if (pool) { console.warn('[CloudDB] 检测到失效连接池,准备重建连接'); await resetPool(); } const candidates = await resolveCloudConfigCandidates(); if (!Array.isArray(candidates) || candidates.length === 0) { throw new Error(NOT_CONFIGURED_MESSAGE); } const errors = []; for (const candidate of candidates) { try { console.log( `[CloudDB] 尝试连接 ${candidate.runtimeInfo.selectedProfile} 配置: ` + `${candidate.runtimeInfo.host}:${candidate.runtimeInfo.port}/${candidate.runtimeInfo.database}` ); await connectWithCandidate(candidate); console.log('[CloudDB] MSSQL 云数据库连接成功'); return pool; } catch (error) { errors.push(error); console.warn(`[CloudDB] ${candidate.runtimeInfo.selectedProfile} 配置连接失败: ${error.message}`); } } const lastError = errors[errors.length - 1]; console.error('[CloudDB] MSSQL 连接失败:', lastError?.message || NOT_CONFIGURED_MESSAGE); throw lastError || new Error(NOT_CONFIGURED_MESSAGE); }; /** * 统一失效「云配置候选」缓存。 * * 抽成单点:数组与负缓存 TTL 时间戳必须同时清,只清数组会让陈旧的 expiresAt 残留, * 下次读取时 `Date.now() < expiresAt` 仍可能成立 → 又退化回「负缓存不失效」。 */ function invalidateCloudConfigCache() { cachedCloudConfigCandidates = null; cachedCloudConfigExpiresAt = 0; } const resetPool = async () => { if (!pool) { activeCloudConfig = null; invalidateCloudConfigCache(); return; } try { await pool.close(); } catch (error) { console.warn('[CloudDB] 关闭异常连接池失败:', error.message); } finally { pool = null; activeCloudConfig = null; invalidateCloudConfigCache(); } }; /** * 执行 SQL 查询 * @param {string} sqlText - SQL 语句 * @param {object} params - 命名参数对象 */ const query = async (sqlText, params = {}, options = {}) => { const maxAttempts = parsePositiveInt(options.maxAttempts, 2); for (let attempt = 1; attempt <= maxAttempts; attempt += 1) { try { const activePool = await getPool(); const resolvedSqlText = rewriteReportReplicaSql(sqlText, activeCloudConfig?.runtimeInfo); const request = activePool.request(); const requestTimeout = parsePositiveInt(options.requestTimeout, 0); if (requestTimeout > 0) { request.timeout = requestTimeout; } if (params && typeof params === 'object') { Object.entries(params).forEach(([key, value]) => { request.input(key, value); }); } const result = await request.query(resolvedSqlText); return result.recordset; } catch (error) { const shouldRetry = attempt < maxAttempts && isTransientError(error); if (!shouldRetry) { throw error; } console.warn(`[CloudDB] 查询失败,准备重连后重试 (${attempt}/${maxAttempts}): ${error.message}`); await resetPool(); } } return []; }; const queryWithFreshConnection = async (sqlText, params = {}, options = {}) => { const maxAttempts = parsePositiveInt(options.maxAttempts, 2); for (let attempt = 1; attempt <= maxAttempts; attempt += 1) { const candidates = await resolveCloudConfigCandidates(); if (!Array.isArray(candidates) || candidates.length === 0) { // 候选为空 = 本轮配置解析失败,且没有任何环境变量兜底。 // // 这里刻意**不**清缓存:失败态已由 resolveCloudConfigCandidates 写成「带 TTL 的负缓存」。 // TTL 内的后续请求快速失败(不必每个请求都去撞系统库的慢超时),到期自动重新解析 —— // 隧道恢复后无需重启后端。旧实现把这次失败永久固化,正是「必须重启后端」的根因。 // // 注:getPool() 里的同款分支行为与此保持一致,两条路径统一由 TTL 负责恢复; // 「坏连接池」由 query() 的 catch → resetPool() 负责,各司其职。 throw new Error(NOT_CONFIGURED_MESSAGE); } let lastError = null; for (const candidate of candidates) { try { return await queryWithCandidateConnection(candidate, sqlText, params, options); } catch (error) { lastError = error; } } const shouldRetry = attempt < maxAttempts && isTransientError(lastError); if (!shouldRetry) { throw lastError || new Error(NOT_CONFIGURED_MESSAGE); } } return []; }; /** * 测试连接 */ const test = async () => { try { // 「测试连接」是用户在系统设置里验证配置的入口,必须读最新配置: // 先显式失效,避免 30s 内的失败态负缓存让用户看到「刚改好却连不上」。 invalidateCloudConfigCache(); const candidates = await resolveCloudConfigCandidates(); if (!Array.isArray(candidates) || candidates.length === 0) { return { success: false, message: NOT_CONFIGURED_MESSAGE }; } const activePool = await getPool(); const result = await activePool.request().query('SELECT 1 as test'); return { success: true, message: '业务数据库连接成功', data: result.recordset, runtime: activeCloudConfig?.runtimeInfo || null, }; } catch (error) { return { success: false, message: `连接失败: ${error.message}` }; } }; /** * 关闭连接 */ const close = async () => { if (pool) { await pool.close(); console.log('[CloudDB] 连接已关闭'); } pool = null; activeCloudConfig = null; invalidateCloudConfigCache(); }; /** * 重新加载配置 */ const reload = async () => { await resetPool(); return getPool(); }; module.exports = { getPool, query, queryWithFreshConnection, createPinnedQueryClient, test, close, reload, getEnvBusinessDatabaseConfig, getEnvBusinessDatabaseSummary, hasCompleteEnvConfig, rewriteReportReplicaSql, getReportReplicaDatabaseName, getReportReplicaSchemaName, getReportReplicaAppSchemaName, CLOUD_DATABASE_CONFIG_NAME, CLOUD_DATABASE_PUBLIC_CONFIG_NAME, };