Cleanup

.gitignore

@@ -57,3 +57,4 @@ csv/**/*
 **/csv/**/*
 !csv/.gitkeep
 inventory/tsconfig.tsbuildinfo
+inventory-server/scripts/.fuse_hidden00000fa20000000a

inventory-server/scripts/import-from-prod.js

@@ -10,9 +10,9 @@ const importPurchaseOrders = require('./import/purchase-orders');
 dotenv.config({ path: path.join(__dirname, "../.env") });
 
 // Constants to control which imports run
-const IMPORT_CATEGORIES = false;
+const IMPORT_CATEGORIES = true;
-const IMPORT_PRODUCTS = false;
+const IMPORT_PRODUCTS = true;
-const IMPORT_ORDERS = false;
+const IMPORT_ORDERS = true;
 const IMPORT_PURCHASE_ORDERS = true;
 
 // Add flag for incremental updates

inventory-server/scripts/import/orders.js

@@ -19,6 +19,13 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
   const missingProducts = new Set();
   let recordsAdded = 0;
   let recordsUpdated = 0;
+  let processedCount = 0;
+  let importedCount = 0;
+  let totalOrderItems = 0;
+  let totalUniqueOrders = 0;
+
+  // Add a cumulative counter for processed orders before the loop
+  let cumulativeProcessedOrders = 0;
 
   try {
     // Insert temporary table creation queries
@@ -86,7 +93,7 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
 
     console.log('Orders: Using last sync time:', lastSyncTime);
 
-    // First get all relevant order items with basic info
+    // First get count of order items
     const [[{ total }]] = await prodConnection.query(`
       SELECT COUNT(*) as total
       FROM order_items oi
@@ -115,7 +122,8 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
         ` : ''}
     `, incrementalUpdate ? [lastSyncTime, lastSyncTime, lastSyncTime] : []);
 
-    console.log('Orders: Found changes:', total);
+    totalOrderItems = total;
+    console.log('Orders: Found changes:', totalOrderItems);
 
     // Get order items in batches
     const [orderItems] = await prodConnection.query(`
@@ -155,9 +163,6 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
 
     console.log('Orders: Processing', orderItems.length, 'order items');
 
-    const totalOrders = orderItems.length;
-    let processed = 0;
-
     // Insert order items in batches
     for (let i = 0; i < orderItems.length; i += 5000) {
       const batch = orderItems.slice(i, Math.min(i + 5000, orderItems.length));
@@ -176,22 +181,30 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
           base_discount = VALUES(base_discount)
       `, values);
 
-      processed += batch.length;
+      processedCount = i + batch.length;
       outputProgress({
         status: "running",
         operation: "Orders import",
-        message: `Loading order items: ${processed} of ${totalOrders}`,
+        message: `Loading order items: ${processedCount} of ${totalOrderItems}`,
-        current: processed,
+        current: processedCount,
-        total: totalOrders
+        total: totalOrderItems
       });
     }
 
     // Get unique order IDs
     const orderIds = [...new Set(orderItems.map(item => item.order_id))];
+    totalUniqueOrders = orderIds.length;
+    console.log('Total unique order IDs:', totalUniqueOrders);
+
+    // Reset processed count for order processing phase
+    processedCount = 0;
 
     // Get order metadata in batches
     for (let i = 0; i < orderIds.length; i += 5000) {
       const batchIds = orderIds.slice(i, i + 5000);
+      console.log(`Processing batch ${i/5000 + 1}, size: ${batchIds.length}`);
+      console.log('Sample of batch IDs:', batchIds.slice(0, 5));
+      
       const [orders] = await prodConnection.query(`
         SELECT 
           o.order_id,
@@ -204,6 +217,14 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
         LEFT JOIN users u ON o.order_cid = u.cid
         WHERE o.order_id IN (?)
       `, [batchIds]);
+      
+      console.log(`Retrieved ${orders.length} orders for ${batchIds.length} IDs`);
+      const duplicates = orders.filter((order, index, self) => 
+        self.findIndex(o => o.order_id === order.order_id) !== index
+      );
+      if (duplicates.length > 0) {
+        console.log('Found duplicates:', duplicates);
+      }
 
       const placeholders = orders.map(() => "(?, ?, ?, ?, ?, ?)").join(",");
       const values = orders.flatMap(order => [
@@ -212,17 +233,27 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
 
       await localConnection.query(`
         INSERT INTO temp_order_meta VALUES ${placeholders}
+        ON DUPLICATE KEY UPDATE
+          date = VALUES(date),
+          customer = VALUES(customer),
+          customer_name = VALUES(customer_name),
+          status = VALUES(status),
+          canceled = VALUES(canceled)
       `, values);
 
+      processedCount = i + orders.length;
       outputProgress({
         status: "running",
         operation: "Orders import",
-        message: `Loading order metadata: ${i + orders.length} of ${orderIds.length}`,
+        message: `Loading order metadata: ${processedCount} of ${totalUniqueOrders}`,
-        current: i + orders.length,
+        current: processedCount,
-        total: orderIds.length
+        total: totalUniqueOrders
       });
     }
 
+    // Reset processed count for final phase
+    processedCount = 0;
+
     // Get promotional discounts in batches
     for (let i = 0; i < orderIds.length; i += 5000) {
       const batchIds = orderIds.slice(i, i + 5000);
@@ -239,6 +270,8 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
 
         await localConnection.query(`
           INSERT INTO temp_order_discounts VALUES ${placeholders}
+          ON DUPLICATE KEY UPDATE
+            discount = VALUES(discount)
         `, values);
       }
     }
@@ -274,6 +307,7 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
           const placeholders = Array(uniqueTaxes.size).fill("(?, ?, ?)").join(",");
           await localConnection.query(`
             INSERT INTO temp_order_taxes VALUES ${placeholders}
+            ON DUPLICATE KEY UPDATE tax = VALUES(tax)
           `, values);
         }
       }
@@ -300,8 +334,6 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
     }
 
     // Now combine all the data and insert into orders table
-    let importedCount = 0;
-
     // Pre-check all products at once instead of per batch
     const allOrderPids = [...new Set(orderItems.map(item => item.pid))];
     const [existingProducts] = allOrderPids.length > 0 ? await localConnection.query(
@@ -343,6 +375,9 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
       // Filter orders and track missing products - do this in a single pass
       const validOrders = [];
       const values = [];
+      const processedOrderItems = new Set(); // Track unique order items
+      const processedOrders = new Set(); // Track unique orders
+      
       for (const order of orders) {
         if (!existingPids.has(order.pid)) {
           missingProducts.add(order.pid);
@@ -351,78 +386,111 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
         }
         validOrders.push(order);
         values.push(...columnNames.map(col => order[col] ?? null));
+        processedOrderItems.add(`${order.order_number}-${order.pid}`);
+        processedOrders.add(order.order_number);
       }
 
       if (validOrders.length > 0) {
         // Pre-compute the placeholders string once
         const singlePlaceholder = `(${columnNames.map(() => "?").join(",")})`;
         const placeholders = Array(validOrders.length).fill(singlePlaceholder).join(",");
+        
+        const result = await localConnection.query(`
+          INSERT INTO orders (${columnNames.join(",")})
+          VALUES ${placeholders}
+          ON DUPLICATE KEY UPDATE
+            SKU = VALUES(SKU),
+            date = VALUES(date),
+            price = VALUES(price),
+            quantity = VALUES(quantity),
+            discount = VALUES(discount),
+            tax = VALUES(tax),
+            tax_included = VALUES(tax_included),
+            shipping = VALUES(shipping),
+            customer = VALUES(customer),
+            customer_name = VALUES(customer_name),
+            status = VALUES(status),
+            canceled = VALUES(canceled),
+            costeach = VALUES(costeach)
+        `, validOrders.map(o => columnNames.map(col => o[col] ?? null)).flat());
+        
+        const affectedRows = result[0].affectedRows;
+        const updates = Math.floor(affectedRows / 2);
+        const inserts = affectedRows - (updates * 2);
+        
+        recordsAdded += inserts;
+        recordsUpdated += updates;
+        importedCount += processedOrderItems.size; // Count unique order items processed
+      }
 
-        // First check which orders exist and get their current values
+      // Update progress based on unique orders processed
-        const [existingOrders] = await localConnection.query(
+      cumulativeProcessedOrders += processedOrders.size;
-          `SELECT ${columnNames.join(",")} FROM orders WHERE (order_number, pid) IN (${validOrders.map(() => "(?,?)").join(",")})`,
+      outputProgress({
-          validOrders.flatMap(o => [o.order_number, o.pid])
+        status: "running",
-        );
+        operation: "Orders import",
-        const existingOrderMap = new Map(
+        message: `Imported ${importedCount} order items (${cumulativeProcessedOrders} of ${totalUniqueOrders} orders processed)`,
-          existingOrders.map(o => [`${o.order_number}-${o.pid}`, o])
+        current: cumulativeProcessedOrders,
-        );
+        total: totalUniqueOrders,
+        elapsed: formatElapsedTime((Date.now() - startTime) / 1000),
+        remaining: estimateRemaining(startTime, cumulativeProcessedOrders, totalUniqueOrders),
+        rate: calculateRate(startTime, cumulativeProcessedOrders)
+      });
+    }
 
-        // Split into inserts and updates
+    // Now try to import any orders that were skipped due to missing products
-        const insertsAndUpdates = validOrders.reduce((acc, order) => {
+    if (skippedOrders.size > 0) {
-          const key = `${order.order_number}-${order.pid}`;
+      try {
-          if (existingOrderMap.has(key)) {
+        outputProgress({
-            const existing = existingOrderMap.get(key);
+          status: "running",
-            // Check if any values are different
+          operation: "Orders import",
-            const hasChanges = columnNames.some(col => {
+          message: `Retrying import of ${skippedOrders.size} orders with previously missing products`,
-              const newVal = order[col] ?? null;
+        });
-              const oldVal = existing[col] ?? null;
-              if (typeof newVal === 'number' && typeof oldVal === 'number') {
-                return Math.abs(newVal - oldVal) > 0.00001; // Allow for tiny floating point differences
-              }
-              return newVal !== oldVal;
-            });
-            if (hasChanges) {
-              acc.updates.push({
-                order_number: order.order_number,
-                pid: order.pid,
-                values: columnNames.map(col => order[col] ?? null)
-              });
-            } else {
-              acc.inserts.push({
-                order_number: order.order_number,
-                pid: order.pid,
-                values: columnNames.map(col => order[col] ?? null)
-              });
-            }
-          } else {
-            acc.inserts.push({
-              order_number: order.order_number,
-              pid: order.pid,
-              values: columnNames.map(col => order[col] ?? null)
-            });
-          }
-          return acc;
-        }, { inserts: [], updates: [] });
 
-        // Handle inserts
+        // Get the orders that were skipped
-        if (insertsAndUpdates.inserts.length > 0) {
+        const [skippedProdOrders] = await localConnection.query(`
-          const insertPlaceholders = Array(insertsAndUpdates.inserts.length).fill(singlePlaceholder).join(",");
+          SELECT DISTINCT
-          
+            oi.order_id as order_number,
-          const insertResult = await localConnection.query(`
+            oi.pid,
-            INSERT INTO orders (${columnNames.join(",")})
+            oi.SKU,
-            VALUES ${insertPlaceholders}
+            om.date,
-          `, insertsAndUpdates.inserts.map(i => i.values).flat());
+            oi.price,
-          
+            oi.quantity,
-          recordsAdded += insertResult[0].affectedRows;
+            oi.base_discount + COALESCE(od.discount, 0) as discount,
-        }
+            COALESCE(ot.tax, 0) as tax,
+            0 as tax_included,
+            0 as shipping,
+            om.customer,
+            om.customer_name,
+            om.status,
+            om.canceled,
+            COALESCE(tc.costeach, 0) as costeach
+          FROM temp_order_items oi
+          JOIN temp_order_meta om ON oi.order_id = om.order_id
+          LEFT JOIN temp_order_discounts od ON oi.order_id = od.order_id AND oi.pid = od.pid
+          LEFT JOIN temp_order_taxes ot ON oi.order_id = ot.order_id AND oi.pid = ot.pid
+          LEFT JOIN temp_order_costs tc ON oi.order_id = tc.order_id AND oi.pid = tc.pid
+          WHERE oi.order_id IN (?)
+        `, [Array.from(skippedOrders)]);
 
-        // Handle updates - now we know these actually have changes
+        // Check which products exist now
-        if (insertsAndUpdates.updates.length > 0) {
+        const skippedPids = [...new Set(skippedProdOrders.map(o => o.pid))];
-          const updatePlaceholders = Array(insertsAndUpdates.updates.length).fill(singlePlaceholder).join(",");
+        const [existingProducts] = skippedPids.length > 0 ? await localConnection.query(
-          
+          "SELECT pid FROM products WHERE pid IN (?)",
-          const updateResult = await localConnection.query(`
+          [skippedPids]
-            INSERT INTO orders (${columnNames.join(",")})
+        ) : [[]];
-            VALUES ${updatePlaceholders}
+        const existingPids = new Set(existingProducts.map(p => p.pid));
+
+        // Filter orders that can now be imported
+        const validOrders = skippedProdOrders.filter(order => existingPids.has(order.pid));
+        const retryOrderItems = new Set(); // Track unique order items in retry
+
+        if (validOrders.length > 0) {
+          const placeholders = validOrders.map(() => `(${columnNames.map(() => "?").join(", ")})`).join(",");
+          const values = validOrders.map(o => columnNames.map(col => o[col] ?? null)).flat();
+
+          const result = await localConnection.query(`
+            INSERT INTO orders (${columnNames.join(", ")})
+            VALUES ${placeholders}
             ON DUPLICATE KEY UPDATE
               SKU = VALUES(SKU),
               date = VALUES(date),
@@ -435,28 +503,37 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
               customer = VALUES(customer),
               customer_name = VALUES(customer_name),
               status = VALUES(status),
-              canceled = VALUES(canceled)
+              canceled = VALUES(canceled),
-          `, insertsAndUpdates.updates.map(u => u.values).flat());
+              costeach = VALUES(costeach)
+          `, values);
+
+          const affectedRows = result[0].affectedRows;
+          const updates = Math.floor(affectedRows / 2);
+          const inserts = affectedRows - (updates * 2);
           
-          recordsUpdated += updateResult[0].affectedRows / 2;  // Each update counts as 2 in affectedRows
+          // Track unique order items
+          validOrders.forEach(order => {
+            retryOrderItems.add(`${order.order_number}-${order.pid}`);
+          });
+          
+          outputProgress({
+            status: "running",
+            operation: "Orders import",
+            message: `Successfully imported ${retryOrderItems.size} previously skipped order items`,
+          });
+
+          // Update the main counters
+          recordsAdded += inserts;
+          recordsUpdated += updates;
+          importedCount += retryOrderItems.size;
         }
-
+      } catch (error) {
-        importedCount += validOrders.length;
+        console.warn('Warning: Failed to retry skipped orders:', error.message);
+        console.warn(`Skipped ${skippedOrders.size} orders due to ${missingProducts.size} missing products`);
       }
-
-      outputProgress({
-        status: "running",
-        operation: "Orders import",
-        message: `Imported ${importedCount} of ${totalOrders} orders`,
-        current: importedCount,
-        total: totalOrders,
-        elapsed: formatElapsedTime((Date.now() - startTime) / 1000),
-        remaining: estimateRemaining(startTime, importedCount, totalOrders),
-        rate: calculateRate(startTime, importedCount)
-      });
     }
 
-    // Clean up temporary tables
+    // Clean up temporary tables after ALL processing is complete
     await localConnection.query(`
       DROP TEMPORARY TABLE IF EXISTS temp_order_items;
       DROP TEMPORARY TABLE IF EXISTS temp_order_meta;
@@ -465,119 +542,6 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
       DROP TEMPORARY TABLE IF EXISTS temp_order_costs;
     `);
 
-    // Import missing products if any
-    if (missingProducts.size > 0) {
-      try {
-        // Import missing products directly without materialization
-        await importMissingProducts(prodConnection, localConnection, Array.from(missingProducts));
-        
-        // Retry skipped orders after importing products
-        if (skippedOrders.size > 0) {
-          outputProgress({
-            status: "running",
-            operation: "Orders import",
-            message: `Retrying import of ${skippedOrders.size} orders with previously missing products`
-          });
-
-          const skippedOrdersArray = Array.from(skippedOrders);
-          const [skippedProdOrders] = skippedOrdersArray.length > 0 ? await prodConnection.query(`
-            SELECT
-              o.order_id,
-              CASE 
-                WHEN o.date_placed = '0000-00-00 00:00:00' OR o.date_placed IS NULL THEN o.stamp
-                ELSE o.date_placed 
-              END as date,
-              o.order_cid,
-              o.bill_firstname,
-              o.bill_lastname,
-              o.order_email,
-              o.order_status,
-              o.date_shipped,
-              o.date_cancelled,
-              oi.prod_pid,
-              oi.prod_itemnumber,
-              oi.prod_price,
-              oi.qty_ordered,
-              oi.qty_back,
-              oi.qty_placed,
-              oi.qty_placed_2,
-              oi.discounted,
-              oi.summary_cogs,
-              oi.summary_profit,
-              oi.summary_orderdate,
-              oi.summary_paiddate,
-              oi.date_added,
-              oi.stamp
-            FROM order_items oi
-            JOIN _order o ON oi.order_id = o.order_id
-            WHERE o.order_id IN (?)
-          `, [skippedOrdersArray]) : [[]];
-
-          // Prepare values for insertion
-          const skippedOrderValues = skippedProdOrders.flatMap(order => {
-            if (!order.date) {
-              console.log(`Warning: Skipped order ${order.order_id} has null date:`, JSON.stringify(order, null, 2));
-              return [];
-            }
-
-            const canceled = order.date_cancelled !== '0000-00-00 00:00:00' ? 1 : 0;
-            const customerName = `${order.bill_firstname} ${order.bill_lastname}`;
-
-            // Create an object with keys based on column names
-            const orderData = {
-              id: order.order_id,
-              order_number: order.order_id,
-              pid: order.prod_pid,
-              SKU: order.prod_itemnumber,
-              date: order.date ? (
-                order.date instanceof Date ? 
-                  order.date.toJSON()?.slice(0,10) || null : 
-                  (typeof order.date === 'string' ? order.date.split(' ')[0] : null)
-              ) : null,
-              price: order.prod_price,
-              quantity: order.qty_ordered,
-              discount: order.discounted,
-              tax: 0, // Placeholder, will be calculated later
-              tax_included: 0, // Placeholder, will be calculated later
-              shipping: 0, // Placeholder, will be calculated later
-              customer: order.order_email,
-              customer_name: customerName,
-              status: order.order_status,
-              canceled: canceled,
-            };
-
-            // Map column names to values, handling missing columns
-            return [columnNames.map(colName => orderData[colName] !== undefined ? orderData[colName] : null)];
-          });
-
-          // Construct the insert query dynamically
-          const skippedPlaceholders = skippedProdOrders.map(() => `(${columnNames.map(() => "?").join(", ")})`).join(",");
-          const skippedInsertQuery = `
-            INSERT INTO orders (${columnNames.join(", ")})
-            VALUES ${skippedPlaceholders}
-            ON DUPLICATE KEY UPDATE
-              ${columnNames.map(col => `${col} = VALUES(${col})`).join(", ")}
-          `;
-
-          // Execute the insert query
-          if (skippedOrderValues.length > 0) {
-            await localConnection.query(skippedInsertQuery, skippedOrderValues.flat());
-          }
-
-          importedCount += skippedProdOrders.length;
-
-          outputProgress({
-            status: "running",
-            operation: "Orders import",
-            message: `Successfully imported ${skippedProdOrders.length} previously skipped orders`,
-          });
-        }
-      } catch (error) {
-        console.warn('Warning: Failed to import missing products:', error.message);
-        console.warn(`Skipped ${skippedOrders.size} orders due to ${missingProducts.size} missing products`);
-      }
-    }
-
     // Only update sync status if we get here (no errors thrown)
     await localConnection.query(`
       INSERT INTO sync_status (table_name, last_sync_timestamp)
@@ -587,9 +551,9 @@ async function importOrders(prodConnection, localConnection, incrementalUpdate =
 
     return {
       status: "complete",
-      totalImported: importedCount,
+      totalImported: Math.floor(importedCount),
       recordsAdded: recordsAdded || 0,
-      recordsUpdated: recordsUpdated || 0,
+      recordsUpdated: Math.floor(recordsUpdated),
       totalSkipped: skippedOrders.size,
       missingProducts: missingProducts.size,
       incrementalUpdate,

inventory-server/scripts/import/products.js

@@ -582,7 +582,21 @@ async function importMissingProducts(prodConnection, localConnection, missingPid
           ELSE 1
         END AS replenishable,
         COALESCE(si.available_local, 0) as stock_quantity,
-        COALESCE(pq.qty, 0) as pending_qty,
+        COALESCE(
+          (SELECT SUM(oi.qty_ordered - oi.qty_placed)
+           FROM order_items oi
+           JOIN _order o ON oi.order_id = o.order_id
+           WHERE oi.prod_pid = p.pid
+             AND o.date_placed != '0000-00-00 00:00:00'
+             AND o.date_shipped = '0000-00-00 00:00:00'
+             AND oi.pick_finished = 0
+             AND oi.qty_back = 0
+             AND o.order_status != 15
+             AND o.order_status < 90
+             AND oi.qty_ordered >= oi.qty_placed
+             AND oi.qty_ordered > 0
+          ), 0
+        ) as pending_qty,
         COALESCE(ci.onpreorder, 0) as preorder_count,
         COALESCE(pnb.inventory, 0) as notions_inv_count,
         COALESCE(pcp.price_each, 0) as price,

inventory-server/scripts/import/purchase-orders.js

@@ -117,7 +117,12 @@ async function importPurchaseOrders(prodConnection, localConnection, incremental
           WHEN r.receiving_id IS NOT NULL THEN
             DATE(r.date_created)
         END as date,
-        NULLIF(p.date_estin, '0000-00-00') as expected_date,
+        CASE 
+          WHEN p.date_estin = '0000-00-00' THEN NULL
+          WHEN p.date_estin IS NULL THEN NULL
+          WHEN p.date_estin NOT REGEXP '^[0-9]{4}-[0-9]{2}-[0-9]{2}$' THEN NULL
+          ELSE p.date_estin
+        END as expected_date,
         COALESCE(p.status, 50) as status,
         p.short_note as notes,
         p.notes as long_note
@@ -359,9 +364,12 @@ async function importPurchaseOrders(prodConnection, localConnection, incremental
 
             function formatDate(dateStr) {
               if (!dateStr) return null;
+              if (dateStr === '0000-00-00' || dateStr === '0000-00-00 00:00:00') return null;
+              if (typeof dateStr === 'string' && !dateStr.match(/^\d{4}-\d{2}-\d{2}/)) return null;
               try {
                 const date = new Date(dateStr);
                 if (isNaN(date.getTime())) return null;
+                if (date.getFullYear() < 1900 || date.getFullYear() > 2100) return null;
                 return date.toISOString().split('T')[0];
               } catch (e) {
                 return null;
@@ -451,7 +459,15 @@ async function importPurchaseOrders(prodConnection, localConnection, incremental
             VALUES ${insertPlaceholders}
           `, insertsAndUpdates.inserts.map(i => i.values).flat());
 
-          recordsAdded += insertResult[0].affectedRows;
+          const affectedRows = insertResult[0].affectedRows;
+          // For an upsert, MySQL counts rows twice for updates
+          // So if affectedRows is odd, we have (updates * 2 + inserts)
+          const updates = Math.floor(affectedRows / 2);
+          const inserts = affectedRows - (updates * 2);
+          
+          recordsAdded += inserts;
+          recordsUpdated += Math.floor(updates); // Ensure we never have fractional updates
+          processed += batchProcessed;
         }
 
         // Handle updates - now we know these actually have changes
@@ -469,10 +485,15 @@ async function importPurchaseOrders(prodConnection, localConnection, incremental
               .join(",")};
           `, insertsAndUpdates.updates.map(u => u.values).flat());
 
-          recordsUpdated += updateResult[0].affectedRows / 2;  // Each update counts as 2 in affectedRows
+          const affectedRows = updateResult[0].affectedRows;
+          // For an upsert, MySQL counts rows twice for updates
+          // So if affectedRows is odd, we have (updates * 2 + inserts)
+          const updates = Math.floor(affectedRows / 2);
+          const inserts = affectedRows - (updates * 2);
+          
+          recordsUpdated += Math.floor(updates); // Ensure we never have fractional updates
+          processed += batchProcessed;
         }
-
-        processed += batchProcessed;
         
         // Update progress based on time interval
         const now = Date.now();