#![cfg(feature = "net")]

use crate::net::sign::{
    decode_public_key_base64, encode_public_key_base64, encode_signature_base64, sign_payload,
    verify_signature_base64, KeyError, KeyMaterial,
};
use crate::net::{
    attestation::{aggregate_attestations, Attestation},
    availability::{self, encode_shares, AvailabilityEvidence},
    blob::BlobJson,
    checkpoint::{
        latest_log_cutoff, load_latest_checkpoint, write_checkpoint, AnchorCheckpoint,
        CheckpointSignature,
    },
    governance::MembershipPolicy,
    schema::{
        AnchorCodecError, AnchorEnvelope, AnchorJson, DaCommitmentJson, ENVELOPE_SCHEMA_VERSION,
        NETWORK_ID, SCHEMA_ENVELOPE,
    },
    stake_registry::StakeRegistry,
};
use crate::{
    build_merkle_proof, compute_fold_digest, julian_genesis_anchor, merkle_root, parse_log_file,
    read_fold_digest_hint,
    rollup::{
        settle_rollup_with_rewards, RollupCommitment, RollupFaultEvidence, RollupSettlementMode,
        ZkRollupProof,
    },
    AnchorVote, EntryAnchor, LedgerAnchor,
};
use base64::{engine::general_purpose::STANDARD as BASE64, Engine as _};
use ed25519_dalek::{Signer, SigningKey};
use futures::StreamExt;
use hex;
use libp2p::{
    gossipsub::{self, IdentTopic, MessageAuthenticity, PublishError, ValidationMode},
    identify, identity,
    kad::{self, store::MemoryStore},
    noise,
    swarm::{NetworkBehaviour, Swarm, SwarmEvent},
    tcp, yamux, Multiaddr, SwarmBuilder,
};
use once_cell::sync::Lazy;
use serde::{Deserialize, Serialize};
use serde_json;
use sha2::{Digest, Sha256};
use std::net::SocketAddr;
use std::{
    collections::{HashMap, HashSet, VecDeque},
    env, fs, io,
    path::{Path, PathBuf},
    str,
    sync::{
        atomic::{AtomicBool, AtomicU64, Ordering},
        Arc,
    },
    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::{Mutex, Semaphore};
use tokio::{select, signal, time};

static TOPIC_ANCHORS: Lazy<IdentTopic> = Lazy::new(|| IdentTopic::new("jrocnet/anchors/v1"));
static TOPIC_EVIDENCE: Lazy<IdentTopic> = Lazy::new(|| IdentTopic::new("jrocnet/evidence/v1"));
static NO_GOSSIP_PEERS_LOGGED: AtomicBool = AtomicBool::new(false);
const MAX_ENVELOPE_BYTES: usize = 64 * 1024;
const MAX_ANCHOR_ENTRIES: usize = 10_000;
const SEEN_CACHE_LIMIT: usize = 2048;
const INVALID_THRESHOLD: usize = 5;
const MAX_HEADER_BYTES: usize = 32 * 1024;
const DEFAULT_MAX_REQUEST_BYTES: usize = 64 * 1024 * 1024;
const DEFAULT_REQUEST_TIMEOUT_MS: u64 = 10_000;
const DEFAULT_BLOB_MAX_CONCURRENCY: usize = 128;

/// Per-namespace limits applied to blob ingestion.
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct NamespaceRule {
    /// Maximum blob size in bytes permitted for this namespace.
    pub max_bytes: Option<usize>,
    /// Retention policy (days) before pruning stored blobs.
    pub retention_days: Option<u64>,
    /// Rate limit in blobs per minute.
    pub max_per_min: Option<u32>,
    /// Minimum fee required to accept a blob.
    pub min_fee: Option<u64>,
    /// Operator reward in basis points (default 50%).
    #[serde(default)]
    pub operator_reward_bps: Option<u16>,
}

fn policy_permits(policy: &dyn MembershipPolicy, key: &[u8]) -> bool {
    let members = policy.current_members();
    if members.is_empty() {
        // Empty list means everyone allowed (bootstrap mode)
        return true;
    }
    members.iter().any(|vk| vk.to_bytes().as_slice() == key)
}

/// Configuration and runtime context for the JULIAN network node.
pub struct NetConfig {
    /// Human-readable node identifier used in logs and envelopes.
    pub node_id: String,
    /// Multiaddr on which the node listens for incoming peers.
    pub listen_addr: Multiaddr,
    /// Optional bootstrap peers dialed on startup.
    pub bootstraps: Vec<Multiaddr>,
    /// Directory containing ledger transcript logs to monitor.
    pub log_dir: PathBuf,
    /// Quorum threshold used when reconciling anchors.
    pub quorum: usize,
    /// Interval between anchor recomputation and gossip broadcasts.
    pub broadcast_interval: Duration,
    /// Signing and libp2p keys backing the node identity.
    pub key_material: KeyMaterial,
    /// Membership governance policy.
    pub membership_policy: Arc<dyn MembershipPolicy>,
    /// Optional checkpoint interval (in broadcasts).
    pub checkpoint_interval: Option<u64>,
    /// Directory used to store blobs and share commitments.
    pub blob_dir: Option<PathBuf>,
    /// TCP socket for the blob ingest server.
    pub blob_listen: Option<SocketAddr>,
    /// Maximum blob size in bytes.
    pub max_blob_bytes: Option<usize>,
    /// Retention window in days for stored blobs.
    pub blob_retention_days: Option<u64>,
    /// Optional per-namespace policy overrides.
    pub blob_policies: Option<HashMap<String, NamespaceRule>>,
    /// Optional bearer token for blob HTTP endpoints.
    pub blob_auth_token: Option<String>,
    /// Maximum concurrent blob HTTP connections.
    pub blob_max_concurrency: usize,
    /// Request read timeout for blob HTTP endpoints.
    pub blob_request_timeout: Duration,
    /// Attestation quorum required for DA commitments.
    pub attestation_quorum: usize,
    /// Path to the stake registry used for fees and slashing.
    pub stake_registry_path: Option<PathBuf>,
    metrics: Arc<Metrics>,
    metrics_addr: Option<SocketAddr>,
}

impl NetConfig {
    /// Constructs a networking configuration for the JULIAN net CLI.
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        node_id: String,
        listen_addr: Multiaddr,
        bootstraps: Vec<Multiaddr>,
        log_dir: PathBuf,
        quorum: usize,
        broadcast_interval: Duration,
        key_material: KeyMaterial,
        metrics_addr: Option<SocketAddr>,
        membership_policy: Arc<dyn MembershipPolicy>,
        checkpoint_interval: Option<u64>,
        blob_dir: Option<PathBuf>,
        blob_listen: Option<SocketAddr>,
        max_blob_bytes: Option<usize>,
        blob_retention_days: Option<u64>,
        blob_policies: Option<HashMap<String, NamespaceRule>>,
        blob_auth_token: Option<String>,
        blob_max_concurrency: Option<usize>,
        blob_request_timeout_ms: Option<u64>,
        attestation_quorum: Option<usize>,
    ) -> Self {
        let attestation_quorum = attestation_quorum.unwrap_or(quorum);
        let stake_registry_path = blob_dir.as_ref().map(|dir| dir.join("stake_registry.json"));
        let blob_max_concurrency = blob_max_concurrency.unwrap_or(DEFAULT_BLOB_MAX_CONCURRENCY);
        let blob_request_timeout =
            Duration::from_millis(blob_request_timeout_ms.unwrap_or(DEFAULT_REQUEST_TIMEOUT_MS));
        Self {
            node_id,
            listen_addr,
            bootstraps,
            log_dir,
            quorum,
            broadcast_interval,
            key_material,
            membership_policy,
            checkpoint_interval,
            blob_dir,
            blob_listen,
            max_blob_bytes,
            blob_retention_days,
            blob_policies,
            blob_auth_token,
            blob_max_concurrency,
            blob_request_timeout,
            attestation_quorum,
            stake_registry_path,
            metrics: Arc::new(Metrics::default()),
            metrics_addr,
        }
    }
}

#[derive(Clone)]
struct BlobServiceConfig {
    base_dir: PathBuf,
    listen: SocketAddr,
    max_bytes: Option<usize>,
    retention_days: Option<u64>,
    policies: Option<HashMap<String, NamespaceRule>>,
    signing: SigningKey,
    verifying_b64: String,
    stake_registry_path: Option<PathBuf>,
    membership_policy: Arc<dyn MembershipPolicy>,
    auth_token: Option<String>,
    max_concurrency: usize,
    request_timeout: Duration,
    rate_limits: Arc<Mutex<HashMap<String, RateState>>>,
}

