const { outputProgress, formatElapsedTime, estimateRemaining, calculateRate } = require('../metrics-new/utils/progress');
const fs = require('fs');
const path = require('path');
const { pipeline } = require('stream');
const { promisify } = require('util');

// Configuration constants to control which tables get imported
const IMPORT_PRODUCT_CURRENT_PRICES = false;
const IMPORT_DAILY_INVENTORY = false; 
const IMPORT_PRODUCT_STAT_HISTORY = true;

// For product stat history, limit to more recent data for faster initial import
const USE_RECENT_MONTHS = 12; // Just use the most recent months for product_stat_history

/**
 * Validates a date from MySQL before inserting it into PostgreSQL
 * @param {string|Date|null} mysqlDate - Date string or object from MySQL
 * @returns {string|null} Valid date string or null if invalid
 */
function validateDate(mysqlDate) {
  // Handle null, undefined, or empty values
  if (!mysqlDate) {
    return null;
  }
  
  // Convert to string if it's not already
  const dateStr = String(mysqlDate);
  
  // Handle MySQL zero dates and empty values
  if (dateStr === '0000-00-00' || 
      dateStr === '0000-00-00 00:00:00' ||
      dateStr.indexOf('0000-00-00') !== -1 ||
      dateStr === '') {
    return null;
  }
  
  // Check if the date is valid
  const date = new Date(mysqlDate);
  
  // If the date is invalid or suspiciously old (pre-1970), return null
  if (isNaN(date.getTime()) || date.getFullYear() < 1970) {
    return null;
  }
  
  return mysqlDate;
}

/**
 * Imports historical data from MySQL to PostgreSQL
 */
