node.rs

//! Sync node implementing structural sync layers 3-5 for blob availability.
//!
//! Layer 1 (validity): Hemera hash verification of every chunk from peers.
//! Layer 3 (completeness): Merkle root over registry; verified on sync.
//! Layer 4 (availability): DAS commitment in FileEntry; sampling on sync.
//! Layer 5 (merge): LWW-Element-Set CRDT with deterministic conflict resolution.
//!
//! Signal ordering (layer 2: VDF, hash chain, equivocation detection) is owned
//! by cybergraph, not this crate. See cybergraph/src/vdf.rs and chain.rs.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use anyhow::{Context, Result};
use iroh::address_lookup::MdnsAddressLookup;
use iroh::endpoint::Connection;
use iroh::protocol::{AcceptError, ProtocolHandler, Router};
use iroh::{Endpoint, EndpointId, RelayMode, SecretKey};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::sync::RwLock;

use crate::das;
use crate::erasure;
use crate::store::{self, FileEntry, GSet};

/// Wire protocol message types.
const MSG_PING: u8 = 1;
const MSG_PONG: u8 = 2;
const MSG_GET_CHUNK: u8 = 5;
const MSG_CHUNK_DATA: u8 = 6;
const MSG_CHUNK_NOT_FOUND: u8 = 7;
const MSG_REGISTRY: u8 = 8;
const MSG_REGISTRY_RESPONSE: u8 = 9;
// Reserved wire opcodes — the delta-sync and heartbeat protocol is specified but
// not yet wired at the transport layer.
#[allow(dead_code)]
const MSG_DELTA_SYNC: u8 = 10;
#[allow(dead_code)]
const MSG_HEARTBEAT: u8 = 11;
#[allow(dead_code)]
const MSG_HEARTBEAT_ACK: u8 = 12;

/// Registry response: includes Merkle root for completeness verification.
#[derive(serde::Serialize, serde::Deserialize)]
struct RegistryResponse {
    registry: GSet,
    merkle_root: String,
}

/// Peer liveness state.
#[derive(Debug)]
pub struct PeerHealth {
    last_seen: u64,
    consecutive_failures: u32,
}

/// Threshold: peer is dead after this many consecutive heartbeat failures.
const DEAD_AFTER_FAILURES: u32 = 3;

#[derive(Debug)]
pub struct SharedState {
    pub registry: GSet,
    pub data_dir: PathBuf,
    pub k: usize,
    pub n: usize,
    pub peers: Vec<String>,
    pub peer_capacities: Vec<u64>,
    pub device_id: String,
    pub peer_health: HashMap<String, PeerHealth>,
    pub last_sync_ts: HashMap<String, u64>,
}

/// ALPN protocol identifier for foculus.
const SYNC_ALPN: &[u8] = b"foculus/0";

/// ProtocolHandler for our sync protocol — dispatched by Router via ALPN.
#[derive(Debug, Clone)]
struct SyncProtocol {
    state: Arc<RwLock<SharedState>>,
}

impl ProtocolHandler for SyncProtocol {
    async fn accept(&self, conn: Connection) -> Result<(), AcceptError> {
        if let Err(e) = handle_connection(conn, self.state.clone()).await {
            eprintln!("sync protocol error: {}", e);
        }
        Ok(())
    }
}

pub struct SyncNode {
    state: Arc<RwLock<SharedState>>,
    endpoint: Endpoint,
    _router: Router,
}

