// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::collections::HashSet;
use std::fs::FileType;
use std::io::ErrorKind;
use std::sync::{Arc, Once};
use std::time::{Duration, SystemTime};

use crate::ReplTargetSizeSummary;
use crate::data_usage_define::{DataUsageCache, DataUsageEntry, DataUsageHash, DataUsageHashMap, SizeSummary, hash_path};
use crate::error::ScannerError;
use crate::scanner_io::ScannerIODisk as _;
use crate::sleeper::DynamicSleeper;
use metrics::{counter, describe_counter};
use rustfs_common::heal_channel::{
    HEAL_DELETE_DANGLING, HealAdmissionResult, HealChannelPriority, HealChannelRequest, HealScanMode,
    send_heal_request_with_admission,
};
use rustfs_common::metrics::{IlmAction, Metric, Metrics, UpdateCurrentPathFn, current_path_updater};
use rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc;
use rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::{GLOBAL_ExpiryState, apply_expiry_rule};
use rustfs_ecstore::bucket::lifecycle::evaluator::Evaluator;
use rustfs_ecstore::bucket::lifecycle::{
    bucket_lifecycle_ops::apply_transition_rule,
    lifecycle::{Event, Lifecycle, ObjectOpts},
};
use rustfs_ecstore::bucket::replication::{ReplicationConfig, ReplicationConfigurationExt as _, queue_replication_heal_internal};
use rustfs_ecstore::bucket::versioning::VersioningApi;
use rustfs_ecstore::bucket::versioning_sys::BucketVersioningSys;
use rustfs_ecstore::cache_value::metacache_set::{ListPathRawOptions, list_path_raw};
use rustfs_ecstore::disk::error::DiskError;
use rustfs_ecstore::disk::{Disk, DiskAPI as _, DiskInfoOptions};
use rustfs_ecstore::error::StorageError;
use rustfs_ecstore::global::is_erasure;
use rustfs_ecstore::pools::{path2_bucket_object, path2_bucket_object_with_base_path};
use rustfs_ecstore::store_api::{ObjectInfo, ObjectToDelete};
use rustfs_ecstore::store_utils::is_reserved_or_invalid_bucket;
use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ReplicationStatusType};
use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration};
use tokio::select;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, warn};

const DATA_USAGE_UPDATE_DIR_CYCLES: u32 = 16;
const DATA_SCANNER_COMPACT_LEAST_OBJECT: usize = 500;
const YIELD_EVERY_N_OBJECTS: u64 = 128;
const DATA_SCANNER_COMPACT_AT_CHILDREN: usize = 10000;
const DATA_SCANNER_COMPACT_AT_FOLDERS: usize = DATA_SCANNER_COMPACT_AT_CHILDREN / 4;
const DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS: usize = 250_000;
const DEFAULT_HEAL_OBJECT_SELECT_PROB: u32 = 1024;
const ENV_DATA_USAGE_UPDATE_DIR_CYCLES: &str = "RUSTFS_DATA_USAGE_UPDATE_DIR_CYCLES";
const ENV_HEAL_OBJECT_SELECT_PROB: &str = "RUSTFS_HEAL_OBJECT_SELECT_PROB";
const ENV_FAILED_OBJECT_TTL_SECS: &str = "RUSTFS_DATA_USAGE_FAILED_OBJECT_TTL_SECS";
const ENV_FAILED_OBJECTS_MAX: &str = "RUSTFS_DATA_USAGE_FAILED_OBJECTS_MAX";
const DEFAULT_FAILED_OBJECT_TTL_SECS: u32 = 86_400;
const DEFAULT_FAILED_OBJECTS_MAX: u32 = 10_000;
const METRIC_SCANNER_INLINE_HEAL_TOTAL: &str = "rustfs_scanner_inline_heal_total";

static SCANNER_INLINE_HEAL_WARN_ONCE: Once = Once::new();
static SCANNER_INLINE_HEAL_METRICS_ONCE: Once = Once::new();

pub fn data_usage_update_dir_cycles() -> u32 {
    rustfs_utils::get_env_u32(ENV_DATA_USAGE_UPDATE_DIR_CYCLES, DATA_USAGE_UPDATE_DIR_CYCLES)
}

pub fn heal_object_select_prob() -> u32 {
    rustfs_utils::get_env_u32(ENV_HEAL_OBJECT_SELECT_PROB, DEFAULT_HEAL_OBJECT_SELECT_PROB)
}

fn scanner_inline_heal_enabled() -> bool {
    scanner_inline_heal_enabled_from_value(std::env::var(rustfs_config::ENV_SCANNER_INLINE_HEAL_ENABLE).ok().as_deref())
}

fn scanner_inline_heal_enabled_from_value(value: Option<&str>) -> bool {
    match value {
        Some(value) => matches!(value.trim().to_ascii_lowercase().as_str(), "1" | "true" | "on" | "yes"),
        None => rustfs_config::DEFAULT_SCANNER_INLINE_HEAL_ENABLE,
    }
}

fn ensure_scanner_inline_heal_metric_registered() {
    SCANNER_INLINE_HEAL_METRICS_ONCE.call_once(|| {
        describe_counter!(
            METRIC_SCANNER_INLINE_HEAL_TOTAL,
            "Total number of inline heal operations executed directly by scanner."
        );
        counter!(METRIC_SCANNER_INLINE_HEAL_TOTAL).increment(0);
    });
}

fn warn_inline_heal_compat_requested() {
    if !scanner_inline_heal_enabled() {
        return;
    }

    SCANNER_INLINE_HEAL_WARN_ONCE.call_once(|| {
        warn!(
            env = rustfs_config::ENV_SCANNER_INLINE_HEAL_ENABLE,
            "Inline scanner heal rollback is no longer supported; continuing to enqueue heal candidates asynchronously"
        );
    });
}

/// Cached folder information for scanning
#[derive(Clone, Debug)]
pub struct CachedFolder {
    pub name: String,
    pub parent: Option<DataUsageHash>,
    pub object_heal_prob_div: u32,
}

/// Type alias for get size function
pub type GetSizeFn = Box<dyn Fn(ScannerItem) -> Result<SizeSummary, StorageError> + Send + Sync>;

fn build_bucket_heal_request(bucket: String, priority: HealChannelPriority) -> HealChannelRequest {
    HealChannelRequest {
        bucket,
        priority,
        ..Default::default()
    }
}

fn build_object_heal_request(
    bucket: String,
    object: String,
    version_id: Option<String>,
    scan_mode: HealScanMode,
    priority: HealChannelPriority,
) -> HealChannelRequest {
    HealChannelRequest {
        bucket,
        object_prefix: Some(object),
        object_version_id: version_id,
        priority,
        scan_mode: Some(scan_mode),
        remove_corrupted: Some(HEAL_DELETE_DANGLING),
        ..Default::default()
    }
}

fn heal_priority_label(priority: HealChannelPriority) -> &'static str {
    match priority {
        HealChannelPriority::Low => "low",
        HealChannelPriority::Normal => "normal",
        HealChannelPriority::High => "high",
        HealChannelPriority::Critical => "critical",
    }
}

fn describe_heal_admission(result: HealAdmissionResult) -> String {
    match result {
        HealAdmissionResult::Accepted | HealAdmissionResult::Merged => result.result_label().to_string(),
        HealAdmissionResult::Full => "queue_full".to_string(),
        HealAdmissionResult::Dropped(reason) => format!("dropped:{}", reason.as_str()),
    }
}

fn record_high_priority_heal_escalation(
    candidate_type: &'static str,
    priority: HealChannelPriority,
    result: HealAdmissionResult,
) {
    counter!(
        "rustfs_heal_candidate_priority_reject_total",
        "type" => candidate_type.to_string(),
        "priority" => heal_priority_label(priority).to_string(),
        "result" => result.result_label().to_string(),
        "reason" => result.reason_label().to_string()
    )
    .increment(1);
}

