Function csv_worker

src/csv/worker.rs:37–165

CSV Worker function that processes phone numbers from the queue

(
    work_rx: async_channel::Receiver<WorkerMessage>,
    result_tx: async_channel::Sender<ResultMessage>,
    counters: Arc<Counters>,
    subnet: String,
    lookup_type: LookupType
)

// CSV Worker function that processes phone numbers from the queue
pub async fn csv_worker(
    work_rx: async_channel::Receiver<WorkerMessage>,
    result_tx: async_channel::Sender<ResultMessage>,
    counters: Arc<Counters>,
    subnet: String,
    lookup_type: LookupType
) {
    let mut client = crate::utils::create_client(Some(&subnet), "");
    let mut last_auth_refresh = std::time::Instant::now();
    let auth_refresh_interval = Duration::from_secs(8 * 60 * 60); // Refresh auth every 8 hours
    
    while let Ok(message) = work_rx.recv().await {
        match message {
            WorkerMessage::CheckPhone { record_id, phone, identifier: _identifier, first_name, last_name, pending_counter } => {
                // Get a reference to the counter for decrementing when done
                let decrement_counter = || {
                    if let Some(counter) = &pending_counter {
                        counter.fetch_sub(1, Ordering::SeqCst);
                    }
                };
                
                // Check if we need to refresh authentication
                if last_auth_refresh.elapsed() >= auth_refresh_interval {
                    if let Ok(_) = auth::get_auth_credentials().await {
                        last_auth_refresh = std::time::Instant::now();
                    }
                }
                
                // Skip processing for completion marker
                if phone.starts_with("COMPLETION_MARKER_") {
                    decrement_counter();
                    continue;
                }
                
                // Process the phone number
                counters.requests.fetch_add(1, Ordering::Relaxed);
                
                // Validate phone number
                let parsed_number = match format!("+{}", phone).parse::<phonenumber::PhoneNumber>() {
                    Ok(number) => number,
                    Err(_) => {
                        counters.success.fetch_add(1, Ordering::Relaxed);
                        decrement_counter();
                        continue;
                    }
                };
                
                if !phonenumber::is_valid(&parsed_number) {
                    counters.success.fetch_add(1, Ordering::Relaxed);
                    decrement_counter();
                    continue;
                }
                
                // Similar to the original worker function but streamlined for CSV mode
                for attempt in 0..3 { // Limited retries
                    let lookup_result = match lookup_type {
                        LookupType::Js => js::lookup(&client, &phone, &first_name, &last_name).await,
                        LookupType::NoJS => nojs::lookup(&client, &phone, &first_name, &last_name).await,