impl SyncNode {
    pub async fn start(data_dir: &Path, k: usize, n: usize, port: u16) -> Result<Self> {
        assert!(n.is_power_of_two());
        assert!(k >= 1 && k <= n);

        std::fs::create_dir_all(data_dir)?;
        std::fs::create_dir_all(data_dir.join("chunks"))?;

        // Load or generate persistent secret key (stable node identity).
        let key_path = data_dir.join("secret.key");
        let secret_key = if key_path.exists() {
            let bytes = std::fs::read(&key_path)?;
            let bytes32: [u8; 32] = bytes.try_into().map_err(|_| anyhow::anyhow!("bad key file"))?;
            SecretKey::from(bytes32)
        } else {
            let key = SecretKey::generate(&mut rand::rng());
            std::fs::write(&key_path, key.to_bytes())?;
            key
        };

        // Create iroh endpoint with mDNS discovery and fixed port.
        let bind_addr = std::net::SocketAddrV4::new(std::net::Ipv4Addr::UNSPECIFIED, port);
        let endpoint = Endpoint::builder()
            .relay_mode(RelayMode::Disabled)
            .secret_key(secret_key)
            .address_lookup(MdnsAddressLookup::builder())
            .bind_addr(bind_addr)
            .context("invalid bind addr")?
            .bind()
            .await
            .context("failed to bind iroh endpoint")?;

        let device_id = endpoint.id().to_string();

        let registry_path = data_dir.join("registry.json");
        let registry = if registry_path.exists() {
            serde_json::from_str(&std::fs::read_to_string(&registry_path)?).unwrap_or_default()
        } else {
            GSet::new()
        };

        let peers_path = data_dir.join("peers.json");
        let peers: Vec<String> = if peers_path.exists() {
            serde_json::from_str(&std::fs::read_to_string(&peers_path)?).unwrap_or_default()
        } else {
            Vec::new()
        };

        let caps_path = data_dir.join("capacities.json");
        let peer_capacities: Vec<u64> = if caps_path.exists() {
            serde_json::from_str(&std::fs::read_to_string(&caps_path)?).unwrap_or_default()
        } else {
            vec![0; peers.len()]
        };

        let state = Arc::new(RwLock::new(SharedState {
            registry,
            data_dir: data_dir.to_path_buf(),
            k,
            n,
            peers,
            peer_capacities,
            device_id: device_id.clone(),
            peer_health: HashMap::new(),
            last_sync_ts: HashMap::new(),
        }));

        // Router dispatches incoming connections by ALPN.
        let proto = SyncProtocol { state: state.clone() };
        let router = Router::builder(endpoint.clone())
            .accept(SYNC_ALPN, proto)
            .spawn();

        println!("node started");
        println!("  id: {}", device_id);
        println!("  addr: {}", serde_json::to_string(&endpoint.addr()).unwrap_or_default());
        println!("  dir: {}", data_dir.display());
        println!("  erasure: k={}, n={}", k, n);

        Ok(Self { state, endpoint, _router: router })
    }

    pub async fn run_daemon(&self, sync_interval_secs: u64) -> Result<()> {
        println!("listening via iroh QUIC + mDNS (Router handles connections)");

        if sync_interval_secs > 0 {
            let state = self.state.clone();
            let ep = self.endpoint.clone();
            println!("auto-sync every {}s", sync_interval_secs);
            tokio::spawn(async move {
                let mut interval =
                    tokio::time::interval(std::time::Duration::from_secs(sync_interval_secs));
                interval.tick().await;
                loop {
                    interval.tick().await;
                    if let Err(e) = background_sync(&state, &ep).await {
                        eprintln!("auto-sync error: {}", e);
                    }
                }
            });
        }

        println!("running... press ctrl-c to stop");
        tokio::signal::ctrl_c().await?;
        println!("\nshutting down...");
        self.save().await?;
        Ok(())
    }

    /// Layer 1+4: put file with hash verification and DAS commitment.
    pub async fn put_file(&self, name: &str, data: &[u8]) -> Result<()> {
        let mut state = self.state.write().await;
        let shards = erasure::encode(data, state.k, state.n);

        let chunks_dir = state.data_dir.join("chunks");
        let mut shard_hashes = Vec::with_capacity(state.n);

        // Layer 4: DAS commitment over shards.
        let commitment = das::commit(&shards, state.k, data.len());
        let das_root = format!("{:?}", commitment.root);

        let placement = capacity_weighted_placement(state.n, &state.peer_capacities);

        for shard in &shards {
            let bytes = shard_to_bytes(shard);
            let hash = cyber_hemera::hash(&bytes);
            let hex = hash.to_hex();

            let keep_local = placement.get(shard.index).copied().unwrap_or(0) == 0;
            if keep_local || state.peers.is_empty() {
                let path = chunks_dir.join(&hex);
                std::fs::write(&path, &bytes)?;
            }

            shard_hashes.push(hex);
        }

        let timestamp = store::now_ms();
        let entry_hash = FileEntry::compute_hash(
            name,
            &shard_hashes,
            timestamp,
            &state.device_id,
        );

        let entry = FileEntry {
            name: name.to_string(),
            original_len: data.len(),
            k: state.k,
            n: state.n,
            shard_hashes,
            timestamp,
            entry_hash,
            device_id: state.device_id.clone(),
            das_root,
            shard_copies: 1, deleted: false,
        };

        state.registry.insert(entry);
        self.save_registry(&state)?;

        let local_count = placement.iter().filter(|&&d| d == 0).count();
        println!(
            "stored '{}' ({} bytes, {} shards, {} local)",
            name,
            data.len(),
            state.n,
            local_count
        );
        Ok(())
    }

