bp_core/control/

server.rs

//! Unix socket control server — runs inside the daemon.
//! Each CLI invocation connects, sends one JSON request, reads one JSON response,
//! then closes the connection.

use crate::{
    coding::{params as coding_params, rlnc},
    control::protocol::{
        ControlRequest, ControlResponse, FileEntry, FlockData, GetFileData, HatchData, InviteData,
        ListFilesData, NetworkQosSummary, PouchStat, PutFileData, StatusData, StorageInfoData,
    },
    error::BpResult,
    identity::Identity,
    network::{
        fragment_gossip::{FragmentIndexAnnouncement, FragmentPointer, RemoteFragmentIndex},
        state::NodeInfo,
        FragmentResponse, NetworkCommand, OutgoingAssignments, QosRegistry, ReputationStore,
        StorageManagerMap,
    },
    service::{ServiceInfo, ServiceRegistry, ServiceStatus, ServiceType},
    storage::{ChunkCipher, FileRegistry, StorageManager},
};
use libp2p::{Multiaddr, PeerId};
use std::{
    collections::HashMap,
    path::Path,
    sync::{Arc, RwLock},
    time::{SystemTime, UNIX_EPOCH},
};
use tokio::{
    io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
    net::{UnixListener, UnixStream},
    sync::mpsc,
};

// ── CEK hint persistence ──────────────────────────────────────────────────────

/// Load persisted CEK hints from `cek_hints.json`.
///
/// Returns an empty map if the file does not exist or cannot be parsed.
/// Called once at daemon startup so files encrypted in previous sessions
/// remain decryptable.
pub fn load_cek_hints() -> HashMap<String, [u8; 32]> {
    let path = match crate::config::cek_hints_path() {
        Ok(p) => p,
        Err(e) => {
            tracing::warn!("cek_hints_path error: {e}");
            return HashMap::new();
        }
    };
    load_cek_hints_at(&path)
}

/// Core deserialization logic — reads from an explicit `path`.
fn load_cek_hints_at(path: &std::path::Path) -> HashMap<String, [u8; 32]> {
    if !path.exists() {
        return HashMap::new();
    }
    let data = match std::fs::read_to_string(path) {
        Ok(d) => d,
        Err(e) => {
            tracing::warn!("Failed to read cek_hints: {e}");
            return HashMap::new();
        }
    };
    let hex_map: HashMap<String, String> = match serde_json::from_str(&data) {
        Ok(m) => m,
        Err(e) => {
            tracing::warn!("Failed to parse cek_hints: {e}");
            return HashMap::new();
        }
    };
    hex_map
        .into_iter()
        .filter_map(|(k, v)| {
            let bytes = hex::decode(&v).ok()?;
            if bytes.len() != 32 {
                return None;
            }
            let mut arr = [0u8; 32];
            arr.copy_from_slice(&bytes);
            Some((k, arr))
        })
        .collect()
}

/// Persist CEK hints to `cek_hints.json`.
///
/// Failures are logged as warnings and silently ignored — a missing hint
/// means the file cannot be decrypted after restart, but it does not
/// corrupt any other state.
fn persist_cek_hints(hints: &HashMap<String, [u8; 32]>) {
    let path = match crate::config::cek_hints_path() {
        Ok(p) => p,
        Err(e) => {
            tracing::warn!("cek_hints_path error: {e}");
            return;
        }
    };
    persist_cek_hints_at(&path, hints);
}

/// Core serialization logic — writes to an explicit `path`.
fn persist_cek_hints_at(path: &std::path::Path, hints: &HashMap<String, [u8; 32]>) {
    // Ensure the parent directory exists before writing.
    if let Some(parent) = path.parent() {
        if let Err(e) = std::fs::create_dir_all(parent) {
            tracing::warn!("Failed to create cek_hints dir: {e}");
            return;
        }
    }
    let hex_map: HashMap<&str, String> = hints
        .iter()
        .map(|(k, v)| (k.as_str(), hex::encode(v)))
        .collect();
    match serde_json::to_string(&hex_map) {
        Ok(json) => {
            if let Err(e) = std::fs::write(path, json) {
                tracing::warn!("Failed to write cek_hints: {e}");
            }
        }
        Err(e) => tracing::warn!("Failed to serialize cek_hints: {e}"),
    }
}

/// Shared daemon state accessible from the control server.
pub struct DaemonState {
    pub identity: Identity,
    pub services: RwLock<ServiceRegistry>,
    pub network_state: Arc<RwLock<crate::network::state::NetworkState>>,
    pub networks: RwLock<Vec<String>>,
    pub net_tx: mpsc::Sender<NetworkCommand>,
    /// One `StorageManager` per active Pouch service, keyed by `service_id`.
    ///
    /// Shared with the network loop so incoming fragment-fetch requests can
    /// be served directly from the P2P event handler.
    pub storage_managers: StorageManagerMap,
    /// Per-peer QoS data updated by the network quality monitor.
    /// Used by `PutFile` to derive adaptive k/n from live stability scores.
    pub qos: Arc<RwLock<QosRegistry>>,
    /// Tracks which fragments were pushed to which remote Pouch peer.
    ///
    /// Updated by the network loop on every `PushFragment`; read by the
    /// quality monitor to select Proof-of-Storage challenge targets.
    pub outgoing_assignments: OutgoingAssignments,
    /// In-memory index of `chunk_id → [(peer_id, fragment_id)]` learned from
    /// gossipped [`FragmentIndexAnnouncement`]s.
    ///
    /// Populated when received from the network; used by `GetFile` to issue
    /// targeted `FetchChunkFragments` requests instead of broadcasting to all
    /// known Pouches.
    pub remote_fragment_index: Arc<RwLock<RemoteFragmentIndex>>,
    /// Maps `chunk_id → BLAKE3(plaintext_chunk)` so that `GetFile` can
    /// re-derive the per-user CEK for decryption after RLNC decode.
    ///
    /// Populated by `PutFile`; persisted only for the daemon’s lifetime
    /// (lost on restart — full persistence comes with the manifest system).
    pub chunk_cek_hints: RwLock<HashMap<String, [u8; 32]>>,
    /// Per-peer reputation state (tier R0–R4, score, uptime, PoS pass rate).
    ///
    /// Updated by the quality monitor and by eviction events.
    pub reputation: RwLock<ReputationStore>,
    /// Local catalogue of files uploaded by this node's Bill services.
    ///
    /// Populated on each successful `PutFile`.
    /// Persisted to `file_registry.json`.
    pub file_registry: RwLock<FileRegistry>,
}

