Add newsletter recommendations

inventory-server/scripts/import/daily-deals.js

@@ -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;