338 lines
12 KiB
JavaScript
338 lines
12 KiB
JavaScript
const { outputProgress, formatElapsedTime, estimateRemaining, calculateRate } = require('../metrics/utils/progress');
|
|
const { importMissingProducts } = require('./products');
|
|
|
|
/**
|
|
* Imports orders from a production MySQL database to a local MySQL database.
|
|
* It can run in two modes:
|
|
* 1. Incremental update mode (default): Only fetch orders that have changed since the last sync time.
|
|
* 2. Full update mode: Fetch all eligible orders within the last 5 years regardless of timestamp.
|
|
*
|
|
* @param {object} prodConnection - A MySQL connection to production DB (MySQL 5.7).
|
|
* @param {object} localConnection - A MySQL connection to local DB (MySQL 8.0).
|
|
* @param {boolean} incrementalUpdate - Set to false for a full sync; true for incremental.
|
|
*
|
|
* @returns {object} Information about the sync operation.
|
|
*/
|
|
async function importOrders(prodConnection, localConnection, incrementalUpdate = true) {
|
|
const startTime = Date.now();
|
|
const skippedOrders = new Set();
|
|
const missingProducts = new Set();
|
|
|
|
try {
|
|
// Get the last sync time
|
|
const [syncInfo] = await localConnection.query(
|
|
"SELECT last_sync_timestamp FROM sync_status WHERE table_name = 'orders'"
|
|
);
|
|
const lastSyncTime = syncInfo?.[0]?.last_sync_timestamp || '1970-01-01';
|
|
|
|
// Retrieve column names for the 'orders' table, skip 'id' since it's auto-increment
|
|
const [columns] = await localConnection.query(`
|
|
SELECT COLUMN_NAME
|
|
FROM INFORMATION_SCHEMA.COLUMNS
|
|
WHERE TABLE_NAME = 'orders'
|
|
ORDER BY ORDINAL_POSITION
|
|
`);
|
|
const columnNames = columns
|
|
.map(col => col.COLUMN_NAME)
|
|
.filter(name => name !== "id");
|
|
|
|
// Build query clauses for incremental vs. full update
|
|
const incrementalWhereClause = incrementalUpdate
|
|
? `AND (
|
|
o.stamp > ?
|
|
OR o.date_modified > ?
|
|
OR o.date_placed > ?
|
|
OR o.date_shipped > ?
|
|
OR oi.stamp > ?
|
|
)`
|
|
: "";
|
|
const incrementalParams = incrementalUpdate
|
|
? [lastSyncTime, lastSyncTime, lastSyncTime, lastSyncTime, lastSyncTime]
|
|
: [];
|
|
|
|
// Count how many orders we need to process
|
|
const [countResult] = await prodConnection.query(
|
|
`
|
|
SELECT COUNT(*) AS total
|
|
FROM order_items oi USE INDEX (PRIMARY)
|
|
JOIN _order o USE INDEX (PRIMARY)
|
|
ON oi.order_id = o.order_id
|
|
WHERE o.order_status >= 15
|
|
AND o.date_placed_onlydate >= DATE_SUB(CURRENT_DATE, INTERVAL 5 YEAR)
|
|
${incrementalWhereClause}
|
|
`,
|
|
incrementalParams
|
|
);
|
|
|
|
const total = countResult[0].total;
|
|
outputProgress({
|
|
operation: `Starting ${incrementalUpdate ? 'incremental' : 'full'} orders import - Fetching ${total} orders`,
|
|
status: "running",
|
|
});
|
|
|
|
let processed = 0;
|
|
// Increase or decrease this if you find a more optimal size
|
|
const batchSize = 20000;
|
|
let offset = 0;
|
|
|
|
// Process in batches for memory efficiency
|
|
while (offset < total) {
|
|
// Fetch orders (initially with tax set to 0, to be updated later)
|
|
const [orders] = await prodConnection.query(
|
|
`
|
|
SELECT
|
|
oi.order_id AS order_number,
|
|
oi.prod_pid AS pid,
|
|
oi.prod_itemnumber AS SKU,
|
|
o.date_placed_onlydate AS date,
|
|
oi.prod_price_reg AS price,
|
|
oi.qty_ordered AS quantity,
|
|
(oi.prod_price_reg - oi.prod_price) AS discount,
|
|
0 AS tax,
|
|
0 AS tax_included,
|
|
ROUND(
|
|
(
|
|
(o.summary_shipping - COALESCE(o.summary_discount_shipping, 0))
|
|
* (oi.prod_price * oi.qty_ordered)
|
|
) / NULLIF(o.summary_subtotal, 0),
|
|
2
|
|
) AS shipping,
|
|
o.order_cid AS customer,
|
|
CONCAT(o.bill_firstname, ' ', o.bill_lastname) AS customer_name,
|
|
'pending' AS status,
|
|
CASE WHEN o.order_status = 15 THEN 1 ELSE 0 END AS canceled
|
|
FROM order_items oi
|
|
FORCE INDEX (PRIMARY)
|
|
JOIN _order o
|
|
ON oi.order_id = o.order_id
|
|
WHERE o.order_status >= 15
|
|
AND o.date_placed_onlydate >= DATE_SUB(CURRENT_DATE, INTERVAL 5 YEAR)
|
|
${incrementalWhereClause}
|
|
LIMIT ? OFFSET ?
|
|
`,
|
|
[...incrementalParams, batchSize, offset]
|
|
);
|
|
|
|
// Fetch the latest tax info for these orders
|
|
if (orders.length > 0) {
|
|
const orderIds = [...new Set(orders.map(o => o.order_number))];
|
|
const [taxInfo] = await prodConnection.query(`
|
|
SELECT oti.order_id, otp.pid, otp.item_taxes_to_collect
|
|
FROM (
|
|
SELECT order_id, MAX(stamp) AS latest_stamp
|
|
FROM order_tax_info
|
|
WHERE order_id IN (?)
|
|
GROUP BY order_id
|
|
) latest
|
|
JOIN order_tax_info oti
|
|
ON oti.order_id = latest.order_id
|
|
AND oti.stamp = latest.latest_stamp
|
|
JOIN order_tax_info_products otp
|
|
ON oti.taxinfo_id = otp.taxinfo_id
|
|
`, [orderIds]);
|
|
|
|
// Map (order_id-pid) -> tax amount
|
|
const taxMap = new Map();
|
|
taxInfo.forEach(t => {
|
|
taxMap.set(`${t.order_id}-${t.pid}`, t.item_taxes_to_collect);
|
|
});
|
|
|
|
// Merge tax into the orders array
|
|
orders.forEach(order => {
|
|
const key = `${order.order_number}-${order.pid}`;
|
|
if (taxMap.has(key)) {
|
|
order.tax = taxMap.get(key) || 0;
|
|
}
|
|
});
|
|
}
|
|
|
|
// Check local DB for existing products to ensure we don't insert orders for missing products
|
|
const orderProductPids = [...new Set(orders.map(o => o.pid))];
|
|
const [existingProducts] = await localConnection.query(
|
|
"SELECT pid FROM products WHERE pid IN (?)",
|
|
[orderProductPids]
|
|
);
|
|
const existingPids = new Set(existingProducts.map(p => p.pid));
|
|
|
|
// Separate valid orders from those referencing missing products
|
|
const validOrders = [];
|
|
for (const order of orders) {
|
|
if (!existingPids.has(order.pid)) {
|
|
missingProducts.add(order.pid);
|
|
skippedOrders.add(order.order_number);
|
|
} else {
|
|
validOrders.push(order);
|
|
}
|
|
}
|
|
|
|
// Bulk insert valid orders
|
|
if (validOrders.length > 0) {
|
|
const placeholders = validOrders
|
|
.map(() => `(${Array(columnNames.length).fill("?").join(",")})`)
|
|
.join(",");
|
|
const updateClauses = columnNames
|
|
.filter(col => col !== "order_number") // don't overwrite primary key
|
|
.map(col => `${col} = VALUES(${col})`)
|
|
.join(",");
|
|
|
|
const upsertQuery = `
|
|
INSERT INTO orders (${columnNames.join(",")})
|
|
VALUES ${placeholders}
|
|
ON DUPLICATE KEY UPDATE ${updateClauses}
|
|
`;
|
|
|
|
await localConnection.query(
|
|
upsertQuery,
|
|
validOrders.flatMap(order => columnNames.map(col => order[col]))
|
|
);
|
|
}
|
|
|
|
processed += orders.length;
|
|
offset += batchSize;
|
|
|
|
outputProgress({
|
|
status: "running",
|
|
operation: "Orders import",
|
|
current: processed,
|
|
total,
|
|
elapsed: formatElapsedTime((Date.now() - startTime) / 1000),
|
|
remaining: estimateRemaining(startTime, processed, total),
|
|
rate: calculateRate(startTime, processed)
|
|
});
|
|
}
|
|
|
|
// If we found missing products, import them and retry the skipped orders
|
|
if (missingProducts.size > 0) {
|
|
outputProgress({
|
|
operation: `Found ${missingProducts.size} missing products, importing them now`,
|
|
status: "running",
|
|
});
|
|
|
|
// Import missing products
|
|
await importMissingProducts(prodConnection, localConnection, [...missingProducts]);
|
|
|
|
// Retry orders that were skipped due to missing products
|
|
if (skippedOrders.size > 0) {
|
|
outputProgress({
|
|
operation: `Retrying ${skippedOrders.size} skipped orders`,
|
|
status: "running",
|
|
});
|
|
|
|
const [retryOrders] = await prodConnection.query(`
|
|
SELECT
|
|
oi.order_id AS order_number,
|
|
oi.prod_pid AS pid,
|
|
oi.prod_itemnumber AS SKU,
|
|
o.date_placed_onlydate AS date,
|
|
oi.prod_price_reg AS price,
|
|
oi.qty_ordered AS quantity,
|
|
(oi.prod_price_reg - oi.prod_price) AS discount,
|
|
0 AS tax,
|
|
0 AS tax_included,
|
|
ROUND(
|
|
(
|
|
(o.summary_shipping - COALESCE(o.summary_discount_shipping, 0))
|
|
* (oi.prod_price * oi.qty_ordered)
|
|
) / NULLIF(o.summary_subtotal, 0),
|
|
2
|
|
) AS shipping,
|
|
o.order_cid AS customer,
|
|
CONCAT(o.bill_firstname, ' ', o.bill_lastname) AS customer_name,
|
|
'pending' AS status,
|
|
CASE WHEN o.order_status = 15 THEN 1 ELSE 0 END AS canceled
|
|
FROM order_items oi
|
|
JOIN _order o ON oi.order_id = o.order_id
|
|
WHERE oi.order_id IN (?)
|
|
`, [[...skippedOrders]]);
|
|
|
|
if (retryOrders.length > 0) {
|
|
// Fetch tax data for these specific retry orders
|
|
const retryOrderIds = [...new Set(retryOrders.map(o => o.order_number))];
|
|
const [retryTaxInfo] = await prodConnection.query(`
|
|
SELECT oti.order_id, otp.pid, otp.item_taxes_to_collect
|
|
FROM (
|
|
SELECT order_id, MAX(stamp) AS latest_stamp
|
|
FROM order_tax_info
|
|
WHERE order_id IN (?)
|
|
GROUP BY order_id
|
|
) latest
|
|
JOIN order_tax_info oti
|
|
ON oti.order_id = latest.order_id
|
|
AND oti.stamp = latest.latest_stamp
|
|
JOIN order_tax_info_products otp
|
|
ON oti.taxinfo_id = otp.taxinfo_id
|
|
`, [retryOrderIds]);
|
|
|
|
const taxMap = new Map();
|
|
retryTaxInfo.forEach(t => {
|
|
taxMap.set(`${t.order_id}-${t.pid}`, t.item_taxes_to_collect);
|
|
});
|
|
|
|
retryOrders.forEach(order => {
|
|
const key = `${order.order_number}-${order.pid}`;
|
|
if (taxMap.has(key)) {
|
|
order.tax = taxMap.get(key) || 0;
|
|
}
|
|
});
|
|
|
|
const placeholders = retryOrders
|
|
.map(() => `(${Array(columnNames.length).fill("?").join(",")})`)
|
|
.join(",");
|
|
const updateClauses = columnNames
|
|
.filter(col => col !== "order_number")
|
|
.map(col => `${col} = VALUES(${col})`)
|
|
.join(",");
|
|
|
|
const upsertQuery = `
|
|
INSERT INTO orders (${columnNames.join(",")})
|
|
VALUES ${placeholders}
|
|
ON DUPLICATE KEY UPDATE ${updateClauses}
|
|
`;
|
|
|
|
await localConnection.query(
|
|
upsertQuery,
|
|
retryOrders.flatMap(order => columnNames.map(col => order[col]))
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Update the sync timestamp
|
|
await localConnection.query(`
|
|
INSERT INTO sync_status (table_name, last_sync_timestamp)
|
|
VALUES ('orders', NOW())
|
|
ON DUPLICATE KEY UPDATE
|
|
last_sync_timestamp = NOW(),
|
|
last_sync_id = LAST_INSERT_ID(last_sync_id)
|
|
`);
|
|
|
|
const endTime = Date.now();
|
|
|
|
outputProgress({
|
|
status: "complete",
|
|
operation: `${incrementalUpdate ? 'Incremental' : 'Full'} orders import completed`,
|
|
current: total,
|
|
total,
|
|
duration: formatElapsedTime((endTime - startTime) / 1000),
|
|
});
|
|
|
|
return {
|
|
status: "complete",
|
|
totalImported: total,
|
|
missingProducts: missingProducts.size,
|
|
retriedOrders: skippedOrders.size,
|
|
incrementalUpdate,
|
|
lastSyncTime
|
|
};
|
|
} catch (error) {
|
|
outputProgress({
|
|
operation: `${incrementalUpdate ? 'Incremental' : 'Full'} orders import failed`,
|
|
status: "error",
|
|
error: error.message,
|
|
});
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
module.exports = importOrders;
|