51b9f9eeed
- 新增库存监控页面、后端服务与 API，支持分页筛选与 CSV 导出 - 库存取值优先聚水潭实时库存(jst_inventoryInfo)，取不到回退日报快照(pm_stockMonitor) - 统计卡与预警分档统一全仓库口径，新增在库 SKU 款数 - 预留聚水潭加工单接入(jushuitanClient + 自检脚本)，凭证到位后可算采购在途
1141 lines
38 KiB
JavaScript
1141 lines
38 KiB
JavaScript
/**
|
||
* 云业务数据库连接模块（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;
|
||
let lastConnectFailure = null;
|
||
|
||
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);
|
||
const DEFAULT_REQUEST_TIMEOUT_MS = parsePositiveInt(process.env.CLOUD_DB_REQUEST_TIMEOUT_MS, 600000);
|
||
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);
|
||
// 连接建立失败后的冷静期（毫秒）。
|
||
// 若业务库被防火墙「丢包」（而非拒绝），每次连接都要死等到 connectTimeout 才会失败；
|
||
// 短时间内重复请求没必要把整套失败流程重跑一遍。
|
||
const CONNECT_FAILURE_COOLDOWN_MS = parsePositiveInt(process.env.CLOUD_DB_CONNECT_FAILURE_COOLDOWN_MS, 15000);
|
||
|
||
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,
|
||
},
|
||
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() {
|
||
if (cachedCloudConfigCandidates) {
|
||
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;
|
||
}
|
||
// 按真实连接目标去重：public / local 两个标签经常指向同一个 host:port，
|
||
// 若按 selectedProfile 去重则视为两个候选，一次请求会把同一个不可达地址串行连接两次。
|
||
const dedupeKey = [
|
||
candidate.runtimeInfo.host,
|
||
candidate.runtimeInfo.port,
|
||
candidate.runtimeInfo.database,
|
||
candidate.runtimeInfo.username,
|
||
candidate.mssqlConfig?.password || '',
|
||
].join('::');
|
||
if (seen.has(dedupeKey)) {
|
||
continue;
|
||
}
|
||
seen.add(dedupeKey);
|
||
candidates.push(candidate);
|
||
}
|
||
|
||
cachedCloudConfigCandidates = candidates;
|
||
return cachedCloudConfigCandidates;
|
||
} catch (error) {
|
||
console.warn('[CloudDB] 无法从数据库读取配置:', error.message);
|
||
} finally {
|
||
if (localPool) {
|
||
await localPool.end().catch((error) => {
|
||
console.warn('[CloudDB] 关闭本地配置库临时连接池失败:', error.message);
|
||
});
|
||
}
|
||
}
|
||
|
||
cachedCloudConfigCandidates = [];
|
||
if (hasCompleteEnvConfig(process.env, 'default')) {
|
||
console.log('[CloudDB] 使用环境变量配置（CLOUD_DB_*）');
|
||
cachedCloudConfigCandidates.push(buildMssqlPoolConfigFromEnv(process.env, 'default'));
|
||
}
|
||
if (hasCompleteEnvConfig(process.env, 'public')) {
|
||
console.log('[CloudDB] 使用环境变量配置（CLOUD_DB_PUBLIC_*）');
|
||
cachedCloudConfigCandidates.push(buildMssqlPoolConfigFromEnv(process.env, 'public'));
|
||
}
|
||
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 (options = {}) => {
|
||
if (isPoolUsable(pool)) {
|
||
return pool;
|
||
}
|
||
|
||
if (
|
||
lastConnectFailure
|
||
&& !options.bypassConnectFailureCache
|
||
&& Date.now() - lastConnectFailure.at < CONNECT_FAILURE_COOLDOWN_MS
|
||
) {
|
||
// 冷静期内直接复用上次的失败结果，避免重复等待连接超时
|
||
throw lastConnectFailure.error;
|
||
}
|
||
|
||
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);
|
||
lastConnectFailure = null;
|
||
console.log('[CloudDB] MSSQL 云数据库连接成功');
|
||
return pool;
|
||
} catch (error) {
|
||
// 附带候选配置信息，便于上层把报错翻译成「连不上哪台库」的可读提示
|
||
error.cloudDbRuntime = candidate.runtimeInfo;
|
||
errors.push(error);
|
||
console.warn(`[CloudDB] ${candidate.runtimeInfo.selectedProfile} 配置连接失败: ${error.message}`);
|
||
}
|
||
}
|
||
|
||
const lastError = errors[errors.length - 1];
|
||
if (lastError) {
|
||
// 标记为「连接建立失败」：这类失败重试没有意义，且要计入冷静期
|
||
lastError.cloudDbConnectFailure = true;
|
||
lastConnectFailure = { at: Date.now(), error: lastError };
|
||
}
|
||
console.error('[CloudDB] MSSQL 连接失败:', lastError?.message || NOT_CONFIGURED_MESSAGE);
|
||
throw lastError || new Error(NOT_CONFIGURED_MESSAGE);
|
||
};
|
||
|
||
const resetPool = async () => {
|
||
lastConnectFailure = null;
|
||
|
||
if (!pool) {
|
||
activeCloudConfig = null;
|
||
cachedCloudConfigCandidates = null;
|
||
return;
|
||
}
|
||
|
||
try {
|
||
await pool.close();
|
||
} catch (error) {
|
||
console.warn('[CloudDB] 关闭异常连接池失败:', error.message);
|
||
} finally {
|
||
pool = null;
|
||
activeCloudConfig = null;
|
||
cachedCloudConfigCandidates = null;
|
||
}
|
||
};
|
||
|
||
/**
|
||
* 执行 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(options);
|
||
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
|
||
&& !error.cloudDbConnectFailure
|
||
&& 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) {
|
||
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 {
|
||
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;
|
||
cachedCloudConfigCandidates = null;
|
||
lastConnectFailure = null;
|
||
};
|
||
|
||
/**
|
||
* 重新加载配置
|
||
*/
|
||
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,
|
||
};
|