Massively enhance and improve this code
use dashmap::DashSet;
use rayon::prelude::*;
use std::{
    collections::VecDeque,
    fs,
    path::Path,
    sync::{atomic::{AtomicUsize, Ordering}, Arc, Mutex},
    time::{Duration, Instant},
};
use chrono::{DateTime, Local, Utc};
use indicatif::{ProgressBar, ProgressStyle};
use rand::{distributions::Alphanumeric, thread_rng, Rng};
use reqwest::{Client, StatusCode};
use thiserror::Error;
use tokio::sync::Mutex as TokioMutex; 
use tokio::{io::AsyncWriteExt, sync::{Semaphore, mpsc}};
use thousands::Separable;
use sha2::{Sha256, Digest};
use mimalloc::MiMalloc;

#[global_allocator]
static GLOBAL: MiMalloc = MiMalloc;

const BASE_URL: &str = "https://litter.catbox.moe/";
const ALLOWED_EXTENSIONS: &[&str] = &["png"];
const CONCURRENT_REQUESTS: usize = 7500;
const TOTAL_REQUESTS: usize = 50_000_000;
const URL_LENGTH: usize = 6;
const OUTPUT_DIR: &str = "output";
const NUM_WORKERS: usize = 7500;
const CONNECTIONS_PER_WORKER: usize = 24;

