cassel/bi-agents
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: use pool instead of single conn for parallel queries

Browse Source 
Single MySQL connection cannot handle concurrent prepared statements
via Promise.all — causes protocol conflicts and JSON parse errors.
Switch all bi.queries.js functions to use pool.execute() (each query
gets its own connection from pool). Bump connectionLimit 10→20.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
root
2026-02-16 23:09:00 -05:00
parent ab7a886836
commit 42803bd946
2 changed files with 45 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/db-rds.js
View File

@@ -9,7 +9,7 @@ const pool = mysql.createPool({
password: process.env.PW_MYSQL, password: process.env.PW_MYSQL,
database: 'cambio_db', database: 'cambio_db',
waitForConnections: true, waitForConnections: true,
connectionLimit: 10, connectionLimit: 20,
}); });
module.exports = pool; module.exports = pool;

103
src/queries/bi.queries.js
View File

@@ -6,19 +6,17 @@ const { pool, fmtDate, fmtTrendRows, calcPrevPeriod } = require('./helpers');
// BI Analytics - Comprehensive data for admin BI dashboard // BI Analytics - Comprehensive data for admin BI dashboard
async function fetchBIData(dataInicio, dataFim, getAgenteName = null) { async function fetchBIData(dataInicio, dataFim, getAgenteName = null) {
const conn = await pool.getConnection(); const { prevStartStr, prevEndStr } = calcPrevPeriod(dataInicio, dataFim);
try {
const { prevStartStr, prevEndStr } = calcPrevPeriod(dataInicio, dataFim);
// Run all 12 queries in parallel // Run all queries in parallel using pool (each gets its own connection)
const [ const [
[kpiBrlUsd], [kpiUsdBrl], [kpiUsdUsd], [uniqueClients], [kpiBrlUsd], [kpiUsdBrl], [kpiUsdUsd], [uniqueClients],
[prevBrlUsd], [prevUsdBrl], [prevUsdUsd], [prevBrlUsd], [prevUsdBrl], [prevUsdUsd],
[trendBrlUsd], [trendUsdBrl], [trendUsdUsd], [trendBrlUsd], [trendUsdBrl], [trendUsdUsd],
[topClients], [retention], [clientsAtRisk], [agentRanking] [topClients], [retention], [clientsAtRisk], [agentRanking]
] = await Promise.all([ ] = await Promise.all([
// 1. BRL→USD KPIs // 1. BRL→USD KPIs
conn.execute(` pool.execute(`
SELECT COUNT(*) as qtd, SELECT COUNT(*) as qtd,
ROUND(COALESCE(SUM(amount_usd), 0), 2) as vol_usd, ROUND(COALESCE(SUM(amount_usd), 0), 2) as vol_usd,
ROUND(COALESCE(SUM(amount_brl), 0), 2) as vol_brl, ROUND(COALESCE(SUM(amount_brl), 0), 2) as vol_brl,
@@ -29,7 +27,7 @@ async function fetchBIData(dataInicio, dataFim, getAgenteName = null) {
WHERE DATE(created_at) >= ? AND DATE(created_at) <= ? WHERE DATE(created_at) >= ? AND DATE(created_at) <= ?
`, [dataInicio, dataFim]), `, [dataInicio, dataFim]),
// 2. USD→BRL KPIs // 2. USD→BRL KPIs
conn.execute(` pool.execute(`
SELECT COUNT(*) as qtd, SELECT COUNT(*) as qtd,
ROUND(COALESCE(SUM(valor), 0), 2) as vol_usd, ROUND(COALESCE(SUM(valor), 0), 2) as vol_usd,
ROUND(COALESCE(SUM(valor_sol), 0), 2) as vol_brl, ROUND(COALESCE(SUM(valor_sol), 0), 2) as vol_brl,
@@ -42,7 +40,7 @@ async function fetchBIData(dataInicio, dataFim, getAgenteName = null) {
AND (pgto IS NULL OR pgto != 'balance') AND (pgto IS NULL OR pgto != 'balance')
`, [dataInicio, dataFim]), `, [dataInicio, dataFim]),
// 3. USD→USD KPIs // 3. USD→USD KPIs
conn.execute(` pool.execute(`
SELECT COUNT(*) as qtd, SELECT COUNT(*) as qtd,
ROUND(COALESCE(SUM(valor), 0), 2) as vol_usd, ROUND(COALESCE(SUM(valor), 0), 2) as vol_usd,
COUNT(DISTINCT id_conta) as clientes COUNT(DISTINCT id_conta) as clientes
@@ -51,7 +49,7 @@ async function fetchBIData(dataInicio, dataFim, getAgenteName = null) {
AND (cotacao IS NULL OR cotacao = 0 OR pgto = 'balance') AND (cotacao IS NULL OR cotacao = 0 OR pgto = 'balance')
`, [dataInicio, dataFim]), `, [dataInicio, dataFim]),
// 4. Unique active clients // 4. Unique active clients
conn.execute(` pool.execute(`
SELECT COUNT(DISTINCT id_conta) as total FROM ( SELECT COUNT(DISTINCT id_conta) as total FROM (
SELECT id_conta FROM br_transaction_to_usa WHERE DATE(created_at) >= ? AND DATE(created_at) <= ? SELECT id_conta FROM br_transaction_to_usa WHERE DATE(created_at) >= ? AND DATE(created_at) <= ?
UNION UNION
@@ -59,31 +57,31 @@ async function fetchBIData(dataInicio, dataFim, getAgenteName = null) {
) all_clients ) all_clients
`, [dataInicio, dataFim, dataInicio, dataFim]), `, [dataInicio, dataFim, dataInicio, dataFim]),
// 5. Previous period // 5. Previous period
conn.execute(` pool.execute(`
SELECT COUNT(*) as qtd, ROUND(COALESCE(SUM(amount_usd),0),2) as vol_usd, SELECT COUNT(*) as qtd, ROUND(COALESCE(SUM(amount_usd),0),2) as vol_usd,
ROUND(COALESCE(SUM((exchange_rate - ptax) / exchange_rate * amount_usd),0),2) as spread_revenue ROUND(COALESCE(SUM((exchange_rate - ptax) / exchange_rate * amount_usd),0),2) as spread_revenue
FROM br_transaction_to_usa WHERE DATE(created_at) >= ? AND DATE(created_at) <= ? FROM br_transaction_to_usa WHERE DATE(created_at) >= ? AND DATE(created_at) <= ?
`, [prevStartStr, prevEndStr]), `, [prevStartStr, prevEndStr]),
conn.execute(` pool.execute(`
SELECT COUNT(*) as qtd, ROUND(COALESCE(SUM(valor),0),2) as vol_usd, SELECT COUNT(*) as qtd, ROUND(COALESCE(SUM(valor),0),2) as vol_usd,
ROUND(COALESCE(SUM((ptax - cotacao) / ptax * valor),0),2) as spread_revenue ROUND(COALESCE(SUM((ptax - cotacao) / ptax * valor),0),2) as spread_revenue
FROM pagamento_br WHERE DATE(created_at) >= ? AND DATE(created_at) <= ? FROM pagamento_br WHERE DATE(created_at) >= ? AND DATE(created_at) <= ?
AND cotacao IS NOT NULL AND cotacao > 0 AND (pgto IS NULL OR pgto != 'balance') AND cotacao IS NOT NULL AND cotacao > 0 AND (pgto IS NULL OR pgto != 'balance')
`, [prevStartStr, prevEndStr]), `, [prevStartStr, prevEndStr]),
conn.execute(` pool.execute(`
SELECT COUNT(*) as qtd, ROUND(COALESCE(SUM(valor),0),2) as vol_usd SELECT COUNT(*) as qtd, ROUND(COALESCE(SUM(valor),0),2) as vol_usd
FROM pagamento_br WHERE DATE(created_at) >= ? AND DATE(created_at) <= ? FROM pagamento_br WHERE DATE(created_at) >= ? AND DATE(created_at) <= ?
AND (cotacao IS NULL OR cotacao = 0 OR pgto = 'balance') AND (cotacao IS NULL OR cotacao = 0 OR pgto = 'balance')
`, [prevStartStr, prevEndStr]), `, [prevStartStr, prevEndStr]),
// 6-8. Trends // 6-8. Trends
conn.execute(` pool.execute(`
SELECT DATE(created_at) as dia, COUNT(*) as qtd, SELECT DATE(created_at) as dia, COUNT(*) as qtd,
ROUND(SUM(amount_usd), 2) as vol_usd, ROUND(SUM(amount_usd), 2) as vol_usd,
ROUND(AVG((exchange_rate - ptax) / exchange_rate * 100), 2) as avg_spread ROUND(AVG((exchange_rate - ptax) / exchange_rate * 100), 2) as avg_spread
FROM br_transaction_to_usa WHERE DATE(created_at) >= ? AND DATE(created_at) <= ? FROM br_transaction_to_usa WHERE DATE(created_at) >= ? AND DATE(created_at) <= ?
GROUP BY DATE(created_at) ORDER BY dia GROUP BY DATE(created_at) ORDER BY dia
`, [dataInicio, dataFim]), `, [dataInicio, dataFim]),
conn.execute(` pool.execute(`
SELECT DATE(created_at) as dia, COUNT(*) as qtd, SELECT DATE(created_at) as dia, COUNT(*) as qtd,
ROUND(SUM(valor), 2) as vol_usd, ROUND(SUM(valor), 2) as vol_usd,
ROUND(AVG(CASE WHEN cotacao > 0 THEN (ptax - cotacao) / ptax * 100 ELSE 0 END), 2) as avg_spread ROUND(AVG(CASE WHEN cotacao > 0 THEN (ptax - cotacao) / ptax * 100 ELSE 0 END), 2) as avg_spread
@@ -91,14 +89,14 @@ async function fetchBIData(dataInicio, dataFim, getAgenteName = null) {
AND cotacao IS NOT NULL AND cotacao > 0 AND (pgto IS NULL OR pgto != 'balance') AND cotacao IS NOT NULL AND cotacao > 0 AND (pgto IS NULL OR pgto != 'balance')
GROUP BY DATE(created_at) ORDER BY dia GROUP BY DATE(created_at) ORDER BY dia
`, [dataInicio, dataFim]), `, [dataInicio, dataFim]),
conn.execute(` pool.execute(`
SELECT DATE(created_at) as dia, COUNT(*) as qtd, ROUND(SUM(valor), 2) as vol_usd SELECT DATE(created_at) as dia, COUNT(*) as qtd, ROUND(SUM(valor), 2) as vol_usd
FROM pagamento_br WHERE DATE(created_at) >= ? AND DATE(created_at) <= ? FROM pagamento_br WHERE DATE(created_at) >= ? AND DATE(created_at) <= ?
AND (cotacao IS NULL OR cotacao = 0 OR pgto = 'balance') AND (cotacao IS NULL OR cotacao = 0 OR pgto = 'balance')
GROUP BY DATE(created_at) ORDER BY dia GROUP BY DATE(created_at) ORDER BY dia
`, [dataInicio, dataFim]), `, [dataInicio, dataFim]),
// 9. Top clients // 9. Top clients
conn.execute(` pool.execute(`
SELECT nome, SUM(vol) as total_usd, SUM(qtd) as total_qtd FROM ( SELECT nome, SUM(vol) as total_usd, SUM(qtd) as total_qtd FROM (
SELECT c.nome, SUM(t.amount_usd) as vol, COUNT(*) as qtd SELECT c.nome, SUM(t.amount_usd) as vol, COUNT(*) as qtd
FROM br_transaction_to_usa t INNER JOIN conta c ON c.id_conta = t.id_conta FROM br_transaction_to_usa t INNER JOIN conta c ON c.id_conta = t.id_conta
@@ -110,7 +108,7 @@ async function fetchBIData(dataInicio, dataFim, getAgenteName = null) {
) combined GROUP BY nome ORDER BY total_usd DESC LIMIT 10 ) combined GROUP BY nome ORDER BY total_usd DESC LIMIT 10
`, [dataInicio, dataFim, dataInicio, dataFim]), `, [dataInicio, dataFim, dataInicio, dataFim]),
// 10. Retention // 10. Retention
conn.execute(` pool.execute(`
SELECT COUNT(DISTINCT prev.id_conta) as prev_clients, SELECT COUNT(DISTINCT prev.id_conta) as prev_clients,
COUNT(DISTINCT CASE WHEN curr.id_conta IS NOT NULL THEN prev.id_conta END) as retained COUNT(DISTINCT CASE WHEN curr.id_conta IS NOT NULL THEN prev.id_conta END) as retained
FROM ( FROM (
@@ -124,7 +122,7 @@ async function fetchBIData(dataInicio, dataFim, getAgenteName = null) {
) curr ON prev.id_conta = curr.id_conta ) curr ON prev.id_conta = curr.id_conta
`, [prevStartStr, prevEndStr, prevStartStr, prevEndStr, dataInicio, dataFim, dataInicio, dataFim]), `, [prevStartStr, prevEndStr, prevStartStr, prevEndStr, dataInicio, dataFim, dataInicio, dataFim]),
// 11. Clients at risk // 11. Clients at risk
conn.execute(` pool.execute(`
SELECT nome, MAX(last_op) as last_op, SUM(vol) as total_usd, SUM(qtd) as total_qtd, SELECT nome, MAX(last_op) as last_op, SUM(vol) as total_usd, SUM(qtd) as total_qtd,
DATEDIFF(CURDATE(), MAX(last_op)) as days_inactive DATEDIFF(CURDATE(), MAX(last_op)) as days_inactive
FROM ( FROM (
@@ -136,7 +134,7 @@ async function fetchBIData(dataInicio, dataFim, getAgenteName = null) {
) combined GROUP BY nome HAVING MAX(last_op) < CURDATE() ORDER BY total_usd DESC LIMIT 20 ) combined GROUP BY nome HAVING MAX(last_op) < CURDATE() ORDER BY total_usd DESC LIMIT 20
`), `),
// 12. Agent ranking // 12. Agent ranking
conn.execute(` pool.execute(`
SELECT agente_id, SUM(vol) as total_usd, SUM(qtd) as total_qtd, SELECT agente_id, SUM(vol) as total_usd, SUM(qtd) as total_qtd,
ROUND(SUM(spread_rev), 2) as total_spread, COUNT(DISTINCT client_id) as clientes ROUND(SUM(spread_rev), 2) as total_spread, COUNT(DISTINCT client_id) as clientes
FROM ( FROM (
@@ -214,15 +212,10 @@ async function fetchBIData(dataInicio, dataFim, getAgenteName = null) {
eficiencia: brl.vol_usd > 0 ? Math.min(100, Math.round(usd.vol_usd / brl.vol_usd * 100)) : 0 eficiencia: brl.vol_usd > 0 ? Math.min(100, Math.round(usd.vol_usd / brl.vol_usd * 100)) : 0
} }
}; };
} finally {
conn.release();
}
} }
// Revenue Analytics - Real P&L by product with dynamic granularity // Revenue Analytics - Real P&L by product with dynamic granularity
async function fetchRevenueAnalytics(dataInicio, dataFim, granularity = 'dia') { async function fetchRevenueAnalytics(dataInicio, dataFim, granularity = 'dia') {
const conn = await pool.getConnection();
try {
const validGran = ['dia', 'mes', 'ano'].includes(granularity) ? granularity : 'dia'; const validGran = ['dia', 'mes', 'ano'].includes(granularity) ? granularity : 'dia';
let periodoInicio, periodoLabel; let periodoInicio, periodoLabel;
@@ -240,7 +233,7 @@ async function fetchRevenueAnalytics(dataInicio, dataFim, granularity = 'dia') {
periodoLabel = "DATE_FORMAT(dia, '%Y-%m-%d')"; periodoLabel = "DATE_FORMAT(dia, '%Y-%m-%d')";
} }
const [rows] = await conn.execute(` const [rows] = await pool.execute(`
WITH limites AS ( WITH limites AS (
SELECT SELECT
CAST(? AS DATE) AS inicio, CAST(? AS DATE) AS inicio,
@@ -316,7 +309,7 @@ async function fetchRevenueAnalytics(dataInicio, dataFim, granularity = 'dia') {
`, [dataInicio, dataFim]); `, [dataInicio, dataFim]);
// Also get totals by product // Also get totals by product
const [totals] = await conn.execute(` const [totals] = await pool.execute(`
WITH limites AS ( WITH limites AS (
SELECT SELECT
CAST(? AS DATE) AS inicio, CAST(? AS DATE) AS inicio,
@@ -401,33 +394,25 @@ async function fetchRevenueAnalytics(dataInicio, dataFim, granularity = 'dia') {
}, },
granularity: validGran granularity: validGran
}; };
} finally {
conn.release();
}
} }
async function fetchBIStrategic(dataInicio, dataFim) { async function fetchBIStrategic(dataInicio, dataFim) {
const conn = await pool.getConnection(); const { prevStartStr, prevEndStr } = calcPrevPeriod(dataInicio, dataFim);
try {
const { prevStartStr, prevEndStr } = calcPrevPeriod(dataInicio, dataFim);
// Run all 4 sections in parallel — each is independent // Run all 4 sections in parallel — each uses pool (own connections)
const [cohorts, expansion, crossSell, maturity] = await Promise.all([ const [cohorts, expansion, crossSell, maturity] = await Promise.all([
_fetchCohorts(conn), _fetchCohorts(),
_fetchExpansion(conn, dataInicio, dataFim, prevStartStr, prevEndStr), _fetchExpansion(dataInicio, dataFim, prevStartStr, prevEndStr),
_fetchCrossSell(conn, dataInicio, dataFim), _fetchCrossSell(dataInicio, dataFim),
_fetchMaturity(conn, dataInicio, dataFim, prevStartStr, prevEndStr) _fetchMaturity(dataInicio, dataFim, prevStartStr, prevEndStr)
]); ]);
return { cohorts, expansion, crossSell, maturity }; return { cohorts, expansion, crossSell, maturity };
} finally {
conn.release();
}
} }
async function _fetchCohorts(conn) { async function _fetchCohorts() {
const [[cohortClients], [activeMonths]] = await Promise.all([ const [[cohortClients], [activeMonths]] = await Promise.all([
conn.execute(` pool.execute(`
SELECT id_conta, DATE_FORMAT(MIN(first_op), '%Y-%m') as cohort_month FROM ( SELECT id_conta, DATE_FORMAT(MIN(first_op), '%Y-%m') as cohort_month FROM (
SELECT id_conta, MIN(created_at) as first_op FROM br_transaction_to_usa GROUP BY id_conta SELECT id_conta, MIN(created_at) as first_op FROM br_transaction_to_usa GROUP BY id_conta
UNION ALL UNION ALL
@@ -436,7 +421,7 @@ async function _fetchCohorts(conn) {
GROUP BY id_conta GROUP BY id_conta
) f GROUP BY id_conta ) f GROUP BY id_conta
`), `),
conn.execute(` pool.execute(`
SELECT id_conta, active_month FROM ( SELECT id_conta, active_month FROM (
SELECT id_conta, DATE_FORMAT(created_at, '%Y-%m') as active_month SELECT id_conta, DATE_FORMAT(created_at, '%Y-%m') as active_month
FROM br_transaction_to_usa GROUP BY id_conta, DATE_FORMAT(created_at, '%Y-%m') FROM br_transaction_to_usa GROUP BY id_conta, DATE_FORMAT(created_at, '%Y-%m')
@@ -489,9 +474,9 @@ async function _fetchCohorts(conn) {
}); });
} }
async function _fetchExpansion(conn, dataInicio, dataFim, prevStartStr, prevEndStr) { async function _fetchExpansion(dataInicio, dataFim, prevStartStr, prevEndStr) {
const [[currRevenue], [prevRevenue]] = await Promise.all([ const [[currRevenue], [prevRevenue]] = await Promise.all([
conn.execute(` pool.execute(`
SELECT id_conta, ROUND(SUM(revenue), 2) as revenue, ROUND(SUM(vol_usd), 2) as vol_usd FROM ( SELECT id_conta, ROUND(SUM(revenue), 2) as revenue, ROUND(SUM(vol_usd), 2) as vol_usd FROM (
SELECT id_conta, SUM((exchange_rate - ptax) / exchange_rate * amount_usd) as revenue, SUM(amount_usd) as vol_usd SELECT id_conta, SUM((exchange_rate - ptax) / exchange_rate * amount_usd) as revenue, SUM(amount_usd) as vol_usd
FROM br_transaction_to_usa WHERE DATE(created_at) >= ? AND DATE(created_at) <= ? FROM br_transaction_to_usa WHERE DATE(created_at) >= ? AND DATE(created_at) <= ?
@@ -503,7 +488,7 @@ async function _fetchExpansion(conn, dataInicio, dataFim, prevStartStr, prevEndS
GROUP BY id_conta GROUP BY id_conta
) c GROUP BY id_conta ) c GROUP BY id_conta
`, [dataInicio, dataFim, dataInicio, dataFim]), `, [dataInicio, dataFim, dataInicio, dataFim]),
conn.execute(` pool.execute(`
SELECT id_conta, ROUND(SUM(revenue), 2) as revenue, ROUND(SUM(vol_usd), 2) as vol_usd FROM ( SELECT id_conta, ROUND(SUM(revenue), 2) as revenue, ROUND(SUM(vol_usd), 2) as vol_usd FROM (
SELECT id_conta, SUM((exchange_rate - ptax) / exchange_rate * amount_usd) as revenue, SUM(amount_usd) as vol_usd SELECT id_conta, SUM((exchange_rate - ptax) / exchange_rate * amount_usd) as revenue, SUM(amount_usd) as vol_usd
FROM br_transaction_to_usa WHERE DATE(created_at) >= ? AND DATE(created_at) <= ? FROM br_transaction_to_usa WHERE DATE(created_at) >= ? AND DATE(created_at) <= ?
@@ -552,8 +537,8 @@ async function _fetchExpansion(conn, dataInicio, dataFim, prevStartStr, prevEndS
return expansion; return expansion;
} }
async function _fetchCrossSell(conn, dataInicio, dataFim) { async function _fetchCrossSell(dataInicio, dataFim) {
const [crossSellData] = await conn.execute(` const [crossSellData] = await pool.execute(`
SELECT c.id_conta, t.vol_usd as pay_vol, p.vol_usd as checkout_vol SELECT c.id_conta, t.vol_usd as pay_vol, p.vol_usd as checkout_vol
FROM ( FROM (
SELECT DISTINCT id_conta FROM br_transaction_to_usa WHERE DATE(created_at) >= ? AND DATE(created_at) <= ? SELECT DISTINCT id_conta FROM br_transaction_to_usa WHERE DATE(created_at) >= ? AND DATE(created_at) <= ?
@@ -585,8 +570,8 @@ async function _fetchCrossSell(conn, dataInicio, dataFim) {
return crossSell; return crossSell;
} }
async function _fetchMaturity(conn, dataInicio, dataFim, prevStartStr, prevEndStr) { async function _fetchMaturity(dataInicio, dataFim, prevStartStr, prevEndStr) {
const [maturityData] = await conn.execute(` const [maturityData] = await pool.execute(`
SELECT id_conta, SELECT id_conta,
MIN(first_op) as first_op, MIN(first_op) as first_op,
FLOOR(DATEDIFF(CURDATE(), MIN(first_op)) / 30.44) as months_active, FLOOR(DATEDIFF(CURDATE(), MIN(first_op)) / 30.44) as months_active,