async function importHistoricalData(
  prodConnection, 
  localConnection, 
  options = {}
) {
  const {
    incrementalUpdate = true,
    oneYearAgo = new Date(new Date().setFullYear(new Date().getFullYear() - 1))
  } = options;

  const oneYearAgoStr = oneYearAgo.toISOString().split('T')[0];
  const startTime = Date.now();
  
  // Use larger batch sizes to improve performance
  const BATCH_SIZE = 5000;  // For fetching from small tables
  const INSERT_BATCH_SIZE = 500; // For inserting to small tables
  const LARGE_BATCH_SIZE = 10000; // For fetching from large tables
  const LARGE_INSERT_BATCH_SIZE = 1000; // For inserting to large tables
  
  // Calculate date for recent data
  const recentDateStr = new Date(
    new Date().setMonth(new Date().getMonth() - USE_RECENT_MONTHS)
  ).toISOString().split('T')[0];

  console.log(`Starting import with:
    - One year ago date: ${oneYearAgoStr}
    - Recent months date: ${recentDateStr} (for product_stat_history)
    - Incremental update: ${incrementalUpdate}
    - Standard batch size: ${BATCH_SIZE}
    - Standard insert batch size: ${INSERT_BATCH_SIZE}
    - Large table batch size: ${LARGE_BATCH_SIZE}
    - Large table insert batch size: ${LARGE_INSERT_BATCH_SIZE}
    - Import product_current_prices: ${IMPORT_PRODUCT_CURRENT_PRICES}
    - Import daily_inventory: ${IMPORT_DAILY_INVENTORY}
    - Import product_stat_history: ${IMPORT_PRODUCT_STAT_HISTORY}`);

  try {
    // Get last sync time for incremental updates
    const lastSyncTimes = {};
    
    if (incrementalUpdate) {
      try {
        const syncResult = await localConnection.query(`
          SELECT table_name, last_sync_timestamp
          FROM sync_status
          WHERE table_name IN (
            'imported_product_current_prices', 
            'imported_daily_inventory', 
            'imported_product_stat_history'
          )
        `);
        
        // Add check for rows existence and type
        if (syncResult && Array.isArray(syncResult.rows)) {
          for (const row of syncResult.rows) {
            lastSyncTimes[row.table_name] = row.last_sync_timestamp;
            console.log(`Last sync time for ${row.table_name}: ${row.last_sync_timestamp}`);
          }
        } else {
          console.warn('Sync status query did not return expected rows. Proceeding without last sync times.');
        }
      } catch (error) {
        console.error('Error fetching sync status:', error);
      }
    }

    // Determine how many tables will be imported
    const tablesCount = [
      IMPORT_PRODUCT_CURRENT_PRICES,
      IMPORT_DAILY_INVENTORY,
      IMPORT_PRODUCT_STAT_HISTORY
    ].filter(Boolean).length;

    // Run all imports sequentially for better reliability
    console.log(`Starting sequential imports for ${tablesCount} tables...`);
    outputProgress({
      status: "running",
      operation: "Historical data import",
      message: `Starting sequential imports for ${tablesCount} tables...`,
      current: 0,
      total: tablesCount,
      elapsed: formatElapsedTime(startTime)
    });

    let progressCount = 0;
    let productCurrentPricesResult = { recordsAdded: 0, recordsUpdated: 0, totalProcessed: 0, errors: [] };
    let dailyInventoryResult = { recordsAdded: 0, recordsUpdated: 0, totalProcessed: 0, errors: [] };
    let productStatHistoryResult = { recordsAdded: 0, recordsUpdated: 0, totalProcessed: 0, errors: [] };

    // Import product current prices
    if (IMPORT_PRODUCT_CURRENT_PRICES) {
      console.log('Importing product current prices...');
      productCurrentPricesResult = await importProductCurrentPrices(
        prodConnection, 
        localConnection, 
        oneYearAgoStr, 
        lastSyncTimes['imported_product_current_prices'], 
        BATCH_SIZE,
        INSERT_BATCH_SIZE,
        incrementalUpdate,
        startTime
      );
      progressCount++;
      
      outputProgress({
        status: "running",
        operation: "Historical data import",
        message: `Completed import ${progressCount} of ${tablesCount}`,
        current: progressCount,
        total: tablesCount,
        elapsed: formatElapsedTime(startTime)
      });
    }
    
    // Import daily inventory
    if (IMPORT_DAILY_INVENTORY) {
      console.log('Importing daily inventory...');
      dailyInventoryResult = await importDailyInventory(
        prodConnection, 
        localConnection, 
        oneYearAgoStr, 
        lastSyncTimes['imported_daily_inventory'], 
        BATCH_SIZE,
        INSERT_BATCH_SIZE,
        incrementalUpdate,
        startTime
      );
      progressCount++;
      
      outputProgress({
        status: "running",
        operation: "Historical data import",
        message: `Completed import ${progressCount} of ${tablesCount}`,
        current: progressCount,
        total: tablesCount,
        elapsed: formatElapsedTime(startTime)
      });
    }
    
    // Import product stat history - using optimized approach
    if (IMPORT_PRODUCT_STAT_HISTORY) {
      console.log('Importing product stat history...');
      productStatHistoryResult = await importProductStatHistory(
        prodConnection, 
        localConnection, 
        recentDateStr, // Use more recent date for this massive table 
        lastSyncTimes['imported_product_stat_history'], 
        LARGE_BATCH_SIZE,
        LARGE_INSERT_BATCH_SIZE,
        incrementalUpdate,
        startTime,
        USE_RECENT_MONTHS // Pass the recent months constant
      );
      progressCount++;
      
      outputProgress({
        status: "running",
        operation: "Historical data import",
        message: `Completed import ${progressCount} of ${tablesCount}`,
        current: progressCount,
        total: tablesCount,
        elapsed: formatElapsedTime(startTime)
      });
    }

    // Aggregate results
    const totalRecordsAdded = 
      productCurrentPricesResult.recordsAdded + 
      dailyInventoryResult.recordsAdded + 
      productStatHistoryResult.recordsAdded;
    
    const totalRecordsUpdated = 
      productCurrentPricesResult.recordsUpdated + 
      dailyInventoryResult.recordsUpdated + 
      productStatHistoryResult.recordsUpdated;
    
    const totalProcessed = 
      productCurrentPricesResult.totalProcessed + 
      dailyInventoryResult.totalProcessed + 
      productStatHistoryResult.totalProcessed;
    
    const allErrors = [
      ...productCurrentPricesResult.errors,
      ...dailyInventoryResult.errors,
      ...productStatHistoryResult.errors
    ];

    // Log import summary
    console.log(`
Historical data import complete:
-------------------------------
Records added:   ${totalRecordsAdded}
Records updated: ${totalRecordsUpdated}
Total processed: ${totalProcessed}
Errors:          ${allErrors.length}
Time taken:      ${formatElapsedTime(startTime)}
    `);

    // Final progress update
    outputProgress({
      status: "complete",
      operation: "Historical data import",
      message: `Import complete. Added: ${totalRecordsAdded}, Updated: ${totalRecordsUpdated}, Errors: ${allErrors.length}`,
      current: tablesCount,
      total: tablesCount,
      elapsed: formatElapsedTime(startTime)
    });

    // Log any errors
    if (allErrors.length > 0) {
      console.log('Errors encountered during import:');
      console.log(JSON.stringify(allErrors, null, 2));
    }

    // Calculate duration
    const endTime = Date.now();
    const durationSeconds = Math.round((endTime - startTime) / 1000);
    const finalStatus = allErrors.length === 0 ? 'complete' : 'failed';
    const errorMessage = allErrors.length > 0 ? JSON.stringify(allErrors) : null;

    // Update import history
    await localConnection.query(`
      INSERT INTO import_history (
        table_name, 
        end_time, 
        duration_seconds, 
        records_added, 
        records_updated, 
        is_incremental, 
        status, 
        error_message,
        additional_info
      )
      VALUES ($1, NOW(), $2, $3, $4, $5, $6, $7, $8)
    `, [
      'historical_data_combined',
      durationSeconds,
      totalRecordsAdded,
      totalRecordsUpdated,
      incrementalUpdate,
      finalStatus,
      errorMessage,
      JSON.stringify({ 
        totalProcessed,
        tablesImported: {
          imported_product_current_prices: IMPORT_PRODUCT_CURRENT_PRICES,
          imported_daily_inventory: IMPORT_DAILY_INVENTORY,
          imported_product_stat_history: IMPORT_PRODUCT_STAT_HISTORY
        }
      })
    ]);

    // Return summary
    return {
      recordsAdded: totalRecordsAdded,
      recordsUpdated: totalRecordsUpdated,
      totalProcessed,
      errors: allErrors,
      timeTaken: formatElapsedTime(startTime)
    };
  } catch (error) {
    console.error('Error importing historical data:', error);
    
    // Final progress update on error
    outputProgress({
      status: "failed",
      operation: "Historical data import",
      message: `Import failed: ${error.message}`,
      elapsed: formatElapsedTime(startTime)
    });

    throw error;
  }
}