fn build_high_priority_heal_admission_error(
    candidate_type: &'static str,
    bucket: &str,
    object: Option<&str>,
    priority: HealChannelPriority,
    result: HealAdmissionResult,
) -> ScannerError {
    let object_text = object.map(|object| format!(", object='{object}'")).unwrap_or_default();
    ScannerError::Other(format!(
        "high-priority heal request was not admitted: type={candidate_type}, bucket='{bucket}'{object_text}, priority={}, admission={}",
        heal_priority_label(priority),
        describe_heal_admission(result)
    ))
}

fn record_heal_candidate_admission(candidate_type: &'static str, priority: HealChannelPriority, result: HealAdmissionResult) {
    counter!(
        "rustfs_heal_candidate_enqueue_total",
        "type" => candidate_type.to_string(),
        "priority" => heal_priority_label(priority).to_string(),
        "result" => result.result_label().to_string()
    )
    .increment(1);

    if matches!(result, HealAdmissionResult::Merged) {
        counter!(
            "rustfs_heal_candidate_merge_total",
            "type" => candidate_type.to_string()
        )
        .increment(1);
    }

    if let HealAdmissionResult::Dropped(reason) = result {
        counter!(
            "rustfs_heal_candidate_drop_total",
            "type" => candidate_type.to_string(),
            "reason" => reason.as_str().to_string()
        )
        .increment(1);
    }
}

async fn send_scanner_heal_request(
    candidate_type: &'static str,
    request: HealChannelRequest,
) -> Result<HealAdmissionResult, ScannerError> {
    let priority = request.priority;
    match send_heal_request_with_admission(request).await {
        Ok(result) => {
            record_heal_candidate_admission(candidate_type, priority, result);
            Ok(result)
        }
        Err(err) => {
            counter!(
                "rustfs_heal_candidate_enqueue_total",
                "type" => candidate_type.to_string(),
                "priority" => heal_priority_label(priority).to_string(),
                "result" => "channel_error".to_string()
            )
            .increment(1);
            Err(ScannerError::Other(err))
        }
    }
}

async fn send_required_scanner_heal_request(
    candidate_type: &'static str,
    bucket: &str,
    object: Option<&str>,
    request: HealChannelRequest,
) -> Result<(), ScannerError> {
    let priority = request.priority;
    let result = send_scanner_heal_request(candidate_type, request).await?;
    if result.is_admitted() {
        return Ok(());
    }

    record_high_priority_heal_escalation(candidate_type, priority, result);
    error!(
        candidate_type,
        bucket,
        object = object.unwrap_or(""),
        priority = heal_priority_label(priority),
        admission = result.result_label(),
        reason = result.reason_label(),
        "High-priority heal request was not admitted; escalating to scanner failure"
    );
    Err(build_high_priority_heal_admission_error(candidate_type, bucket, object, priority, result))
}

/// Scanner item representing a file during scanning
#[derive(Clone, Debug)]
pub struct ScannerItem {
    pub path: String,
    pub bucket: String,
    pub prefix: String,
    pub object_name: String,
    pub file_type: FileType,
    pub lifecycle: Option<Arc<BucketLifecycleConfiguration>>,
    pub replication: Option<Arc<ReplicationConfig>>,
    pub heal_enabled: bool,
    pub heal_bitrot: bool,
    pub debug: bool,
}

impl ScannerItem {
    /// Get the object path (prefix + object_name)
    pub fn object_path(&self) -> String {
        if self.prefix.is_empty() {
            self.object_name.clone()
        } else {
            path_join_buf(&[&self.prefix, &self.object_name])
        }
    }

    /// Transform meta directory by splitting prefix and extracting object name
    /// This converts a directory path like "bucket/dir1/dir2/file" to prefix="bucket/dir1/dir2" and object_name="file"
    pub fn transform_meta_dir(&mut self) {
        let prefix = self.prefix.clone(); // Clone to avoid borrow checker issues
        let split: Vec<&str> = prefix.split(SLASH_SEPARATOR).collect();

        if split.len() > 1 {
            let prefix_parts: Vec<&str> = split[..split.len() - 1].to_vec();
            self.prefix = path_join_buf(&prefix_parts);
        } else {
            self.prefix = String::new();
        }

        // Object name is the last element
        self.object_name = split.last().unwrap_or(&"").to_string();
    }

    pub async fn apply_actions(
        &mut self,
        object_infos: Vec<ObjectInfo>,
        lock_retention: Option<Arc<ObjectLockConfiguration>>,
        size_summary: &mut SizeSummary,
    ) {
        if object_infos.is_empty() {
            debug!("apply_actions: no object infos for object: {}", self.object_path());
            return;
        }
        debug!("apply_actions: applying actions for object: {}", self.object_path());

        let versioning_config = match BucketVersioningSys::get(&self.bucket).await {
            Ok(versioning_config) => versioning_config,
            Err(_) => {
                warn!("apply_actions: Failed to get versioning configuration for bucket {}", self.bucket);
                return;
            }
        };

        let Some(lifecycle) = self.lifecycle.as_ref() else {
            let mut cumulative_size = 0;
            for oi in object_infos.iter() {
                let actual_size = match oi.get_actual_size() {
                    Ok(size) => size,
                    Err(_) => {
                        warn!("apply_actions: Failed to get actual size for object {}", oi.name);
                        continue;
                    }
                };

                let size = self.heal_actions(oi, actual_size, size_summary).await;

                size_summary.actions_accounting(oi, size, actual_size);

                cumulative_size += size;
            }

            self.alert_excessive_versions(object_infos.len(), cumulative_size);

            debug!("apply_actions: done for now no lifecycle config");
            return;
        };

        let object_opts = object_infos
            .iter()
            .map(ObjectOpts::from_object_info)
            .collect::<Vec<ObjectOpts>>();

        let events = match Evaluator::new(lifecycle.clone())
            .with_lock_retention(lock_retention)
            .with_replication_config(self.replication.clone())
            .eval(&object_opts)
            .await
        {
            Ok(events) => events,
            Err(e) => {
                warn!("apply_actions: Failed to evaluate lifecycle for object: {}", e);
                return;
            }
        };
        let mut to_delete_objs: Vec<ObjectToDelete> = Vec::new();
        let mut noncurrent_events: Vec<Event> = Vec::new();
        let mut cumulative_size = 0;
        let mut remaining_versions = object_infos.len();
        'eventLoop: {
            for (i, event) in events.iter().enumerate() {
                let oi = &object_infos[i];
                let actual_size = match oi.get_actual_size() {
                    Ok(size) => size,
                    Err(_) => {
                        warn!("apply_actions: Failed to get actual size for object {}", oi.name);
                        0
                    }
                };

                let mut size = actual_size;

                let done_ilm = Metrics::time_ilm(event.action);
                match event.action {
                    IlmAction::DeleteAllVersionsAction | IlmAction::DelMarkerDeleteAllVersionsAction => {
                        remaining_versions = 0;
                        debug!("apply_actions: applying expiry rule for object: {} {}", oi.name, event.action);
                        apply_expiry_rule(event, &LcEventSrc::Scanner, oi).await;
                        done_ilm(1)();
                        break 'eventLoop;
                    }

                    IlmAction::DeleteAction | IlmAction::DeleteRestoredAction | IlmAction::DeleteRestoredVersionAction => {
                        if !versioning_config.prefix_enabled(&self.object_path()) && event.action == IlmAction::DeleteAction {
                            remaining_versions -= 1;
                            size = 0;
                        }

                        debug!("apply_actions: applying expiry rule for object: {} {}", oi.name, event.action);
                        apply_expiry_rule(event, &LcEventSrc::Scanner, oi).await;
                        done_ilm(1)();
                    }
                    IlmAction::DeleteVersionAction => {
                        remaining_versions -= 1;
                        size = 0;
                        if let Some(opt) = object_opts.get(i) {
                            to_delete_objs.push(ObjectToDelete {
                                object_name: opt.name.clone(),
                                version_id: opt.version_id,
                                ..Default::default()
                            });
                        }
                        noncurrent_events.push(event.clone());
                        done_ilm(1)();
                    }
                    IlmAction::TransitionAction | IlmAction::TransitionVersionAction => {
                        debug!("apply_actions: applying transition rule for object: {} {}", oi.name, event.action);
                        apply_transition_rule(event, &LcEventSrc::Scanner, oi).await;
                        done_ilm(1)();
                    }

                    IlmAction::NoneAction | IlmAction::ActionCount => {
                        size = self.heal_actions(oi, actual_size, size_summary).await;
                    }
                }

                size_summary.actions_accounting(oi, size, actual_size);

                cumulative_size += size;
            }
        }