/// Accept connections on the Unix socket and dispatch requests.
pub async fn run_control_server(
    socket_path: impl AsRef<Path>,
    state: Arc<DaemonState>,
) -> BpResult<()> {
    // Remove stale socket from a previous run.
    let path = socket_path.as_ref();
    if path.exists() {
        std::fs::remove_file(path).ok();
    }

    let listener =
        UnixListener::bind(path).map_err(|e| crate::error::BpError::Control(e.to_string()))?;

    tracing::info!("Control socket listening at {:?}", path);

    loop {
        match listener.accept().await {
            Ok((stream, _)) => {
                let state = Arc::clone(&state);
                tokio::spawn(async move {
                    if let Err(e) = handle_connection(stream, state).await {
                        tracing::warn!("Control connection error: {}", e);
                    }
                });
            }
            Err(e) => {
                tracing::error!("Accept error on control socket: {}", e);
            }
        }
    }
}

async fn handle_connection(stream: UnixStream, state: Arc<DaemonState>) -> anyhow::Result<()> {
    let (read_half, mut write_half) = stream.into_split();
    let mut reader = BufReader::new(read_half);
    let mut line = String::new();

    reader.read_line(&mut line).await?;
    let line = line.trim();
    if line.is_empty() {
        return Ok(());
    }

    let request: ControlRequest = match serde_json::from_str(line) {
        Ok(r) => r,
        Err(e) => {
            let resp = ControlResponse::err(format!("Invalid request JSON: {}", e));
            send_response(&mut write_half, &resp).await?;
            return Ok(());
        }
    };

    let response = dispatch(request, &state).await;
    send_response(&mut write_half, &response).await?;
    Ok(())
}

async fn send_response(
    writer: &mut tokio::net::unix::OwnedWriteHalf,
    resp: &ControlResponse,
) -> anyhow::Result<()> {
    let mut json = serde_json::to_string(resp)?;
    json.push('\n');
    writer.write_all(json.as_bytes()).await?;
    Ok(())
}