    /// Layer 1: get file with hash verification of every fetched chunk.
    pub async fn get_file(&self, name: &str) -> Result<Vec<u8>> {
        let state = self.state.read().await;
        let entry = state
            .registry
            .get(name)
            .ok_or_else(|| anyhow::anyhow!("file '{}' not found", name))?
            .clone();
        let peers = state.peers.clone();
        let chunks_dir = state.data_dir.join("chunks");
        drop(state);

        let mut available_shards = Vec::new();

        for (idx, expected_hash) in entry.shard_hashes.iter().enumerate() {
            // Try local.
            let path = chunks_dir.join(expected_hash);
            if path.exists() {
                let bytes = std::fs::read(&path)?;
                // Layer 1: verify even local chunks (could be corrupted on disk).
                if !store::verify_chunk(&bytes, expected_hash) {
                    eprintln!(
                        "warning: local chunk {} corrupted, removing",
                        &expected_hash[..16]
                    );
                    std::fs::remove_file(&path)?;
                } else {
                    available_shards.push(bytes_to_shard(idx, &bytes));
                    if available_shards.len() >= entry.k {
                        break;
                    }
                    continue;
                }
            }

            // Try peers — Layer 1: verify hash of fetched data.
            for peer in &peers {
                match fetch_chunk_from_peer(&self.endpoint, peer, expected_hash).await {
                    Ok(bytes) => {
                        if store::verify_chunk(&bytes, expected_hash) {
                            std::fs::write(&path, &bytes)?;
                            available_shards.push(bytes_to_shard(idx, &bytes));
                            break;
                        } else {
                            eprintln!(
                                "warning: peer {} sent bad chunk {}, hash mismatch",
                                peer,
                                &expected_hash[..16]
                            );
                        }
                    }
                    Err(_) => continue,
                }
            }

            if available_shards.len() >= entry.k {
                break;
            }
        }

        if available_shards.len() < entry.k {
            anyhow::bail!(
                "only {} of {} required shards available for '{}'",
                available_shards.len(),
                entry.k,
                name
            );
        }

        Ok(erasure::decode(
            &available_shards,
            entry.k,
            entry.n,
            entry.original_len,
        ))
    }

    /// Layer 3+5: sync with peer — verify Merkle root, then merge + fetch.
    pub async fn sync_with(&self, peer: &str) -> Result<()> {
        println!("syncing with {}...", peer);

        let response = fetch_registry_with_proof(&self.endpoint, peer).await?;
        let remote_count = response.registry.len();

        // Layer 3: verify Merkle root matches claimed registry.
        let computed_root = response.registry.merkle_root();
        if computed_root != response.merkle_root {
            anyhow::bail!(
                "completeness check failed: peer {} sent registry with mismatched Merkle root\n  claimed: {}\n  computed: {}",
                peer,
                &response.merkle_root[..16],
                &computed_root[..16]
            );
        }

        // Layer 5: validated CRDT merge — verify every entry from peer.
        let mut state = self.state.write().await;
        let before = state.registry.len();
        let (_accepted, merge_rejected) = state.registry.validated_merge(&response.registry);
        let after = state.registry.len();
        self.save_registry(&state)?;

        if merge_rejected > 0 {
            eprintln!(
                "warning: rejected {} invalid entries from {} (forged hash, future timestamp, etc)",
                merge_rejected, peer
            );
        }

        println!(
            "registry: {} local + {} remote → {} merged ({} new, {} rejected)",
            before,
            remote_count,
            after,
            after - before,
            merge_rejected
        );

        // Layer 1: fetch missing chunks with hash verification.
        let chunks_dir = state.data_dir.join("chunks");
        let mut fetched = 0;
        let mut rejected = 0;

        for entry in state.registry.files.values() {
            for hash_hex in &entry.shard_hashes {
                let path = chunks_dir.join(hash_hex);
                if !path.exists() {
                    match fetch_chunk_from_peer(&self.endpoint, peer, hash_hex).await {
                        Ok(bytes) => {
                            if store::verify_chunk(&bytes, hash_hex) {
                                std::fs::write(&path, &bytes)?;
                                fetched += 1;
                            } else {
                                rejected += 1;
                                eprintln!(
                                    "rejected chunk {} from {}: hash mismatch",
                                    &hash_hex[..16],
                                    peer
                                );
                            }
                        }
                        Err(_) => {}
                    }
                }
            }
        }

        println!("fetched {} chunks from {} ({} rejected)", fetched, peer, rejected);
        Ok(())
    }