        if !to_delete_objs.is_empty()
            && let Some(event) = noncurrent_events.first().cloned()
        {
            GLOBAL_ExpiryState
                .write()
                .await
                .enqueue_by_newer_noncurrent(&self.bucket, to_delete_objs, event)
                .await;
        }
        self.alert_excessive_versions(remaining_versions, cumulative_size);
    }

    async fn heal_actions(&mut self, oi: &ObjectInfo, actual_size: i64, size_summary: &mut SizeSummary) -> i64 {
        if self.heal_enabled {
            warn_inline_heal_compat_requested();
            self.enqueue_heal(oi).await;
        }

        self.heal_replication(oi, size_summary).await;

        actual_size
    }

    async fn heal_replication(&mut self, oi: &ObjectInfo, size_summary: &mut SizeSummary) {
        if oi.version_id.is_none_or(|v| v.is_nil()) {
            return;
        }

        let Some(replication) = self.replication.clone() else {
            return;
        };

        let roi = queue_replication_heal_internal(&oi.bucket, oi.clone(), (*replication).clone(), 0).await;
        if oi.delete_marker || oi.version_purge_status.is_empty() {
            return;
        }

        for (arn, target_status) in roi.target_statuses.iter() {
            if !size_summary.repl_target_stats.contains_key(arn.as_str()) {
                size_summary
                    .repl_target_stats
                    .insert(arn.clone(), ReplTargetSizeSummary::default());
            }

            if let Some(repl_target_size_summary) = size_summary.repl_target_stats.get_mut(arn.as_str()) {
                match target_status {
                    ReplicationStatusType::Pending => {
                        repl_target_size_summary.pending_size += roi.size;
                        repl_target_size_summary.pending_count += 1;
                        size_summary.pending_size += roi.size;
                        size_summary.pending_count += 1;
                    }
                    ReplicationStatusType::Failed => {
                        repl_target_size_summary.failed_size += roi.size;
                        repl_target_size_summary.failed_count += 1;
                        size_summary.failed_size += roi.size;
                        size_summary.failed_count += 1;
                    }
                    ReplicationStatusType::Completed | ReplicationStatusType::CompletedLegacy => {
                        repl_target_size_summary.replicated_size += roi.size;
                        repl_target_size_summary.replicated_count += 1;
                        size_summary.replicated_size += roi.size;
                        size_summary.replicated_count += 1;
                    }
                    _ => {}
                }
            }
        }

        if oi.replication_status == ReplicationStatusType::Replica {
            size_summary.replica_size += roi.size;
            size_summary.replica_count += 1;
        }
    }

    async fn enqueue_heal(&mut self, oi: &ObjectInfo) {
        let done_heal = Metrics::time(Metric::HealAbandonedObject);
        debug!(
            "enqueue_heal: bucket: {}, object_path: {}, version_id: {}",
            self.bucket,
            self.object_path(),
            oi.version_id.unwrap_or_default()
        );

        let scan_mode = if self.heal_bitrot {
            HealScanMode::Deep
        } else {
            HealScanMode::Normal
        };

        let result = send_scanner_heal_request(
            "object",
            build_object_heal_request(
                self.bucket.clone(),
                self.object_path(),
                oi.version_id
                    .and_then(|v| if v.is_nil() { None } else { Some(v.to_string()) }),
                scan_mode,
                HealChannelPriority::Low,
            ),
        )
        .await;

        match result {
            Ok(HealAdmissionResult::Accepted | HealAdmissionResult::Merged) => {}
            Ok(result @ (HealAdmissionResult::Full | HealAdmissionResult::Dropped(_))) => {
                warn!(
                    bucket = %self.bucket,
                    object = %self.object_path(),
                    admission = %describe_heal_admission(result),
                    "enqueue_heal: low-priority heal request was not admitted"
                );
            }
            Err(e) => warn!("enqueue_heal: failed to submit heal request: {}", e),
        }
        done_heal();
    }

    fn alert_excessive_versions(&self, _object_infos_length: usize, _cumulative_size: i64) {
        // TODO: Implement alerting for excessive versions
    }
}

/// Folder scanner for scanning directory structures
pub struct FolderScanner {
    root: String,
    old_cache: DataUsageCache,
    new_cache: DataUsageCache,
    update_cache: DataUsageCache,

    data_usage_scanner_debug: bool,
    heal_object_select: u32,
    scan_mode: HealScanMode,

    failed_object_ttl_secs: u64,
    failed_objects_max: usize,

    sleeper: DynamicSleeper,
    // should_heal: Arc<dyn Fn() -> bool + Send + Sync>,
    disks: Vec<Arc<Disk>>,
    disks_quorum: usize,

    updates: Option<mpsc::Sender<DataUsageEntry>>,
    last_update: SystemTime,

    update_current_path: UpdateCurrentPathFn,

    skip_heal: Arc<std::sync::atomic::AtomicBool>,
    local_disk: Arc<Disk>,
}

