// 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 crate::module_switches::{heal_enabled_from_env, scanner_enabled_from_env};
use crate::storage::storage_api::runtime_sources_consumer::EndpointServerPools;
use jiff::Timestamp;
use rmp_serde::Deserializer;
use rustfs_heal::HealOperationsSnapshot;
use rustfs_heal_contracts::heal_channel::HealScanMode;
use rustfs_scanner::scanner::BackgroundHealInfo;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::io::Cursor;

use super::super::encode_msgpack_map;

const NODE_HEAL_STATUS_PREVIOUS_VERSION: u8 = 1;
const NODE_HEAL_STATUS_VERSION: u8 = 2;
const NODE_HEAL_STATUS_MAX_SIZE: usize = 64 * 1024;
const NODE_REPLACEMENT_RECOVERY_STATUS_VERSION: u8 = 1;
const NODE_REPLACEMENT_RECOVERY_STATUS_MAX_SIZE: usize = 64 * 1024;
const HEAL_TOPOLOGY_FINGERPRINT_DOMAIN: &[u8] = b"rustfs-heal-topology-v1\0";

fn chrono_to_jiff_timestamp(timestamp: chrono::DateTime<chrono::Utc>) -> Timestamp {
    let seconds = timestamp.timestamp();
    let nanoseconds = match i32::try_from(timestamp.timestamp_subsec_nanos()) {
        Ok(nanoseconds) => nanoseconds,
        Err(_) => {
            return if seconds < 0 { Timestamp::MIN } else { Timestamp::MAX };
        }
    };

    match Timestamp::new(seconds, nanoseconds) {
        Ok(timestamp) => timestamp,
        Err(_) => {
            if seconds < 0 {
                Timestamp::MIN
            } else {
                Timestamp::MAX
            }
        }
    }
}

