use crate::cli::Commands;
use crate::commands;
use crate::config::CONFIG;
use crate::core::error::BcmrError;
use crate::core::remote::{self, parse_remote_path, RemotePath};
use crate::ui::progress::{self, ProgressRenderer};
use crate::ui::utils::format_bytes;
use anyhow::{bail, Result};
use parking_lot::Mutex;
use std::path::PathBuf;
use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};
use std::sync::Arc;
use tokio::signal::ctrl_c;
use tokio::time::Duration;

pub fn is_plain_mode(args: &Commands) -> bool {
    args.is_tui_mode() || CONFIG.progress.style.eq_ignore_ascii_case("plain")
}

pub struct ProgressRunner {
    progress: Arc<Mutex<Box<dyn ProgressRenderer>>>,
    ticker_handle: tokio::task::JoinHandle<()>,
}

impl ProgressRunner {
    pub fn new(total_size: u64, plain: bool, silent: bool) -> std::io::Result<Self> {
        let renderer = progress::create_renderer(total_size, plain, silent)?;
        let progress = Arc::new(Mutex::new(renderer));

        let ticker = Arc::clone(&progress);
        let ticker_handle = tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_millis(100));
            loop {
                interval.tick().await;
                ticker.lock().tick();
            }
        });

        let signal = Arc::clone(&progress);
        tokio::spawn(async move {
            if let Ok(()) = ctrl_c().await {
                commands::copy::cleanup_partial_files();
                let _ = signal.lock().finish();
                std::process::exit(130);
            }
        });

        Ok(Self {
            progress,
            ticker_handle,
        })
    }

    pub fn progress(&self) -> &Arc<Mutex<Box<dyn ProgressRenderer>>> {
        &self.progress
    }

    pub fn inc_callback(&self) -> impl Fn(u64) + Send + Sync + Clone + 'static {
        let p = Arc::clone(&self.progress);
        move |n| p.lock().inc_current(n)
    }

    pub fn skip_callback(&self) -> impl Fn(u64) + Send + Sync + Clone + 'static {
        let p = Arc::clone(&self.progress);
        move |n| p.lock().inc_skipped(n)
    }

    pub fn file_callback(&self) -> impl Fn(&str, u64) + Send + Sync + Clone + 'static {
        let p = Arc::clone(&self.progress);
        move |name, size| p.lock().set_current_file(name, size)
    }

    pub fn set_parallel_mode(&self, worker_count: usize) {
        self.progress.lock().set_parallel_mode(worker_count);
    }

    pub fn finish_ok(self) -> Result<()> {
        self.ticker_handle.abort();
        self.progress.lock().finish()?;
        Ok(())
    }

    pub fn finish_err(self, msg: String) -> Result<()> {
        self.ticker_handle.abort();
        let _ = self.progress.lock().finish();
        bail!("{}", msg);
    }
}

pub struct TransferItem {
    pub local_path: PathBuf,
    pub remote: RemotePath,
    pub size: u64,
    pub is_upload: bool,
}

const COMPRESSED_EXTENSIONS: &[&str] = &[
    "gz", "bz2", "xz", "zst", "lz4", "zip", "rar", "7z", "jpg", "jpeg", "png", "gif", "webp",
    "avif", "heic", "mp4", "mkv", "avi", "mov", "webm", "mp3", "aac", "ogg", "flac", "opus", "pdf",
    "docx", "xlsx", "pptx", "dmg", "iso", "whl", "egg",
];

fn should_compress(items: &[TransferItem]) -> bool {
    let (mut compressible, mut total) = (0u64, 0u64);
    for item in items {
        total += item.size;
        let ext = std::path::Path::new(if item.is_upload {
            item.local_path.to_str().unwrap_or("")
        } else {
            &item.remote.path
        })
        .extension()
        .unwrap_or_default()
        .to_string_lossy()
        .to_lowercase();
        if !COMPRESSED_EXTENSIONS.contains(&ext.as_str()) {
            compressible += item.size;
        }
    }
    total > 0 && compressible * 100 / total > 30
}