impl FolderScanner {
    fn now_secs() -> u64 {
        SystemTime::now()
            .duration_since(SystemTime::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs()
    }

    fn should_skip_failed(&self, path: &str) -> bool {
        let ttl = self.failed_object_ttl_secs;
        if ttl == 0 {
            return false;
        }

        let Some(last_failed) = self.new_cache.info.failed_objects.get(path) else {
            return false;
        };

        let now = Self::now_secs();
        now.saturating_sub(*last_failed) < ttl
    }

    fn record_failed(&mut self, path: &str) {
        let ttl = self.failed_object_ttl_secs;
        if ttl == 0 {
            return;
        }

        let now = Self::now_secs();
        self.new_cache.info.failed_objects.insert(path.to_string(), now);

        let max_entries = self.failed_objects_max;
        if max_entries > 0 && self.new_cache.info.failed_objects.len() > max_entries {
            self.prune_failed_objects(now, ttl);
        }
    }

    fn prune_failed_objects_cache(&mut self) {
        let ttl = self.failed_object_ttl_secs;
        if ttl == 0 {
            return;
        }

        let now = Self::now_secs();
        self.prune_failed_objects(now, ttl);
    }

    fn prune_failed_objects(&mut self, now: u64, ttl: u64) {
        let max_entries = self.failed_objects_max;
        let failed = &mut self.new_cache.info.failed_objects;
        if failed.is_empty() {
            return;
        }

        failed.retain(|_, ts| now.saturating_sub(*ts) < ttl);

        if max_entries == 0 {
            return;
        }

        if failed.len() <= max_entries {
            return;
        }

        let mut entries: Vec<(String, u64)> = failed.iter().map(|(k, v)| (k.clone(), *v)).collect();
        entries.sort_by(|(k1, ts1), (k2, ts2)| ts1.cmp(ts2).then_with(|| k1.cmp(k2)));

        let remove_count = failed.len().saturating_sub(max_entries);
        for (key, _) in entries.into_iter().take(remove_count) {
            failed.remove(&key);
        }
    }

    pub async fn should_heal(&self) -> bool {
        if self.skip_heal.load(std::sync::atomic::Ordering::Relaxed) {
            return false;
        }
        if self.heal_object_select == 0 {
            return false;
        }

        if self
            .local_disk
            .disk_info(&DiskInfoOptions::default())
            .await
            .unwrap_or_default()
            .healing
        {
            self.skip_heal.store(true, std::sync::atomic::Ordering::Relaxed);
            return false;
        }

        true
    }

    /// Set heal object select probability
    pub fn set_heal_object_select(&mut self, prob: u32) {
        self.heal_object_select = prob;
    }

    /// Set debug mode
    pub fn set_debug(&mut self, debug: bool) {
        self.data_usage_scanner_debug = debug;
    }

    /// Send update if enough time has passed
    /// Should be called on a regular basis when the new_cache contains more recent total than previously.
    /// May or may not send an update upstream.
    pub async fn send_update(&mut self) {
        // Send at most an update every minute.
        if self.updates.is_none() {
            return;
        }

        let elapsed = self.last_update.elapsed().unwrap_or(Duration::from_secs(0));
        if elapsed < Duration::from_secs(60) {
            return;
        }

        if let Some(flat) = self.update_cache.size_recursive(&self.new_cache.info.name)
            && let Some(ref updates) = self.updates
        {
            // Try to send without blocking
            if let Err(e) = updates.send(flat.clone()).await {
                error!("send_update: failed to send update: {}", e);
            }
            self.last_update = SystemTime::now();
        }
    }

    /// Scan a folder recursively
    /// Files found in the folders will be added to new_cache.
    #[allow(clippy::never_loop)]
    #[allow(unused_assignments)]
    pub async fn scan_folder(
        &mut self,
        ctx: CancellationToken,
        folder: CachedFolder,
        into: &mut DataUsageEntry,
    ) -> Result<(), ScannerError> {
        let done_folder = Metrics::time(Metric::ScanFolder);

        if ctx.is_cancelled() {
            return Err(ScannerError::Other("Operation cancelled".to_string()));
        }

        let this_hash = hash_path(&folder.name);
        // Store initial compaction state.
        let was_compacted = into.compacted;

        loop {
            if ctx.is_cancelled() {
                return Err(ScannerError::Other("Operation cancelled".to_string()));
            }

            self.prune_failed_objects_cache();

            let mut abandoned_children: DataUsageHashMap = HashSet::new();
            if !into.compacted {
                abandoned_children = self.old_cache.find_children_copy(this_hash.clone());
            }

            debug!("scan_folder : {}/{}", &self.root, &folder.name);
            let (_, prefix) = path2_bucket_object_with_base_path(&self.root, &folder.name);

            let active_life_cycle = if self
                .old_cache
                .info
                .lifecycle
                .as_ref()
                .is_some_and(|v| v.has_active_rules(&prefix))
            {
                self.old_cache.info.lifecycle.clone()
            } else {
                None
            };

            let active_replication =
                if self.old_cache.info.replication.as_ref().is_some_and(|v| {
                    !v.is_empty() && v.config.as_ref().is_some_and(|config| config.has_active_rules(&prefix, true))
                }) {
                    self.old_cache.info.replication.clone()
                } else {
                    None
                };

            self.sleeper.sleep_folder().await;

            let mut existing_folders: Vec<CachedFolder> = Vec::new();
            let mut new_folders: Vec<CachedFolder> = Vec::new();
            let mut found_objects = false;
            let mut object_count: u64 = 0;

            let dir_path = path_join_buf(&[&self.root, &folder.name]);

            debug!("scan_folder: dir_path: {:?}", dir_path);

            let mut dir_reader = match tokio::fs::read_dir(&dir_path).await {
                Ok(dir_reader) => dir_reader,
                Err(e) if e.kind() == ErrorKind::NotFound => {
                    warn!("scan_folder: directory disappeared before read {}: {}", dir_path, e);
                    return Ok(());
                }
                Err(e) => return Err(ScannerError::Io(e)),
            };

            loop {
                let entry = match dir_reader.next_entry().await {
                    Ok(Some(entry)) => entry,
                    Ok(None) => break,
                    Err(e) if e.kind() == ErrorKind::NotFound => {
                        warn!("scan_folder: directory disappeared during iteration {}: {}", dir_path, e);
                        break;
                    }
                    Err(e) if e.kind() == ErrorKind::NotADirectory => {
                        warn!("scan_folder: path became non-directory during iteration {}: {}", dir_path, e);
                        break;
                    }
                    Err(e) => return Err(ScannerError::Io(e)),
                };
                let file_name = entry.file_name().to_string_lossy().to_string();
                if file_name.is_empty() || file_name == "." || file_name == ".." {
                    continue;
                }

                let file_path = entry.path().to_string_lossy().to_string();

                let trim_dir_name = file_path.strip_prefix(&dir_path).unwrap_or(&file_path);

                let entry_name = path_join_buf(&[&folder.name, trim_dir_name]);

                if entry_name.is_empty() || entry_name == folder.name {
                    continue;
                }

                // Ignore entries that disappeared during traversal or hit symlink
                // loops, but propagate other walk errors.
                let mut entry_type = match entry.file_type().await {
                    Ok(entry_type) => entry_type,
                    Err(e) if e.kind() == ErrorKind::NotFound => {
                        warn!("scan_folder: entry disappeared before type lookup {}: {}", entry_name, e);
                        continue;
                    }
                    Err(e) if e.kind() == ErrorKind::TooManyLinks => {
                        warn!("scan_folder: entry hit symlink loop before type lookup {}: {}", entry_name, e);
                        continue;
                    }
                    Err(e) => return Err(ScannerError::Io(e)),
                };

                if entry_type.is_symlink() {
                    let metadata = match tokio::fs::metadata(&file_path).await {
                        Ok(metadata) => metadata,
                        Err(e) if e.kind() == ErrorKind::NotFound => {
                            warn!("scan_folder: symlink target disappeared before metadata lookup {}: {}", file_path, e);
                            continue;
                        }
                        Err(e) if e.kind() == ErrorKind::TooManyLinks => {
                            warn!("scan_folder: symlink target hit loop before metadata lookup {}: {}", file_path, e);
                            continue;
                        }
                        Err(e) => return Err(ScannerError::Io(e)),
                    };

                    if metadata.is_dir() {
                        warn!("scan_folder: ignoring symlinked directory {}", file_path);
                        continue;
                    }

                    entry_type = metadata.file_type();
                }

                // ok

                let (bucket, prefix) = path2_bucket_object_with_base_path(self.root.as_str(), &entry_name);
                if bucket.is_empty() {
                    break;
                }

                if is_reserved_or_invalid_bucket(&bucket, false) {
                    break;
                }

                if ctx.is_cancelled() {
                    break;
                }

                if entry_type.is_dir() {
                    let h = hash_path(&entry_name);

                    if h == this_hash {
                        continue;
                    }

                    let exists = self.old_cache.cache.contains_key(&h.key());

                    let this = CachedFolder {
                        name: entry_name.clone(),
                        parent: Some(this_hash.clone()),
                        object_heal_prob_div: folder.object_heal_prob_div,
                    };

                    abandoned_children.remove(&h.key());

                    if exists {
                        existing_folders.push(this);
                        self.update_cache
                            .copy_with_children(&self.old_cache, &h, &Some(this_hash.clone()));
                    } else {
                        new_folders.push(this);
                    }
                    continue;
                }

                let timer = self.sleeper.timer();

                let heal_enabled = this_hash.mod_alt(
                    self.old_cache.info.next_cycle as u32 / folder.object_heal_prob_div,
                    self.heal_object_select / folder.object_heal_prob_div,
                ) && self.should_heal().await;

                let mut item = ScannerItem {
                    path: file_path,
                    bucket,
                    prefix: rustfs_utils::path::dir(&prefix),
                    object_name: file_name,
                    lifecycle: active_life_cycle.clone(),
                    replication: active_replication.clone(),
                    heal_enabled,
                    heal_bitrot: self.scan_mode == HealScanMode::Deep,
                    debug: self.data_usage_scanner_debug,
                    file_type: entry_type,
                };

                // If this path is already known as failed, just skip it.
                // We intentionally do NOT call `record_failed` or bump `failed_objects` here,
                // because the failure was recorded when the original error occurred
                // (e.g. in the get_size error branch below). This branch only accounts
                // for subsequent skips of already-failed paths.
                if self.should_skip_failed(&item.path) {
                    continue;
                }

                let sz = match self.local_disk.get_size(item.clone()).await {
                    Ok(sz) => sz,
                    Err(e) => {
                        let is_skip_file = matches!(e, StorageError::Io(ref io) if io.to_string() == "skip file");

                        if !is_skip_file {
                            // Track failed objects to prevent infinite retry loops
                            into.failed_objects += 1;
                            self.record_failed(&item.path);

                            // Only log non-skip errors to avoid noise
                            warn!("scan_folder: failed to get size for item {}: {}", item.path, e);
                        }

                        timer.sleep().await;
                        continue;
                    }
                };

                found_objects = true;

                item.transform_meta_dir();

                abandoned_children.remove(&path_join_buf(&[&item.bucket, &item.object_path()]));

                into.add_sizes(&sz);
                into.objects += 1;
                object_count += 1;

                timer.sleep().await;

                if object_count.is_multiple_of(YIELD_EVERY_N_OBJECTS) {
                    tokio::task::yield_now().await;
                }
            }

            if found_objects && is_erasure().await {
                // If we found an object in erasure mode, we skip subdirs (only datadirs)...
                info!("scan_folder: done for now found an object in erasure mode");
                break;
            }

            // If we have many subfolders, compact ourself.
            let should_compact = (self.new_cache.info.name != folder.name
                && existing_folders.len() + new_folders.len() >= DATA_SCANNER_COMPACT_AT_FOLDERS)
                || existing_folders.len() + new_folders.len() >= DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS;

            // TODO: Check for excess folders and send events

            if !into.compacted && should_compact {
                into.compacted = true;
                new_folders.append(&mut existing_folders);

                existing_folders.clear();

                if self.data_usage_scanner_debug {
                    debug!("scan_folder: Preemptively compacting: {}, entries: {}", folder.name, new_folders.len());
                }
            }

            if !into.compacted {
                for folder_item in &existing_folders {
                    let h = hash_path(&folder_item.name);
                    self.update_cache.copy_with_children(&self.old_cache, &h, &folder_item.parent);
                }
            }

            // Scan new folders
            for folder_item in new_folders {
                if ctx.is_cancelled() {
                    return Err(ScannerError::Other("Operation cancelled".to_string()));
                }

                let h = hash_path(&folder_item.name);
                // Add new folders to the update tree so totals update for these.
                if !into.compacted {
                    let mut found_any = false;
                    let mut parent = this_hash.clone();
                    let update_cache_name_hash = hash_path(&self.update_cache.info.name);

                    while parent != update_cache_name_hash {
                        let parent_key = parent.key();
                        let e = self.update_cache.find(&parent_key);
                        if e.is_none_or(|v| v.compacted) {
                            found_any = true;
                            break;
                        }
                        if let Some(next) = self.update_cache.search_parent(&parent) {
                            parent = next;
                        } else {
                            found_any = true;
                            break;
                        }
                    }
                    if !found_any {
                        // Add non-compacted empty entry.
                        self.update_cache
                            .replace_hashed(&h, &Some(this_hash.clone()), &DataUsageEntry::default());
                    }
                }

                (self.update_current_path)(&folder_item.name).await;

                if into.compacted {
                    // In compacted mode child totals are accumulated directly into the parent entry.
                    let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), into));
                    fut.await.map_err(|e| ScannerError::Other(e.to_string()))?;
                    tokio::task::yield_now().await;
                } else {
                    let mut dst = DataUsageEntry::default();

                    // Use Box::pin for recursive async call
                    let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), &mut dst));
                    if let Err(e) = fut.await {
                        warn!("scan_folder: failed to scan child folder {}: {}", folder_item.name, e);
                        continue;
                    }
                    tokio::task::yield_now().await;

                    let h = DataUsageHash(folder_item.name.clone());
                    into.add_child(&h);
                    // We scanned a folder, optionally send update.
                    self.update_cache.delete_recursive(&h);
                    self.update_cache.copy_with_children(&self.new_cache, &h, &folder_item.parent);
                    self.send_update().await;
                }

                if !into.compacted && self.update_cache.find(&this_hash.key()).is_some_and(|v| !v.compacted) {
                    self.update_cache.delete_recursive(&h);
                    self.update_cache
                        .copy_with_children(&self.new_cache, &h, &Some(this_hash.clone()));
                }
            }

            // Scan existing folders
            for mut folder_item in existing_folders {
                if ctx.is_cancelled() {
                    return Err(ScannerError::Other("Operation cancelled".to_string()));
                }

                let h = hash_path(&folder_item.name);

                if !into.compacted && self.old_cache.is_compacted(&h) {
                    let next_cycle = self.old_cache.info.next_cycle as u32;
                    if !h.mod_(next_cycle, data_usage_update_dir_cycles()) {
                        // Transfer and add as child...
                        self.new_cache.copy_with_children(&self.old_cache, &h, &folder_item.parent);
                        into.add_child(&h);
                        continue;
                    }

                    folder_item.object_heal_prob_div = data_usage_update_dir_cycles();
                }

                (self.update_current_path)(&folder_item.name).await;

                if into.compacted {
                    // In compacted mode child totals are accumulated directly into the parent entry.
                    let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), into));
                    fut.await.map_err(|e| ScannerError::Other(e.to_string()))?;
                    tokio::task::yield_now().await;
                } else {
                    let mut dst = DataUsageEntry::default();

                    // Use Box::pin for recursive async call
                    let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), &mut dst));
                    if let Err(e) = fut.await {
                        warn!("scan_folder: failed to scan child folder {}: {}", folder_item.name, e);
                        continue;
                    }
                    tokio::task::yield_now().await;

                    let h = DataUsageHash(folder_item.name.clone());
                    into.add_child(&h);
                    // We scanned a folder, optionally send update.
                    self.update_cache.delete_recursive(&h);
                    self.update_cache.copy_with_children(&self.new_cache, &h, &folder_item.parent);
                    self.send_update().await;
                }
            }

            // Scan for healing
            if abandoned_children.is_empty() || !self.should_heal().await {
                debug!("scan_folder: done for now abandoned children are empty or we are not healing");
                // If we are not heal scanning, return now.
                break;
            }

            if self.disks.is_empty() || self.disks_quorum == 0 {
                debug!("scan_folder: done for now disks are empty or quorum is 0");
                break;
            }

            let mut resolver = MetadataResolutionParams {
                dir_quorum: self.disks_quorum,
                obj_quorum: self.disks_quorum,
                bucket: "".to_string(),
                strict: false,
                ..Default::default()
            };

            for name in abandoned_children {
                if !self.should_heal().await {
                    break;
                }

                let (bucket, prefix) = path2_bucket_object(name.as_str());

                if bucket != resolver.bucket {
                    send_required_scanner_heal_request(
                        "bucket",
                        &bucket,
                        None,
                        build_bucket_heal_request(bucket.clone(), HealChannelPriority::High),
                    )
                    .await?;
                }

                resolver.bucket = bucket.clone();

                let child_ctx = ctx.child_token();

                let (agreed_tx, mut agreed_rx) = mpsc::channel::<String>(1);
                let (partial_tx, mut partial_rx) = mpsc::channel::<MetaCacheEntries>(1);
                let (finished_tx, mut finished_rx) = mpsc::channel::<Vec<Option<DiskError>>>(1);

                let disks = self.disks.iter().cloned().map(Some).collect();
                let disks_quorum = self.disks_quorum;
                let bucket_clone = bucket.clone();
                let prefix_clone = prefix.clone();
                let child_ctx_clone = child_ctx.clone();
                let agreed_tx = agreed_tx.clone();
                let partial_tx = partial_tx.clone();
                let finished_tx = finished_tx.clone();

                tokio::spawn(async move {
                    if let Err(e) = list_path_raw(
                        child_ctx_clone.clone(),
                        ListPathRawOptions {
                            disks,
                            bucket: bucket_clone.clone(),
                            path: prefix_clone.clone(),
                            recursive: true,
                            report_not_found: true,
                            min_disks: disks_quorum,
                            agreed: Some(Box::new(move |entry: MetaCacheEntry| {
                                let entry_name = entry.name.clone();
                                let agreed_tx = agreed_tx.clone();
                                Box::pin(async move {
                                    if let Err(e) = agreed_tx.send(entry_name).await {
                                        error!("scan_folder: list_path_raw: failed to send entry name: {}: {}", entry.name, e);
                                    }
                                })
                            })),
                            partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
                                let partial_tx = partial_tx.clone();
                                Box::pin(async move {
                                    if let Err(e) = partial_tx.send(entries).await {
                                        error!("scan_folder: list_path_raw: failed to send partial err: {}", e);
                                    }
                                })
                            })),
                            finished: Some(Box::new(move |errs: &[Option<DiskError>]| {
                                let finished_tx = finished_tx.clone();
                                let errs_clone = errs.to_vec();
                                Box::pin(async move {
                                    if let Err(e) = finished_tx.send(errs_clone).await {
                                        error!("scan_folder: list_path_raw: failed to send finished errs: {}", e);
                                    }
                                })
                            })),
                            ..Default::default()
                        },
                    )
                    .await
                    {
                        error!("scan_folder: failed to list path: {}/{}: {}", bucket_clone, prefix_clone, e);
                    }
                });

                let mut found_objects = false;

                loop {
                    select! {
                        Some(entry_name) = agreed_rx.recv() => {
                            (self.update_current_path)(&entry_name).await;
                        }
                        Some(entries) = partial_rx.recv() => {
                            if !self.should_heal().await {
                                child_ctx.cancel();
                                break;
                            }

                         let entry_option =  match entries.resolve(resolver.clone()){
                            Some(entry) => {
                                Some(entry)
                            }
                            None => {
                               let (entry,_) = entries.first_found();
                               entry
                            }
                           };


                           let Some(entry) = entry_option else {
                            break;
                           };

                           (self.update_current_path)(&entry.name).await;

                           if entry.is_dir() {
                            continue;
                           }




                           let fivs = match entry.file_info_versions(&bucket) {
                            Ok(fivs) => fivs,
                            Err(e) => {
                                error!("scan_folder: list_path_raw: failed to get file info versions: {}", e);
                                send_required_scanner_heal_request(
                                    "object",
                                    &bucket,
                                    Some(&entry.name),
                                    build_object_heal_request(
                                        bucket.clone(),
                                        entry.name.clone(),
                                        None,
                                        self.scan_mode,
                                        HealChannelPriority::High,
                                    ),
                                )
                                .await?;
                                found_objects = true;
                                continue;
                            }
                           };

                           for fiv in fivs.versions {

                            send_required_scanner_heal_request(
                                "object",
                                &bucket,
                                Some(&entry.name),
                                build_object_heal_request(
                                    bucket.clone(),
                                    entry.name.clone(),
                                    fiv.version_id.and_then(|v| if v.is_nil() { None } else { Some(v.to_string()) }),
                                    self.scan_mode,
                                    HealChannelPriority::High,
                                ),
                            )
                            .await?;
                            found_objects = true;
                           }


                        }
                        Some(errs) = finished_rx.recv() => {
                            error!("scan_folder: list_path_raw: failed to get finished errs: {:?}", errs);
                            child_ctx.cancel();
                        }
                        _ = child_ctx.cancelled() => {
                            break;
                        }
                    }
                }

                if found_objects {
                    let folder_item = CachedFolder {
                        name: name.clone(),
                        parent: Some(this_hash.clone()),
                        object_heal_prob_div: 1,
                    };

                    if into.compacted {
                        // In compacted mode child totals are accumulated directly into the parent entry.
                        let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), into));
                        fut.await.map_err(|e| ScannerError::Other(e.to_string()))?;
                        tokio::task::yield_now().await;
                    } else {
                        let mut dst = DataUsageEntry::default();

                        // Use Box::pin for recursive async call
                        let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), &mut dst));
                        if let Err(e) = fut.await {
                            warn!("scan_folder: failed to scan child folder {}: {}", folder_item.name, e);
                            continue;
                        }
                        tokio::task::yield_now().await;

                        let h = DataUsageHash(folder_item.name.clone());
                        into.add_child(&h);
                        // We scanned a folder, optionally send update.
                        self.update_cache.delete_recursive(&h);
                        self.update_cache.copy_with_children(&self.new_cache, &h, &folder_item.parent);
                        self.send_update().await;
                    }
                }
            }

            break;
        }

        if !was_compacted {
            self.new_cache.replace_hashed(&this_hash, &folder.parent, into);
        }

        if !into.compacted
            && self.new_cache.info.name != folder.name
            && let Some(mut flat) = self.new_cache.size_recursive(&this_hash.key())
        {
            flat.compacted = true;
            let mut should_compact = false;

            if flat.objects < DATA_SCANNER_COMPACT_LEAST_OBJECT {
                should_compact = true;
            } else {
                // Compact if we only have objects as children...
                should_compact = true;
                for k in &into.children {
                    if let Some(v) = self.new_cache.cache.get(k)
                        && (!v.children.is_empty() || v.objects > 1)
                    {
                        should_compact = false;
                        break;
                    }
                }
            }

            if should_compact {
                self.new_cache.delete_recursive(&this_hash);
                self.new_cache.replace_hashed(&this_hash, &folder.parent, &flat);
            }
        }

        // Compact if too many children...
        if !into.compacted {
            let done_compact = Metrics::time(Metric::CompactFolder);
            self.new_cache.reduce_children_of(
                &this_hash,
                DATA_SCANNER_COMPACT_AT_CHILDREN,
                self.new_cache.info.name != folder.name,
            );
            done_compact();
        }

        if self.update_cache.cache.contains_key(&this_hash.key()) && !was_compacted {
            // Replace if existed before.
            if let Some(flat) = self.new_cache.size_recursive(&this_hash.key()) {
                self.update_cache.delete_recursive(&this_hash);
                self.update_cache.replace_hashed(&this_hash, &folder.parent, &flat);
            }
        }

        done_folder();

        Ok(())
    }

    pub fn as_mut_new_cache(&mut self) -> &mut DataUsageCache {
        &mut self.new_cache
    }
}