/**
 * Imports product_current_prices data from MySQL to PostgreSQL
 */
async function importProductCurrentPrices(
  prodConnection, 
  localConnection, 
  oneYearAgoStr, 
  lastSyncTime, 
  batchSize,
  insertBatchSize,
  incrementalUpdate,
  startTime
) {
  let recordsAdded = 0;
  let recordsUpdated = 0;
  let totalProcessed = 0;
  let errors = [];
  let offset = 0;
  let allProcessed = false;

  try {
    // Get total count for progress reporting
    const [countResult] = await prodConnection.query(`
      SELECT COUNT(*) as total
      FROM product_current_prices
      WHERE (date_active >= ? OR date_deactive >= ?)
        ${incrementalUpdate && lastSyncTime ? `AND date_deactive > ?` : ''}
    `, [oneYearAgoStr, oneYearAgoStr, ...(incrementalUpdate && lastSyncTime ? [lastSyncTime] : [])]);

    const totalCount = countResult[0].total;

    outputProgress({
      status: "running",
      operation: "Historical data import - Product Current Prices",
      message: `Found ${totalCount} records to process`,
      current: 0,
      total: totalCount,
      elapsed: formatElapsedTime(startTime)
    });

    // Process in batches for better performance
    while (!allProcessed) {
      try {
        // Fetch batch from production
        const [rows] = await prodConnection.query(`
          SELECT
            price_id,
            pid,
            qty_buy,
            is_min_qty_buy,
            price_each,
            qty_limit,
            no_promo,
            checkout_offer,
            active,
            date_active,
            date_deactive
          FROM product_current_prices
          WHERE (date_active >= ? OR date_deactive >= ?)
            ${incrementalUpdate && lastSyncTime ? `AND date_deactive > ?` : ''}
          ORDER BY price_id
          LIMIT ? OFFSET ?
        `, [
          oneYearAgoStr, 
          oneYearAgoStr, 
          ...(incrementalUpdate && lastSyncTime ? [lastSyncTime] : []),
          batchSize,
          offset
        ]);

        if (rows.length === 0) {
          allProcessed = true;
          break;
        }

        // Process rows in smaller batches for better performance
        for (let i = 0; i < rows.length; i += insertBatchSize) {
          const batch = rows.slice(i, i + insertBatchSize);
          
          if (batch.length === 0) continue;
          
          try {
            // Build parameterized query to handle NULL values properly
            const values = [];
            const placeholders = [];
            let placeholderIndex = 1;
            
            for (const row of batch) {
              const rowPlaceholders = [
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`
              ];
              
              placeholders.push(`(${rowPlaceholders.join(', ')})`);
              
              values.push(
                row.price_id,
                row.pid,
                row.qty_buy,
                row.is_min_qty_buy ? true : false,
                row.price_each,
                row.qty_limit, // PostgreSQL will handle null values properly
                row.no_promo ? true : false,
                row.checkout_offer ? true : false,
                row.active ? true : false,
                validateDate(row.date_active),
                validateDate(row.date_deactive)
              );
            }
            
            // Execute batch insert
            const result = await localConnection.query(`
              WITH ins AS (
                INSERT INTO imported_product_current_prices (
                  price_id, pid, qty_buy, is_min_qty_buy, price_each, qty_limit, 
                  no_promo, checkout_offer, active, date_active, date_deactive
                )
                VALUES ${placeholders.join(',\n')}
                ON CONFLICT (price_id) DO UPDATE SET
                  pid = EXCLUDED.pid,
                  qty_buy = EXCLUDED.qty_buy,
                  is_min_qty_buy = EXCLUDED.is_min_qty_buy,
                  price_each = EXCLUDED.price_each,
                  qty_limit = EXCLUDED.qty_limit,
                  no_promo = EXCLUDED.no_promo,
                  checkout_offer = EXCLUDED.checkout_offer,
                  active = EXCLUDED.active,
                  date_active = EXCLUDED.date_active,
                  date_deactive = EXCLUDED.date_deactive,
                  updated = CURRENT_TIMESTAMP
                RETURNING (xmax = 0) AS inserted
              )
              SELECT 
                COUNT(*) FILTER (WHERE inserted) AS inserted_count, 
                COUNT(*) FILTER (WHERE NOT inserted) AS updated_count
              FROM ins
            `, values);
            
            // Safely update counts based on the result
            if (result && result.rows && result.rows.length > 0) {
              const insertedCount = parseInt(result.rows[0].inserted_count || 0);
              const updatedCount = parseInt(result.rows[0].updated_count || 0);
              
              recordsAdded += insertedCount;
              recordsUpdated += updatedCount;
            }
          } catch (error) {
            console.error(`Error in batch import of product_current_prices at offset ${i}:`, error);
            errors.push({
              table: 'imported_product_current_prices',
              batchOffset: i,
              batchSize: batch.length,
              error: error.message
            });
          }
        }

        totalProcessed += rows.length;
        offset += rows.length;

        // Update progress
        outputProgress({
          status: "running",
          operation: "Historical data import - Product Current Prices",
          message: `Processed ${totalProcessed} of ${totalCount} records`,
          current: totalProcessed,
          total: totalCount,
          elapsed: formatElapsedTime(startTime),
          remaining: estimateRemaining(startTime, totalProcessed, totalCount),
          rate: calculateRate(startTime, totalProcessed)
        });
      } catch (error) {
        console.error('Error in batch import of product_current_prices:', error);
        errors.push({
          table: 'imported_product_current_prices',
          error: error.message,
          offset: offset,
          batchSize: batchSize
        });
        
        // Try to continue with next batch
        offset += batchSize;
      }
    }

    // Update sync status
    await localConnection.query(`
      INSERT INTO sync_status (table_name, last_sync_timestamp)
      VALUES ('imported_product_current_prices', NOW())
      ON CONFLICT (table_name) DO UPDATE SET
        last_sync_timestamp = NOW()
    `);

    return { recordsAdded, recordsUpdated, totalProcessed, errors };
  } catch (error) {
    console.error('Error in product current prices import:', error);
    return { 
      recordsAdded, 
      recordsUpdated, 
      totalProcessed, 
      errors: [...errors, { 
        table: 'imported_product_current_prices',
        error: error.message
      }] 
    };
  }
}

/**
 * Imports daily_inventory data from MySQL to PostgreSQL
 */
async function importDailyInventory(
  prodConnection, 
  localConnection, 
  oneYearAgoStr, 
  lastSyncTime, 
  batchSize,
  insertBatchSize,
  incrementalUpdate,
  startTime
) {
  let recordsAdded = 0;
  let recordsUpdated = 0;
  let totalProcessed = 0;
  let errors = [];
  let offset = 0;
  let allProcessed = false;

  try {
    // Get total count for progress reporting
    const [countResult] = await prodConnection.query(`
      SELECT COUNT(*) as total
      FROM daily_inventory
      WHERE date >= ?
        ${incrementalUpdate && lastSyncTime ? `AND stamp > ?` : ''}
    `, [oneYearAgoStr, ...(incrementalUpdate && lastSyncTime ? [lastSyncTime] : [])]);

    const totalCount = countResult[0].total;

    outputProgress({
      status: "running",
      operation: "Historical data import - Daily Inventory",
      message: `Found ${totalCount} records to process`,
      current: 0,
      total: totalCount,
      elapsed: formatElapsedTime(startTime)
    });

    // Process in batches for better performance
    while (!allProcessed) {
      try {
        // Fetch batch from production
        const [rows] = await prodConnection.query(`
          SELECT
            date,
            pid,
            amountsold,
            times_sold,
            qtyreceived,
            price,
            costeach,
            stamp
          FROM daily_inventory
          WHERE date >= ?
            ${incrementalUpdate && lastSyncTime ? `AND stamp > ?` : ''}
          ORDER BY date, pid
          LIMIT ? OFFSET ?
        `, [
          oneYearAgoStr, 
          ...(incrementalUpdate && lastSyncTime ? [lastSyncTime] : []),
          batchSize,
          offset
        ]);

        if (rows.length === 0) {
          allProcessed = true;
          break;
        }

        // Process rows in smaller batches for better performance
        for (let i = 0; i < rows.length; i += insertBatchSize) {
          const batch = rows.slice(i, i + insertBatchSize);
          
          if (batch.length === 0) continue;
          
          try {
            // Build parameterized query to handle NULL values properly
            const values = [];
            const placeholders = [];
            let placeholderIndex = 1;
            
            for (const row of batch) {
              const rowPlaceholders = [
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`
              ];
              
              placeholders.push(`(${rowPlaceholders.join(', ')})`);
              
              values.push(
                validateDate(row.date),
                row.pid,
                row.amountsold || 0,
                row.times_sold || 0,
                row.qtyreceived || 0,
                row.price || 0,
                row.costeach || 0,
                validateDate(row.stamp)
              );
            }
            
            // Execute batch insert
            const result = await localConnection.query(`
              WITH ins AS (
                INSERT INTO imported_daily_inventory (
                  date, pid, amountsold, times_sold, qtyreceived, price, costeach, stamp
                )
                VALUES ${placeholders.join(',\n')}
                ON CONFLICT (date, pid) DO UPDATE SET
                  amountsold = EXCLUDED.amountsold,
                  times_sold = EXCLUDED.times_sold,
                  qtyreceived = EXCLUDED.qtyreceived,
                  price = EXCLUDED.price,
                  costeach = EXCLUDED.costeach,
                  stamp = EXCLUDED.stamp,
                  updated = CURRENT_TIMESTAMP
                RETURNING (xmax = 0) AS inserted
              )
              SELECT 
                COUNT(*) FILTER (WHERE inserted) AS inserted_count, 
                COUNT(*) FILTER (WHERE NOT inserted) AS updated_count
              FROM ins
            `, values);
            
            // Safely update counts based on the result
            if (result && result.rows && result.rows.length > 0) {
              const insertedCount = parseInt(result.rows[0].inserted_count || 0);
              const updatedCount = parseInt(result.rows[0].updated_count || 0);
              
              recordsAdded += insertedCount;
              recordsUpdated += updatedCount;
            }
          } catch (error) {
            console.error(`Error in batch import of daily_inventory at offset ${i}:`, error);
            errors.push({
              table: 'imported_daily_inventory',
              batchOffset: i,
              batchSize: batch.length,
              error: error.message
            });
          }
        }

        totalProcessed += rows.length;
        offset += rows.length;

        // Update progress
        outputProgress({
          status: "running",
          operation: "Historical data import - Daily Inventory",
          message: `Processed ${totalProcessed} of ${totalCount} records`,
          current: totalProcessed,
          total: totalCount,
          elapsed: formatElapsedTime(startTime),
          remaining: estimateRemaining(startTime, totalProcessed, totalCount),
          rate: calculateRate(startTime, totalProcessed)
        });
      } catch (error) {
        console.error('Error in batch import of daily_inventory:', error);
        errors.push({
          table: 'imported_daily_inventory',
          error: error.message,
          offset: offset,
          batchSize: batchSize
        });
        
        // Try to continue with next batch
        offset += batchSize;
      }
    }

    // Update sync status
    await localConnection.query(`
      INSERT INTO sync_status (table_name, last_sync_timestamp)
      VALUES ('imported_daily_inventory', NOW())
      ON CONFLICT (table_name) DO UPDATE SET
        last_sync_timestamp = NOW()
    `);

    return { recordsAdded, recordsUpdated, totalProcessed, errors };
  } catch (error) {
    console.error('Error in daily inventory import:', error);
    return { 
      recordsAdded, 
      recordsUpdated, 
      totalProcessed, 
      errors: [...errors, { 
        table: 'imported_daily_inventory',
        error: error.message
      }] 
    };
  }
}

