use std::path::{Path, PathBuf};
use std::sync::{
    Arc, OnceLock,
    atomic::{AtomicBool, Ordering},
};

use ignore::WalkBuilder;
use rayon::ThreadPool;
use rayon::prelude::*;
use serde::{Deserialize, Serialize};

use crate::error::{RebeccaError, Result, ScanFailure, ScanFailurePhase};
use crate::model::DeleteMode;
use crate::parallelism::{bounded_parallelism_budget, run_scoped_parallel_work};
use crate::plan::{CleanupTarget, CleanupTargetIssueReason};
use crate::safety::{
    PATH_DOES_NOT_EXIST_REASON, PathDisposition, assess_existing_path, is_reparse_like,
};

#[derive(Debug, Clone, Default)]
pub struct ScanCancellationToken {
    cancelled: Arc<AtomicBool>,
}

static SCAN_THREAD_POOL: OnceLock<ThreadPool> = OnceLock::new();

impl ScanCancellationToken {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn cancel(&self) {
        self.cancelled.store(true, Ordering::SeqCst);
    }

    pub fn is_cancelled(&self) -> bool {
        self.cancelled.load(Ordering::SeqCst)
    }
}

#[derive(Debug, Clone, Copy)]
pub enum ScanProgressEvent<'a> {
    FileMeasured {
        path: &'a Path,
        file_size: u64,
        files_scanned: u64,
        bytes_scanned: u64,
    },
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct ScanReport {
    pub bytes_scanned: u64,
    pub files_scanned: u64,
    pub directories_scanned: u64,
}

impl ScanReport {
    fn record_file(&mut self, bytes: u64) {
        self.files_scanned = self.files_scanned.saturating_add(1);
        self.bytes_scanned = self.bytes_scanned.saturating_add(bytes);
    }

    fn record_directory(&mut self) {
        self.directories_scanned = self.directories_scanned.saturating_add(1);
    }
}

#[derive(Debug, Clone, Default)]
pub struct ScanEngine;

impl ScanEngine {
    pub fn new() -> Self {
        Self
    }

    pub fn measure_path(&self, path: &Path) -> Result<ScanReport> {
        self.measure_path_with_progress(path, &ScanCancellationToken::new(), |_| {})
    }

    pub fn measure_path_with_progress<F>(
        &self,
        path: &Path,
        cancellation: &ScanCancellationToken,
        progress: F,
    ) -> Result<ScanReport>
    where
        F: for<'a> FnMut(ScanProgressEvent<'a>),
    {
        check_not_cancelled(cancellation)?;
        let metadata = root_metadata(path)?;

        if is_reparse_like(&metadata) {
            return Err(RebeccaError::SafetyBlocked(
                "symlink or reparse point traversal is disabled".to_string(),
            ));
        }

        if metadata.is_file() {
            let file_size = metadata.len();
            let mut progress = progress;
            let mut report = ScanReport::default();
            report.record_file(file_size);
            progress(ScanProgressEvent::FileMeasured {
                path,
                file_size,
                files_scanned: report.files_scanned,
                bytes_scanned: report.bytes_scanned,
            });
            return Ok(report);
        }

        if metadata.is_dir() {
            return IgnoreWalkerAdapter.measure_directory(path, cancellation, progress);
        }

        Ok(ScanReport::default())
    }

    pub fn measure_target(&self, target: ScanTargetRequest) -> CleanupTarget {
        match assess_existing_path(&target.path) {
            PathDisposition::Allowed => match self.measure_path(&target.path) {
                Ok(report) => CleanupTarget::allowed(
                    target.rule_id,
                    target.path,
                    report.bytes_scanned,
                    target.mode,
                ),
                Err(err) => CleanupTarget::failed_with_reason_code(
                    target.rule_id,
                    target.path,
                    target.mode,
                    0,
                    CleanupTargetIssueReason::ScanFailed,
                    err.to_string(),
                ),
            },
            PathDisposition::Missing => CleanupTarget::skipped_with_reason_code(
                target.rule_id,
                target.path,
                target.mode,
                CleanupTargetIssueReason::SafetyPolicySkipped,
                PATH_DOES_NOT_EXIST_REASON,
            ),
            PathDisposition::Skipped(reason) => CleanupTarget::skipped_with_reason_code(
                target.rule_id,
                target.path,
                target.mode,
                CleanupTargetIssueReason::SafetyPolicySkipped,
                reason,
            ),
            PathDisposition::Blocked(reason) => CleanupTarget::blocked_with_reason_code(
                target.rule_id,
                target.path,
                target.mode,
                CleanupTargetIssueReason::SafetyPolicyBlocked,
                reason,
            ),
        }
    }

    pub fn measure_targets(&self, targets: Vec<ScanTargetRequest>) -> Vec<CleanupTarget> {
        let scanner = self;
        let mut scanned: Vec<_> = run_scoped_scan(|| {
            targets
                .into_par_iter()
                .map(|target| scanner.measure_target(target))
                .collect()
        });

        scanned.sort_by(|left, right| {
            left.rule_id
                .cmp(&right.rule_id)
                .then_with(|| left.path.cmp(&right.path))
        });

        scanned
    }
}