/// Scan a data folder
/// This function scans the basepath+cache.info.name and returns an updated cache.
/// The returned cache will always be valid, but may not be updated from the existing.
/// Throttling between operations is controlled by the provided [`DynamicSleeper`].
/// If the supplied context is canceled the function will return at the first chance.
#[allow(clippy::too_many_arguments)]
pub async fn scan_data_folder(
    ctx: CancellationToken,
    disks: Vec<Arc<Disk>>,
    local_disk: Arc<Disk>,
    cache: DataUsageCache,
    updates: Option<mpsc::Sender<DataUsageEntry>>,
    scan_mode: HealScanMode,
    sleeper: DynamicSleeper,
) -> Result<DataUsageCache, ScannerError> {
    use crate::data_usage_define::DATA_USAGE_ROOT;

    ensure_scanner_inline_heal_metric_registered();

    // Check that we're not trying to scan the root
    if cache.info.name.is_empty() || cache.info.name == DATA_USAGE_ROOT {
        return Err(ScannerError::Other("internal error: root scan attempted".to_string()));
    }

    // Get disk path
    let base_path = local_disk.path().to_string_lossy().to_string();

    let (update_current_path, close_disk) = current_path_updater(&base_path, &cache.info.name);

    // Create skip_heal flag
    let is_erasure_mode = is_erasure().await;
    let skip_heal = Arc::new(std::sync::atomic::AtomicBool::new(!is_erasure_mode || cache.info.skip_healing));

    // Create heal_object_select flag
    let heal_object_select = if is_erasure_mode && !cache.info.skip_healing {
        heal_object_select_prob()
    } else {
        0
    };

    let disks_quorum = disks.len() / 2;

    let failed_object_ttl = rustfs_utils::get_env_u32(ENV_FAILED_OBJECT_TTL_SECS, DEFAULT_FAILED_OBJECT_TTL_SECS) as u64;
    let failed_objects_max = rustfs_utils::get_env_u32(ENV_FAILED_OBJECTS_MAX, DEFAULT_FAILED_OBJECTS_MAX) as usize;

    // Create folder scanner
    let mut scanner = FolderScanner {
        root: base_path,
        old_cache: cache.clone(),
        new_cache: DataUsageCache {
            info: cache.info.clone(),
            ..Default::default()
        },
        update_cache: DataUsageCache {
            info: cache.info.clone(),
            ..Default::default()
        },
        data_usage_scanner_debug: false,
        heal_object_select,
        scan_mode,
        failed_object_ttl_secs: failed_object_ttl,
        failed_objects_max,
        sleeper,
        disks,
        disks_quorum,
        updates,
        last_update: SystemTime::UNIX_EPOCH,
        update_current_path,
        skip_heal,
        local_disk,
    };

    // Check if context is cancelled
    if ctx.is_cancelled() {
        return Err(ScannerError::Other("Operation cancelled".to_string()));
    }

    // Read top level in bucket
    let mut root = DataUsageEntry::default();
    let folder = CachedFolder {
        name: cache.info.name.clone(),
        parent: None,
        object_heal_prob_div: 1,
    };

    // Scan the folder
    match scanner.scan_folder(ctx, folder, &mut root).await {
        Ok(()) => {
            // Get the new cache and finalize it
            let new_cache = scanner.as_mut_new_cache();
            new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN);
            new_cache.info.last_update = Some(SystemTime::now());
            new_cache.info.next_cycle = cache.info.next_cycle;

            close_disk().await;
            Ok(new_cache.clone())
        }
        Err(e) => {
            close_disk().await;
            // No useful information, return original cache
            Err(e)
        }
    }
}