#[derive(Debug, Clone)]
struct RateState {
    window_start: Instant,
    count: u32,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct EvidenceRecord {
    namespace: String,
    blob_hash: String,
    pk: String,
    reason: String,
    ts: u64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct EvidenceEnvelope {
    public_key: String,
    signature: String,
    evidence: EvidenceRecord,
}

fn sign_evidence(sk: &SigningKey, ev: &EvidenceRecord) -> EvidenceEnvelope {
    let payload = serde_json::to_vec(ev).expect("evidence encode");
    let sig = sk.sign(&payload);
    EvidenceEnvelope {
        public_key: encode_public_key_base64(&sk.verifying_key()),
        signature: BASE64.encode(sig.to_bytes()),
        evidence: ev.clone(),
    }
}

struct HttpRequest {
    method: String,
    path: String,
    headers: HashMap<String, String>,
    body: Vec<u8>,
}

async fn run_blob_service(cfg: BlobServiceConfig) {
    if let Err(err) = fs::create_dir_all(&cfg.base_dir) {
        eprintln!("blob dir init error: {err}");
        return;
    }
    let _retention_days = cfg.retention_days;
    let listener = match TcpListener::bind(cfg.listen).await {
        Ok(l) => l,
        Err(err) => {
            eprintln!("failed to bind blob listener {}: {err}", cfg.listen);
            return;
        }
    };
    println!("QSYS|mod=BLOB|evt=LISTEN|addr={}", cfg.listen);
    let limiter = Arc::new(Semaphore::new(cfg.max_concurrency));
    loop {
        match listener.accept().await {
            Ok((mut stream, _addr)) => {
                let permit = match limiter.clone().acquire_owned().await {
                    Ok(permit) => permit,
                    Err(_) => {
                        eprintln!("blob accept error: limiter closed");
                        continue;
                    }
                };
                let cfg_clone = cfg.clone();
                tokio::spawn(async move {
                    let _permit = permit;
                    if let Err(err) = handle_blob_connection(&mut stream, cfg_clone).await {
                        eprintln!("blob connection error: {err}");
                    }
                });
            }
            Err(err) => {
                eprintln!("blob accept error: {err}");
                break;
            }
        }
    }
}

async fn handle_blob_connection(stream: &mut TcpStream, cfg: BlobServiceConfig) -> io::Result<()> {
    let max_body = cfg.max_bytes.unwrap_or(DEFAULT_MAX_REQUEST_BYTES);
    let req = match read_http_request(stream, MAX_HEADER_BYTES, max_body, cfg.request_timeout).await
    {
        Ok(r) => r,
        Err(err) => {
            let resp = build_response(
                "400 Bad Request",
                format!("{{\"error\":\"{err}\"}}"),
                "application/json",
            );
            let _ = stream.write_all(&resp).await;
            let _ = stream.shutdown().await;
            return Ok(());
        }
    };
    if let Some(token) = cfg.auth_token.as_deref() {
        let mut ok = false;
        if let Some(auth) = req.headers.get("authorization") {
            if let Some(bearer) = auth.strip_prefix("Bearer ") {
                ok = bearer == token;
            }
        }
        if let Some(key) = req.headers.get("x-api-key") {
            ok = ok || key == token;
        }
        if !ok {
            let resp = build_response(
                "401 Unauthorized",
                "{\"error\":\"unauthorized\"}".to_string(),
                "application/json",
            );
            let _ = stream.write_all(&resp).await;
            let _ = stream.shutdown().await;
            return Ok(());
        }
    }
    let (status, body, content_type) = match req.method.as_str() {
        "GET" if req.path == "/healthz" => (
            "200 OK".to_string(),
            format!(
                "{{\"status\":\"ok\",\"network\":\"{}\",\"version\":\"{}\"}}",
                NETWORK_ID,
                env!("CARGO_PKG_VERSION")
            ),
            "application/json".to_string(),
        ),
        "POST" if req.path.starts_with("/submit_blob") => {
            match handle_submit_blob(&req, &cfg).await {
                Ok(json) => ("200 OK".to_string(), json, "application/json".to_string()),
                Err(err) => (
                    "400 Bad Request".to_string(),
                    format!("{{\"error\":\"{err}\"}}"),
                    "application/json".to_string(),
                ),
            }
        }
        "GET" if req.path.starts_with("/commitment/") => {
            let json = handle_commitment(&req, &cfg).unwrap_or_else(|e| e);
            if json.starts_with("{") {
                ("200 OK".to_string(), json, "application/json".to_string())
            } else {
                ("404 Not Found".to_string(), json, "text/plain".to_string())
            }
        }
        "GET" if req.path.starts_with("/sample/") => {
            let (code, body) = handle_sample(&req, &cfg).unwrap_or((404, "Not Found".to_string()));
            let status = match code {
                200 => "200 OK",
                400 => "400 Bad Request",
                _ => "404 Not Found",
            };
            (
                status.to_string(),
                body,
                if code == 200 {
                    "application/json"
                } else {
                    "text/plain"
                }
                .to_string(),
            )
        }
        "GET" if req.path.starts_with("/prove_storage/") => {
            let (code, body) =
                handle_prove_storage(&req, &cfg).unwrap_or((404, "Not Found".to_string()));
            let status = match code {
                200 => "200 OK",
                400 => "400 Bad Request",
                _ => "404 Not Found",
            };
            (
                status.to_string(),
                body,
                if code == 200 {
                    "application/json"
                } else {
                    "text/plain"
                }
                .to_string(),
            )
        }
        "POST" if req.path.starts_with("/rollup_settle") => {
            match handle_rollup_settle(&req, &cfg) {
                Ok(json) => ("200 OK".to_string(), json, "application/json".to_string()),
                Err(err) => (
                    "400 Bad Request".to_string(),
                    format!("{{\"error\":\"{err}\"}}"),
                    "application/json".to_string(),
                ),
            }
        }
        _ => (
            "404 Not Found".to_string(),
            "Not Found".to_string(),
            "text/plain".to_string(),
        ),
    };

    let resp = build_response(&status, body, &content_type);
    stream.write_all(&resp).await?;
    stream.shutdown().await
}

async fn read_http_request(
    stream: &mut TcpStream,
    max_header_bytes: usize,
    max_body_bytes: usize,
    timeout: Duration,
) -> io::Result<HttpRequest> {
    let mut buf = Vec::new();
    let mut header_end = None;
    loop {
        let mut tmp = [0u8; 1024];
        let n = time::timeout(timeout, stream.read(&mut tmp))
            .await
            .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "read timeout"))??;
        if n == 0 {
            break;
        }
        buf.extend_from_slice(&tmp[..n]);
        if buf.len() > max_header_bytes && header_end.is_none() {
            return Err(io::Error::new(
                io::ErrorKind::InvalidData,
                "header too large",
            ));
        }
        if let Some(pos) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
            header_end = Some(pos + 4);
            break;
        }
    }
    let end = header_end
        .ok_or_else(|| io::Error::new(io::ErrorKind::UnexpectedEof, "malformed request"))?;
    let header_str = str::from_utf8(&buf[..end])
        .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "invalid header"))?;
    let mut lines = header_str.split("\r\n").filter(|l| !l.is_empty());
    let request_line = lines
        .next()
        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "missing request line"))?;
    let mut parts = request_line.split_whitespace();
    let method = parts.next().unwrap_or("").to_string();
    let path = parts.next().unwrap_or("").to_string();
    let mut headers = HashMap::new();
    for line in lines {
        if let Some((name, value)) = line.split_once(":") {
            headers.insert(name.trim().to_ascii_lowercase(), value.trim().to_string());
        }
    }
    let content_len: usize = headers
        .get("content-length")
        .and_then(|v| v.parse().ok())
        .unwrap_or(0);
    if content_len > max_body_bytes {
        return Err(io::Error::new(
            io::ErrorKind::InvalidData,
            "content-length exceeds limit",
        ));
    }
    let mut body = if end < buf.len() {
        buf[end..].to_vec()
    } else {
        Vec::new()
    };
    while body.len() < content_len {
        let remaining = content_len - body.len();
        let mut tmp = vec![0u8; remaining.min(8192)];
        let n = time::timeout(timeout, stream.read(&mut tmp))
            .await
            .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "read timeout"))??;
        if n == 0 {
            break;
        }
        body.extend_from_slice(&tmp[..n]);
    }
    if body.len() < content_len {
        return Err(io::Error::new(
            io::ErrorKind::UnexpectedEof,
            "incomplete request body",
        ));
    }
    Ok(HttpRequest {
        method,
        path,
        headers,
        body,
    })
}

fn build_response(status: &str, body: String, content_type: &str) -> Vec<u8> {
    format!(
        "HTTP/1.1 {status}\r\nContent-Length: {}\r\nContent-Type: {content_type}\r\nConnection: close\r\n\r\n{body}",
        body.len()
    )
    .into_bytes()
}

fn sanitize_token(token: &str) -> Option<String> {
    if token
        .chars()
        .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
    {
        Some(token.to_string())
    } else {
        None
    }
}

fn namespace_rule<'a>(cfg: &'a BlobServiceConfig, namespace: &str) -> Option<NamespaceRule> {
    cfg.policies
        .as_ref()
        .and_then(|m| m.get(namespace).cloned())
}

fn read_shares(
    base: &Path,
    namespace: &str,
    hash: &str,
    total: usize,
) -> Result<Vec<Vec<u8>>, NetworkError> {
    let (_, _, share_dir) = blob_paths(base, namespace, hash);
    let mut shares = Vec::with_capacity(total);
    for idx in 0..total {
        let path = share_dir.join(format!("{idx}.share"));
        let data = fs::read(&path)
            .map_err(|err| NetworkError::Io(format!("failed to read {}: {err}", path.display())))?;
        shares.push(data);
    }
    Ok(shares)
}

fn share_hashes(shares: &[Vec<u8>]) -> Vec<[u8; 32]> {
    shares
        .iter()
        .map(|s| {
            let mut hasher = Sha256::new();
            hasher.update(s);
            hasher.finalize().into()
        })
        .collect()
}

