<?php
/** 
 * EVP - CSV Importer Scheduler
 * This file linked to scheduler.opt.php
 **/

set_time_limit(0);
ini_set('display_errors', 1);

require_once('../../config.php');

// Add MySQL connection check function
function ensure_mysql_connection($db, $log)
{
    try {
        if (!@$db->link->query("SELECT 1")) {
            throw new Exception("Connection lost");
        }
    } catch (Exception $e) {
        $log->system_debug("MySQL connection lost, attempting to reconnect...");
        // Close connection if still active
        if ($db->link) {
            @$db->link->close();
        }
        // Reconnect to database
        $db->connectDB(); // Using public connectDB method if available, or recreate connection

        try {
            if ($db->link->query("SELECT 1")) {
                $log->system_debug("MySQL reconnection successful");
                return true;
            } else {
                $log->system_debug("MySQL reconnection failed: " . mysqli_connect_error());
                return false;
            }
        } catch (Exception $e) {
            $log->system_debug("MySQL reconnection failed: " . $e->getMessage());
            return false;
        }
    }
    return true;
}

$log->system_debug("Import Scheduler started.");

$script_start_time = microtime(true);

// Calculate appropriate batch size based on available memory and time limit
$memory_mb = 1024; // Hardcoded to 1024MB for performance
$time_limit = (int) ini_get('max_execution_time');
if ($time_limit == 0) {
    $time_limit = 120; // default conservative value
}

// Empirical: 2000 rows/sec, 2KB/row, but add safety factors
$estimated_rows_per_second = 2000;
$time_safety_factor = 0.8; // Use 80% of available time
$available_processing_time = $time_limit * $time_safety_factor;
$optimal_batch_size = (int) ($estimated_rows_per_second * $available_processing_time);

$memory_safety_factor = 0.7; // Use 70% of available memory
$memory_based_limit = ($memory_mb * 1024 * $memory_safety_factor) / 2; // 2KB per row

// Take the smaller of the two limits
$batchSize = min($optimal_batch_size, $memory_based_limit);
$batchSize = max(2000, $batchSize); // Remove upper clamp of 10,000

// If total_lines is known and less than batchSize, use total_lines
if (isset($total_lines) && $total_lines > 0 && $total_lines < $batchSize) {
    $batchSize = $total_lines;
}

$max_rows_to_process = $batchSize; // Ensure this is always defined

$log->system_debug("Calculated batch size: $batchSize rows (based on $time_limit second limit and $memory_mb MB memory)");

$task_query = "SELECT * FROM evp_task WHERE status = 'importing' LIMIT 1";
$task_result = $db->select($task_query);

