const mysql = require('mysql2/promise'); const path = require('path'); require('dotenv').config({ path: path.resolve(__dirname, '..', '.env') }); // Helper function to format elapsed time function formatElapsedTime(startTime) { const elapsed = Date.now() - startTime; const seconds = Math.floor(elapsed / 1000); const minutes = Math.floor(seconds / 60); const hours = Math.floor(minutes / 60); if (hours > 0) { return `${hours}h ${minutes % 60}m`; } else if (minutes > 0) { return `${minutes}m ${seconds % 60}s`; } else { return `${seconds}s`; } } // Helper function to estimate remaining time function estimateRemaining(startTime, current, total) { if (current === 0) return null; const elapsed = Date.now() - startTime; const rate = current / elapsed; const remaining = (total - current) / rate; const minutes = Math.floor(remaining / 60000); const seconds = Math.floor((remaining % 60000) / 1000); if (minutes > 0) { return `${minutes}m ${seconds}s`; } else { return `${seconds}s`; } } // Helper function to calculate rate function calculateRate(startTime, current) { const elapsed = (Date.now() - startTime) / 1000; // Convert to seconds return elapsed > 0 ? Math.round(current / elapsed) : 0; } // Helper function to output progress function outputProgress(data) { process.stdout.write(JSON.stringify(data) + '\n'); } // Helper function to log errors function logError(error, context) { console.error(JSON.stringify({ status: 'error', error: error.message || error, context })); } // Database configuration const dbConfig = { host: process.env.DB_HOST, user: process.env.DB_USER, password: process.env.DB_PASSWORD, database: process.env.DB_NAME, waitForConnections: true, connectionLimit: 10, queueLimit: 0 }; // Add cancel handler let isCancelled = false; function cancelCalculation() { isCancelled = true; process.stdout.write(JSON.stringify({ status: 'cancelled', operation: 'Calculation cancelled', current: 0, total: 0, elapsed: null, remaining: null, rate: 0 }) + '\n'); process.exit(0); } async function calculateMetrics() { let pool; const startTime = Date.now(); let processedCount = 0; let totalProducts = 0; // Initialize at the top try { isCancelled = false; pool = mysql.createPool(dbConfig); const connection = await pool.getConnection(); try { // Get total number of products const [countResult] = await connection.query('SELECT COUNT(*) as total FROM products'); totalProducts = countResult[0].total; // Initial progress outputProgress({ status: 'running', operation: 'Processing products', current: processedCount, total: totalProducts, elapsed: '0s', remaining: 'Calculating...', rate: 0 }); // Process in batches of 100 const batchSize = 100; for (let offset = 0; offset < totalProducts; offset += batchSize) { if (isCancelled) { throw new Error('Operation cancelled'); } const [products] = await connection.query('SELECT product_id FROM products LIMIT ? OFFSET ?', [batchSize, offset]); processedCount += products.length; // Update progress after each batch outputProgress({ status: 'running', operation: 'Processing products', current: processedCount, total: totalProducts, elapsed: formatElapsedTime(startTime), remaining: estimateRemaining(startTime, processedCount, totalProducts), rate: calculateRate(startTime, processedCount) }); // Process the batch for (const product of products) { // Calculate sales metrics const [salesMetrics] = await connection.query(` SELECT SUM(o.quantity) as total_quantity_sold, SUM((o.price - COALESCE(o.discount, 0)) * o.quantity) as total_revenue, SUM(COALESCE(p.cost_price, 0) * o.quantity) as total_cost, MAX(o.date) as last_sale_date FROM orders o JOIN products p ON o.product_id = p.product_id WHERE o.canceled = 0 AND o.product_id = ? GROUP BY o.product_id `, [product.product_id]); // Calculate purchase metrics const [purchaseMetrics] = await connection.query(` SELECT SUM(received) as total_quantity_purchased, SUM(cost_price * received) as total_cost, MAX(date) as last_purchase_date, MAX(received_date) as last_received_date, AVG(DATEDIFF(received_date, date)) as avg_lead_time_days FROM purchase_orders WHERE status = 'closed' AND received > 0 AND product_id = ? GROUP BY product_id `, [product.product_id]); // Get current stock const [stockInfo] = await connection.query(` SELECT stock_quantity, cost_price FROM products WHERE product_id = ? `, [product.product_id]); // Calculate metrics const metrics = salesMetrics[0] || {}; const purchases = purchaseMetrics[0] || {}; const stock = stockInfo[0] || {}; const daily_sales_avg = metrics.total_quantity_sold ? metrics.total_quantity_sold / 30 : 0; const weekly_sales_avg = metrics.total_quantity_sold ? metrics.total_quantity_sold / 4 : 0; const monthly_sales_avg = metrics.total_quantity_sold || 0; // Update product metrics await connection.query(` INSERT INTO product_metrics ( product_id, last_calculated_at, daily_sales_avg, weekly_sales_avg, monthly_sales_avg, days_of_inventory, weeks_of_inventory, reorder_point, safety_stock, avg_margin_percent, total_revenue, avg_lead_time_days, last_purchase_date, last_received_date, abc_class, stock_status ) VALUES ( ?, NOW(), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ? ) ON DUPLICATE KEY UPDATE last_calculated_at = VALUES(last_calculated_at), daily_sales_avg = VALUES(daily_sales_avg), weekly_sales_avg = VALUES(weekly_sales_avg), monthly_sales_avg = VALUES(monthly_sales_avg), days_of_inventory = VALUES(days_of_inventory), weeks_of_inventory = VALUES(weeks_of_inventory), reorder_point = VALUES(reorder_point), safety_stock = VALUES(safety_stock), avg_margin_percent = VALUES(avg_margin_percent), total_revenue = VALUES(total_revenue), avg_lead_time_days = VALUES(avg_lead_time_days), last_purchase_date = VALUES(last_purchase_date), last_received_date = VALUES(last_received_date), stock_status = VALUES(stock_status) `, [ product.product_id, daily_sales_avg, weekly_sales_avg, monthly_sales_avg, daily_sales_avg ? stock.stock_quantity / daily_sales_avg : null, weekly_sales_avg ? stock.stock_quantity / weekly_sales_avg : null, Math.ceil(daily_sales_avg * 14), // 14 days reorder point Math.ceil(daily_sales_avg * 7), // 7 days safety stock metrics.total_revenue ? ((metrics.total_revenue - metrics.total_cost) / metrics.total_revenue) * 100 : 0, metrics.total_revenue || 0, purchases.avg_lead_time_days || 0, purchases.last_purchase_date, purchases.last_received_date, daily_sales_avg === 0 ? 'New' : stock.stock_quantity <= Math.ceil(daily_sales_avg * 7) ? 'Critical' : stock.stock_quantity <= Math.ceil(daily_sales_avg * 14) ? 'Reorder' : stock.stock_quantity > (daily_sales_avg * 90) ? 'Overstocked' : 'Healthy' ]); } } // Update progress for ABC classification outputProgress({ status: 'running', operation: 'Calculating ABC classification', current: totalProducts, total: totalProducts, elapsed: formatElapsedTime(startTime), remaining: estimateRemaining(startTime, totalProducts, totalProducts), rate: calculateRate(startTime, totalProducts) }); // Calculate ABC classification await connection.query(` WITH revenue_percentiles AS ( SELECT product_id, total_revenue, PERCENT_RANK() OVER (ORDER BY total_revenue DESC) as revenue_percentile FROM product_metrics WHERE total_revenue > 0 ) UPDATE product_metrics pm JOIN revenue_percentiles rp ON pm.product_id = rp.product_id SET pm.abc_class = CASE WHEN rp.revenue_percentile < 0.2 THEN 'A' WHEN rp.revenue_percentile < 0.5 THEN 'B' ELSE 'C' END; `); // Update progress for time-based aggregates outputProgress({ status: 'running', operation: 'Calculating time-based aggregates', current: totalProducts, total: totalProducts, elapsed: formatElapsedTime(startTime), remaining: estimateRemaining(startTime, totalProducts, totalProducts), rate: calculateRate(startTime, totalProducts) }); // Calculate time-based aggregates await connection.query('TRUNCATE TABLE product_time_aggregates;'); await connection.query(` INSERT INTO product_time_aggregates ( product_id, year, month, total_quantity_sold, total_revenue, total_cost, order_count, stock_received, stock_ordered, avg_price, profit_margin ) WITH sales_data AS ( SELECT o.product_id, YEAR(o.date) as year, MONTH(o.date) as month, SUM(o.quantity) as total_quantity_sold, SUM((o.price - COALESCE(o.discount, 0)) * o.quantity) as total_revenue, SUM(COALESCE(p.cost_price, 0) * o.quantity) as total_cost, COUNT(DISTINCT o.order_number) as order_count, AVG(o.price - COALESCE(o.discount, 0)) as avg_price, CASE WHEN SUM((o.price - COALESCE(o.discount, 0)) * o.quantity) = 0 THEN 0 ELSE ((SUM((o.price - COALESCE(o.discount, 0)) * o.quantity) - SUM(COALESCE(p.cost_price, 0) * o.quantity)) / SUM((o.price - COALESCE(o.discount, 0)) * o.quantity)) * 100 END as profit_margin FROM orders o JOIN products p ON o.product_id = p.product_id WHERE o.canceled = 0 GROUP BY o.product_id, YEAR(o.date), MONTH(o.date) ), purchase_data AS ( SELECT product_id, YEAR(date) as year, MONTH(date) as month, SUM(received) as stock_received, SUM(ordered) as stock_ordered FROM purchase_orders WHERE status = 'closed' GROUP BY product_id, YEAR(date), MONTH(date) ) SELECT s.product_id, s.year, s.month, s.total_quantity_sold, s.total_revenue, s.total_cost, s.order_count, COALESCE(p.stock_received, 0) as stock_received, COALESCE(p.stock_ordered, 0) as stock_ordered, s.avg_price, s.profit_margin FROM sales_data s LEFT JOIN purchase_data p ON s.product_id = p.product_id AND s.year = p.year AND s.month = p.month UNION SELECT p.product_id, p.year, p.month, 0 as total_quantity_sold, 0 as total_revenue, 0 as total_cost, 0 as order_count, p.stock_received, p.stock_ordered, 0 as avg_price, 0 as profit_margin FROM purchase_data p LEFT JOIN sales_data s ON p.product_id = s.product_id AND p.year = s.year AND p.month = s.month WHERE s.product_id IS NULL `); // Update progress for vendor metrics outputProgress({ status: 'running', operation: 'Calculating vendor metrics', current: totalProducts, total: totalProducts, elapsed: formatElapsedTime(startTime), remaining: estimateRemaining(startTime, totalProducts, totalProducts), rate: calculateRate(startTime, totalProducts) }); // Calculate vendor metrics await connection.query(` INSERT INTO vendor_metrics ( vendor, last_calculated_at, avg_lead_time_days, on_time_delivery_rate, order_fill_rate, total_orders, total_late_orders ) SELECT vendor, NOW() as last_calculated_at, COALESCE(AVG(DATEDIFF(received_date, date)), 0) as avg_lead_time_days, COALESCE((COUNT(CASE WHEN DATEDIFF(received_date, date) <= 14 THEN 1 END) * 100.0 / NULLIF(COUNT(*), 0)), 0) as on_time_delivery_rate, COALESCE((SUM(received) * 100.0 / NULLIF(SUM(ordered), 0)), 0) as order_fill_rate, COUNT(DISTINCT po_id) as total_orders, COUNT(CASE WHEN DATEDIFF(received_date, date) > 14 THEN 1 END) as total_late_orders FROM purchase_orders WHERE status = 'closed' GROUP BY vendor ON DUPLICATE KEY UPDATE last_calculated_at = VALUES(last_calculated_at), avg_lead_time_days = VALUES(avg_lead_time_days), on_time_delivery_rate = VALUES(on_time_delivery_rate), order_fill_rate = VALUES(order_fill_rate), total_orders = VALUES(total_orders), total_late_orders = VALUES(total_late_orders) `); // Final success message outputProgress({ status: 'complete', operation: 'Metrics calculation complete', current: totalProducts, total: totalProducts, elapsed: formatElapsedTime(startTime), remaining: '0s', rate: calculateRate(startTime, totalProducts) }); } catch (error) { if (isCancelled) { outputProgress({ status: 'cancelled', operation: 'Calculation cancelled', current: processedCount, total: totalProducts || 0, // Use 0 if not yet defined elapsed: formatElapsedTime(startTime), remaining: null, rate: calculateRate(startTime, processedCount) }); } else { outputProgress({ status: 'error', operation: 'Error: ' + error.message, current: processedCount, total: totalProducts || 0, // Use 0 if not yet defined elapsed: formatElapsedTime(startTime), remaining: null, rate: calculateRate(startTime, processedCount) }); } throw error; } finally { connection.release(); } } finally { if (pool) { await pool.end(); } } } // Export both functions module.exports = calculateMetrics; module.exports.cancelCalculation = cancelCalculation; // Run directly if called from command line if (require.main === module) { calculateMetrics().catch(error => { if (!error.message.includes('Operation cancelled')) { console.error('Error:', error); } process.exit(1); }); }