async fn run_parallel_transfers(
    items: Vec<TransferItem>,
    parallel: usize,
    progress: &Arc<Mutex<Box<dyn ProgressRenderer>>>,
    opts: &remote::RemoteTransferOptions,
) -> Result<(), BcmrError> {
    let semaphore = Arc::new(tokio::sync::Semaphore::new(parallel));
    let slot_pool: Arc<Mutex<Vec<usize>>> = Arc::new(Mutex::new((0..parallel).rev().collect()));
    let errors: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
    let mut handles = Vec::new();

    for item in items {
        let sem = Arc::clone(&semaphore);
        let pool = Arc::clone(&slot_pool);
        let prog = Arc::clone(progress);
        let errs = Arc::clone(&errors);
        let task_opts = opts.clone();

        handles.push(tokio::spawn(async move {
            let _permit = sem.acquire().await.unwrap();
            let slot = match pool.lock().pop() {
                Some(s) => s,
                None => {
                    errs.lock().push("no available worker slot".to_string());
                    return;
                }
            };

            let file_name = if item.is_upload {
                item.local_path
                    .file_name()
                    .unwrap_or_default()
                    .to_string_lossy()
                    .to_string()
            } else {
                item.remote
                    .path
                    .rsplit('/')
                    .next()
                    .unwrap_or(&item.remote.path)
                    .to_string()
            };

            let worker_bytes = Arc::new(AtomicU64::new(0));
            let fsize = item.size;

            let wb = Arc::clone(&worker_bytes);
            let p = Arc::clone(&prog);
            let fname = file_name.clone();
            let progress_cb = move |n: u64| {
                let total = wb.fetch_add(n, AtomicOrdering::Relaxed) + n;
                let mut guard = p.lock();
                guard.inc_current(n);
                guard.update_worker(slot, &fname, fsize, total);
            };

            let p2 = Arc::clone(&prog);
            let noop_file_cb = move |name: &str, size: u64| {
                p2.lock().update_worker(slot, name, size, 0);
            };

            let p_skip = Arc::clone(&prog);
            let skip_cb = move |n: u64| {
                p_skip.lock().inc_skipped(n);
            };

            let result = if item.is_upload {
                remote::upload_file(
                    &item.local_path,
                    &item.remote,
                    &progress_cb,
                    &skip_cb,
                    &noop_file_cb,
                    &task_opts,
                )
                .await
            } else {
                remote::download_file(
                    &item.remote,
                    &item.local_path,
                    &progress_cb,
                    &skip_cb,
                    &noop_file_cb,
                    item.size,
                    &task_opts,
                )
                .await
            };

            if let Err(e) = result {
                errs.lock().push(e.to_string());
            }

            prog.lock().finish_worker(slot);
            pool.lock().push(slot);
        }));
    }

    for handle in handles {
        handle.await?;
    }

    let errs = errors.lock();
    if !errs.is_empty() {
        return Err(BcmrError::InvalidInput(errs.join("; ")));
    }
    Ok(())
}

fn collect_upload_files(
    local_src: &std::path::Path,
    remote_base: &RemotePath,
    excludes: &[regex::Regex],
) -> Result<Vec<TransferItem>> {
    use crate::core::traversal;

    let mut items = Vec::new();

    for entry in traversal::walk(local_src, true, false, 1, excludes) {
        let entry = entry?;
        let path = entry.path();
        let rel = path.strip_prefix(local_src)?;

        if path.is_file() {
            items.push(TransferItem {
                local_path: path.to_path_buf(),
                remote: remote_base.join(&rel.to_string_lossy()),
                size: entry.metadata()?.len(),
                is_upload: true,
            });
        }
    }

    Ok(items)
}