/// Dispatch a `ControlRequest` to the appropriate handler.
async fn dispatch(req: ControlRequest, state: &Arc<DaemonState>) -> ControlResponse {
    match req {
        // ── Ping ──────────────────────────────────────────────────────────
        ControlRequest::Ping => ControlResponse::ok("pong"),

        // ── Status ────────────────────────────────────────────────────────
        ControlRequest::Status => {
            let services = state.services.read().unwrap();
            let networks = state.networks.read().unwrap();
            let ns = state.network_state.read().unwrap();
            let own_peer = state.identity.peer_id.to_string();

            // Own reputation tier, score, and dynamic availability factor.
            let (rep_tier, rep_score, avail_factor) = {
                let (tier_str, score, tier_enum) = {
                    let rep = state.reputation.read().unwrap();
                    match rep.get(&own_peer) {
                        Some(r) => (r.tier.to_string(), r.reputation_score, r.tier),
                        None => (
                            crate::network::ReputationTier::R1.to_string(),
                            0i64,
                            crate::network::ReputationTier::R1,
                        ),
                    }
                };
                // k/N: fraction of raw bid usable as recoverable content.
                // N = remote Pouch peers + own node (score 0.4 if unobserved).
                let factor = if tier_enum == crate::network::ReputationTier::R0 {
                    0.0_f64
                } else {
                    let ph = tier_enum.qos_target_ph();
                    let qos = state.qos.read().unwrap();
                    let own_score = qos
                        .get(&own_peer)
                        .map(|q| q.stability_score())
                        .unwrap_or(0.4);
                    let mut stabilities = qos.all_stability_scores();
                    stabilities.push(own_score);
                    crate::coding::params::compute_network_storage_factor(&stabilities, ph)
                };
                (tier_str, score, factor)
            };

            // Per-Pouch storage stats.
            let pouch_stats: Vec<crate::control::protocol::PouchStat> = {
                let managers = state.storage_managers.read().unwrap();
                services
                    .all()
                    .iter()
                    .filter(|s| s.service_type == ServiceType::Pouch)
                    .map(|s| {
                        let tier_label = s.metadata.get("tier").and_then(|v| v.as_str()).map(|t| {
                            format!(
                                "{} — {}",
                                t,
                                match t {
                                    "T1" => "Pebble",
                                    "T2" => "Stone",
                                    "T3" => "Boulder",
                                    "T4" => "Rock",
                                    "T5" => "Monolith",
                                    _ => "?",
                                }
                            )
                        });
                        if let Some(sm_lock) = managers.get(&s.id) {
                            let sm = sm_lock.read().unwrap();
                            crate::control::protocol::PouchStat {
                                service_id: s.id.clone(),
                                network_id: s.network_id.clone(),
                                storage_tier: tier_label,
                                storage_bid_bytes: sm.meta.storage_bytes_bid,
                                storage_used_bytes: sm.meta.storage_bytes_used,
                                available_bytes: ((sm.meta.available_bytes() as f64) * avail_factor)
                                    .round()
                                    as u64,
                            }
                        } else {
                            crate::control::protocol::PouchStat {
                                service_id: s.id.clone(),
                                network_id: s.network_id.clone(),
                                storage_tier: tier_label,
                                storage_bid_bytes: 0,
                                storage_used_bytes: 0,
                                available_bytes: 0,
                            }
                        }
                    })
                    .collect()
            };

            // Network QoS summary.
            let qos_peer_count = state.qos.read().unwrap().peer_count();
            let network_qos = {
                let qos = state.qos.read().unwrap();
                let rep = state.reputation.read().unwrap();
                let peer_count = qos.peer_count();
                if peer_count == 0 {
                    None
                } else {
                    let scores = qos.all_stability_scores();
                    let avg_stability = if scores.is_empty() {
                        0.0
                    } else {
                        scores.iter().sum::<f64>() / scores.len() as f64
                    };
                    let mut tier_counts: std::collections::HashMap<String, usize> =
                        std::collections::HashMap::new();
                    for peer_id in qos.peer_ids() {
                        let tier = rep.tier(peer_id).to_string();
                        *tier_counts.entry(tier).or_insert(0) += 1;
                    }
                    Some(NetworkQosSummary {
                        observed_peers: peer_count,
                        avg_stability,
                        tier_counts,
                    })
                }
            };

            let data = StatusData {
                peer_id: own_peer,
                fingerprint: state.identity.fingerprint.clone(),
                alias: state.identity.profile.alias.clone(),
                local_services: services.all().into_iter().cloned().collect(),
                networks: networks.clone(),
                known_peers: ns.len().max(qos_peer_count),
                version: env!("CARGO_PKG_VERSION").to_string(),
                reputation_tier: rep_tier,
                reputation_score: rep_score,
                pouch_stats,
                network_qos,
            };
            ControlResponse::ok(data)
        }

        // ── AnnounceNow ───────────────────────────────────────────────────
        ControlRequest::AnnounceNow => {
            // Re-broadcast NodeInfo for every running service, then pause
            // 2 s to let gossipsub deliver the messages to peers.
            let svcs: Vec<(String, ServiceType, String)> = {
                state
                    .services
                    .read()
                    .unwrap()
                    .all()
                    .iter()
                    .filter(|s| s.status == ServiceStatus::Running)
                    .map(|s| (s.id.clone(), s.service_type, s.network_id.clone()))
                    .collect()
            };
            if svcs.is_empty() {
                tracing::debug!("AnnounceNow: no running services, skipping");
            } else {
                for (id, stype, net) in &svcs {
                    announce_self(state, id, *stype, net).await;
                }
                tracing::debug!(
                    "AnnounceNow: announced {} service(s), waiting 2s",
                    svcs.len()
                );
                tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
            }
            ControlResponse::ok("announced")
        }

        // ── Hatch ─────────────────────────────────────────────────────────
        ControlRequest::Hatch {
            service_type,
            network_id,
            metadata,
        } => {
            // Ensure we're subscribed to that network's gossip
            let already_joined = {
                let nets = state.networks.read().unwrap();
                nets.contains(&network_id)
            };
            if !already_joined {
                let _ = state
                    .net_tx
                    .send(NetworkCommand::JoinNetwork {
                        network_id: network_id.clone(),
                    })
                    .await;
                state.networks.write().unwrap().push(network_id.clone());
            }

            // Enforce one-Pouch-per-network: two correlated Pouches on the
            // same node defeat distributed redundancy.
            if service_type == ServiceType::Pouch {
                let already_exists =
                    state.services.read().unwrap().all().iter().any(|s| {
                        s.service_type == ServiceType::Pouch && s.network_id == network_id
                    });
                if already_exists {
                    return ControlResponse::err(format!(
                        "a Pouch for network '{}' already exists on this node; \
                         two correlated Pouches defeat distributed redundancy",
                        network_id
                    ));
                }
            }

            let info = ServiceInfo::new(service_type, network_id.clone(), metadata.clone());
            let service_id = info.id.clone();

            // Set to Running immediately — scope the lock guard so it is dropped
            // BEFORE the .await below (std::sync guards are !Send across await).
            {
                let mut reg = state.services.write().unwrap();
                let mut running_info = info.clone();
                running_info.status = ServiceStatus::Running;
                reg.register(running_info);
            } // ← lock released here

            // Initialise a StorageManager when hatching a Pouch service.
            if service_type == ServiceType::Pouch {
                let quota = metadata
                    .get("storage_bytes")
                    .or_else(|| metadata.get("storage_bytes_bid"))
                    .and_then(|v| v.as_u64())
                    .unwrap_or(0);

                match StorageManager::init(network_id.clone(), service_id.clone(), quota) {
                    Ok(sm) => {
                        state
                            .storage_managers
                            .write()
                            .unwrap()
                            .insert(service_id.clone(), Arc::new(RwLock::new(sm)));
                        tracing::info!(service_id=%service_id, quota_bytes=%quota, "StorageManager initialised");
                    }
                    Err(e) => {
                        tracing::warn!("Failed to init StorageManager for {}: {}", service_id, e);
                    }
                }
            }

            // Announce to the network (await-safe: no lock held)
            announce_self(state, &service_id, service_type, &network_id).await;

            ControlResponse::ok(HatchData {
                service_id: service_id.clone(),
                service_type,
                network_id,
                message: format!("🦤 {} service hatched — id: {}", service_type, service_id),
            })
        }

        // ── Farewell ──────────────────────────────────────────────────────
        ControlRequest::Farewell { service_id } => {
            let mut reg = state.services.write().unwrap();
            match reg.remove(&service_id) {
                Some(info) => ControlResponse::ok(serde_json::json!({
                    "service_id": info.id,
                    "service_type": info.service_type,
                    "message": format!("Service {} stopped", info.id),
                })),
                None => ControlResponse::err(format!("No service with id '{}'", service_id)),
            }
        }
        // ── FarewellEvict ─────────────────────────────────────────────────
        ControlRequest::FarewellEvict { service_id } => {
            let now_secs = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap_or_default()
                .as_secs();

            // Look up the service.
            let (network_id, service_type, peer_id_str) = {
                let reg = state.services.read().unwrap();
                match reg.get(&service_id) {
                    Some(info) => (
                        info.network_id.clone(),
                        info.service_type,
                        state.identity.peer_id.to_string(),
                    ),
                    None => {
                        return ControlResponse::err(format!(
                            "No service with id '{}'",
                            service_id
                        ));
                    }
                }
            };

            // Collect storage stats and purge disk (if Pouch).
            let (chunk_count, fragment_count, bytes_freed) = {
                let mut managers = state.storage_managers.write().unwrap();
                if let Some(sm_lock) = managers.remove(&service_id) {
                    let sm = sm_lock.read().unwrap();
                    let summary = sm.storage_summary();
                    if let Err(e) = sm.purge() {
                        tracing::warn!("Failed to purge storage for {}: {}", service_id, e);
                    }
                    summary
                } else {
                    (0, 0, 0)
                }
            };

            // Announce eviction via gossip before removing from registry.
            {
                let mut meta = HashMap::new();
                meta.insert("evicting".to_string(), serde_json::Value::Bool(true));
                let info = NodeInfo {
                    peer_id: peer_id_str,
                    user_fingerprint: state.identity.fingerprint.clone(),
                    user_alias: state.identity.profile.alias.clone(),
                    service_type,
                    service_id: service_id.clone(),
                    network_id: network_id.clone(),
                    listen_addrs: vec![],
                    announced_at: now_secs,
                    metadata: meta,
                };
                if let Ok(payload) = serde_json::to_vec(&info) {
                    let _ = state
                        .net_tx
                        .send(NetworkCommand::Announce {
                            network_id: network_id.clone(),
                            payload,
                        })
                        .await;
                }
            }

            // Record reputation eviction.
            {
                let local_peer = state.identity.peer_id.to_string();
                let mut rep = state.reputation.write().unwrap();
                rep.get_or_create(&local_peer, now_secs)
                    .evict_without_notice(now_secs);
            }

            // Remove from registry.
            state.services.write().unwrap().remove(&service_id);

            tracing::info!(
                service_id = %service_id,
                chunks = chunk_count,
                fragments = fragment_count,
                bytes = bytes_freed,
                "Pouch evicted"
            );

            ControlResponse::ok(serde_json::json!({
                "service_id": service_id,
                "network_id": network_id,
                "chunks_removed": chunk_count,
                "fragments_removed": fragment_count,
                "bytes_freed": bytes_freed,
                "message": format!(
                    "Pouch {} evicted — {} chunk(s), {} fragment(s), {} bytes freed. \
                     Network peers will rebalance via Proof-of-Storage.",
                    service_id, chunk_count, fragment_count, bytes_freed
                ),
            }))
        }
        // ── Pause ─────────────────────────────────────────────────────────
        ControlRequest::Pause {
            service_id,
            eta_minutes,
        } => {
            let now_secs = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap_or_default()
                .as_secs();

            let (network_id, service_type) = {
                let mut reg = state.services.write().unwrap();
                match reg.get_mut(&service_id) {
                    Some(info) => {
                        if info.status == crate::service::ServiceStatus::Stopped
                            || matches!(info.status, crate::service::ServiceStatus::Stopping)
                        {
                            return ControlResponse::err(format!(
                                "Service '{}' is not running",
                                service_id
                            ));
                        }
                        info.status = crate::service::ServiceStatus::Paused {
                            eta_minutes,
                            paused_at: now_secs,
                        };
                        info.metadata
                            .insert("maintenance".into(), serde_json::Value::Bool(true));
                        info.metadata
                            .insert("eta_minutes".into(), serde_json::Value::from(eta_minutes));
                        (info.network_id.clone(), info.service_type)
                    }
                    None => {
                        return ControlResponse::err(format!(
                            "No service with id '{}'",
                            service_id
                        ));
                    }
                }
            };

            announce_self(state, &service_id, service_type, &network_id).await;

            ControlResponse::ok(serde_json::json!({
                "service_id": service_id,
                "status": "paused",
                "eta_minutes": eta_minutes,
                "message": format!(
                    "Service {} paused — announcing maintenance (ETA {} min)",
                    service_id, eta_minutes
                ),
            }))
        }

        // ── Resume ────────────────────────────────────────────────────────
        ControlRequest::Resume { service_id } => {
            let (network_id, service_type) = {
                let mut reg = state.services.write().unwrap();
                match reg.get_mut(&service_id) {
                    Some(info) => {
                        if !matches!(info.status, crate::service::ServiceStatus::Paused { .. }) {
                            return ControlResponse::err(format!(
                                "Service '{}' is not paused",
                                service_id
                            ));
                        }
                        info.status = crate::service::ServiceStatus::Running;
                        info.metadata.remove("maintenance");
                        info.metadata.remove("eta_minutes");
                        (info.network_id.clone(), info.service_type)
                    }
                    None => {
                        return ControlResponse::err(format!(
                            "No service with id '{}'",
                            service_id
                        ));
                    }
                }
            };

            announce_self(state, &service_id, service_type, &network_id).await;

            ControlResponse::ok(serde_json::json!({
                "service_id": service_id,
                "status": "running",
                "message": format!("Service {} resumed", service_id),
            }))
        }

        // ── Flock ─────────────────────────────────────────────────────────
        ControlRequest::Flock => {
            let services = state.services.read().unwrap();
            let networks = state.networks.read().unwrap();
            let ns = state.network_state.read().unwrap();
            let data = FlockData {
                local_services: services.all().into_iter().cloned().collect(),
                known_peers: ns.all().into_iter().cloned().collect(),
                networks: networks.clone(),
                peer_count: ns.len(),
            };
            ControlResponse::ok(data)
        }

        // ── Join ──────────────────────────────────────────────────────────
        ControlRequest::Join { network_id } => {
            {
                let nets = state.networks.read().unwrap();
                if nets.contains(&network_id) {
                    return ControlResponse::err(format!(
                        "Already a member of network '{}'",
                        network_id
                    ));
                }
            }
            // Ensure a NetworkMetaKey exists for this network.
            // If none was installed via an invite token, we generate a fresh
            // random one (this node is creating the network).
            if let Err(e) = crate::storage::manifest::NetworkMetaKey::load_or_create(&network_id) {
                tracing::warn!(network = %network_id, "Failed to ensure NetworkMetaKey: {e}");
            }
            let _ = state
                .net_tx
                .send(NetworkCommand::JoinNetwork {
                    network_id: network_id.clone(),
                })
                .await;
            state.networks.write().unwrap().push(network_id.clone());
            ControlResponse::ok(serde_json::json!({
                "network_id": network_id,
                "message": format!("Joined network '{}'", network_id),
            }))
        }

        // ── Leave ─────────────────────────────────────────────────────────
        ControlRequest::Leave { network_id, force } => {
            // Collect blocking services on this network.
            let blocking: Vec<ServiceInfo> = {
                let reg = state.services.read().unwrap();
                reg.all()
                    .iter()
                    .filter(|s| s.network_id == network_id)
                    .map(|s| (*s).clone())
                    .collect()
            };

            if !blocking.is_empty() && !force {
                // v1: return helpful error with per-service hints.
                let blocking_json: Vec<serde_json::Value> = blocking
                    .iter()
                    .map(|s| {
                        let hint = if s.service_type == ServiceType::Pouch {
                            format!("bp farewell {} --evict", s.id)
                        } else {
                            format!("bp farewell {}", s.id)
                        };
                        serde_json::json!({
                            "id":   s.id,
                            "type": s.service_type.to_string(),
                            "hint": hint,
                        })
                    })
                    .collect();
                return ControlResponse::ok(serde_json::json!({
                    "network_id": network_id,
                    "blocked": true,
                    "blocking_services": blocking_json,
                    "message": format!(
                        "Cannot leave '{}': {} active service(s) must be stopped first \
                         (see 'blocking_services' for commands, or use --force to auto-evict)",
                        network_id,
                        blocking.len()
                    ),
                }));
            }

            // force=true (or no blocking services): auto-evict then leave.
            let now_secs = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap_or_default()
                .as_secs();
            let mut evicted: Vec<serde_json::Value> = Vec::new();

            for svc in &blocking {
                let service_id = svc.id.clone();
                let service_type = svc.service_type;
                let peer_id_str = state.identity.peer_id.to_string();

                if service_type == ServiceType::Pouch {
                    // Purge on-disk storage.
                    {
                        let mut managers = state.storage_managers.write().unwrap();
                        if let Some(sm_lock) = managers.remove(&service_id) {
                            let sm = sm_lock.read().unwrap();
                            if let Err(e) = sm.purge() {
                                tracing::warn!(
                                    "leave --force: failed to purge storage for {}: {}",
                                    service_id,
                                    e
                                );
                            }
                        }
                    }
                    // Gossip eviction notice before removing from registry.
                    {
                        let mut meta = HashMap::new();
                        meta.insert("evicting".to_string(), serde_json::Value::Bool(true));
                        let info = NodeInfo {
                            peer_id: peer_id_str.clone(),
                            user_fingerprint: state.identity.fingerprint.clone(),
                            user_alias: state.identity.profile.alias.clone(),
                            service_type,
                            service_id: service_id.clone(),
                            network_id: network_id.clone(),
                            listen_addrs: vec![],
                            announced_at: now_secs,
                            metadata: meta,
                        };
                        if let Ok(payload) = serde_json::to_vec(&info) {
                            let _ = state
                                .net_tx
                                .send(NetworkCommand::Announce {
                                    network_id: network_id.clone(),
                                    payload,
                                })
                                .await;
                        }
                    }
                    // Record reputation eviction.
                    {
                        let mut rep = state.reputation.write().unwrap();
                        rep.get_or_create(&peer_id_str, now_secs)
                            .evict_without_notice(now_secs);
                    }
                    tracing::info!(
                        service_id = %service_id,
                        network_id = %network_id,
                        "Pouch auto-evicted by leave --force"
                    );
                } else {
                    tracing::info!(
                        service_id = %service_id,
                        service_type = %service_type,
                        network_id = %network_id,
                        "service stopped by leave --force"
                    );
                }

                // Remove from registry regardless of type.
                state.services.write().unwrap().remove(&service_id);
                evicted.push(serde_json::json!({
                    "service_id":   service_id,
                    "service_type": service_type.to_string(),
                    "evicted":       service_type == ServiceType::Pouch,
                }));
            }

            // Unsubscribe gossip and remove from active_networks.
            let _ = state
                .net_tx
                .send(NetworkCommand::LeaveNetwork {
                    network_id: network_id.clone(),
                })
                .await;
            state.networks.write().unwrap().retain(|n| n != &network_id);

            let msg = if evicted.is_empty() {
                format!("Left network '{}'", network_id)
            } else {
                format!(
                    "Left network '{}' — {} service(s) auto-evicted",
                    network_id,
                    evicted.len()
                )
            };
            ControlResponse::ok(serde_json::json!({
                "network_id": network_id,
                "services_auto_evicted": evicted,
                "message": msg,
            }))
        }

        // ── ConnectRelay ──────────────────────────────────────────────────
        ControlRequest::ConnectRelay { relay_addr } => {
            let addr: Multiaddr = match relay_addr.parse() {
                Ok(a) => a,
                Err(e) => {
                    return ControlResponse::err(format!("Invalid relay multiaddr: {}", e));
                }
            };
            let _ = state
                .net_tx
                .send(NetworkCommand::DialRelay { relay_addr: addr })
                .await;
            ControlResponse::ok(serde_json::json!({
                "relay_addr": relay_addr,
                "message": format!("Dialing relay '{}'", relay_addr),
            }))
        }
        // ── CreateInvite ──────────────────────────────────────────────────
        ControlRequest::CreateInvite {
            network_id,
            invitee_fingerprint,
            invite_password,
            ttl_hours,
        } => {
            let ttl = ttl_hours.unwrap_or(24);
            let expires_at =
                (chrono::Utc::now() + chrono::Duration::hours(ttl as i64)).timestamp() as u64;
            match crate::invite::create_invite(
                &state.identity,
                &network_id,
                invitee_fingerprint,
                ttl,
                &invite_password,
            ) {
                Ok(blob) => ControlResponse::ok(InviteData {
                    blob,
                    network_id,
                    expires_at,
                    inviter_fingerprint: state.identity.fingerprint.clone(),
                }),
                Err(e) => ControlResponse::err(format!("Failed to create invite: {e}")),
            }
        }
        // ── PutFile ───────────────────────────────────────────────────────
        ControlRequest::PutFile {
            chunk_data,
            ph,
            q_target,
            network_id,
            file_name,
        } => {
            let ph = ph.unwrap_or(0.999);
            let q_target = q_target.unwrap_or(1.0);

            // Find a Pouch StorageManager for this network.
            let managers_snap: Vec<(String, Arc<RwLock<StorageManager>>)> = {
                let map = state.storage_managers.read().unwrap();
                map.iter()
                    .map(|(id, sm)| (id.clone(), Arc::clone(sm)))
                    .collect()
            };

            // Filter to managers whose network matches (or any if network_id is empty).
            let candidates: Vec<_> = managers_snap
                .iter()
                .filter(|(_, sm)| {
                    let meta = &sm.read().unwrap().meta;
                    network_id.is_empty() || meta.network_id == network_id
                })
                .collect();

            // bill/post nodes have no local Pouch — they act as pure coordinators
            // that encode the chunk and distribute all fragments to remote Pouches.
            // We only fail if there are neither local nor remote Pouches.
            let has_local_pouch = !candidates.is_empty();

            // ── Compute adaptive k/n from live QoS data ──────────────────────
            let pouch_peer_ids: Vec<String> = {
                let ns = state.network_state.read().unwrap();
                ns.in_network(&network_id)
                    .into_iter()
                    .filter(|n| n.service_type == ServiceType::Pouch)
                    .map(|n| n.peer_id.clone())
                    .collect()
            };

            // For peers not yet observed via Ping, assume 0.8 stability
            // (benefit of the doubt for new entries).
            let stabilities: Vec<f64> = {
                let qos = state.qos.read().unwrap();
                if pouch_peer_ids.is_empty() {
                    vec![0.8; candidates.len().max(1)]
                } else {
                    pouch_peer_ids
                        .iter()
                        .map(|pid| match qos.get(pid) {
                            Some(p) if p.is_observed() => p.stability_score(),
                            _ => 0.8,
                        })
                        .collect()
                }
            };

            let (k, n, q, pe) =
                match coding_params::compute_coding_params(&stabilities, ph, q_target) {
                    Ok(p) => {
                        let pe = coding_params::effective_recovery_probability(&stabilities, p.k);
                        (p.k, p.n, p.q, pe)
                    }
                    Err(e) => {
                        tracing::warn!("compute_coding_params failed ({e}), using fallback");
                        let peer_n = stabilities.len().max(2);
                        let fallback_k = (peer_n / 2).max(1);
                        let fallback_q = (peer_n - fallback_k) as f64 / fallback_k as f64;
                        let pe =
                            coding_params::effective_recovery_probability(&stabilities, fallback_k);
                        (fallback_k, peer_n, fallback_q, pe)
                    }
                };

            tracing::info!(
                network = %network_id, k = %k, n = %n, q = %q,
                ph = %ph, pe = %pe, peers = %stabilities.len(),
                "PutFile: computed coding params"
            );

            // Encrypt the chunk with a per-user CEK before RLNC encoding.
            // The CEK is derived from the owner’s secret material and the
            // BLAKE3 hash of the plaintext — Pouch nodes never see plaintext.
            let plaintext_hash: [u8; 32] = *blake3::hash(&chunk_data).as_bytes();
            let secret_mat = state.identity.secret_material();
            let cipher = ChunkCipher::for_user(&secret_mat, &plaintext_hash);
            let encrypted_chunk = match cipher.encrypt(&chunk_data) {
                Ok(b) => b,
                Err(e) => return ControlResponse::err(format!("Chunk encryption error: {e}")),
            };

            let fragments = match rlnc::encode(&encrypted_chunk, k, n) {
                Ok(f) => f,
                Err(e) => return ControlResponse::err(format!("Encode error: {e}")),
            };

            let chunk_id = match fragments.first() {
                Some(f) => f.chunk_id.clone(),
                None => return ControlResponse::err("Encoding produced no fragments"),
            };

            // Store the plaintext hash so GetFile can re-derive the CEK.
            state
                .chunk_cek_hints
                .write()
                .unwrap()
                .insert(chunk_id.clone(), plaintext_hash);
            // Persist hints so the CEK survives a daemon restart.
            {
                let hints = state.chunk_cek_hints.read().unwrap();
                persist_cek_hints(&hints);
            }

            // Store all fragments locally if a local Pouch is available.
            let mut stored = 0usize;
            if has_local_pouch {
                let (_, sm_arc) = candidates[0];
                let mut sm = sm_arc.write().unwrap();
                for fragment in &fragments {
                    match sm.store_fragment(fragment) {
                        Ok(()) => stored += 1,
                        Err(e) => {
                            tracing::warn!(chunk_id=%chunk_id, "store_fragment failed: {e}");
                            break;
                        }
                    }
                }
            } // write lock released before async work

            // ── Distribute fragments to remote Pouch peers (round-robin) ────
            let remote_pouches: Vec<PeerId> = {
                let ns = state.network_state.read().unwrap();
                let local_peer = state.identity.peer_id.to_string();
                ns.in_network(&network_id)
                    .into_iter()
                    .filter(|n| n.service_type == ServiceType::Pouch && n.peer_id != local_peer)
                    .filter_map(|n| n.peer_id.parse::<PeerId>().ok())
                    .collect()
            };

            // Fail only if there is nowhere to store fragments at all.
            if stored == 0 && remote_pouches.is_empty() {
                return ControlResponse::err(format!(
                    "No Pouches available on network '{}' — hatch a Pouch or wait for Pouch peers to join",
                    network_id
                ));
            }

            let mut distributed = 0usize;
            let mut index_pointers: Vec<FragmentPointer> = Vec::new();
            if !remote_pouches.is_empty() {
                for (i, fragment) in fragments.iter().enumerate() {
                    let peer = remote_pouches[i % remote_pouches.len()];
                    if state
                        .net_tx
                        .send(NetworkCommand::PushFragment {
                            peer_id: peer,
                            chunk_id: chunk_id.clone(),
                            fragment_id: fragment.id.clone(),
                            data: fragment.to_bytes(),
                        })
                        .await
                        .is_ok()
                    {
                        distributed += 1;
                        index_pointers.push(FragmentPointer {
                            peer_id: peer.to_string(),
                            fragment_id: fragment.id.clone(),
                        });
                    }
                }
            }

            // Publish the fragment-index announcement so peers learn which
            // Pouch holds each fragment.
            if !index_pointers.is_empty() {
                let ann = FragmentIndexAnnouncement {
                    network_id: network_id.clone(),
                    chunk_id: chunk_id.clone(),
                    announced_at: chrono::Utc::now().timestamp() as u64,
                    pointers: index_pointers,
                };
                if let Ok(payload) = serde_json::to_vec(&ann) {
                    let _ = state
                        .net_tx
                        .send(NetworkCommand::AnnounceIndex {
                            network_id: network_id.clone(),
                            payload,
                        })
                        .await;
                    // Also update our own local index eagerly.
                    if let Ok(mut idx) = state.remote_fragment_index.write() {
                        idx.upsert(ann);
                    }
                }
            }

            tracing::info!(
                chunk_id=%chunk_id, stored=%stored, distributed=%distributed,
                total=%fragments.len(), "PutFile stored + distributed"
            );

            // Record the upload in the local file registry so `bp ls` can list it.
            if let Some(name) = file_name {
                let entry = crate::storage::StoredFileEntry {
                    file_name: name,
                    size_bytes: chunk_data.len() as u64,
                    chunk_id: chunk_id.clone(),
                    network_id: network_id.clone(),
                    uploaded_at: chrono::Utc::now().timestamp() as u64,
                };
                if let Ok(path) = crate::config::file_registry_path() {
                    state
                        .file_registry
                        .write()
                        .unwrap()
                        .insert_and_save(entry, &path);
                }
            }

            ControlResponse::ok(PutFileData {
                chunk_id: chunk_id.clone(),
                k,
                n,
                q,
                ph,
                pe,
                fragments_stored: stored,
                fragments_distributed: distributed,
                message: format!(
                    "Stored {stored}/{total} locally, pushed {distributed} to {peers} remote pouch(es) — chunk_id: {chunk_id}",
                    total = fragments.len(),
                    peers = remote_pouches.len(),
                ),
            })
        }

        // ── GetFile ───────────────────────────────────────────────────────
        ControlRequest::GetFile {
            chunk_id,
            network_id,
        } => {
            let managers_snap: Vec<Arc<RwLock<StorageManager>>> = {
                let map = state.storage_managers.read().unwrap();
                map.values()
                    .filter(|sm| {
                        let meta = &sm.read().unwrap().meta;
                        network_id.is_empty() || meta.network_id == network_id
                    })
                    .map(Arc::clone)
                    .collect()
            };

            // bill/post nodes have no local Pouch: skip the local load step and
            // go straight to remote fragment fetching.
            let remote_only = managers_snap.is_empty();

            // Collect fragments from all matching local managers.
            let mut all_fragments = Vec::new();
            for sm_arc in &managers_snap {
                let sm = sm_arc.read().unwrap();
                let metas = sm.index.fragments_for_chunk(&chunk_id).to_vec();
                for meta in metas {
                    match sm.load_fragment(&chunk_id, &meta.fragment_id) {
                        Ok(f) => all_fragments.push(f),
                        Err(e) => tracing::warn!("load_fragment failed: {e}"),
                    }
                }
            }

            let local_count = all_fragments.len();

            // Determine k from existing fragments (if any) to know if we need more.
            let k_needed = all_fragments.first().map(|f| f.k).unwrap_or(0);

            // ── Fetch from remote Pouches if local fragments < k ────────────
            // When there is no local Pouch (remote_only), always attempt remote fetch
            // since k_needed == 0 would otherwise skip this block entirely.
            let mut fragments_remote = 0usize;
            if remote_only || (k_needed > 0 && all_fragments.len() < k_needed) {
                let local_peer = state.identity.peer_id.to_string();

                // Build list of remote peers to query: prefer the fragment
                // index (targeted fetch) and fall back to broadcasting to all
                // known Pouches if the index has no entries for this chunk.
                let target_peers: Vec<PeerId> = {
                    let idx = state.remote_fragment_index.read().unwrap();
                    let ptrs = idx.pointers_for(&chunk_id);
                    if !ptrs.is_empty() {
                        // Deduplicate peer IDs from the index.
                        let mut seen = std::collections::HashSet::new();
                        ptrs.iter()
                            .filter(|p| p.peer_id != local_peer)
                            .filter_map(|p| {
                                if seen.insert(p.peer_id.clone()) {
                                    p.peer_id.parse::<PeerId>().ok()
                                } else {
                                    None
                                }
                            })
                            .collect()
                    } else {
                        // Fallback: broadcast to all known Pouches in the network.
                        let ns = state.network_state.read().unwrap();
                        ns.in_network(&network_id)
                            .into_iter()
                            .filter(|n| {
                                n.service_type == ServiceType::Pouch && n.peer_id != local_peer
                            })
                            .filter_map(|n| n.peer_id.parse::<PeerId>().ok())
                            .collect()
                    }
                };

                // Send FetchChunkFragments to each target peer.
                let mut receivers = Vec::new();
                for peer in &target_peers {
                    let (tx, rx) = tokio::sync::oneshot::channel();
                    if state
                        .net_tx
                        .send(NetworkCommand::FetchChunkFragments {
                            peer_id: *peer,
                            chunk_id: chunk_id.clone(),
                            resp_tx: tx,
                        })
                        .await
                        .is_ok()
                    {
                        receivers.push(rx);
                    }
                }

                // Await responses with a timeout.
                use crate::coding::rlnc::EncodedFragment;
                for rx in receivers {
                    match tokio::time::timeout(std::time::Duration::from_secs(10), rx).await {
                        Ok(Ok(FragmentResponse::FoundMany { fragments })) => {
                            for (frag_id, bytes) in fragments {
                                if all_fragments.len() >= k_needed {
                                    break;
                                }
                                match EncodedFragment::from_bytes(frag_id, chunk_id.clone(), &bytes)
                                {
                                    Ok(f) => {
                                        all_fragments.push(f);
                                        fragments_remote += 1;
                                    }
                                    Err(e) => tracing::warn!("Remote fragment parse error: {e}"),
                                }
                            }
                        }
                        Ok(Ok(FragmentResponse::Found { data })) => {
                            if all_fragments.len() < k_needed {
                                let frag_id = uuid::Uuid::new_v4().to_string();
                                match EncodedFragment::from_bytes(frag_id, chunk_id.clone(), &data)
                                {
                                    Ok(f) => {
                                        all_fragments.push(f);
                                        fragments_remote += 1;
                                    }
                                    Err(e) => tracing::warn!("Remote fragment parse error: {e}"),
                                }
                            }
                        }
                        Ok(Ok(_)) => {} // NotFound, Stored, StoreFailed — skip
                        Ok(Err(_)) => tracing::debug!("Remote fetch: oneshot dropped"),
                        Err(_) => tracing::debug!("Remote fetch: timeout"),
                    }
                    if all_fragments.len() >= k_needed {
                        break;
                    }
                }
            }

            if all_fragments.is_empty() {
                return ControlResponse::err(format!(
                    "Chunk '{chunk_id}' not found locally or on remote peers"
                ));
            }

            let fragments_used = all_fragments.len();
            match rlnc::decode(&all_fragments) {
                Ok(encrypted_data) => {
                    // Decrypt the recovered chunk using the per-user CEK.
                    // Re-derive the CEK from the stored plaintext hash hint.
                    let plaintext_hash = {
                        state
                            .chunk_cek_hints
                            .read()
                            .unwrap()
                            .get(&chunk_id)
                            .copied()
                    };
                    let data = match plaintext_hash {
                        Some(ph) => {
                            let secret_mat = state.identity.secret_material();
                            let cipher = ChunkCipher::for_user(&secret_mat, &ph);
                            match cipher.decrypt(&encrypted_data) {
                                Ok(d) => d,
                                Err(e) => {
                                    return ControlResponse::err(format!(
                                        "Chunk decryption failed: {e}"
                                    ))
                                }
                            }
                        }
                        None => {
                            return ControlResponse::err(format!(
                                "CEK hint not found for chunk '{}' — daemon may have restarted",
                                chunk_id
                            ))
                        }
                    };
                    tracing::info!(
                        chunk_id=%chunk_id, local=%local_count,
                        remote=%fragments_remote, total=%fragments_used,
                        "GetFile decoded + decrypted"
                    );
                    ControlResponse::ok(GetFileData {
                        chunk_id,
                        data,
                        fragments_used,
                        fragments_remote,
                    })
                }
                Err(e) => ControlResponse::err(format!("Decode failed: {e}")),
            }
        }

        // ── StorageInfo ───────────────────────────────────────────────────
        ControlRequest::StorageInfo { network_id } => {
            let managers = state.storage_managers.read().unwrap();
            let services = state.services.read().unwrap();
            let registry = state.file_registry.read().unwrap();

            // k/N availability factor: dynamic, from current network QoS.
            let avail_factor = {
                let own_peer = state.identity.peer_id.to_string();
                let tier_enum = {
                    let rep = state.reputation.read().unwrap();
                    match rep.get(&own_peer) {
                        Some(r) => r.tier,
                        None => crate::network::ReputationTier::R1,
                    }
                };
                if tier_enum == crate::network::ReputationTier::R0 {
                    0.0_f64
                } else {
                    let ph = tier_enum.qos_target_ph();
                    let qos = state.qos.read().unwrap();
                    let own_score = qos
                        .get(&own_peer)
                        .map(|q| q.stability_score())
                        .unwrap_or(0.4);
                    let mut stabilities = qos.all_stability_scores();
                    stabilities.push(own_score);
                    crate::coding::params::compute_network_storage_factor(&stabilities, ph)
                }
            };

            let pouches: Vec<PouchStat> = services
                .all()
                .iter()
                .filter(|s| {
                    s.service_type == ServiceType::Pouch
                        && (network_id.is_empty() || s.network_id == network_id)
                })
                .map(|s| {
                    let tier_label = s.metadata.get("tier").and_then(|v| v.as_str()).map(|t| {
                        format!(
                            "{} — {}",
                            t,
                            match t {
                                "T1" => "Pebble",
                                "T2" => "Stone",
                                "T3" => "Boulder",
                                "T4" => "Rock",
                                "T5" => "Monolith",
                                _ => "?",
                            }
                        )
                    });
                    if let Some(sm_lock) = managers.get(&s.id) {
                        let sm = sm_lock.read().unwrap();
                        PouchStat {
                            service_id: s.id.clone(),
                            network_id: s.network_id.clone(),
                            storage_tier: tier_label,
                            storage_bid_bytes: sm.meta.storage_bytes_bid,
                            storage_used_bytes: sm.meta.storage_bytes_used,
                            available_bytes: ((sm.meta.available_bytes() as f64) * avail_factor)
                                .round() as u64,
                        }
                    } else {
                        PouchStat {
                            service_id: s.id.clone(),
                            network_id: s.network_id.clone(),
                            storage_tier: tier_label,
                            storage_bid_bytes: 0,
                            storage_used_bytes: 0,
                            available_bytes: 0,
                        }
                    }
                })
                .collect();

            let total_bid_bytes: u64 = pouches.iter().map(|p| p.storage_bid_bytes).sum();
            let total_used_bytes: u64 = pouches.iter().map(|p| p.storage_used_bytes).sum();
            let total_available_bytes: u64 = pouches.iter().map(|p| p.available_bytes).sum();

            let net_filter = if network_id.is_empty() {
                ""
            } else {
                &network_id
            };
            let file_entries = registry.list(net_filter);
            let total_files_uploaded = file_entries.len();
            let total_uploaded_bytes: u64 = file_entries.iter().map(|e| e.size_bytes).sum();

            ControlResponse::ok(StorageInfoData {
                pouches,
                total_bid_bytes,
                total_used_bytes,
                total_available_bytes,
                total_files_uploaded,
                total_uploaded_bytes,
            })
        }

        // ── ListFiles ─────────────────────────────────────────────────────
        ControlRequest::ListFiles { network_id } => {
            let registry = state.file_registry.read().unwrap();
            let net_filter = if network_id.is_empty() {
                ""
            } else {
                &network_id
            };
            let entries: Vec<FileEntry> = registry
                .list(net_filter)
                .into_iter()
                .map(|e| FileEntry {
                    file_name: e.file_name.clone(),
                    size_bytes: e.size_bytes,
                    chunk_id: e.chunk_id.clone(),
                    network_id: e.network_id.clone(),
                    uploaded_at: e.uploaded_at,
                })
                .collect();
            let total_bytes = entries.iter().map(|e| e.size_bytes).sum();
            let total_files = entries.len();
            ControlResponse::ok(ListFilesData {
                files: entries,
                network_id: network_id.clone(),
                total_files,
                total_bytes,
            })
        }
    }
}