async fn handle_submit_blob(req: &HttpRequest, cfg: &BlobServiceConfig) -> Result<String, String> {
    let namespace = req
        .headers
        .get("x-namespace")
        .and_then(|v| sanitize_token(v))
        .unwrap_or_else(|| "default".to_string());
    let fee: Option<u64> = req.headers.get("x-fee").and_then(|v| v.parse().ok());
    let publisher_pk = req.headers.get("x-publisher").cloned();
    let publisher_sig = req.headers.get("x-publisher-sig").cloned();
    if let Some(rule) = namespace_rule(cfg, &namespace) {
        if let Some(max_per_min) = rule.max_per_min {
            if max_per_min > 0 {
                let mut limits = cfg.rate_limits.lock().await;
                let entry = limits.entry(namespace.clone()).or_insert(RateState {
                    window_start: Instant::now(),
                    count: 0,
                });
                if entry.window_start.elapsed() >= Duration::from_secs(60) {
                    entry.window_start = Instant::now();
                    entry.count = 0;
                }
                if entry.count >= max_per_min {
                    return Err("rate limit exceeded".into());
                }
                entry.count += 1;
            }
        }
        if let Some(max) = rule.max_bytes {
            if req.body.len() > max {
                return Err(format!("blob exceeds max_bytes for namespace {namespace}"));
            }
        }
        if let Some(min_fee) = rule.min_fee {
            if fee.unwrap_or(0) < min_fee {
                return Err("fee below namespace minimum".into());
            }
        }
    }
    if let Some(max) = cfg.max_bytes {
        if req.body.len() > max {
            return Err("blob exceeds max_bytes".into());
        }
    }
    let blob = BlobJson::from_bytes(&namespace, &req.body);
    let data_shards = 4u8;
    let parity_shards = 2u8;
    let (shares, commitment) = encode_shares(&req.body, data_shards, parity_shards)
        .map_err(|e| format!("encode error: {e}"))?;

    let (meta_path, blob_path, share_dir) = blob_paths(&cfg.base_dir, &namespace, &blob.hash);
    if let Some(parent) = meta_path.parent() {
        fs::create_dir_all(parent).map_err(|err| format!("create dirs: {err}"))?;
    }
    fs::write(&blob_path, &req.body).map_err(|err| format!("write blob: {err}"))?;
    fs::create_dir_all(&share_dir).map_err(|err| format!("create share dir: {err}"))?;
    for (idx, share) in shares.iter().enumerate() {
        let path = share_dir.join(format!("{idx}.share"));
        fs::write(&path, share).map_err(|err| format!("write share: {err}"))?;
    }

    let sig = sign_payload(&cfg.signing, commitment.share_root.as_bytes());
    let operator_att = StoredAttestation {
        pk: cfg.verifying_b64.clone(),
        sig: encode_signature_base64(&sig),
    };
    let mut attestations = vec![operator_att];
    if let (Some(att_pk), Some(att_sig)) = (
        req.headers.get("x-attestation-pk"),
        req.headers.get("x-attestation-sig"),
    ) {
        if verify_signature_base64(att_pk, commitment.share_root.as_bytes(), att_sig).is_ok() {
            attestations.push(StoredAttestation {
                pk: att_pk.clone(),
                sig: att_sig.clone(),
            });
        }
    }
    let meta = BlobMeta {
        namespace: namespace.clone(),
        hash: blob.hash.clone(),
        size: blob.size,
        data_shards,
        parity_shards,
        share_root: commitment.share_root.clone(),
        pedersen_root: Some(commitment.pedersen_root.clone()),
        attestations,
        publisher_pk,
    };
    if let Some(pk) = meta.publisher_pk.as_ref() {
        if let Some(sig) = publisher_sig.as_ref() {
            verify_signature_base64(pk, commitment.share_root.as_bytes(), sig)
                .map_err(|_| "invalid x-publisher-sig signature".to_string())?;
        } else {
            return Err("missing x-publisher-sig header for publisher".into());
        }
    }
    save_blob_meta(&cfg.base_dir, &meta).map_err(|err| err.to_string())?;
    // Fees and rewards (split between operator and attestors by stake).
    if let (Some(path), Some(amount)) = (&cfg.stake_registry_path, fee) {
        let payer = meta
            .publisher_pk
            .clone()
            .unwrap_or_else(|| cfg.verifying_b64.clone());
        match StakeRegistry::load(path) {
            Ok(mut reg) => {
                reg.debit_fee(&payer, amount)
                    .map_err(|err| format!("fee debit failed: {err}"))?;
                let ns_rule = namespace_rule(cfg, &namespace).unwrap_or_default();
                let op_bps = ns_rule.operator_reward_bps.unwrap_or(5000) as u64;
                let operator_cut = amount.saturating_mul(op_bps).saturating_div(10_000);
                let attestor_pool = amount.saturating_sub(operator_cut);
                reg.credit_reward(&cfg.verifying_b64, operator_cut);
                let attestors = meta.attestations.clone();
                let total_weight: u64 = attestors.iter().filter_map(|a| reg.stake_for(&a.pk)).sum();
                if total_weight == 0 {
                    reg.credit_reward(&cfg.verifying_b64, attestor_pool);
                } else {
                    for att in attestors {
                        if let Some(w) = reg.stake_for(&att.pk) {
                            let share = attestor_pool.saturating_mul(w) / total_weight;
                            if share > 0 {
                                reg.credit_reward(&att.pk, share);
                            }
                        }
                    }
                }
                reg.save(path)
                    .map_err(|err| format!("failed to persist stake registry: {err}"))?;
            }
            Err(err) => return Err(format!("failed to load stake registry: {err}")),
        }
    }

    let response = serde_json::json!({
        "status": "ok",
        "size": blob.size,
        "hash": blob.hash,
        "share_root": meta.share_root,
        "pedersen_root": meta.pedersen_root,
        "data_shards": data_shards,
        "parity_shards": parity_shards,
    });
    Ok(response.to_string())
}

fn handle_rollup_settle(req: &HttpRequest, cfg: &BlobServiceConfig) -> Result<String, String> {
    #[derive(Deserialize)]
    struct RollupSettleRequest {
        namespace: String,
        share_root: String,
        #[serde(default)]
        pedersen_root: Option<String>,
        payer_pk: String,
        #[serde(default)]
        operator_pk: Option<String>,
        #[serde(default)]
        attesters: Option<Vec<String>>,
        fee: u64,
        #[serde(default)]
        mode: Option<String>,
        #[serde(default)]
        proof_b64: Option<String>,
        #[serde(default)]
        public_inputs_b64: Option<String>,
        #[serde(default)]
        merkle_path_b64: Option<String>,
    }
    let req_body: RollupSettleRequest =
        serde_json::from_slice(&req.body).map_err(|e| format!("decode error: {e}"))?;
    let registry_path = cfg
        .stake_registry_path
        .as_ref()
        .ok_or_else(|| "stake registry not configured".to_string())?;
    let pedersen_root = req_body
        .pedersen_root
        .clone()
        .unwrap_or_else(|| req_body.share_root.clone());
    let commitment = RollupCommitment {
        namespace: req_body.namespace.clone(),
        share_root: req_body.share_root.clone(),
        pedersen_root: Some(pedersen_root),
        settlement_slot: None,
    };
    let operator_pk = req_body
        .operator_pk
        .clone()
        .unwrap_or_else(|| req_body.payer_pk.clone());
    let attesters = req_body.attesters.unwrap_or_default();
    let mode = req_body.mode.unwrap_or_else(|| "optimistic".to_string());

    let zk_proof = if let (Some(p_b64), Some(pi_b64), Some(mp_b64)) = (
        req_body.proof_b64.as_ref(),
        req_body.public_inputs_b64.as_ref(),
        req_body.merkle_path_b64.as_ref(),
    ) {
        let proof = BASE64
            .decode(p_b64.as_bytes())
            .map_err(|e| format!("proof decode: {e}"))?;
        let public_inputs = BASE64
            .decode(pi_b64.as_bytes())
            .map_err(|e| format!("public inputs decode: {e}"))?;
        let merkle_path = BASE64
            .decode(mp_b64.as_bytes())
            .map_err(|e| format!("merkle path decode: {e}"))?;
        ZkRollupProof {
            proof,
            public_inputs,
            merkle_path,
        }
    } else {
        ZkRollupProof {
            proof: Vec::new(),
            public_inputs: Vec::new(),
            merkle_path: Vec::new(),
        }
    };

    let mode_enum = if mode == "zk" {
        RollupSettlementMode::Zk(zk_proof)
    } else {
        RollupSettlementMode::Optimistic(Vec::new())
    };

    match settle_rollup_with_rewards(
        registry_path,
        commitment.clone(),
        &req_body.payer_pk,
        &operator_pk,
        &attesters,
        req_body.fee,
        mode_enum,
    ) {
        Ok(receipt) => Ok(serde_json::to_string(&serde_json::json!({
            "status": "ok",
            "payer": receipt.payer,
            "fee": receipt.fee,
            "commitment": receipt.commitment.share_root
        }))
        .unwrap_or_else(|_| "{}".to_string())),
        Err(fault) => {
            let outbox = cfg.base_dir.join("evidence_outbox.jsonl");
            append_rollup_fault_evidence(&outbox, &fault);
            Err(format!("rollup fault: {}", fault.reason))
        }
    }
}

fn handle_commitment(req: &HttpRequest, cfg: &BlobServiceConfig) -> Result<String, String> {
    let parts: Vec<&str> = req.path.split('/').collect();
    if parts.len() < 4 {
        return Err("Not Found".into());
    }
    let namespace = sanitize_token(parts[2]).ok_or_else(|| "Not Found".to_string())?;
    let hash = sanitize_token(parts[3]).ok_or_else(|| "Not Found".to_string())?;
    let meta = load_blob_meta(&cfg.base_dir, &namespace, &hash)
        .map_err(|e| e.to_string())?
        .ok_or_else(|| "Not Found".to_string())?;
    let attestations: Vec<_> = meta
        .attestations
        .iter()
        .map(|a| serde_json::json!({"pk": a.pk, "sig": a.sig}))
        .collect();
    let body = serde_json::json!({
        "namespace": meta.namespace,
        "hash": meta.hash,
        "size": meta.size,
        "share_root": meta.share_root,
        "pedersen_root": meta.pedersen_root,
        "data_shards": meta.data_shards,
        "parity_shards": meta.parity_shards,
        "shares": (meta.data_shards as usize + meta.parity_shards as usize),
        "attestations": attestations,
        "publisher_pk": meta.publisher_pk,
    });
    Ok(body.to_string())
}

fn parse_count(query: Option<&str>) -> usize {
    query
        .and_then(|q| {
            q.split('&').find_map(|kv| {
                let mut parts = kv.split('=');
                if parts.next() == Some("count") {
                    parts.next().and_then(|v| v.parse().ok())
                } else {
                    None
                }
            })
        })
        .unwrap_or(1)
}

fn pick_slash_target(meta: &BlobMeta) -> Option<String> {
    meta.publisher_pk
        .clone()
        .or_else(|| meta.attestations.first().map(|a| a.pk.clone()))
}

