matt/dashboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Set up drop in replacement routes for data coming from acot connection that previously came from klaviyo

Browse Source
This commit is contained in:
Matt
2025-05-30 09:39:20 -04:00
parent 9db11531d6
commit 21185e23cf
12 changed files with 1686 additions and 275 deletions
Download Patch File Download Diff File Expand all files Collapse all files

269
dashboard-server/acot-server/db/connection.js
View File

@@ -2,15 +2,12 @@ const { Client } = require('ssh2');
const mysql = require('mysql2/promise');
const fs = require('fs');
// Connection pooling and cache configuration
const connectionCache = {
ssh: null,
dbConnection: null,
lastUsed: 0,
isConnecting: false,
connectionPromise: null,
// Cache expiration time in milliseconds (5 minutes)
expirationTime: 5 * 60 * 1000,
// Connection pool configuration
const connectionPool = {
connections: [],
maxConnections: 20,
currentConnections: 0,
pendingRequests: [],
// Cache for query results (key: query string, value: {data, timestamp})
queryCache: new Map(),
// Cache duration for different query types in milliseconds
@@ -19,90 +16,139 @@ const connectionCache = {
'products': 5 * 60 * 1000, // 5 minutes for products
'orders': 60 * 1000, // 1 minute for orders
'default': 60 * 1000 // 1 minute default
},
// Circuit breaker state
circuitBreaker: {
failures: 0,
lastFailure: 0,
isOpen: false,
threshold: 5,
timeout: 30000 // 30 seconds
}
};
/**
* Get a database connection with connection pooling
* @returns {Promise<{ssh: object, connection: object}>} The SSH and database connection
* Get a database connection from the pool
* @returns {Promise<{connection: object, release: function}>} The database connection and release function
*/
async function getDbConnection() {
const now = Date.now();
// Check if we need to refresh the connection due to inactivity
const needsRefresh = !connectionCache.ssh ||
!connectionCache.dbConnection ||
(now - connectionCache.lastUsed > connectionCache.expirationTime);
// If connection is still valid, update last used time and return existing connection
if (!needsRefresh) {
connectionCache.lastUsed = now;
return {
ssh: connectionCache.ssh,
connection: connectionCache.dbConnection
};
}
// If another request is already establishing a connection, wait for that promise
if (connectionCache.isConnecting && connectionCache.connectionPromise) {
try {
await connectionCache.connectionPromise;
return {
ssh: connectionCache.ssh,
connection: connectionCache.dbConnection
};
} catch (error) {
// If that connection attempt failed, we'll try again below
console.error('Error waiting for existing connection:', error);
return new Promise(async (resolve, reject) => {
// Check circuit breaker
const now = Date.now();
if (connectionPool.circuitBreaker.isOpen) {
if (now - connectionPool.circuitBreaker.lastFailure > connectionPool.circuitBreaker.timeout) {
// Reset circuit breaker
connectionPool.circuitBreaker.isOpen = false;
connectionPool.circuitBreaker.failures = 0;
console.log('Circuit breaker reset');
} else {
reject(new Error('Circuit breaker is open - too many connection failures'));
return;
}
}
}
// Close existing connections if they exist
if (connectionCache.dbConnection) {
try {
await connectionCache.dbConnection.end();
} catch (error) {
console.error('Error closing existing database connection:', error);
}
}
if (connectionCache.ssh) {
try {
connectionCache.ssh.end();
} catch (error) {
console.error('Error closing existing SSH connection:', error);
}
}
// Mark that we're establishing a new connection
connectionCache.isConnecting = true;
// Create a new promise for this connection attempt
connectionCache.connectionPromise = setupSshTunnel().then(tunnel => {
const { ssh, stream, dbConfig } = tunnel;
return mysql.createConnection({
...dbConfig,
stream
}).then(connection => {
// Store the new connections
connectionCache.ssh = ssh;
connectionCache.dbConnection = connection;
connectionCache.lastUsed = Date.now();
connectionCache.isConnecting = false;
return {
ssh,
connection
};
// Check if there's an available connection in the pool
if (connectionPool.connections.length > 0) {
const conn = connectionPool.connections.pop();
console.log(`Using pooled connection. Pool size: ${connectionPool.connections.length}`);
resolve({
connection: conn.connection,
release: () => releaseConnection(conn)
});
return;
}
// If we haven't reached max connections, create a new one
if (connectionPool.currentConnections < connectionPool.maxConnections) {
try {
console.log(`Creating new connection. Current: ${connectionPool.currentConnections}/${connectionPool.maxConnections}`);
connectionPool.currentConnections++;
const tunnel = await setupSshTunnel();
const { ssh, stream, dbConfig } = tunnel;
const connection = await mysql.createConnection({
...dbConfig,
stream
});
const conn = { ssh, connection, inUse: true, created: Date.now() };
console.log('Database connection established');
// Reset circuit breaker on successful connection
if (connectionPool.circuitBreaker.failures > 0) {
connectionPool.circuitBreaker.failures = 0;
connectionPool.circuitBreaker.isOpen = false;
}
resolve({
connection: conn.connection,
release: () => releaseConnection(conn)
});
} catch (error) {
connectionPool.currentConnections--;
// Track circuit breaker failures
connectionPool.circuitBreaker.failures++;
connectionPool.circuitBreaker.lastFailure = Date.now();
if (connectionPool.circuitBreaker.failures >= connectionPool.circuitBreaker.threshold) {
connectionPool.circuitBreaker.isOpen = true;
console.log(`Circuit breaker opened after ${connectionPool.circuitBreaker.failures} failures`);
}
reject(error);
}
return;
}
// Pool is full, queue the request with timeout
console.log('Connection pool full, queuing request...');
const timeoutId = setTimeout(() => {
// Remove from queue if still there
const index = connectionPool.pendingRequests.findIndex(req => req.resolve === resolve);
if (index !== -1) {
connectionPool.pendingRequests.splice(index, 1);
reject(new Error('Connection pool queue timeout after 15 seconds'));
}
}, 15000);
connectionPool.pendingRequests.push({
resolve,
reject,
timeoutId,
timestamp: Date.now()
});
}).catch(error => {
connectionCache.isConnecting = false;
throw error;
});
}
/**
* Release a connection back to the pool
*/
function releaseConnection(conn) {
conn.inUse = false;
// Wait for the connection to be established
return connectionCache.connectionPromise;
// Check if there are pending requests
if (connectionPool.pendingRequests.length > 0) {
const { resolve, timeoutId } = connectionPool.pendingRequests.shift();
// Clear the timeout since we're serving the request
if (timeoutId) {
clearTimeout(timeoutId);
}
conn.inUse = true;
console.log(`Serving queued request. Queue length: ${connectionPool.pendingRequests.length}`);
resolve({
connection: conn.connection,
release: () => releaseConnection(conn)
});
} else {
// Return to pool
connectionPool.connections.push(conn);
console.log(`Connection returned to pool. Pool size: ${connectionPool.connections.length}, Active: ${connectionPool.currentConnections}`);
}
}
/**
@@ -114,10 +160,10 @@ async function getDbConnection() {
*/
async function getCachedQuery(cacheKey, queryType, queryFn) {
// Get cache duration based on query type
const cacheDuration = connectionCache.cacheDuration[queryType] || connectionCache.cacheDuration.default;
const cacheDuration = connectionPool.cacheDuration[queryType] || connectionPool.cacheDuration.default;
// Check if we have a valid cached result
const cachedResult = connectionCache.queryCache.get(cacheKey);
const cachedResult = connectionPool.queryCache.get(cacheKey);
const now = Date.now();
if (cachedResult && (now - cachedResult.timestamp < cacheDuration)) {
@@ -130,7 +176,7 @@ async function getCachedQuery(cacheKey, queryType, queryFn) {
const result = await queryFn();
// Cache the result
connectionCache.queryCache.set(cacheKey, {
connectionPool.queryCache.set(cacheKey, {
data: result,
timestamp: now
});
@@ -192,10 +238,10 @@ async function setupSshTunnel() {
*/
function clearQueryCache(cacheKey) {
if (cacheKey) {
connectionCache.queryCache.delete(cacheKey);
connectionPool.queryCache.delete(cacheKey);
console.log(`Cleared cache for key: ${cacheKey}`);
} else {
connectionCache.queryCache.clear();
connectionPool.queryCache.clear();
console.log('Cleared all query cache');
}
}
@@ -205,34 +251,47 @@ function clearQueryCache(cacheKey) {
* Useful for server shutdown or manual connection reset
*/
async function closeAllConnections() {
if (connectionCache.dbConnection) {
// Close all pooled connections
for (const conn of connectionPool.connections) {
try {
await connectionCache.dbConnection.end();
console.log('Closed database connection');
await conn.connection.end();
conn.ssh.end();
console.log('Closed pooled connection');
} catch (error) {
console.error('Error closing database connection:', error);
console.error('Error closing pooled connection:', error);
}
connectionCache.dbConnection = null;
}
if (connectionCache.ssh) {
try {
connectionCache.ssh.end();
console.log('Closed SSH connection');
} catch (error) {
console.error('Error closing SSH connection:', error);
}
connectionCache.ssh = null;
}
// Reset pool state
connectionPool.connections = [];
connectionPool.currentConnections = 0;
connectionPool.pendingRequests = [];
connectionPool.queryCache.clear();
connectionCache.lastUsed = 0;
connectionCache.isConnecting = false;
connectionCache.connectionPromise = null;
console.log('All connections closed and pool reset');
}
/**
* Get connection pool status for debugging
*/
function getPoolStatus() {
return {
poolSize: connectionPool.connections.length,
activeConnections: connectionPool.currentConnections,
maxConnections: connectionPool.maxConnections,
pendingRequests: connectionPool.pendingRequests.length,
cacheSize: connectionPool.queryCache.size,
queuedRequests: connectionPool.pendingRequests.map(req => ({
waitTime: Date.now() - req.timestamp,
hasTimeout: !!req.timeoutId
}))
};
}
module.exports = {
getDbConnection,
getCachedQuery,
clearQueryCache,
closeAllConnections
closeAllConnections,
getPoolStatus
};