fn resolve_upload_remote(
    src: &std::path::Path,
    rdest: &RemotePath,
    multi_source: bool,
) -> RemotePath {
    if multi_source || rdest.path == "." {
        rdest.join(&src.file_name().unwrap_or_default().to_string_lossy())
    } else {
        rdest.clone()
    }
}

pub async fn handle_remote_copy(
    args: &Commands,
    sources: &[std::path::PathBuf],
    dest: &std::path::Path,
    excludes: &[regex::Regex],
) -> Result<()> {
    let dest_str = dest.to_string_lossy();
    let remote_dest = parse_remote_path(&dest_str);
    let is_upload = remote_dest.is_some();

    let compression_mode = CONFIG.scp.compression.to_lowercase();
    let compress = match compression_mode.as_str() {
        "force" => true,
        "off" | "never" => false,
        _ => {
            if is_upload {
                let probe: Vec<TransferItem> = sources
                    .iter()
                    .filter_map(|s| {
                        if parse_remote_path(&s.to_string_lossy()).is_some() {
                            return None;
                        }
                        let size = s.metadata().ok()?.len();
                        Some(TransferItem {
                            local_path: s.clone(),
                            remote: RemotePath {
                                user: None,
                                host: String::new(),
                                path: String::new(),
                            },
                            size,
                            is_upload: true,
                        })
                    })
                    .collect();
                should_compress(&probe)
            } else {
                let probe: Vec<TransferItem> = sources
                    .iter()
                    .filter_map(|s| {
                        let rp = parse_remote_path(&s.to_string_lossy())?;
                        Some(TransferItem {
                            local_path: PathBuf::new(),
                            remote: rp,
                            size: 1,
                            is_upload: false,
                        })
                    })
                    .collect();
                should_compress(&probe)
            }
        }
    };
    if args.is_sync() {
        eprintln!("Warning: --sync has no effect on remote transfers");
    }

    remote::set_ssh_compression(compress);

    let check_target = if let Some(ref rd) = remote_dest {
        rd.clone()
    } else {
        let src_str = sources[0].to_string_lossy();
        parse_remote_path(&src_str).ok_or_else(|| anyhow::anyhow!("No remote path found"))?
    };
    remote::validate_ssh_connection(&check_target).await?;

    let parallel = args.get_parallel().unwrap_or(CONFIG.scp.parallel_transfers);

    if let Some(ref rdest) = remote_dest {
        handle_remote_upload(args, sources, rdest, parallel, excludes).await
    } else {
        handle_remote_download(args, sources, dest, parallel, excludes).await
    }
}