fn handle_sample(
    req: &HttpRequest,
    cfg: &BlobServiceConfig,
) -> Result<(u16, String), (u16, String)> {
    let parts: Vec<&str> = req.path.split('?').collect();
    let path_part = parts[0];
    let query_part = if parts.len() > 1 {
        Some(parts[1])
    } else {
        None
    };
    let segs: Vec<&str> = path_part.split('/').collect();
    if segs.len() < 4 {
        return Err((404, "Not Found".into()));
    }
    let namespace = sanitize_token(segs[2]).ok_or((404, "Not Found".into()))?;
    let hash = sanitize_token(segs[3]).ok_or((404, "Not Found".into()))?;
    let meta = load_blob_meta(&cfg.base_dir, &namespace, &hash)
        .map_err(|e| (400, e.to_string()))?
        .ok_or((404, "Not Found".into()))?;
    let total = meta.data_shards as usize + meta.parity_shards as usize;
    let shares = match read_shares(&cfg.base_dir, &namespace, &hash, total) {
        Ok(s) => s,
        Err(_err) => {
            if let Some(pk) = meta.publisher_pk.as_ref() {
                record_slash_with_registry(
                    &cfg.membership_policy,
                    &cfg.stake_registry_path,
                    pk,
                    "blob-missing",
                );
            }
            let ev_path = cfg.base_dir.join("evidence_outbox.jsonl");
            let ev = availability::build_missing_share_evidence(&namespace, &hash, 0);
            append_availability_evidence(&ev_path, &ev);
            return Err((404, "Not Found".into()));
        }
    };
    let share_hashes = share_hashes(&shares);
    let root_hex = hex::encode(crate::merkle_root(&share_hashes));
    if root_hex != meta.share_root {
        let ev_path = cfg.base_dir.join("evidence_outbox.jsonl");
        let ev = AvailabilityEvidence {
            namespace: namespace.clone(),
            blob_hash: hash.clone(),
            idx: 0,
            share: None,
            reason: "share_root_mismatch".into(),
        };
        append_availability_evidence(&ev_path, &ev);
        return Err((400, "share_root mismatch".into()));
    }
    let pedersen_root = availability::pedersen_merkle_root(&share_hashes);
    let pedersen_root_hex = hex::encode(pedersen_root);
    if let Some(stored) = &meta.pedersen_root {
        if &pedersen_root_hex != stored {
            let ev_path = cfg.base_dir.join("evidence_outbox.jsonl");
            let ev = AvailabilityEvidence {
                namespace: namespace.clone(),
                blob_hash: hash.clone(),
                idx: 0,
                share: None,
                reason: "pedersen_root_mismatch".into(),
            };
            append_availability_evidence(&ev_path, &ev);
            return Err((400, "pedersen_root mismatch".into()));
        }
    }
    let mut indices: Vec<usize> = (0..shares.len()).collect();
    let count = parse_count(query_part);
    indices.truncate(count.min(indices.len()));
    let mut sampled = Vec::new();
    for idx in indices {
        let proof = build_merkle_proof(&share_hashes, idx).ok_or((400, "bad index".into()))?;
        let pedersen_proof =
            availability::pedersen_share_proof(&share_hashes, idx).map_err(|e| (400, e))?;
        sampled.push(serde_json::json!({
            "idx": idx,
            "data": BASE64.encode(&shares[idx]),
            "leaf": hex::encode(share_hashes[idx]),
            "proof": proof.path.iter().map(|n| serde_json::json!({"left": n.left, "sibling": hex::encode(n.sibling)})).collect::<Vec<_>>(),
            "pedersen_proof": pedersen_proof.path.iter().map(|n| serde_json::json!({"left": n.left, "hash": hex::encode(n.sibling)})).collect::<Vec<_>>(),
        }));
    }
    let attestations: Vec<_> = meta
        .attestations
        .iter()
        .map(|a| serde_json::json!({"pk": a.pk, "sig": a.sig}))
        .collect();
    let body = serde_json::json!({
        "namespace": meta.namespace,
        "hash": meta.hash,
        "size": meta.size,
        "share_root": meta.share_root,
        "pedersen_root": meta.pedersen_root,
        "data_shards": meta.data_shards,
        "parity_shards": meta.parity_shards,
        "shares": sampled,
        "attestations": attestations,
    });
    Ok((200, body.to_string()))
}

fn handle_prove_storage(
    req: &HttpRequest,
    cfg: &BlobServiceConfig,
) -> Result<(u16, String), (u16, String)> {
    let evidence_log = cfg.base_dir.join("evidence.jsonl");
    let evidence_outbox = cfg.base_dir.join("evidence_outbox.jsonl");
    let segs: Vec<&str> = req.path.split('/').collect();
    if segs.len() < 5 {
        return Err((404, "Not Found".into()));
    }
    let namespace = sanitize_token(segs[2]).ok_or((404, "Not Found".into()))?;
    let hash = sanitize_token(segs[3]).ok_or((404, "Not Found".into()))?;
    let idx: usize = segs[4].parse().map_err(|_| (400, "invalid index".into()))?;
    let meta = load_blob_meta(&cfg.base_dir, &namespace, &hash)
        .map_err(|e| (400, e.to_string()))?
        .ok_or((404, "Not Found".into()))?;
    let total = meta.data_shards as usize + meta.parity_shards as usize;
    let shares = match read_shares(&cfg.base_dir, &namespace, &hash, total) {
        Ok(s) => s,
        Err(_err) => {
            if let Some(pk) = pick_slash_target(&meta) {
                record_slash_with_registry(
                    &cfg.membership_policy,
                    &cfg.stake_registry_path,
                    &pk,
                    "blob-missing",
                );
                append_evidence(
                    &evidence_log,
                    &meta.namespace,
                    &meta.hash,
                    &pk,
                    "blob-missing",
                );
                let ev = availability::build_missing_share_evidence(&namespace, &hash, idx);
                append_availability_evidence(&evidence_outbox, &ev);
            }
            return Err((404, "Not Found".into()));
        }
    };
    if idx >= shares.len() {
        return Err((404, "Not Found".into()));
    }
    let share_hashes = share_hashes(&shares);
    let proof = build_merkle_proof(&share_hashes, idx).ok_or((400, "bad index".into()))?;
    let body = serde_json::json!({
        "namespace": meta.namespace,
        "hash": meta.hash,
        "idx": idx,
        "share": BASE64.encode(&shares[idx]),
        "share_root": meta.share_root,
        "leaf": hex::encode(share_hashes[idx]),
        "proof": proof.path.iter().map(|n| serde_json::json!({"left": n.left, "sibling": hex::encode(n.sibling)})).collect::<Vec<_>>(),
    });
    Ok((200, body.to_string()))
}

struct PayloadCache {
    seen: HashSet<[u8; 32]>,
    order: VecDeque<[u8; 32]>,
    metrics: Arc<Metrics>,
}

impl PayloadCache {
    fn new(metrics: Arc<Metrics>) -> Self {
        Self {
            seen: HashSet::new(),
            order: VecDeque::new(),
            metrics,
        }
    }

    fn insert(&mut self, digest: [u8; 32]) -> bool {
        if self.seen.contains(&digest) {
            return false;
        }
        self.seen.insert(digest);
        self.order.push_back(digest);
        if self.order.len() > SEEN_CACHE_LIMIT {
            if let Some(old) = self.order.pop_front() {
                self.seen.remove(&old);
                self.metrics.inc_lrucache_evictions();
            }
        }
        true
    }
}

#[derive(Default)]
struct Metrics {
    anchors_received_total: AtomicU64,
    anchors_verified_total: AtomicU64,
    invalid_envelopes_total: AtomicU64,
    lrucache_evictions_total: AtomicU64,
    finality_events_total: AtomicU64,
    gossipsub_rejects_total: AtomicU64,
}

impl Metrics {
    fn inc_anchors_received(&self) {
        self.anchors_received_total.fetch_add(1, Ordering::Relaxed);
    }

    fn inc_anchors_verified(&self) {
        self.anchors_verified_total.fetch_add(1, Ordering::Relaxed);
    }

    fn inc_invalid_envelopes(&self) {
        self.invalid_envelopes_total.fetch_add(1, Ordering::Relaxed);
    }

    fn inc_lrucache_evictions(&self) {
        self.lrucache_evictions_total
            .fetch_add(1, Ordering::Relaxed);
    }

    fn inc_finality_events(&self) {
        self.finality_events_total.fetch_add(1, Ordering::Relaxed);
    }

    fn inc_gossipsub_rejects(&self) {
        self.gossipsub_rejects_total.fetch_add(1, Ordering::Relaxed);
    }

    fn render(&self) -> String {
        format!(
            "# TYPE anchors_received_total counter\nanchors_received_total {}\n\
# TYPE anchors_verified_total counter\nanchors_verified_total {}\n\
# TYPE invalid_envelopes_total counter\ninvalid_envelopes_total {}\n\
# TYPE lrucache_evictions_total counter\nlrucache_evictions_total {}\n\
# TYPE finality_events_total counter\nfinality_events_total {}\n\
# TYPE gossipsub_rejects_total counter\ngossipsub_rejects_total {}\n",
            self.anchors_received_total.load(Ordering::Relaxed),
            self.anchors_verified_total.load(Ordering::Relaxed),
            self.invalid_envelopes_total.load(Ordering::Relaxed),
            self.lrucache_evictions_total.load(Ordering::Relaxed),
            self.finality_events_total.load(Ordering::Relaxed),
            self.gossipsub_rejects_total.load(Ordering::Relaxed),
        )
    }
}

/// Errors surfaced by the networking runtime.
#[derive(Debug)]
pub enum NetworkError {
    /// Ledger anchor could not be produced or verified.
    Anchor(String),
    /// JSON or base64 codec failure.
    Codec(String),
    /// Filesystem interaction failure.
    Io(String),
    /// Key derivation or signature failure.
    Key(String),
    /// Underlying libp2p API returned an error.
    Libp2p(String),
    /// Evidence sender not permitted.
    Policy(String),
}

impl std::fmt::Display for NetworkError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Anchor(msg) => write!(f, "anchor error: {msg}"),
            Self::Codec(msg) => write!(f, "codec error: {msg}"),
            Self::Io(msg) => write!(f, "I/O error: {msg}"),
            Self::Key(msg) => write!(f, "key error: {msg}"),
            Self::Libp2p(msg) => write!(f, "libp2p error: {msg}"),
            Self::Policy(msg) => write!(f, "policy error: {msg}"),
        }
    }
}

impl std::error::Error for NetworkError {}

impl From<AnchorCodecError> for NetworkError {
    fn from(err: AnchorCodecError) -> Self {
        Self::Codec(err.to_string())
    }
}

impl From<KeyError> for NetworkError {
    fn from(err: KeyError) -> Self {
        Self::Key(err.to_string())
    }
}

impl From<io::Error> for NetworkError {
    fn from(err: io::Error) -> Self {
        Self::Io(err.to_string())
    }
}

#[derive(NetworkBehaviour)]
struct JrocBehaviour {
    pub gossipsub: gossipsub::Behaviour,
    pub identify: identify::Behaviour,
    pub kademlia: kad::Behaviour<MemoryStore>,
}

