Make calculations incremental
This commit is contained in:
@@ -4,9 +4,39 @@ const { getConnection } = require('./utils/db');
|
||||
async function calculateTimeAggregates(startTime, totalProducts, processedCount = 0, isCancelled = false) {
|
||||
const connection = await getConnection();
|
||||
let success = false;
|
||||
let processedOrders = 0;
|
||||
const BATCH_SIZE = 5000;
|
||||
|
||||
try {
|
||||
// Get last calculation timestamp
|
||||
const [lastCalc] = await connection.query(`
|
||||
SELECT last_calculation_timestamp
|
||||
FROM calculate_status
|
||||
WHERE module_name = 'time_aggregates'
|
||||
`);
|
||||
const lastCalculationTime = lastCalc[0]?.last_calculation_timestamp || '1970-01-01';
|
||||
|
||||
// Get total count of products needing updates
|
||||
if (!totalProducts) {
|
||||
const [productCount] = await connection.query(`
|
||||
SELECT COUNT(DISTINCT p.pid) as count
|
||||
FROM products p
|
||||
LEFT JOIN orders o ON p.pid = o.pid AND o.updated > ?
|
||||
WHERE p.updated > ?
|
||||
OR o.pid IS NOT NULL
|
||||
`, [lastCalculationTime, lastCalculationTime]);
|
||||
totalProducts = productCount[0].count;
|
||||
}
|
||||
|
||||
if (totalProducts === 0) {
|
||||
console.log('No products need time aggregate updates');
|
||||
return {
|
||||
processedProducts: 0,
|
||||
processedOrders: 0,
|
||||
processedPurchaseOrders: 0,
|
||||
success: true
|
||||
};
|
||||
}
|
||||
|
||||
if (isCancelled) {
|
||||
outputProgress({
|
||||
status: 'cancelled',
|
||||
@@ -31,14 +61,6 @@ async function calculateTimeAggregates(startTime, totalProducts, processedCount
|
||||
};
|
||||
}
|
||||
|
||||
// Get order count that will be processed
|
||||
const [orderCount] = await connection.query(`
|
||||
SELECT COUNT(*) as count
|
||||
FROM orders o
|
||||
WHERE o.canceled = false
|
||||
`);
|
||||
processedOrders = orderCount[0].count;
|
||||
|
||||
outputProgress({
|
||||
status: 'running',
|
||||
operation: 'Starting time aggregates calculation',
|
||||
@@ -55,180 +77,98 @@ async function calculateTimeAggregates(startTime, totalProducts, processedCount
|
||||
}
|
||||
});
|
||||
|
||||
// Initial insert of time-based aggregates
|
||||
await connection.query(`
|
||||
INSERT INTO product_time_aggregates (
|
||||
pid,
|
||||
year,
|
||||
month,
|
||||
total_quantity_sold,
|
||||
total_revenue,
|
||||
total_cost,
|
||||
order_count,
|
||||
stock_received,
|
||||
stock_ordered,
|
||||
avg_price,
|
||||
profit_margin,
|
||||
inventory_value,
|
||||
gmroi
|
||||
)
|
||||
WITH monthly_sales AS (
|
||||
SELECT
|
||||
o.pid,
|
||||
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 ((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
|
||||
ELSE 0
|
||||
END as profit_margin,
|
||||
p.cost_price * p.stock_quantity as inventory_value,
|
||||
COUNT(DISTINCT DATE(o.date)) as active_days
|
||||
FROM orders o
|
||||
JOIN products p ON o.pid = p.pid
|
||||
WHERE o.canceled = false
|
||||
GROUP BY o.pid, YEAR(o.date), MONTH(o.date)
|
||||
),
|
||||
monthly_stock AS (
|
||||
SELECT
|
||||
// Process in batches
|
||||
let lastPid = 0;
|
||||
while (true) {
|
||||
if (isCancelled) break;
|
||||
|
||||
const [batch] = await connection.query(`
|
||||
SELECT DISTINCT p.pid
|
||||
FROM products p
|
||||
LEFT JOIN orders o ON p.pid = o.pid AND o.updated > ?
|
||||
WHERE p.pid > ?
|
||||
AND (
|
||||
p.updated > ?
|
||||
OR o.id IS NOT NULL
|
||||
)
|
||||
ORDER BY p.pid
|
||||
LIMIT ?
|
||||
`, [lastCalculationTime, lastPid, lastCalculationTime, BATCH_SIZE]);
|
||||
|
||||
if (batch.length === 0) break;
|
||||
|
||||
// Calculate and update time aggregates for this batch
|
||||
await connection.query(`
|
||||
INSERT INTO product_time_aggregates (
|
||||
pid,
|
||||
YEAR(date) as year,
|
||||
MONTH(date) as month,
|
||||
SUM(received) as stock_received,
|
||||
SUM(ordered) as stock_ordered
|
||||
FROM purchase_orders
|
||||
GROUP BY pid, YEAR(date), MONTH(date)
|
||||
)
|
||||
SELECT
|
||||
s.pid,
|
||||
s.year,
|
||||
s.month,
|
||||
s.total_quantity_sold,
|
||||
s.total_revenue,
|
||||
s.total_cost,
|
||||
s.order_count,
|
||||
COALESCE(ms.stock_received, 0) as stock_received,
|
||||
COALESCE(ms.stock_ordered, 0) as stock_ordered,
|
||||
s.avg_price,
|
||||
s.profit_margin,
|
||||
s.inventory_value,
|
||||
CASE
|
||||
WHEN s.inventory_value > 0 THEN
|
||||
(s.total_revenue - s.total_cost) / s.inventory_value
|
||||
ELSE 0
|
||||
END as gmroi
|
||||
FROM monthly_sales s
|
||||
LEFT JOIN monthly_stock ms
|
||||
ON s.pid = ms.pid
|
||||
AND s.year = ms.year
|
||||
AND s.month = ms.month
|
||||
UNION
|
||||
SELECT
|
||||
p.pid,
|
||||
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,
|
||||
(SELECT cost_price * stock_quantity FROM products WHERE pid = p.pid) as inventory_value,
|
||||
0 as gmroi
|
||||
FROM monthly_stock p
|
||||
LEFT JOIN monthly_sales s
|
||||
ON p.pid = s.pid
|
||||
AND p.year = s.year
|
||||
AND p.month = s.month
|
||||
WHERE s.pid IS NULL
|
||||
ON DUPLICATE KEY UPDATE
|
||||
total_quantity_sold = VALUES(total_quantity_sold),
|
||||
total_revenue = VALUES(total_revenue),
|
||||
total_cost = VALUES(total_cost),
|
||||
order_count = VALUES(order_count),
|
||||
stock_received = VALUES(stock_received),
|
||||
stock_ordered = VALUES(stock_ordered),
|
||||
avg_price = VALUES(avg_price),
|
||||
profit_margin = VALUES(profit_margin),
|
||||
inventory_value = VALUES(inventory_value),
|
||||
gmroi = VALUES(gmroi)
|
||||
`);
|
||||
|
||||
processedCount = Math.floor(totalProducts * 0.60);
|
||||
outputProgress({
|
||||
status: 'running',
|
||||
operation: 'Base time aggregates calculated, updating financial metrics',
|
||||
current: processedCount,
|
||||
total: totalProducts,
|
||||
elapsed: formatElapsedTime(startTime),
|
||||
remaining: estimateRemaining(startTime, processedCount, totalProducts),
|
||||
rate: calculateRate(startTime, processedCount),
|
||||
percentage: ((processedCount / totalProducts) * 100).toFixed(1),
|
||||
timing: {
|
||||
start_time: new Date(startTime).toISOString(),
|
||||
end_time: new Date().toISOString(),
|
||||
elapsed_seconds: Math.round((Date.now() - startTime) / 1000)
|
||||
}
|
||||
});
|
||||
|
||||
if (isCancelled) return {
|
||||
processedProducts: processedCount,
|
||||
processedOrders,
|
||||
processedPurchaseOrders: 0,
|
||||
success
|
||||
};
|
||||
|
||||
// Update with financial metrics
|
||||
await connection.query(`
|
||||
UPDATE product_time_aggregates pta
|
||||
JOIN (
|
||||
year,
|
||||
month,
|
||||
total_quantity_sold,
|
||||
total_revenue,
|
||||
total_cost,
|
||||
order_count,
|
||||
avg_price,
|
||||
profit_margin,
|
||||
inventory_value,
|
||||
gmroi
|
||||
)
|
||||
SELECT
|
||||
p.pid,
|
||||
YEAR(o.date) as year,
|
||||
MONTH(o.date) as month,
|
||||
SUM(o.quantity) as total_quantity_sold,
|
||||
SUM(o.quantity * o.price) as total_revenue,
|
||||
SUM(o.quantity * p.cost_price) as total_cost,
|
||||
COUNT(DISTINCT o.order_number) as order_count,
|
||||
AVG(o.price) as avg_price,
|
||||
CASE
|
||||
WHEN SUM(o.quantity * o.price) > 0
|
||||
THEN ((SUM(o.quantity * o.price) - SUM(o.quantity * p.cost_price)) / SUM(o.quantity * o.price)) * 100
|
||||
ELSE 0
|
||||
END as profit_margin,
|
||||
p.cost_price * p.stock_quantity as inventory_value,
|
||||
SUM(o.quantity * (o.price - p.cost_price)) as gross_profit,
|
||||
COUNT(DISTINCT DATE(o.date)) as days_in_period
|
||||
CASE
|
||||
WHEN p.cost_price * p.stock_quantity > 0
|
||||
THEN (SUM(o.quantity * (o.price - p.cost_price))) / (p.cost_price * p.stock_quantity)
|
||||
ELSE 0
|
||||
END as gmroi
|
||||
FROM products p
|
||||
LEFT JOIN orders o ON p.pid = o.pid
|
||||
WHERE o.canceled = false
|
||||
INNER JOIN orders o ON p.pid = o.pid
|
||||
AND o.canceled = false
|
||||
AND o.date >= DATE_SUB(CURRENT_DATE, INTERVAL 12 MONTH)
|
||||
WHERE p.pid IN (?)
|
||||
GROUP BY p.pid, YEAR(o.date), MONTH(o.date)
|
||||
) fin ON pta.pid = fin.pid
|
||||
AND pta.year = fin.year
|
||||
AND pta.month = fin.month
|
||||
SET
|
||||
pta.inventory_value = COALESCE(fin.inventory_value, 0),
|
||||
pta.gmroi = CASE
|
||||
WHEN COALESCE(fin.inventory_value, 0) > 0 AND fin.days_in_period > 0 THEN
|
||||
(COALESCE(fin.gross_profit, 0) * (365.0 / fin.days_in_period)) / COALESCE(fin.inventory_value, 0)
|
||||
ELSE 0
|
||||
END
|
||||
`);
|
||||
HAVING year IS NOT NULL AND month IS NOT NULL
|
||||
ON DUPLICATE KEY UPDATE
|
||||
total_quantity_sold = VALUES(total_quantity_sold),
|
||||
total_revenue = VALUES(total_revenue),
|
||||
total_cost = VALUES(total_cost),
|
||||
order_count = VALUES(order_count),
|
||||
avg_price = VALUES(avg_price),
|
||||
profit_margin = VALUES(profit_margin),
|
||||
inventory_value = VALUES(inventory_value),
|
||||
gmroi = VALUES(gmroi)
|
||||
`, [batch.map(row => row.pid)]);
|
||||
|
||||
processedCount = Math.floor(totalProducts * 0.65);
|
||||
outputProgress({
|
||||
status: 'running',
|
||||
operation: 'Financial metrics updated',
|
||||
current: processedCount,
|
||||
total: totalProducts,
|
||||
elapsed: formatElapsedTime(startTime),
|
||||
remaining: estimateRemaining(startTime, processedCount, totalProducts),
|
||||
rate: calculateRate(startTime, processedCount),
|
||||
percentage: ((processedCount / totalProducts) * 100).toFixed(1),
|
||||
timing: {
|
||||
start_time: new Date(startTime).toISOString(),
|
||||
end_time: new Date().toISOString(),
|
||||
elapsed_seconds: Math.round((Date.now() - startTime) / 1000)
|
||||
}
|
||||
});
|
||||
lastPid = batch[batch.length - 1].pid;
|
||||
processedCount += batch.length;
|
||||
|
||||
outputProgress({
|
||||
status: 'running',
|
||||
operation: 'Processing time aggregates batch',
|
||||
current: processedCount,
|
||||
total: totalProducts,
|
||||
elapsed: formatElapsedTime(startTime),
|
||||
remaining: estimateRemaining(startTime, processedCount, totalProducts),
|
||||
rate: calculateRate(startTime, processedCount),
|
||||
percentage: ((processedCount / totalProducts) * 100).toFixed(1),
|
||||
timing: {
|
||||
start_time: new Date(startTime).toISOString(),
|
||||
end_time: new Date().toISOString(),
|
||||
elapsed_seconds: Math.round((Date.now() - startTime) / 1000)
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// If we get here, everything completed successfully
|
||||
success = true;
|
||||
@@ -242,7 +182,7 @@ async function calculateTimeAggregates(startTime, totalProducts, processedCount
|
||||
|
||||
return {
|
||||
processedProducts: processedCount,
|
||||
processedOrders,
|
||||
processedOrders: 0,
|
||||
processedPurchaseOrders: 0,
|
||||
success
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user