if ($task_result) {

    while ($task = $task_result->fetch_assoc()) {
        $csv_name = $task['csv_name'];
        $user_id = $task['user_id'];

        // ATOMIC LOCK: Only one process can claim this task
        $current_pid = getmypid();
        $lock_expiry = date('Y-m-d H:i:s', strtotime('+30 minutes'));
        $lock_sql = "UPDATE evp_task
            SET meta = JSON_SET(
                meta,
                '$.import_locked', true,
                '$.import_pid', $current_pid,
                '$.import_locked_until', '$lock_expiry'
            )
            WHERE csv_name = '$csv_name'
              AND user_id = '$user_id'
              AND status = 'importing'
              AND (JSON_EXTRACT(meta, '$.import_locked') IS NULL OR JSON_EXTRACT(meta, '$.import_locked') = false OR JSON_EXTRACT(meta, '$.import_locked_until') < NOW())";
        $db->update($lock_sql);
        if ($db->link->affected_rows == 0) {
            $log->system_debug("Could not acquire lock for CSV: $csv_name. Another process is working on it.");
            continue;
        }
        $log->system_debug("Lock acquired for CSV: $csv_name with PID: $current_pid until: $lock_expiry");

        $log->system_debug("Processing task for CSV: $csv_name, User ID: $user_id");        // Register shutdown function here, after $csv_name is set
        register_shutdown_function(function () use ($log, $csv_name, $script_start_time, $db, $user_id) {
            $error = error_get_last();
            $max_exec_time = round(microtime(true) - $script_start_time, 4);
            if ($error !== null) {
                $log->system_debug("[SHUTDOWN] Script exited with error for CSV: $csv_name | Type: {$error['type']} | Message: {$error['message']} | File: {$error['file']} | Line: {$error['line']} | Max script run time: $max_exec_time s");

                // If fatal error, update task status with error info
                if (
                    $error['type'] === E_ERROR || $error['type'] === E_PARSE || $error['type'] === E_CORE_ERROR ||
                    $error['type'] === E_COMPILE_ERROR || $error['type'] === E_USER_ERROR
                ) {

                    // Try to recover task status if fatal error
                    try {
                        // Reconnect to database if needed
                        if ($db && $db->link && !$db->link->ping()) {
                            @$db->link->close();
                            $db->connectDB();
                        }

                        if ($db && $db->link) {
                            $error_msg = mysqli_real_escape_string($db->link, substr($error['message'], 0, 255));
                            $db->update("UPDATE evp_task SET meta = JSON_SET(meta, '$.last_error', '{$error_msg}') WHERE csv_name = '$csv_name' AND user_id = '$user_id'");
                            $log->system_debug("[SHUTDOWN] Updated task meta with error information");
                        }
                    } catch (Exception $e) {
                        $log->system_debug("[SHUTDOWN] Failed to update task status after fatal error: " . $e->getMessage());
                    }
                }
            } else {
                $log->system_debug("[SHUTDOWN] Script exited normally for CSV: $csv_name | Max script run time: $max_exec_time s");
            }
        });        // Load import_offset and other metadata from task meta
        $meta = json_decode($task['meta'], true) ?: [];
        $import_offset = isset($meta['import_offset']) ? (int) $meta['import_offset'] : 0;
        $total_lines = isset($meta['total_lines']) ? (int) $meta['total_lines'] : 0;
        $detected_encoding = isset($meta['detected_encoding']) ? $meta['detected_encoding'] : null;

        // --- CUMULATIVE COUNTERS INIT (persisted in meta) ---
        if (!isset($meta['cumulative_imported']))
            $meta['cumulative_imported'] = 0;
        if (!isset($meta['cumulative_duplicates']))
            $meta['cumulative_duplicates'] = 0;
        if (!isset($meta['cumulative_skipped']))
            $meta['cumulative_skipped'] = 0;
        if (!isset($meta['cumulative_total']))
            $meta['cumulative_total'] = 0;

        // Fetch cc_data and check_mailbox from evp_email_list_metadata
        $metadata_query = "SELECT cc_data, check_mailbox FROM evp_email_list_metadata WHERE csv_file_name = '$csv_name' AND total = '0' AND user_id = '$user_id' LIMIT 1";
        $log->system_debug("Metadata Query: $metadata_query");
        $metadata_result = $db->select($metadata_query);
        if ($metadata_result) {
            $metadata = $metadata_result->fetch_assoc();
            $cc_data = unserialize($metadata['cc_data']);
            if (is_string($cc_data)) {
                $cc_data = json_decode($cc_data, true);
                $log->system_debug("cc_data (after json_decode): " . json_encode($cc_data));
            }
            $log->system_debug("cc_data: " . json_encode($cc_data));
            $email_name_val = $cc_data['email_name'];
            $log->system_debug("Email name value: $email_name_val");
            $check_mailbox = $metadata['check_mailbox'];


            // Apply condition for check_mailbox
            if ($check_mailbox == 'true') {
                $mail_box = 'pending';
            } elseif ($check_mailbox == NULL) {
                $mail_box = NULL;
            }            // Initialize email meta (do not overwrite the task meta variable)
            $email_meta = '{"retry_counter":0,"last_try":0,"retry_times":[]}';

            $file_path = $s_path . '/tmp/csv/' . $csv_name . '.csv';
            if (!file_exists($file_path)) {
                $log->system_debug("File not found: $file_path");
                $db->update("UPDATE evp_task SET status = 'failed' WHERE csv_name = '$csv_name' AND user_id = '$user_id'");
                continue;
            }

            $handle = fopen($file_path, 'rb');
            if (!$handle) {
                $log->system_debug("Failed to open file: $file_path");
                continue;
            }
            $file_header = fgetcsv($handle, 0, ",", "\"", "\\");
            $log->system_debug("File header: " . implode(',', $file_header));

            // Skip to import_offset
            $current_line = 0;
            while ($current_line < $import_offset && ($line = fgets($handle)) !== false) {
                $current_line++;
            }            // Check if we already have stored encoding in meta
            $encoding_detected = 'UTF-8'; // Default encoding

            if (!empty($detected_encoding)) {
                // Use stored encoding from previous run
                $encoding_detected = $detected_encoding;
                $log->system_debug("Using stored encoding from previous run: $encoding_detected");
            } else {
                $log->system_debug("No stored encoding found, detecting from file content");
                // First run - use more advanced detection with multiple samples from different parts of the file
                $possible_encodings = ['UTF-8', 'ISO-8859-1', 'Windows-1252', 'ASCII', 'SJIS', 'EUC-JP', 'JIS', 'GBK', 'BIG5', 'CP1251', 'KOI8-R', 'ISO-8859-15'];
                $encoding_samples = [];

                // Save current position
                $pos = ftell($handle);

                // Sample from beginning, middle, and end of file for better detection
                $file_positions = [
                    0, // Beginning
                    filesize($file_path) / 3, // 1/3 of the way
                    (filesize($file_path) / 3) * 2 // 2/3 of the way
                ];

                foreach ($file_positions as $position) {
                    // Seek to position, making sure it's an integer
                    fseek($handle, (int) $position);

                    // Get to the start of a new line to avoid partial lines
                    if ($position > 0) {
                        fgets($handle);
                    }

                    // Sample 5 lines from each position
                    for ($i = 0; $i < 5; $i++) {
                        $line = fgets($handle);
                        if ($line === false)
                            break;

                        // Try to detect encoding
                        $encoding = mb_detect_encoding($line, $possible_encodings, true);
                        if ($encoding) {
                            $encoding_samples[] = $encoding;
                        }
                    }
                }

                // Use the most common encoding detected, fallback to UTF-8
                if (count($encoding_samples) > 0) {
                    $encoding_counts = array_count_values($encoding_samples);
                    arsort($encoding_counts);
                    $encoding_detected = key($encoding_counts);
                }

                // Store detected encoding in meta for future runs
                $meta['detected_encoding'] = $encoding_detected;
                $db->update("UPDATE evp_task SET meta = JSON_SET(meta, '$.detected_encoding', '$encoding_detected') WHERE csv_name = '$csv_name' AND user_id = '$user_id'");

                $log->system_debug("Detected encoding from multiple file sections: $encoding_detected");

                // Return to previous position
                fseek($handle, $pos);
            }            // Count total lines if not already stored in meta (first run only)
            // --- FIX: Always count the real number of lines if total_lines is not set or is less than the actual file line count ---
            $need_to_count_lines = false;
            if ($total_lines == 0) {
                $need_to_count_lines = true;
            } else {
                // Double-check: if the file has more lines than total_lines, recount
                $file_line_count = 0;
                $file_for_count = fopen($file_path, 'rb');
                if ($file_for_count) {
                    // Skip header
                    fgets($file_for_count);
                    while (fgets($file_for_count) !== false) {
                        $file_line_count++;
                    }
                    fclose($file_for_count);
                }
                if ($file_line_count > $total_lines) {
                    $log->system_debug("Detected more lines in file ($file_line_count) than stored total_lines ($total_lines), recounting.");
                    $need_to_count_lines = true;
                }
            }
            if ($need_to_count_lines) {
                $log->system_debug("Counting total lines in CSV file (ignoring estimate, using real file count)");
                $file_for_count = fopen($file_path, 'rb');
                $file_line_count = 0;
                if ($file_for_count) {
                    // Skip header
                    fgets($file_for_count);
                    while (fgets($file_for_count) !== false) {
                        $file_line_count++;
                    }
                    fclose($file_for_count);
                }
                $total_lines = $file_line_count;
                $meta['total_lines'] = $total_lines;
                $db->update("UPDATE evp_task SET meta = JSON_SET(meta, '$.total_lines', $total_lines) WHERE csv_name = '$csv_name' AND user_id = '$user_id'");
                $log->system_debug("Total lines in CSV file (counted): $total_lines");
            }            // Skip to import_offset position to continue where we left off
            fseek($handle, 0);
            fgetcsv($handle, 0, ",", "\"", "\\"); // skip header

            // If we already counted lines and need to resume from offset, no need to skip again
            $current_line = 0;
            if ($import_offset > 0) {
                while ($current_line < $import_offset && ($line = fgets($handle)) !== false) {
                    $current_line++;
                }
                $log->system_debug("Skipped to offset: $current_line");
            }

            $skipped = 0;
            $all_valid_emails = [];
            $email_to_row = [];
            $rows = [];
            $processed = 0;
            $totalInserted = 0;

            // --- FIX: Always define dataC for SQL insert ---
            $dataC = '';
            if (!empty($cc_data)) {
                foreach ($cc_data as $key => $val) {
                    if ($key !== 'email_name') {
                        $dataC .= "$key, ";
                    }
                }
            }

            // --- FIX: Always define em_status and start_date before use ---
            $em_status = 'Not Verify';
            $start_date = date('Y-m-d H:i:s');

            // --- FIX: Always define time_safety_margin, rows_since_time_check, last_time_check, max_allowed_time before use ---
            $time_safety_margin = 10; // seconds reserved to avoid hitting max_execution_time
            $rows_since_time_check = 0;
            $last_time_check = microtime(true);
            $max_allowed_time = $script_start_time + $time_limit - $time_safety_margin;

            $batchRows = [];
            $save_data_sql_base = "INSERT IGNORE INTO evp_email_list (csv_file_name, $dataC email_name, email_status, mail_box, create_date, user_id, meta) VALUES";
            $save = 0; // Ensure $save is always defined, even if no emails are processed

            while ($processed < $max_rows_to_process && ($line = fgets($handle)) !== false) {
                $rows_since_time_check++;
                if ($rows_since_time_check >= 5000) {
                    if (microtime(true) > $max_allowed_time) {
                        $log->system_debug("Approaching PHP execution time limit after processing $processed rows, stopping batch early");
                        break;
                    }
                    $last_time_check = microtime(true);
                    $rows_since_time_check = 0;
                }
                if ($encoding_detected !== 'UTF-8') {
                    $line = mb_convert_encoding($line, 'UTF-8', $encoding_detected);
                }
                $row = str_getcsv($line, ",", "\"", "\\");
                if (!$row || count($row) < 1) {
                    $skipped++;
                    continue;
                }
                $dataC = '';
                if (!empty($cc_data)) {
                    foreach ($cc_data as $key => $val) {
                        if ($key !== 'email_name') {
                            $dataC .= "$key, ";
                        }
                    }
                }
                if (isset($row[$email_name_val])) {
                    $emails = explode(",", $row[$email_name_val]);
                    foreach ($emails as $email) {
                        $email = strtolower(trim(clear_quotes($email)));
                        if (!strlen($email)) {
                            $skipped++;
                            continue;
                        }
                        if (!preg_match('/^[^\s@]+@[^\s@]+\.[^\s@]+$/', $email)) {
                            $skipped++;
                            continue;
                        }
                        if (!filter_var($email, FILTER_VALIDATE_EMAIL)) {
                            $skipped++;
                            continue;
                        }
                        $all_valid_emails[] = $email;
                        if (!isset($email_to_row[$email])) {
                            $email_to_row[$email] = $row; // Only first occurrence
                        }
                    }
                } else {
                    $log->system_debug("Skipping row: missing expected email_name_val index ($email_name_val)");
                    $skipped++;
                }
                $rows[] = $row;
                $processed++;
                $current_line++;
            }
            $unique_emails = array_keys($email_to_row);
            $duplicate = count($all_valid_emails) - count($unique_emails);
            $total = count($all_valid_emails) + $skipped;

            $batchRows = [];
            $save_data_sql_base = "INSERT IGNORE INTO evp_email_list (csv_file_name, $dataC email_name, email_status, mail_box, create_date, user_id, meta) VALUES";

            foreach ($unique_emails as $email) {
                $row = $email_to_row[$email];
                $dataV = '';
                if (!empty($cc_data)) {
                    foreach ($cc_data as $key => $val) {
                        if ($key !== 'email_name') {
                            $value = null;
                            if (isset($row[$val])) {
                                $value = clear_quotes($row[$val]);
                                $value = ($value !== null) ? mysqli_real_escape_string($db->link, $value) : '';
                            }
                            $dataV .= "'" . $value . "', ";
                        }
                    }
                }
                $csv_name_esc = mysqli_real_escape_string($db->link, $csv_name);
                $em_status_esc = mysqli_real_escape_string($db->link, $em_status);
                $mail_box_esc = mysqli_real_escape_string($db->link, (string) $mail_box);
                $start_date_esc = mysqli_real_escape_string($db->link, $start_date);
                $user_id_esc = mysqli_real_escape_string($db->link, $user_id);
                $meta_esc = mysqli_real_escape_string($db->link, $email_meta);
                $batchRows[] = "('$csv_name_esc', $dataV '$email', '$em_status_esc', '$mail_box_esc', '$start_date_esc', '$user_id_esc', '$meta_esc')";
                $save++;
            }

            // Insert the batch of emails using optimized insertion methods
            if (count($batchRows) > 0) {
                $insert_start_time = microtime(true);
                $processed_chunks = 0;
                $rows_inserted = 0;
                $remaining_time = max(5, $time_limit - (microtime(true) - $script_start_time) - $time_safety_margin);

                $log->system_debug("Beginning database insertion with approximately {$remaining_time}s remaining");

                // Calculate optimal chunk size based on batch size and remaining time
                $optimal_chunk_size = min(5000, max(1000, floor($batchSize / 10)));
                if (count($batchRows) < $optimal_chunk_size) {
                    $save_data_sql = $save_data_sql_base . "\n" . implode(",\n", $batchRows);
                    $log->system_debug("[SQL Preview] First 500 chars: " . substr($save_data_sql, 0, 500));
                    $result = $db->insert($save_data_sql);
                    $affected = $db->link->affected_rows;
                    if ($result) {
                        $max_exec_time = round(microtime(true) - $script_start_time, 4);
                        $log->system_debug("Inserted batch of " . count($batchRows) . " records for CSV: $csv_name | Max script run time: $max_exec_time s | Rows affected: $affected");
                        $totalInserted += count($batchRows);
                    } else {
                        $log->system_debug("Failed to insert batch for CSV: $csv_name. Error: " . mysqli_error($db->link));
                        $log->system_debug("[FAILED SQL] First 500 chars: " . substr($save_data_sql, 0, 500));
                    }
                } else {
                    $chunks = array_chunk($batchRows, $optimal_chunk_size);
                    $total_chunks = count($chunks);
                    foreach ($chunks as $index => $chunk) {
                        $time_now = microtime(true);
                        $elapsed_insert_time = $time_now - $script_start_time;
                        // Only break if we are truly close to the time limit (e.g., < 10s left)
                        if ($time_limit - $elapsed_insert_time < 10) {
                            $log->system_debug("Approaching PHP execution time limit during DB insert (" . round($elapsed_insert_time, 2) . "s), breaking insert loop early at chunk $index");
                            break;
                        }
                        $save_data_sql = $save_data_sql_base . "\n" . implode(",\n", $chunk);
                        $log->system_debug("[SQL Preview] First 500 chars: " . substr($save_data_sql, 0, 500));
                        $result = $db->insert($save_data_sql);
                        $affected = $db->link->affected_rows;
                        if ($result) {
                            $rows_inserted += count($chunk);
                            $totalInserted += count($chunk);
                            $log->system_debug("Inserted chunk of " . count($chunk) . " records for CSV: $csv_name | Rows affected: $affected");
                        } else {
                            $log->system_debug("Failed to insert chunk for CSV: $csv_name. Error: " . mysqli_error($db->link));
                            $log->system_debug("[FAILED SQL] First 500 chars: " . substr($save_data_sql, 0, 500));
                            break;
                        }
                        $processed_chunks++;
                        if ($processed_chunks % 10 == 0) {
                            $db->link->query("COMMIT");
                            $db->link->query("START TRANSACTION");
                            $current_progress = ($current_line / $total_lines) * 100;
                            $meta['progress_percentage'] = min(100, round($current_progress, 2));
                            $meta_json = json_encode($meta);
                            $db->update("UPDATE evp_task SET meta = '" . mysqli_real_escape_string($db->link, $meta_json) . "' WHERE csv_name = '$csv_name' AND user_id = '$user_id'");
                            $log->system_debug("Intermediate commit: $rows_inserted rows inserted in $processed_chunks chunks so far");
                        }
                    }
                    $max_exec_time = round(microtime(true) - $script_start_time, 4);
                    $log->system_debug("Inserted $totalInserted records in $processed_chunks/$total_chunks chunks for CSV: $csv_name | Max script run time: $max_exec_time s");
                }
                $commit_result = $db->link->commit();
                if ($commit_result) {
                    $log->system_debug("Transaction committed successfully for CSV: $csv_name");
                } else {
                    $log->system_debug("Transaction commit FAILED for CSV: $csv_name. Error: " . mysqli_error($db->link));
                }
            }
            $db->link->query("COMMIT");

            // Update total inserted (used for metadata updates)
            $save = $totalInserted;            // The meta variable now preserves all existing data, no need to reload              // Update metadata with more information
            $meta['import_offset'] = $current_line;
            $meta['detected_encoding'] = $encoding_detected;
            $meta['total_lines'] = $total_lines;

            // Add progress tracking based on actual total lines
            $completion_percentage = ($total_lines > 0)
                ? round(($current_line / $total_lines) * 100, 2)
                : 0;

            // Cap at 100% to avoid confusion
            $completion_percentage = min(100, $completion_percentage);

            $meta['progress_percentage'] = $completion_percentage;
            $meta['last_processed_time'] = time();
            $meta['batch_processing_time'] = round(microtime(true) - $script_start_time, 4);
            $meta['processed_rows'] = $current_line;

            // Track performance metrics
            $memory_usage_mb = round(memory_get_usage(true) / 1024 / 1024, 2);
            $meta['memory_usage_mb'] = $memory_usage_mb;

            // Adjust next batch size based on performance if needed
            if ($memory_usage_mb > ($memory_mb * 0.8)) {
                // We're using too much memory, reduce batch size for next run
                $log->system_debug("Memory usage high ($memory_usage_mb MB), consider reducing batch size for future runs");
            }

            // After batch processing, update cumulative counters
            $meta['cumulative_imported'] += $save;
            $meta['cumulative_duplicates'] += $duplicate;
            $meta['cumulative_skipped'] += $skipped;
            $meta['cumulative_total'] += $total;

            // --- CUMULATIVE COUNTERS: update meta with cumulative totals ---
            $meta['imported_total'] = isset($meta['imported_total']) ? $meta['imported_total'] + $save : $save;
            $meta['duplicate_total'] = isset($meta['duplicate_total']) ? $meta['duplicate_total'] + $duplicate : $duplicate;
            $meta['skipped_total'] = isset($meta['skipped_total']) ? $meta['skipped_total'] + $skipped : $skipped;
            $meta['parsed_total'] = isset($meta['parsed_total']) ? $meta['parsed_total'] + $total : $total;

            $meta_json = json_encode($meta);
            $db->update("UPDATE evp_task SET meta = '" . mysqli_real_escape_string($db->link, $meta_json) . "' WHERE csv_name = '$csv_name' AND user_id = '$user_id'");
            $log->system_debug("[RESUME CHECK] import_offset saved as $current_line for CSV: $csv_name. Progress: {$completion_percentage}%");
            // --- RELEASE LOCK FOR NEXT BATCH ---
            $meta['import_locked'] = false;
            $meta['import_pid'] = null;
            $meta['import_locked_until'] = null;
            $meta_json = json_encode($meta);
            $db->update("UPDATE evp_task SET meta = '" . mysqli_real_escape_string($db->link, $meta_json) . "' WHERE csv_name = '$csv_name' AND user_id = '$user_id'");
            $log->system_debug('[LOCK RELEASE] Lock released for CSV: ' . $csv_name);
            // If not EOF or if we've processed total_lines, determine if we should continue or complete
            if (!feof($handle) && ($total_lines == 0 || $current_line < $total_lines)) {
                $log->system_debug("Batch processed, import_offset updated to $current_line. " . $totalInserted . " emails inserted. Progress: {$completion_percentage}% complete. Exiting for next run.");
                fclose($handle);

                // Aggressively free memory before exit
                $all_emails = $rows = $batchRows = $valid_emails = $unique_emails = null;
                unset($all_emails, $rows, $batchRows, $valid_emails, $unique_emails);

                // Force garbage collection if available
                if (function_exists('gc_collect_cycles')) {
                    gc_collect_cycles();
                }

                // Log memory usage after cleanup
                $final_memory = round(memory_get_usage(true) / 1024 / 1024, 2);
                $log->system_debug("Final memory usage after cleanup: $final_memory MB");

                exit();
            }

            // If we've reached EOF or processed all known lines
            $log->system_debug("Reached end of file or processed all $total_lines lines.");
            fclose($handle);

            // --- Use cumulative totals for summary and metadata ---
            $imported_total = isset($meta['imported_total']) ? $meta['imported_total'] : $save;
            $duplicate_total = isset($meta['duplicate_total']) ? $meta['duplicate_total'] : $duplicate;
            $skipped_total = isset($meta['skipped_total']) ? $meta['skipped_total'] : $skipped;
            $parsed_total = isset($meta['parsed_total']) ? $meta['parsed_total'] : $total;
            $log->system_debug("CSV Import Summary for $csv_name: Imported (unique, committed): $imported_total, Duplicates: $duplicate_total, Skipped/Invalid: $skipped_total, Total parsed: $parsed_total");

            // Ensure database connection before final updates
            ensure_mysql_connection($db, $log);
            // --- Update metadata with cumulative imported total ---
            $listing_metadata_sql = "UPDATE evp_email_list_metadata SET total = '$imported_total', scan_status = 'pending' WHERE csv_file_name = '$csv_name' AND user_id = '$user_id'";
            if ($db->insert($listing_metadata_sql)) {
                $log->system_debug("Inserted metadata for CSV: $csv_name");
            } else {
                $log->system_debug("Failed to insert metadata for CSV: $csv_name. Error: " . mysqli_error($db->link));
            }

            $update_task_sql = "UPDATE evp_task SET status = 'completed' WHERE csv_name = '$csv_name' AND user_id = '$user_id'";
            if ($db->update($update_task_sql)) {
                $log->system_debug("Updated task status to completed for CSV: $csv_name");

                // Check if the task status has been updated to 'completed'
                $check_task_sql = "SELECT status FROM evp_task WHERE csv_name = '$csv_name' AND user_id = '$user_id' AND status = 'completed'";
                $check_task_result = $db->select($check_task_sql);
                if ($check_task_result && $check_task_result->num_rows > 0) {
                    // Delete the completed task
                    $delete_task_sql = "DELETE FROM evp_task WHERE csv_name = '$csv_name' AND user_id = '$user_id' AND status = 'completed'";
                    if ($db->delete($delete_task_sql)) {
                        $log->system_debug("Deleted completed task for CSV: $csv_name");

                        // Send email notification to the user with cumulative totals
                        $mail = new SendMail();
                        $mail->user_csv_import_notify($user_id, $csv_name, $imported_total, $duplicate_total, $skipped_total);

                    } else {
                        $log->system_debug("Failed to delete completed task for CSV: $csv_name. Error: " . mysqli_error($db->link));
                    }
                } else {
                    $log->system_debug("Task status not updated to completed for CSV: $csv_name");
                }
            } else {
                $log->system_debug("Failed to update task status for CSV: $csv_name. Error: " . mysqli_error($db->link));
            }

            $log->system_debug("FINAL CSV Import Results for $csv_name: Imported (unique, committed): $imported_total, Duplicates: $duplicate_total, Skipped/Invalid: $skipped_total, Total parsed: $parsed_total");

            unlink($file_path);
        } else {
            $log->system_debug("Importer Tasks: Total Emails Not Empty.");
            exit();
        }
    }
} else {
    $log->system_debug("No Importer tasks found.");
    exit();
}

function sanitize_text($data)
{
    global $db;
    return mysqli_real_escape_string($db->link, htmlspecialchars(stripslashes(trim($data))));
}

// Define the clear_quotes function
function clear_quotes($strClear)
{
    return preg_replace('/^(\'(.*)\'|"(.*)")$/', '$2$3', $strClear);
}

$log->system_debug("Import Scheduler finished.");
?>