/// Starts the JULIAN Protocol libp2p node and continues until Ctrl+C.
///
/// The swarm listens on `cfg.listen_addr`, optionally dials bootstrap peers,
/// recomputes anchors at `cfg.broadcast_interval`, and emits gossip messages
/// for every new anchor. Incoming envelopes are verified and reconciled with
/// the local ledger according to `cfg.quorum`.
pub async fn run_network(cfg: NetConfig) -> Result<(), NetworkError> {
    let local_key_bytes = cfg.key_material.verifying.to_bytes();
    if !policy_permits(cfg.membership_policy.as_ref(), &local_key_bytes) {
        return Err(NetworkError::Key(
            "local key not permitted by identity policy".to_string(),
        ));
    }
    fs::create_dir_all(&cfg.log_dir).map_err(|err| {
        NetworkError::Io(format!(
            "failed to create log dir {}: {err}",
            cfg.log_dir.display()
        ))
    })?;
    if let Some(blob_dir) = cfg.blob_dir.as_ref() {
        fs::create_dir_all(blob_dir).map_err(|err| {
            NetworkError::Io(format!(
                "failed to create blob dir {}: {err}",
                blob_dir.display()
            ))
        })?;
    }
    let mut swarm = build_swarm(&cfg)?;
    Swarm::listen_on(&mut swarm, cfg.listen_addr.clone())
        .map_err(|err| NetworkError::Libp2p(format!("{err:?}")))?;
    for addr in &cfg.bootstraps {
        if let Err(err) = Swarm::dial(&mut swarm, addr.clone()) {
            eprintln!("dial {addr} failed: {err}");
        }
    }

    let mut ticker = time::interval(cfg.broadcast_interval);
    ticker.set_missed_tick_behavior(time::MissedTickBehavior::Delay);

    let metrics = cfg.metrics.clone();
    if let Some(addr) = cfg.metrics_addr {
        let metrics_clone = metrics.clone();
        tokio::spawn(async move {
            if let Err(err) = run_metrics_server(addr, metrics_clone).await {
                eprintln!("metrics server error: {err}");
            }
        });
        println!("QSYS|mod=METRICS|evt=LISTEN|addr={addr}");
    }

    if let (Some(blob_dir), Some(blob_listen)) = (cfg.blob_dir.clone(), cfg.blob_listen) {
        let blob_cfg = BlobServiceConfig {
            base_dir: blob_dir,
            listen: blob_listen,
            max_bytes: cfg.max_blob_bytes,
            retention_days: cfg.blob_retention_days,
            policies: cfg.blob_policies.clone(),
            signing: cfg.key_material.signing.clone(),
            verifying_b64: encode_public_key_base64(&cfg.key_material.verifying),
            stake_registry_path: cfg.stake_registry_path.clone(),
            membership_policy: cfg.membership_policy.clone(),
            auth_token: cfg.blob_auth_token.clone(),
            max_concurrency: cfg.blob_max_concurrency,
            request_timeout: cfg.blob_request_timeout,
            rate_limits: Arc::new(Mutex::new(HashMap::new())),
        };
        tokio::spawn(run_blob_service(blob_cfg));
    }

    let mut seen_payloads = PayloadCache::new(metrics.clone());
    let mut invalid_counters: HashMap<libp2p::PeerId, usize> = HashMap::new();
    let mut last_payload = Vec::new();
    let mut last_publish: Option<Instant> = None;
    let mut broadcast_counter: u64 = 0;

    let local_peer = cfg.key_material.libp2p.public().to_peer_id();

    println!(
        "QSYS|mod=NET|evt=LISTEN|node={} peer={} addr={}",
        cfg.node_id, local_peer, cfg.listen_addr
    );

    loop {
        select! {
            _ = ticker.tick() => {
                if let Err(err) = broadcast_local_anchor(
                    &mut swarm,
                    &cfg,
                    &mut last_payload,
                    &mut last_publish,
                    &mut broadcast_counter,
                    &metrics,
                )
                .await
                {
                    metrics.inc_gossipsub_rejects();
                    eprintln!("broadcast error: {err}");
                }
                if let Err(err) = broadcast_evidence(&mut swarm, &cfg) {
                    eprintln!("evidence broadcast error: {err}");
                }
            }
            event = swarm.select_next_some() => {
                if let Err(err) = handle_event(event, &cfg, &mut seen_payloads, &mut invalid_counters, &metrics).await {
                    eprintln!("network error: {err}");
                }
            }
            _ = signal::ctrl_c() => {
                println!("JROC-NET node shutting down");
                return Ok(());
            }
        }
    }
}

fn build_swarm(cfg: &NetConfig) -> Result<Swarm<JrocBehaviour>, NetworkError> {
    let identity = cfg.key_material.libp2p.clone();

    let builder = SwarmBuilder::with_existing_identity(identity)
        .with_tokio()
        .with_tcp(
            tcp::Config::default(),
            noise::Config::new,
            yamux::Config::default,
        )
        .map_err(|err| NetworkError::Libp2p(format!("{err:?}")))?
        .with_behaviour(|key| {
            build_behaviour(key).map_err(|err| {
                let boxed: Box<dyn std::error::Error + Send + Sync> = Box::new(err);
                boxed
            })
        })
        .map_err(|err| NetworkError::Libp2p(format!("{err:?}")))?
        .with_swarm_config(|cfg| cfg.with_idle_connection_timeout(Duration::from_secs(60)));

    Ok(builder.build())
}

fn build_behaviour(key: &identity::Keypair) -> Result<JrocBehaviour, NetworkError> {
    let peer_id = key.public().to_peer_id();

    let gossipsub_config = gossipsub::ConfigBuilder::default()
        .validation_mode(ValidationMode::Strict)
        .message_id_fn(|message: &gossipsub::Message| {
            let mut hasher = Sha256::new();
            hasher.update(&message.data);
            gossipsub::MessageId::from(hasher.finalize().to_vec())
        })
        .build()
        .map_err(|err| NetworkError::Libp2p(format!("{err:?}")))?;

    let mut gossipsub =
        gossipsub::Behaviour::new(MessageAuthenticity::Signed(key.clone()), gossipsub_config)
            .map_err(|err| NetworkError::Libp2p(format!("{err:?}")))?;
    gossipsub
        .subscribe(&TOPIC_ANCHORS)
        .map_err(|err| NetworkError::Libp2p(format!("{err:?}")))?;

    let identify_config = identify::Config::new("jrocnet/1.0.0".into(), key.public())
        .with_push_listen_addr_updates(true);
    let identify = identify::Behaviour::new(identify_config);

    let store = MemoryStore::new(peer_id);
    let kademlia = kad::Behaviour::with_config(peer_id, store, kad::Config::default());

    Ok(JrocBehaviour {
        gossipsub,
        identify,
        kademlia,
    })
}

fn evidence_outbox(cfg: &NetConfig) -> Option<PathBuf> {
    cfg.blob_dir
        .as_ref()
        .map(|d| d.join("evidence_outbox.jsonl"))
}

fn broadcast_evidence(
    swarm: &mut Swarm<JrocBehaviour>,
    cfg: &NetConfig,
) -> Result<(), NetworkError> {
    let Some(path) = evidence_outbox(cfg) else {
        return Ok(());
    };
    let contents = match fs::read_to_string(&path) {
        Ok(c) => c,
        Err(err) => {
            if err.kind() == io::ErrorKind::NotFound {
                return Ok(());
            }
            return Err(NetworkError::Io(err.to_string()));
        }
    };
    if contents.trim().is_empty() {
        return Ok(());
    }
    let lines: Vec<&str> = contents.lines().filter(|l| !l.trim().is_empty()).collect();
    if lines.is_empty() {
        return Ok(());
    }
    fs::write(&path, "").map_err(|e| NetworkError::Io(e.to_string()))?;
    for line in lines {
        if let Ok(record) = serde_json::from_str::<EvidenceRecord>(line) {
            let env = sign_evidence(&cfg.key_material.signing, &record);
            if let Ok(msg) = serde_json::to_vec(&env) {
                let _ = swarm
                    .behaviour_mut()
                    .gossipsub
                    .publish(TOPIC_EVIDENCE.clone(), msg);
            }
        } else if let Ok(av) = serde_json::from_str::<AvailabilityEvidence>(line) {
            let msg = serde_json::to_vec(&av).map_err(|e| NetworkError::Codec(e.to_string()))?;
            let _ = swarm
                .behaviour_mut()
                .gossipsub
                .publish(TOPIC_EVIDENCE.clone(), msg);
        } else if let Ok(rf) = serde_json::from_str::<RollupFaultEvidence>(line) {
            let msg = serde_json::to_vec(&rf).map_err(|e| NetworkError::Codec(e.to_string()))?;
            let _ = swarm
                .behaviour_mut()
                .gossipsub
                .publish(TOPIC_EVIDENCE.clone(), msg);
        }
    }
    Ok(())
}