#[derive(Debug, Clone)]
pub struct ScanTargetRequest {
    rule_id: String,
    path: PathBuf,
    mode: DeleteMode,
}

impl ScanTargetRequest {
    pub fn new(rule_id: impl Into<String>, path: PathBuf, mode: DeleteMode) -> Self {
        Self {
            rule_id: rule_id.into(),
            path,
            mode,
        }
    }
}

#[derive(Debug, Clone, Copy)]
struct IgnoreWalkerAdapter;

impl IgnoreWalkerAdapter {
    fn measure_directory<F>(
        self,
        path: &Path,
        cancellation: &ScanCancellationToken,
        mut progress: F,
    ) -> Result<ScanReport>
    where
        F: for<'a> FnMut(ScanProgressEvent<'a>),
    {
        let mut report = ScanReport::default();
        let walker = cleanup_walk_builder(path).build();

        for entry in walker {
            check_not_cancelled(cancellation)?;
            let entry = entry
                .map_err(|err| RebeccaError::ScanFailed(ScanFailure::directory_walk(path, &err)))?;
            let metadata = entry_metadata(entry.path())?;

            if is_reparse_like(&metadata) {
                continue;
            }

            if metadata.is_dir() {
                report.record_directory();
            }

            if metadata.is_file() {
                let file_size = metadata.len();
                report.record_file(file_size);
                progress(ScanProgressEvent::FileMeasured {
                    path: entry.path(),
                    file_size,
                    files_scanned: report.files_scanned,
                    bytes_scanned: report.bytes_scanned,
                });
            }
        }

        Ok(report)
    }
}

fn cleanup_walk_builder(path: &Path) -> WalkBuilder {
    let mut builder = WalkBuilder::new(path);
    builder.standard_filters(false).follow_links(false);
    builder
}

fn root_metadata(path: &Path) -> Result<std::fs::Metadata> {
    std::fs::symlink_metadata(path).map_err(|err| {
        RebeccaError::ScanFailed(ScanFailure::from_io(
            path,
            ScanFailurePhase::RootMetadata,
            &err,
        ))
    })
}

fn entry_metadata(path: &Path) -> Result<std::fs::Metadata> {
    std::fs::symlink_metadata(path).map_err(|err| {
        RebeccaError::ScanFailed(ScanFailure::from_io(
            path,
            ScanFailurePhase::EntryMetadata,
            &err,
        ))
    })
}

fn check_not_cancelled(cancellation: &ScanCancellationToken) -> Result<()> {
    if cancellation.is_cancelled() {
        return Err(RebeccaError::OperationCancelled(
            "scan was cancelled".to_string(),
        ));
    }

    Ok(())
}

pub fn scan_parallelism_budget() -> usize {
    bounded_parallelism_budget()
}

pub(crate) fn run_scoped_scan<R, F>(work: F) -> R
where
    F: FnOnce() -> R + Send,
    R: Send,
{
    run_scoped_parallel_work(&SCAN_THREAD_POOL, "scan", work)
}

#[cfg(test)]
mod tests {
    use super::{run_scoped_scan, scan_parallelism_budget};
    use crate::executor::cleanup_parallelism_budget;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};

    #[test]
    fn scan_parallelism_budget_stays_bounded() {
        let budget = scan_parallelism_budget();

        assert!((2..=8).contains(&budget));
    }

    #[test]
    fn run_scoped_scan_executes_work() {
        let counter = Arc::new(AtomicUsize::new(0));
        let counter_ref = Arc::clone(&counter);

        run_scoped_scan(move || {
            counter_ref.fetch_add(1, Ordering::SeqCst);
        });

        assert_eq!(counter.load(Ordering::SeqCst), 1);
    }

    #[test]
    fn scan_and_cleanup_parallelism_budgets_match() {
        assert_eq!(scan_parallelism_budget(), cleanup_parallelism_budget());
    }
}
