Restore accidentally removed files, a few additional import/calculation fixes
This commit is contained in:
167
inventory-server/scripts/import/daily-deals.js
Normal file
167
inventory-server/scripts/import/daily-deals.js
Normal file
@@ -0,0 +1,167 @@
|
||||
const { outputProgress, formatElapsedTime } = require('../metrics-new/utils/progress');
|
||||
|
||||
/**
|
||||
* Import daily deals from production MySQL to local PostgreSQL.
|
||||
*
|
||||
* Production has two tables:
|
||||
* - product_daily_deals (deal_id, deal_date, pid, price_id)
|
||||
* - product_current_prices (price_id, pid, price_each, active, ...)
|
||||
*
|
||||
* We join them in the prod query to denormalize the deal price, avoiding
|
||||
* the need to sync the full product_current_prices table.
|
||||
*
|
||||
* On each sync:
|
||||
* 1. Fetch deals from the last 7 days (plus today) from production
|
||||
* 2. Upsert into local table
|
||||
* 3. Hard delete local deals older than 7 days past their deal_date
|
||||
*/
|
||||
async function importDailyDeals(prodConnection, localConnection) {
|
||||
outputProgress({
|
||||
operation: "Starting daily deals import",
|
||||
status: "running",
|
||||
});
|
||||
|
||||
const startTime = Date.now();
|
||||
|
||||
try {
|
||||
await localConnection.query('BEGIN');
|
||||
|
||||
// Fetch recent daily deals from production (MySQL 5.7, no CTEs)
|
||||
// Join product_current_prices to get the actual deal price
|
||||
// Only grab last 7 days + today + tomorrow (for pre-scheduled deals)
|
||||
const [deals] = await prodConnection.query(`
|
||||
SELECT
|
||||
pdd.deal_id,
|
||||
pdd.deal_date,
|
||||
pdd.pid,
|
||||
pdd.price_id,
|
||||
pcp.price_each as deal_price
|
||||
FROM product_daily_deals pdd
|
||||
LEFT JOIN product_current_prices pcp ON pcp.price_id = pdd.price_id
|
||||
WHERE pdd.deal_date >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)
|
||||
AND pdd.deal_date <= DATE_ADD(CURDATE(), INTERVAL 1 DAY)
|
||||
ORDER BY pdd.deal_date DESC, pdd.pid
|
||||
`);
|
||||
|
||||
outputProgress({
|
||||
status: "running",
|
||||
operation: "Daily deals import",
|
||||
message: `Fetched ${deals.length} deals from production`,
|
||||
elapsed: formatElapsedTime(startTime),
|
||||
});
|
||||
|
||||
let totalInserted = 0;
|
||||
let totalUpdated = 0;
|
||||
|
||||
if (deals.length > 0) {
|
||||
// Batch upsert — filter to only PIDs that exist locally
|
||||
const pids = [...new Set(deals.map(d => d.pid))];
|
||||
const existingResult = await localConnection.query(
|
||||
`SELECT pid FROM products WHERE pid = ANY($1)`,
|
||||
[pids]
|
||||
);
|
||||
const existingPids = new Set(
|
||||
(Array.isArray(existingResult) ? existingResult[0] : existingResult)
|
||||
.rows.map(r => Number(r.pid))
|
||||
);
|
||||
|
||||
const validDeals = deals.filter(d => existingPids.has(Number(d.pid)));
|
||||
|
||||
if (validDeals.length > 0) {
|
||||
// Build batch upsert
|
||||
const values = validDeals.flatMap(d => [
|
||||
d.deal_date,
|
||||
d.pid,
|
||||
d.price_id,
|
||||
d.deal_price ?? null,
|
||||
]);
|
||||
|
||||
const placeholders = validDeals
|
||||
.map((_, i) => `($${i * 4 + 1}, $${i * 4 + 2}, $${i * 4 + 3}, $${i * 4 + 4})`)
|
||||
.join(',');
|
||||
|
||||
const upsertQuery = `
|
||||
WITH upserted AS (
|
||||
INSERT INTO product_daily_deals (deal_date, pid, price_id, deal_price)
|
||||
VALUES ${placeholders}
|
||||
ON CONFLICT (deal_date, pid) DO UPDATE SET
|
||||
price_id = EXCLUDED.price_id,
|
||||
deal_price = EXCLUDED.deal_price
|
||||
WHERE
|
||||
product_daily_deals.price_id IS DISTINCT FROM EXCLUDED.price_id OR
|
||||
product_daily_deals.deal_price IS DISTINCT FROM EXCLUDED.deal_price
|
||||
RETURNING
|
||||
CASE WHEN xmax = 0 THEN true ELSE false END as is_insert
|
||||
)
|
||||
SELECT
|
||||
COUNT(*) FILTER (WHERE is_insert) as inserted,
|
||||
COUNT(*) FILTER (WHERE NOT is_insert) as updated
|
||||
FROM upserted
|
||||
`;
|
||||
|
||||
const result = await localConnection.query(upsertQuery, values);
|
||||
const queryResult = Array.isArray(result) ? result[0] : result;
|
||||
totalInserted = parseInt(queryResult.rows[0].inserted) || 0;
|
||||
totalUpdated = parseInt(queryResult.rows[0].updated) || 0;
|
||||
}
|
||||
|
||||
const skipped = deals.length - validDeals.length;
|
||||
if (skipped > 0) {
|
||||
console.log(`Skipped ${skipped} deals (PIDs not in local products table)`);
|
||||
}
|
||||
}
|
||||
|
||||
// Hard delete deals older than 7 days past their deal_date
|
||||
const deleteResult = await localConnection.query(`
|
||||
DELETE FROM product_daily_deals
|
||||
WHERE deal_date < CURRENT_DATE - INTERVAL '7 days'
|
||||
`);
|
||||
const deletedCount = deleteResult.rowCount ??
|
||||
(Array.isArray(deleteResult) ? deleteResult[0]?.rowCount : 0) ?? 0;
|
||||
|
||||
// Update sync status
|
||||
await localConnection.query(`
|
||||
INSERT INTO sync_status (table_name, last_sync_timestamp)
|
||||
VALUES ('product_daily_deals', NOW())
|
||||
ON CONFLICT (table_name) DO UPDATE SET
|
||||
last_sync_timestamp = NOW()
|
||||
`);
|
||||
|
||||
await localConnection.query('COMMIT');
|
||||
|
||||
outputProgress({
|
||||
status: "complete",
|
||||
operation: "Daily deals import completed",
|
||||
message: `Inserted ${totalInserted}, updated ${totalUpdated}, deleted ${deletedCount} expired`,
|
||||
current: totalInserted + totalUpdated,
|
||||
total: totalInserted + totalUpdated,
|
||||
duration: formatElapsedTime(startTime),
|
||||
});
|
||||
|
||||
return {
|
||||
status: "complete",
|
||||
recordsAdded: totalInserted,
|
||||
recordsUpdated: totalUpdated,
|
||||
recordsDeleted: deletedCount,
|
||||
totalRecords: totalInserted + totalUpdated,
|
||||
};
|
||||
} catch (error) {
|
||||
console.error("Error importing daily deals:", error);
|
||||
|
||||
try {
|
||||
await localConnection.query('ROLLBACK');
|
||||
} catch (rollbackError) {
|
||||
console.error("Error during rollback:", rollbackError);
|
||||
}
|
||||
|
||||
outputProgress({
|
||||
status: "error",
|
||||
operation: "Daily deals import failed",
|
||||
error: error.message,
|
||||
});
|
||||
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = importDailyDeals;
|
||||
Reference in New Issue
Block a user