fn handle_evidence_message(cfg: &NetConfig, data: &[u8]) -> Result<(), NetworkError> {
    // Try to parse signed evidence envelope first.
    if let Ok(env) = serde_json::from_slice::<EvidenceEnvelope>(data) {
        let payload =
            serde_json::to_vec(&env.evidence).map_err(|e| NetworkError::Codec(e.to_string()))?;
        verify_signature_base64(&env.public_key, &payload, &env.signature)?;
        // Basic validation: ensure peer is permitted by policy.
        let vk = decode_public_key_base64(&env.public_key)
            .map_err(|e| NetworkError::Codec(e.to_string()))?;
        if !policy_permits(cfg.membership_policy.as_ref(), &vk.to_bytes()) {
            return Err(NetworkError::Policy("evidence sender not permitted".into()));
        }
        if let Some(path) = cfg.blob_dir.as_ref().map(|d| d.join("evidence.jsonl")) {
            append_evidence(
                &path,
                &env.evidence.namespace,
                &env.evidence.blob_hash,
                &env.evidence.pk,
                &env.evidence.reason,
            );
        }
        record_slash_with_registry(
            &cfg.membership_policy,
            &cfg.stake_registry_path,
            &env.evidence.pk,
            &env.evidence.reason,
        );
        return Ok(());
    }
    // Fallback: raw availability or rollup evidence.
    if let Ok(av) = serde_json::from_slice::<AvailabilityEvidence>(data) {
        if let Some(path) = cfg.blob_dir.as_ref().map(|d| d.join("evidence.jsonl")) {
            append_availability_evidence(&path, &av);
        }
        if let Some(pk) = pick_slash_target(&BlobMeta {
            namespace: av.namespace.clone(),
            hash: av.blob_hash.clone(),
            size: 0,
            data_shards: 0,
            parity_shards: 0,
            share_root: String::new(),
            pedersen_root: None,
            attestations: Vec::new(),
            publisher_pk: None,
        }) {
            record_slash_with_registry(
                &cfg.membership_policy,
                &cfg.stake_registry_path,
                &pk,
                "availability-fault",
            );
        }
        return Ok(());
    }
    if let Ok(rf) = serde_json::from_slice::<RollupFaultEvidence>(data) {
        if let Some(path) = cfg.blob_dir.as_ref().map(|d| d.join("evidence.jsonl")) {
            append_availability_evidence(
                &path,
                &AvailabilityEvidence {
                    namespace: rf.namespace.clone(),
                    blob_hash: rf.commitment.clone(),
                    idx: 0,
                    share: rf.payload.clone(),
                    reason: rf.reason.clone(),
                },
            );
        }
        if let Some(pk) = pick_slash_target(&BlobMeta {
            namespace: rf.namespace.clone(),
            hash: rf.commitment.clone(),
            size: 0,
            data_shards: 0,
            parity_shards: 0,
            share_root: String::new(),
            pedersen_root: None,
            attestations: Vec::new(),
            publisher_pk: None,
        }) {
            record_slash_with_registry(
                &cfg.membership_policy,
                &cfg.stake_registry_path,
                &pk,
                "rollup-fault",
            );
        }
        return Ok(());
    }
    Ok(())
}

async fn broadcast_local_anchor(
    swarm: &mut Swarm<JrocBehaviour>,
    cfg: &NetConfig,
    last_payload: &mut Vec<u8>,
    last_publish: &mut Option<Instant>,
    broadcast_counter: &mut u64,
    metrics: &Arc<Metrics>,
) -> Result<(), NetworkError> {
    if !policy_permits(
        cfg.membership_policy.as_ref(),
        &cfg.key_material.verifying.to_bytes(),
    ) {
        return Err(NetworkError::Key(
            "local key not permitted by identity policy".to_string(),
        ));
    }
    let ledger = load_anchor_from_logs(&cfg.log_dir)?;
    let timestamp_ms = now_millis();
    let anchor_json = AnchorJson::from_ledger(
        cfg.node_id.clone(),
        cfg.quorum,
        &ledger,
        timestamp_ms,
        latest_da_commitments(&cfg.blob_dir),
        evidence_root(&cfg.blob_dir),
    )?;
    let payload =
        serde_json::to_vec(&anchor_json).map_err(|err| NetworkError::Codec(err.to_string()))?;
    if *last_payload == payload {
        return Ok(());
    }
    if let Some(prev) = last_publish {
        if prev.elapsed() < cfg.broadcast_interval {
            return Ok(());
        }
    }
    let signature = sign_payload(&cfg.key_material.signing, &payload);
    let signature_b64 = encode_signature_base64(&signature);
    let envelope = AnchorEnvelope {
        schema: SCHEMA_ENVELOPE.to_string(),
        schema_version: ENVELOPE_SCHEMA_VERSION,
        public_key: encode_public_key_base64(&cfg.key_material.verifying),
        node_id: cfg.node_id.clone(),
        payload: BASE64.encode(&payload),
        signature: signature_b64.clone(),
    };
    let message =
        serde_json::to_vec(&envelope).map_err(|err| NetworkError::Codec(err.to_string()))?;
    match swarm
        .behaviour_mut()
        .gossipsub
        .publish(TOPIC_ANCHORS.clone(), message)
    {
        Ok(_) => {
            NO_GOSSIP_PEERS_LOGGED.store(false, Ordering::Relaxed);
        }
        Err(PublishError::InsufficientPeers) => {
            if !NO_GOSSIP_PEERS_LOGGED.swap(true, Ordering::Relaxed) {
                println!("QSYS|mod=ANCHOR|evt=STANDBY|reason=awaiting_peers");
            }
            return Ok(());
        }
        Err(err) => {
            metrics.inc_gossipsub_rejects();
            return Err(NetworkError::Libp2p(err.to_string()));
        }
    }
    *last_payload = payload;
    *last_publish = Some(Instant::now());
    println!(
        "QSYS|mod=ANCHOR|evt=BROADCAST|entries={}",
        ledger.entries.len()
    );
    if let Some(interval) = cfg.checkpoint_interval {
        if interval > 0 {
            *broadcast_counter = broadcast_counter.saturating_add(1);
            if *broadcast_counter % interval == 0 {
                let checkpoint = AnchorCheckpoint::new(
                    *broadcast_counter,
                    anchor_json.clone(),
                    vec![CheckpointSignature {
                        node_id: cfg.node_id.clone(),
                        public_key: encode_public_key_base64(&cfg.key_material.verifying),
                        signature: signature_b64,
                    }],
                    latest_log_cutoff(&cfg.log_dir),
                );
                if let Err(err) = write_checkpoint(&cfg.log_dir.join("checkpoints"), &checkpoint) {
                    eprintln!("checkpoint write failed: {err}");
                } else {
                    println!(
                        "QSYS|mod=CHECKPOINT|evt=RECORDED|epoch={} entries={}",
                        checkpoint.epoch,
                        ledger.entries.len()
                    );
                }
            }
        }
    }
    Ok(())
}

async fn handle_event(
    event: SwarmEvent<JrocBehaviourEvent>,
    cfg: &NetConfig,
    seen_payloads: &mut PayloadCache,
    invalid_counters: &mut HashMap<libp2p::PeerId, usize>,
    metrics: &Arc<Metrics>,
) -> Result<(), NetworkError> {
    match event {
        SwarmEvent::NewListenAddr { address, .. } => {
            println!("QSYS|mod=NET|evt=LISTEN|addr={address}");
        }
        SwarmEvent::Behaviour(JrocBehaviourEvent::Gossipsub(event)) => match event {
            gossipsub::Event::Message {
                propagation_source,
                message,
                ..
            } => {
                if message.topic == TOPIC_EVIDENCE.hash() {
                    handle_evidence_message(cfg, &message.data)?;
                    return Ok(());
                }
                metrics.inc_anchors_received();
                if message.data.len() > MAX_ENVELOPE_BYTES {
                    metrics.inc_gossipsub_rejects();
                    record_invalid(invalid_counters, propagation_source, metrics);
                    return Ok(());
                }
                let digest = sha256_digest(&message.data);
                if !seen_payloads.insert(digest) {
                    metrics.inc_gossipsub_rejects();
                    return Ok(());
                }
                let envelope: AnchorEnvelope = serde_json::from_slice(&message.data)
                    .map_err(|err| NetworkError::Codec(err.to_string()))?;
                envelope.validate()?;
                let payload = BASE64
                    .decode(envelope.payload.as_bytes())
                    .map_err(|err| NetworkError::Codec(err.to_string()))?;
                if payload.len() > MAX_ENVELOPE_BYTES {
                    metrics.inc_gossipsub_rejects();
                    record_invalid(invalid_counters, propagation_source, metrics);
                    return Ok(());
                }
                verify_signature_base64(&envelope.public_key, &payload, &envelope.signature)?;
                let remote_verifying = decode_public_key_base64(&envelope.public_key)
                    .map_err(|err| NetworkError::Codec(err.to_string()))?;
                let remote_key_bytes = remote_verifying.to_bytes();
                if !policy_permits(cfg.membership_policy.as_ref(), &remote_key_bytes) {
                    metrics.inc_gossipsub_rejects();
                    record_invalid(invalid_counters, propagation_source, metrics);
                    println!(
                        "rejecting peer {}: identity not permitted by policy",
                        envelope.node_id
                    );
                    return Ok(());
                }
                let payload_str = std::str::from_utf8(&payload)
                    .map_err(|err| NetworkError::Codec(err.to_string()))?;
                let anchor_json = AnchorJson::from_json_str(payload_str)
                    .map_err(|err| NetworkError::Codec(err.to_string()))?;
                if anchor_json.network != NETWORK_ID {
                    metrics.inc_gossipsub_rejects();
                    record_invalid(invalid_counters, propagation_source, metrics);
                    return Ok(());
                }
                if anchor_json.entries.len() > MAX_ANCHOR_ENTRIES {
                    metrics.inc_gossipsub_rejects();
                    record_invalid(invalid_counters, propagation_source, metrics);
                    return Ok(());
                }
                // DA gating: require commitments and verify share roots + attestation QC; require persisted QC.
                if anchor_json.da_commitments.is_empty() {
                    metrics.inc_gossipsub_rejects();
                    println!(
                        "rejecting peer {}: missing DA commitments in anchor",
                        envelope.node_id
                    );
                    return Ok(());
                }
                if let Some(blob_dir) = cfg.blob_dir.as_ref() {
                    for da in &anchor_json.da_commitments {
                        let meta = load_blob_meta(blob_dir, &da.namespace, &da.blob_hash)
                            .ok()
                            .flatten();
                        let Some(meta) = meta else {
                            metrics.inc_gossipsub_rejects();
                            println!(
                                "rejecting peer {}: missing blob {} in {}",
                                envelope.node_id, da.blob_hash, da.namespace
                            );
                            return Ok(());
                        };
                        if meta.share_root != da.share_root {
                            metrics.inc_gossipsub_rejects();
                            println!(
                                "rejecting peer {}: share_root mismatch for {}",
                                envelope.node_id, da.blob_hash
                            );
                            return Ok(());
                        }
                        if let (Some(meta_p), Some(da_p)) = (&meta.pedersen_root, &da.pedersen_root)
                        {
                            if meta_p != da_p {
                                metrics.inc_gossipsub_rejects();
                                println!(
                                    "rejecting peer {}: pedersen_root mismatch for {}",
                                    envelope.node_id, da.blob_hash
                                );
                                return Ok(());
                            }
                        }
                        if da.pedersen_root.is_none() {
                            metrics.inc_gossipsub_rejects();
                            println!(
                                "rejecting peer {}: pedersen_root missing for {}",
                                envelope.node_id, da.blob_hash
                            );
                            return Ok(());
                        }
                        let attestations: Vec<_> = meta
                            .attestations
                            .iter()
                            .map(|a| a.to_attestation(&meta.share_root, &meta.pedersen_root))
                            .collect();
                        let qc =
                            aggregate_attestations(&attestations, cfg.attestation_quorum, |pk| {
                                lookup_stake(cfg, pk)
                            })
                            .map_err(|e| NetworkError::Codec(e.to_string()))?;
                        if !qc.quorum_reached {
                            metrics.inc_gossipsub_rejects();
                            println!(
                                "rejecting peer {}: DA quorum not met for {}",
                                envelope.node_id, da.blob_hash
                            );
                            return Ok(());
                        }
                        // Persist QC for evidence and rollup settlement.
                        if let Some(log_dir) = cfg.blob_dir.as_ref() {
                            let qc_path = log_dir
                                .join(&da.namespace)
                                .join(format!("{}.qc", da.blob_hash));
                            if let Some(parent) = qc_path.parent() {
                                let _ = fs::create_dir_all(parent);
                            }
                            if let Ok(bytes) = serde_json::to_vec(&qc) {
                                let _ = fs::write(&qc_path, bytes);
                            }
                            // Reward attesters (best-effort).
                            if let Some(path) = &cfg.stake_registry_path {
                                if let Ok(mut reg) = StakeRegistry::load(path) {
                                    for signer in &qc.signers {
                                        reg.credit_reward(signer, 1);
                                    }
                                    let _ = reg.save(path);
                                }
                            }
                        }
                        // Require QC file to exist (stake-weighted gating).
                        if let Some(log_dir) = cfg.blob_dir.as_ref() {
                            let qc_path = log_dir
                                .join(&da.namespace)
                                .join(format!("{}.qc", da.blob_hash));
                            if !qc_path.exists() {
                                metrics.inc_gossipsub_rejects();
                                println!(
                                    "rejecting peer {}: missing QC for {}",
                                    envelope.node_id, da.blob_hash
                                );
                                return Ok(());
                            }
                        }
                    }
                }
                let remote_anchor = anchor_json.clone().into_ledger()?;
                let local_anchor = load_anchor_from_logs(&cfg.log_dir)?;
                let local_key_bytes = cfg.key_material.verifying.to_bytes();
                let votes = [
                    AnchorVote {
                        anchor: &local_anchor,
                        public_key: &local_key_bytes,
                    },
                    AnchorVote {
                        anchor: &remote_anchor,
                        public_key: &remote_key_bytes,
                    },
                ];
                match crate::reconcile_anchors_with_quorum(&votes, cfg.quorum) {
                    Ok(()) => {
                        metrics.inc_anchors_verified();
                        metrics.inc_finality_events();
                        println!(
                            "QSYS|mod=QUORUM|evt=FINALIZED|peer={} entries={}",
                            envelope.node_id,
                            remote_anchor.entries.len()
                        );
                    }
                    Err(err) => {
                        println!("anchor divergence with peer {}: {}", envelope.node_id, err);
                        if let Err(slash_err) =
                            cfg.membership_policy.record_slash(&remote_verifying)
                        {
                            eprintln!(
                                "failed to record slash for {}: {}",
                                envelope.node_id, slash_err
                            );
                        } else {
                            println!(
                                "peer {} marked as slashed due to conflicting anchor",
                                envelope.node_id
                            );
                        }
                    }
                }
            }
            _ => {}
        },
        _ => {}
    }
    Ok(())
}