    pub async fn add_peer(&self, addr: &str, capacity: u64) -> Result<()> {
        let mut state = self.state.write().await;
        if !state.peers.contains(&addr.to_string()) {
            state.peers.push(addr.to_string());
            state.peer_capacities.push(capacity);
            save_peers(&state)?;
            println!("added peer: {} (capacity: {})", addr, format_bytes(capacity));
        } else {
            println!("peer already known: {}", addr);
        }
        Ok(())
    }

    pub async fn sync_all(&self) -> Result<()> {
        let state = self.state.read().await;
        let peers = state.peers.clone();
        drop(state);
        if peers.is_empty() {
            println!("no peers configured. use 'add-peer' first.");
            return Ok(());
        }
        for peer in &peers {
            if let Err(e) = self.sync_with(peer).await {
                eprintln!("sync with {} failed: {}", peer, e);
            }
        }
        Ok(())
    }

    pub async fn list_files(&self) -> Vec<String> {
        let state = self.state.read().await;
        state.registry.list().iter().map(|s| s.to_string()).collect()
    }

    /// Delete a file (tombstone). Shards will be GC'd on next gc() call.
    pub async fn delete_file(&self, name: &str) -> Result<()> {
        let mut state = self.state.write().await;
        let existing = state.registry.get_raw(name)
            .ok_or_else(|| anyhow::anyhow!("file '{}' not found", name))?
            .clone();

        if existing.deleted {
            anyhow::bail!("file '{}' already deleted", name);
        }

        // Insert tombstone with higher timestamp.
        let timestamp = store::now_ms();
        let shard_hashes = existing.shard_hashes.clone();
        let device_id = state.device_id.clone();
        let entry_hash = FileEntry::compute_hash(
            name, &shard_hashes, timestamp, &device_id,
        );

        state.registry.insert(FileEntry {
            name: name.to_string(),
            original_len: 0,
            k: existing.k,
            n: existing.n,
            shard_hashes,
            timestamp,
            entry_hash,
            device_id,
            das_root: "0".repeat(64),
            shard_copies: 1,
            deleted: true,
        });

        self.save_registry(&state)?;
        println!("deleted '{}' (tombstone)", name);
        Ok(())
    }

    /// Garbage collect: remove orphaned chunks not referenced by any live file.
    pub async fn gc(&self) -> Result<(usize, u64)> {
        let state = self.state.read().await;
        let live_hashes = state.registry.live_shard_hashes();
        let chunks_dir = state.data_dir.join("chunks");
        drop(state);

        let mut store = store::ChunkStore::new(&chunks_dir, 0)?;
        let (removed, freed) = store.gc(&live_hashes)?;
        println!("gc: removed {} orphaned chunks, freed {} bytes", removed, freed);
        Ok((removed, freed))
    }

    /// Integrity audit: hash-check all local chunks.
    pub async fn audit(&self) -> Result<(usize, usize)> {
        let state = self.state.read().await;
        let chunks_dir = state.data_dir.join("chunks");
        drop(state);

        let store = store::ChunkStore::new(&chunks_dir, 0)?;
        let (ok, corrupt, corrupt_hashes) = store.audit()?;

        if !corrupt_hashes.is_empty() {
            eprintln!("audit: {} corrupt chunks detected, removing", corrupt);
            let mut store_mut = store::ChunkStore::new(&chunks_dir, 0)?;
            for h in &corrupt_hashes {
                store_mut.remove(h)?;
            }
        }

        println!("audit: {} ok, {} corrupt (removed)", ok, corrupt);
        Ok((ok, corrupt))
    }