async fn handle_remote_upload(
    args: &Commands,
    sources: &[std::path::PathBuf],
    rdest: &RemotePath,
    parallel: usize,
    excludes: &[regex::Regex],
) -> Result<()> {
    let excludes = excludes.to_vec();
    let mut total_size = 0u64;
    for src in sources {
        if parse_remote_path(&src.to_string_lossy()).is_some() {
            bail!("Cannot copy between two remote hosts. Use local as intermediary.");
        }
        if src.is_file() {
            total_size += src.metadata()?.len();
        } else if src.is_dir() && args.is_recursive() {
            total_size +=
                commands::copy::get_total_size(std::slice::from_ref(src), true, args, &[]).await?;
        } else if src.is_dir() {
            bail!(
                "Source '{}' is a directory. Use -r flag for recursive copy.",
                src.display()
            );
        } else {
            bail!("Source '{}' not found", src.display());
        }
    }

    if args.is_dry_run() {
        println!(
            "Dry-run: would upload {} to {}",
            format_bytes(total_size as f64),
            rdest
        );
        for src in sources {
            if src.is_file() {
                println!(
                    "  {} -> {}",
                    src.display(),
                    resolve_upload_remote(src, rdest, sources.len() > 1)
                );
            } else if src.is_dir() && args.is_recursive() {
                let dir_remote = rdest.join(&src.file_name().unwrap_or_default().to_string_lossy());
                for item in collect_upload_files(src, &dir_remote, &excludes)? {
                    println!("  {} -> {}", item.local_path.display(), item.remote);
                }
            }
        }
        return Ok(());
    }

    let opts = remote::RemoteTransferOptions {
        preserve: args.is_preserve(),
        verify: args.is_verify(),
        resume: args.is_resume(),
        strict: args.is_strict(),
        append: args.is_append(),
    };

    let runner = ProgressRunner::new(total_size, is_plain_mode(args), false)?;
    runner.progress().lock().set_operation_type("Uploading");
    let multi_source = sources.len() > 1;

    if parallel > 1 {
        runner.set_parallel_mode(parallel);

        let mut items: Vec<TransferItem> = Vec::new();
        for src in sources {
            if src.is_file() {
                let file_remote = resolve_upload_remote(src, rdest, multi_source);
                items.push(TransferItem {
                    local_path: src.clone(),
                    remote: file_remote,
                    size: src.metadata()?.len(),
                    is_upload: true,
                });
            } else if src.is_dir() && args.is_recursive() {
                let dir_remote = rdest.join(&src.file_name().unwrap_or_default().to_string_lossy());
                remote::ensure_remote_tree(src, &dir_remote).await?;
                items.extend(collect_upload_files(src, &dir_remote, &excludes)?);
            }
        }

        run_parallel_transfers(items, parallel, runner.progress(), &opts)
            .await
            .map_err(|e| anyhow::anyhow!("{}", e))?;
    } else {
        for src in sources {
            if src.is_file() {
                let file_remote = resolve_upload_remote(src, rdest, multi_source);
                remote::upload_file(
                    src,
                    &file_remote,
                    &runner.inc_callback(),
                    &runner.skip_callback(),
                    &runner.file_callback(),
                    &opts,
                )
                .await?;
            } else if src.is_dir() && args.is_recursive() {
                let dir_remote = rdest.join(&src.file_name().unwrap_or_default().to_string_lossy());
                remote::upload_directory(
                    src,
                    &dir_remote,
                    &runner.inc_callback(),
                    &runner.skip_callback(),
                    &runner.file_callback(),
                    &excludes,
                    &opts,
                )
                .await?;
            }
        }
    }

    runner.finish_ok()
}