fn sha256_digest(data: &[u8]) -> [u8; 32] {
    let mut hasher = Sha256::new();
    hasher.update(data);
    let mut out = [0u8; 32];
    out.copy_from_slice(&hasher.finalize());
    out
}

fn record_invalid(
    map: &mut HashMap<libp2p::PeerId, usize>,
    peer: libp2p::PeerId,
    metrics: &Arc<Metrics>,
) {
    let entry = map.entry(peer).or_insert(0);
    *entry += 1;
    metrics.inc_invalid_envelopes();
    if *entry >= INVALID_THRESHOLD {
        println!("peer {peer} exceeded invalid envelope threshold");
        *entry = 0;
    }
}

async fn run_metrics_server(addr: SocketAddr, metrics: Arc<Metrics>) -> io::Result<()> {
    let listener = TcpListener::bind(addr).await?;
    loop {
        let (mut stream, _) = listener.accept().await?;
        let metrics = metrics.clone();
        tokio::spawn(async move {
            if let Err(err) = respond_with_metrics(&mut stream, metrics).await {
                eprintln!("metrics connection error: {err}");
            }
        });
    }
}

async fn respond_with_metrics(
    stream: &mut tokio::net::TcpStream,
    metrics: Arc<Metrics>,
) -> io::Result<()> {
    let mut buf = [0u8; 1024];
    let mut read = 0usize;
    loop {
        if read == buf.len() {
            break;
        }
        let n = stream.read(&mut buf[read..]).await?;
        if n == 0 {
            break;
        }
        read += n;
        if read >= 4 && &buf[read - 4..read] == b"\r\n\r\n" {
            break;
        }
    }

    let request_line = std::str::from_utf8(&buf[..read]).unwrap_or("");
    let path = request_line
        .lines()
        .next()
        .and_then(|line| {
            let mut parts = line.split_whitespace();
            let _method = parts.next()?;
            let path = parts.next()?;
            Some(path)
        })
        .unwrap_or("/metrics");

    if path != "/" && path != "/metrics" {
        let response = b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n";
        stream.write_all(response).await?;
        stream.shutdown().await?;
        return Ok(());
    }

    let body = metrics.render();
    let response = format!(
        "HTTP/1.1 200 OK\r\nContent-Type: text/plain; version=0.0.4\r\nContent-Length: {}\r\n\r\n{}",
        body.len(),
        body
    );
    stream.write_all(response.as_bytes()).await?;
    stream.shutdown().await
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct StoredAttestation {
    pk: String,
    sig: String,
}

impl StoredAttestation {
    fn to_attestation(&self, share_root: &str, pedersen_root: &Option<String>) -> Attestation {
        Attestation {
            share_root: share_root.to_string(),
            pedersen_root: pedersen_root.clone(),
            public_key: self.pk.clone(),
            signature: self.sig.clone(),
            ts: Some(now_millis()),
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct BlobMeta {
    namespace: String,
    hash: String,
    size: u64,
    data_shards: u8,
    parity_shards: u8,
    share_root: String,
    #[serde(default)]
    pedersen_root: Option<String>,
    #[serde(default)]
    attestations: Vec<StoredAttestation>,
    #[serde(default)]
    publisher_pk: Option<String>,
}

fn blob_paths(base: &Path, namespace: &str, hash: &str) -> (PathBuf, PathBuf, PathBuf) {
    let ns_dir = base.join(namespace);
    let meta_path = ns_dir.join(format!("{hash}.meta"));
    let blob_path = ns_dir.join(format!("{hash}.blob"));
    let share_dir = ns_dir.join(hash).join("shares");
    (meta_path, blob_path, share_dir)
}

fn load_blob_meta(
    base: &Path,
    namespace: &str,
    hash: &str,
) -> Result<Option<BlobMeta>, NetworkError> {
    let (meta_path, _blob_path, _share_dir) = blob_paths(base, namespace, hash);
    if !meta_path.exists() {
        return Ok(None);
    }
    let bytes = fs::read(&meta_path).map_err(|err| {
        NetworkError::Io(format!("failed to read {}: {err}", meta_path.display()))
    })?;
    let meta: BlobMeta = serde_json::from_slice(&bytes).map_err(|err| {
        NetworkError::Codec(format!("invalid blob meta {}: {err}", meta_path.display()))
    })?;
    Ok(Some(meta))
}

fn save_blob_meta(base: &Path, meta: &BlobMeta) -> Result<(), NetworkError> {
    let (meta_path, _, _) = blob_paths(base, &meta.namespace, &meta.hash);
    if let Some(parent) = meta_path.parent() {
        fs::create_dir_all(parent).map_err(|err| {
            NetworkError::Io(format!("failed to create {}: {err}", parent.display()))
        })?;
    }
    let data = serde_json::to_vec_pretty(meta)
        .map_err(|err| NetworkError::Codec(format!("meta encode error: {err}")))?;
    fs::write(&meta_path, data)
        .map_err(|err| NetworkError::Io(format!("failed to write {}: {err}", meta_path.display())))
}

fn evidence_root(blob_dir: &Option<PathBuf>) -> Option<String> {
    let Some(dir) = blob_dir.as_ref() else {
        return None;
    };
    let path = dir.join("evidence.jsonl");
    let contents = std::fs::read_to_string(&path).ok()?;
    let mut leaves = Vec::new();
    for line in contents.lines() {
        if line.trim().is_empty() {
            continue;
        }
        let mut hasher = Sha256::new();
        hasher.update(line.as_bytes());
        leaves.push(hasher.finalize().into());
    }
    if leaves.is_empty() {
        return None;
    }
    Some(hex::encode(crate::merkle_root(&leaves)))
}

fn latest_da_commitments(blob_dir: &Option<PathBuf>) -> Vec<DaCommitmentJson> {
    let mut map: HashMap<String, (SystemTime, DaCommitmentJson)> = HashMap::new();
    let Some(dir) = blob_dir.as_ref() else {
        return Vec::new();
    };
    let Ok(ns_entries) = fs::read_dir(dir) else {
        return Vec::new();
    };
    for ns in ns_entries.flatten() {
        let ns_path = ns.path();
        let Some(_ns_name) = ns_path.file_name().and_then(|n| n.to_str()) else {
            continue;
        };
        if !ns_path.is_dir() {
            continue;
        }
        let Ok(meta_entries) = fs::read_dir(&ns_path) else {
            continue;
        };
        for entry in meta_entries.flatten() {
            let path = entry.path();
            if path.extension().and_then(|e| e.to_str()) != Some("meta") {
                continue;
            }
            let meta_bytes = match fs::read(&path) {
                Ok(bytes) => bytes,
                Err(_) => continue,
            };
            let meta: BlobMeta = match serde_json::from_slice(&meta_bytes) {
                Ok(m) => m,
                Err(_) => continue,
            };
            let mtime = entry
                .metadata()
                .and_then(|m| m.modified())
                .unwrap_or(SystemTime::UNIX_EPOCH);
            let dc = DaCommitmentJson {
                namespace: meta.namespace.clone(),
                blob_hash: meta.hash.clone(),
                share_root: meta.share_root.clone(),
                pedersen_root: meta.pedersen_root.clone(),
                attestation_qc: None,
            };
            match map.get(&meta.namespace) {
                Some((seen_time, _)) if *seen_time >= mtime => {}
                _ => {
                    map.insert(meta.namespace.clone(), (mtime, dc));
                }
            }
        }
    }
    map.into_iter().map(|(_, v)| v.1).collect()
}

fn lookup_stake(cfg: &NetConfig, pk_b64: &str) -> Option<u64> {
    let vk = decode_public_key_base64(pk_b64).ok()?;
    if let Some(weight) = cfg.membership_policy.stake_for(&vk) {
        return Some(weight);
    }
    if let Some(path) = &cfg.stake_registry_path {
        if let Ok(reg) = StakeRegistry::load(path) {
            if let Some(w) = reg.stake_for(pk_b64) {
                return Some(w);
            }
        }
    }
    Some(1)
}

fn append_evidence(path: &Path, namespace: &str, blob_hash: &str, pk: &str, reason: &str) {
    let record = serde_json::json!({
        "namespace": namespace,
        "blob_hash": blob_hash,
        "pk": pk,
        "reason": reason,
        "ts": now_millis(),
    });
    if let Some(parent) = path.parent() {
        let _ = fs::create_dir_all(parent);
    }
    let line = format!("{}\n", record);
    if let Err(err) = fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(path)
        .and_then(|mut f| std::io::Write::write_all(&mut f, line.as_bytes()))
    {
        eprintln!("failed to write evidence: {err}");
    }
}

fn append_availability_evidence(path: &Path, ev: &AvailabilityEvidence) {
    if let Some(parent) = path.parent() {
        let _ = fs::create_dir_all(parent);
    }
    match serde_json::to_string(ev) {
        Ok(line) => {
            let line = format!("{}\n", line);
            if let Err(err) = fs::OpenOptions::new()
                .create(true)
                .append(true)
                .open(path)
                .and_then(|mut f| std::io::Write::write_all(&mut f, line.as_bytes()))
            {
                eprintln!("failed to write availability evidence: {err}");
            }
        }
        Err(err) => eprintln!("failed to encode availability evidence: {err}"),
    }
}

fn append_rollup_fault_evidence(path: &Path, ev: &RollupFaultEvidence) {
    if let Some(parent) = path.parent() {
        let _ = fs::create_dir_all(parent);
    }
    match serde_json::to_string(ev) {
        Ok(line) => {
            let line = format!("{}\n", line);
            if let Err(err) = fs::OpenOptions::new()
                .create(true)
                .append(true)
                .open(path)
                .and_then(|mut f| std::io::Write::write_all(&mut f, line.as_bytes()))
            {
                eprintln!("failed to write rollup fault evidence: {err}");
            }
        }
        Err(err) => eprintln!("failed to encode rollup fault evidence: {err}"),
    }
}

fn record_slash_with_registry(
    policy: &Arc<dyn MembershipPolicy>,
    registry_path: &Option<PathBuf>,
    pk_b64: &str,
    reason: &str,
) {
    if let Ok(vk) = decode_public_key_base64(pk_b64) {
        if let Err(err) = policy.record_slash(&vk) {
            eprintln!("failed to record slash in policy: {err}");
        }
    }
    if let Some(path) = registry_path {
        match StakeRegistry::load(path) {
            Ok(mut reg) => {
                reg.slash(pk_b64);
                if let Err(err) = reg.save(path) {
                    eprintln!("failed to persist stake registry: {err}");
                } else {
                    println!("slash recorded for {pk_b64} ({reason})");
                }
            }
            Err(err) => eprintln!("failed to load stake registry for slashing: {err}"),
        }
    }
}
#[cfg(test)]
mod tests {
    use super::*;
    use crate::transcript_digest;
    use std::fs;
    use std::sync::atomic::Ordering;
    use std::time::SystemTime;

    fn temp_path(name: &str) -> PathBuf {
        let mut base = std::env::temp_dir();
        let nanos = SystemTime::now()
            .duration_since(SystemTime::UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        base.push(format!("{}_{}", name, nanos));
        base
    }

    #[test]
    fn payload_cache_rejects_duplicates() {
        let metrics = Arc::new(Metrics::default());
        let mut cache = PayloadCache::new(metrics.clone());
        let digest = [1u8; 32];
        assert!(cache.insert(digest));
        assert!(!cache.insert(digest));
    }

    #[test]
    fn payload_cache_eviction_tracks_metric() {
        let metrics = Arc::new(Metrics::default());
        let mut cache = PayloadCache::new(metrics.clone());
        for i in 0..(SEEN_CACHE_LIMIT + 1) {
            let mut digest = [0u8; 32];
            digest[..8].copy_from_slice(&(i as u64).to_le_bytes());
            cache.insert(digest);
        }
        assert!(metrics.lrucache_evictions_total.load(Ordering::Relaxed) >= 1);
    }

    #[test]
    fn identical_logs_yield_identical_anchors() {
        let dir = temp_path("jroc_net_logs");
        fs::create_dir_all(&dir).unwrap();
        let log_path = dir.join("ledger_0000.txt");
        let challenges = vec![1, 2, 3];
        let round_sums = vec![4, 5, 6];
        let final_value = 7;
        let hash = transcript_digest(&challenges, &round_sums, final_value);
        let content = format!(
            "statement:Test Statement\ntranscript:{}\nround_sums:{}\nfinal:{}\nhash:{}\n",
            challenges
                .iter()
                .map(|v| v.to_string())
                .collect::<Vec<_>>()
                .join(" "),
            round_sums
                .iter()
                .map(|v| v.to_string())
                .collect::<Vec<_>>()
                .join(" "),
            final_value,
            crate::transcript_digest_to_hex(&hash)
        );
        fs::write(&log_path, content).unwrap();

        let anchor_a = load_anchor_from_logs(&dir).unwrap();
        let anchor_b = load_anchor_from_logs(&dir).unwrap();
        assert_eq!(anchor_a.entries, anchor_b.entries);

        fs::remove_dir_all(&dir).unwrap();
    }

    #[test]
    fn tampered_log_is_rejected() {
        let dir = temp_path("jroc_net_logs_tampered");
        fs::create_dir_all(&dir).unwrap();
        let log_path = dir.join("ledger_0000.txt");
        let content = "statement:Demo\ntranscript:1\nround_sums:2\nfinal:3\nhash:999\n";
        fs::write(&log_path, content).unwrap();
        let result = load_anchor_from_logs(&dir);
        assert!(result.is_err());
        fs::remove_dir_all(&dir).unwrap();
    }
}

fn load_anchor_from_logs(path: &Path) -> Result<LedgerAnchor, NetworkError> {
    let mut cutoff: Option<String> = None;
    let mut anchor_from_checkpoint = false;
    let anchor = match load_latest_checkpoint(path) {
        Ok(Some(checkpoint)) => {
            anchor_from_checkpoint = true;
            let (anchor, cp_cutoff) = checkpoint
                .clone()
                .into_ledger()
                .map_err(|err| NetworkError::Anchor(err.to_string()))?;
            cutoff = cp_cutoff;
            anchor
        }
        Ok(None) => julian_genesis_anchor(),
        Err(err) => return Err(NetworkError::Anchor(err.to_string())),
    };
    let mut entries = anchor.entries;
    let mut metadata = anchor.metadata;
    if !anchor_from_checkpoint {
        metadata.challenge_mode = None;
        metadata.fold_digest = None;
    }
    metadata
        .crate_version
        .get_or_insert_with(|| env!("CARGO_PKG_VERSION").to_string());
    let mut files: Vec<PathBuf> = std::fs::read_dir(path)
        .map_err(|err| NetworkError::Io(format!("failed to read {}: {err}", path.display())))?
        .filter_map(|entry| entry.ok().map(|e| e.path()))
        .filter(|p| p.is_file() && is_ledger_file(p))
        .collect();
    files.sort();
    for file in files {
        if let Some(ref cutoff_name) = cutoff {
            if let Some(name) = file.file_name().and_then(|n| n.to_str()) {
                if name <= cutoff_name.as_str() {
                    continue;
                }
            }
        }
        let parsed = parse_log_file(&file).map_err(NetworkError::Anchor)?;
        if let Some(mode) = parsed.metadata.challenge_mode {
            match &mut metadata.challenge_mode {
                None => metadata.challenge_mode = Some(mode),
                Some(existing) if existing != &mode => {
                    return Err(NetworkError::Anchor(format!(
                        "{} challenge_mode {} conflicts with existing {}",
                        file.display(),
                        mode,
                        existing
                    )));
                }
                _ => {}
            }
        }
        if let Some(digest) = parsed.metadata.fold_digest {
            if let Some(existing) = &metadata.fold_digest {
                if existing != &digest && anchor_from_checkpoint {
                    return Err(NetworkError::Anchor(format!(
                        "{} fold_digest conflicts with existing value",
                        file.display()
                    )));
                }
            }
            metadata.fold_digest = Some(digest);
        }
        let entry_hashes = vec![parsed.digest];
        entries.push(EntryAnchor {
            statement: parsed.statement,
            merkle_root: merkle_root(&entry_hashes),
            hashes: entry_hashes,
        });
    }
    if entries.is_empty() {
        entries = julian_genesis_anchor().entries;
    }
    if let Some(digest) = read_fold_digest_hint(path).map_err(NetworkError::Anchor)? {
        if let Some(existing) = &metadata.fold_digest {
            if existing != &digest && anchor_from_checkpoint {
                return Err(NetworkError::Anchor(
                    "fold_digest hint conflicts with checkpoint metadata".to_string(),
                ));
            }
        }
        metadata.fold_digest = Some(digest);
    }
    let mut anchor = LedgerAnchor { entries, metadata };
    if anchor.metadata.fold_digest.is_none() {
        anchor.metadata.fold_digest = Some(compute_fold_digest(&anchor));
    }
    Ok(anchor)
}

fn is_ledger_file(path: &Path) -> bool {
    match path.file_name().and_then(|n| n.to_str()) {
        Some(name) => name.starts_with("ledger_") && name.ends_with(".txt"),
        None => false,
    }
}

fn now_millis() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as u64
}