fn jiff_to_chrono_datetime(timestamp: Timestamp) -> chrono::DateTime<chrono::Utc> {
    chrono::DateTime::<chrono::Utc>::from(std::time::SystemTime::from(timestamp))
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct HealControlCoordinator {
    pub grid_host: String,
    pub is_local: bool,
}

pub(crate) fn heal_control_coordinator(endpoint_pools: &EndpointServerPools) -> Result<HealControlCoordinator, String> {
    endpoint_pools
        .get_nodes()
        .into_iter()
        .next()
        .map(|node| HealControlCoordinator {
            grid_host: node.grid_host,
            is_local: node.is_local,
        })
        .ok_or_else(|| "heal control topology has no coordinator".to_string())
}

fn hash_sized(hasher: &mut Sha256, value: &[u8]) -> Result<(), String> {
    let len = u64::try_from(value.len()).map_err(|_| "heal topology field length cannot be represented".to_string())?;
    hasher.update(len.to_be_bytes());
    hasher.update(value);
    Ok(())
}

/// Returns a node-independent digest of the complete pool/set/disk layout.
/// Locality is deliberately excluded because it differs on every peer.
pub(crate) fn heal_topology_fingerprint(endpoint_pools: &EndpointServerPools) -> Result<String, String> {
    if endpoint_pools.as_ref().is_empty() {
        return Err("heal topology has no storage pools".to_string());
    }
    let mut hasher = Sha256::new();
    hasher.update(HEAL_TOPOLOGY_FINGERPRINT_DOMAIN);
    hasher.update(
        u64::try_from(endpoint_pools.as_ref().len())
            .map_err(|_| "heal topology pool count cannot be represented".to_string())?
            .to_be_bytes(),
    );

    for (pool_index, pool) in endpoint_pools.as_ref().iter().enumerate() {
        if pool.set_count == 0 || pool.drives_per_set == 0 {
            return Err(format!("heal topology pool {pool_index} has an empty set or drive layout"));
        }
        let expected_endpoints = pool
            .set_count
            .checked_mul(pool.drives_per_set)
            .ok_or_else(|| "heal topology endpoint count overflow".to_string())?;
        if pool.endpoints.as_ref().len() != expected_endpoints {
            return Err(format!(
                "heal topology pool {pool_index} has {} endpoints, expected {expected_endpoints}",
                pool.endpoints.as_ref().len()
            ));
        }

        hasher.update(
            u64::try_from(pool_index)
                .map_err(|_| "heal topology pool index cannot be represented")?
                .to_be_bytes(),
        );
        hasher.update(
            u64::try_from(pool.set_count)
                .map_err(|_| "heal topology set count cannot be represented")?
                .to_be_bytes(),
        );
        hasher.update(
            u64::try_from(pool.drives_per_set)
                .map_err(|_| "heal topology drive count cannot be represented")?
                .to_be_bytes(),
        );
        hasher.update([u8::from(pool.legacy)]);

        let mut entries = Vec::with_capacity(pool.endpoints.as_ref().len());
        for (endpoint_index, endpoint) in pool.endpoints.as_ref().iter().enumerate() {
            let declared_pool = usize::try_from(endpoint.pool_idx)
                .map_err(|_| format!("heal topology endpoint {endpoint_index} has an invalid pool index"))?;
            if declared_pool != pool_index {
                return Err(format!("heal topology endpoint declares pool {declared_pool}, expected {pool_index}"));
            }
            let set_index = usize::try_from(endpoint.set_idx)
                .map_err(|_| format!("heal topology endpoint {endpoint_index} has an invalid set index"))?;
            let disk_index = usize::try_from(endpoint.disk_idx)
                .map_err(|_| format!("heal topology endpoint {endpoint_index} has an invalid disk index"))?;
            if set_index >= pool.set_count || disk_index >= pool.drives_per_set {
                return Err(format!(
                    "heal topology endpoint position {pool_index}/{set_index}/{disk_index} is out of range"
                ));
            }
            entries.push((set_index, disk_index, endpoint.url.as_str()));
        }
        entries.sort_by(|left, right| left.0.cmp(&right.0).then(left.1.cmp(&right.1)));
        for pair in entries.windows(2) {
            if pair[0].0 == pair[1].0 && pair[0].1 == pair[1].1 {
                return Err(format!(
                    "heal topology contains duplicate endpoint position {pool_index}/{}/{}",
                    pair[0].0, pair[0].1
                ));
            }
        }
        for (set_index, disk_index, endpoint) in entries {
            hasher.update(
                u64::try_from(set_index)
                    .map_err(|_| "heal topology set index cannot be represented")?
                    .to_be_bytes(),
            );
            hasher.update(
                u64::try_from(disk_index)
                    .map_err(|_| "heal topology disk index cannot be represented")?
                    .to_be_bytes(),
            );
            hash_sized(&mut hasher, endpoint.as_bytes())?;
        }
    }

    Ok(hex_simd::encode_to_string(hasher.finalize(), hex_simd::AsciiCase::Lower))
}

pub(crate) type NodeHealProgress = rustfs_heal::HealProgress;

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct NodeHealProgressV1 {
    objects_scanned: u64,
    objects_healed: u64,
    objects_failed: u64,
    bytes_processed: u64,
}

impl From<&NodeHealProgress> for NodeHealProgressV1 {
    fn from(progress: &NodeHealProgress) -> Self {
        Self {
            objects_scanned: progress.objects_scanned,
            objects_healed: progress.objects_healed,
            objects_failed: progress.objects_failed,
            bytes_processed: progress.bytes_processed,
        }
    }
}

impl From<NodeHealProgressV1> for NodeHealProgress {
    fn from(progress: NodeHealProgressV1) -> Self {
        Self {
            objects_scanned: progress.objects_scanned,
            objects_healed: progress.objects_healed,
            objects_failed: progress.objects_failed,
            bytes_processed: progress.bytes_processed,
            ..Default::default()
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct NodeHealInfo {
    bitrot_start_time: Option<Timestamp>,
    bitrot_start_cycle: u64,
    current_scan_mode: HealScanMode,
}

impl From<BackgroundHealInfo> for NodeHealInfo {
    fn from(info: BackgroundHealInfo) -> Self {
        Self {
            bitrot_start_time: info.bitrot_start_time.map(chrono_to_jiff_timestamp),
            bitrot_start_cycle: info.bitrot_start_cycle,
            current_scan_mode: info.current_scan_mode,
        }
    }
}

impl From<NodeHealInfo> for BackgroundHealInfo {
    fn from(info: NodeHealInfo) -> Self {
        Self {
            bitrot_start_time: info.bitrot_start_time.map(jiff_to_chrono_datetime),
            bitrot_start_cycle: info.bitrot_start_cycle,
            current_scan_mode: info.current_scan_mode,
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub(crate) struct NodeHealStatusSnapshot {
    version: u8,
    pub services_enabled: bool,
    pub initialized: bool,
    info: NodeHealInfo,
    pub operations: HealOperationsSnapshot,
    pub progress: Option<NodeHealProgress>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct NodeHealStatusSnapshotV1 {
    version: u8,
    services_enabled: bool,
    initialized: bool,
    info: NodeHealInfo,
    operations: HealOperationsSnapshot,
    progress: Option<NodeHealProgressV1>,
}

impl From<&NodeHealStatusSnapshot> for NodeHealStatusSnapshotV1 {
    fn from(snapshot: &NodeHealStatusSnapshot) -> Self {
        Self {
            version: NODE_HEAL_STATUS_PREVIOUS_VERSION,
            services_enabled: snapshot.services_enabled,
            initialized: snapshot.initialized,
            info: snapshot.info.clone(),
            operations: snapshot.operations,
            progress: snapshot.progress.as_ref().map(NodeHealProgressV1::from),
        }
    }
}

impl From<NodeHealStatusSnapshotV1> for NodeHealStatusSnapshot {
    fn from(snapshot: NodeHealStatusSnapshotV1) -> Self {
        Self {
            version: snapshot.version,
            services_enabled: snapshot.services_enabled,
            initialized: snapshot.initialized,
            info: snapshot.info,
            operations: snapshot.operations,
            progress: snapshot.progress.map(NodeHealProgress::from),
        }
    }
}

#[derive(Deserialize)]
struct NodeHealStatusVersion {
    version: u8,
}

impl NodeHealStatusSnapshot {
    #[cfg(test)]
    pub(crate) fn for_test(
        services_enabled: bool,
        initialized: bool,
        info: BackgroundHealInfo,
        operations: HealOperationsSnapshot,
        progress: Option<NodeHealProgress>,
    ) -> Self {
        Self {
            version: NODE_HEAL_STATUS_VERSION,
            services_enabled,
            initialized,
            info: info.into(),
            operations,
            progress,
        }
    }

    pub(crate) fn info(&self) -> BackgroundHealInfo {
        BackgroundHealInfo {
            bitrot_start_time: self.info.bitrot_start_time.map(jiff_to_chrono_datetime),
            bitrot_start_cycle: self.info.bitrot_start_cycle,
            current_scan_mode: self.info.current_scan_mode,
        }
    }
}

pub(crate) async fn capture_node_heal_status(info: BackgroundHealInfo) -> NodeHealStatusSnapshot {
    let progress = rustfs_heal::current_heal_progress_snapshot().await;

    NodeHealStatusSnapshot {
        version: NODE_HEAL_STATUS_VERSION,
        services_enabled: scanner_enabled_from_env() || heal_enabled_from_env(),
        initialized: rustfs_heal::heal_runtime_initialized(),
        info: info.into(),
        operations: rustfs_heal::current_heal_operations_snapshot().await,
        progress,
    }
}

pub(crate) fn encode_node_heal_status(snapshot: &NodeHealStatusSnapshot, protocol_version: u32) -> Result<Vec<u8>, String> {
    let encoded = if protocol_version < rustfs_protos::BACKGROUND_HEAL_STATUS_PROTOCOL_VERSION {
        encode_msgpack_map(&NodeHealStatusSnapshotV1::from(snapshot))
    } else {
        let mut snapshot = snapshot.clone();
        snapshot.version = NODE_HEAL_STATUS_VERSION;
        encode_msgpack_map(&snapshot)
    };
    encoded.map_err(|err| format!("failed to encode node heal status: {err}"))
}

pub(crate) fn decode_node_heal_status(data: &[u8]) -> Result<NodeHealStatusSnapshot, String> {
    if data.len() > NODE_HEAL_STATUS_MAX_SIZE {
        return Err("node heal status exceeds size limit".to_string());
    }
    let decode_version = || {
        let mut deserializer = Deserializer::new(Cursor::new(data));
        NodeHealStatusVersion::deserialize(&mut deserializer)
            .map(|version| (version, deserializer))
            .map_err(|err| format!("failed to decode node heal status: {err}"))
    };
    let (version, deserializer) = decode_version()?;
    if usize::try_from(deserializer.get_ref().position()).ok() != Some(data.len()) {
        return Err("node heal status contains trailing data".to_string());
    }

    let mut deserializer = Deserializer::new(Cursor::new(data));
    let snapshot = match version.version {
        NODE_HEAL_STATUS_PREVIOUS_VERSION => {
            NodeHealStatusSnapshotV1::deserialize(&mut deserializer).map(NodeHealStatusSnapshot::from)
        }
        NODE_HEAL_STATUS_VERSION => NodeHealStatusSnapshot::deserialize(&mut deserializer),
        version => return Err(format!("unsupported node heal status version: {version}")),
    }
    .map_err(|err| format!("failed to decode node heal status: {err}"))?;
    if usize::try_from(deserializer.get_ref().position()).ok() != Some(data.len()) {
        return Err("node heal status contains trailing data".to_string());
    }
    Ok(snapshot)
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub(crate) struct NodeReplacementRecoveryStatusSnapshot {
    version: u8,
    pub snapshot: rustfs_heal::ReplacementRecoverySnapshot,
}

impl NodeReplacementRecoveryStatusSnapshot {
    pub(crate) fn new(snapshot: rustfs_heal::ReplacementRecoverySnapshot) -> Self {
        Self {
            version: NODE_REPLACEMENT_RECOVERY_STATUS_VERSION,
            snapshot,
        }
    }
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct NodeReplacementRecoveryStatusSnapshotWire {
    version: u8,
    snapshot: ReplacementRecoverySnapshotWire,
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct ReplacementRecoverySnapshotWire {
    records: Vec<ReplacementRecoveryRecordWire>,
    definitive: bool,
    reason: Option<String>,
}

impl From<ReplacementRecoverySnapshotWire> for rustfs_heal::ReplacementRecoverySnapshot {
    fn from(snapshot: ReplacementRecoverySnapshotWire) -> Self {
        Self {
            records: snapshot.records.into_iter().map(Into::into).collect(),
            definitive: snapshot.definitive,
            reason: snapshot.reason,
        }
    }
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct ReplacementRecoveryRecordWire {
    task_id: String,
    state: rustfs_heal::ReplacementRecoveryState,
    generation: Option<String>,
    set_disk_id: Option<String>,
    target_slots: Vec<String>,
    reason: Option<String>,
    verified_at: Option<u64>,
}

impl From<ReplacementRecoveryRecordWire> for rustfs_heal::ReplacementRecoveryRecord {
    fn from(record: ReplacementRecoveryRecordWire) -> Self {
        Self {
            task_id: record.task_id,
            state: record.state,
            generation: record.generation,
            set_disk_id: record.set_disk_id,
            target_slots: record.target_slots,
            reason: record.reason,
            verified_at: record.verified_at,
        }
    }
}

pub(crate) async fn capture_node_replacement_recovery_status() -> NodeReplacementRecoveryStatusSnapshot {
    NodeReplacementRecoveryStatusSnapshot::new(rustfs_heal::current_replacement_recovery_snapshot().await)
}

pub(crate) fn encode_node_replacement_recovery_status(
    snapshot: &NodeReplacementRecoveryStatusSnapshot,
) -> Result<Vec<u8>, String> {
    encode_msgpack_map(snapshot).map_err(|err| format!("failed to encode replacement recovery status: {err}"))
}

pub(crate) fn decode_node_replacement_recovery_status(data: &[u8]) -> Result<NodeReplacementRecoveryStatusSnapshot, String> {
    if data.len() > NODE_REPLACEMENT_RECOVERY_STATUS_MAX_SIZE {
        return Err("replacement recovery status exceeds size limit".to_string());
    }
    let mut deserializer = Deserializer::new(Cursor::new(data));
    let wire = NodeReplacementRecoveryStatusSnapshotWire::deserialize(&mut deserializer)
        .map_err(|err| format!("failed to decode replacement recovery status: {err}"))?;
    if usize::try_from(deserializer.get_ref().position()).ok() != Some(data.len()) {
        return Err("replacement recovery status contains trailing data".to_string());
    }
    if wire.version != NODE_REPLACEMENT_RECOVERY_STATUS_VERSION {
        return Err(format!("unsupported replacement recovery status version: {}", wire.version));
    }
    Ok(NodeReplacementRecoveryStatusSnapshot {
        version: wire.version,
        snapshot: wire.snapshot.into(),
    })
}

#[cfg(test)]
mod tests {
    use super::{
        NODE_HEAL_STATUS_MAX_SIZE, NODE_HEAL_STATUS_PREVIOUS_VERSION, NODE_HEAL_STATUS_VERSION, NodeHealProgress,
        NodeHealStatusSnapshot, NodeReplacementRecoveryStatusSnapshot, decode_node_heal_status,
        decode_node_replacement_recovery_status, encode_node_heal_status, encode_node_replacement_recovery_status,
        heal_control_coordinator, heal_topology_fingerprint,
    };
    use crate::storage::storage_api::{
        Endpoint,
        ecstore_layout::{EndpointServerPools, Endpoints, PoolEndpoints},
    };
    use chrono::SecondsFormat;
    use rustfs_heal::{HealOperationsSnapshot, ReplacementRecoveryRecord, ReplacementRecoverySnapshot, ReplacementRecoveryState};
    use rustfs_scanner::scanner::BackgroundHealInfo;

    fn topology_endpoints(last_host: &str) -> EndpointServerPools {
        let endpoints = ["node-a", "node-b", "node-c", last_host]
            .into_iter()
            .enumerate()
            .map(|(index, host)| {
                let mut endpoint = Endpoint::try_from(format!("http://{host}:9000/disk{}", index + 1).as_str())
                    .expect("test endpoint should parse");
                endpoint.set_pool_index(0);
                endpoint.set_set_index(index / 2);
                endpoint.set_disk_index(index % 2);
                endpoint.is_local = index == 0;
                endpoint
            })
            .collect::<Vec<_>>();
        EndpointServerPools::from(vec![PoolEndpoints {
            legacy: false,
            set_count: 2,
            drives_per_set: 2,
            endpoints: Endpoints::from(endpoints),
            cmd_line: String::new(),
            platform: String::new(),
        }])
    }

    #[test]
    fn heal_topology_fingerprint_is_node_independent_and_layout_complete() {
        let topology = topology_endpoints("node-d");
        let expected = heal_topology_fingerprint(&topology).expect("valid topology should hash");

        let mut other_node_view = topology;
        for endpoint in other_node_view.as_mut()[0].endpoints.as_mut() {
            endpoint.is_local = endpoint.host_port() == "node-c:9000";
        }
        assert_eq!(
            heal_topology_fingerprint(&other_node_view).expect("locality must not affect the hash"),
            expected
        );

        let mut reordered = topology_endpoints("node-d");
        reordered.as_mut()[0].endpoints.as_mut().reverse();
        assert_eq!(
            heal_topology_fingerprint(&reordered).expect("explicit positions must normalize endpoint order"),
            expected
        );

        let mut legacy = topology_endpoints("node-d");
        legacy.as_mut()[0].legacy = true;
        assert_ne!(heal_topology_fingerprint(&legacy).expect("legacy topology should hash"), expected);

        let changed = topology_endpoints("node-e");
        assert_ne!(heal_topology_fingerprint(&changed).expect("changed topology should still hash"), expected);
    }

    #[test]
    fn heal_control_coordinator_is_stable_across_node_views() {
        let topology = topology_endpoints("node-d");
        let coordinator = heal_control_coordinator(&topology).expect("valid topology should select a coordinator");
        assert_eq!(coordinator.grid_host, "http://node-a:9000");
        assert!(coordinator.is_local);

        let mut other_node_view = topology;
        for endpoint in other_node_view.as_mut()[0].endpoints.as_mut() {
            endpoint.is_local = endpoint.host_port() == "node-c:9000";
        }
        let coordinator = heal_control_coordinator(&other_node_view).expect("all nodes should select the same coordinator");
        assert_eq!(coordinator.grid_host, "http://node-a:9000");
        assert!(!coordinator.is_local);
        assert!(heal_control_coordinator(&EndpointServerPools::default()).is_err());
    }

    #[test]
    fn heal_topology_fingerprint_rejects_duplicate_positions() {
        let mut topology = topology_endpoints("node-d");
        topology.as_mut()[0].endpoints.as_mut()[1].set_set_index(0);
        topology.as_mut()[0].endpoints.as_mut()[1].set_disk_index(0);
        let err = heal_topology_fingerprint(&topology).expect_err("duplicate position must fail closed");
        assert!(err.contains("duplicate endpoint position"));
    }

    #[test]
    fn heal_topology_fingerprint_rejects_uninitialized_indices_and_empty_layouts() {
        let mut uninitialized = topology_endpoints("node-d");
        uninitialized.as_mut()[0].endpoints.as_mut()[0].pool_idx = -1;
        let err = heal_topology_fingerprint(&uninitialized).expect_err("negative index must fail closed");
        assert!(err.contains("invalid pool index"));

        let empty = EndpointServerPools::default();
        let err = heal_topology_fingerprint(&empty).expect_err("empty topology must fail closed");
        assert!(err.contains("no storage pools"));

        let mut cases = Vec::new();
        let mut zero_sets = topology_endpoints("node-d");
        zero_sets.as_mut()[0].set_count = 0;
        cases.push(("zero sets", zero_sets, "empty set or drive layout"));
        let mut zero_drives = topology_endpoints("node-d");
        zero_drives.as_mut()[0].drives_per_set = 0;
        cases.push(("zero drives", zero_drives, "empty set or drive layout"));
        let mut missing_endpoint = topology_endpoints("node-d");
        missing_endpoint.as_mut()[0].endpoints.as_mut().pop();
        cases.push(("missing endpoint", missing_endpoint, "has 3 endpoints, expected 4"));
        let mut wrong_pool = topology_endpoints("node-d");
        wrong_pool.as_mut()[0].endpoints.as_mut()[0].set_pool_index(1);
        cases.push(("wrong pool", wrong_pool, "declares pool 1, expected 0"));
        let mut negative_set = topology_endpoints("node-d");
        negative_set.as_mut()[0].endpoints.as_mut()[0].set_idx = -1;
        cases.push(("negative set", negative_set, "invalid set index"));
        let mut negative_disk = topology_endpoints("node-d");
        negative_disk.as_mut()[0].endpoints.as_mut()[0].disk_idx = -1;
        cases.push(("negative disk", negative_disk, "invalid disk index"));
        let mut out_of_range_set = topology_endpoints("node-d");
        out_of_range_set.as_mut()[0].endpoints.as_mut()[0].set_set_index(2);
        cases.push(("set out of range", out_of_range_set, "is out of range"));
        let mut out_of_range_disk = topology_endpoints("node-d");
        out_of_range_disk.as_mut()[0].endpoints.as_mut()[0].set_disk_index(2);
        cases.push(("disk out of range", out_of_range_disk, "is out of range"));

        for (name, topology, expected) in cases {
            let err = heal_topology_fingerprint(&topology).expect_err(name);
            assert!(err.contains(expected), "{name}: {err}");
        }
    }

    #[test]
    fn node_heal_status_round_trip_is_versioned() {
        let snapshot = NodeHealStatusSnapshot::for_test(
            true,
            true,
            BackgroundHealInfo::default(),
            HealOperationsSnapshot {
                queue_length: 2,
                active_tasks: 1,
                ..Default::default()
            },
            Some(NodeHealProgress {
                kind: rustfs_heal::heal::progress::HealProgressKind::ObjectSweep,
                objects_scanned: 7,
                objects_healed: 5,
                objects_failed: 1,
                skipped_objects: 1,
                objects_total_count: 10,
                objects_total_size: 2048,
                bytes_processed: 1024,
                progress_percentage: 50.0,
                progress_state: rustfs_heal::heal::progress::HealProgressState::Running,
                baseline_generation: Some(42),
                baseline_known: true,
                ..Default::default()
            }),
        );

        let encoded = encode_node_heal_status(&snapshot, rustfs_protos::BACKGROUND_HEAL_STATUS_PROTOCOL_VERSION)
            .expect("snapshot should encode");
        let decoded = decode_node_heal_status(&encoded).expect("snapshot should decode");
        assert_eq!(decoded.version, NODE_HEAL_STATUS_VERSION);
        assert_eq!(decoded.operations.queue_length, 2);
        assert_eq!(decoded.progress, snapshot.progress);
    }

    #[test]
    fn node_heal_status_v1_encoding_preserves_rolling_compatibility() {
        let snapshot = NodeHealStatusSnapshot::for_test(
            true,
            true,
            BackgroundHealInfo::default(),
            HealOperationsSnapshot::default(),
            Some(NodeHealProgress {
                objects_scanned: 7,
                objects_healed: 5,
                objects_failed: 1,
                skipped_objects: 3,
                bytes_processed: 1024,
                progress_state: rustfs_heal::heal::progress::HealProgressState::Running,
                baseline_known: true,
                ..Default::default()
            }),
        );

        let encoded =
            encode_node_heal_status(&snapshot, u32::from(NODE_HEAL_STATUS_PREVIOUS_VERSION)).expect("v1 snapshot should encode");
        let wire: serde_json::Value = rmp_serde::from_slice(&encoded).expect("v1 snapshot should decode as JSON");
        let progress = wire["progress"].as_object().expect("v1 progress should be a map");
        assert_eq!(wire["version"], NODE_HEAL_STATUS_PREVIOUS_VERSION);
        assert_eq!(progress.len(), 4);
        assert_eq!(progress["objectsScanned"], 7);
        assert!(!progress.contains_key("skippedObjects"));

        let decoded = decode_node_heal_status(&encoded).expect("v1 snapshot should decode");
        let progress = decoded.progress.as_ref().expect("v1 progress should be present");
        assert_eq!(decoded.version, NODE_HEAL_STATUS_PREVIOUS_VERSION);
        assert_eq!(progress.objects_scanned, 7);
        assert_eq!(progress.progress_state, rustfs_heal::heal::progress::HealProgressState::Unknown);
        assert!(!progress.baseline_known);

        let upgraded = encode_node_heal_status(&decoded, rustfs_protos::BACKGROUND_HEAL_STATUS_PROTOCOL_VERSION)
            .expect("decoded v1 snapshot should upgrade to v2");
        let upgraded = decode_node_heal_status(&upgraded).expect("upgraded snapshot should decode");
        assert_eq!(upgraded.version, NODE_HEAL_STATUS_VERSION);
        assert_eq!(upgraded.progress.as_ref(), Some(progress));
    }

    #[test]
    fn node_heal_status_rejects_unknown_version() {
        let mut snapshot = NodeHealStatusSnapshot::for_test(
            false,
            false,
            BackgroundHealInfo::default(),
            HealOperationsSnapshot::default(),
            None,
        );
        snapshot.version += 1;

        let encoded = rmp_serde::to_vec_named(&snapshot).expect("snapshot should encode");
        let err = decode_node_heal_status(&encoded).expect_err("unknown version should fail closed");
        assert!(err.contains("unsupported node heal status version"));
    }

    #[test]
    fn node_heal_status_accepts_fixed_v1_wire_keys() {
        let fixture = serde_json::json!({
            "version": 1,
            "servicesEnabled": true,
            "initialized": true,
            "info": {"bitrotStartTime": null, "bitrotStartCycle": 9, "currentScanMode": 1},
            "operations": {
                "queueLength": 2, "activeTasks": 1, "retryingTasks": 0,
                "queuedByPriority": {"low": 0, "normal": 2, "high": 0, "urgent": 0},
                "activeByPriority": {"low": 0, "normal": 0, "high": 1, "urgent": 0},
                "retryingByPriority": {"low": 0, "normal": 0, "high": 0, "urgent": 0},
                "queuedBySource": {"scanner": 2, "admin": 0, "autoHeal": 0, "internal": 0, "readRepair": 0},
                "activeBySource": {"scanner": 0, "admin": 1, "autoHeal": 0, "internal": 0, "readRepair": 0},
                "retryingBySource": {"scanner": 0, "admin": 0, "autoHeal": 0, "internal": 0, "readRepair": 0}
            },
            "progress": {
                "objectsScanned": 7,
                "objectsHealed": 5,
                "objectsFailed": 1,
                "bytesProcessed": 1024
            }
        });
        let encoded = rmp_serde::to_vec_named(&fixture).expect("fixture should encode");
        let decoded = decode_node_heal_status(&encoded).expect("fixed v1 fixture should decode");
        assert_eq!(decoded.info().bitrot_start_cycle, 9);
        assert_eq!(decoded.operations.queue_length, 2);
        assert_eq!(decoded.operations.queued_by_source.mrf, 0);
        let progress = decoded.progress.expect("legacy progress should decode");
        assert_eq!(progress.objects_scanned, 7);
        assert!(!progress.baseline_known);
        assert_eq!(progress.progress_state, rustfs_heal::heal::progress::HealProgressState::Unknown);
    }

    #[test]
    fn node_heal_status_timestamp_json_remains_rfc3339() {
        let fixture = serde_json::json!({
            "version": 1,
            "servicesEnabled": true,
            "initialized": true,
            "info": {"bitrotStartTime": "2023-11-14T22:13:20.123456Z", "bitrotStartCycle": 9, "currentScanMode": 1},
            "operations": {
                "queueLength": 2, "activeTasks": 1, "retryingTasks": 0,
                "queuedByPriority": {"low": 0, "normal": 2, "high": 0, "urgent": 0},
                "activeByPriority": {"low": 0, "normal": 0, "high": 1, "urgent": 0},
                "retryingByPriority": {"low": 0, "normal": 0, "high": 0, "urgent": 0},
                "queuedBySource": {"scanner": 2, "admin": 0, "autoHeal": 0, "internal": 0, "readRepair": 0},
                "activeBySource": {"scanner": 0, "admin": 1, "autoHeal": 0, "internal": 0, "readRepair": 0},
                "retryingBySource": {"scanner": 0, "admin": 0, "autoHeal": 0, "internal": 0, "readRepair": 0}
            },
            "progress": null
        });
        let encoded = rmp_serde::to_vec_named(&fixture).expect("fixture should encode");
        let decoded = decode_node_heal_status(&encoded).expect("timestamp fixture should decode");
        let info = decoded.info();

        assert_eq!(
            info.bitrot_start_time
                .expect("bitrot start time should be preserved")
                .to_rfc3339_opts(SecondsFormat::Micros, true),
            "2023-11-14T22:13:20.123456Z"
        );

        let started_at = chrono::DateTime::parse_from_rfc3339("2023-11-14T22:13:20.123456Z")
            .expect("fixture timestamp should parse")
            .with_timezone(&chrono::Utc);
        let snapshot = NodeHealStatusSnapshot::for_test(
            true,
            true,
            BackgroundHealInfo {
                bitrot_start_time: Some(started_at),
                bitrot_start_cycle: 9,
                current_scan_mode: rustfs_heal_contracts::heal_channel::HealScanMode::Deep,
            },
            HealOperationsSnapshot::default(),
            None,
        );
        let encoded = encode_node_heal_status(&snapshot, rustfs_protos::BACKGROUND_HEAL_STATUS_PROTOCOL_VERSION)
            .expect("snapshot should encode");
        let encoded_json: serde_json::Value = rmp_serde::from_slice(&encoded).expect("encoded snapshot should decode as JSON");
        assert_eq!(encoded_json["info"]["bitrotStartTime"], serde_json::json!("2023-11-14T22:13:20.123456Z"));
    }

    #[test]
    fn node_heal_status_rejects_nested_unknown_trailing_truncated_and_oversized_data() {
        let mut fixture = serde_json::json!({
            "version": 1, "servicesEnabled": true, "initialized": true,
            "info": {"bitrotStartTime": null, "bitrotStartCycle": 0, "currentScanMode": 0, "unknown": true},
            "operations": HealOperationsSnapshot::default(), "progress": null
        });
        let encoded = rmp_serde::to_vec_named(&fixture).expect("fixture should encode");
        assert!(
            decode_node_heal_status(&encoded)
                .expect_err("nested unknown field must fail")
                .contains("unknown")
        );

        fixture["info"].as_object_mut().expect("info object").remove("unknown");
        let mut encoded = rmp_serde::to_vec_named(&fixture).expect("fixture should encode");
        encoded.push(0);
        assert!(
            decode_node_heal_status(&encoded)
                .expect_err("trailing data must fail")
                .contains("trailing")
        );
        encoded.truncate(encoded.len() / 2);
        assert!(decode_node_heal_status(&encoded).is_err());
        assert!(
            decode_node_heal_status(&vec![0; NODE_HEAL_STATUS_MAX_SIZE + 1])
                .expect_err("oversized status must fail")
                .contains("size limit")
        );
    }

    #[test]
    fn node_replacement_recovery_status_round_trip_is_versioned() {
        let snapshot = NodeReplacementRecoveryStatusSnapshot::new(ReplacementRecoverySnapshot {
            records: vec![ReplacementRecoveryRecord {
                task_id: "11111111-1111-4111-8111-111111111111".to_string(),
                state: ReplacementRecoveryState::Completed,
                generation: Some("11111111-1111-4111-8111-111111111111".to_string()),
                set_disk_id: Some("pool_0_set_0".to_string()),
                target_slots: vec!["http://node-a:9000/mnt/disk1".to_string()],
                reason: None,
                verified_at: Some(42),
            }],
            definitive: true,
            reason: None,
        });

        let encoded = encode_node_replacement_recovery_status(&snapshot).expect("snapshot should encode");
        let decoded = decode_node_replacement_recovery_status(&encoded).expect("snapshot should decode");

        assert_eq!(decoded, snapshot);
    }

    #[test]
    fn node_replacement_recovery_status_rejects_unknown_trailing_and_oversized_data() {
        let mut snapshot = NodeReplacementRecoveryStatusSnapshot::new(ReplacementRecoverySnapshot {
            records: Vec::new(),
            definitive: true,
            reason: None,
        });
        snapshot.version += 1;
        let encoded = encode_node_replacement_recovery_status(&snapshot).expect("snapshot should encode");
        assert!(
            decode_node_replacement_recovery_status(&encoded)
                .expect_err("unknown version should fail closed")
                .contains("unsupported replacement recovery status version")
        );

        snapshot.version = super::NODE_REPLACEMENT_RECOVERY_STATUS_VERSION;
        let mut encoded = encode_node_replacement_recovery_status(&snapshot).expect("snapshot should encode");
        encoded.push(0);
        assert!(
            decode_node_replacement_recovery_status(&encoded)
                .expect_err("trailing data must fail")
                .contains("trailing")
        );
        assert!(
            decode_node_replacement_recovery_status(&vec![0; super::NODE_REPLACEMENT_RECOVERY_STATUS_MAX_SIZE + 1])
                .expect_err("oversized status must fail")
                .contains("size limit")
        );

        let fixture = serde_json::json!({
            "version": super::NODE_REPLACEMENT_RECOVERY_STATUS_VERSION,
            "snapshot": {
                "records": [],
                "definitive": true,
                "reason": null,
                "futureField": true,
            }
        });
        let encoded = rmp_serde::to_vec_named(&fixture).expect("fixture should encode");
        assert!(
            decode_node_replacement_recovery_status(&encoded)
                .expect_err("nested unknown field must fail")
                .contains("futureField")
        );
    }
}