async fn handle_remote_download(
    args: &Commands,
    sources: &[std::path::PathBuf],
    dest_local: &std::path::Path,
    parallel: usize,
    excludes: &[regex::Regex],
) -> Result<()> {
    let excludes = excludes.to_vec();

    let mut remote_sources = Vec::new();
    for src in sources {
        let rsrc = parse_remote_path(&src.to_string_lossy()).ok_or_else(|| {
            anyhow::anyhow!("Mixed local/remote sources without remote destination")
        })?;
        let size = remote::remote_total_size(&rsrc, args.is_recursive()).await?;
        remote_sources.push((rsrc, size));
    }

    let total_size: u64 = remote_sources.iter().map(|(_, s)| *s).sum();

    if args.is_dry_run() {
        println!(
            "Dry-run: would download {} to {}",
            format_bytes(total_size as f64),
            dest_local.display()
        );
        for (rsrc, _) in &remote_sources {
            let info = remote::remote_stat(rsrc).await?;
            if info.is_dir && args.is_recursive() {
                let dir_name = rsrc.path.rsplit('/').next().unwrap_or(&rsrc.path);
                let local_dir = if dest_local.is_dir() {
                    dest_local.join(dir_name)
                } else {
                    dest_local.to_path_buf()
                };
                let entries = remote::remote_list_files(rsrc).await?;
                for (rel_path, _, is_dir_entry) in &entries {
                    if *is_dir_entry {
                        continue;
                    }
                    if crate::core::traversal::is_excluded(
                        std::path::Path::new(rel_path),
                        &excludes,
                    ) {
                        continue;
                    }
                    println!(
                        "  {} -> {}",
                        rsrc.join(rel_path),
                        local_dir.join(rel_path).display()
                    );
                }
            } else {
                let local_path = if dest_local.is_dir() {
                    dest_local.join(rsrc.path.rsplit('/').next().unwrap_or(&rsrc.path))
                } else {
                    dest_local.to_path_buf()
                };
                println!("  {} -> {}", rsrc, local_path.display());
            }
        }
        return Ok(());
    }

    let opts = remote::RemoteTransferOptions {
        preserve: args.is_preserve(),
        verify: args.is_verify(),
        resume: args.is_resume(),
        strict: args.is_strict(),
        append: args.is_append(),
    };

    let runner = ProgressRunner::new(total_size, is_plain_mode(args), false)?;
    runner.progress().lock().set_operation_type("Downloading");

    if parallel > 1 {
        runner.set_parallel_mode(parallel);

        let mut items: Vec<TransferItem> = Vec::new();
        for (rsrc, _) in &remote_sources {
            let info = remote::remote_stat(rsrc).await?;
            if info.is_dir {
                if !args.is_recursive() {
                    bail!("Remote source '{}' is a directory. Use -r flag.", rsrc);
                }
                let dir_name = rsrc.path.rsplit('/').next().unwrap_or(&rsrc.path);
                let local_dir = if dest_local.is_dir() {
                    dest_local.join(dir_name)
                } else {
                    dest_local.to_path_buf()
                };
                let entries = remote::remote_list_files(rsrc).await?;
                for (rel_path, _, is_dir_entry) in &entries {
                    if *is_dir_entry
                        && !crate::core::traversal::is_excluded(
                            std::path::Path::new(rel_path),
                            &excludes,
                        )
                    {
                        tokio::fs::create_dir_all(local_dir.join(rel_path)).await?;
                    }
                }
                for (rel_path, size, is_dir_entry) in &entries {
                    if *is_dir_entry {
                        continue;
                    }
                    if crate::core::traversal::is_excluded(
                        std::path::Path::new(rel_path),
                        &excludes,
                    ) {
                        continue;
                    }
                    items.push(TransferItem {
                        local_path: local_dir.join(rel_path),
                        remote: rsrc.join(rel_path),
                        size: *size,
                        is_upload: false,
                    });
                }
            } else {
                let local_path = if dest_local.is_dir() {
                    dest_local.join(rsrc.path.rsplit('/').next().unwrap_or(&rsrc.path))
                } else {
                    dest_local.to_path_buf()
                };
                items.push(TransferItem {
                    local_path,
                    remote: rsrc.clone(),
                    size: info.size,
                    is_upload: false,
                });
            }
        }

        run_parallel_transfers(items, parallel, runner.progress(), &opts)
            .await
            .map_err(|e| anyhow::anyhow!("{}", e))?;
    } else {
        for (rsrc, _) in &remote_sources {
            let info = remote::remote_stat(rsrc).await?;
            let inc = runner.inc_callback();
            let skip = runner.skip_callback();
            let file_cb = runner.file_callback();

            if info.is_dir {
                if !args.is_recursive() {
                    bail!("Remote source '{}' is a directory. Use -r flag.", rsrc);
                }
                let dir_name = rsrc.path.rsplit('/').next().unwrap_or(&rsrc.path);
                let local_dir = if dest_local.is_dir() {
                    dest_local.join(dir_name)
                } else {
                    dest_local.to_path_buf()
                };
                if !local_dir.exists() {
                    tokio::fs::create_dir_all(&local_dir).await?;
                }
                remote::download_directory(
                    rsrc, &local_dir, &inc, &skip, &file_cb, &excludes, &opts,
                )
                .await?;
            } else {
                let local_path = if dest_local.is_dir() {
                    dest_local.join(rsrc.path.rsplit('/').next().unwrap_or(&rsrc.path))
                } else {
                    dest_local.to_path_buf()
                };
                remote::download_file(rsrc, &local_path, &inc, &skip, &file_cb, info.size, &opts)
                    .await?;
            }
        }
    }

    runner.finish_ok()
}