    /// Heartbeat: check liveness of all peers.
    /// Returns (alive_count, dead_peers).
    pub async fn heartbeat(&self) -> Result<(usize, Vec<String>)> {
        let state = self.state.read().await;
        let peers = state.peers.clone();
        drop(state);

        let mut alive = 0;
        let mut dead = Vec::new();

        for peer in &peers {
            let is_alive = ping_peer(&self.endpoint, peer).await;

            let mut state = self.state.write().await;
            let health = state.peer_health
                .entry(peer.clone())
                .or_insert(PeerHealth { last_seen: 0, consecutive_failures: 0 });

            if is_alive {
                health.last_seen = store::now_ms();
                health.consecutive_failures = 0;
                alive += 1;
            } else {
                health.consecutive_failures += 1;
                if health.consecutive_failures >= DEAD_AFTER_FAILURES {
                    dead.push(peer.clone());
                }
            }
        }

        Ok((alive, dead))
    }

    /// Compute deterministic (k, n) from the number of alive peers.
    /// All nodes using the same algorithm + same device set = same result.
    pub fn compute_params(alive_devices: usize, redundancy_f: usize) -> (usize, usize) {
        if alive_devices < 2 {
            return (1, 1);
        }
        let n = alive_devices.min(8).next_power_of_two();
        let f = redundancy_f.min(n - 1);
        let k = n - f;
        (k, n)
    }

    pub async fn status(&self) -> (usize, usize, usize, usize) {
        let state = self.state.read().await;
        let files = state.registry.len();
        let peers = state.peers.len();
        let chunks = std::fs::read_dir(state.data_dir.join("chunks"))
            .map(|d| d.count())
            .unwrap_or(0);
        (files, peers, chunks, 0)
    }

    pub fn node_id(&self) -> String {
        self.endpoint.id().to_string()
    }

    pub async fn save(&self) -> Result<()> {
        let state = self.state.read().await;
        self.save_registry(&state)
    }

    fn save_registry(&self, state: &SharedState) -> Result<()> {
        let path = state.data_dir.join("registry.json");
        let json = serde_json::to_string_pretty(&state.registry)?;
        std::fs::write(path, json)?;
        Ok(())
    }
}

// ── Background auto-sync ──

async fn background_sync(state: &Arc<RwLock<SharedState>>, ep: &Endpoint) -> Result<()> {
    {
        let mut s = state.write().await;
        let path = s.data_dir.join("registry.json");
        if path.exists() {
            if let Ok(disk_reg) = serde_json::from_str::<GSet>(&std::fs::read_to_string(&path)?) {
                s.registry.merge(&disk_reg);
            }
        }
    }

    let s = state.read().await;
    let peers = s.peers.clone();
    let chunks_dir = s.data_dir.join("chunks");
    drop(s);

    if peers.is_empty() {
        return Ok(());
    }

    let mut total_new_files = 0;
    let mut total_fetched = 0;

    for peer in &peers {
        // Layer 3: fetch registry with proof.
        let response = match fetch_registry_with_proof(ep, peer).await {
            Ok(r) => r,
            Err(_) => continue,
        };

        let computed_root = response.registry.merkle_root();
        if computed_root != response.merkle_root {
            eprintln!("[auto-sync] completeness check failed for peer {}", peer);
            continue;
        }

        let mut s = state.write().await;
        let before = s.registry.len();
        let (_accepted, _rejected) = s.registry.validated_merge(&response.registry);
        let after = s.registry.len();
        total_new_files += after - before;

        if after > before {
            let path = s.data_dir.join("registry.json");
            std::fs::write(&path, serde_json::to_string_pretty(&s.registry)?)?;
        }

        // Layer 1: fetch + verify chunks.
        for entry in s.registry.files.values() {
            for hash_hex in &entry.shard_hashes {
                let path = chunks_dir.join(hash_hex);
                if !path.exists() {
                    if let Ok(bytes) = fetch_chunk_from_peer(ep, peer, hash_hex).await {
                        if store::verify_chunk(&bytes, hash_hex) {
                            std::fs::write(&path, &bytes)?;
                            total_fetched += 1;
                        }
                    }
                }
            }
        }
    }

    if total_new_files > 0 || total_fetched > 0 {
        println!(
            "[auto-sync] {} new files, {} chunks fetched",
            total_new_files, total_fetched
        );
    }
    Ok(())
}

