const path = require('path'); const dotenv = require('dotenv'); const mysql = require('mysql2/promise'); const planningDb = require('./planningDatabase'); dotenv.config(); dotenv.config({ path: path.resolve(__dirname, '../../.env') }); dotenv.config({ path: path.resolve(__dirname, '../.env') }); function parsePositiveInt(value, fallback) { const parsed = Number.parseInt(value, 10); return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback; } 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 getDedicatedConfig() { if (parseBoolean(process.env.MARKET_REPLICA_DB_DISABLED, false)) { return null; } const host = String(process.env.MARKET_REPLICA_DB_HOST || process.env.MARKET_SYNC_SOURCE_HOST || '').trim(); const database = String( process.env.MARKET_REPLICA_DB_NAME || process.env.MARKET_REPLICA_DB_DATABASE || process.env.MARKET_SYNC_SOURCE_DATABASE || '' ).trim(); const user = String(process.env.MARKET_REPLICA_DB_USER || process.env.MARKET_SYNC_SOURCE_USER || '').trim(); if (!host || !database || !user) { return null; } return { host, port: parsePositiveInt(process.env.MARKET_REPLICA_DB_PORT || process.env.MARKET_SYNC_SOURCE_PORT, 3306), user, password: process.env.MARKET_REPLICA_DB_PASSWORD || process.env.MARKET_SYNC_SOURCE_PASSWORD || '', database, charset: process.env.MARKET_REPLICA_DB_CHARSET || 'utf8mb4', supportBigNumbers: true, bigNumberStrings: true, waitForConnections: true, connectionLimit: parsePositiveInt(process.env.MARKET_REPLICA_DB_CONNECTION_LIMIT, 3), maxIdle: parsePositiveInt(process.env.MARKET_REPLICA_DB_MAX_IDLE, 2), idleTimeout: parsePositiveInt(process.env.MARKET_REPLICA_DB_IDLE_TIMEOUT_MS, 30000), queueLimit: parsePositiveInt(process.env.MARKET_REPLICA_DB_QUEUE_LIMIT, 20), connectTimeout: parsePositiveInt(process.env.MARKET_REPLICA_DB_CONNECT_TIMEOUT_MS, 8000), enableKeepAlive: true, keepAliveInitialDelay: 0, ssl: parseBoolean(process.env.MARKET_REPLICA_DB_SSL, false) ? { rejectUnauthorized: false } : false, }; } let dedicatedPool = null; let dedicatedConfig = null; let dedicatedUnavailableUntil = 0; let dedicatedLastError = null; function isDedicatedFallbackEnabled() { return parseBoolean(process.env.MARKET_REPLICA_DB_FALLBACK_TO_PLANNING, true); } function getFallbackCooldownMs() { return parsePositiveInt(process.env.MARKET_REPLICA_DB_FALLBACK_COOLDOWN_MS, 60000); } function isDedicatedTemporarilyUnavailable() { return dedicatedUnavailableUntil > Date.now(); } function markDedicatedUnavailable(error) { dedicatedLastError = error; dedicatedUnavailableUntil = Date.now() + getFallbackCooldownMs(); } function isMissingSchemaObjectError(error) { const code = String(error?.code || '').trim().toUpperCase(); const errno = Number(error?.errno || 0); const message = String(error?.message || '').toLowerCase(); return code === 'ER_NO_SUCH_TABLE' || errno === 1146 || message.includes("doesn't exist") || message.includes('no such table'); } function isDedicatedConnectionError(error) { const code = String(error?.code || '').trim().toUpperCase(); const message = String(error?.message || '').toLowerCase(); return [ 'ECONNREFUSED', 'ECONNRESET', 'EHOSTUNREACH', 'ENETUNREACH', 'ENOTFOUND', 'ETIMEDOUT', 'ER_CON_COUNT_ERROR', 'ER_TOO_MANY_USER_CONNECTIONS', 'PROTOCOL_CONNECTION_LOST', 'PROTOCOL_ENQUEUE_AFTER_FATAL_ERROR', 'PROTOCOL_SEQUENCE_TIMEOUT', ].includes(code) || message.includes('connection is closed') || message.includes('connection lost') || message.includes('connect etimedout') || message.includes('too many connections') || message.includes('max_user_connections'); } async function closeDedicatedPool() { if (dedicatedPool) { await dedicatedPool.end().catch(() => {}); } dedicatedPool = null; } function getPool() { if (isDedicatedTemporarilyUnavailable()) { return null; } if (!dedicatedConfig) { dedicatedConfig = getDedicatedConfig(); } if (!dedicatedConfig) { return null; } if (!dedicatedPool) { dedicatedPool = mysql.createPool(dedicatedConfig); } return dedicatedPool; } function withQueryTimeout(sql, options = {}) { const timeout = parsePositiveInt( options.requestTimeout || process.env.MARKET_REPLICA_DB_QUERY_TIMEOUT_MS, 8000 ); return timeout ? { sql, timeout } : sql; } async function runWithDedicatedFallback(operation, fallbackOperation) { const pool = getPool(); if (!pool) { return fallbackOperation(); } try { return await operation(pool); } catch (error) { if (!isDedicatedFallbackEnabled()) { throw error; } if (isMissingSchemaObjectError(error)) { console.warn(`[MarketReplicaDB] 专用市场库缺少当前查询对象,仅本次回退企划库: ${error.message}`); return fallbackOperation(); } console.warn(`[MarketReplicaDB] 专用市场库不可用,${Math.round(getFallbackCooldownMs() / 1000)} 秒内回退企划库: ${error.message}`); markDedicatedUnavailable(error); await closeDedicatedPool(); return fallbackOperation(); } } async function query(sql, params = [], options = {}) { return runWithDedicatedFallback( async (pool) => { const [rows] = await pool.query(withQueryTimeout(sql, options), params); return rows; }, () => planningDb.query(sql, params, options) ); } async function queryDedicated(sql, params = [], options = {}) { const pool = getPool(); if (!pool) { const error = new Error( dedicatedLastError ? `市场专用数据库暂不可用: ${dedicatedLastError.message}` : '市场专用数据库未配置' ); error.code = 'MARKET_REPLICA_DEDICATED_UNAVAILABLE'; throw error; } try { const [rows] = await pool.query(withQueryTimeout(sql, options), params); return rows; } catch (error) { if (isDedicatedConnectionError(error)) { markDedicatedUnavailable(error); } throw error; } } async function execute(sql, params = [], options = {}) { return runWithDedicatedFallback( async (pool) => { const [result] = await pool.query(withQueryTimeout(sql, options), params); return result; }, () => planningDb.execute(sql, params, options) ); } async function getRuntimeInfo() { const pool = getPool(); if (!pool) { const runtime = await planningDb.getRuntimeInfo(); return { ...runtime, marketReplicaDedicated: false, marketReplicaFallback: 'planning_database', marketReplicaLastError: dedicatedLastError?.message || '', marketReplicaRetryAt: dedicatedUnavailableUntil || null, }; } return { runtimeProfile: 'market-replica-dedicated', selectedProfile: 'market-replica', source: 'env', configName: 'market_replica_database', host: dedicatedConfig.host, port: dedicatedConfig.port, database: dedicatedConfig.database, charset: dedicatedConfig.charset, ssl: Boolean(dedicatedConfig.ssl), marketReplicaDedicated: true, fallbackOrder: ['market-replica', 'planning_database'], }; } async function test() { const pool = getPool(); if (!pool) { const fallbackResult = await planningDb.test(); return { ...fallbackResult, message: fallbackResult.success ? '市场洞察专用库当前不可用,已回退企划数据库' : fallbackResult.message, runtime: await getRuntimeInfo().catch(() => fallbackResult.runtime), }; } try { const [rows] = await pool.query('SELECT 1 AS test'); return { success: true, message: '市场洞察专用数据库连接成功', data: rows, runtime: await getRuntimeInfo(), }; } catch (error) { markDedicatedUnavailable(error); await closeDedicatedPool(); const fallbackResult = await planningDb.test().catch((fallbackError) => ({ success: false, message: fallbackError.message || '企划数据库连接失败', })); return { success: Boolean(fallbackResult.success), message: fallbackResult.success ? `市场洞察专用库连接失败,已回退企划数据库: ${error.message}` : error.message || '市场洞察专用数据库连接失败', runtime: await getRuntimeInfo().catch(() => null), }; } } async function close() { await closeDedicatedPool(); dedicatedConfig = null; dedicatedUnavailableUntil = 0; dedicatedLastError = null; } async function reload() { await close(); return getPool(); } module.exports = { query, queryDedicated, execute, getRuntimeInfo, test, close, reload, _test: { isDedicatedConnectionError, isMissingSchemaObjectError, }, };