#[cfg(test)]
mod tests {
    use crate::SCANNER_SLEEPER;

    use super::*;
    use rustfs_ecstore::disk::{DiskOption, endpoint::Endpoint, new_disk};
    use serial_test::serial;
    #[cfg(unix)]
    use std::os::unix::fs::{PermissionsExt, symlink};
    use std::sync::atomic::AtomicBool;
    use uuid::Uuid;

    async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
        let temp_dir = std::env::temp_dir().join(format!("rustfs-scanner-test-{}", Uuid::new_v4()));
        tokio::fs::create_dir_all(&temp_dir)
            .await
            .expect("failed to create test directory");

        let endpoint = Endpoint::try_from(temp_dir.to_string_lossy().as_ref()).expect("failed to create endpoint");
        let disk = new_disk(
            &endpoint,
            &DiskOption {
                cleanup: false,
                health_check: false,
            },
        )
        .await
        .expect("failed to create disk");

        let update_current_path: UpdateCurrentPathFn = Arc::new(|_: &str| Box::pin(async {}));

        let scanner = FolderScanner {
            root: temp_dir.to_string_lossy().to_string(),
            old_cache: DataUsageCache::default(),
            new_cache: DataUsageCache::default(),
            update_cache: DataUsageCache::default(),
            data_usage_scanner_debug: false,
            heal_object_select: 0,
            scan_mode: HealScanMode::Normal,
            failed_object_ttl_secs: u64::MAX,
            failed_objects_max: usize::MAX,
            sleeper: SCANNER_SLEEPER.clone(),
            disks: Vec::new(),
            disks_quorum: 0,
            updates: None,
            last_update: SystemTime::UNIX_EPOCH,
            update_current_path,
            skip_heal: Arc::new(AtomicBool::new(false)),
            local_disk: disk,
        };