/// Build a `NodeInfo` from the current daemon state for a given service and
/// broadcast it via gossipsub.
async fn announce_self(
    state: &Arc<DaemonState>,
    service_id: &str,
    service_type: ServiceType,
    network_id: &str,
) {
    let listen_addrs: Vec<String> = vec![]; // Populated once we know our Multiaddrs

    let info = NodeInfo {
        peer_id: state.identity.peer_id.to_string(),
        user_fingerprint: state.identity.fingerprint.clone(),
        user_alias: state.identity.profile.alias.clone(),
        service_type,
        service_id: service_id.to_string(),
        network_id: network_id.to_string(),
        listen_addrs,
        announced_at: chrono::Utc::now().timestamp() as u64,
        metadata: HashMap::new(),
    };

    match serde_json::to_vec(&info) {
        Ok(payload) => {
            let _ = state
                .net_tx
                .send(NetworkCommand::Announce {
                    network_id: network_id.to_string(),
                    payload,
                })
                .await;
        }
        Err(e) => tracing::warn!("Failed to serialize NodeInfo: {}", e),
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// Unit tests
// ─────────────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::HashMap;

    /// `load_cek_hints_at` on a non-existent path returns an empty map.
    #[test]
    fn load_cek_hints_missing_file_returns_empty() {
        let dir = tempdir();
        let path = dir.join("cek_hints.json");
        let hints = load_cek_hints_at(&path);
        assert!(hints.is_empty(), "expected empty map for missing file");
    }

    /// `persist_cek_hints_at` + `load_cek_hints_at` round-trip.
    #[test]
    fn cek_hints_persist_load_roundtrip() {
        let dir = tempdir();
        let path = dir.join("cek_hints.json");

        let mut hints: HashMap<String, [u8; 32]> = HashMap::new();
        let key_a = *b"abcdefghijklmnopqrstuvwxyz012345";
        let key_b = *b"ABCDEFGHIJKLMNOPQRSTUVWXYZ678901";
        hints.insert("chunk-aaa".to_string(), key_a);
        hints.insert("chunk-bbb".to_string(), key_b);

        persist_cek_hints_at(&path, &hints);
        let loaded = load_cek_hints_at(&path);

        assert_eq!(loaded.len(), 2, "wrong number of hints loaded");
        assert_eq!(loaded["chunk-aaa"], key_a);
        assert_eq!(loaded["chunk-bbb"], key_b);
    }

    /// `load_cek_hints_at` with a corrupt JSON file returns an empty map without panicking.
    #[test]
    fn load_cek_hints_corrupt_file_returns_empty() {
        let dir = tempdir();
        let path = dir.join("cek_hints.json");
        std::fs::write(&path, b"NOT_VALID_JSON").unwrap();

        let hints = load_cek_hints_at(&path);
        assert!(hints.is_empty(), "corrupt file must yield empty map");
    }

    // ── helpers ───────────────────────────────────────────────────────────

    fn tempdir() -> std::path::PathBuf {
        let mut path = std::env::temp_dir();
        path.push(format!("bp_test_{}", uuid::Uuid::new_v4()));
        std::fs::create_dir_all(&path).unwrap();
        path
    }
}