// ── Capacity-weighted placement ──

fn capacity_weighted_placement(n_shards: usize, peer_capacities: &[u64]) -> Vec<usize> {
    let n_devices = peer_capacities.len() + 1;
    if n_devices == 0 || n_shards == 0 {
        return vec![0; n_shards];
    }
    let mut caps: Vec<u64> = Vec::with_capacity(n_devices);
    caps.push(u64::MAX);
    caps.extend_from_slice(peer_capacities);
    let total_cap: u128 = caps.iter().map(|&c| c.max(1) as u128).sum();
    let mut alloc = vec![0usize; n_devices];
    let mut assigned = 0;
    for d in 0..n_devices {
        let share = (n_shards as u128 * caps[d].max(1) as u128 / total_cap) as usize;
        alloc[d] = share;
        assigned += share;
    }
    let mut remainder = n_shards.saturating_sub(assigned);
    let mut order: Vec<usize> = (0..n_devices).collect();
    order.sort_by(|&a, &b| caps[b].cmp(&caps[a]));
    for &d in &order {
        if remainder == 0 { break; }
        alloc[d] += 1;
        remainder -= 1;
    }
    let mut placement = Vec::with_capacity(n_shards);
    for (device_idx, &count) in alloc.iter().enumerate() {
        for _ in 0..count {
            placement.push(device_idx);
        }
    }
    placement.truncate(n_shards);
    placement
}

fn save_peers(state: &SharedState) -> Result<()> {
    std::fs::write(
        state.data_dir.join("peers.json"),
        serde_json::to_string_pretty(&state.peers)?,
    )?;
    std::fs::write(
        state.data_dir.join("capacities.json"),
        serde_json::to_string_pretty(&state.peer_capacities)?,
    )?;
    Ok(())
}

fn format_bytes(bytes: u64) -> String {
    if bytes == 0 { return "unlimited".to_string(); }
    const GB: u64 = 1_000_000_000;
    const MB: u64 = 1_000_000;
    if bytes >= GB { format!("{:.1} GB", bytes as f64 / GB as f64) }
    else if bytes >= MB { format!("{:.1} MB", bytes as f64 / MB as f64) }
    else { format!("{} B", bytes) }
}

// ── Server: handle incoming connections ──

async fn handle_connection(conn: Connection, state: Arc<RwLock<SharedState>>) -> Result<()> {
    let (mut send, mut recv) = conn.accept_bi().await?;
    let msg_type = recv.read_u8().await?;

    match msg_type {
        MSG_PING => {
            send.write_u8(MSG_PONG).await?;
        }
        MSG_GET_CHUNK => {
            let hash_hex = read_string(&mut recv).await?;
            let state = state.read().await;
            let path = state.data_dir.join("chunks").join(&hash_hex);
            if path.exists() {
                let data = std::fs::read(&path)?;
                send.write_u8(MSG_CHUNK_DATA).await?;
                write_bytes(&mut send, &data).await?;
            } else {
                send.write_u8(MSG_CHUNK_NOT_FOUND).await?;
            }
        }
        MSG_REGISTRY => {
            let state = state.read().await;
            let registry: GSet = {
                let path = state.data_dir.join("registry.json");
                if path.exists() {
                    serde_json::from_str(&std::fs::read_to_string(&path)?).unwrap_or_default()
                } else {
                    state.registry.clone()
                }
            };
            let merkle_root = registry.merkle_root();
            let response = RegistryResponse { registry, merkle_root };
            send.write_u8(MSG_REGISTRY_RESPONSE).await?;
            write_bytes(&mut send, serde_json::to_string(&response)?.as_bytes()).await?;
        }
        _ => {}
    }

    send.flush().await?;
    send.finish()?;
    // Wait for the peer to close the connection — keeps it alive
    // until the client has read our response.
    conn.closed().await;
    Ok(())
}