        (scanner, temp_dir)
    }

    struct TestGuard {
        temp_dir: Option<std::path::PathBuf>,
    }

    impl TestGuard {
        fn new(ttl: u64, max: usize, scanner: &mut FolderScanner, temp_dir: std::path::PathBuf) -> Self {
            scanner.failed_object_ttl_secs = ttl;
            scanner.failed_objects_max = max;
            Self {
                temp_dir: Some(temp_dir),
            }
        }
    }

    impl Drop for TestGuard {
        fn drop(&mut self) {
            if let Some(temp_dir) = self.temp_dir.take() {
                let _ = std::fs::remove_dir_all(&temp_dir);
            }
        }
    }

    #[tokio::test]
    #[serial]
    async fn test_should_skip_failed_respects_ttl() {
        let (mut scanner, temp_dir) = build_test_scanner().await;
        let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir);
        let now = FolderScanner::now_secs();

        scanner
            .new_cache
            .info
            .failed_objects
            .insert("recent".to_string(), now.saturating_sub(10));
        scanner
            .new_cache
            .info
            .failed_objects
            .insert("expired".to_string(), now.saturating_sub(120));

        assert!(scanner.should_skip_failed("recent"));
        assert!(!scanner.should_skip_failed("expired"));
    }

    #[tokio::test]
    #[serial]
    async fn test_record_failed_ttl_zero_noop() {
        let (mut scanner, temp_dir) = build_test_scanner().await;
        let _guard = TestGuard::new(0, 100, &mut scanner, temp_dir);

        scanner.record_failed("path1");
        assert!(scanner.new_cache.info.failed_objects.is_empty());

        let now = FolderScanner::now_secs();
        scanner.new_cache.info.failed_objects.insert("path2".to_string(), now);
        assert!(!scanner.should_skip_failed("path2"));
    }

    #[tokio::test]
    #[serial]
    async fn test_record_failed_prunes_to_max_entries() {
        let (mut scanner, temp_dir) = build_test_scanner().await;
        let _guard = TestGuard::new(1000, 2, &mut scanner, temp_dir);
        let now = FolderScanner::now_secs();

        scanner
            .new_cache
            .info
            .failed_objects
            .insert("old1".to_string(), now.saturating_sub(50));
        scanner
            .new_cache
            .info
            .failed_objects
            .insert("old2".to_string(), now.saturating_sub(40));
        scanner
            .new_cache
            .info
            .failed_objects
            .insert("old3".to_string(), now.saturating_sub(30));

        scanner.record_failed("new");

        assert_eq!(scanner.new_cache.info.failed_objects.len(), 2);
        assert!(scanner.new_cache.info.failed_objects.contains_key("new"));
        assert!(scanner.new_cache.info.failed_objects.contains_key("old3"));
        assert!(!scanner.new_cache.info.failed_objects.contains_key("old1"));
        assert!(!scanner.new_cache.info.failed_objects.contains_key("old2"));
    }

    #[tokio::test]
    #[serial]
    async fn test_prune_failed_objects_cache_drops_expired() {
        let (mut scanner, temp_dir) = build_test_scanner().await;
        let _guard = TestGuard::new(5, 10, &mut scanner, temp_dir);
        let now = FolderScanner::now_secs();

        scanner
            .new_cache
            .info
            .failed_objects
            .insert("expired".to_string(), now.saturating_sub(10));
        scanner
            .new_cache
            .info
            .failed_objects
            .insert("fresh".to_string(), now.saturating_sub(2));

        scanner.prune_failed_objects_cache();

        assert_eq!(scanner.new_cache.info.failed_objects.len(), 1);
        assert!(scanner.new_cache.info.failed_objects.contains_key("fresh"));
    }

    #[tokio::test]
    #[serial]
    async fn test_prune_failed_objects_max_zero_keeps_fresh() {
        let (mut scanner, temp_dir) = build_test_scanner().await;
        let _guard = TestGuard::new(60, 0, &mut scanner, temp_dir);
        let now = FolderScanner::now_secs();

        scanner
            .new_cache
            .info
            .failed_objects
            .insert("fresh1".to_string(), now.saturating_sub(5));
        scanner
            .new_cache
            .info
            .failed_objects
            .insert("fresh2".to_string(), now.saturating_sub(10));
        scanner
            .new_cache
            .info
            .failed_objects
            .insert("expired".to_string(), now.saturating_sub(120));

        scanner.prune_failed_objects_cache();

        assert_eq!(scanner.new_cache.info.failed_objects.len(), 2);
        assert!(scanner.new_cache.info.failed_objects.contains_key("fresh1"));
        assert!(scanner.new_cache.info.failed_objects.contains_key("fresh2"));
        assert!(!scanner.new_cache.info.failed_objects.contains_key("expired"));
    }

    #[test]
    fn test_scanner_inline_heal_enabled_defaults_to_false() {
        assert!(!scanner_inline_heal_enabled_from_value(None));
    }

    #[test]
    fn test_scanner_inline_heal_enabled_reads_env_override() {
        assert!(scanner_inline_heal_enabled_from_value(Some("true")));
        assert!(scanner_inline_heal_enabled_from_value(Some("YES")));
        assert!(scanner_inline_heal_enabled_from_value(Some("1")));
        assert!(!scanner_inline_heal_enabled_from_value(Some("false")));
    }

    #[test]
    fn test_build_object_heal_request_omits_nil_version_id() {
        let request = build_object_heal_request(
            "bucket".to_string(),
            "path/to/object".to_string(),
            None,
            HealScanMode::Deep,
            HealChannelPriority::Low,
        );

        assert_eq!(request.bucket, "bucket");
        assert_eq!(request.object_prefix.as_deref(), Some("path/to/object"));
        assert!(request.object_version_id.is_none());
        assert_eq!(request.scan_mode, Some(HealScanMode::Deep));
        assert_eq!(request.priority, HealChannelPriority::Low);
        assert_eq!(request.remove_corrupted, Some(HEAL_DELETE_DANGLING));
    }

    #[test]
    fn test_heal_priority_label_matches_priority_names() {
        assert_eq!(heal_priority_label(HealChannelPriority::Low), "low");
        assert_eq!(heal_priority_label(HealChannelPriority::Normal), "normal");
        assert_eq!(heal_priority_label(HealChannelPriority::High), "high");
        assert_eq!(heal_priority_label(HealChannelPriority::Critical), "critical");
    }

    #[test]
    fn test_describe_heal_admission_formats_unadmitted_results() {
        assert_eq!(describe_heal_admission(HealAdmissionResult::Accepted), "accepted");
        assert_eq!(describe_heal_admission(HealAdmissionResult::Merged), "merged");
        assert_eq!(describe_heal_admission(HealAdmissionResult::Full), "queue_full");
        assert_eq!(
            describe_heal_admission(HealAdmissionResult::Dropped(
                rustfs_common::heal_channel::HealAdmissionDropReason::QueueFull
            )),
            "dropped:queue_full"
        );
    }

    #[test]
    fn test_build_high_priority_heal_admission_error_contains_context() {
        let err = build_high_priority_heal_admission_error(
            "object",
            "bucket-a",
            Some("path/to/object"),
            HealChannelPriority::High,
            HealAdmissionResult::Full,
        );

        let err_text = err.to_string();
        assert!(err_text.contains("type=object"));
        assert!(err_text.contains("bucket='bucket-a'"));
        assert!(err_text.contains("object='path/to/object'"));
        assert!(err_text.contains("priority=high"));
        assert!(err_text.contains("admission=queue_full"));
    }

    #[tokio::test]
    async fn test_heal_actions_returns_actual_size_without_inline_heal() {
        let temp_dir = std::env::temp_dir();
        let file_type = std::fs::metadata(&temp_dir).unwrap().file_type();

        let mut item = ScannerItem {
            path: temp_dir.join("object").to_string_lossy().to_string(),
            bucket: "bucket".to_string(),
            prefix: "".to_string(),
            object_name: "object".to_string(),
            file_type,
            lifecycle: None,
            replication: None,
            heal_enabled: true,
            heal_bitrot: true,
            debug: false,
        };
        let object_info = ObjectInfo {
            bucket: "bucket".to_string(),
            name: "object".to_string(),
            ..Default::default()
        };
        let mut size_summary = SizeSummary::default();

        let size = item.heal_actions(&object_info, 123, &mut size_summary).await;
        assert_eq!(size, 123);
    }

    #[tokio::test]
    #[serial]
    #[cfg(unix)]
    async fn test_scan_folder_skips_unreadable_child_directory() {
        let (mut scanner, temp_dir) = build_test_scanner().await;
        let _guard = TestGuard::new(60, 0, &mut scanner, temp_dir.clone());

        let bucket_dir = temp_dir.join("bucket");
        let good_dir = bucket_dir.join("good");
        let bad_dir = bucket_dir.join("bad");

        std::fs::create_dir_all(&good_dir).expect("failed to create good dir");
        std::fs::create_dir_all(&bad_dir).expect("failed to create bad dir");
        std::fs::set_permissions(&bad_dir, std::fs::Permissions::from_mode(0o000)).expect("failed to remove bad dir permissions");

        scanner.old_cache.info.name = "bucket".to_string();
        scanner.new_cache.info.name = "bucket".to_string();
        scanner.update_cache.info.name = "bucket".to_string();

        let folder = CachedFolder {
            name: "bucket".to_string(),
            parent: None,
            object_heal_prob_div: 1,
        };

        let mut into = DataUsageEntry::default();
        let result = scanner.scan_folder(CancellationToken::new(), folder, &mut into).await;

        std::fs::set_permissions(&bad_dir, std::fs::Permissions::from_mode(0o755))
            .expect("failed to restore bad dir permissions");

        assert!(result.is_ok(), "expected unreadable child directory to be skipped");
    }

    #[tokio::test]
    #[serial]
    #[cfg(unix)]
    async fn test_scan_folder_ignores_symlinked_child_directory() {
        let (mut scanner, temp_dir) = build_test_scanner().await;
        let _guard = TestGuard::new(60, 0, &mut scanner, temp_dir.clone());

        let bucket_dir = temp_dir.join("bucket");
        let target_dir = bucket_dir.join("target");
        let link_dir = bucket_dir.join("link");

        std::fs::create_dir_all(&target_dir).expect("failed to create target dir");
        symlink(&target_dir, &link_dir).expect("failed to create symlinked dir");

        scanner.old_cache.info.name = "bucket".to_string();
        scanner.new_cache.info.name = "bucket".to_string();
        scanner.update_cache.info.name = "bucket".to_string();

        let folder = CachedFolder {
            name: "bucket".to_string(),
            parent: None,
            object_heal_prob_div: 1,
        };

        let mut into = DataUsageEntry::default();
        let result = scanner.scan_folder(CancellationToken::new(), folder, &mut into).await;

        assert!(result.is_ok(), "expected symlinked child directory to be ignored");
        assert_eq!(into.failed_objects, 0, "expected ignored symlink not to count as a failed object");
    }
}