#[derive(Error, Debug)]
enum MyError {
    #[error("Request error: {0}")]
    Reqwest(#[from] reqwest::Error),

    #[error("IO error: {0}")]
    Io(#[from] std::io::Error),

    #[error("Serialization error: {0}")]
    Serde(#[from] serde_json::Error),

    #[error("Join error: {0}")]
    Join(#[from] tokio::task::JoinError),

    #[error("Anyhow error: {0}")]
    Anyhow(#[from] anyhow::Error),
}
struct ConnectionPool {
    connections: Vec<Client>,
}

impl ConnectionPool {
    fn new(size: usize) -> Self {
        let connections = (0..size)
            .map(|_| {
                Client::builder()
                    .pool_max_idle_per_host(100)
                    .pool_idle_timeout(Some(Duration::from_secs(30)))
                    .tcp_keepalive(Some(Duration::from_secs(60)))
                    .build()
                    .unwrap()
            })
            .collect();
        ConnectionPool { connections }
    }

    fn get_connection(&self) -> &Client {
        &self.connections[thread_rng().gen_range(0..self.connections.len())]
    }
}

#[tokio::main(flavor = "multi_thread", worker_threads = 32)]
async fn main() -> Result<(), MyError> {
    fs::create_dir_all(OUTPUT_DIR)?;
    sort_files_by_type().await?;
    process_files_in_batches()?;

    let url_generation_start_time = Instant::now();
    let urls = Arc::new(Mutex::new(generate_urls(TOTAL_REQUESTS)));
    let url_generation_time = url_generation_start_time.elapsed();

    spdlog::info!("Generated {} URLs in {:?}", TOTAL_REQUESTS.separate_with_commas(), url_generation_time);
    let found_urls = Arc::new(DashSet::with_capacity(TOTAL_REQUESTS)); // Changed to DashSet
    let total_found = Arc::new(AtomicUsize::new(0));
    let semaphore = Arc::new(Semaphore::new(CONCURRENT_REQUESTS));
    let overall_progress_bar = create_progress_bar("Overall Progress", TOTAL_REQUESTS);

    let (task_sender, task_receiver) = mpsc::channel(TOTAL_REQUESTS);
    let task_receiver = Arc::new(TokioMutex::new(task_receiver));

    let worker_handles: Vec<_> = (0..NUM_WORKERS)
        .map(|_| {
            let urls = Arc::clone(&urls);
            let found_urls = Arc::clone(&found_urls);
            let total_found = Arc::clone(&total_found);
            let semaphore = Arc::clone(&semaphore);
            let overall_progress_bar = overall_progress_bar.clone();
            let task_receiver = Arc::clone(&task_receiver);
            let connection_pool = Arc::new(ConnectionPool::new(CONNECTIONS_PER_WORKER));

            tokio::spawn(async move {
                worker_loop(
                    urls,
                    found_urls,
                    total_found,
                    semaphore,
                    overall_progress_bar,
                    task_receiver,
                    connection_pool,
                )
                .await
            })
        })
        .collect();

    let request_start_time = Instant::now();

    // Initial task distribution
    for _ in 0..TOTAL_REQUESTS {
        task_sender.send(()).await.unwrap();
    }

    // Wait for all workers to complete
    for handle in worker_handles {
        handle.await??;
    }

    let request_time = request_start_time.elapsed();

    overall_progress_bar.finish_with_message("Finished");
    spdlog::info!(
        "Found {} unique urls in {:?}. URL generation took {:?}",
        total_found.load(Ordering::Relaxed),
        request_time,
        url_generation_time
    );
    Ok(())
}
async fn worker_loop(
    urls: Arc<Mutex<VecDeque<String>>>,
    found_urls: Arc<DashSet<String>>, // Changed to DashSet
    total_found: Arc<AtomicUsize>,
    semaphore: Arc<Semaphore>,
    overall_progress_bar: ProgressBar,
    task_receiver: Arc<TokioMutex<mpsc::Receiver<()>>>,
    connection_pool: Arc<ConnectionPool>,
) -> Result<(), MyError> {
    loop {
        let should_continue = {
            let mut receiver = task_receiver.lock().await;
            match receiver.try_recv() {
                Ok(_) => true,
                Err(mpsc::error::TryRecvError::Empty) => {
                    tokio::time::sleep(Duration::from_millis(10)).await;
                    continue;
                }
                Err(mpsc::error::TryRecvError::Disconnected) => false,
            }
        };

        if !should_continue {
            break;
        }

        let url = {
            let mut urls = urls.lock().unwrap();
            urls.pop_front()
        };

        if let Some(url) = url {
            let _permit = semaphore.acquire().await.unwrap();
            if let Err(_) = process_url(&connection_pool, &url, &found_urls, &total_found).await { }
            overall_progress_bar.inc(1);
        }
    }
    Ok(())
}

async fn process_url(
    connection_pool: &ConnectionPool,
    url: &str,
    found_urls: &Arc<DashSet<String>>, // Changed to DashSet
    total_found: &Arc<AtomicUsize>,
) -> Result<(), MyError> {
    if found_urls.contains(url) {
        return Ok(());
    }

    let client = connection_pool.get_connection();
    let response = client.head(url).send().await?;

    if response.status() == StatusCode::OK {
        if let Some(last_modified) = extract_last_modified(&response) {
            let content_length: u64 = response
                .headers()
                .get("content-length")
                .and_then(|value| value.to_str().ok())
                .and_then(|len_str| len_str.parse().ok())
                .unwrap_or(0);

            if found_urls.insert(url.to_string()) { // Changed: insert returns true if the value was newly inserted
                if let Err(err) = download_file(client, url, last_modified, content_length).await {
                    spdlog::error!("Error downloading {}: {}", url, err);
                    return Err(err);
                }
                total_found.fetch_add(1, Ordering::Relaxed);
            }
        }
    }
    Ok(())
}

fn process_files_in_batches() -> Result<(), MyError> {
    let batch_size = 1000;
    let entries: Vec<_> = fs::read_dir(OUTPUT_DIR)?.collect::<Result<_, _>>()?;
    let total_files = entries.len();
    let progress_bar = create_progress_bar("Processing files", total_files);

    for chunk in entries.chunks(batch_size) {
        chunk.par_iter().try_for_each(|entry| -> Result<(), MyError> {
            let path = entry.path();
            if path.is_file() {
                let hash = calculate_file_hash(&path)?;
                spdlog::info!("File: {:?}, Hash: {}", path, hash);

                if let Some(kind) = infer::get_from_path(&path)? {
                    let type_dir = Path::new(OUTPUT_DIR).join(kind.mime_type());
                    fs::create_dir_all(&type_dir)?;
                    let new_path = type_dir.join(path.file_name().unwrap());
                    fs::rename(&path, &new_path)?;
                }
            }
            progress_bar.inc(1);
            Ok(())
        })?;
    }

    progress_bar.finish_with_message("File processing completed");
    Ok(())
}


fn generate_urls(total_urls: usize) -> VecDeque<String> {
    (0..total_urls)
        .into_par_iter()
        .map(|_| {
            let random_string: String = thread_rng()
                .sample_iter(&Alphanumeric)
                .take(URL_LENGTH)
                .map(|c| c.to_ascii_lowercase() as char)
                .collect();
            format!("{}{}.{}", BASE_URL, random_string, ALLOWED_EXTENSIONS[0])
        })
        .collect()
}

fn extract_last_modified(response: &reqwest::Response) -> Option<DateTime<Utc>> {
    response
        .headers()
        .get("last-modified")
        .and_then(|value| value.to_str().ok())
        .and_then(|date_str| DateTime::parse_from_rfc2822(date_str).ok())
        .map(|dt| dt.into())
}

fn create_progress_bar(prefix_message: &str, total_requests: usize) -> ProgressBar {
    let pb = ProgressBar::new(total_requests as u64);
    pb.set_style(
        ProgressStyle::default_bar()
            .template("{prefix:.bold.bright_cyan} {spinner:.green} [{elapsed_precise}] [{bar:40.gradient(red,yellow,green)}] {pos:>7}/{len:7} ({eta})")
            .unwrap()
            .progress_chars("█▓▒░ ")
    );
    pb.set_prefix(prefix_message.to_string());
    pb
}
async fn download_file(
    client: &Client,
    url: &str,
    last_modified: DateTime<Utc>,
    total_bytes: u64,
) -> Result<String, MyError> {
    let formatted_time = last_modified.with_timezone(&Local).format("%Y-%m-%d %H:%M:%S");
    spdlog::info!(
        "Found: [{}] - {} - {}",
        url,
        formatted_time,
        human_readable_size(total_bytes)
    );

    let response = client.get(url).send().await?;

    if !response.status().is_success() {
        return Err(MyError::Reqwest(reqwest::Error::from(
            response.error_for_status().unwrap_err(),
        )));
    }

    let file_name = Path::new(url)
        .file_name()
        .unwrap()
        .to_string_lossy()
        .to_string();

    let file_path = Path::new(OUTPUT_DIR).join(file_name);
    let mut file = tokio::fs::File::create(&file_path).await?;

    let bytes = response.bytes().await?;
    file.write_all(&bytes).await?;
    file.flush().await?;

    sort_files_by_type().await?;
    process_files_in_batches()?;

    Ok(file_path.to_str().unwrap().to_string())
}


fn human_readable_size(size: u64) -> String {
    let mut size = size as f64;
    let units = ["bytes", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB"];
    let mut i = 0;
    while size >= 1024.0 {
        size /= 1024.0;
        i += 1;
    }
    format!("{:.2} {}", size, units[i])
}

async fn sort_files_by_type() -> Result<(), MyError> {
    let entries: Vec<_> = fs::read_dir(OUTPUT_DIR)?.collect::<Result<_, _>>()?;
    
    entries.par_iter().try_for_each(|entry| {
        let path = entry.path();
        if path.is_file() {
            if let Some(kind) = infer::get_from_path(&path)? {
                let type_dir = Path::new(OUTPUT_DIR).join(kind.mime_type());
                fs::create_dir_all(&type_dir)?;
                let new_path = type_dir.join(path.file_name().unwrap());
                fs::rename(&path, &new_path)?;
            }
        }
        Ok::<(), MyError>(())
    })?;
    
    Ok(())
}

fn calculate_file_hash(path: &Path) -> Result<String, MyError> {
    let mut file = std::fs::File::open(path)?;
    let mut hasher = Sha256::new();
    std::io::copy(&mut file, &mut hasher)?;
    let hash = hasher.finalize();
    Ok(format!("{:x}", hash))
}