// ── Client: fetch from peers ──

async fn connect_peer(ep: &Endpoint, peer: &str) -> Result<Connection> {
    // Try parsing as EndpointId first, then as EndpointAddr JSON.
    let endpoint_addr: iroh::EndpointAddr = if let Ok(id) = peer.parse::<EndpointId>() {
        iroh::EndpointAddr::new(id)
    } else if let Ok(addr) = serde_json::from_str::<iroh::EndpointAddr>(peer) {
        addr
    } else {
        anyhow::bail!("invalid peer: expected node ID or JSON EndpointAddr")
    };
    ep.connect(endpoint_addr, SYNC_ALPN)
        .await
        .context("failed to connect to peer")
}

async fn fetch_chunk_from_peer(ep: &Endpoint, peer: &str, hash_hex: &str) -> Result<Vec<u8>> {
    let conn = connect_peer(ep, peer).await?;
    let (mut send, mut recv) = conn.open_bi().await?;
    send.write_u8(MSG_GET_CHUNK).await?;
    write_bytes(&mut send, hash_hex.as_bytes()).await?;
    send.flush().await?;
    let response = recv.read_u8().await?;
    if response == MSG_CHUNK_DATA {
        read_bytes(&mut recv).await
    } else {
        anyhow::bail!("chunk not found on peer")
    }
}

async fn fetch_registry_with_proof(ep: &Endpoint, peer: &str) -> Result<RegistryResponse> {
    let conn = connect_peer(ep, peer).await?;
    let (mut send, mut recv) = conn.open_bi().await?;
    send.write_u8(MSG_REGISTRY).await?;
    send.flush().await?;
    let response = recv.read_u8().await?;
    if response == MSG_REGISTRY_RESPONSE {
        let data = read_bytes(&mut recv).await?;
        let json = String::from_utf8(data)?;
        Ok(serde_json::from_str(&json)?)
    } else {
        anyhow::bail!("unexpected response type {}", response)
    }
}

async fn ping_peer(ep: &Endpoint, peer: &str) -> bool {
    let conn = match tokio::time::timeout(
        std::time::Duration::from_secs(3),
        connect_peer(ep, peer),
    ).await {
        Ok(Ok(c)) => c,
        _ => return false,
    };
    let bi = match tokio::time::timeout(
        std::time::Duration::from_secs(2),
        conn.open_bi(),
    ).await {
        Ok(Ok(bi)) => bi,
        _ => return false,
    };
    let (mut send, mut recv) = bi;
    if send.write_u8(MSG_PING).await.is_err() { return false; }
    matches!(
        tokio::time::timeout(std::time::Duration::from_secs(2), recv.read_u8()).await,
        Ok(Ok(MSG_PONG))
    )
}

// ── Wire helpers (generic over AsyncRead/AsyncWrite) ──

async fn write_bytes(w: &mut (impl AsyncWriteExt + Unpin), data: &[u8]) -> Result<()> {
    w.write_u32(data.len() as u32).await?;
    w.write_all(data).await?;
    w.flush().await?;
    Ok(())
}

async fn read_bytes(r: &mut (impl AsyncReadExt + Unpin)) -> Result<Vec<u8>> {
    let len = r.read_u32().await? as usize;
    let mut buf = vec![0u8; len];
    r.read_exact(&mut buf).await?;
    Ok(buf)
}

async fn read_string(r: &mut (impl AsyncReadExt + Unpin)) -> Result<String> {
    Ok(String::from_utf8(read_bytes(r).await?)?)
}

fn shard_to_bytes(shard: &erasure::Shard) -> Vec<u8> {
    let mut bytes = Vec::with_capacity(shard.data.len() * 8);
    for &elem in &shard.data {
        bytes.extend_from_slice(&elem.as_u64().to_le_bytes());
    }
    bytes
}

fn bytes_to_shard(index: usize, bytes: &[u8]) -> erasure::Shard {
    let mut data = Vec::with_capacity(bytes.len() / 8);
    for chunk in bytes.chunks(8) {
        if chunk.len() == 8 {
            data.push(nebu::Goldilocks::new(u64::from_le_bytes(
                chunk.try_into().unwrap(),
            )));
        }
    }
    erasure::Shard { index, data }
}