/**
 * Imports product_stat_history data from MySQL to PostgreSQL
 * Using fast direct inserts without conflict checking
 */
async function importProductStatHistory(
  prodConnection, 
  localConnection, 
  recentDateStr, // Use more recent date instead of one year ago
  lastSyncTime, 
  batchSize,
  insertBatchSize,
  incrementalUpdate,
  startTime,
  recentMonths // Add parameter for recent months
) {
  let recordsAdded = 0;
  let recordsUpdated = 0;
  let totalProcessed = 0;
  let errors = [];
  let offset = 0;
  let allProcessed = false;
  let lastRateCheck = Date.now();
  let lastProcessed = 0;

  try {
    // Get total count for progress reporting
    const [countResult] = await prodConnection.query(`
      SELECT COUNT(*) as total
      FROM product_stat_history
      WHERE date >= ?
        ${incrementalUpdate && lastSyncTime ? `AND date > ?` : ''}
    `, [recentDateStr, ...(incrementalUpdate && lastSyncTime ? [lastSyncTime] : [])]);

    const totalCount = countResult[0].total;
    console.log(`Found ${totalCount} records to process in product_stat_history (using recent date: ${recentDateStr})`);

    // Progress indicator
    outputProgress({
      status: "running",
      operation: "Historical data import - Product Stat History",
      message: `Found ${totalCount} records to process (last ${recentMonths} months only)`,
      current: 0,
      total: totalCount,
      elapsed: formatElapsedTime(startTime)
    });

    // If not incremental, truncate the table first for better performance
    if (!incrementalUpdate) {
      console.log('Truncating imported_product_stat_history for full import...');
      await localConnection.query('TRUNCATE TABLE imported_product_stat_history');
    } else if (lastSyncTime) {
      // For incremental updates, delete records that will be reimported
      console.log(`Deleting records from imported_product_stat_history since ${lastSyncTime}...`);
      await localConnection.query('DELETE FROM imported_product_stat_history WHERE date > $1', [lastSyncTime]);
    }

    // Process in batches for better performance
    while (!allProcessed) {
      try {
        // Fetch batch from production with minimal filtering and no sorting
        const [rows] = await prodConnection.query(`
          SELECT
            pid,
            date,
            COALESCE(score, 0) as score,
            COALESCE(score2, 0) as score2,
            COALESCE(qty_in_baskets, 0) as qty_in_baskets,
            COALESCE(qty_sold, 0) as qty_sold,
            COALESCE(notifies_set, 0) as notifies_set,
            COALESCE(visibility_score, 0) as visibility_score,
            COALESCE(health_score, 0) as health_score,
            COALESCE(sold_view_score, 0) as sold_view_score
          FROM product_stat_history
          WHERE date >= ?
            ${incrementalUpdate && lastSyncTime ? `AND date > ?` : ''}
          LIMIT ? OFFSET ?
        `, [
          recentDateStr, 
          ...(incrementalUpdate && lastSyncTime ? [lastSyncTime] : []),
          batchSize,
          offset
        ]);

        if (rows.length === 0) {
          allProcessed = true;
          break;
        }

        // Process rows in smaller batches for better performance
        for (let i = 0; i < rows.length; i += insertBatchSize) {
          const batch = rows.slice(i, i + insertBatchSize);
          
          if (batch.length === 0) continue;
          
          try {
            // Build parameterized query to handle NULL values properly
            const values = [];
            const placeholders = [];
            let placeholderIndex = 1;
            
            for (const row of batch) {
              const rowPlaceholders = [
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`,
                `$${placeholderIndex++}`
              ];
              
              placeholders.push(`(${rowPlaceholders.join(', ')})`);
              
              values.push(
                row.pid,
                validateDate(row.date),
                row.score,
                row.score2,
                row.qty_in_baskets,
                row.qty_sold,
                row.notifies_set,
                row.visibility_score,
                row.health_score,
                row.sold_view_score
              );
            }
            
            // Execute direct batch insert without conflict checking
            await localConnection.query(`
              INSERT INTO imported_product_stat_history (
                pid, date, score, score2, qty_in_baskets, qty_sold, notifies_set,
                visibility_score, health_score, sold_view_score
              )
              VALUES ${placeholders.join(',\n')}
            `, values);
            
            // All inserts are new records when using this approach
            recordsAdded += batch.length;
          } catch (error) {
            console.error(`Error in batch insert of product_stat_history at offset ${i}:`, error);
            errors.push({
              table: 'imported_product_stat_history',
              batchOffset: i,
              batchSize: batch.length,
              error: error.message
            });
          }
        }

        totalProcessed += rows.length;
        offset += rows.length;

        // Calculate current rate every 10 seconds or 100,000 records
        const now = Date.now();
        if (now - lastRateCheck > 10000 || totalProcessed - lastProcessed > 100000) {
          const timeElapsed = (now - lastRateCheck) / 1000; // seconds
          const recordsProcessed = totalProcessed - lastProcessed;
          const currentRate = Math.round(recordsProcessed / timeElapsed);
          
          console.log(`Current import rate: ${currentRate} records/second`);
          
          lastRateCheck = now;
          lastProcessed = totalProcessed;
        }

        // Update progress
        outputProgress({
          status: "running",
          operation: "Historical data import - Product Stat History",
          message: `Processed ${totalProcessed} of ${totalCount} records`,
          current: totalProcessed,
          total: totalCount,
          elapsed: formatElapsedTime(startTime),
          remaining: estimateRemaining(startTime, totalProcessed, totalCount),
          rate: calculateRate(startTime, totalProcessed)
        });
      } catch (error) {
        console.error('Error in batch import of product_stat_history:', error);
        errors.push({
          table: 'imported_product_stat_history',
          error: error.message,
          offset: offset,
          batchSize: batchSize
        });
        
        // Try to continue with next batch
        offset += batchSize;
      }
    }

    // Update sync status
    await localConnection.query(`
      INSERT INTO sync_status (table_name, last_sync_timestamp)
      VALUES ('imported_product_stat_history', NOW())
      ON CONFLICT (table_name) DO UPDATE SET
        last_sync_timestamp = NOW()
    `);

    return { recordsAdded, recordsUpdated, totalProcessed, errors };
  } catch (error) {
    console.error('Error in product stat history import:', error);
    return { 
      recordsAdded, 
      recordsUpdated, 
      totalProcessed, 
      errors: [...errors, { 
        table: 'imported_product_stat_history',
        error: error.message
      }] 
    };
  }
}

module.exports = importHistoricalData;