use std::collections::{HashMap, HashSet};
use std::net::SocketAddr;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;

use tokio::sync::Mutex;
use tracing::{debug, info, trace, warn};

use crate::activity::{ActivityCategory, ActivityLevel, ActivityLog};
use crate::blob::BlobStore;
use crate::content::verify_post_id;
use crate::crypto;
use crate::protocol::{
    read_message_type, read_payload, write_typed_message, AnchorReferral,
    AnchorReferralRequestPayload, AnchorReferralResponsePayload, AnchorRegisterPayload,
    AudienceRequestPayload, AudienceResponsePayload, BlobHeaderDiffPayload,
    BlobHeaderRequestPayload, BlobHeaderResponsePayload, BlobRequestPayload, BlobResponsePayload,
    CircleProfileUpdatePayload, GroupKeyDistributePayload, GroupKeyRequestPayload,
    GroupKeyResponsePayload, InitialExchangePayload, MeshPreferPayload,
    MessageType, NodeListUpdatePayload, PostDownstreamRegisterPayload,
    PostNotificationPayload, PostPushPayload,
    ProfileUpdatePayload, PullSyncRequestPayload, PullSyncResponsePayload,
    RefuseRedirectPayload, RelayIntroducePayload, RelayIntroduceResultPayload, SessionRelayPayload,
    SocialAddressUpdatePayload, SocialCheckinPayload, SocialDisconnectNoticePayload,
    SyncPost, VisibilityUpdatePayload, WormQueryPayload, WormResponsePayload,
    ReplicationRequestPayload, ReplicationResponsePayload, ALPN_V2,
};
use crate::storage::Storage;
use crate::types::{
    DeviceProfile, NodeId, PeerSlotKind, PeerWithAddress, PostId, PostVisibility, ReachMethod,
    SessionReachMethod, SocialRouteEntry, SocialStatus, WormId, WormResult,
};

const MAX_PAYLOAD: usize = 64 * 1024 * 1024; // 64 MB

const WORM_FAN_OUT_TIMEOUT_MS: u64 = 500;
const WORM_BLOOM_TIMEOUT_MS: u64 = 1500;
const WORM_TOTAL_TIMEOUT_MS: u64 = 3000;
const WORM_COOLDOWN_MS: i64 = 300_000; // 5 min
const WORM_DEDUP_EXPIRY_MS: u64 = 10_000; // 10 sec
const SESSION_IDLE_TIMEOUT_MS: u64 = 300_000; // 5 min
#[allow(dead_code)]
const RELAY_COOLDOWN_MS: i64 = 300_000; // 5 min
const RELAY_INTRO_DEDUP_EXPIRY_MS: u64 = 30_000; // 30 sec
#[allow(dead_code)]
const RELAY_INTRO_TIMEOUT_MS: u64 = 15_000; // 15 sec
const HOLE_PUNCH_TIMEOUT_MS: u64 = 30_000; // 30 sec overall window
const HOLE_PUNCH_ATTEMPT_MS: u64 = 2_000; // 2 sec per attempt before retry
/// Max bytes relayed per pipe before closing
const RELAY_MAX_BYTES: u64 = 50 * 1024 * 1024; // 50 MB
/// Relay pipe idle timeout
const RELAY_PIPE_IDLE_MS: u64 = 120_000; // 2 min
/// How long a preferred peer can be unreachable before being pruned (7 days)
const PREFERRED_UNREACHABLE_PRUNE_MS: u64 = 7 * 24 * 60 * 60 * 1000;
/// How long reconnect watchers live before expiry (30 days)
const WATCHER_EXPIRY_MS: i64 = 30 * 24 * 60 * 60 * 1000;
/// Max pending introductions per target in 5 minutes
#[allow(dead_code)]
const RELAY_TARGET_RATE_LIMIT: usize = 5;

/// Grace period before removing disconnected peers from referral list
const REFERRAL_DISCONNECT_GRACE_MS: u64 = 120_000; // 2 min
/// Soft cap on referral list size (affects max_uses tiering)
const REFERRAL_LIST_CAP: usize = 50;

/// Zombie connection timeout: no stream activity for this long = dead
const ZOMBIE_TIMEOUT_MS: u64 = 600_000; // 10 minutes

/// Mesh keepalive interval: send a lightweight ping to prevent zombie reaping + NAT timeout
const MESH_KEEPALIVE_INTERVAL_SECS: u64 = 30;

/// Relay introduction identifier for deduplication
pub type IntroId = [u8; 16];

/// Result of initial exchange: accepted or refused with optional redirect peer.
pub enum ExchangeResult {
    Accepted,
    Refused { redirect: Option<PeerWithAddress> },
}

/// Hole punch: try all addresses in parallel, retrying every HOLE_PUNCH_ATTEMPT_MS
/// for up to HOLE_PUNCH_TIMEOUT_MS total. Returns the first successful connection
/// or None if all attempts fail.
pub(crate) async fn hole_punch_parallel(
    endpoint: &iroh::Endpoint,
    target: &NodeId,
    addresses: &[String],
) -> Option<iroh::endpoint::Connection> {
    use crate::protocol::ALPN_V2;

    let addrs: Vec<iroh::EndpointAddr> = addresses
        .iter()
        .filter_map(|addr_str| {
            let sock = normalize_addr(addr_str.parse::<std::net::SocketAddr>().ok()?);
            let eid = iroh::EndpointId::from_bytes(target).ok()?;
            Some(iroh::EndpointAddr::from(eid).with_ip_addr(sock))
        })
        .collect();

    if addrs.is_empty() {
        return None;
    }

    let deadline = tokio::time::Instant::now()
        + std::time::Duration::from_millis(HOLE_PUNCH_TIMEOUT_MS);
    let mut attempt = 0u32;

    while tokio::time::Instant::now() < deadline {
        attempt += 1;

        // Spawn a connect attempt to every address in parallel
        let mut handles = Vec::new();
        for addr in &addrs {
            let ep = endpoint.clone();
            let a = addr.clone();
            handles.push(tokio::spawn(async move {
                tokio::time::timeout(
                    std::time::Duration::from_millis(HOLE_PUNCH_ATTEMPT_MS),
                    ep.connect(a, ALPN_V2),
                ).await
            }));
        }

        // Wait for all to finish — return first success
        for handle in handles {
            if let Ok(Ok(Ok(conn))) = handle.await {
                tracing::info!(
                    peer = hex::encode(target),
                    attempt,
                    "Hole punch succeeded"
                );
                return Some(conn);
            }
        }

        tracing::debug!(
            peer = hex::encode(target),
            attempt,
            "Hole punch attempt failed, retrying"
        );
    }

    tracing::debug!(peer = hex::encode(target), attempts = attempt, "Hole punch exhausted");
    None
}

/// Timeout for each individual scan connect attempt (200ms → ~20 in-flight at 100/sec)
const SCAN_CONNECT_TIMEOUT_MS: u64 = 200;
/// Scan rate: one attempt every 10ms = 100 ports/sec
const SCAN_INTERVAL_MS: u64 = 10;
/// How often to punch peer's anchor-observed address during scanning (seconds).
/// Each punch checks if the peer has opened a firewall port matching our actual port.
const SCAN_PUNCH_INTERVAL_SECS: u64 = 2;
/// Maximum scan duration (seconds) — accept the cost for otherwise-impossible connections
const SCAN_MAX_DURATION_SECS: u64 = 300; // 5 minutes

/// Advanced hole punch with port scanning fallback for EDM/port-restricted NAT.
///
/// **Role-based behavior** (each side calls this independently):
///
/// - **Scanner** (our mapping = EDM or Unknown): Walk outward from peer's anchor-observed
///   base port at ~100/sec. Each probe opens a firewall entry on our NAT so that when
///   the peer punches us, their packet can land. Also punch every 2s to peer's observed
///   address to check if they've opened their port for us.
///
/// - **Puncher** (our mapping = EIM): Our port is stable — the peer's scanner will find
///   us. We just punch every 2s to peer's anchor-observed address, which also keeps our
///   NAT mapping alive and checks if the peer's scan has opened their firewall for us.
///
/// For both-EDM pairs: both sides scan + punch simultaneously.
pub(crate) async fn hole_punch_with_scanning(
    endpoint: &iroh::Endpoint,
    target: &NodeId,
    addresses: &[String],
    our_profile: crate::types::NatProfile,
    peer_profile: crate::types::NatProfile,
) -> Option<iroh::endpoint::Connection> {
    // Step 1: Standard hole punch (one quick round to anchor-observed address)
    let quick_result = hole_punch_single(endpoint, target, addresses).await;
    if quick_result.is_some() {
        return quick_result;
    }

    // Step 2: Decide whether to scan
    if !our_profile.should_try_scanning(&peer_profile) {
        // Neither side is EDM — standard punch is all we can do, try full window
        return hole_punch_parallel(endpoint, target, addresses).await;
    }

    // Parse anchor-observed address (first in list, injected by relay)
    let observed_addr = addresses.first()
        .and_then(|a| a.parse::<std::net::SocketAddr>().ok())
        .map(|s| normalize_addr(s));

    let (scan_ip, base_port) = match observed_addr {
        Some(s) => (s.ip(), s.port()),
        None => return None,
    };

    let eid = match iroh::EndpointId::from_bytes(target).ok() {
        Some(e) => e,
        None => return None,
    };

    // Determine our role based on mapping type
    let we_scan = our_profile.mapping != crate::types::NatMapping::EndpointIndependent;
    // EIM = stable port, no need to scan. EDM/Unknown = our port changes, need to scan.

    let role = if we_scan { "scanner+puncher" } else { "puncher" };
    tracing::info!(
        peer = hex::encode(target),
        our_mapping = %our_profile.mapping,
        peer_mapping = %peer_profile.mapping,
        role,
        scan_ip = %scan_ip,
        base_port,
        "Advanced NAT traversal started"
    );

    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(SCAN_MAX_DURATION_SECS);
    let (found_tx, mut found_rx) = tokio::sync::mpsc::channel::<iroh::endpoint::Connection>(1);
    let mut join_set = tokio::task::JoinSet::new();

    // Punch address: single anchor-observed address for the peer
    let punch_addr = iroh::EndpointAddr::from(eid).with_ip_addr(
        std::net::SocketAddr::new(scan_ip, base_port)
    );

    // Set up punch interval (both roles punch every 2s)
    let mut punch_interval = tokio::time::interval(std::time::Duration::from_secs(SCAN_PUNCH_INTERVAL_SECS));
    punch_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    // Skip the first tick (we already tried quick punch above)
    punch_interval.tick().await;

    if we_scan {
        // Scanner role: walk outward from base port at ~100/sec + punch every 2s
        let mut port_iter = PortWalkIter::new(base_port);
        let mut scan_interval = tokio::time::interval(std::time::Duration::from_millis(SCAN_INTERVAL_MS));
        scan_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);

        let mut ports_tried: u32 = 0;
        let mut last_log_count: u32 = 0;

        loop {
            tokio::select! {
                // Rate-limited port scan: one attempt per tick (~100/sec)
                _ = scan_interval.tick() => {
                    if tokio::time::Instant::now() >= deadline {
                        break;
                    }
                    if let Some(port) = port_iter.next() {
                        let ep = endpoint.clone();
                        let tx = found_tx.clone();
                        let addr = iroh::EndpointAddr::from(eid).with_ip_addr(
                            std::net::SocketAddr::new(scan_ip, port)
                        );
                        join_set.spawn(async move {
                            if let Ok(Ok(conn)) = tokio::time::timeout(
                                std::time::Duration::from_millis(SCAN_CONNECT_TIMEOUT_MS),
                                ep.connect(addr, ALPN_V2),
                            ).await {
                                let _ = tx.send(conn).await;
                            }
                        });
                        ports_tried += 1;

                        // Log progress every 1000 ports
                        if ports_tried - last_log_count >= 1000 {
                            last_log_count = ports_tried;
                            tracing::info!(
                                peer = hex::encode(target),
                                ports_tried,
                                current_offset = port_iter.current_offset(),
                                "Port scan progress"
                            );
                        }
                    } else {
                        break; // exhausted all valid ports
                    }
                }

                // Every 2s: punch peer's anchor-observed address
                _ = punch_interval.tick() => {
                    let ep = endpoint.clone();
                    let tx = found_tx.clone();
                    let addr = punch_addr.clone();
                    join_set.spawn(async move {
                        if let Ok(Ok(conn)) = tokio::time::timeout(
                            std::time::Duration::from_millis(SCAN_CONNECT_TIMEOUT_MS),
                            ep.connect(addr, ALPN_V2),
                        ).await {
                            let _ = tx.send(conn).await;
                        }
                    });
                }

                // Success!
                result = found_rx.recv() => {
                    if let Some(conn) = result {
                        tracing::info!(
                            peer = hex::encode(target),
                            ports_tried,
                            "Advanced NAT: scan+punch succeeded"
                        );
                        join_set.abort_all();
                        return Some(conn);
                    }
                    break;
                }
            }
        }

        join_set.abort_all();
        tracing::info!(peer = hex::encode(target), ports_tried, "Advanced NAT: scan exhausted");
    } else {
        // Puncher-only role (EIM): our port is stable, peer scans to find us.
        // We just punch every 2s to peer's observed address to keep our mapping
        // alive and check if their scan has opened their firewall for us.
        loop {
            tokio::select! {
                _ = punch_interval.tick() => {
                    if tokio::time::Instant::now() >= deadline {
                        break;
                    }
                    let ep = endpoint.clone();
                    let tx = found_tx.clone();
                    let addr = punch_addr.clone();
                    join_set.spawn(async move {
                        if let Ok(Ok(conn)) = tokio::time::timeout(
                            std::time::Duration::from_millis(SCAN_CONNECT_TIMEOUT_MS),
                            ep.connect(addr, ALPN_V2),
                        ).await {
                            let _ = tx.send(conn).await;
                        }
                    });
                }

                result = found_rx.recv() => {
                    if let Some(conn) = result {
                        tracing::info!(
                            peer = hex::encode(target),
                            "Advanced NAT: punch succeeded (peer's scan opened their firewall)"
                        );
                        join_set.abort_all();
                        return Some(conn);
                    }
                    break;
                }
            }
        }

        join_set.abort_all();
        tracing::info!(peer = hex::encode(target), "Advanced NAT: punch-only exhausted (peer may not be scanning)");
    }

    None
}

/// Iterator that walks outward from a base port: base, base+1, base-1, base+2, base-2, ...
/// Skips ports outside [1, 65535].
struct PortWalkIter {
    base: u16,
    offset: u32,
    tried_plus: bool, // within current offset, have we tried base+offset?
}

impl PortWalkIter {
    fn new(base: u16) -> Self {
        Self { base, offset: 0, tried_plus: false }
    }

    fn current_offset(&self) -> u32 {
        self.offset
    }
}

impl Iterator for PortWalkIter {
    type Item = u16;

    fn next(&mut self) -> Option<u16> {
        // offset 0: just the base port
        if self.offset == 0 {
            self.offset = 1;
            self.tried_plus = false;
            return Some(self.base);
        }

        // For each offset > 0: try base+offset, then base-offset
        loop {
            if self.offset > 65535 {
                return None; // exhausted
            }

            if !self.tried_plus {
                self.tried_plus = true;
                let candidate = self.base as u32 + self.offset;
                if candidate <= 65535 {
                    return Some(candidate as u16);
                }
                // Fall through to try minus
            }

            // Try base - offset
            self.tried_plus = false;
            self.offset += 1;
            let prev_offset = self.offset - 1;
            if (self.base as u32) >= prev_offset && self.base as u32 - prev_offset >= 1 {
                return Some((self.base as u32 - prev_offset) as u16);
            }
            // Both sides out of range at this offset, try next
        }
    }
}

/// Quick single punch to the anchor-observed address only (first in list).
/// Used as the initial attempt before escalating to scanning.
async fn hole_punch_single(
    endpoint: &iroh::Endpoint,
    target: &NodeId,
    addresses: &[String],
) -> Option<iroh::endpoint::Connection> {
    let addr = addresses.first()
        .and_then(|a| a.parse::<std::net::SocketAddr>().ok())
        .map(|s| normalize_addr(s))?;
    let eid = iroh::EndpointId::from_bytes(target).ok()?;
    let endpoint_addr = iroh::EndpointAddr::from(eid).with_ip_addr(addr);

    match tokio::time::timeout(
        std::time::Duration::from_millis(HOLE_PUNCH_ATTEMPT_MS),
        endpoint.connect(endpoint_addr, ALPN_V2),
    ).await {
        Ok(Ok(conn)) => {
            tracing::info!(peer = hex::encode(target), "Quick single punch succeeded");
            Some(conn)
        }
        _ => None,
    }
}


/// Normalize IPv4-mapped IPv6 addresses (e.g. [::ffff:1.2.3.4]:port) to plain IPv4.
/// Dual-stack servers report IPv4 peers as mapped-v6 but v4-only clients can't reach them.
pub fn normalize_addr(addr: std::net::SocketAddr) -> std::net::SocketAddr {
    match addr {
        std::net::SocketAddr::V6(v6) => {
            if let Some(v4) = v6.ip().to_ipv4_mapped() {
                std::net::SocketAddr::new(std::net::IpAddr::V4(v4), v6.port())
            } else {
                addr
            }
        }
        _ => addr,
    }
}

pub struct MeshConnection {
    pub node_id: NodeId,
    pub connection: iroh::endpoint::Connection,
    pub slot_kind: PeerSlotKind,
    pub connected_at: u64,
    /// Remote address as seen from our side of the QUIC connection
    pub remote_addr: Option<std::net::SocketAddr>,
    /// Last time a stream was accepted on this connection (for zombie detection)
    pub last_activity: Arc<AtomicU64>,
}

pub struct SessionConnection {
    pub node_id: NodeId,
    pub connection: iroh::endpoint::Connection,
    pub created_at: u64,
    pub last_active_at: u64,
    pub reach_method: SessionReachMethod,
    /// Remote address as seen from our side of the QUIC connection
    pub remote_addr: Option<std::net::SocketAddr>,
}

pub struct PullSyncStats {
    pub posts_received: usize,
    pub visibility_updates: usize,
}

/// Entry in the anchor's referral list — connection-backed, self-pruning.
struct ReferralEntry {
    node_id: NodeId,
    addresses: Vec<String>,
    #[allow(dead_code)]
    registered_at: u64,
    use_count: u32,
    disconnected_at: Option<u64>,
}

pub struct ConnectionManager {
    connections: HashMap<NodeId, MeshConnection>,
    endpoint: iroh::Endpoint,
    storage: Arc<Mutex<Storage>>,
    our_node_id: NodeId,
    #[allow(dead_code)]
    is_anchor: Arc<AtomicBool>,
    diff_seq: AtomicU64,
    #[allow(dead_code)]
    secret_seed: [u8; 32],
    blob_store: Arc<BlobStore>,
    /// Dedup map for worm queries: worm_id → timestamp_ms
    seen_worms: HashMap<WormId, u64>,
    /// Last broadcast N1 set (for computing diffs)
    last_n1_set: HashSet<NodeId>,
    /// Last broadcast N2 set (for computing diffs)
    last_n2_set: HashSet<NodeId>,
    /// Max preferred (bilateral) mesh slots
    preferred_slots: usize,
    /// Max local (diverse) mesh slots
    local_slots: usize,
    /// Max wide (bloom-sourced) mesh slots
    wide_slots: usize,
    /// Session connections: short-lived, tracked, separate from mesh slots
    sessions: HashMap<NodeId, SessionConnection>,
    /// Max session slots
    session_slots: usize,
    /// Dedup map for relay introductions: intro_id → timestamp_ms
    seen_intros: HashMap<IntroId, u64>,
    /// Active relay pipe count (we are the intermediary)
    active_relay_pipes: Arc<AtomicU64>,
    /// Max concurrent relay pipes
    max_relay_pipes: usize,
    /// Device profile (for resource limits)
    #[allow(dead_code)]
    device_profile: DeviceProfile,
    /// Peers known to be unreachable directly: node_id → last_failed_at_ms
    /// Learned over the session — cleared on restart, updated on connect success/failure
    unreachable_peers: HashMap<NodeId, u64>,
    /// Anchor-side referral list: connected peers available for referral
    referral_list: HashMap<NodeId, ReferralEntry>,
    /// Channel to signal the growth loop to wake up and seek diverse peers
    growth_tx: Option<tokio::sync::mpsc::Sender<()>>,
    /// Channel to signal the recovery loop when mesh drops below threshold
    recovery_tx: Option<tokio::sync::mpsc::Sender<()>>,
    activity_log: Arc<std::sync::Mutex<ActivityLog>>,
    /// UPnP external address (prepended to self-reported addresses in anchor registration)
    upnp_external_addr: Option<SocketAddr>,
    /// Stable bind address (from --bind flag), used for anchor advertised address
    bind_addr: Option<SocketAddr>,
    /// Our detected NAT type (from STUN probing on startup)
    nat_type: crate::types::NatType,
    /// Our detected NAT mapping (from STUN probing on startup)
    nat_mapping: crate::types::NatMapping,
    /// Our detected NAT filtering (from anchor filter probe, starts Unknown)
    pub(crate) nat_filtering: crate::types::NatFiltering,
    /// Anchor probe: last successful probe timestamp (0 = never)
    last_probe_success_ms: u64,
    /// Anchor probe: consecutive failure count
    probe_failure_streak: u8,
    /// When this ConnectionManager was created
    started_at_ms: u64,
    /// Dedup map for anchor probes: probe_id → timestamp_ms
    seen_probes: HashMap<[u8; 16], u64>,
    /// Whether this node's HTTP server is running and externally reachable
    pub(crate) http_capable: bool,
    /// External HTTP address (ip:port) if known
    pub(crate) http_addr: Option<String>,
    /// Sticky N1 entries: NodeIds to report in N1 share until expiry (ms).
    /// Used to advertise the bootstrap anchor for 24h after isolation recovery.
    sticky_n1: HashMap<NodeId, u64>,
}

impl ConnectionManager {
    pub fn new(
        endpoint: iroh::Endpoint,
        storage: Arc<Mutex<Storage>>,
        our_node_id: NodeId,
        is_anchor: Arc<AtomicBool>,
        secret_seed: [u8; 32],
        blob_store: Arc<BlobStore>,
        profile: DeviceProfile,
        activity_log: Arc<std::sync::Mutex<ActivityLog>>,
        upnp_external_addr: Option<SocketAddr>,
        bind_addr: Option<SocketAddr>,
        nat_type: crate::types::NatType,
        nat_mapping: crate::types::NatMapping,
    ) -> Self {
        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis() as u64;
        Self {
            connections: HashMap::new(),
            endpoint,
            storage,
            our_node_id,
            is_anchor,
            diff_seq: AtomicU64::new(0),
            secret_seed,
            blob_store,
            seen_worms: HashMap::new(),
            last_n1_set: HashSet::new(),
            last_n2_set: HashSet::new(),
            preferred_slots: profile.preferred_slots(),
            local_slots: profile.local_slots(),
            wide_slots: profile.wide_slots(),
            sessions: HashMap::new(),
            session_slots: profile.session_slots(),
            seen_intros: HashMap::new(),
            active_relay_pipes: Arc::new(AtomicU64::new(0)),
            max_relay_pipes: profile.max_relay_pipes(),
            device_profile: profile,
            unreachable_peers: HashMap::new(),
            referral_list: HashMap::new(),
            growth_tx: None,
            recovery_tx: None,
            activity_log,
            upnp_external_addr,
            bind_addr,
            nat_type,
            nat_mapping,
            // Filtering starts Unknown — even "Public" nodes may only be public on
            // IPv6 while NATed on IPv4. The anchor filter probe determines this reliably.
            nat_filtering: crate::types::NatFiltering::Unknown,
            last_probe_success_ms: 0,
            probe_failure_streak: 0,
            started_at_ms: now,
            seen_probes: HashMap::new(),
            http_capable: false,
            http_addr: None,
            sticky_n1: HashMap::new(),
        }
    }

    /// Our detected NAT type
    pub fn nat_type(&self) -> crate::types::NatType {
        self.nat_type
    }

    /// Our detected NAT mapping
    pub fn nat_mapping(&self) -> crate::types::NatMapping {
        self.nat_mapping
    }

    /// Our NAT profile (mapping + filtering)
    pub fn our_nat_profile(&self) -> crate::types::NatProfile {
        crate::types::NatProfile::new(self.nat_mapping, self.nat_filtering)
    }

    /// Whether this node is a candidate for anchor status (has UPnP/public, enough connections, uptime)
    pub fn is_anchor_candidate(&self) -> bool {
        let now = now_ms();
        // Must have UPnP mapping or public IPv6
        let has_public_addr = self.upnp_external_addr.is_some()
            || self.endpoint.addr().ip_addrs().any(|s| crate::network::is_publicly_routable(&s));
        if !has_public_addr {
            return false;
        }
        // Must have enough connections (at least 50)
        if self.connections.len() < 50 {
            return false;
        }
        // Must have been running for at least 2 hours
        let two_hours_ms = 2 * 60 * 60 * 1000;
        if now.saturating_sub(self.started_at_ms) < two_hours_ms {
            return false;
        }
        // Not mobile
        if self.device_profile == crate::types::DeviceProfile::Mobile {
            return false;
        }
        true
    }

    /// Whether an anchor probe is due (candidate minus probe-success check, last probe > 30 min ago)
    pub fn probe_due(&self) -> bool {
        let now = now_ms();
        let has_public_addr = self.upnp_external_addr.is_some()
            || self.endpoint.addr().ip_addrs().any(|s| crate::network::is_publicly_routable(&s));
        if !has_public_addr {
            return false;
        }
        if self.connections.len() < 50 {
            return false;
        }
        let two_hours_ms = 2 * 60 * 60 * 1000;
        if now.saturating_sub(self.started_at_ms) < two_hours_ms {
            return false;
        }
        if self.device_profile == crate::types::DeviceProfile::Mobile {
            return false;
        }
        // Last probe must be > 30 min ago
        let thirty_min_ms = 30 * 60 * 1000;
        now.saturating_sub(self.last_probe_success_ms) > thirty_min_ms
    }

    /// Initiate an anchor self-verification probe.
    /// Selects a stranger from N2, sends probe request via the reporter.
    /// Returns true if probe succeeded (we're reachable).
    pub async fn initiate_anchor_probe(&mut self) -> anyhow::Result<bool> {
        use crate::protocol::{
            AnchorProbeRequestPayload, AnchorProbeResultPayload,
            MessageType, read_message_type, read_payload, write_typed_message,
        };

        let our_connections: HashSet<NodeId> = self.connections.keys().copied().collect();
        let (witness, reporter) = {
            let s = self.storage.lock().await;
            match s.random_n2_stranger(&our_connections)? {
                Some(pair) => pair,
                None => {
                    debug!("No N2 stranger available for anchor probe");
                    return Ok(false);
                }
            }
        };

        // Build our external address for the probe target
        let target_addr = match self.build_anchor_advertised_addr() {
            Some(addr) => addr,
            None => {
                debug!("No advertised address for anchor probe");
                return Ok(false);
            }
        };

        // Our globally-routable addresses for the witness to deliver result
        let mut candidate_addrs: Vec<String> = self.endpoint.addr().ip_addrs()
            .filter(|s| crate::network::is_publicly_routable(s))
            .map(|s| s.to_string())
            .collect();
        if let Some(ref ext) = self.upnp_external_addr {
            let ext_str = ext.to_string();
            if !candidate_addrs.contains(&ext_str) {
                candidate_addrs.insert(0, ext_str);
            }
        }

        let probe_id: [u8; 16] = {
            use rand::Rng;
            let mut id = [0u8; 16];
            rand::rng().fill(&mut id);
            id
        };

        let payload = AnchorProbeRequestPayload {
            target_addr,
            witness,
            candidate: self.our_node_id,
            candidate_addresses: candidate_addrs,
            probe_id,
        };

        // Send to reporter via bi-stream
        let reporter_conn = match self.connections.get(&reporter) {
            Some(mc) => mc.connection.clone(),
            None => {
                debug!("Reporter not connected for anchor probe");
                return Ok(false);
            }
        };

        let result = tokio::time::timeout(
            std::time::Duration::from_secs(20),
            async {
                let (mut send, mut recv) = reporter_conn.open_bi().await?;
                write_typed_message(&mut send, MessageType::AnchorProbeRequest, &payload).await?;
                send.finish()?;

                let msg_type = read_message_type(&mut recv).await?;
                if msg_type != MessageType::AnchorProbeResult {
                    anyhow::bail!("expected AnchorProbeResult, got {:?}", msg_type);
                }
                let result: AnchorProbeResultPayload = read_payload(&mut recv, 4096).await?;
                Ok::<_, anyhow::Error>(result)
            }
        ).await;

        match result {
            Ok(Ok(probe_result)) => {
                if probe_result.reachable {
                    self.last_probe_success_ms = now_ms();
                    self.probe_failure_streak = 0;
                    self.log_activity(ActivityLevel::Info, ActivityCategory::Anchor,
                        "Anchor probe succeeded — confirmed reachable".to_string(), None);
                    info!("Anchor probe succeeded — confirmed reachable");
                    Ok(true)
                } else {
                    self.probe_failure_streak += 1;
                    self.log_activity(ActivityLevel::Warn, ActivityCategory::Anchor,
                        format!("Anchor probe failed (streak: {})", self.probe_failure_streak), None);
                    warn!("Anchor probe failed (streak: {})", self.probe_failure_streak);
                    if self.probe_failure_streak >= 2 {
                        self.is_anchor.store(false, Ordering::Relaxed);
                        self.log_activity(ActivityLevel::Warn, ActivityCategory::Anchor,
                            "Anchor status revoked (2 consecutive failures)".to_string(), None);
                        warn!("Anchor status revoked (2 consecutive probe failures)");
                    }
                    Ok(false)
                }
            }
            Ok(Err(e)) => {
                debug!(error = %e, "Anchor probe communication error");
                Ok(false)
            }
            Err(_) => {
                self.probe_failure_streak += 1;
                self.log_activity(ActivityLevel::Warn, ActivityCategory::Anchor,
                    format!("Anchor probe timed out (streak: {})", self.probe_failure_streak), None);
                if self.probe_failure_streak >= 2 {
                    self.is_anchor.store(false, Ordering::Relaxed);
                    self.log_activity(ActivityLevel::Warn, ActivityCategory::Anchor,
                        "Anchor status revoked (2 consecutive failures)".to_string(), None);
                    warn!("Anchor status revoked (2 consecutive probe failures)");
                }
                Ok(false)
            }
        }
    }

    /// Handle an incoming AnchorProbeRequest — either as reporter (forward to witness) or as witness (cold connect)
    pub async fn handle_anchor_probe_request(
        &mut self,
        payload: crate::protocol::AnchorProbeRequestPayload,
        mut send: iroh::endpoint::SendStream,
    ) -> anyhow::Result<()> {
        use crate::protocol::{
            AnchorProbeResultPayload, MessageType, read_message_type, read_payload,
            write_typed_message,
        };

        let now = now_ms();

        // Dedup check
        if let Some(&seen_at) = self.seen_probes.get(&payload.probe_id) {
            if now - seen_at < 30_000 {
                let result = AnchorProbeResultPayload {
                    probe_id: payload.probe_id,
                    reachable: false,
                    observed_addr: None,
                };
                write_typed_message(&mut send, MessageType::AnchorProbeResult, &result).await?;
                send.finish()?;
                return Ok(());
            }
        }
        self.seen_probes.insert(payload.probe_id, now);

        // Are we the WITNESS? (witness == our_node_id)
        if payload.witness == self.our_node_id {
            info!(
                candidate = hex::encode(payload.candidate),
                target_addr = %payload.target_addr,
                "Anchor probe: we are witness, cold connecting"
            );

            // Raw QUIC connect to target_addr, 15s timeout. NO cascade, NO hole punch.
            let reachable = match payload.target_addr.parse::<std::net::SocketAddr>() {
                Ok(sock_addr) => {
                    let eid = match iroh::EndpointId::from_bytes(&payload.candidate) {
                        Ok(eid) => eid,
                        Err(_) => {
                            let result = AnchorProbeResultPayload {
                                probe_id: payload.probe_id,
                                reachable: false,
                                observed_addr: None,
                            };
                            write_typed_message(&mut send, MessageType::AnchorProbeResult, &result).await?;
                            send.finish()?;
                            return Ok(());
                        }
                    };
                    let addr = iroh::EndpointAddr::from(eid).with_ip_addr(sock_addr);
                    match tokio::time::timeout(
                        std::time::Duration::from_secs(15),
                        self.endpoint.connect(addr, ALPN_V2),
                    ).await {
                        Ok(Ok(_conn)) => true,
                        _ => false,
                    }
                }
                Err(_) => false,
            };

            let result = AnchorProbeResultPayload {
                probe_id: payload.probe_id,
                reachable,
                observed_addr: None,
            };

            // Send result back through the bi-stream (back through reporter)
            write_typed_message(&mut send, MessageType::AnchorProbeResult, &result).await?;
            send.finish()?;
            return Ok(());
        }

        // We are the REPORTER — forward to the witness
        if let Some(witness_conn) = self.connections.get(&payload.witness) {
            let witness_connection = witness_conn.connection.clone();
            let probe_id = payload.probe_id;

            // Forward the probe request to the witness
            let forward_result = tokio::time::timeout(
                std::time::Duration::from_secs(20),
                async {
                    let (mut w_send, mut w_recv) = witness_connection.open_bi().await?;
                    write_typed_message(&mut w_send, MessageType::AnchorProbeRequest, &payload).await?;
                    w_send.finish()?;
                    let msg_type = read_message_type(&mut w_recv).await?;
                    if msg_type != MessageType::AnchorProbeResult {
                        anyhow::bail!("expected AnchorProbeResult from witness, got {:?}", msg_type);
                    }
                    let result: AnchorProbeResultPayload = read_payload(&mut w_recv, 4096).await?;
                    Ok::<_, anyhow::Error>(result)
                }
            ).await;

            let result = match forward_result {
                Ok(Ok(r)) => r,
                _ => AnchorProbeResultPayload {
                    probe_id,
                    reachable: false,
                    observed_addr: None,
                },
            };

            write_typed_message(&mut send, MessageType::AnchorProbeResult, &result).await?;
            send.finish()?;
        } else {
            // Witness not connected to us
            let result = AnchorProbeResultPayload {
                probe_id: payload.probe_id,
                reachable: false,
                observed_addr: None,
            };
            write_typed_message(&mut send, MessageType::AnchorProbeResult, &result).await?;
            send.finish()?;
        }

        Ok(())
    }

    /// Build the stable advertised address for this anchor node.
    /// Returns None if not an anchor or no suitable address found.
    pub fn build_anchor_advertised_addr(&self) -> Option<String> {
        if !self.is_anchor.load(Ordering::Relaxed) {
            return None;
        }
        // Priority: UPnP external addr (has public IP + correct port)
        if let Some(ref ext) = self.upnp_external_addr {
            return Some(ext.to_string());
        }
        // If --bind was used, combine bind port with first public IP
        if let Some(bind) = self.bind_addr {
            let port = bind.port();
            for sa in self.endpoint.addr().ip_addrs() {
                if crate::network::is_publicly_routable(&sa) {
                    return Some(SocketAddr::new(sa.ip(), port).to_string());
                }
            }
        }
        // Fallback: use first publicly-routable bound address (e.g. public IPv6)
        for sa in self.endpoint.bound_sockets() {
            if crate::network::is_publicly_routable(&sa) {
                return Some(sa.to_string());
            }
        }
        for sa in self.endpoint.addr().ip_addrs() {
            if crate::network::is_publicly_routable(&sa) {
                return Some(sa.to_string());
            }
        }
        None
    }

    pub(crate) fn log_activity(&self, level: ActivityLevel, cat: ActivityCategory, msg: String, peer: Option<NodeId>) {
        if let Ok(mut log) = self.activity_log.try_lock() {
            log.log(level, cat, msg, peer);
        }
    }

    /// Public accessor for logging from static methods that hold a conn_mgr lock.
    pub fn log_activity_pub(&self, level: ActivityLevel, cat: ActivityCategory, msg: String, peer: Option<NodeId>) {
        self.log_activity(level, cat, msg, peer);
    }

    /// Set the growth loop signal sender.
    pub fn set_growth_tx(&mut self, tx: tokio::sync::mpsc::Sender<()>) {
        self.growth_tx = Some(tx);
    }

    /// Signal the growth loop to wake up and seek diverse peers.
    /// Non-blocking: drops the signal if the channel is full (coalescing).
    pub fn notify_growth(&self) {
        if let Some(tx) = &self.growth_tx {
            let _ = tx.try_send(());
        }
    }

    /// Set the recovery loop signal sender.
    pub fn set_recovery_tx(&mut self, tx: tokio::sync::mpsc::Sender<()>) {
        self.recovery_tx = Some(tx);
    }

    /// Signal the recovery loop when mesh is critically low.
    /// Non-blocking: drops the signal if the channel is full (coalescing).
    pub fn notify_recovery(&self) {
        if let Some(tx) = &self.recovery_tx {
            let _ = tx.try_send(());
        }
    }

    /// Score N2 candidates for growth loop diversity selection.
    /// Returns candidates sorted by diversity score (highest first), excluding
    /// already-connected peers, self, and unreachable peers.
    pub async fn score_n2_candidates(&self) -> Vec<(NodeId, f64)> {
        let candidates = {
            let storage = self.storage.lock().await;
            storage.score_n2_candidates_batch().unwrap_or_default()
        };

        let mut scored: Vec<(NodeId, f64)> = candidates
            .into_iter()
            .filter(|(nid, _, _)| {
                *nid != self.our_node_id
                    && !self.connections.contains_key(nid)
                    && !self.is_likely_unreachable(nid)
            })
            .map(|(nid, reporter_count, in_n3)| {
                let score = 1.0 / (reporter_count as f64)
                    + if !in_n3 { 0.3 } else { 0.0 };
                (nid, score)
            })
            .collect();

        scored.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
        scored
    }

    /// How many local slots are available for the growth loop to fill.
    pub fn available_local_slots(&self) -> usize {
        let local_count = self.count_kind(PeerSlotKind::Local);
        self.local_slots.saturating_sub(local_count)
    }

    /// Accept an incoming connection, returning false if slots are full.
    pub fn accept_connection(
        &mut self,
        conn: iroh::endpoint::Connection,
        remote_node_id: NodeId,
        remote_addr: Option<std::net::SocketAddr>,
    ) -> bool {
        if self.connections.contains_key(&remote_node_id) {
            debug!(peer = hex::encode(remote_node_id), "Replacing existing connection");
        }

        let total_slots = self.preferred_slots + self.local_slots + self.wide_slots;
        let total = self.connections.len();
        if total >= total_slots && !self.connections.contains_key(&remote_node_id) {
            debug!(peer = hex::encode(remote_node_id), "Slots full, rejecting");
            return false;
        }

        let now = now_ms();
        let slot_kind = if self.count_kind(PeerSlotKind::Local) < self.local_slots {
            PeerSlotKind::Local
        } else {
            PeerSlotKind::Wide
        };

        self.connections.insert(
            remote_node_id,
            MeshConnection {
                node_id: remote_node_id,
                connection: conn,
                slot_kind,
                connected_at: now,
                remote_addr: remote_addr.map(normalize_addr),
                last_activity: Arc::new(AtomicU64::new(now)),
            },
        );

        // If peer was on referral list and reconnected, clear disconnected_at
        self.mark_referral_reconnected(&remote_node_id);

        true
    }

    /// Register an already-established connection into the mesh.
    /// The QUIC connect must happen OUTSIDE the conn_mgr lock to avoid blocking
    /// all other tasks (diff cycle, accept loop, keepalive, rebalance) during
    /// the potentially long connection timeout.
    pub async fn register_connection(
        &mut self,
        peer_id: NodeId,
        conn: iroh::endpoint::Connection,
        addrs: &[std::net::SocketAddr],
        slot_kind: PeerSlotKind,
    ) {
        let now = now_ms();
        let connect_addr = addrs.first().copied();

        // Direct connect succeeded — mark peer as directly reachable
        self.mark_reachable(&peer_id);

        self.connections.insert(
            peer_id,
            MeshConnection {
                node_id: peer_id,
                connection: conn,
                slot_kind,
                connected_at: now,
                remote_addr: connect_addr.map(normalize_addr),
                last_activity: Arc::new(AtomicU64::new(now)),
            },
        );

        // If peer was on referral list and reconnected, clear disconnected_at
        self.mark_referral_reconnected(&peer_id);

        // Persist address to peers table so it survives restart
        if !addrs.is_empty() {
            let storage = self.storage.lock().await;
            let _ = storage.upsert_peer(&peer_id, addrs, None);
            drop(storage);
        }

        // Record in mesh_peers table + touch social route
        {
            let storage = self.storage.lock().await;
            let _ = storage.add_mesh_peer(&peer_id, slot_kind, 0);
            if storage.has_social_route(&peer_id).unwrap_or(false) {
                let _ = storage.touch_social_route_connect(&peer_id, addrs, ReachMethod::Direct);
                // Gather watcher data before dropping storage
                let watchers = storage.get_reconnect_watchers(&peer_id).unwrap_or_default();
                let route = storage.get_social_route(&peer_id).ok().flatten();
                let _ = storage.clear_reconnect_watchers(&peer_id);
                drop(storage);
                // Notify watchers asynchronously (no storage reference held)
                if !watchers.is_empty() {
                    if let Some(route) = route {
                        self.notify_watchers(watchers, &peer_id, &route).await;
                    }
                }
            }
        }

        info!(peer = hex::encode(peer_id), kind = %slot_kind, "Connected to peer");
    }

    /// Establish an outgoing mesh connection with a 15s timeout on the QUIC connect.
    /// Used by rebalance_slots() and reconnect_preferred() which hold &mut self.
    /// Note: holds the conn_mgr lock during the connect. For lock-free connecting,
    /// use Network::connect_to_peer() which connects outside the lock.
    pub async fn connect_to(
        &mut self,
        peer_id: NodeId,
        addr: iroh::EndpointAddr,
        slot_kind: PeerSlotKind,
    ) -> anyhow::Result<()> {
        if self.connections.contains_key(&peer_id) {
            return Ok(()); // Already connected
        }

        let addrs: Vec<std::net::SocketAddr> = addr.ip_addrs().copied().collect();
        if !addrs.is_empty() {
            let storage = self.storage.lock().await;
            let _ = storage.upsert_peer(&peer_id, &addrs, None);
        }

        // 15s timeout to limit lock contention (QUIC default can be 60+s)
        let conn = tokio::time::timeout(
            std::time::Duration::from_secs(15),
            self.endpoint.connect(addr, ALPN_V2),
        ).await
            .map_err(|_| anyhow::anyhow!("connect timed out (15s)"))?
            .map_err(|e| anyhow::anyhow!("connect failed: {e}"))?;

        self.register_connection(peer_id, conn, &addrs, slot_kind).await;
        Ok(())
    }

    /// Do the initial exchange after connecting: N1/N2 node lists + profile + deletes + peer addresses (both directions).
    pub async fn do_initial_exchange(
        &self,
        conn: &iroh::endpoint::Connection,
        remote_node_id: NodeId,
    ) -> anyhow::Result<()> {
        // Build our payload
        let our_payload = {
            let storage = self.storage.lock().await;
            let n1 = storage.build_n1_share()?;
            let n2 = storage.build_n2_share()?;
            let profile = storage.get_profile(&self.our_node_id)?;
            let deletes = storage.list_delete_records()?;
            let post_ids = storage.list_post_ids()?;
            let peer_addresses = storage.build_peer_addresses_for(&self.our_node_id)?;
            let our_profile = crate::types::NatProfile::from_nat_type(self.nat_type);
            InitialExchangePayload {
                n1_node_ids: n1,
                n2_node_ids: n2,
                profile,
                deletes,
                post_ids,
                peer_addresses,
                anchor_addr: self.build_anchor_advertised_addr(),
                your_observed_addr: None,
                nat_type: Some(self.nat_type.to_string()),
                nat_mapping: Some(our_profile.mapping.to_string()),
                nat_filtering: Some(our_profile.filtering.to_string()),
                http_capable: self.http_capable,
                http_addr: self.http_addr.clone(),
                device_role: None,
                cache_pressure: None,
            }
        };

        // Open bi-stream for initial exchange
        let (mut send, mut recv) = conn.open_bi().await?;

        // Send our payload
        write_typed_message(&mut send, MessageType::InitialExchange, &our_payload).await?;
        send.finish()?;

        // Read their payload
        let msg_type = read_message_type(&mut recv).await?;
        if msg_type != MessageType::InitialExchange {
            anyhow::bail!("expected InitialExchange, got {:?}", msg_type);
        }
        let their_payload: InitialExchangePayload = read_payload(&mut recv, MAX_PAYLOAD).await?;

        // Process their data
        let storage = self.storage.lock().await;

        // Their N1 → our N2 (tagged to this reporter)
        // Filter out our own ID and already-connected peers (they'd waste candidate slots)
        let filtered_n1: Vec<NodeId> = their_payload.n1_node_ids.iter()
            .filter(|nid| **nid != self.our_node_id && !self.connections.contains_key(*nid))
            .copied()
            .collect();
        storage.set_peer_n1(&remote_node_id, &filtered_n1)?;
        debug!(peer = hex::encode(remote_node_id), raw = their_payload.n1_node_ids.len(), stored = filtered_n1.len(), "Stored peer N1 as our N2 (filtered self+mesh)");

        // Their N2 → our N3 (tagged to this reporter)
        let filtered_n2: Vec<NodeId> = their_payload.n2_node_ids.iter()
            .filter(|nid| **nid != self.our_node_id)
            .copied()
            .collect();
        storage.set_peer_n2(&remote_node_id, &filtered_n2)?;
        debug!(peer = hex::encode(remote_node_id), raw = their_payload.n2_node_ids.len(), stored = filtered_n2.len(), "Stored peer N2 as our N3 (filtered self)");

        // Store their profile
        if let Some(profile) = their_payload.profile {
            let _ = storage.store_profile(&profile);
        }

        // Process delete records
        for dr in their_payload.deletes {
            if crypto::verify_delete_signature(&dr.author, &dr.post_id, &dr.signature) {
                let _ = storage.store_delete(&dr);
                let _ = storage.apply_delete(&dr);
            }
        }

        // Store their peer_addresses (N+10:Addresses)
        for pa in &their_payload.peer_addresses {
            if let Ok(nid) = crate::parse_node_id_hex(&pa.n) {
                let addrs: Vec<std::net::SocketAddr> = pa.a.iter().filter_map(|a| a.parse().ok()).collect();
                if !addrs.is_empty() {
                    let _ = storage.upsert_peer(&nid, &addrs, None);
                }
            }
        }

        // Record replicas from overlapping post_ids
        let our_post_ids: HashSet<[u8; 32]> = storage.list_post_ids()?.into_iter().collect();
        for pid in &their_payload.post_ids {
            if our_post_ids.contains(pid) {
                let _ = storage.record_replica(pid, &remote_node_id);
            }
        }

        // Process anchor's advertised address
        if let Some(ref anchor_addr_str) = their_payload.anchor_addr {
            if let Ok(sock) = anchor_addr_str.parse::<std::net::SocketAddr>() {
                let _ = storage.upsert_known_anchor(&remote_node_id, &[sock]);
                let _ = storage.upsert_peer(&remote_node_id, &[sock], None);
                info!(peer = hex::encode(remote_node_id), addr = %sock, "Stored anchor's advertised address");
            }
        }

        // Log observed address (STUN-like feedback)
        if let Some(ref observed) = their_payload.your_observed_addr {
            info!(observed_addr = %observed, reporter = hex::encode(remote_node_id), "Peer reports our address as");
        }

        // Store peer's NAT type
        if let Some(ref nat_str) = their_payload.nat_type {
            let nat = crate::types::NatType::from_str_label(nat_str);
            let _ = storage.set_peer_nat_type(&remote_node_id, nat);
        }

        // Store peer's NAT profile (mapping + filtering) if provided
        if their_payload.nat_mapping.is_some() || their_payload.nat_filtering.is_some() {
            let mapping = their_payload.nat_mapping.as_deref()
                .map(crate::types::NatMapping::from_str_label)
                .unwrap_or(crate::types::NatMapping::Unknown);
            let filtering = their_payload.nat_filtering.as_deref()
                .map(crate::types::NatFiltering::from_str_label)
                .unwrap_or(crate::types::NatFiltering::Unknown);
            let profile = crate::types::NatProfile::new(mapping, filtering);
            let _ = storage.set_peer_nat_profile(&remote_node_id, &profile);
        }

        Ok(())
    }

    /// Handle the responder side of initial exchange.
    pub async fn handle_initial_exchange(
        &self,
        mut send: iroh::endpoint::SendStream,
        mut recv: iroh::endpoint::RecvStream,
        remote_node_id: NodeId,
    ) -> anyhow::Result<()> {
        // Read their payload (message type byte already consumed by caller)
        let their_payload: InitialExchangePayload = read_payload(&mut recv, MAX_PAYLOAD).await?;

        // Build and send our payload
        let our_payload = {
            let storage = self.storage.lock().await;
            let n1 = storage.build_n1_share()?;
            let n2 = storage.build_n2_share()?;
            let profile = storage.get_profile(&self.our_node_id)?;
            let deletes = storage.list_delete_records()?;
            let post_ids = storage.list_post_ids()?;
            let peer_addresses = storage.build_peer_addresses_for(&self.our_node_id)?;
            let our_profile = crate::types::NatProfile::from_nat_type(self.nat_type);
            InitialExchangePayload {
                n1_node_ids: n1,
                n2_node_ids: n2,
                profile,
                deletes,
                post_ids,
                peer_addresses,
                anchor_addr: self.build_anchor_advertised_addr(),
                your_observed_addr: None, // no remote_addr available in this method
                nat_type: Some(self.nat_type.to_string()),
                nat_mapping: Some(our_profile.mapping.to_string()),
                nat_filtering: Some(our_profile.filtering.to_string()),
                http_capable: self.http_capable,
                http_addr: self.http_addr.clone(),
                device_role: None,
                cache_pressure: None,
            }
        };

        write_typed_message(&mut send, MessageType::InitialExchange, &our_payload).await?;
        send.finish()?;

        // Process their data
        let storage = self.storage.lock().await;

        // Their N1 → our N2 (filter out self + already-connected peers)
        let filtered_n1: Vec<NodeId> = their_payload.n1_node_ids.iter()
            .filter(|nid| **nid != self.our_node_id && !self.connections.contains_key(*nid))
            .copied()
            .collect();
        storage.set_peer_n1(&remote_node_id, &filtered_n1)?;
        debug!(peer = hex::encode(remote_node_id), raw = their_payload.n1_node_ids.len(), stored = filtered_n1.len(), "Stored peer N1 as our N2 (filtered self+mesh)");

        // Their N2 → our N3 (filter out self)
        let filtered_n2: Vec<NodeId> = their_payload.n2_node_ids.iter()
            .filter(|nid| **nid != self.our_node_id)
            .copied()
            .collect();
        storage.set_peer_n2(&remote_node_id, &filtered_n2)?;

        if let Some(profile) = their_payload.profile {
            let _ = storage.store_profile(&profile);
        }

        for dr in their_payload.deletes {
            if crypto::verify_delete_signature(&dr.author, &dr.post_id, &dr.signature) {
                let _ = storage.store_delete(&dr);
                let _ = storage.apply_delete(&dr);
            }
        }

        // Store their peer_addresses (N+10:Addresses)
        for pa in &their_payload.peer_addresses {
            if let Ok(nid) = crate::parse_node_id_hex(&pa.n) {
                let addrs: Vec<std::net::SocketAddr> = pa.a.iter().filter_map(|a| a.parse().ok()).collect();
                if !addrs.is_empty() {
                    let _ = storage.upsert_peer(&nid, &addrs, None);
                }
            }
        }

        let our_post_ids: HashSet<[u8; 32]> = storage.list_post_ids()?.into_iter().collect();
        for pid in &their_payload.post_ids {
            if our_post_ids.contains(pid) {
                let _ = storage.record_replica(pid, &remote_node_id);
            }
        }

        Ok(())
    }

    /// Compute N1/N2 changes and broadcast NodeListUpdate to all connected peers.
    pub async fn broadcast_routing_diff(&mut self) -> anyhow::Result<usize> {
        let seq = self.diff_seq.fetch_add(1, Ordering::Relaxed) + 1;

        let (current_n1, current_n2) = {
            let storage = self.storage.lock().await;
            let n1: HashSet<NodeId> = storage.build_n1_share()?.into_iter().collect();
            let n2: HashSet<NodeId> = storage.build_n2_share()?.into_iter().collect();
            (n1, n2)
        };

        let n1_added: Vec<NodeId> = current_n1.difference(&self.last_n1_set).copied().collect();
        let n1_removed: Vec<NodeId> = self.last_n1_set.difference(&current_n1).copied().collect();
        let n2_added: Vec<NodeId> = current_n2.difference(&self.last_n2_set).copied().collect();
        let n2_removed: Vec<NodeId> = self.last_n2_set.difference(&current_n2).copied().collect();

        if n1_added.is_empty() && n1_removed.is_empty() && n2_added.is_empty() && n2_removed.is_empty() {
            return Ok(0);
        }

        let payload = NodeListUpdatePayload {
            seq,
            n1_added,
            n1_removed,
            n2_added,
            n2_removed,
        };
        let mut sent_count = 0;

        for (peer_id, pc) in &self.connections {
            let result = async {
                let mut send = pc.connection.open_uni().await?;
                write_typed_message(&mut send, MessageType::NodeListUpdate, &payload).await?;
                send.finish()?;
                anyhow::Ok(())
            }
            .await;

            match result {
                Ok(()) => sent_count += 1,
                Err(e) => {
                    debug!(
                        peer = hex::encode(peer_id),
                        error = %e,
                        "Failed to send node list update"
                    );
                }
            }
        }

        // Update last sets
        self.last_n1_set = current_n1;
        self.last_n2_set = current_n2;

        Ok(sent_count)
    }

    /// Process a node list update from a peer: their N1 changes → our N2, their N2 changes → our N3.
    pub async fn process_routing_diff(
        &self,
        reporter: &NodeId,
        diff: NodeListUpdatePayload,
    ) -> anyhow::Result<usize> {
        let storage = self.storage.lock().await;
        let mut count = 0;

        // Their N1 added → add to our N2 (filter self + already-connected)
        if !diff.n1_added.is_empty() {
            let filtered: Vec<NodeId> = diff.n1_added.iter()
                .filter(|nid| **nid != self.our_node_id && !self.connections.contains_key(*nid))
                .copied()
                .collect();
            if !filtered.is_empty() {
                storage.add_peer_n1(reporter, &filtered)?;
            }
            count += filtered.len();
        }

        // Their N1 removed → remove from our N2
        if !diff.n1_removed.is_empty() {
            storage.remove_peer_n1(reporter, &diff.n1_removed)?;
            count += diff.n1_removed.len();
        }

        // Their N2 added → add to our N3 (filter self)
        if !diff.n2_added.is_empty() {
            let filtered: Vec<NodeId> = diff.n2_added.iter()
                .filter(|nid| **nid != self.our_node_id)
                .copied()
                .collect();
            if !filtered.is_empty() {
                storage.add_peer_n2(reporter, &filtered)?;
            }
            count += filtered.len();
        }

        // Their N2 removed → remove from our N3
        if !diff.n2_removed.is_empty() {
            storage.remove_peer_n2(reporter, &diff.n2_removed)?;
            count += diff.n2_removed.len();
        }

        // Update last_diff_seq
        let _ = storage.update_mesh_peer_seq(reporter, diff.seq);

        Ok(count)
    }

    /// Broadcast a visibility update to all connected peers.
    pub async fn broadcast_visibility_update(
        &self,
        update: &crate::types::VisibilityUpdate,
    ) -> usize {
        let payload = VisibilityUpdatePayload {
            updates: vec![update.clone()],
        };

        let mut sent = 0;
        for (peer_id, pc) in &self.connections {
            let result = async {
                let mut send = pc.connection.open_uni().await?;
                write_typed_message(&mut send, MessageType::VisibilityUpdate, &payload).await?;
                send.finish()?;
                anyhow::Ok(())
            }
            .await;

            match result {
                Ok(()) => sent += 1,
                Err(e) => {
                    debug!(peer = hex::encode(peer_id), error = %e, "Failed to push visibility update");
                }
            }
        }

        sent
    }

    /// Handle an incoming post notification: if we follow the author, pull the post.
    /// `conn` is a fallback connection for ephemeral callers (not persistently connected).
    pub async fn handle_post_notification(
        &self,
        from: &NodeId,
        notification: PostNotificationPayload,
        conn: Option<&iroh::endpoint::Connection>,
    ) -> anyhow::Result<bool> {
        let dominated = {
            let storage = self.storage.lock().await;
            // Already have this post?
            if storage.get_post(&notification.post_id)?.is_some() {
                return Ok(false);
            }
            // Do we follow the author?
            let follows = storage.list_follows()?;
            follows.contains(&notification.author)
        };

        if !dominated {
            return Ok(false);
        }

        // We follow the author and don't have the post — pull it from the notifier
        let pull_conn = match self.connections.get(from) {
            Some(pc) => pc.connection.clone(),
            None => match conn {
                Some(c) => c.clone(),
                None => return Ok(false),
            },
        };

        let (our_follows, follows_sync) = {
            let storage = self.storage.lock().await;
            (
                storage.list_follows()?,
                storage.get_follows_with_last_sync().unwrap_or_default(),
            )
        };

        let (mut send, mut recv) = pull_conn.open_bi().await?;
        let request = PullSyncRequestPayload {
            follows: our_follows,
            have_post_ids: vec![],  // v4: empty, using since_ms instead
            since_ms: follows_sync,
        };
        write_typed_message(&mut send, MessageType::PullSyncRequest, &request).await?;
        send.finish()?;

        let _resp_type = read_message_type(&mut recv).await?;
        let response: PullSyncResponsePayload =
            read_payload(&mut recv, MAX_PAYLOAD).await?;

        let now_ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis() as u64;
        let mut stored = false;
        let mut new_post_ids: Vec<PostId> = Vec::new();
        let mut synced_authors: HashSet<NodeId> = HashSet::new();

        // Brief lock 1: store posts
        {
            let storage = self.storage.lock().await;
            for sp in &response.posts {
                if verify_post_id(&sp.id, &sp.post) && !storage.is_deleted(&sp.id)? {
                    let _ = storage.store_post_with_visibility(&sp.id, &sp.post, &sp.visibility);
                    new_post_ids.push(sp.id);
                    synced_authors.insert(sp.post.author);
                    if sp.id == notification.post_id {
                        stored = true;
                    }
                }
            }
        }
        // Lock RELEASED

        // Brief lock 2: upstream + last_sync + visibility updates
        {
            let storage = self.storage.lock().await;
            for pid in &new_post_ids {
                let prio = storage.get_post_upstreams(pid).map(|v| v.len() as u8).unwrap_or(0);
                let _ = storage.add_post_upstream(pid, from, prio);
            }
            for author in &synced_authors {
                let _ = storage.update_follow_last_sync(author, now_ms);
            }
            for vu in &response.visibility_updates {
                if let Some(post) = storage.get_post(&vu.post_id)? {
                    if post.author == vu.author {
                        let _ = storage.update_post_visibility(&vu.post_id, &vu.visibility);
                    }
                }
            }
        }

        // Register as downstream for new posts (cap at 50 to avoid flooding)
        if !new_post_ids.is_empty() {
            let reg_conn = pull_conn.clone();
            tokio::spawn(async move {
                for post_id in new_post_ids.into_iter().take(50) {
                    let payload = PostDownstreamRegisterPayload { post_id };
                    if let Ok(mut send) = reg_conn.open_uni().await {
                        let _ = write_typed_message(&mut send, MessageType::PostDownstreamRegister, &payload).await;
                        let _ = send.finish();
                    }
                }
            });
        }

        Ok(stored)
    }

    /// Pull posts from a connected peer.
    pub async fn pull_from_peer(&self, peer_id: &NodeId) -> anyhow::Result<PullSyncStats> {
        let pc = self
            .connections
            .get(peer_id)
            .ok_or_else(|| anyhow::anyhow!("not connected to {}", hex::encode(peer_id)))?;

        let (our_follows, follows_sync) = {
            let storage = self.storage.lock().await;
            (
                storage.list_follows()?,
                storage.get_follows_with_last_sync().unwrap_or_default(),
            )
        };

        let request = PullSyncRequestPayload {
            follows: our_follows,
            have_post_ids: vec![],  // v4: empty, using since_ms instead
            since_ms: follows_sync,
        };

        let (mut send, mut recv) = pc.connection.open_bi().await?;
        write_typed_message(&mut send, MessageType::PullSyncRequest, &request).await?;
        send.finish()?;

        let msg_type = read_message_type(&mut recv).await?;
        if msg_type != MessageType::PullSyncResponse {
            anyhow::bail!("expected PullSyncResponse, got {:?}", msg_type);
        }
        let response: PullSyncResponsePayload = read_payload(&mut recv, MAX_PAYLOAD).await?;

        let mut posts_received = 0;
        let mut vis_updates = 0;
        let mut new_post_ids: Vec<PostId> = Vec::new();
        let now_ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis() as u64;

        // Brief lock 1: store posts
        let mut synced_authors: HashSet<NodeId> = HashSet::new();
        {
            let storage = self.storage.lock().await;
            for sp in &response.posts {
                if storage.is_deleted(&sp.id)? {
                    continue;
                }
                if verify_post_id(&sp.id, &sp.post) {
                    if storage.store_post_with_visibility(&sp.id, &sp.post, &sp.visibility)? {
                        new_post_ids.push(sp.id);
                        posts_received += 1;
                    }
                    synced_authors.insert(sp.post.author);
                }
            }
        }
        // Lock RELEASED

        // Brief lock 2: upstream + last_sync + visibility updates
        {
            let storage = self.storage.lock().await;
            for pid in &new_post_ids {
                let prio = storage.get_post_upstreams(pid).map(|v| v.len() as u8).unwrap_or(0);
                let _ = storage.add_post_upstream(pid, peer_id, prio);
            }
            for author in &synced_authors {
                let _ = storage.update_follow_last_sync(author, now_ms);
            }
            for vu in response.visibility_updates {
                if vu.author != *peer_id {
                    // Only accept visibility updates authored by the responding peer
                    // (or their forwarded data — but for now, only from the peer)
                }
                if let Some(post) = storage.get_post(&vu.post_id)? {
                    if post.author == vu.author {
                        if storage.update_post_visibility(&vu.post_id, &vu.visibility)? {
                            vis_updates += 1;
                        }
                    }
                }
            }
        }

        // Register as downstream with the sender for new posts (cap at 50 to avoid flooding)
        if !new_post_ids.is_empty() {
            let conn = pc.connection.clone();
            tokio::spawn(async move {
                for post_id in new_post_ids.into_iter().take(50) {
                    let payload = PostDownstreamRegisterPayload { post_id };
                    if let Ok(mut send) = conn.open_uni().await {
                        let _ = write_typed_message(&mut send, MessageType::PostDownstreamRegister, &payload).await;
                        let _ = send.finish();
                    }
                }
            });
        }

        Ok(PullSyncStats {
            posts_received,
            visibility_updates: vis_updates,
        })
    }

    /// Fetch engagement headers (reactions, comments, policies) for posts due for check from a peer.
    /// Uses tiered check rates: active posts checked more often, cold posts less frequently.
    pub async fn fetch_engagement_from_peer(&self, peer_id: &NodeId) -> anyhow::Result<usize> {
        let pc = self
            .connections
            .get(peer_id)
            .ok_or_else(|| anyhow::anyhow!("not connected to {}", hex::encode(peer_id)))?;

        // Brief lock: gather only posts DUE for engagement check (tiered frequency)
        let post_headers: Vec<([u8; 32], u64)> = {
            let storage = self.storage.lock().await;
            let due_ids = storage.get_posts_due_for_engagement_check()?;
            due_ids
                .into_iter()
                .map(|pid| {
                    let ts = storage
                        .get_blob_header(&pid)
                        .ok()
                        .flatten()
                        .map(|(_, ts)| ts)
                        .unwrap_or(0);
                    (pid, ts)
                })
                .collect()
        };
        // Lock RELEASED — all network I/O happens without the lock

        let now_ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis() as u64;

        let mut updated = 0;
        // Request headers in batches to avoid opening too many streams
        for chunk in post_headers.chunks(20) {
            // Collect all results for this chunk WITHOUT holding the lock
            let mut results: Vec<([u8; 32], Option<(String, crate::types::BlobHeader)>)> = Vec::new();
            for (post_id, current_ts) in chunk {
                let result: anyhow::Result<Option<(String, crate::types::BlobHeader)>> = async {
                    let (mut send, mut recv) = pc.connection.open_bi().await?;
                    let request = BlobHeaderRequestPayload {
                        post_id: *post_id,
                        current_updated_at: *current_ts,
                    };
                    write_typed_message(&mut send, MessageType::BlobHeaderRequest, &request).await?;
                    send.finish()?;

                    let msg_type = read_message_type(&mut recv).await?;
                    if msg_type != MessageType::BlobHeaderResponse {
                        anyhow::bail!("expected BlobHeaderResponse");
                    }
                    let response: BlobHeaderResponsePayload =
                        read_payload(&mut recv, MAX_PAYLOAD).await?;

                    if response.updated {
                        if let Some(json) = response.header_json {
                            if let Ok(header) = serde_json::from_str::<crate::types::BlobHeader>(&json) {
                                return Ok(Some((json, header)));
                            }
                        }
                    }
                    Ok(None)
                }
                .await;
                match result {
                    Ok(header_opt) => results.push((*post_id, header_opt)),
                    Err(e) => {
                        trace!(post_id = hex::encode(post_id), error = %e, "Failed to fetch engagement header");
                    }
                }
            }

            // Single lock for ALL writes in this chunk
            if !results.is_empty() {
                let storage = self.storage.lock().await;
                for (post_id, header_opt) in &results {
                    let _ = storage.update_post_last_check(post_id, now_ms);
                    if let Some((json, header)) = header_opt {
                        let _ = storage.store_blob_header(
                            &header.post_id,
                            &header.author,
                            json,
                            header.updated_at,
                        );
                        // store_reaction / store_comment are tombstone-aware:
                        // they compare timestamps and respect deleted_at fields.
                        for reaction in &header.reactions {
                            let _ = storage.store_reaction(reaction);
                        }
                        for comment in &header.comments {
                            let _ = storage.store_comment(comment);
                        }
                        let _ = storage.set_comment_policy(&header.post_id, &header.policy);
                        let _ = storage.update_post_last_engagement(post_id, now_ms);
                        updated += 1;
                    }
                }
                drop(storage);
            }
            // Lock RELEASED before next chunk
        }

        Ok(updated)
    }

    /// Handle an incoming pull request from a peer.
    pub async fn handle_pull_request(
        &self,
        remote_node_id: NodeId,
        mut recv: iroh::endpoint::RecvStream,
        mut send: iroh::endpoint::SendStream,
    ) -> anyhow::Result<()> {
        let request: PullSyncRequestPayload = read_payload(&mut recv, MAX_PAYLOAD).await?;

        let their_follows: HashSet<NodeId> = request.follows.into_iter().collect();
        let their_post_ids: HashSet<[u8; 32]> = request.have_post_ids.into_iter().collect();

        // Protocol v4: build per-author since_ms lookup
        let since_ms_map: HashMap<NodeId, u64> = request.since_ms.into_iter().collect();
        let use_since_ms = !since_ms_map.is_empty();

        // Phase 1: Brief lock — load data
        let (all_posts, group_members) = {
            let storage = self.storage.lock().await;
            let posts = storage.list_posts_with_visibility()?;
            let members = storage.get_all_group_members().unwrap_or_default();
            (posts, members)
        };
        // Lock RELEASED

        // Phase 2: Filter without lock (pure CPU)
        let mut candidates_to_send = Vec::new();
        let mut vis_updates_to_send = Vec::new();

        for (id, post, visibility) in all_posts {
            let should_send =
                crate::network::should_send_post(&post, &visibility, &remote_node_id, &their_follows, &group_members);

            if !should_send {
                continue;
            }

            // Determine if peer already has this post
            let peer_has_post = if use_since_ms {
                // v4 path: filter by per-author timestamp (60s fudge for clock skew)
                if let Some(&since) = since_ms_map.get(&post.author) {
                    post.timestamp_ms <= since + 60_000
                } else {
                    false // no since_ms for this author — they want everything
                }
            } else {
                // Legacy path: use have_post_ids
                their_post_ids.contains(&id)
            };

            if !peer_has_post {
                candidates_to_send.push((id, post, visibility));
            } else {
                // They already have the post — send visibility update if we authored it
                if post.author == self.our_node_id {
                    vis_updates_to_send.push(crate::types::VisibilityUpdate {
                        post_id: id,
                        author: self.our_node_id,
                        visibility,
                    });
                }
            }
        }

        // Phase 3: Brief re-lock for is_deleted checks on filtered posts
        let (posts, vis_updates) = {
            let storage = self.storage.lock().await;
            let posts_to_send: Vec<SyncPost> = candidates_to_send.into_iter()
                .filter(|(id, _, _)| !storage.is_deleted(id).unwrap_or(false))
                .map(|(id, post, visibility)| SyncPost { id, post, visibility })
                .collect();
            (posts_to_send, vis_updates_to_send)
        };

        let response = PullSyncResponsePayload {
            posts,
            visibility_updates: vis_updates,
        };

        write_typed_message(&mut send, MessageType::PullSyncResponse, &response).await?;
        send.finish()?;

        Ok(())
    }

    /// Handle an address resolution request: check connections, social routes, then peer records.
    /// If target is disconnected, register the requester as a watcher.
    pub async fn handle_address_request(
        &self,
        mut recv: iroh::endpoint::RecvStream,
        mut send: iroh::endpoint::SendStream,
        requester: NodeId,
    ) -> anyhow::Result<()> {
        let req: crate::protocol::AddressRequestPayload =
            read_payload(&mut recv, MAX_PAYLOAD).await?;

        // Check if target is directly connected to us
        if let Some(_pc) = self.connections.get(&req.target) {
            let storage = self.storage.lock().await;
            let addr = storage.get_peer_record(&req.target)?
                .and_then(|r| r.addresses.first().map(|a| a.to_string()));
            let response = crate::protocol::AddressResponsePayload {
                target: req.target,
                address: addr,
                disconnected_at: None,
                peer_addresses: vec![],
            };
            write_typed_message(&mut send, MessageType::AddressResponse, &response).await?;
            send.finish()?;
            return Ok(());
        }

        let storage = self.storage.lock().await;

        // Check social routes (richer info)
        if let Some(route) = storage.get_social_route(&req.target)? {
            match route.status {
                SocialStatus::Online => {
                    let addr = route.addresses.first().map(|a| a.to_string());
                    let response = crate::protocol::AddressResponsePayload {
                        target: req.target,
                        address: addr,
                        disconnected_at: None,
                        peer_addresses: route.peer_addresses.clone(),
                    };
                    write_typed_message(&mut send, MessageType::AddressResponse, &response).await?;
                    send.finish()?;
                    return Ok(());
                }
                SocialStatus::Disconnected => {
                    let _ = storage.add_reconnect_watcher(&req.target, &requester);
                    let response = crate::protocol::AddressResponsePayload {
                        target: req.target,
                        address: None,
                        disconnected_at: Some(route.last_seen_ms),
                        peer_addresses: route.peer_addresses.clone(),
                    };
                    write_typed_message(&mut send, MessageType::AddressResponse, &response).await?;
                    send.finish()?;
                    return Ok(());
                }
            }
        }

        // Fall back to peer record
        let address = storage.get_peer_record(&req.target)?
            .and_then(|r| r.addresses.first().map(|a| a.to_string()));

        let response = crate::protocol::AddressResponsePayload {
            target: req.target,
            address,
            disconnected_at: None,
            peer_addresses: vec![],
        };
        write_typed_message(&mut send, MessageType::AddressResponse, &response).await?;
        send.finish()?;

        Ok(())
    }

    /// Resolve a peer's address using connections, social routes, and N2/N3 referral chain.
    pub async fn resolve_address(&self, target: &NodeId) -> anyhow::Result<Option<String>> {
        // Check if target is directly connected
        if self.connections.contains_key(target) {
            let storage = self.storage.lock().await;
            return Ok(storage.get_peer_record(target)?
                .and_then(|r| r.addresses.first().map(|a| a.to_string())));
        }

        // Check social routes
        {
            let storage = self.storage.lock().await;
            if let Some(route) = storage.get_social_route(target)? {
                if route.status == SocialStatus::Online {
                    if let Some(addr) = route.addresses.first() {
                        return Ok(Some(addr.to_string()));
                    }
                }
            }
        }

        // N2 lookup: ask tagged reporter for address
        let n2_reporters = {
            let storage = self.storage.lock().await;
            storage.find_in_n2(target)?
        };
        for reporter in &n2_reporters {
            if let Some(pc) = self.connections.get(reporter) {
                let result = async {
                    let (mut send, mut recv) = pc.connection.open_bi().await?;
                    let req = crate::protocol::AddressRequestPayload { target: *target };
                    write_typed_message(&mut send, MessageType::AddressRequest, &req).await?;
                    send.finish()?;
                    let _resp_type = read_message_type(&mut recv).await?;
                    let resp: crate::protocol::AddressResponsePayload =
                        read_payload(&mut recv, MAX_PAYLOAD).await?;
                    anyhow::Ok(resp.address)
                }.await;
                if let Ok(Some(addr)) = result {
                    return Ok(Some(addr));
                }
            }
        }

        // N3 lookup: ask tagged reporter (chains one more hop)
        let n3_reporters = {
            let storage = self.storage.lock().await;
            storage.find_in_n3(target)?
        };
        for reporter in &n3_reporters {
            if let Some(pc) = self.connections.get(reporter) {
                let result = async {
                    let (mut send, mut recv) = pc.connection.open_bi().await?;
                    let req = crate::protocol::AddressRequestPayload { target: *target };
                    write_typed_message(&mut send, MessageType::AddressRequest, &req).await?;
                    send.finish()?;
                    let _resp_type = read_message_type(&mut recv).await?;
                    let resp: crate::protocol::AddressResponsePayload =
                        read_payload(&mut recv, MAX_PAYLOAD).await?;
                    anyhow::Ok(resp.address)
                }.await;
                if let Ok(Some(addr)) = result {
                    return Ok(Some(addr));
                }
            }
        }

        Ok(None)
    }

    /// Resolve a peer's address from all local sources (no network requests).
    /// Checks: peers table → social route cache.
    pub async fn resolve_peer_addr_local(&self, peer_id: &NodeId) -> Option<iroh::EndpointAddr> {
        let endpoint_id = iroh::EndpointId::from_bytes(peer_id).ok()?;
        let storage = self.storage.lock().await;

        // 1. Peers table
        if let Ok(Some(rec)) = storage.get_peer_record(peer_id) {
            if let Some(addr) = rec.addresses.first() {
                return Some(iroh::EndpointAddr::from(endpoint_id).with_ip_addr(*addr));
            }
        }
        // 2. Social route cache
        if let Ok(Some(route)) = storage.get_social_route(peer_id) {
            if let Some(addr) = route.addresses.first() {
                return Some(iroh::EndpointAddr::from(endpoint_id).with_ip_addr(*addr));
            }
        }
        None
    }

    /// Pick a random connected peer with a known address (for RefuseRedirect).
    pub async fn pick_random_redirect_peer(&self, exclude: &NodeId) -> Option<PeerWithAddress> {
        let candidates: Vec<NodeId> = self.connections.keys()
            .filter(|nid| *nid != exclude && **nid != self.our_node_id)
            .copied()
            .collect();
        if candidates.is_empty() {
            return None;
        }
        let storage = self.storage.lock().await;
        for nid in &candidates {
            if let Ok(Some(rec)) = storage.get_peer_record(nid) {
                if let Some(addr) = rec.addresses.first() {
                    return Some(PeerWithAddress {
                        n: hex::encode(nid),
                        a: vec![addr.to_string()],
                    });
                }
            }
        }
        None
    }

    // ---- Worm lookup (wide-bloom + 11-needle) ----

    /// Initiate a worm content search — find a post (and optionally its author node).
    /// Uses the extended worm with post_id/blob_id fields.
    pub async fn initiate_content_search(
        &self,
        target: &NodeId,
        post_id: Option<PostId>,
        blob_id: Option<[u8; 32]>,
    ) -> anyhow::Result<Option<WormResult>> {
        // Gather needle_peers: target's recent_peers from stored profile (up to 10)
        let needle_peers: Vec<NodeId> = {
            let storage = self.storage.lock().await;
            let mut rp = storage.get_recent_peers(target)?;
            rp.truncate(10);
            rp
        };

        let mut all_needles = vec![*target];
        all_needles.extend_from_slice(&needle_peers);

        let worm_id: WormId = rand::random();
        let visited = vec![self.our_node_id];

        let result = tokio::time::timeout(
            std::time::Duration::from_millis(WORM_TOTAL_TIMEOUT_MS),
            self.originator_worm_cascade(target, &needle_peers, &worm_id, &visited, &all_needles, post_id, blob_id),
        )
        .await;

        match result {
            Ok(Ok(Some(wr))) => Ok(Some(wr)),
            Ok(Ok(None)) => Ok(None),
            Ok(Err(e)) => {
                debug!(error = %e, "Content search failed");
                Ok(None)
            }
            Err(_) => {
                debug!("Content search timed out");
                Ok(None)
            }
        }
    }

    /// Initiate a worm lookup using the wide-bloom + 11-needle algorithm.
    /// Called by the originator node. Returns WormResult if found.
    pub async fn initiate_worm_lookup(&self, target: &NodeId) -> anyhow::Result<Option<WormResult>> {
        // Check cooldown
        {
            let storage = self.storage.lock().await;
            if storage.is_worm_cooldown(target, WORM_COOLDOWN_MS)? {
                debug!(target = hex::encode(target), "Worm lookup on cooldown");
                return Ok(None);
            }
        }

        // Gather needle_peers: target's recent_peers from stored profile (up to 10)
        let needle_peers: Vec<NodeId> = {
            let storage = self.storage.lock().await;
            let mut rp = storage.get_recent_peers(target)?;
            rp.truncate(10);
            rp
        };

        // All 11 IDs = [target] + needle_peers
        let mut all_needles = vec![*target];
        all_needles.extend_from_slice(&needle_peers);

        // Generate worm_id
        let worm_id: WormId = rand::random();
        let visited = vec![self.our_node_id];

        // Run with total timeout
        let result = tokio::time::timeout(
            std::time::Duration::from_millis(WORM_TOTAL_TIMEOUT_MS),
            self.originator_worm_cascade(target, &needle_peers, &worm_id, &visited, &all_needles, None, None),
        )
        .await;

        match result {
            Ok(Ok(Some(wr))) => Ok(Some(wr)),
            Ok(Ok(None)) => {
                let storage = self.storage.lock().await;
                let _ = storage.record_worm_miss(target);
                Ok(None)
            }
            Ok(Err(e)) => {
                debug!(target = hex::encode(target), error = %e, "Worm lookup failed");
                let storage = self.storage.lock().await;
                let _ = storage.record_worm_miss(target);
                Err(e)
            }
            Err(_) => {
                debug!(target = hex::encode(target), "Worm lookup timed out");
                let storage = self.storage.lock().await;
                let _ = storage.record_worm_miss(target);
                Ok(None)
            }
        }
    }

    /// Originator-driven worm cascade: local check → fan-out → wide-bloom.
    async fn originator_worm_cascade(
        &self,
        target: &NodeId,
        needle_peers: &[NodeId],
        worm_id: &WormId,
        visited: &[NodeId],
        all_needles: &[NodeId],
        post_id: Option<PostId>,
        blob_id: Option<[u8; 32]>,
    ) -> anyhow::Result<Option<WormResult>> {
        // Step 0: Local check — find any of the 11 IDs in our connections, N2, or N3
        {
            // Check direct connections first
            for needle in all_needles {
                if self.connections.contains_key(needle) {
                    let storage = self.storage.lock().await;
                    let addr = storage.get_peer_record(needle)?
                        .and_then(|r| r.addresses.first().map(|a| a.to_string()));
                    drop(storage);
                    return Ok(Some(WormResult {
                        node_id: *needle,
                        addresses: addr.into_iter().collect(),
                        reporter: self.our_node_id,
                        freshness_ms: 0,
                        post_holder: None,
                        blob_holder: None,
                    }));
                }
            }

            // Check N2/N3
            let storage = self.storage.lock().await;
            let found_entries = storage.find_any_in_n2_n3(all_needles)?;
            if let Some((found_id, _reporter, _level)) = found_entries.first() {
                drop(storage);
                let address = self.resolve_address(found_id).await.unwrap_or(None);
                if let Some(addr) = address {
                    return Ok(Some(WormResult {
                        node_id: *found_id,
                        addresses: vec![addr],
                        reporter: self.our_node_id,
                        freshness_ms: 0,
                        post_holder: None,
                        blob_holder: None,
                    }));
                }
                return Ok(Some(WormResult {
                    node_id: *found_id,
                    addresses: vec![],
                    reporter: self.our_node_id,
                    freshness_ms: 0,
                    post_holder: None,
                    blob_holder: None,
                }));
            }
        }

        // Step 2: Fan-out — send WormQuery{ttl=0} to all connected peers, collect ALL responses
        let peer_conns: Vec<(NodeId, iroh::endpoint::Connection)> = {
            let visited_set: HashSet<NodeId> = visited.iter().copied().collect();
            self.connections
                .iter()
                .filter(|(nid, _)| !visited_set.contains(*nid))
                .map(|(nid, pc)| (*nid, pc.connection.clone()))
                .collect()
        };

        if !peer_conns.is_empty() {
            let fan_out_payload = WormQueryPayload {
                worm_id: *worm_id,
                target: *target,
                needle_peers: needle_peers.to_vec(),
                ttl: 0,
                visited: visited.to_vec(),
                post_id: post_id.clone(),
                blob_id,
            };

            let (hit, wide_referrals) = tokio::time::timeout(
                std::time::Duration::from_millis(WORM_FAN_OUT_TIMEOUT_MS),
                Self::fan_out_worm_query_all(&peer_conns, &fan_out_payload),
            )
            .await
            .unwrap_or((None, vec![]));

            if let Some(wr) = hit {
                return Ok(Some(wr));
            }

            // Step 3: Wide-bloom — connect to referred wide peers, send WormQuery{ttl=1}
            if !wide_referrals.is_empty() {
                let bloom_payload = WormQueryPayload {
                    worm_id: *worm_id,
                    target: *target,
                    needle_peers: needle_peers.to_vec(),
                    ttl: 1,
                    visited: visited.to_vec(),
                    post_id: post_id.clone(),
                    blob_id,
                };

                let bloom_result = tokio::time::timeout(
                    std::time::Duration::from_millis(WORM_BLOOM_TIMEOUT_MS),
                    self.bloom_to_wide_peers(&wide_referrals, &bloom_payload),
                )
                .await;

                if let Ok(Some(wr)) = bloom_result {
                    return Ok(Some(wr));
                }
            }
        }

        Ok(None)
    }

    /// Send a PostFetch request to a specific peer and return the post if found.
    async fn send_post_fetch(
        &self,
        holder: &NodeId,
        post_id: &PostId,
    ) -> anyhow::Result<Option<crate::protocol::SyncPost>> {
        use crate::protocol::{PostFetchRequestPayload, PostFetchResponsePayload};

        // Get connection to the holder (mesh or session)
        let conn = self.connections.get(holder)
            .map(|pc| pc.connection.clone())
            .or_else(|| self.sessions.get(holder).map(|sc| sc.connection.clone()));

        let conn = match conn {
            Some(c) => c,
            None => {
                // Try to connect via endpoint (resolve address first)
                let addr = self.resolve_address(holder).await.unwrap_or(None);
                if let Some(addr_str) = addr {
                    let endpoint_id = iroh::EndpointId::from_bytes(holder)
                        .map_err(|_| anyhow::anyhow!("Invalid endpoint ID"))?;
                    let mut ep_addr = iroh::EndpointAddr::from(endpoint_id);
                    if let Ok(sock) = addr_str.parse::<std::net::SocketAddr>() {
                        ep_addr = ep_addr.with_ip_addr(sock);
                    }
                    self.endpoint.connect(ep_addr, ALPN_V2).await?
                } else {
                    return Err(anyhow::anyhow!("No connection or address for post holder"));
                }
            }
        };

        let (mut send, mut recv) = conn.open_bi().await?;
        let req = PostFetchRequestPayload { post_id: *post_id };
        write_typed_message(&mut send, MessageType::PostFetchRequest, &req).await?;
        send.finish()?;

        let msg_type = tokio::time::timeout(
            std::time::Duration::from_secs(10),
            read_message_type(&mut recv),
        ).await??;

        if msg_type != MessageType::PostFetchResponse {
            return Ok(None);
        }

        let resp: PostFetchResponsePayload = read_payload(&mut recv, MAX_PAYLOAD).await?;
        Ok(resp.post)
    }

    /// Send a TcpPunchRequest to a peer, asking them to punch a TCP hole toward a browser IP.
    /// Returns the peer's HTTP address if the punch succeeded.
    async fn send_tcp_punch(
        &self,
        holder: &NodeId,
        browser_ip: String,
        post_id: &PostId,
    ) -> anyhow::Result<Option<String>> {
        use crate::protocol::{TcpPunchRequestPayload, TcpPunchResultPayload};

        // Get connection to the holder (mesh or session)
        let conn = self.connections.get(holder)
            .map(|pc| pc.connection.clone())
            .or_else(|| self.sessions.get(holder).map(|sc| sc.connection.clone()));

        let conn = match conn {
            Some(c) => c,
            None => {
                // Try to connect via endpoint
                let addr = self.resolve_address(holder).await.unwrap_or(None);
                if let Some(addr_str) = addr {
                    let endpoint_id = iroh::EndpointId::from_bytes(holder)
                        .map_err(|_| anyhow::anyhow!("Invalid endpoint ID"))?;
                    let mut ep_addr = iroh::EndpointAddr::from(endpoint_id);
                    if let Ok(sock) = addr_str.parse::<std::net::SocketAddr>() {
                        ep_addr = ep_addr.with_ip_addr(sock);
                    }
                    self.endpoint.connect(ep_addr, ALPN_V2).await?
                } else {
                    return Err(anyhow::anyhow!("No connection or address for punch target"));
                }
            }
        };

        let (mut send, mut recv) = conn.open_bi().await?;
        let req = TcpPunchRequestPayload {
            browser_ip,
            post_id: *post_id,
        };
        write_typed_message(&mut send, MessageType::TcpPunchRequest, &req).await?;
        send.finish()?;

        let msg_type = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            read_message_type(&mut recv),
        ).await??;

        if msg_type != MessageType::TcpPunchResult {
            return Ok(None);
        }

        let resp: TcpPunchResultPayload = read_payload(&mut recv, 4096).await?;
        if resp.success {
            Ok(resp.http_addr)
        } else {
            Ok(None)
        }
    }

    /// Fan out a worm query to multiple peers, collecting ALL responses.
    /// Returns (first_hit, all_wide_referrals).
    async fn fan_out_worm_query_all(
        peer_conns: &[(NodeId, iroh::endpoint::Connection)],
        payload: &WormQueryPayload,
    ) -> (Option<WormResult>, Vec<(NodeId, String)>) {
        use tokio::task::JoinSet;

        let mut set = JoinSet::new();
        for (_nid, conn) in peer_conns {
            let conn = conn.clone();
            let payload = WormQueryPayload {
                worm_id: payload.worm_id,
                target: payload.target,
                needle_peers: payload.needle_peers.clone(),
                ttl: payload.ttl,
                visited: payload.visited.clone(),
                post_id: payload.post_id.clone(),
                blob_id: payload.blob_id,
            };
            set.spawn(async move { Self::send_worm_query_raw(&conn, &payload).await });
        }

        let mut first_hit: Option<WormResult> = None;
        let mut wide_referrals: Vec<(NodeId, String)> = Vec::new();

        while let Some(result) = set.join_next().await {
            if let Ok(Ok(resp)) = result {
                // Collect wide referral regardless of hit
                if let Some((ref_id, ref_addr)) = resp.wide_referral {
                    wide_referrals.push((ref_id, ref_addr));
                }

                // Treat content-found as a hit too (post/blob holder without node match)
                let is_hit = resp.found || resp.post_holder.is_some() || resp.blob_holder.is_some();
                if is_hit && first_hit.is_none() {
                    let found_node = resp.found_id.unwrap_or(payload.target);
                    first_hit = Some(WormResult {
                        node_id: found_node,
                        addresses: resp.addresses.clone(),
                        reporter: resp.reporter.unwrap_or([0u8; 32]),
                        freshness_ms: 0,
                        post_holder: resp.post_holder,
                        blob_holder: resp.blob_holder,
                    });
                    set.abort_all();
                }
            }
        }

        (first_hit, wide_referrals)
    }

    /// Connect to referred wide peers and send WormQuery{ttl=1}.
    /// First hit wins (abort rest).
    async fn bloom_to_wide_peers(
        &self,
        referrals: &[(NodeId, String)],
        payload: &WormQueryPayload,
    ) -> Option<WormResult> {
        use tokio::task::JoinSet;

        // Deduplicate referrals by node_id
        let mut seen = HashSet::new();
        let unique_referrals: Vec<&(NodeId, String)> = referrals
            .iter()
            .filter(|(nid, _)| {
                // Skip if it's us, already connected, or duplicate
                *nid != self.our_node_id
                    && !self.connections.contains_key(nid)
                    && seen.insert(*nid)
            })
            .collect();

        if unique_referrals.is_empty() {
            return None;
        }

        debug!(count = unique_referrals.len(), "Bloom round: connecting to wide peers");

        let mut set = JoinSet::new();
        let endpoint = self.endpoint.clone();

        for (ref_id, ref_addr) in unique_referrals {
            let endpoint = endpoint.clone();
            let ref_id = *ref_id;
            let ref_addr = ref_addr.clone();
            let payload = WormQueryPayload {
                worm_id: payload.worm_id,
                target: payload.target,
                needle_peers: payload.needle_peers.clone(),
                ttl: payload.ttl,
                visited: payload.visited.clone(),
                post_id: payload.post_id.clone(),
                blob_id: payload.blob_id,
            };

            set.spawn(async move {
                // Parse address and connect
                let endpoint_id = iroh::EndpointId::from_bytes(&ref_id).ok()?;
                let mut addr = iroh::EndpointAddr::from(endpoint_id);
                if let Ok(sock) = ref_addr.parse::<std::net::SocketAddr>() {
                    addr = addr.with_ip_addr(sock);
                }

                let conn = endpoint.connect(addr, ALPN_V2).await.ok()?;
                let resp = Self::send_worm_query_raw(&conn, &payload).await.ok()?;

                let is_hit = resp.found || resp.post_holder.is_some() || resp.blob_holder.is_some();
                if is_hit {
                    let found_node = resp.found_id.unwrap_or([0u8; 32]);
                    Some(WormResult {
                        node_id: found_node,
                        addresses: resp.addresses,
                        reporter: resp.reporter.unwrap_or([0u8; 32]),
                        freshness_ms: 0,
                        post_holder: resp.post_holder,
                        blob_holder: resp.blob_holder,
                    })
                } else {
                    None
                }
            });
        }

        while let Some(result) = set.join_next().await {
            if let Ok(Some(wr)) = result {
                set.abort_all();
                return Some(wr);
            }
        }

        None
    }

    /// Send a WormQuery on a bi-stream and read the raw WormResponsePayload.
    async fn send_worm_query_raw(
        conn: &iroh::endpoint::Connection,
        payload: &WormQueryPayload,
    ) -> anyhow::Result<WormResponsePayload> {
        let (mut send, mut recv) = conn.open_bi().await?;
        write_typed_message(&mut send, MessageType::WormQuery, payload).await?;
        send.finish()?;

        let msg_type = read_message_type(&mut recv).await?;
        if msg_type != MessageType::WormResponse {
            anyhow::bail!("expected WormResponse, got {:?}", msg_type);
        }
        let resp: WormResponsePayload = read_payload(&mut recv, MAX_PAYLOAD).await?;
        Ok(resp)
    }

    /// Handle an incoming WormQuery from a peer.
    /// Checks local map for ALL needles. If ttl > 0, also fans out to own peers.
    /// Always includes a wide_referral in the response if available.
    pub async fn handle_worm_query(
        &mut self,
        payload: WormQueryPayload,
        mut send: iroh::endpoint::SendStream,
        from_peer: NodeId,
    ) -> anyhow::Result<()> {
        let now = now_ms();

        // Dedup: check if we've already seen this worm_id
        if let Some(&seen_at) = self.seen_worms.get(&payload.worm_id) {
            if now - seen_at < WORM_DEDUP_EXPIRY_MS {
                let resp = WormResponsePayload {
                    worm_id: payload.worm_id,
                    found: false,
                    found_id: None,
                    addresses: vec![],
                    reporter: None,
                    hop: None,
                    wide_referral: None,
                    post_holder: None,
                    blob_holder: None,
                };
                write_typed_message(&mut send, MessageType::WormResponse, &resp).await?;
                send.finish()?;
                return Ok(());
            }
        }

        // Register this worm_id
        self.seen_worms.insert(payload.worm_id, now);

        // Prune old entries periodically
        if self.seen_worms.len() > 1000 {
            self.seen_worms
                .retain(|_, ts| now - *ts < WORM_DEDUP_EXPIRY_MS);
        }

        // Check for post/blob content locally (CDN tree, replicas, blob store)
        let mut post_holder: Option<NodeId> = None;
        let mut blob_holder: Option<NodeId> = None;

        if let Some(ref post_id) = payload.post_id {
            let found = {
                let store = self.storage.lock().await;
                // Direct: do we have this post?
                if store.get_post_with_visibility(post_id).ok().flatten().is_some() {
                    Some(self.our_node_id)
                } else {
                    // CDN tree: do any of our downstream hosts have it?
                    let downstream = store.get_post_downstream(post_id).unwrap_or_default();
                    if !downstream.is_empty() {
                        Some(downstream[0])
                    } else {
                        None
                    }
                }
            };
            post_holder = found;
        }

        if let Some(ref blob_id) = payload.blob_id {
            // Check if we have the blob locally
            if self.blob_store.get(blob_id).ok().flatten().is_some() {
                blob_holder = Some(self.our_node_id);
            } else {
                // Check CDN: do we know who has it via blob post ownership?
                let store = self.storage.lock().await;
                if let Ok(Some(pid)) = store.get_blob_post_id(blob_id) {
                    let downstream = store.get_post_downstream(&pid).unwrap_or_default();
                    if !downstream.is_empty() {
                        blob_holder = Some(downstream[0]);
                    }
                }
            }
        }

        // Build needle list: target + needle_peers
        let mut all_needles = vec![payload.target];
        all_needles.extend_from_slice(&payload.needle_peers);

        // Step 1: Check connections + N2/N3 for ALL needle IDs
        let local_result = {
            // Check direct connections first
            let mut found = None;
            for needle in &all_needles {
                if self.connections.contains_key(needle) {
                    let storage = self.storage.lock().await;
                    let addr = storage.get_peer_record(needle)?
                        .and_then(|r| r.addresses.first().map(|a| a.to_string()));
                    found = Some((*needle, addr.into_iter().collect::<Vec<_>>(), 0u64));
                    break;
                }
            }
            if found.is_none() {
                let storage = self.storage.lock().await;
                let entries = storage.find_any_in_n2_n3(&all_needles)?;
                if let Some((found_id, _reporter, _level)) = entries.first() {
                    drop(storage);
                    let address = self.resolve_address(found_id).await.unwrap_or(None);
                    found = Some((*found_id, address.into_iter().collect::<Vec<_>>(), 0u64));
                }
            }
            found
        };

        // If we found content (post or blob) but no node, still respond immediately
        let content_found = post_holder.is_some() || blob_holder.is_some();

        if let Some((found_id, addresses, _updated_at)) = local_result {
            let wide_referral = self.pick_random_wide_referral(&from_peer).await;

            let resp = WormResponsePayload {
                worm_id: payload.worm_id,
                found: true,
                found_id: Some(found_id),
                addresses,
                reporter: Some(self.our_node_id),
                hop: None,
                wide_referral,
                post_holder,
                blob_holder,
            };
            write_typed_message(&mut send, MessageType::WormResponse, &resp).await?;
            send.finish()?;
            return Ok(());
        }

        if content_found {
            let wide_referral = self.pick_random_wide_referral(&from_peer).await;

            let resp = WormResponsePayload {
                worm_id: payload.worm_id,
                found: false,
                found_id: None,
                addresses: vec![],
                reporter: Some(self.our_node_id),
                hop: None,
                wide_referral,
                post_holder,
                blob_holder,
            };
            write_typed_message(&mut send, MessageType::WormResponse, &resp).await?;
            send.finish()?;
            return Ok(());
        }

        // Step 2: If ttl > 0, fan out to our own connected peers with ttl=0
        if payload.ttl > 0 {
            let visited_set: HashSet<NodeId> = payload.visited.iter().copied().collect();
            let peer_conns: Vec<(NodeId, iroh::endpoint::Connection)> = self
                .connections
                .iter()
                .filter(|(nid, _)| !visited_set.contains(*nid) && **nid != from_peer)
                .map(|(nid, pc)| (*nid, pc.connection.clone()))
                .collect();

            if !peer_conns.is_empty() {
                let fan_payload = WormQueryPayload {
                    worm_id: payload.worm_id,
                    target: payload.target,
                    needle_peers: payload.needle_peers.clone(),
                    ttl: 0,
                    visited: payload.visited.clone(),
                    post_id: payload.post_id.clone(),
                    blob_id: payload.blob_id,
                };

                let (hit, _referrals) = tokio::time::timeout(
                    std::time::Duration::from_millis(WORM_FAN_OUT_TIMEOUT_MS),
                    Self::fan_out_worm_query_all(&peer_conns, &fan_payload),
                )
                .await
                .unwrap_or((None, vec![]));

                if let Some(wr) = hit {
                    let wide_referral = self.pick_random_wide_referral(&from_peer).await;
                    let resp = WormResponsePayload {
                        worm_id: payload.worm_id,
                        found: true,
                        found_id: Some(wr.node_id),
                        addresses: wr.addresses,
                        reporter: Some(wr.reporter),
                        hop: None,
                        wide_referral,
                        post_holder: wr.post_holder,
                        blob_holder: wr.blob_holder,
                    };
                    write_typed_message(&mut send, MessageType::WormResponse, &resp).await?;
                    send.finish()?;
                    return Ok(());
                }
            }
        }

        // Not found — respond with wide_referral for the originator's bloom round
        let wide_referral = self.pick_random_wide_referral(&from_peer).await;
        let resp = WormResponsePayload {
            worm_id: payload.worm_id,
            found: false,
            found_id: None,
            addresses: vec![],
            reporter: None,
            hop: None,
            wide_referral,
            post_holder: None,
            blob_holder: None,
        };
        write_typed_message(&mut send, MessageType::WormResponse, &resp).await?;
        send.finish()?;

        Ok(())
    }

    /// Pick a random wide-connected peer to include as a referral in worm responses.
    /// Returns (node_id, address_string) if a suitable peer with a known address is found.
    async fn pick_random_wide_referral(&self, exclude: &NodeId) -> Option<(NodeId, String)> {
        // Prefer wide-slot peers, but any connected peer with a known address works
        let candidates: Vec<(NodeId, PeerSlotKind)> = self
            .connections
            .iter()
            .filter(|(nid, _)| *nid != exclude && **nid != self.our_node_id)
            .map(|(nid, pc)| (*nid, pc.slot_kind))
            .collect();

        if candidates.is_empty() {
            return None;
        }

        // Prefer wide peers
        let wide: Vec<_> = candidates
            .iter()
            .filter(|(_, kind)| *kind == PeerSlotKind::Wide)
            .collect();

        // Shuffle candidates to pick randomly
        let ordered: Vec<&(NodeId, PeerSlotKind)> = if !wide.is_empty() {
            wide
        } else {
            candidates.iter().collect()
        };

        // Try each candidate until we find one with a known address
        let storage = self.storage.lock().await;
        for candidate in ordered {
            let nid = candidate.0;
            if let Ok(Some(rec)) = storage.get_peer_record(&nid) {
                if let Some(addr) = rec.addresses.first() {
                    return Some((nid, addr.to_string()));
                }
            }
        }

        None
    }

    /// Disconnect a peer, cleaning up N2/N3 entries.
    pub async fn disconnect_peer(&mut self, peer_id: &NodeId) {
        if let Some(pc) = self.connections.remove(peer_id) {
            drop(pc);
        }

        // Mark disconnected in referral list (anchor-side)
        self.mark_referral_disconnected(peer_id);

        let storage = self.storage.lock().await;
        // Remove their N2 contributions (their N1 share → our N2)
        let _ = storage.clear_peer_n2(peer_id);
        // Remove their N3 contributions (their N2 share → our N3)
        let _ = storage.clear_peer_n3(peer_id);
        // Remove from active mesh peers (but keep in peers table for reconnection)
        let _ = storage.remove_mesh_peer(peer_id);
        // Mark social route as disconnected
        if storage.has_social_route(peer_id).unwrap_or(false) {
            let _ = storage.set_social_route_status(peer_id, SocialStatus::Disconnected);
        }

        let remaining = self.connections.len();
        debug!(peer = hex::encode(peer_id), remaining, "Disconnected peer");
        self.log_activity(ActivityLevel::Info, ActivityCategory::Connection, format!("Disconnected, {} remaining", remaining), Some(*peer_id));

        // If mesh is completely empty, trigger immediate recovery
        if self.connections.is_empty() {
            info!("Mesh empty, triggering recovery");
            self.log_activity(ActivityLevel::Warn, ActivityCategory::Connection, "Mesh empty".into(), None);
            self.notify_recovery();
        }
    }

    /// Notify watchers that a previously disconnected peer has reconnected.
    /// Takes pre-gathered data to avoid holding &Storage across await points.
    async fn notify_watchers(&self, watchers: Vec<NodeId>, peer_id: &NodeId, route: &SocialRouteEntry) {
        let payload = SocialAddressUpdatePayload {
            node_id: *peer_id,
            addresses: route.addresses.iter().map(|a: &std::net::SocketAddr| a.to_string()).collect(),
            peer_addresses: route.peer_addresses.clone(),
        };

        let mut notified = 0;
        for watcher in &watchers {
            if let Some(pc) = self.connections.get(watcher) {
                let result = async {
                    let mut send = pc.connection.open_uni().await?;
                    write_typed_message(&mut send, MessageType::SocialAddressUpdate, &payload).await?;
                    send.finish()?;
                    anyhow::Ok(())
                }.await;
                if result.is_ok() {
                    notified += 1;
                }
            }
        }

        if notified > 0 {
            info!(peer = hex::encode(peer_id), notified, "Notified watchers of reconnection");
        }
    }

    /// Send a SocialCheckin to a specific peer. Returns their checkin reply info if successful.
    pub async fn send_social_checkin(
        &self,
        peer_id: &NodeId,
        our_node_id: &NodeId,
        our_addresses: &[String],
        our_peer_addresses: &[PeerWithAddress],
    ) -> anyhow::Result<SocialCheckinPayload> {
        let pc = self.connections.get(peer_id)
            .ok_or_else(|| anyhow::anyhow!("not connected to peer"))?;

        let payload = SocialCheckinPayload {
            node_id: *our_node_id,
            addresses: our_addresses.to_vec(),
            peer_addresses: our_peer_addresses.to_vec(),
        };

        let (mut send, mut recv) = pc.connection.open_bi().await?;
        write_typed_message(&mut send, MessageType::SocialCheckin, &payload).await?;
        send.finish()?;

        let msg_type = read_message_type(&mut recv).await?;
        if msg_type != MessageType::SocialCheckin {
            anyhow::bail!("expected SocialCheckin reply, got {:?}", msg_type);
        }
        let reply: SocialCheckinPayload = read_payload(&mut recv, MAX_PAYLOAD).await?;
        Ok(reply)
    }

    /// Rebalance connection slots: remove dead connections, prune stale N2/N3 entries.
    /// Returns list of newly connected peer IDs (caller should spawn run_mesh_streams for them).
    pub async fn rebalance_slots(&mut self) -> anyhow::Result<Vec<NodeId>> {
        self.log_activity(ActivityLevel::Info, ActivityCategory::Rebalance, "Rebalance started".into(), None);

        // 1. Remove dead + zombie connections
        let mut dead = Vec::new();
        let now = now_ms();
        for (peer_id, pc) in &self.connections {
            if pc.connection.close_reason().is_some() {
                dead.push(*peer_id);
            } else if now.saturating_sub(pc.last_activity.load(Ordering::Relaxed)) > ZOMBIE_TIMEOUT_MS {
                let idle_secs = now.saturating_sub(pc.last_activity.load(Ordering::Relaxed)) / 1000;
                info!(peer = hex::encode(peer_id), idle_secs, "Zombie connection detected (no activity)");
                self.log_activity(ActivityLevel::Warn, ActivityCategory::Rebalance, format!("Zombie (idle {}s)", idle_secs), Some(*peer_id));
                dead.push(*peer_id);
            }
        }

        for peer_id in &dead {
            info!(peer = hex::encode(peer_id), "Removing dead connection");
            self.log_activity(ActivityLevel::Info, ActivityCategory::Rebalance, "Removed dead connection".into(), Some(*peer_id));
            self.disconnect_peer(peer_id).await;
        }

        // 2. Prune stale N2/N3 entries (5 hours) + stale watchers (30 days)
        {
            let storage = self.storage.lock().await;
            let pruned = storage.prune_n2_n3(5 * 60 * 60 * 1000)?;
            if pruned > 0 {
                info!(pruned, "Pruned stale N2/N3 entries");
            }
            let _ = storage.prune_stale_watchers(WATCHER_EXPIRY_MS);
        }

        // 3. Diversity scoring: find low-diversity peers for potential eviction
        {
            let storage = self.storage.lock().await;
            let connected: Vec<NodeId> = self.connections.keys().copied().collect();
            let mut zero_diversity = Vec::new();
            for peer_id in &connected {
                let unique = storage.count_unique_n2_for_reporter(peer_id, &[]).unwrap_or(0);
                if unique == 0 {
                    zero_diversity.push(*peer_id);
                }
            }
            if !zero_diversity.is_empty() {
                debug!(count = zero_diversity.len(), "Peers with zero unique N2 contributions");
            }
        }

        let mut newly_connected: Vec<NodeId> = Vec::new();

        // Priority 0 (NEW): Reconnect preferred peers
        {
            let preferred_peers = {
                let storage = self.storage.lock().await;
                storage.list_preferred_peers().unwrap_or_default()
            };

            let now = now_ms();
            for peer_id in &preferred_peers {
                if self.connections.contains_key(peer_id) || *peer_id == self.our_node_id {
                    continue;
                }
                // Prune preferred peers unreachable for 7+ days
                if let Some(&failed_at) = self.unreachable_peers.get(peer_id) {
                    if now.saturating_sub(failed_at) > PREFERRED_UNREACHABLE_PRUNE_MS {
                        info!(peer = hex::encode(peer_id), "Removing preferred peer unreachable for 7 days+");
                        let storage = self.storage.lock().await;
                        let _ = storage.remove_preferred_peer(peer_id);
                        continue;
                    }
                }

                // Evict lowest-diversity non-preferred peer if at capacity
                let total_slots = self.preferred_slots + self.local_slots + self.wide_slots;
                if self.connections.len() >= total_slots {
                    let evict_candidate = self.find_non_preferred_eviction_candidate().await;
                    if let Some(evict_id) = evict_candidate {
                        info!(
                            evicting = hex::encode(evict_id),
                            for_preferred = hex::encode(peer_id),
                            "Evicting non-preferred peer for preferred reconnection"
                        );
                        self.log_activity(ActivityLevel::Info, ActivityCategory::Rebalance, format!("Evicting {} for preferred {}", &hex::encode(evict_id)[..8], &hex::encode(peer_id)[..8]), Some(evict_id));
                        self.disconnect_peer(&evict_id).await;
                    } else {
                        debug!(peer = hex::encode(peer_id), "No non-preferred peer to evict");
                        continue;
                    }
                }

                if let Err(e) = self.reconnect_preferred(peer_id).await {
                    debug!(peer = hex::encode(peer_id), error = %e, "Preferred reconnection failed");
                } else if self.connections.contains_key(peer_id) {
                    newly_connected.push(*peer_id);
                }
            }
        }

        // Priority 1+2: Fill empty local slots with diverse candidates
        let local_count = self.count_kind(PeerSlotKind::Local);
        if local_count < self.local_slots {
            let candidates: Vec<(NodeId, Option<String>)> = {
                let storage = self.storage.lock().await;
                let mut cands = Vec::new();

                // Priority 1: reconnect recently-dead non-preferred peers
                for peer_id in &dead {
                    if *peer_id == self.our_node_id {
                        continue;
                    }
                    // Skip preferred peers — handled above
                    if storage.is_preferred_peer(peer_id).unwrap_or(false) {
                        continue;
                    }
                    if let Ok(Some(rec)) = storage.get_peer_record(peer_id) {
                        let addr = rec.addresses.first().map(|a| a.to_string());
                        cands.push((*peer_id, addr));
                    }
                }

                // Priority 2: handled by growth loop (reactive, one-at-a-time)
                cands
            };

            let slots_available = self.local_slots - local_count;
            let to_connect = candidates.len().min(slots_available);
            if to_connect > 0 {
                debug!(candidates = candidates.len(), connecting = to_connect, "Filling local slots");
            }

            for (peer_id, addr_str) in candidates.into_iter().take(slots_available) {
                let resolved_addr = if let Some(addr_s) = addr_str {
                    Some(addr_s)
                } else {
                    match self.resolve_address(&peer_id).await {
                        Ok(addr) => addr,
                        Err(_) => None,
                    }
                };

                if let Some(addr_s) = resolved_addr {
                    let endpoint_id = match iroh::EndpointId::from_bytes(&peer_id) {
                        Ok(eid) => eid,
                        Err(_) => continue,
                    };
                    let mut addr = iroh::EndpointAddr::from(endpoint_id);
                    if let Ok(sock) = addr_s.parse::<std::net::SocketAddr>() {
                        addr = addr.with_ip_addr(sock);
                    }
                    match self.connect_to(peer_id, addr, PeerSlotKind::Local).await {
                        Ok(()) => {
                            info!(peer = hex::encode(peer_id), "Auto-connected to diverse peer");
                            newly_connected.push(peer_id);
                        }
                        Err(e) => {
                            debug!(peer = hex::encode(peer_id), error = %e, "Auto-connect failed");
                        }
                    }
                }
            }
        }

        // Initial exchange for newly connected peers is done by Network::rebalance()
        // outside the conn_mgr lock to avoid blocking other operations.

        // 5. Reap idle session connections (keeps anchor + referral sessions alive)
        self.reap_idle_sessions(SESSION_IDLE_TIMEOUT_MS).await;

        // 6. Prune stale relay intro dedup entries
        {
            let now = now_ms();
            if self.seen_intros.len() > 500 {
                self.seen_intros
                    .retain(|_, ts| now - *ts < RELAY_INTRO_DEDUP_EXPIRY_MS);
            }
        }

        if !dead.is_empty() {
            info!(removed = dead.len(), "Rebalance complete");
            self.log_activity(ActivityLevel::Info, ActivityCategory::Rebalance, format!("Complete, removed {}", dead.len()), None);
        } else {
            self.log_activity(ActivityLevel::Info, ActivityCategory::Rebalance, "Complete, no changes".into(), None);
        }

        // Backstop: signal growth loop to fill any remaining local slots
        self.notify_growth();

        Ok(newly_connected)
    }

    /// Find the lowest-diversity non-preferred peer to evict.
    async fn find_non_preferred_eviction_candidate(&self) -> Option<NodeId> {
        let storage = self.storage.lock().await;
        let mut worst: Option<(NodeId, usize)> = None;
        for (peer_id, mc) in &self.connections {
            if mc.slot_kind == PeerSlotKind::Preferred {
                continue; // Never evict preferred
            }
            let unique = storage.count_unique_n2_for_reporter(peer_id, &[]).unwrap_or(0);
            match &worst {
                None => worst = Some((*peer_id, unique)),
                Some((_, worst_unique)) if unique < *worst_unique => worst = Some((*peer_id, unique)),
                _ => {}
            }
        }
        worst.map(|(nid, _)| nid)
    }

    pub fn is_connected(&self, peer_id: &NodeId) -> bool {
        self.connections.contains_key(peer_id)
    }

    pub fn connected_peers(&self) -> Vec<NodeId> {
        self.connections.keys().cloned().collect()
    }

    pub fn connection_count(&self) -> usize {
        self.connections.len()
    }

    /// Get connection info for display: (node_id, slot_kind, connected_at)
    /// Get a reference to the connections map (for Network to access connections).
    pub fn connections_ref(&self) -> &HashMap<NodeId, MeshConnection> {
        &self.connections
    }

    pub fn connection_info(&self) -> Vec<(NodeId, PeerSlotKind, u64)> {
        self.connections
            .values()
            .map(|pc| (pc.node_id, pc.slot_kind, pc.connected_at))
            .collect()
    }

    fn count_kind(&self, kind: PeerSlotKind) -> usize {
        self.connections.values().filter(|pc| pc.slot_kind == kind).count()
    }

    // ---- Preferred peer negotiation ----

    /// Request bilateral preferred peer status with a connected mesh peer.
    /// On success, both sides persist the agreement and upgrade the slot.
    pub async fn request_prefer(&mut self, peer_id: &NodeId) -> anyhow::Result<bool> {
        let pc = self.connections.get(peer_id)
            .ok_or_else(|| anyhow::anyhow!("peer not connected"))?;

        // Check if we have room
        let preferred_count = self.count_kind(PeerSlotKind::Preferred);
        if preferred_count >= self.preferred_slots {
            anyhow::bail!("preferred slots full ({}/{})", preferred_count, self.preferred_slots);
        }

        let request = MeshPreferPayload {
            requesting: true,
            accepted: false,
            reject_reason: None,
        };

        let (mut send, mut recv) = pc.connection.open_bi().await?;
        write_typed_message(&mut send, MessageType::MeshPrefer, &request).await?;
        send.finish()?;

        let msg_type = read_message_type(&mut recv).await?;
        if msg_type != MessageType::MeshPrefer {
            anyhow::bail!("expected MeshPrefer response, got {:?}", msg_type);
        }
        let response: MeshPreferPayload = read_payload(&mut recv, 4096).await?;

        if response.accepted {
            // Persist agreement
            let storage = self.storage.lock().await;
            storage.add_preferred_peer(peer_id)?;
            storage.add_mesh_peer(peer_id, PeerSlotKind::Preferred, 100)?;
            drop(storage);

            // Upgrade slot in-memory
            if let Some(mc) = self.connections.get_mut(peer_id) {
                mc.slot_kind = PeerSlotKind::Preferred;
            }
            info!(peer = hex::encode(peer_id), "Preferred peer agreement established");
            Ok(true)
        } else {
            debug!(
                peer = hex::encode(peer_id),
                reason = ?response.reject_reason,
                "Preferred peer request rejected"
            );
            Ok(false)
        }
    }

    /// Handle an incoming MeshPrefer request from a connected peer.
    pub async fn handle_mesh_prefer(
        &mut self,
        from_peer: NodeId,
        mut send: iroh::endpoint::SendStream,
        mut recv: iroh::endpoint::RecvStream,
    ) -> anyhow::Result<()> {
        let request: MeshPreferPayload = read_payload(&mut recv, 4096).await?;

        if !request.requesting {
            // Not a request — ignore
            return Ok(());
        }

        // Check if we have room for a preferred peer
        let preferred_count = self.count_kind(PeerSlotKind::Preferred);
        let can_accept = preferred_count < self.preferred_slots
            && self.connections.contains_key(&from_peer);

        let response = if can_accept {
            // Persist agreement
            let storage = self.storage.lock().await;
            storage.add_preferred_peer(&from_peer)?;
            storage.add_mesh_peer(&from_peer, PeerSlotKind::Preferred, 100)?;
            drop(storage);

            // Upgrade slot in-memory
            if let Some(mc) = self.connections.get_mut(&from_peer) {
                mc.slot_kind = PeerSlotKind::Preferred;
            }

            info!(peer = hex::encode(from_peer), "Accepted preferred peer request");

            MeshPreferPayload {
                requesting: false,
                accepted: true,
                reject_reason: None,
            }
        } else {
            let reason = if !self.connections.contains_key(&from_peer) {
                "not connected".to_string()
            } else {
                format!("preferred slots full ({}/{})", preferred_count, self.preferred_slots)
            };
            debug!(peer = hex::encode(from_peer), reason = %reason, "Rejecting preferred peer request");
            MeshPreferPayload {
                requesting: false,
                accepted: false,
                reject_reason: Some(reason),
            }
        };

        write_typed_message(&mut send, MessageType::MeshPrefer, &response).await?;
        send.finish()?;
        Ok(())
    }

    /// Reconnect a preferred peer via relay introduction if direct connect fails.
    pub async fn reconnect_preferred(&mut self, peer_id: &NodeId) -> anyhow::Result<()> {
        if self.connections.contains_key(peer_id) {
            return Ok(()); // Already connected
        }

        // Try direct connect first (from peers table or social route)
        let addr_str = if !self.is_likely_unreachable(peer_id) {
            let storage = self.storage.lock().await;
            let addr = storage.get_peer_record(peer_id)?
                .and_then(|r| r.addresses.first().map(|a| a.to_string()))
                .or_else(|| {
                    storage.get_social_route(peer_id).ok().flatten()
                        .and_then(|r| r.addresses.first().map(|a| a.to_string()))
                });
            drop(storage);
            addr
        } else {
            None
        };

        if let Some(addr_s) = addr_str {
            let endpoint_id = iroh::EndpointId::from_bytes(peer_id)?;
            let mut addr = iroh::EndpointAddr::from(endpoint_id);
            if let Ok(sock) = addr_s.parse::<std::net::SocketAddr>() {
                addr = addr.with_ip_addr(sock);
            }
            match tokio::time::timeout(
                std::time::Duration::from_millis(HOLE_PUNCH_TIMEOUT_MS),
                self.connect_to(*peer_id, addr, PeerSlotKind::Preferred),
            ).await {
                Ok(Ok(())) => {
                    self.mark_reachable(peer_id);
                    info!(peer = hex::encode(peer_id), "Preferred peer reconnected directly");
                    return Ok(());
                }
                Ok(Err(e)) => {
                    debug!(peer = hex::encode(peer_id), error = %e, "Direct reconnect failed, trying relay");
                    self.mark_unreachable(peer_id);
                }
                Err(_) => {
                    debug!(peer = hex::encode(peer_id), "Direct reconnect timed out, trying relay");
                    self.mark_unreachable(peer_id);
                }
            }
        }

        // Try relay introduction (with timeout to avoid holding lock forever)
        let relays = self.find_relays_for(peer_id).await;
        for (relay_peer, ttl) in relays {
            let introduce_result = match tokio::time::timeout(
                std::time::Duration::from_millis(RELAY_INTRO_TIMEOUT_MS),
                self.send_relay_introduce(&relay_peer, peer_id, ttl),
            ).await {
                Ok(r) => r,
                Err(_) => {
                    debug!(relay = hex::encode(relay_peer), "Relay introduce timed out");
                    continue;
                }
            };
            match introduce_result {
                Ok(result) if result.accepted => {
                    let our_profile = self.our_nat_profile();
                    let peer_profile = {
                        let s = self.storage.lock().await;
                        s.get_peer_nat_profile(peer_id)
                    };
                    if let Some(conn) = hole_punch_with_scanning(&self.endpoint, peer_id, &result.target_addresses, our_profile, peer_profile).await {
                        // Register as preferred mesh peer
                        self.register_connection(*peer_id, conn, &[], PeerSlotKind::Preferred).await;
                        self.mark_reachable(peer_id);
                        info!(
                            peer = hex::encode(peer_id),
                            relay = hex::encode(relay_peer),
                            "Preferred peer reconnected via relay hole punch"
                        );
                        return Ok(());
                    }
                }
                Ok(_) => {} // Not accepted, try next relay
                Err(e) => {
                    debug!(relay = hex::encode(relay_peer), error = %e, "Relay introduce failed");
                }
            }
        }

        debug!(peer = hex::encode(peer_id), "Could not reconnect preferred peer");
        Ok(())
    }

    // ---- Session connection management ----

    /// Add a session connection. Evicts oldest idle session if at capacity.
    pub fn add_session(
        &mut self,
        node_id: NodeId,
        connection: iroh::endpoint::Connection,
        reach_method: SessionReachMethod,
        remote_addr: Option<std::net::SocketAddr>,
    ) {
        let now = now_ms();

        // Evict oldest idle session if at capacity
        if self.sessions.len() >= self.session_slots && !self.sessions.contains_key(&node_id) {
            let oldest = self
                .sessions
                .iter()
                .min_by_key(|(_, s)| s.last_active_at)
                .map(|(nid, _)| *nid);
            if let Some(old_nid) = oldest {
                debug!(peer = hex::encode(old_nid), "Evicting idle session for new session");
                self.sessions.remove(&old_nid);
            }
        }

        self.sessions.insert(
            node_id,
            SessionConnection {
                node_id,
                connection,
                created_at: now,
                last_active_at: now,
                reach_method,
                remote_addr,
            },
        );
        info!(peer = hex::encode(node_id), method = %reach_method, "Added session connection");
    }

    /// Get a session connection (does NOT touch last_active_at — caller should call touch_session).
    pub fn get_session(&self, node_id: &NodeId) -> Option<&SessionConnection> {
        self.sessions.get(node_id)
    }

    /// Remove a session connection.
    pub fn remove_session(&mut self, node_id: &NodeId) {
        if self.sessions.remove(node_id).is_some() {
            debug!(peer = hex::encode(node_id), "Removed session connection");
        }
    }

    /// Update last_active_at for a session.
    pub fn touch_session(&mut self, node_id: &NodeId) {
        if let Some(session) = self.sessions.get_mut(node_id) {
            session.last_active_at = now_ms();
        }
    }

    /// Close sessions idle longer than the given timeout.
    pub async fn reap_idle_sessions(&mut self, idle_timeout_ms: u64) {
        let now = now_ms();

        // Build set of sessions that have a reason to stay alive
        let mut keep_alive: std::collections::HashSet<NodeId> = std::collections::HashSet::new();

        // Anchor side: peers on our referral list need their session kept
        for nid in self.referral_list.keys() {
            if self.sessions.contains_key(nid) && !self.connections.contains_key(nid) {
                keep_alive.insert(*nid);
            }
        }

        // Client side: known anchors we're session-connected to (mesh was full)
        {
            let storage = self.storage.lock().await;
            for nid in self.sessions.keys() {
                if storage.is_peer_anchor(nid).unwrap_or(false) && !self.connections.contains_key(nid) {
                    keep_alive.insert(*nid);
                }
            }
        }

        let mut reaped = Vec::new();
        for (nid, session) in &self.sessions {
            if keep_alive.contains(nid) {
                continue; // Session has a reason to stay alive
            }
            if now.saturating_sub(session.last_active_at) > idle_timeout_ms {
                reaped.push(*nid);
            }
        }
        for nid in &reaped {
            self.sessions.remove(nid);
        }
        if !reaped.is_empty() {
            info!(count = reaped.len(), kept = keep_alive.len(), "Reaped idle sessions");
        }
    }

    /// Get session info for display: (node_id, reach_method, last_active_at)
    pub fn session_info(&self) -> Vec<(NodeId, SessionReachMethod, u64)> {
        self.sessions
            .values()
            .map(|s| (s.node_id, s.reach_method, s.last_active_at))
            .collect()
    }

    /// Check if a node has a session connection.
    pub fn has_session(&self, node_id: &NodeId) -> bool {
        self.sessions.contains_key(node_id)
    }

    /// Check if a node is connected via mesh OR session.
    pub fn is_connected_or_session(&self, node_id: &NodeId) -> bool {
        self.connections.contains_key(node_id) || self.sessions.contains_key(node_id)
    }

    /// Get list of session peer NodeIds.
    pub fn session_peer_ids(&self) -> Vec<NodeId> {
        self.sessions.keys().copied().collect()
    }

    /// Get the active relay pipe count reference (for relay pipe tasks).
    pub fn active_relay_pipes(&self) -> &Arc<AtomicU64> {
        &self.active_relay_pipes
    }

    /// Check if we can accept more relay pipes.
    pub fn can_accept_relay_pipe(&self) -> bool {
        self.active_relay_pipes.load(Ordering::Relaxed) < self.max_relay_pipes as u64
    }

    /// Get our node ID.
    pub fn our_node_id(&self) -> &NodeId {
        &self.our_node_id
    }

    /// Get a connection (mesh or session) to a peer, if any.
    pub(crate) fn get_any_connection(&self, peer: &NodeId) -> Option<iroh::endpoint::Connection> {
        self.connections.get(peer)
            .map(|pc| pc.connection.clone())
            .or_else(|| self.sessions.get(peer).map(|s| s.connection.clone()))
    }

    /// Get a clone of the storage Arc (for standalone exchange functions).
    pub fn storage_ref(&self) -> Arc<Mutex<Storage>> {
        Arc::clone(&self.storage)
    }

    // ---- NAT reachability tracking ----

    /// 30 minutes — how long we consider a "unreachable" verdict valid
    const UNREACHABLE_EXPIRY_MS: u64 = 1_800_000;

    /// Mark a peer as directly reachable (clear any unreachable flag).
    pub fn mark_reachable(&mut self, node_id: &NodeId) {
        if self.unreachable_peers.remove(node_id).is_some() {
            debug!(peer = hex::encode(node_id), "Peer now marked directly reachable");
        }
    }

    /// Mark a peer as not directly reachable (behind NAT or offline).
    pub fn mark_unreachable(&mut self, node_id: &NodeId) {
        let now = now_ms();
        self.unreachable_peers.insert(*node_id, now);
        debug!(peer = hex::encode(node_id), "Peer marked unreachable (direct failed)");
    }

    /// Behavioral NAT filtering inference: update peer's profile based on connection outcome.
    /// Call after a successful hole punch to refine filtering classification.
    /// `used_scanning` = true if scanning was required (standard punch failed first).
    pub async fn infer_nat_filtering(&self, node_id: &NodeId, used_scanning: bool) {
        let s = self.storage.lock().await;
        let mut profile = s.get_peer_nat_profile(node_id);
        if profile.mapping == crate::types::NatMapping::EndpointDependent {
            if used_scanning {
                // Scanning succeeded where standard punch failed → port-restricted
                if profile.filtering != crate::types::NatFiltering::PortRestricted {
                    profile.filtering = crate::types::NatFiltering::PortRestricted;
                    let _ = s.set_peer_nat_profile(node_id, &profile);
                    info!(peer = hex::encode(node_id), "NAT filtering inferred: port-restricted (from scanning outcome)");
                }
            } else {
                // Standard punch succeeded despite EDM → open/address-restricted
                if profile.filtering != crate::types::NatFiltering::Open {
                    profile.filtering = crate::types::NatFiltering::Open;
                    let _ = s.set_peer_nat_profile(node_id, &profile);
                    info!(peer = hex::encode(node_id), "NAT filtering inferred: open (standard punch succeeded)");
                }
            }
        }
    }

    /// Check if a peer is likely unreachable directly (failed recently).
    pub fn is_likely_unreachable(&self, node_id: &NodeId) -> bool {
        if let Some(&failed_at) = self.unreachable_peers.get(node_id) {
            now_ms().saturating_sub(failed_at) < Self::UNREACHABLE_EXPIRY_MS
        } else {
            false
        }
    }

    // ---- Relay introduction ----

    /// Find relay peers that can introduce us to a target.
    /// Returns up to 3 candidates as (relay_node_id, ttl) where ttl=0 means relay
    /// knows target directly (N2), ttl=1 means relay chains through their peer (N3).
    pub async fn find_relays_for(&self, target: &NodeId) -> Vec<(NodeId, u8)> {
        let mut candidates = Vec::new();
        let storage = self.storage.lock().await;

        // Step 1 (NEW): Check target's preferred_tree from social_routes (~100 NodeIds)
        // Intersect with our connections → TTL=0 candidates (they know target or are stably nearby)
        if let Ok(Some(route)) = storage.get_social_route(target) {
            if !route.preferred_tree.is_empty() {
                // Prefer our own preferred peers first within the tree
                for tree_node in &route.preferred_tree {
                    if candidates.len() >= 3 {
                        break;
                    }
                    if let Some(mc) = self.connections.get(tree_node) {
                        if mc.slot_kind == PeerSlotKind::Preferred
                            && !candidates.iter().any(|(nid, _)| nid == tree_node)
                        {
                            candidates.push((*tree_node, 0));
                        }
                    }
                }
                // Then any connected tree node
                for tree_node in &route.preferred_tree {
                    if candidates.len() >= 3 {
                        break;
                    }
                    if self.connections.contains_key(tree_node)
                        && !candidates.iter().any(|(nid, _)| nid == tree_node)
                    {
                        candidates.push((*tree_node, 0));
                    }
                }
            }
        }

        // Step 2: Our mesh peers that have target in their N1 (our N2 entry tagged to them)
        if candidates.len() < 3 {
            if let Ok(reporters) = storage.find_in_n2(target) {
                for reporter in reporters {
                    if self.connections.contains_key(&reporter)
                        && !candidates.iter().any(|(nid, _)| *nid == reporter)
                        && candidates.len() < 3
                    {
                        candidates.push((reporter, 0));
                    }
                }
            }
        }

        // Step 3: Intersect preferred_tree with N3 → TTL=1 candidates
        if candidates.len() < 3 {
            if let Ok(Some(route)) = storage.get_social_route(target) {
                for tree_node in &route.preferred_tree {
                    if candidates.len() >= 3 {
                        break;
                    }
                    if let Ok(reporters) = storage.find_in_n3(tree_node) {
                        for reporter in reporters {
                            if self.connections.contains_key(&reporter)
                                && !candidates.iter().any(|(nid, _)| *nid == reporter)
                                && candidates.len() < 3
                            {
                                candidates.push((reporter, 1));
                            }
                        }
                    }
                }
            }
        }

        // Step 4: Fallback — full N3 scan for target
        if candidates.len() < 3 {
            if let Ok(reporters) = storage.find_in_n3(target) {
                for reporter in reporters {
                    if self.connections.contains_key(&reporter)
                        && !candidates.iter().any(|(nid, _)| *nid == reporter)
                        && candidates.len() < 3
                    {
                        candidates.push((reporter, 1));
                    }
                }
            }
        }

        candidates
    }

    /// Send a RelayIntroduce request to a relay peer and wait for the result.
    pub async fn send_relay_introduce(
        &self,
        relay_peer: &NodeId,
        target: &NodeId,
        ttl: u8,
    ) -> anyhow::Result<RelayIntroduceResultPayload> {
        let pc = self.connections.get(relay_peer)
            .ok_or_else(|| anyhow::anyhow!("relay peer not connected"))?;

        let intro_id: IntroId = rand::random();
        let mut our_addrs: Vec<String> = self.endpoint.addr().ip_addrs()
            .filter(|s| crate::network::is_publicly_routable(s))
            .map(|s| s.to_string())
            .collect();
        // Prepend UPnP external address if available
        if let Some(ref ext) = self.upnp_external_addr {
            let ext_str = ext.to_string();
            if !our_addrs.contains(&ext_str) {
                our_addrs.insert(0, ext_str);
            }
        }

        let payload = RelayIntroducePayload {
            intro_id,
            target: *target,
            requester: self.our_node_id,
            requester_addresses: our_addrs,
            ttl,
        };

        let (mut send, mut recv) = pc.connection.open_bi().await?;
        write_typed_message(&mut send, MessageType::RelayIntroduce, &payload).await?;
        send.finish()?;

        let msg_type = read_message_type(&mut recv).await?;
        if msg_type != MessageType::RelayIntroduceResult {
            anyhow::bail!("expected RelayIntroduceResult, got {:?}", msg_type);
        }
        let result: RelayIntroduceResultPayload = read_payload(&mut recv, MAX_PAYLOAD).await?;
        Ok(result)
    }

    /// Handle an incoming RelayIntroduce — either as relay (forward) or as target (respond).
    /// `conn_mgr_arc` is passed so the spawned hole-punch task can register the
    /// resulting connection without holding the lock during the 30 s punch window.
    pub async fn handle_relay_introduce(
        &mut self,
        payload: RelayIntroducePayload,
        mut send: iroh::endpoint::SendStream,
        from_peer: NodeId,
        conn_mgr_arc: Arc<Mutex<Self>>,
    ) -> anyhow::Result<()> {
        let now = now_ms();

        // Dedup check
        if let Some(&seen_at) = self.seen_intros.get(&payload.intro_id) {
            if now - seen_at < RELAY_INTRO_DEDUP_EXPIRY_MS {
                let result = RelayIntroduceResultPayload {
                    intro_id: payload.intro_id,
                    accepted: false,
                    target_addresses: vec![],
                    relay_available: false,
                    reject_reason: Some("duplicate intro".to_string()),
                };
                write_typed_message(&mut send, MessageType::RelayIntroduceResult, &result).await?;
                send.finish()?;
                return Ok(());
            }
        }
        self.seen_intros.insert(payload.intro_id, now);

        // Are WE the target?
        if payload.target == self.our_node_id {
            // Respond with our globally-routable addresses only (no Docker bridge / private IPs)
            let mut our_addrs: Vec<String> = self.endpoint.addr().ip_addrs()
                .filter(|s| crate::network::is_publicly_routable(s))
                .map(|s| s.to_string())
                .collect();
            // Prepend UPnP external address if available
            if let Some(ref ext) = self.upnp_external_addr {
                let ext_str = ext.to_string();
                if !our_addrs.contains(&ext_str) {
                    our_addrs.insert(0, ext_str);
                }
            }

            let result = RelayIntroduceResultPayload {
                intro_id: payload.intro_id,
                accepted: true,
                target_addresses: our_addrs,
                relay_available: false,
                reject_reason: None,
            };
            write_typed_message(&mut send, MessageType::RelayIntroduceResult, &result).await?;
            send.finish()?;

            // Hole punch: filter to routable addresses only (skip Docker bridge IPs etc.)
            let routable_requester_addrs: Vec<String> = payload.requester_addresses.iter()
                .filter(|a| a.parse::<std::net::SocketAddr>().map_or(false, |s| crate::network::is_publicly_routable(&s)))
                .cloned()
                .collect();
            info!(
                requester = hex::encode(payload.requester),
                addrs = ?routable_requester_addrs,
                "Relay introduction: we are target, hole punching to requester"
            );
            self.log_activity(ActivityLevel::Info, ActivityCategory::Relay, format!("We are target, hole punching to {}", &hex::encode(payload.requester)[..8]), Some(payload.requester));

            let endpoint = self.endpoint.clone();
            let requester = payload.requester;
            let requester_addrs = routable_requester_addrs;
            let storage = Arc::clone(&self.storage);
            let our_node_id = self.our_node_id;
            let our_nat_type = self.nat_type;
            let our_http_capable = self.http_capable;
            let our_http_addr = self.http_addr.clone();
            let our_nat_profile = self.our_nat_profile();
            let peer_nat_profile = {
                let s = self.storage.lock().await;
                s.get_peer_nat_profile(&requester)
            };
            tokio::spawn(async move {
                if let Some(conn) = hole_punch_with_scanning(&endpoint, &requester, &requester_addrs, our_nat_profile, peer_nat_profile).await {
                    // Register as session with the peer's address for relay introduction
                    let remote_sock = requester_addrs.iter()
                        .filter_map(|a| a.parse::<std::net::SocketAddr>().ok())
                        .find(|s| crate::network::is_publicly_routable(s));
                    let mut cm = conn_mgr_arc.lock().await;
                    if cm.is_connected(&requester) {
                        // Initiator already connected to us (their punch succeeded first)
                        return;
                    }
                    cm.add_session(requester, conn, SessionReachMethod::HolePunch, remote_sock);
                    cm.mark_reachable(&requester);
                    cm.log_activity(
                        ActivityLevel::Info,
                        ActivityCategory::Relay,
                        format!("Target-side hole punch succeeded to {}", &hex::encode(requester)[..8]),
                        Some(requester),
                    );
                    // Run initial exchange so both sides know each other
                    let storage_clone = Arc::clone(&storage);
                    if let Some(session) = cm.sessions.get(&requester) {
                        let session_conn = session.connection.clone();
                        drop(cm); // release lock before async work
                        match initial_exchange_connect(&storage_clone, &our_node_id, &session_conn, requester, None, our_nat_type, our_http_capable, our_http_addr.clone(), None, None).await {
                            Ok(ExchangeResult::Accepted) => {
                                tracing::info!(peer = hex::encode(requester), "Target-side: initial exchange after hole punch");
                            }
                            Ok(ExchangeResult::Refused { .. }) => {
                                tracing::debug!(peer = hex::encode(requester), "Target-side: mesh refused after hole punch (ok, initiator will drive)");
                            }
                            Err(e) => {
                                tracing::debug!(peer = hex::encode(requester), error = %e, "Target-side: initial exchange failed (ok, initiator will drive)");
                            }
                        }
                    }
                }
            });

            return Ok(());
        }

        // We are a RELAY — forward the introduction to the target

        // Check if target is directly connected (our N1)
        if let Some(target_pc) = self.connections.get(&payload.target) {
            let target_conn = target_pc.connection.clone();
            let target_observed_addr = target_pc.remote_addr;
            let relay_available = self.can_accept_relay_pipe();

            // Look up the requester's observed address from mesh or session
            let requester_observed_addr = self.connections.get(&from_peer)
                .and_then(|pc| pc.remote_addr)
                .or_else(|| self.sessions.get(&from_peer).and_then(|s| s.remote_addr));

            // Build forwarded payload with requester's real public address injected
            let mut forwarded_payload = payload.clone();
            if let Some(addr) = requester_observed_addr {
                let addr_str = addr.to_string();
                if !forwarded_payload.requester_addresses.contains(&addr_str) {
                    forwarded_payload.requester_addresses.insert(0, addr_str);
                    info!(
                        requester = hex::encode(payload.requester),
                        observed_addr = %addr,
                        "Relay injecting requester's observed public address"
                    );
                }
            }

            // Forward to target
            let forward_result = async {
                let (mut fwd_send, mut fwd_recv) = target_conn.open_bi().await?;
                write_typed_message(&mut fwd_send, MessageType::RelayIntroduce, &forwarded_payload).await?;
                fwd_send.finish()?;

                let msg_type = read_message_type(&mut fwd_recv).await?;
                if msg_type != MessageType::RelayIntroduceResult {
                    anyhow::bail!("expected RelayIntroduceResult from target, got {:?}", msg_type);
                }
                let mut result: RelayIntroduceResultPayload = read_payload(&mut fwd_recv, MAX_PAYLOAD).await?;
                result.relay_available = relay_available;

                // Inject target's observed public address into the result
                if let Some(addr) = target_observed_addr {
                    let addr_str = addr.to_string();
                    if !result.target_addresses.contains(&addr_str) {
                        result.target_addresses.insert(0, addr_str);
                    }
                }

                anyhow::Ok(result)
            }.await;

            match forward_result {
                Ok(result) => {
                    write_typed_message(&mut send, MessageType::RelayIntroduceResult, &result).await?;
                    send.finish()?;
                    info!(
                        requester = hex::encode(payload.requester),
                        target = hex::encode(payload.target),
                        accepted = result.accepted,
                        target_observed = ?target_observed_addr,
                        requester_observed = ?requester_observed_addr,
                        "Relayed introduction (direct)"
                    );
                    self.log_activity(ActivityLevel::Info, ActivityCategory::Relay, format!("Forwarded introduction {} -> {}", &hex::encode(payload.requester)[..8], &hex::encode(payload.target)[..8]), None);
                }
                Err(e) => {
                    let result = RelayIntroduceResultPayload {
                        intro_id: payload.intro_id,
                        accepted: false,
                        target_addresses: vec![],
                        relay_available: false,
                        reject_reason: Some(format!("relay forward failed: {}", e)),
                    };
                    write_typed_message(&mut send, MessageType::RelayIntroduceResult, &result).await?;
                    send.finish()?;
                }
            }
            return Ok(());
        }

        // Check if target is connected via session (e.g. refused mesh but still reachable)
        if let Some(target_session) = self.sessions.get(&payload.target) {
            let target_conn = target_session.connection.clone();
            let target_observed_addr = target_session.remote_addr;
            let relay_available = self.can_accept_relay_pipe();

            // Look up requester's observed address from mesh or session
            let requester_observed_addr = self.connections.get(&from_peer)
                .and_then(|pc| pc.remote_addr)
                .or_else(|| self.sessions.get(&from_peer).and_then(|s| s.remote_addr));

            let mut forwarded_payload = payload.clone();
            if let Some(addr) = requester_observed_addr {
                let addr_str = addr.to_string();
                if !forwarded_payload.requester_addresses.contains(&addr_str) {
                    forwarded_payload.requester_addresses.insert(0, addr_str);
                    info!(
                        requester = hex::encode(payload.requester),
                        observed_addr = %addr,
                        "Relay injecting requester's observed public address (via session)"
                    );
                }
            }

            let forward_result = async {
                let (mut fwd_send, mut fwd_recv) = target_conn.open_bi().await?;
                write_typed_message(&mut fwd_send, MessageType::RelayIntroduce, &forwarded_payload).await?;
                fwd_send.finish()?;

                let msg_type = read_message_type(&mut fwd_recv).await?;
                if msg_type != MessageType::RelayIntroduceResult {
                    anyhow::bail!("expected RelayIntroduceResult from target (session), got {:?}", msg_type);
                }
                let mut result: RelayIntroduceResultPayload = read_payload(&mut fwd_recv, MAX_PAYLOAD).await?;
                result.relay_available = relay_available;

                // Inject target's observed public address into the result
                if let Some(addr) = target_observed_addr {
                    let addr_str = addr.to_string();
                    if !result.target_addresses.contains(&addr_str) {
                        result.target_addresses.insert(0, addr_str);
                    }
                }

                anyhow::Ok(result)
            }.await;

            match forward_result {
                Ok(result) => {
                    write_typed_message(&mut send, MessageType::RelayIntroduceResult, &result).await?;
                    send.finish()?;
                    info!(
                        requester = hex::encode(payload.requester),
                        target = hex::encode(payload.target),
                        accepted = result.accepted,
                        "Relayed introduction (via session)"
                    );
                    self.log_activity(ActivityLevel::Info, ActivityCategory::Relay, format!("Forwarded introduction {} -> {} (session)", &hex::encode(payload.requester)[..8], &hex::encode(payload.target)[..8]), None);
                }
                Err(e) => {
                    let result = RelayIntroduceResultPayload {
                        intro_id: payload.intro_id,
                        accepted: false,
                        target_addresses: vec![],
                        relay_available: false,
                        reject_reason: Some(format!("relay forward to session failed: {}", e)),
                    };
                    write_typed_message(&mut send, MessageType::RelayIntroduceResult, &result).await?;
                    send.finish()?;
                }
            }
            return Ok(());
        }

        // Target not directly connected — try forwarding with ttl
        if payload.ttl > 0 {
            // Find an N2 reporter for the target and forward with ttl-1
            let reporters = {
                let storage = self.storage.lock().await;
                storage.find_in_n2(&payload.target).unwrap_or_default()
            };

            for reporter in reporters {
                if reporter == from_peer || reporter == self.our_node_id {
                    continue;
                }
                if let Some(reporter_pc) = self.connections.get(&reporter) {
                    let reporter_conn = reporter_pc.connection.clone();
                    let relay_available = self.can_accept_relay_pipe();

                    // Inject requester's observed address if we have it
                    let mut req_addrs = payload.requester_addresses.clone();
                    if let Some(addr) = self.connections.get(&from_peer).and_then(|pc| pc.remote_addr) {
                        let addr_str = addr.to_string();
                        if !req_addrs.contains(&addr_str) {
                            req_addrs.insert(0, addr_str);
                        }
                    }

                    let forwarded_payload = RelayIntroducePayload {
                        intro_id: payload.intro_id,
                        target: payload.target,
                        requester: payload.requester,
                        requester_addresses: req_addrs,
                        ttl: payload.ttl - 1,
                    };

                    let forward_result = async {
                        let (mut fwd_send, mut fwd_recv) = reporter_conn.open_bi().await?;
                        write_typed_message(&mut fwd_send, MessageType::RelayIntroduce, &forwarded_payload).await?;
                        fwd_send.finish()?;

                        let msg_type = read_message_type(&mut fwd_recv).await?;
                        if msg_type != MessageType::RelayIntroduceResult {
                            anyhow::bail!("expected RelayIntroduceResult from chain, got {:?}", msg_type);
                        }
                        let mut result: RelayIntroduceResultPayload = read_payload(&mut fwd_recv, MAX_PAYLOAD).await?;
                        result.relay_available = relay_available;
                        anyhow::Ok(result)
                    }.await;

                    match forward_result {
                        Ok(result) => {
                            write_typed_message(&mut send, MessageType::RelayIntroduceResult, &result).await?;
                            send.finish()?;
                            info!(
                                requester = hex::encode(payload.requester),
                                target = hex::encode(payload.target),
                                via = hex::encode(reporter),
                                "Relayed introduction (chained)"
                            );
                            return Ok(());
                        }
                        Err(e) => {
                            debug!(reporter = hex::encode(reporter), error = %e, "Chain forward failed, trying next reporter");
                            continue;
                        }
                    }
                }
            }
        }

        // Cannot relay — target not reachable through us
        warn!(
            requester = hex::encode(payload.requester),
            target = hex::encode(payload.target),
            ttl = payload.ttl,
            our_connections = self.connections.len(),
            "Relay introduction: target not reachable through us"
        );
        self.log_activity(ActivityLevel::Warn, ActivityCategory::Relay, format!("Target {} not reachable through relay", &hex::encode(payload.target)[..8]), Some(payload.target));
        let result = RelayIntroduceResultPayload {
            intro_id: payload.intro_id,
            accepted: false,
            target_addresses: vec![],
            relay_available: false,
            reject_reason: Some("target not reachable through relay".to_string()),
        };
        write_typed_message(&mut send, MessageType::RelayIntroduceResult, &result).await?;
        send.finish()?;

        Ok(())
    }

    /// Handle an incoming SessionRelay request — splice two bi-streams.
    pub async fn handle_session_relay(
        conn_mgr: Arc<Mutex<Self>>,
        mut requester_recv: iroh::endpoint::RecvStream,
        mut requester_send: iroh::endpoint::SendStream,
        from_peer: NodeId,
    ) -> anyhow::Result<()> {
        let payload: SessionRelayPayload = read_payload(&mut requester_recv, 4096).await?;

        // Check capacity and find target connection
        let (target_conn, active_pipes) = {
            let cm = conn_mgr.lock().await;
            if !cm.can_accept_relay_pipe() {
                // Reject — at capacity
                let result = RelayIntroduceResultPayload {
                    intro_id: payload.intro_id,
                    accepted: false,
                    target_addresses: vec![],
                    relay_available: false,
                    reject_reason: Some("relay at capacity".to_string()),
                };
                write_typed_message(&mut requester_send, MessageType::RelayIntroduceResult, &result).await?;
                requester_send.finish()?;
                return Ok(());
            }

            let target_conn = cm.connections.get(&payload.target)
                .map(|pc| pc.connection.clone());
            (target_conn, Arc::clone(cm.active_relay_pipes()))
        };

        let target_conn = match target_conn {
            Some(c) => c,
            None => {
                let result = RelayIntroduceResultPayload {
                    intro_id: payload.intro_id,
                    accepted: false,
                    target_addresses: vec![],
                    relay_available: false,
                    reject_reason: Some("target not connected to relay".to_string()),
                };
                write_typed_message(&mut requester_send, MessageType::RelayIntroduceResult, &result).await?;
                requester_send.finish()?;
                return Ok(());
            }
        };

        // Open bi-stream to target
        let (mut target_send, mut target_recv) = target_conn.open_bi().await?;

        // Send SessionRelay message to target so they know what's happening
        write_typed_message(&mut target_send, MessageType::SessionRelay, &payload).await?;

        // Increment active pipe count
        active_pipes.fetch_add(1, Ordering::Relaxed);

        info!(
            requester = hex::encode(from_peer),
            target = hex::encode(payload.target),
            "Starting relay pipe"
        );

        // Run the copy loop with byte limit and idle timeout
        let result = Self::run_relay_pipe(
            &mut requester_recv,
            &mut requester_send,
            &mut target_recv,
            &mut target_send,
        ).await;

        // Decrement active pipe count
        active_pipes.fetch_sub(1, Ordering::Relaxed);

        match result {
            Ok(bytes) => {
                info!(
                    requester = hex::encode(from_peer),
                    target = hex::encode(payload.target),
                    bytes,
                    "Relay pipe closed"
                );
            }
            Err(e) => {
                debug!(
                    requester = hex::encode(from_peer),
                    target = hex::encode(payload.target),
                    error = %e,
                    "Relay pipe error"
                );
            }
        }

        Ok(())
    }

    /// Run a bidirectional relay pipe with byte limit and idle timeout.
    /// Returns total bytes transferred.
    async fn run_relay_pipe(
        requester_recv: &mut iroh::endpoint::RecvStream,
        requester_send: &mut iroh::endpoint::SendStream,
        target_recv: &mut iroh::endpoint::RecvStream,
        target_send: &mut iroh::endpoint::SendStream,
    ) -> anyhow::Result<u64> {
        let mut total_bytes: u64 = 0;
        let mut buf_a = vec![0u8; 16384];
        let mut buf_b = vec![0u8; 16384];

        loop {
            let idle_timeout = tokio::time::sleep(std::time::Duration::from_millis(RELAY_PIPE_IDLE_MS));

            tokio::select! {
                // requester → target
                result = requester_recv.read(&mut buf_a) => {
                    match result {
                        Ok(Some(n)) => {
                            total_bytes += n as u64;
                            if total_bytes > RELAY_MAX_BYTES {
                                anyhow::bail!("relay byte limit exceeded");
                            }
                            target_send.write_all(&buf_a[..n]).await?;
                        }
                        Ok(None) => {
                            // requester finished sending
                            let _ = target_send.finish();
                            break;
                        }
                        Err(e) => {
                            anyhow::bail!("requester recv error: {}", e);
                        }
                    }
                }
                // target → requester
                result = target_recv.read(&mut buf_b) => {
                    match result {
                        Ok(Some(n)) => {
                            total_bytes += n as u64;
                            if total_bytes > RELAY_MAX_BYTES {
                                anyhow::bail!("relay byte limit exceeded");
                            }
                            requester_send.write_all(&buf_b[..n]).await?;
                        }
                        Ok(None) => {
                            // target finished sending
                            let _ = requester_send.finish();
                            break;
                        }
                        Err(e) => {
                            anyhow::bail!("target recv error: {}", e);
                        }
                    }
                }
                _ = idle_timeout => {
                    anyhow::bail!("relay pipe idle timeout");
                }
            }
        }

        Ok(total_bytes)
    }

    /// Run the per-connection stream accept loop. Dispatches incoming streams by MessageType.
    /// `last_activity` is updated on each successful stream accept for zombie detection.
    pub async fn run_mesh_streams(
        conn_mgr: Arc<Mutex<Self>>,
        conn: iroh::endpoint::Connection,
        remote_node_id: NodeId,
        last_activity: Arc<AtomicU64>,
    ) {
        let our_stable_id = conn.stable_id();
        let keepalive_interval = std::time::Duration::from_secs(MESH_KEEPALIVE_INTERVAL_SECS);
        loop {
            tokio::select! {
                uni_result = conn.accept_uni() => {
                    match uni_result {
                        Ok(mut recv) => {
                            last_activity.store(now_ms(), Ordering::Relaxed);
                            let cm = Arc::clone(&conn_mgr);
                            let remote = remote_node_id;
                            tokio::spawn(async move {
                                if let Err(e) = Self::handle_uni_stream(&cm, &mut recv, remote).await {
                                    debug!(peer = hex::encode(remote), error = %e, "Uni-stream handler failed");
                                }
                            });
                        }
                        Err(e) => {
                            debug!(peer = hex::encode(remote_node_id), error = %e, "accept_uni failed, peer disconnected");
                            break;
                        }
                    }
                }
                bi_result = conn.accept_bi() => {
                    match bi_result {
                        Ok((send, recv)) => {
                            last_activity.store(now_ms(), Ordering::Relaxed);
                            let cm = Arc::clone(&conn_mgr);
                            let remote = remote_node_id;
                            tokio::spawn(async move {
                                if let Err(e) = Self::handle_bi_stream(&cm, recv, send, remote).await {
                                    debug!(peer = hex::encode(remote), error = %e, "Bi-stream handler failed");
                                }
                            });
                        }
                        Err(e) => {
                            debug!(peer = hex::encode(remote_node_id), error = %e, "accept_bi failed, peer disconnected");
                            break;
                        }
                    }
                }
                _ = tokio::time::sleep(keepalive_interval) => {
                    // Send lightweight keepalive ping — keeps NAT mapping alive
                    // and prevents zombie detection on the remote side
                    if let Ok(mut send) = conn.open_uni().await {
                        let _ = send.write_all(&[MessageType::MeshKeepalive.as_byte()]).await;
                        let _ = send.finish();
                        // Update our own last_activity so we don't zombie-detect this
                        // connection if the remote stops sending keepalives to us
                        last_activity.store(now_ms(), Ordering::Relaxed);
                    } else {
                        debug!(peer = hex::encode(remote_node_id), "Keepalive send failed, peer disconnected");
                        break;
                    }
                }
            }
        }

        // Connection ended — only clean up if this is still the active connection
        // (a reconnect may have already replaced our entry with a newer connection)
        let mut cm = conn_mgr.lock().await;
        let is_current = cm.connections.get(&remote_node_id)
            .map_or(false, |pc| pc.connection.stable_id() == our_stable_id);
        if is_current {
            cm.disconnect_peer(&remote_node_id).await;
        } else {
            debug!(peer = hex::encode(remote_node_id), "Skipping disconnect — connection was replaced by reconnect");
        }
    }

    // ---- Anchor referral methods ----

    /// Anchor-side: register a peer in the referral list.
    /// Uses the observed remote address from the peers table (the NAT-mapped public IP
    /// seen by the anchor from the QUIC connection) rather than self-reported addresses,
    /// which are often private/LAN IPs for NAT'd peers.
    pub async fn handle_anchor_register(&mut self, payload: AnchorRegisterPayload) {
        if !self.is_anchor.load(Ordering::Relaxed) {
            return;
        }
        if !self.connections.contains_key(&payload.node_id) && !self.sessions.contains_key(&payload.node_id) {
            debug!(peer = hex::encode(payload.node_id), "AnchorRegister from non-connected/session peer, ignoring");
            return;
        }

        // Prefer observed remote address (NAT-mapped public IP) over self-reported
        let addresses = {
            let storage = self.storage.lock().await;
            let observed = storage.get_peer_record(&payload.node_id)
                .ok().flatten()
                .map(|r| r.addresses).unwrap_or_default();
            if observed.is_empty() {
                // Fall back to self-reported addresses
                payload.addresses
            } else {
                // Use observed addresses (public IP as seen by anchor)
                observed.iter().map(|a| a.to_string()).collect()
            }
        };

        let now = now_ms();
        self.referral_list.insert(payload.node_id, ReferralEntry {
            node_id: payload.node_id,
            addresses,
            registered_at: now,
            use_count: 0,
            disconnected_at: None,
        });
        debug!(
            peer = hex::encode(payload.node_id),
            list_size = self.referral_list.len(),
            "Anchor: registered peer in referral list"
        );
    }

    /// Anchor-side: pick referrals from the list, applying tiered usage and self-pruning.
    pub fn pick_referrals(&mut self, exclude: &NodeId, count: usize) -> Vec<AnchorReferral> {
        let now = now_ms();

        // Prune: remove entries where disconnected_at is >2 min ago
        self.referral_list.retain(|_, entry| {
            match entry.disconnected_at {
                Some(disc_at) if now.saturating_sub(disc_at) > REFERRAL_DISCONNECT_GRACE_MS => false,
                _ => true,
            }
        });

        // Tiered usage policy
        let list_len = self.referral_list.len();
        let max_uses: u32 = if list_len < REFERRAL_LIST_CAP {
            3
        } else if list_len == REFERRAL_LIST_CAP {
            2
        } else {
            1
        };

        // Filter eligible entries
        let mut eligible: Vec<&NodeId> = self.referral_list.iter()
            .filter(|(nid, entry)| {
                *nid != exclude
                    && entry.use_count < max_uses
                    && entry.disconnected_at.is_none()
            })
            .map(|(nid, _)| nid)
            .collect();

        // Sort: lowest use_count first, then most recent registration
        eligible.sort_by(|a, b| {
            let ea = &self.referral_list[*a];
            let eb = &self.referral_list[*b];
            ea.use_count.cmp(&eb.use_count)
                .then(eb.registered_at.cmp(&ea.registered_at))
        });

        eligible.truncate(count);
        let picked_ids: Vec<NodeId> = eligible.into_iter().copied().collect();

        let mut result = Vec::with_capacity(picked_ids.len());
        for nid in &picked_ids {
            if let Some(entry) = self.referral_list.get_mut(nid) {
                entry.use_count += 1;
                result.push(AnchorReferral {
                    node_id: entry.node_id,
                    addresses: entry.addresses.clone(),
                });
            }
        }

        // Self-prune: remove entries that reached max_uses
        self.referral_list.retain(|_, entry| entry.use_count < max_uses);

        result
    }

    /// Mark a peer as disconnected in the referral list.
    pub fn mark_referral_disconnected(&mut self, node_id: &NodeId) {
        if let Some(entry) = self.referral_list.get_mut(node_id) {
            entry.disconnected_at = Some(now_ms());
        }
    }

    /// Mark a peer as reconnected in the referral list.
    pub fn mark_referral_reconnected(&mut self, node_id: &NodeId) {
        if let Some(entry) = self.referral_list.get_mut(node_id) {
            entry.disconnected_at = None;
        }
    }

    /// Anchor-side: handle a referral request (bi-stream).
    /// Supplements from mesh peers when the explicit referral list is sparse.
    pub async fn handle_anchor_referral_request(
        &mut self,
        payload: AnchorReferralRequestPayload,
        mut send: iroh::endpoint::SendStream,
    ) -> anyhow::Result<()> {
        // Also register the requester (they provide addresses, they're connected)
        self.handle_anchor_register(AnchorRegisterPayload {
            node_id: payload.requester,
            addresses: payload.requester_addresses,
        }).await;

        let mut referrals = self.pick_referrals(&payload.requester, 3);

        // Auto-refer from mesh peers when referral list is sparse
        if referrals.len() < 3 {
            let referred_ids: std::collections::HashSet<NodeId> = referrals.iter().map(|r| r.node_id).collect();
            let mut mesh_candidates: Vec<_> = self.connections.iter()
                .filter(|(nid, _)| **nid != payload.requester && !referred_ids.contains(*nid) && **nid != self.our_node_id)
                .filter_map(|(nid, pc)| pc.remote_addr.map(|a| (*nid, a.to_string())))
                .collect();
            // Shuffle for variety
            use rand::seq::SliceRandom;
            mesh_candidates.shuffle(&mut rand::rng());
            for (nid, addr) in mesh_candidates.into_iter().take(3 - referrals.len()) {
                referrals.push(AnchorReferral { node_id: nid, addresses: vec![addr] });
            }
        }

        info!(
            requester = hex::encode(payload.requester),
            referral_count = referrals.len(),
            "Anchor: serving referrals"
        );

        let response = AnchorReferralResponsePayload { referrals };
        write_typed_message(&mut send, MessageType::AnchorReferralResponse, &response).await?;
        send.finish()?;
        Ok(())
    }

    /// Client-side: request referrals from an anchor peer (mesh or session).
    pub async fn request_anchor_referrals(
        &mut self,
        anchor_peer: &NodeId,
    ) -> anyhow::Result<Vec<AnchorReferral>> {
        let conn = if let Some(pc) = self.connections.get(anchor_peer) {
            pc.connection.clone()
        } else if let Some(session) = self.sessions.get(anchor_peer) {
            session.connection.clone()
        } else {
            anyhow::bail!("anchor peer not connected (mesh or session)");
        };

        let our_addrs: Vec<String> = self.endpoint.addr().ip_addrs()
            .map(|s| s.to_string())
            .collect();

        let request = AnchorReferralRequestPayload {
            requester: self.our_node_id,
            requester_addresses: our_addrs,
        };

        let (mut send, mut recv) = conn.open_bi().await?;
        write_typed_message(&mut send, MessageType::AnchorReferralRequest, &request).await?;
        send.finish()?;

        let msg_type = read_message_type(&mut recv).await?;
        if msg_type != MessageType::AnchorReferralResponse {
            anyhow::bail!("expected AnchorReferralResponse, got {:?}", msg_type);
        }
        let response: AnchorReferralResponsePayload = read_payload(&mut recv, 4096).await?;

        // Touch session last_active to prevent idle reaping
        if let Some(session) = self.sessions.get_mut(anchor_peer) {
            session.last_active_at = now_ms();
        }

        Ok(response.referrals)
    }

    /// Client-side: register our address with an anchor peer (mesh or session).
    pub async fn send_anchor_register(&mut self, anchor_peer: &NodeId) -> anyhow::Result<()> {
        let conn = if let Some(pc) = self.connections.get(anchor_peer) {
            pc.connection.clone()
        } else if let Some(session) = self.sessions.get(anchor_peer) {
            session.connection.clone()
        } else {
            anyhow::bail!("anchor peer not connected (mesh or session)");
        };

        let mut our_addrs: Vec<String> = self.endpoint.addr().ip_addrs()
            .map(|s| s.to_string())
            .collect();
        // Prepend UPnP external address (most useful for remote peers)
        if let Some(ref ext) = self.upnp_external_addr {
            let ext_str = ext.to_string();
            if !our_addrs.contains(&ext_str) {
                our_addrs.insert(0, ext_str);
            }
        }
        // Prepend stable anchor advertised address
        if let Some(anchor_addr) = self.build_anchor_advertised_addr() {
            if !our_addrs.contains(&anchor_addr) {
                our_addrs.insert(0, anchor_addr);
            }
        }

        let payload = AnchorRegisterPayload {
            node_id: self.our_node_id,
            addresses: our_addrs,
        };

        let mut send = conn.open_uni().await?;
        write_typed_message(&mut send, MessageType::AnchorRegister, &payload).await?;
        send.finish()?;

        // Touch session last_active to prevent idle reaping
        if let Some(session) = self.sessions.get_mut(anchor_peer) {
            session.last_active_at = now_ms();
        }

        debug!(anchor = hex::encode(anchor_peer), "Registered with anchor");

        Ok(())
    }

    /// Handle an incoming NAT filter probe request (anchor side).
    /// Static method — does NOT hold conn_mgr lock during the 2s probe.
    pub async fn handle_nat_filter_probe_static(
        payload: crate::protocol::NatFilterProbePayload,
        observed_addr: Option<std::net::SocketAddr>,
        mut send: iroh::endpoint::SendStream,
    ) -> anyhow::Result<()> {
        let observed = match observed_addr {
            Some(addr) => addr,
            None => {
                info!(peer = hex::encode(payload.node_id), "NAT filter probe: no observed address, returning unreachable");
                let result = crate::protocol::NatFilterProbeResultPayload { reachable: false };
                write_typed_message(&mut send, MessageType::NatFilterProbeResult, &result).await?;
                send.finish()?;
                return Ok(());
            }
        };

        info!(
            peer = hex::encode(payload.node_id),
            observed = %observed,
            "NAT filter probe: testing reachability from different port"
        );

        // Create a temporary endpoint on a random port and try to connect
        let reachable = match Self::probe_from_different_port(&payload.node_id, observed).await {
            Ok(r) => r,
            Err(e) => {
                debug!(error = %e, "NAT filter probe: test failed");
                false
            }
        };

        info!(
            peer = hex::encode(payload.node_id),
            reachable,
            "NAT filter probe result — sending response"
        );

        let result = crate::protocol::NatFilterProbeResultPayload { reachable };
        write_typed_message(&mut send, MessageType::NatFilterProbeResult, &result).await?;
        send.finish()?;
        info!(peer = hex::encode(payload.node_id), "NAT filter probe response sent");
        Ok(())
    }

    /// Try to connect to a peer from a fresh temporary endpoint (different source port).
    /// Returns true if we could reach them (address-restricted or better),
    /// false if we couldn't (port-restricted).
    async fn probe_from_different_port(
        target: &NodeId,
        observed_addr: std::net::SocketAddr,
    ) -> anyhow::Result<bool> {
        // Build a temporary endpoint with a random port
        let secret_key = iroh::SecretKey::generate(&mut rand::rng());
        let temp_ep = iroh::Endpoint::builder()
            .alpns(vec![ALPN_V2.to_vec()])
            .secret_key(secret_key)
            .bind()
            .await?;

        let eid = iroh::EndpointId::from_bytes(target)?;
        let addr = iroh::EndpointAddr::from(eid).with_ip_addr(observed_addr);

        // Try connecting with a short timeout (2s)
        let result = tokio::time::timeout(
            std::time::Duration::from_secs(2),
            temp_ep.connect(addr, ALPN_V2),
        ).await;

        // Clean up the temporary endpoint
        temp_ep.close().await;

        match result {
            Ok(Ok(_conn)) => Ok(true),   // reached them — Open filtering
            Ok(Err(_)) => Ok(false),      // connection error — likely PortRestricted
            Err(_) => Ok(false),          // timeout — PortRestricted
        }
    }

    pub async fn handle_uni_stream(
        conn_mgr: &Arc<Mutex<Self>>,
        recv: &mut iroh::endpoint::RecvStream,
        remote_node_id: NodeId,
    ) -> anyhow::Result<()> {
        let msg_type = read_message_type(recv).await?;

        match msg_type {
            MessageType::NodeListUpdate => {
                let diff: NodeListUpdatePayload = read_payload(recv, MAX_PAYLOAD).await?;
                let has_n1_additions = !diff.n1_added.is_empty();
                let cm = conn_mgr.lock().await;
                let count = cm.process_routing_diff(&remote_node_id, diff).await?;
                if count > 0 {
                    debug!(peer = hex::encode(remote_node_id), count, "Applied node list update");
                }
                if has_n1_additions {
                    cm.notify_growth();
                }
            }
            MessageType::ProfileUpdate => {
                let payload: ProfileUpdatePayload = read_payload(recv, MAX_PAYLOAD).await?;
                let cm = conn_mgr.lock().await;
                let storage = cm.storage.lock().await;
                for profile in payload.profiles {
                    let _ = storage.store_profile(&profile);
                }
            }
            MessageType::DeleteRecord => {
                let payload: crate::protocol::DeleteRecordPayload =
                    read_payload(recv, MAX_PAYLOAD).await?;
                let cm = conn_mgr.lock().await;

                // Collect blob CIDs + CDN peers before async work
                let mut blob_cleanup: Vec<([u8; 32], Vec<(NodeId, Vec<String>)>, Option<(NodeId, Vec<String>)>)> = Vec::new();
                {
                    let storage = cm.storage.lock().await;
                    for dr in &payload.records {
                        if crypto::verify_delete_signature(&dr.author, &dr.post_id, &dr.signature) {
                            // Collect blobs for CDN cleanup before deleting
                            let blob_cids = storage.get_blobs_for_post(&dr.post_id).unwrap_or_default();
                            for cid in blob_cids {
                                let downstream = storage.get_blob_downstream(&cid).unwrap_or_default();
                                let upstream = storage.get_blob_upstream(&cid).ok().flatten();
                                blob_cleanup.push((cid, downstream, upstream));
                            }
                            let _ = storage.store_delete(dr);
                            let _ = storage.apply_delete(dr);
                            // Delete blob metadata + CDN metadata
                            let deleted_cids = storage.delete_blobs_for_post(&dr.post_id).unwrap_or_default();
                            for cid in &deleted_cids {
                                let _ = storage.cleanup_cdn_for_blob(cid);
                                let _ = cm.blob_store.delete(cid);
                            }
                        }
                    }
                }

                // Send CDN delete notices (async, best-effort)
                for (cid, downstream, upstream) in &blob_cleanup {
                    // Notify downstream (with our upstream info for tree healing)
                    let upstream_info = upstream.as_ref().map(|(nid, addrs)| {
                        PeerWithAddress {
                            n: hex::encode(nid),
                            a: addrs.clone(),
                        }
                    });
                    let ds_payload = crate::protocol::BlobDeleteNoticePayload {
                        cid: *cid,
                        upstream_node: upstream_info,
                    };
                    for (ds_nid, _) in downstream {
                        if let Some(pc) = cm.connections_ref().get(ds_nid) {
                            if let Ok(mut send) = pc.connection.open_uni().await {
                                let _ = write_typed_message(&mut send, MessageType::BlobDeleteNotice, &ds_payload).await;
                                let _ = send.finish();
                            }
                        }
                    }
                    // Notify upstream (no upstream info — just "remove me")
                    if let Some((up_nid, _)) = upstream {
                        let up_payload = crate::protocol::BlobDeleteNoticePayload {
                            cid: *cid,
                            upstream_node: None,
                        };
                        if let Some(pc) = cm.connections_ref().get(up_nid) {
                            if let Ok(mut send) = pc.connection.open_uni().await {
                                let _ = write_typed_message(&mut send, MessageType::BlobDeleteNotice, &up_payload).await;
                                let _ = send.finish();
                            }
                        }
                    }
                }
            }
            MessageType::VisibilityUpdate => {
                let payload: crate::protocol::VisibilityUpdatePayload =
                    read_payload(recv, MAX_PAYLOAD).await?;
                let cm = conn_mgr.lock().await;
                let storage = cm.storage.lock().await;
                for vu in payload.updates {
                    if let Some(post) = storage.get_post(&vu.post_id)? {
                        if post.author == vu.author {
                            let _ = storage.update_post_visibility(&vu.post_id, &vu.visibility);
                        }
                    }
                }
            }
            MessageType::PostNotification => {
                let notification: PostNotificationPayload =
                    read_payload(recv, MAX_PAYLOAD).await?;
                info!(
                    peer = hex::encode(remote_node_id),
                    post_id = hex::encode(notification.post_id),
                    author = hex::encode(notification.author),
                    "Received post notification"
                );
                let cm = conn_mgr.lock().await;
                match cm.handle_post_notification(&remote_node_id, notification, None).await {
                    Ok(true) => {
                        info!(peer = hex::encode(remote_node_id), "Pulled post from notification");
                    }
                    Ok(false) => {
                        info!(peer = hex::encode(remote_node_id), "Post notification ignored (not following or already have)");
                    }
                    Err(e) => {
                        warn!(peer = hex::encode(remote_node_id), error = %e, "Post notification pull failed");
                    }
                }
            }
            MessageType::PostPush => {
                let push: PostPushPayload = read_payload(recv, MAX_PAYLOAD).await?;
                let cm = conn_mgr.lock().await;
                let storage = cm.storage.lock().await;
                if !storage.is_deleted(&push.post.id)?
                    && storage.get_post(&push.post.id)?.is_none()
                    && crate::content::verify_post_id(&push.post.id, &push.post.post)
                {
                    let _ = storage.store_post_with_visibility(
                        &push.post.id,
                        &push.post.post,
                        &push.post.visibility,
                    );
                    let prio = storage.get_post_upstreams(&push.post.id).map(|v| v.len() as u8).unwrap_or(0);
                    let _ = storage.add_post_upstream(&push.post.id, &remote_node_id, prio);
                    info!(
                        peer = hex::encode(remote_node_id),
                        post_id = hex::encode(push.post.id),
                        "Received direct post push"
                    );
                }
            }
            MessageType::AudienceRequest => {
                let req: AudienceRequestPayload = read_payload(recv, MAX_PAYLOAD).await?;
                info!(
                    peer = hex::encode(remote_node_id),
                    requester = hex::encode(req.requester),
                    "Received audience request"
                );
                let cm = conn_mgr.lock().await;
                let storage = cm.storage.lock().await;
                // Store as inbound pending request
                let _ = storage.store_audience(
                    &req.requester,
                    crate::types::AudienceDirection::Inbound,
                    crate::types::AudienceStatus::Pending,
                );
            }
            MessageType::AudienceResponse => {
                let resp: AudienceResponsePayload = read_payload(recv, MAX_PAYLOAD).await?;
                let status = if resp.approved { "approved" } else { "denied" };
                info!(
                    peer = hex::encode(remote_node_id),
                    responder = hex::encode(resp.responder),
                    status,
                    "Received audience response"
                );
                let cm = conn_mgr.lock().await;
                let storage = cm.storage.lock().await;
                let new_status = if resp.approved {
                    crate::types::AudienceStatus::Approved
                } else {
                    crate::types::AudienceStatus::Denied
                };
                let _ = storage.store_audience(
                    &resp.responder,
                    crate::types::AudienceDirection::Outbound,
                    new_status,
                );
            }
            MessageType::SocialAddressUpdate => {
                let payload: SocialAddressUpdatePayload = read_payload(recv, MAX_PAYLOAD).await?;
                let cm = conn_mgr.lock().await;
                let storage = cm.storage.lock().await;
                if storage.has_social_route(&payload.node_id).unwrap_or(false) {
                    let addrs: Vec<std::net::SocketAddr> = payload.addresses.iter()
                        .filter_map(|a| a.parse().ok()).collect();
                    let _ = storage.touch_social_route_connect(
                        &payload.node_id, &addrs, ReachMethod::Referral,
                    );
                    let _ = storage.update_social_route_peer_addrs(
                        &payload.node_id, &payload.peer_addresses,
                    );
                }
                debug!(
                    peer = hex::encode(remote_node_id),
                    target = hex::encode(payload.node_id),
                    "Received social address update"
                );
            }
            MessageType::ManifestPush => {
                let payload: crate::protocol::ManifestPushPayload = read_payload(recv, MAX_PAYLOAD).await?;
                let cm = conn_mgr.lock().await;
                let storage = cm.storage.lock().await;
                let mut stored_entries: Vec<crate::protocol::ManifestPushEntry> = Vec::new();
                for entry in &payload.manifests {
                    if !crate::crypto::verify_manifest_signature(&entry.manifest.author_manifest) {
                        continue;
                    }
                    // Only store if newer than what we have
                    let dominated = storage.get_cdn_manifest(&entry.cid).ok().flatten()
                        .and_then(|json| serde_json::from_str::<crate::types::CdnManifest>(&json).ok())
                        .map(|existing| existing.author_manifest.updated_at >= entry.manifest.author_manifest.updated_at)
                        .unwrap_or(false);
                    if dominated {
                        continue;
                    }
                    let manifest_json = match serde_json::to_string(&entry.manifest) {
                        Ok(j) => j,
                        Err(_) => continue,
                    };
                    let _ = storage.store_cdn_manifest(
                        &entry.cid,
                        &manifest_json,
                        &entry.manifest.author_manifest.author,
                        entry.manifest.author_manifest.updated_at,
                    );
                    stored_entries.push(entry.clone());
                }
                // Gather downstream peers for relay before dropping locks
                let mut relay_targets: Vec<(NodeId, crate::protocol::ManifestPushPayload)> = Vec::new();
                for entry in &stored_entries {
                    let downstream = storage.get_blob_downstream(&entry.cid).unwrap_or_default();
                    for (ds_nid, _) in downstream {
                        if ds_nid == remote_node_id {
                            continue;
                        }
                        relay_targets.push((ds_nid, crate::protocol::ManifestPushPayload {
                            manifests: vec![entry.clone()],
                        }));
                    }
                }
                let stored = stored_entries.len();

                // Phase 5: Gather post IDs from manifests for discovery
                let our_follows: std::collections::HashSet<NodeId> =
                    storage.list_follows().unwrap_or_default().into_iter().collect();
                let mut discovery_posts: Vec<(PostId, NodeId)> = Vec::new();
                for entry in &stored_entries {
                    let am = &entry.manifest.author_manifest;
                    let author = am.author;
                    // Collect post IDs from the manifest's neighborhood
                    let mut candidate_ids: Vec<PostId> = Vec::new();
                    candidate_ids.push(am.post_id);
                    for me in &am.previous_posts {
                        candidate_ids.push(me.post_id);
                    }
                    for me in &am.following_posts {
                        candidate_ids.push(me.post_id);
                    }
                    for pid in candidate_ids {
                        // Only discover posts from authors we follow
                        if !our_follows.contains(&author) {
                            continue;
                        }
                        // Only discover posts we don't have locally
                        if storage.get_post(&pid).ok().flatten().is_some() {
                            continue;
                        }
                        discovery_posts.push((pid, author));
                        // Cap at 10 posts per manifest push to avoid storms
                        if discovery_posts.len() >= 10 {
                            break;
                        }
                    }
                    if discovery_posts.len() >= 10 {
                        break;
                    }
                }

                drop(storage);
                // Relay to downstream (best-effort via mesh connections)
                for (ds_nid, relay_payload) in &relay_targets {
                    if let Some(pc) = cm.connections_ref().get(ds_nid) {
                        if let Ok(mut send) = pc.connection.open_uni().await {
                            let _ = write_typed_message(&mut send, MessageType::ManifestPush, relay_payload).await;
                            let _ = send.finish();
                        }
                    }
                }

                // Phase 5: Spawn post discovery task (non-blocking)
                if !discovery_posts.is_empty() {
                    let cm_arc = conn_mgr.clone();
                    let sender_id = remote_node_id;
                    tokio::spawn(async move {
                        // Brief lock: get connection handle only
                        let conn = {
                            let cm = cm_arc.lock().await;
                            cm.connections_ref().get(&sender_id).map(|pc| pc.connection.clone())
                        };
                        // cm lock RELEASED

                        let Some(conn) = conn else { return };

                        let mut fetched = 0usize;
                        for (post_id, _author) in &discovery_posts {
                            if fetched >= 10 { break; }

                            // PostFetch network I/O WITHOUT any lock
                            let result = async {
                                use crate::protocol::{PostFetchRequestPayload, PostFetchResponsePayload};
                                let (mut send, mut recv) = conn.open_bi().await?;
                                let req = PostFetchRequestPayload { post_id: *post_id };
                                write_typed_message(&mut send, MessageType::PostFetchRequest, &req).await?;
                                send.finish()?;
                                let msg_type = tokio::time::timeout(
                                    std::time::Duration::from_secs(10),
                                    read_message_type(&mut recv),
                                ).await??;
                                if msg_type != MessageType::PostFetchResponse {
                                    return anyhow::Ok(None);
                                }
                                let resp: PostFetchResponsePayload = read_payload(&mut recv, MAX_PAYLOAD).await?;
                                anyhow::Ok(resp.post)
                            }.await;

                            match result {
                                Ok(Some(sync_post)) => {
                                    if crate::content::verify_post_id(&sync_post.id, &sync_post.post) {
                                        // Brief re-acquire for storage writes only
                                        let stored = {
                                            let cm = cm_arc.lock().await;
                                            let storage = cm.storage.lock().await;
                                            if storage.store_post_with_visibility(&sync_post.id, &sync_post.post, &sync_post.visibility).unwrap_or(false) {
                                                let prio = storage.get_post_upstreams(&sync_post.id).map(|v| v.len() as u8).unwrap_or(0);
                                                let _ = storage.add_post_upstream(&sync_post.id, &sender_id, prio);
                                                let now = std::time::SystemTime::now()
                                                    .duration_since(std::time::UNIX_EPOCH)
                                                    .unwrap_or_default()
                                                    .as_millis() as u64;
                                                let _ = storage.update_follow_last_sync(&sync_post.post.author, now);
                                                true
                                            } else {
                                                false
                                            }
                                        };
                                        // cm lock RELEASED — register downstream without lock
                                        if stored {
                                            let reg = crate::protocol::PostDownstreamRegisterPayload { post_id: sync_post.id };
                                            if let Ok(mut send) = conn.open_uni().await {
                                                let _ = write_typed_message(&mut send, MessageType::PostDownstreamRegister, &reg).await;
                                                let _ = send.finish();
                                            }
                                            fetched += 1;
                                        }
                                    }
                                }
                                _ => {}
                            }
                        }
                        if fetched > 0 {
                            debug!(discovered = fetched, "ManifestPush post discovery");
                        }
                    });
                }

                drop(cm);
                debug!(peer = hex::encode(remote_node_id), stored, relayed = relay_targets.len(), "Received manifest push");
            }
            MessageType::SocialDisconnectNotice => {
                let payload: SocialDisconnectNoticePayload = read_payload(recv, MAX_PAYLOAD).await?;
                let cm = conn_mgr.lock().await;
                let storage = cm.storage.lock().await;
                if storage.has_social_route(&payload.node_id).unwrap_or(false) {
                    let _ = storage.set_social_route_status(&payload.node_id, SocialStatus::Disconnected);
                }
                debug!(
                    peer = hex::encode(remote_node_id),
                    target = hex::encode(payload.node_id),
                    "Received social disconnect notice"
                );
            }
            MessageType::BlobDeleteNotice => {
                let payload: crate::protocol::BlobDeleteNoticePayload =
                    read_payload(recv, MAX_PAYLOAD).await?;
                let cm = conn_mgr.lock().await;
                let storage = cm.storage.lock().await;
                let cid = payload.cid;

                // Check if sender was our upstream for this blob
                let was_upstream = storage.get_blob_upstream(&cid).ok().flatten()
                    .map(|(nid, _)| nid == remote_node_id)
                    .unwrap_or(false);

                if was_upstream {
                    // Sender was our upstream — clear it
                    let _ = storage.remove_blob_upstream(&cid);

                    // If they provided their upstream, store it as our new upstream
                    if let Some(ref new_up) = payload.upstream_node {
                        if let Ok(nid_bytes) = hex::decode(&new_up.n) {
                            if let Ok(nid) = <[u8; 32]>::try_from(nid_bytes.as_slice()) {
                                let _ = storage.store_blob_upstream(&cid, &nid, &new_up.a);
                            }
                        }
                    }
                } else {
                    // Sender was our downstream — remove them
                    let _ = storage.remove_blob_downstream(&cid, &remote_node_id);
                }

                info!(
                    peer = hex::encode(remote_node_id),
                    cid = hex::encode(cid),
                    was_upstream,
                    "Received blob delete notice"
                );
            }
            MessageType::GroupKeyDistribute => {
                let payload: GroupKeyDistributePayload = read_payload(recv, MAX_PAYLOAD).await?;
                let cm = conn_mgr.lock().await;

                // Verify the sender is the admin
                if payload.admin != remote_node_id {
                    warn!(peer = hex::encode(remote_node_id), "GroupKeyDistribute from non-admin, ignoring");
                } else {
                    let storage = cm.storage.lock().await;
                    let record = crate::types::GroupKeyRecord {
                        group_id: payload.group_id,
                        circle_name: payload.circle_name.clone(),
                        epoch: payload.epoch,
                        group_public_key: payload.group_public_key,
                        admin: payload.admin,
                        created_at: now_ms(),
                    };
                    let _ = storage.create_group_key(&record, None);

                    // Find our wrapped key and unwrap the group seed
                    for mk in &payload.member_keys {
                        let _ = storage.store_group_member_key(&payload.group_id, mk);
                        if mk.member == cm.our_node_id {
                            match crypto::unwrap_group_key(
                                &cm.secret_seed,
                                &payload.admin,
                                &mk.wrapped_group_key,
                            ) {
                                Ok(seed) => {
                                    let _ = storage.store_group_seed(&payload.group_id, payload.epoch, &seed);
                                    info!(
                                        circle = %payload.circle_name,
                                        epoch = payload.epoch,
                                        "Received and unwrapped group key"
                                    );
                                }
                                Err(e) => {
                                    warn!(error = %e, "Failed to unwrap group key");
                                }
                            }
                        }
                    }
                }
            }
            MessageType::CircleProfileUpdate => {
                let payload: CircleProfileUpdatePayload = read_payload(recv, MAX_PAYLOAD).await?;
                let cm = conn_mgr.lock().await;

                // Try to decrypt if we have the group seed
                let storage = cm.storage.lock().await;
                let decrypted = storage
                    .get_group_seed(&payload.group_id, payload.epoch)
                    .ok()
                    .flatten()
                    .and_then(|seed| {
                        let gk = storage.get_group_key(&payload.group_id).ok()??;
                        let json = crypto::decrypt_group_post(
                            &payload.encrypted_payload,
                            &seed,
                            &gk.group_public_key,
                            &payload.wrapped_cek,
                        )
                        .ok()?;
                        // Tombstone: empty string means deleted
                        if json.is_empty() {
                            let _ = storage.delete_circle_profile(&payload.author, &payload.circle_name);
                            return None;
                        }
                        serde_json::from_str::<crate::types::CircleProfile>(&json).ok()
                    });

                if let Some(cp) = decrypted {
                    // Store decrypted + encrypted form
                    let _ = storage.store_remote_circle_profile(
                        &payload.author,
                        &payload.circle_name,
                        &cp,
                        &payload.encrypted_payload,
                        &payload.wrapped_cek,
                        &payload.group_id,
                        payload.epoch,
                    );
                    debug!(
                        peer = hex::encode(remote_node_id),
                        author = hex::encode(payload.author),
                        circle = %payload.circle_name,
                        "Decrypted and stored circle profile"
                    );
                } else {
                    // Can't decrypt — store encrypted form only for relay
                    let _ = storage.store_encrypted_circle_profile(
                        &payload.author,
                        &payload.circle_name,
                        &payload.encrypted_payload,
                        &payload.wrapped_cek,
                        &payload.group_id,
                        payload.epoch,
                        payload.updated_at,
                    );
                    debug!(
                        peer = hex::encode(remote_node_id),
                        author = hex::encode(payload.author),
                        circle = %payload.circle_name,
                        "Stored encrypted circle profile (no group key)"
                    );
                }

                // Relay to other connected peers (gossip)
                for (peer_id, mc) in cm.connections_ref() {
                    if *peer_id == remote_node_id {
                        continue;
                    }
                    if let Ok(mut send) = mc.connection.open_uni().await {
                        let _ = write_typed_message(
                            &mut send,
                            MessageType::CircleProfileUpdate,
                            &payload,
                        )
                        .await;
                        let _ = send.finish();
                    }
                }
            }
            MessageType::AnchorRegister => {
                let payload: AnchorRegisterPayload = read_payload(recv, 4096).await?;
                let mut cm = conn_mgr.lock().await;
                cm.handle_anchor_register(payload).await;
            }
            MessageType::MeshKeepalive => {
                // No-op — last_activity already updated on stream accept
            }
            MessageType::BlobHeaderDiff => {
                let payload: BlobHeaderDiffPayload = read_payload(recv, MAX_PAYLOAD).await?;
                let cm = conn_mgr.lock().await;
                cm.handle_blob_header_diff(payload, remote_node_id).await;
            }
            MessageType::PostDownstreamRegister => {
                let payload: PostDownstreamRegisterPayload = read_payload(recv, MAX_PAYLOAD).await?;
                let cm = conn_mgr.lock().await;
                let storage = cm.storage.lock().await;
                let _ = storage.add_post_downstream(&payload.post_id, &remote_node_id);
                drop(storage);
                trace!(
                    peer = hex::encode(remote_node_id),
                    post = hex::encode(payload.post_id),
                    "Registered as post downstream"
                );
            }
            other => {
                warn!(msg_type = ?other, "Unexpected message type on uni-stream");
            }
        }

        Ok(())
    }

    async fn handle_bi_stream(
        conn_mgr: &Arc<Mutex<Self>>,
        mut recv: iroh::endpoint::RecvStream,
        send: iroh::endpoint::SendStream,
        remote_node_id: NodeId,
    ) -> anyhow::Result<()> {
        let msg_type = read_message_type(&mut recv).await?;
        Self::handle_bi_stream_typed(conn_mgr, recv, send, remote_node_id, msg_type).await
    }

    /// Handle a bi-stream where the message type has already been read.
    /// Used by handle_incoming_connection (ephemeral accept loop).
    pub async fn handle_bi_stream_typed(
        conn_mgr: &Arc<Mutex<Self>>,
        mut recv: iroh::endpoint::RecvStream,
        mut send: iroh::endpoint::SendStream,
        remote_node_id: NodeId,
        msg_type: MessageType,
    ) -> anyhow::Result<()> {
        match msg_type {
            MessageType::PullSyncRequest => {
                let cm = conn_mgr.lock().await;
                cm.handle_pull_request(remote_node_id, recv, send).await?;
            }
            MessageType::InitialExchange => {
                let (storage, our_node_id, anchor_addr, our_nat_type, our_http_capable, our_http_addr) = {
                    let cm = conn_mgr.lock().await;
                    (cm.storage_ref(), *cm.our_node_id(), cm.build_anchor_advertised_addr(), cm.nat_type(), cm.http_capable, cm.http_addr.clone())
                };
                initial_exchange_accept(&storage, &our_node_id, send, recv, remote_node_id, anchor_addr, None, our_nat_type, our_http_capable, our_http_addr, None, None)
                    .await?;
            }
            MessageType::AddressRequest => {
                let cm = conn_mgr.lock().await;
                cm.handle_address_request(recv, send, remote_node_id).await?;
            }
            MessageType::SocialCheckin => {
                let payload: SocialCheckinPayload = read_payload(&mut recv, MAX_PAYLOAD).await?;
                let reply = {
                    let cm = conn_mgr.lock().await;
                    let storage = cm.storage.lock().await;
                    // Update their social route
                    if storage.has_social_route(&payload.node_id).unwrap_or(false) {
                        let addrs: Vec<std::net::SocketAddr> = payload.addresses.iter()
                            .filter_map(|a| a.parse().ok()).collect();
                        let _ = storage.touch_social_route_connect(
                            &payload.node_id, &addrs, ReachMethod::Direct,
                        );
                        let _ = storage.update_social_route_peer_addrs(
                            &payload.node_id, &payload.peer_addresses,
                        );
                    }
                    // Build reply (gather data before dropping locks)
                    let our_addrs: Vec<String> = cm.endpoint.addr().ip_addrs()
                        .map(|s| s.to_string()).collect();
                    SocialCheckinPayload {
                        node_id: cm.our_node_id,
                        addresses: our_addrs,
                        peer_addresses: vec![],
                    }
                };
                write_typed_message(&mut send, MessageType::SocialCheckin, &reply).await?;
                send.finish()?;
                debug!(peer = hex::encode(remote_node_id), "Handled social checkin");
            }
            MessageType::WormQuery => {
                let payload: WormQueryPayload = read_payload(&mut recv, MAX_PAYLOAD).await?;
                debug!(
                    peer = hex::encode(remote_node_id),
                    target = hex::encode(payload.target),
                    ttl = payload.ttl,
                    "Received worm query"
                );
                let mut cm = conn_mgr.lock().await;
                cm.handle_worm_query(payload, send, remote_node_id).await?;
            }
            MessageType::PostFetchRequest => {
                let payload: crate::protocol::PostFetchRequestPayload = read_payload(&mut recv, 4096).await?;
                debug!(
                    peer = hex::encode(remote_node_id),
                    post = hex::encode(payload.post_id),
                    "Received PostFetch request"
                );
                let cm = conn_mgr.lock().await;
                let result = {
                    let store = cm.storage.lock().await;
                    store.get_post_with_visibility(&payload.post_id).ok().flatten()
                };
                let resp = if let Some((post, visibility)) = result {
                    if matches!(visibility, PostVisibility::Public) {
                        crate::protocol::PostFetchResponsePayload {
                            post_id: payload.post_id,
                            found: true,
                            post: Some(SyncPost {
                                id: payload.post_id,
                                post,
                                visibility,
                            }),
                        }
                    } else {
                        crate::protocol::PostFetchResponsePayload {
                            post_id: payload.post_id,
                            found: false,
                            post: None,
                        }
                    }
                } else {
                    crate::protocol::PostFetchResponsePayload {
                        post_id: payload.post_id,
                        found: false,
                        post: None,
                    }
                };
                write_typed_message(&mut send, MessageType::PostFetchResponse, &resp).await?;
                send.finish()?;
            }
            MessageType::TcpPunchRequest => {
                let payload: crate::protocol::TcpPunchRequestPayload = read_payload(&mut recv, 4096).await?;
                info!(
                    peer = hex::encode(remote_node_id),
                    browser_ip = %payload.browser_ip,
                    post = hex::encode(payload.post_id),
                    "Received TcpPunch request"
                );
                // Validate: we hold this post and it's public
                let (valid, http_port, http_addr) = {
                    let cm = conn_mgr.lock().await;
                    let has_post = {
                        let store = cm.storage.lock().await;
                        store.get_post_with_visibility(&payload.post_id)
                            .ok().flatten()
                            .map(|(_, v)| matches!(v, PostVisibility::Public))
                            .unwrap_or(false)
                    };
                    let port = cm.endpoint.bound_sockets().first()
                        .map(|s| s.port()).unwrap_or(0);
                    (has_post && cm.http_capable, port, cm.http_addr.clone())
                };
                let resp = if valid {
                    // Parse browser IP and execute TCP punch
                    if let Ok(browser_ip) = payload.browser_ip.parse::<std::net::IpAddr>() {
                        let punched = crate::http::tcp_punch(http_port, browser_ip).await;
                        crate::protocol::TcpPunchResultPayload {
                            success: punched,
                            http_addr,
                        }
                    } else {
                        crate::protocol::TcpPunchResultPayload { success: false, http_addr: None }
                    }
                } else {
                    crate::protocol::TcpPunchResultPayload { success: false, http_addr: None }
                };
                write_typed_message(&mut send, MessageType::TcpPunchResult, &resp).await?;
                send.finish()?;
            }
            MessageType::BlobRequest => {
                let payload: BlobRequestPayload = read_payload(&mut recv, 4096).await?;
                let cm = conn_mgr.lock().await;
                let data = cm.blob_store.get(&payload.cid)?;
                let response = match data {
                    Some(bytes) => {
                        // Check delivery budget before serving
                        if !cm.blob_store.consume_delivery_budget(bytes.len() as u64) {
                            debug!(
                                peer = hex::encode(remote_node_id),
                                cid = hex::encode(payload.cid),
                                blob_size = bytes.len(),
                                "Delivery budget exhausted, declining blob request"
                            );
                            BlobResponsePayload {
                                cid: payload.cid,
                                found: false,
                                data_b64: String::new(),
                                manifest: None,
                                cdn_registered: false,
                                cdn_redirect_peers: vec![],
                            }
                        } else {
                        use base64::Engine;

                        // Load manifest if available, wrap in CdnManifest
                        let storage = cm.storage.lock().await;
                        let manifest: Option<crate::types::CdnManifest> = storage
                            .get_cdn_manifest(&payload.cid)
                            .ok()
                            .flatten()
                            .and_then(|json| {
                                // Try as AuthorManifest first (author-side), then as CdnManifest (relay)
                                if let Ok(am) = serde_json::from_str::<crate::types::AuthorManifest>(&json) {
                                    let ds_count = storage.get_blob_downstream_count(&payload.cid).unwrap_or(0);
                                    Some(crate::types::CdnManifest {
                                        author_manifest: am,
                                        host: cm.our_node_id,
                                        host_addresses: vec![], // Filled by caller if needed
                                        source: cm.our_node_id,
                                        source_addresses: vec![],
                                        downstream_count: ds_count,
                                    })
                                } else {
                                    // Already a CdnManifest (from a relay/fetch)
                                    serde_json::from_str(&json).ok()
                                }
                            });

                        // Try to register requester as downstream
                        let (cdn_registered, cdn_redirect_peers) = if !payload.requester_addresses.is_empty() {
                            let ok = storage.add_blob_downstream(
                                &payload.cid,
                                &remote_node_id,
                                &payload.requester_addresses,
                            ).unwrap_or(false);
                            if ok {
                                (true, vec![])
                            } else {
                                // Full — provide downstream list as redirect candidates
                                let downstream = storage.get_blob_downstream(&payload.cid).unwrap_or_default();
                                let redirects: Vec<PeerWithAddress> = downstream.into_iter()
                                    .map(|(nid, addrs)| PeerWithAddress {
                                        n: hex::encode(nid),
                                        a: addrs,
                                    })
                                    .collect();
                                (false, redirects)
                            }
                        } else {
                            (false, vec![])
                        };
                        drop(storage);

                        BlobResponsePayload {
                            cid: payload.cid,
                            found: true,
                            data_b64: base64::engine::general_purpose::STANDARD.encode(&bytes),
                            manifest,
                            cdn_registered,
                            cdn_redirect_peers,
                        }
                        } // end delivery budget else
                    }
                    None => BlobResponsePayload {
                        cid: payload.cid,
                        found: false,
                        data_b64: String::new(),
                        manifest: None,
                        cdn_registered: false,
                        cdn_redirect_peers: vec![],
                    },
                };
                drop(cm);
                // 15MB limit for base64 overhead on 10MB blobs + manifest
                write_typed_message(&mut send, MessageType::BlobResponse, &response).await?;
                send.finish()?;
                debug!(peer = hex::encode(remote_node_id), found = response.found, cdn_reg = response.cdn_registered, "Handled blob request");
            }
            MessageType::ManifestRefreshRequest => {
                let payload: crate::protocol::ManifestRefreshRequestPayload = read_payload(&mut recv, 1024).await?;
                let cm = conn_mgr.lock().await;
                let storage = cm.storage.lock().await;
                let response = match storage.get_cdn_manifest(&payload.cid).ok().flatten() {
                    Some(json) => {
                        // Build CdnManifest from stored AuthorManifest
                        let manifest = if let Ok(am) = serde_json::from_str::<crate::types::AuthorManifest>(&json) {
                            if am.updated_at > payload.current_updated_at {
                                let ds_count = storage.get_blob_downstream_count(&payload.cid).unwrap_or(0);
                                Some(crate::types::CdnManifest {
                                    author_manifest: am,
                                    host: cm.our_node_id,
                                    host_addresses: vec![],
                                    source: cm.our_node_id,
                                    source_addresses: vec![],
                                    downstream_count: ds_count,
                                })
                            } else {
                                None
                            }
                        } else {
                            None
                        };
                        crate::protocol::ManifestRefreshResponsePayload {
                            cid: payload.cid,
                            updated: manifest.is_some(),
                            manifest,
                        }
                    }
                    None => crate::protocol::ManifestRefreshResponsePayload {
                        cid: payload.cid,
                        updated: false,
                        manifest: None,
                    },
                };
                drop(storage);
                drop(cm);
                write_typed_message(&mut send, MessageType::ManifestRefreshResponse, &response).await?;
                send.finish()?;
                debug!(peer = hex::encode(remote_node_id), updated = response.updated, "Handled manifest refresh request");
            }
            MessageType::GroupKeyRequest => {
                let payload: GroupKeyRequestPayload = read_payload(&mut recv, 4096).await?;
                let cm = conn_mgr.lock().await;
                let storage = cm.storage.lock().await;

                let response = match storage.get_group_key(&payload.group_id)? {
                    Some(record) if record.admin == cm.our_node_id => {
                        // We're the admin — check if requester is still a circle member
                        let members = storage.get_circle_members(&record.circle_name)?;
                        if members.contains(&remote_node_id) {
                            // Wrap group seed for requester
                            let seed = storage.get_group_seed(&payload.group_id, record.epoch)?;
                            let member_key = if let Some(seed) = seed {
                                crypto::wrap_group_key_for_member(
                                    &cm.secret_seed,
                                    &remote_node_id,
                                    &seed,
                                ).ok().map(|wrapped| crate::types::GroupMemberKey {
                                    member: remote_node_id,
                                    epoch: record.epoch,
                                    wrapped_group_key: wrapped,
                                })
                            } else {
                                None
                            };
                            GroupKeyResponsePayload {
                                group_id: payload.group_id,
                                epoch: record.epoch,
                                group_public_key: record.group_public_key,
                                admin: cm.our_node_id,
                                member_key,
                            }
                        } else {
                            // Requester not a member — no key
                            GroupKeyResponsePayload {
                                group_id: payload.group_id,
                                epoch: record.epoch,
                                group_public_key: record.group_public_key,
                                admin: cm.our_node_id,
                                member_key: None,
                            }
                        }
                    }
                    _ => {
                        // Not admin or no record
                        GroupKeyResponsePayload {
                            group_id: payload.group_id,
                            epoch: 0,
                            group_public_key: [0u8; 32],
                            admin: cm.our_node_id,
                            member_key: None,
                        }
                    }
                };
                drop(storage);
                drop(cm);
                write_typed_message(&mut send, MessageType::GroupKeyResponse, &response).await?;
                send.finish()?;
                debug!(peer = hex::encode(remote_node_id), group_id = hex::encode(payload.group_id), "Handled group key request");
            }
            MessageType::RelayIntroduce => {
                let payload: RelayIntroducePayload = read_payload(&mut recv, MAX_PAYLOAD).await?;
                debug!(
                    peer = hex::encode(remote_node_id),
                    target = hex::encode(payload.target),
                    requester = hex::encode(payload.requester),
                    ttl = payload.ttl,
                    "Received relay introduce"
                );
                let cm_arc = Arc::clone(conn_mgr);
                let mut cm = conn_mgr.lock().await;
                cm.handle_relay_introduce(payload, send, remote_node_id, cm_arc).await?;
            }
            MessageType::SessionRelay => {
                let cm = Arc::clone(conn_mgr);
                Self::handle_session_relay(cm, recv, send, remote_node_id).await?;
            }
            MessageType::MeshPrefer => {
                let mut cm = conn_mgr.lock().await;
                cm.handle_mesh_prefer(remote_node_id, send, recv).await?;
            }
            MessageType::AnchorReferralRequest => {
                let payload: AnchorReferralRequestPayload = read_payload(&mut recv, 4096).await?;
                let mut cm = conn_mgr.lock().await;
                cm.handle_anchor_referral_request(payload, send).await?;
            }
            MessageType::AnchorProbeRequest => {
                let payload: crate::protocol::AnchorProbeRequestPayload = read_payload(&mut recv, 4096).await?;
                let mut cm = conn_mgr.lock().await;
                cm.handle_anchor_probe_request(payload, send).await?;
            }
            MessageType::NatFilterProbe => {
                let payload: crate::protocol::NatFilterProbePayload = read_payload(&mut recv, 256).await?;
                // Only hold lock briefly to look up observed address, then release before 2s probe
                let observed_addr = {
                    let cm = conn_mgr.lock().await;
                    cm.connections.get(&payload.node_id)
                        .and_then(|pc| pc.remote_addr)
                        .or_else(|| cm.sessions.get(&payload.node_id).and_then(|s| s.remote_addr))
                };
                ConnectionManager::handle_nat_filter_probe_static(payload, observed_addr, send).await?;
            }
            MessageType::BlobHeaderRequest => {
                let payload: BlobHeaderRequestPayload = read_payload(&mut recv, MAX_PAYLOAD).await?;
                let (header_json, _updated_at) = {
                    let cm = conn_mgr.lock().await;
                    let storage = cm.storage.lock().await;
                    match storage.get_blob_header(&payload.post_id) {
                        Ok(Some((json, ts))) if ts > payload.current_updated_at => (Some(json), ts),
                        Ok(_) => (None, 0),
                        Err(_) => (None, 0),
                    }
                };
                let response = BlobHeaderResponsePayload {
                    post_id: payload.post_id,
                    updated: header_json.is_some(),
                    header_json,
                };
                write_typed_message(&mut send, MessageType::BlobHeaderResponse, &response).await?;
            }
            MessageType::ReplicationRequest => {
                // Limit to 3 concurrent replication handlers to prevent overload
                static REPLICATION_SEMAPHORE: std::sync::LazyLock<tokio::sync::Semaphore> =
                    std::sync::LazyLock::new(|| tokio::sync::Semaphore::new(3));
                let _permit = REPLICATION_SEMAPHORE.acquire().await
                    .map_err(|_| anyhow::anyhow!("replication semaphore closed"))?;

                let payload: ReplicationRequestPayload = read_payload(&mut recv, MAX_PAYLOAD).await?;
                let (accepted, rejected, needs_pull) = {
                    let cm = conn_mgr.lock().await;
                    let storage = cm.storage.lock().await;
                    let mut acc = Vec::new();
                    let mut rej = Vec::new();
                    let mut to_pull = Vec::new();
                    // Estimate ~1 MB per post with blobs for budget tracking
                    let est_bytes_per_post: u64 = 1024 * 1024;
                    let mut budget_used: u64 = 0;
                    let budget_cap: u64 = 20 * est_bytes_per_post; // cap per request

                    for pid in &payload.post_ids {
                        // Already have it — accept for free
                        if storage.get_post(pid).ok().flatten().is_some() {
                            acc.push(*pid);
                            continue;
                        }
                        // Check budget before accepting a post we need to pull
                        if budget_used + est_bytes_per_post > budget_cap {
                            rej.push(*pid);
                            continue;
                        }
                        budget_used += est_bytes_per_post;
                        acc.push(*pid);
                        to_pull.push(*pid);
                    }

                    // Register as downstream for all accepted posts
                    for pid in &acc {
                        let _ = storage.add_post_downstream(pid, &remote_node_id);
                    }

                    (acc, rej, to_pull)
                };
                let response = ReplicationResponsePayload { accepted: accepted.clone(), rejected };
                write_typed_message(&mut send, MessageType::ReplicationResponse, &response).await?;
                send.finish()?;
                let accepted_count = accepted.len();
                let needs_pull_count = needs_pull.len();
                debug!(
                    peer = hex::encode(remote_node_id),
                    accepted = accepted_count,
                    rejected = response.rejected.len(),
                    needs_pull = needs_pull_count,
                    "Handled replication request"
                );
                // Actively fetch posts we accepted but don't have from the requester
                if !needs_pull.is_empty() {
                    let cm_arc = conn_mgr.clone();
                    let sender = remote_node_id;
                    tokio::spawn(async move {
                        let conn = {
                            let cm = cm_arc.lock().await;
                            cm.connections_ref().get(&sender).map(|pc| pc.connection.clone())
                                .or_else(|| cm.sessions.get(&sender).map(|sc| sc.connection.clone()))
                        };
                        let Some(conn) = conn else { return };
                        let mut fetched = 0usize;
                        for post_id in &needs_pull {
                            // PostFetch without holding any lock
                            let result: anyhow::Result<Option<crate::protocol::SyncPost>> = async {
                                let (mut send, mut recv) = conn.open_bi().await?;
                                let req = crate::protocol::PostFetchRequestPayload { post_id: *post_id };
                                write_typed_message(&mut send, MessageType::PostFetchRequest, &req).await?;
                                send.finish()?;
                                let msg_type = read_message_type(&mut recv).await?;
                                if msg_type != MessageType::PostFetchResponse {
                                    return Ok(None);
                                }
                                let resp: crate::protocol::PostFetchResponsePayload = read_payload(&mut recv, MAX_PAYLOAD).await?;
                                Ok(resp.post)
                            }.await;

                            if let Ok(Some(sp)) = result {
                                if crate::content::verify_post_id(&sp.id, &sp.post) {
                                    let attachments = sp.post.attachments.clone();
                                    let post_author = sp.post.author;
                                    let cm = cm_arc.lock().await;
                                    let storage = cm.storage.lock().await;
                                    let _ = storage.store_post_with_visibility(&sp.id, &sp.post, &sp.visibility);
                                    let prio = storage.get_post_upstreams(&sp.id).map(|v| v.len() as u8).unwrap_or(0);
                                    let _ = storage.add_post_upstream(&sp.id, &sender, prio);
                                    let blob_store = cm.blob_store.clone();
                                    drop(storage);
                                    drop(cm);
                                    fetched += 1;

                                    // Fetch blobs for this post from the requester
                                    for att in &attachments {
                                        if blob_store.has(&att.cid) { continue; }
                                        let blob_result: anyhow::Result<()> = async {
                                            let (mut bs, mut br) = conn.open_bi().await?;
                                            let req = BlobRequestPayload {
                                                cid: att.cid,
                                                requester_addresses: vec![],
                                            };
                                            write_typed_message(&mut bs, MessageType::BlobRequest, &req).await?;
                                            bs.finish()?;
                                            let mt = read_message_type(&mut br).await?;
                                            if mt != MessageType::BlobResponse { return Ok(()); }
                                            let resp: BlobResponsePayload = read_payload(&mut br, MAX_PAYLOAD).await?;
                                            if resp.found {
                                                use base64::Engine;
                                                let data = base64::engine::general_purpose::STANDARD.decode(resp.data_b64.as_bytes())?;
                                                blob_store.store(&att.cid, &data)?;
                                                let cm = cm_arc.lock().await;
                                                let storage = cm.storage.lock().await;
                                                let _ = storage.record_blob(&att.cid, post_id, &post_author, data.len() as u64, &att.mime_type, att.size_bytes);
                                                let _ = storage.add_post_upstream(&att.cid, &sender, 0);
                                            }
                                            Ok(())
                                        }.await;
                                        if let Err(e) = blob_result {
                                            debug!(cid = hex::encode(att.cid), error = %e, "Replication blob fetch failed");
                                        }
                                    }
                                }
                            }
                        }
                        if fetched > 0 {
                            debug!(fetched, peer = hex::encode(sender), "Fetched replicated posts from requester");
                        }
                    });
                }
            }
            other => {
                warn!(msg_type = ?other, "Unexpected message type on bi-stream");
            }
        }

        Ok(())
    }

    /// Handle an incoming BlobHeaderDiff — store engagement ops and re-propagate to downstream + upstream.
    async fn handle_blob_header_diff(&self, payload: BlobHeaderDiffPayload, sender: NodeId) {
        use crate::types::BlobHeaderDiffOp;

        // Gather policy + audience data, then drop lock immediately
        let (policy, approved_audience, downstream, upstreams) = {
            let storage = self.storage.lock().await;
            let policy = storage.get_comment_policy(&payload.post_id)
                .ok()
                .flatten()
                .unwrap_or_default();
            let approved = storage.list_audience(
                crate::types::AudienceDirection::Inbound,
                Some(crate::types::AudienceStatus::Approved),
            ).unwrap_or_default();
            let downstream = storage.get_post_downstream(&payload.post_id).unwrap_or_default();
            let upstreams: Vec<NodeId> = storage.get_post_upstreams(&payload.post_id)
                .unwrap_or_default()
                .into_iter()
                .map(|(nid, _)| nid)
                .collect();
            (policy, approved, downstream, upstreams)
        };

        // Filter ops using gathered data (no lock held)
        let audience_set: std::collections::HashSet<NodeId> = approved_audience.iter().map(|a| a.node_id).collect();

        // Apply ops in a short lock acquisition
        {
            let storage = self.storage.lock().await;
            for op in &payload.ops {
                match op {
                    BlobHeaderDiffOp::AddReaction(reaction) => {
                        if policy.blocklist.contains(&reaction.reactor) {
                            continue;
                        }
                        if let crate::types::ReactPermission::None = policy.allow_reacts {
                            continue;
                        }
                        // Verify signature (skip if empty for backward compat with old nodes)
                        if !reaction.signature.is_empty() && !crate::crypto::verify_reaction_signature(
                            &reaction.reactor,
                            &payload.post_id,
                            &reaction.emoji,
                            reaction.timestamp_ms,
                            &reaction.signature,
                        ) {
                            continue; // Skip forged reactions
                        }
                        let _ = storage.store_reaction(reaction);
                    }
                    BlobHeaderDiffOp::RemoveReaction { reactor, emoji, post_id } => {
                        if *reactor == sender || sender == payload.author {
                            let _ = storage.remove_reaction(reactor, post_id, emoji);
                        }
                    }
                    BlobHeaderDiffOp::AddComment(comment) => {
                        if policy.blocklist.contains(&comment.author) {
                            continue;
                        }
                        match policy.allow_comments {
                            crate::types::CommentPermission::None => continue,
                            crate::types::CommentPermission::AudienceOnly => {
                                if !audience_set.contains(&comment.author) {
                                    continue;
                                }
                            }
                            crate::types::CommentPermission::Public => {}
                        }
                        if !crate::crypto::verify_comment_signature(
                            &comment.author,
                            &payload.post_id,
                            &comment.content,
                            comment.timestamp_ms,
                            &comment.signature,
                        ) {
                            continue; // Skip forged comments
                        }
                        let _ = storage.store_comment(comment);
                    }
                    BlobHeaderDiffOp::EditComment { author, post_id, timestamp_ms, new_content } => {
                        // Trust-based: only the comment author can edit
                        if *author == sender || sender == payload.author {
                            let _ = storage.edit_comment(author, post_id, *timestamp_ms, new_content);
                        }
                    }
                    BlobHeaderDiffOp::DeleteComment { author, post_id, timestamp_ms } => {
                        // Trust-based: comment author or post author can delete
                        if *author == sender || sender == payload.author {
                            let _ = storage.delete_comment(author, post_id, *timestamp_ms);
                        }
                    }
                    BlobHeaderDiffOp::SetPolicy(new_policy) => {
                        if sender == payload.author {
                            let _ = storage.set_comment_policy(&payload.post_id, new_policy);
                        }
                    }
                    BlobHeaderDiffOp::ThreadSplit { new_post_id } => {
                        let _ = storage.store_thread_meta(&crate::types::ThreadMeta {
                            post_id: *new_post_id,
                            parent_post_id: payload.post_id,
                        });
                    }
                    BlobHeaderDiffOp::WriteReceiptSlot { post_id, slot_index, data } => {
                        // Store encrypted bytes directly — no decryption needed on relay nodes
                        if let Ok(Some((json, _ts))) = storage.get_blob_header(post_id) {
                            if let Ok(mut header) = serde_json::from_str::<crate::types::BlobHeader>(&json) {
                                let idx = *slot_index as usize;
                                while header.receipt_slots.len() <= idx {
                                    header.receipt_slots.push(vec![0u8; 64]); // padding
                                }
                                header.receipt_slots[idx] = data.clone();
                                let now_ms = std::time::SystemTime::now()
                                    .duration_since(std::time::UNIX_EPOCH)
                                    .map(|d| d.as_millis() as u64)
                                    .unwrap_or(0);
                                header.updated_at = now_ms;
                                if let Ok(new_json) = serde_json::to_string(&header) {
                                    let _ = storage.store_blob_header(post_id, &header.author, &new_json, now_ms);
                                }
                            }
                        }
                    }
                    BlobHeaderDiffOp::WriteCommentSlot { post_id, slot_index, data } => {
                        if let Ok(Some((json, _ts))) = storage.get_blob_header(post_id) {
                            if let Ok(mut header) = serde_json::from_str::<crate::types::BlobHeader>(&json) {
                                let idx = *slot_index as usize;
                                while header.comment_slots.len() <= idx {
                                    header.comment_slots.push(vec![0u8; 256]);
                                }
                                header.comment_slots[idx] = data.clone();
                                let now_ms = std::time::SystemTime::now()
                                    .duration_since(std::time::UNIX_EPOCH)
                                    .map(|d| d.as_millis() as u64)
                                    .unwrap_or(0);
                                header.updated_at = now_ms;
                                if let Ok(new_json) = serde_json::to_string(&header) {
                                    let _ = storage.store_blob_header(post_id, &header.author, &new_json, now_ms);
                                }
                            }
                        }
                    }
                    BlobHeaderDiffOp::AddCommentSlots { post_id, count: _, slots } => {
                        if let Ok(Some((json, _ts))) = storage.get_blob_header(post_id) {
                            if let Ok(mut header) = serde_json::from_str::<crate::types::BlobHeader>(&json) {
                                for slot in slots {
                                    header.comment_slots.push(slot.clone());
                                }
                                let now_ms = std::time::SystemTime::now()
                                    .duration_since(std::time::UNIX_EPOCH)
                                    .map(|d| d.as_millis() as u64)
                                    .unwrap_or(0);
                                header.updated_at = now_ms;
                                if let Ok(new_json) = serde_json::to_string(&header) {
                                    let _ = storage.store_blob_header(post_id, &header.author, &new_json, now_ms);
                                }
                            }
                        }
                    }
                    BlobHeaderDiffOp::Unknown => {} // future ops — silently skip
                }
            }

            // Rebuild blob header JSON from current DB state so pull-based sync gets fresh data.
            // Use _with_tombstones so tombstones propagate through the pull path.
            let reactions = storage.get_reactions_with_tombstones(&payload.post_id).unwrap_or_default();
            let comments = storage.get_comments_with_tombstones(&payload.post_id).unwrap_or_default();
            let policy = storage.get_comment_policy(&payload.post_id).ok().flatten().unwrap_or_default();
            let (existing_header_json, _) = storage.get_blob_header(&payload.post_id)
                .ok()
                .flatten()
                .unwrap_or((String::new(), 0));
            let mut header: crate::types::BlobHeader = serde_json::from_str(&existing_header_json).unwrap_or_else(|_| {
                crate::types::BlobHeader {
                    post_id: payload.post_id,
                    author: payload.author,
                    reactions: vec![],
                    comments: vec![],
                    policy: crate::types::CommentPolicy::default(),
                    updated_at: 0,
                    thread_splits: vec![],
                    receipt_slots: vec![],
                    comment_slots: vec![],
                }
            });
            header.reactions = reactions;
            header.comments = comments;
            header.policy = policy;
            header.updated_at = payload.timestamp_ms;
            // Look up actual post author (don't trust payload.author)
            let actual_author = storage.get_post(&payload.post_id)
                .ok().flatten()
                .map(|p| p.author)
                .unwrap_or(payload.author); // fallback if post not stored yet
            header.author = actual_author;
            if let Ok(json) = serde_json::to_string(&header) {
                let _ = storage.store_blob_header(&payload.post_id, &actual_author, &json, payload.timestamp_ms);
            }
            // Phase 4: Update last_engagement_ms when engagement arrives via diff
            let _ = storage.update_post_last_engagement(&payload.post_id, payload.timestamp_ms);
        }

        // Collect all targets (downstream + all upstreams), then send in a single batched task
        let mut targets: Vec<iroh::endpoint::Connection> = Vec::new();
        for peer_id in downstream {
            if peer_id == sender { continue; }
            if let Some(conn) = self.connections.get(&peer_id).map(|mc| mc.connection.clone())
                .or_else(|| self.sessions.get(&peer_id).map(|sc| sc.connection.clone()))
            {
                targets.push(conn);
            }
        }
        // Phase 6: Try all upstreams, not just one
        for up in &upstreams {
            if *up != sender {
                if let Some(conn) = self.connections.get(up).map(|mc| mc.connection.clone())
                    .or_else(|| self.sessions.get(up).map(|sc| sc.connection.clone()))
                {
                    targets.push(conn);
                }
            }
        }
        if !targets.is_empty() {
            let payload_clone = payload.clone();
            tokio::spawn(async move {
                // Send to up to 10 concurrently, then batch the rest
                use tokio::task::JoinSet;
                let mut set = JoinSet::new();
                for conn in targets {
                    let p = payload_clone.clone();
                    set.spawn(async move {
                        if let Ok(mut send) = conn.open_uni().await {
                            let _ = write_typed_message(&mut send, MessageType::BlobHeaderDiff, &p).await;
                            let _ = send.finish();
                        }
                    });
                    // Cap concurrency at 10
                    if set.len() >= 10 {
                        let _ = set.join_next().await;
                    }
                }
                while set.join_next().await.is_some() {}
            });
        }
    }

    /// Get the endpoint reference.
    pub fn endpoint(&self) -> &iroh::Endpoint {
        &self.endpoint
    }
}

// ============================================================================
// ConnHandle + ConnectionActor: actor-based interface to ConnectionManager
// ============================================================================

use tokio::sync::{mpsc, oneshot};

/// Response types for actor commands that return data.
pub enum ConnResponse {
    Bool(bool),
    Usize(usize),
    NodeId(NodeId),
    OptConnection(Option<iroh::endpoint::Connection>),
    Peers(Vec<NodeId>),
    ConnectionInfo(Vec<(NodeId, PeerSlotKind, u64)>),
    SessionInfo(Vec<(NodeId, SessionReachMethod, u64)>),
    NatProfile(crate::types::NatProfile),
    NatType(crate::types::NatType),
    NatMapping(crate::types::NatMapping),
    NatFiltering(crate::types::NatFiltering),
    OptString(Option<String>),
    OptEndpointAddr(Option<iroh::EndpointAddr>),
    Endpoint(iroh::Endpoint),
    SecretSeed([u8; 32]),
    Storage(Arc<Mutex<Storage>>),
    BlobStore(Arc<BlobStore>),
    ActiveRelayPipes(Arc<AtomicU64>),
    Referrals(Vec<AnchorReferral>),
    OptSocketAddr(Option<SocketAddr>),
    IsAnchorAtomicBool(Arc<AtomicBool>),
    ActivityLogRef(Arc<std::sync::Mutex<ActivityLog>>),
    ScoreVec(Vec<(NodeId, f64)>),
    OptPeerWithAddress(Option<PeerWithAddress>),
    ConnectionMap(Vec<(NodeId, iroh::endpoint::Connection, PeerSlotKind, Arc<AtomicU64>)>),
    DiffData(DiffSnapshot),
    Unit,
}

/// Snapshot of data needed for routing diff computation.
pub struct DiffSnapshot {
    pub our_node_id: NodeId,
    pub connections: Vec<(NodeId, iroh::endpoint::Connection)>,
    pub diff_seq: u64,
    pub n1_added: Vec<NodeId>,
    pub n1_removed: Vec<NodeId>,
    pub n2_added: Vec<NodeId>,
    pub n2_removed: Vec<NodeId>,
}

/// Commands sent to the ConnectionActor via the ConnHandle channel.
pub enum ConnCommand {
    // --- Reads ---
    IsConnected {
        peer: NodeId,
        reply: oneshot::Sender<bool>,
    },
    ConnectionCount {
        reply: oneshot::Sender<usize>,
    },
    ConnectedPeers {
        reply: oneshot::Sender<Vec<NodeId>>,
    },
    ConnectionInfo {
        reply: oneshot::Sender<Vec<(NodeId, PeerSlotKind, u64)>>,
    },
    GetConnection {
        peer: NodeId,
        reply: oneshot::Sender<Option<iroh::endpoint::Connection>>,
    },
    GetAnyConnection {
        peer: NodeId,
        reply: oneshot::Sender<Option<iroh::endpoint::Connection>>,
    },
    /// Get all mesh connections (node_id, connection, slot_kind, last_activity)
    GetConnectionMap {
        reply: oneshot::Sender<Vec<(NodeId, iroh::endpoint::Connection, PeerSlotKind, Arc<AtomicU64>)>>,
    },
    OurNodeId {
        reply: oneshot::Sender<NodeId>,
    },
    OurNatProfile {
        reply: oneshot::Sender<crate::types::NatProfile>,
    },
    NatType {
        reply: oneshot::Sender<crate::types::NatType>,
    },
    NatMapping {
        reply: oneshot::Sender<crate::types::NatMapping>,
    },
    NatFiltering {
        reply: oneshot::Sender<crate::types::NatFiltering>,
    },
    SessionInfo {
        reply: oneshot::Sender<Vec<(NodeId, SessionReachMethod, u64)>>,
    },
    HasSession {
        peer: NodeId,
        reply: oneshot::Sender<bool>,
    },
    IsConnectedOrSession {
        peer: NodeId,
        reply: oneshot::Sender<bool>,
    },
    SessionPeerIds {
        reply: oneshot::Sender<Vec<NodeId>>,
    },
    AvailableLocalSlots {
        reply: oneshot::Sender<usize>,
    },
    IsLikelyUnreachable {
        peer: NodeId,
        reply: oneshot::Sender<bool>,
    },
    GetEndpoint {
        reply: oneshot::Sender<iroh::Endpoint>,
    },
    GetSecretSeed {
        reply: oneshot::Sender<[u8; 32]>,
    },
    GetStorage {
        reply: oneshot::Sender<Arc<Mutex<Storage>>>,
    },
    GetBlobStore {
        reply: oneshot::Sender<Arc<BlobStore>>,
    },
    ActiveRelayPipes {
        reply: oneshot::Sender<Arc<AtomicU64>>,
    },
    CanAcceptRelayPipe {
        reply: oneshot::Sender<bool>,
    },
    BuildAnchorAdvertisedAddr {
        reply: oneshot::Sender<Option<String>>,
    },
    ResolvePeerAddrLocal {
        peer: NodeId,
        reply: oneshot::Sender<Option<iroh::EndpointAddr>>,
    },
    PickRandomRedirectPeer {
        exclude: NodeId,
        reply: oneshot::Sender<Option<PeerWithAddress>>,
    },
    IsAnchorCandidate {
        reply: oneshot::Sender<bool>,
    },
    ProbeDue {
        reply: oneshot::Sender<bool>,
    },
    IsAnchorRef {
        reply: oneshot::Sender<Arc<AtomicBool>>,
    },
    GetActivityLog {
        reply: oneshot::Sender<Arc<std::sync::Mutex<ActivityLog>>>,
    },
    ScoreN2Candidates {
        reply: oneshot::Sender<Vec<(NodeId, f64)>>,
    },
    GetPeerObservedAddr {
        peer: NodeId,
        reply: oneshot::Sender<Option<SocketAddr>>,
    },
    GetPeerLastActivity {
        peer: NodeId,
        reply: oneshot::Sender<Option<Arc<AtomicU64>>>,
    },

    // --- Mutations (with reply) ---
    AcceptConnection {
        conn: iroh::endpoint::Connection,
        remote_node_id: NodeId,
        remote_addr: Option<SocketAddr>,
        reply: oneshot::Sender<bool>,
    },
    RegisterConnection {
        peer_id: NodeId,
        conn: iroh::endpoint::Connection,
        addrs: Vec<SocketAddr>,
        slot_kind: PeerSlotKind,
        reply: oneshot::Sender<()>,
    },
    DisconnectPeer {
        peer: NodeId,
        reply: oneshot::Sender<()>,
    },
    AddSession {
        node_id: NodeId,
        conn: iroh::endpoint::Connection,
        reach_method: SessionReachMethod,
        remote_addr: Option<SocketAddr>,
        reply: oneshot::Sender<()>,
    },
    RemoveSession {
        peer: NodeId,
        reply: oneshot::Sender<()>,
    },
    TouchSession {
        peer: NodeId,
        reply: oneshot::Sender<()>,
    },
    ReapIdleSessions {
        idle_timeout_ms: u64,
        reply: oneshot::Sender<()>,
    },
    MarkReachable {
        peer: NodeId,
    },
    MarkUnreachable {
        peer: NodeId,
    },
    SetNatFiltering {
        filtering: crate::types::NatFiltering,
    },
    AddStickyN1 {
        peer: NodeId,
        duration_ms: u64,
    },
    SetGrowthTx {
        tx: tokio::sync::mpsc::Sender<()>,
    },
    SetRecoveryTx {
        tx: tokio::sync::mpsc::Sender<()>,
    },

    // --- Fire-and-forget ---
    NotifyGrowth,
    NotifyRecovery,
    LogActivity {
        level: ActivityLevel,
        cat: ActivityCategory,
        msg: String,
        peer: Option<NodeId>,
    },

    // --- Dedup checks ---
    CheckSeenWorm {
        worm_id: WormId,
        reply: oneshot::Sender<bool>,
    },
    CheckSeenIntro {
        intro_id: IntroId,
        reply: oneshot::Sender<bool>,
    },
    CheckSeenProbe {
        probe_id: [u8; 16],
        reply: oneshot::Sender<bool>,
    },
    RecordProbeSuccess,
    RecordProbeFailure,

    // --- Referral management ---
    HandleAnchorRegister {
        payload: AnchorRegisterPayload,
        reply: oneshot::Sender<()>,
    },
    PickReferrals {
        exclude: NodeId,
        count: usize,
        reply: oneshot::Sender<Vec<AnchorReferral>>,
    },
    MarkReferralDisconnected {
        node_id: NodeId,
    },
    MarkReferralReconnected {
        node_id: NodeId,
    },

    // --- Diff/routing ---
    ProcessRoutingDiff {
        from_peer: NodeId,
        payload: NodeListUpdatePayload,
        reply: oneshot::Sender<anyhow::Result<usize>>,
    },
    /// Get snapshot of connections + diff data for external broadcast
    GetDiffData {
        reply: oneshot::Sender<DiffSnapshot>,
    },

    // --- Relay/address lookups (state reads, no network I/O) ---
    FindRelaysFor {
        target: NodeId,
        reply: oneshot::Sender<Vec<(NodeId, u8)>>,
    },
    GetUpnpExternalAddr {
        reply: oneshot::Sender<Option<SocketAddr>>,
    },
    TouchSessionIfExists {
        peer: NodeId,
    },

    // --- Complex operations (actor processes, may block command queue during I/O) ---
    RebalanceSlots {
        reply: oneshot::Sender<anyhow::Result<Vec<NodeId>>>,
    },
    WormLookup {
        target: NodeId,
        reply: oneshot::Sender<anyhow::Result<Option<WormResult>>>,
    },
    ContentSearch {
        target: NodeId,
        post_id: Option<PostId>,
        blob_id: Option<[u8; 32]>,
        reply: oneshot::Sender<anyhow::Result<Option<WormResult>>>,
    },
    PostFetch {
        holder: NodeId,
        post_id: PostId,
        reply: oneshot::Sender<anyhow::Result<Option<crate::protocol::SyncPost>>>,
    },
    ResolveAddress {
        target: NodeId,
        reply: oneshot::Sender<anyhow::Result<Option<String>>>,
    },
    PullFromPeer {
        peer: NodeId,
        reply: oneshot::Sender<anyhow::Result<PullSyncStats>>,
    },
    FetchEngagement {
        peer: NodeId,
        reply: oneshot::Sender<anyhow::Result<usize>>,
    },
    InitiateAnchorProbe {
        reply: oneshot::Sender<anyhow::Result<bool>>,
    },
    TcpPunch {
        holder: NodeId,
        browser_ip: String,
        post_id: PostId,
        reply: oneshot::Sender<anyhow::Result<Option<String>>>,
    },
}

/// Cheap-to-clone handle for sending commands to the ConnectionActor.
/// Replaces `Arc<Mutex<ConnectionManager>>` at call sites.
#[derive(Clone)]
pub struct ConnHandle {
    tx: mpsc::Sender<ConnCommand>,
    /// Whether this node's HTTP server is running (set once at startup)
    http_capable: Arc<AtomicBool>,
    /// External HTTP address if known (set once at startup)
    http_addr: Arc<std::sync::Mutex<Option<String>>>,
    /// CDN device role (set once at startup by Network)
    device_role_val: Arc<std::sync::Mutex<Option<crate::types::DeviceRole>>>,
}

impl ConnHandle {
    /// Create a ConnHandle from a command channel sender.
    pub fn new(tx: mpsc::Sender<ConnCommand>) -> Self {
        Self {
            tx,
            http_capable: Arc::new(AtomicBool::new(false)),
            http_addr: Arc::new(std::sync::Mutex::new(None)),
            device_role_val: Arc::new(std::sync::Mutex::new(None)),
        }
    }

    /// Set HTTP capability info (called once after HTTP server starts).
    pub fn set_http_info(&self, capable: bool, addr: Option<String>) {
        self.http_capable.store(capable, Ordering::Relaxed);
        *self.http_addr.lock().unwrap() = addr;
    }

    /// Set CDN device role (called once at Network startup).
    pub fn set_device_role(&self, role: crate::types::DeviceRole) {
        *self.device_role_val.lock().unwrap() = Some(role);
    }

    /// Get CDN device role, if set.
    pub fn device_role(&self) -> Option<crate::types::DeviceRole> {
        *self.device_role_val.lock().unwrap()
    }

    /// Whether this node is HTTP-capable.
    pub fn is_http_capable(&self) -> bool {
        self.http_capable.load(Ordering::Relaxed)
    }

    /// External HTTP address if known.
    pub fn http_addr(&self) -> Option<String> {
        self.http_addr.lock().unwrap().clone()
    }

    // === Read operations ===

    pub async fn is_connected(&self, peer: &NodeId) -> bool {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::IsConnected { peer: *peer, reply: tx }).await;
        rx.await.unwrap_or(false)
    }

    pub async fn connection_count(&self) -> usize {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::ConnectionCount { reply: tx }).await;
        rx.await.unwrap_or(0)
    }

    pub async fn connected_peers(&self) -> Vec<NodeId> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::ConnectedPeers { reply: tx }).await;
        rx.await.unwrap_or_default()
    }

    pub async fn connection_info(&self) -> Vec<(NodeId, PeerSlotKind, u64)> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::ConnectionInfo { reply: tx }).await;
        rx.await.unwrap_or_default()
    }

    pub async fn get_connection(&self, peer: &NodeId) -> Option<iroh::endpoint::Connection> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::GetConnection { peer: *peer, reply: tx }).await;
        rx.await.ok().flatten()
    }

    pub async fn get_any_connection(&self, peer: &NodeId) -> Option<iroh::endpoint::Connection> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::GetAnyConnection { peer: *peer, reply: tx }).await;
        rx.await.ok().flatten()
    }

    /// Get all mesh connections as (node_id, connection, slot_kind, last_activity).
    pub async fn get_connection_map(&self) -> Vec<(NodeId, iroh::endpoint::Connection, PeerSlotKind, Arc<AtomicU64>)> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::GetConnectionMap { reply: tx }).await;
        rx.await.unwrap_or_default()
    }

    pub async fn our_node_id(&self) -> NodeId {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::OurNodeId { reply: tx }).await;
        rx.await.unwrap_or([0u8; 32])
    }

    pub async fn our_nat_profile(&self) -> crate::types::NatProfile {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::OurNatProfile { reply: tx }).await;
        rx.await.unwrap_or_else(|_| crate::types::NatProfile::new(
            crate::types::NatMapping::EndpointIndependent,
            crate::types::NatFiltering::Unknown,
        ))
    }

    pub async fn nat_type(&self) -> crate::types::NatType {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::NatType { reply: tx }).await;
        rx.await.unwrap_or(crate::types::NatType::Unknown)
    }

    pub async fn nat_filtering(&self) -> crate::types::NatFiltering {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::NatFiltering { reply: tx }).await;
        rx.await.unwrap_or(crate::types::NatFiltering::Unknown)
    }

    pub async fn session_info(&self) -> Vec<(NodeId, SessionReachMethod, u64)> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::SessionInfo { reply: tx }).await;
        rx.await.unwrap_or_default()
    }

    pub async fn has_session(&self, peer: &NodeId) -> bool {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::HasSession { peer: *peer, reply: tx }).await;
        rx.await.unwrap_or(false)
    }

    pub async fn is_connected_or_session(&self, peer: &NodeId) -> bool {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::IsConnectedOrSession { peer: *peer, reply: tx }).await;
        rx.await.unwrap_or(false)
    }

    pub async fn session_peer_ids(&self) -> Vec<NodeId> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::SessionPeerIds { reply: tx }).await;
        rx.await.unwrap_or_default()
    }

    pub async fn available_local_slots(&self) -> usize {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::AvailableLocalSlots { reply: tx }).await;
        rx.await.unwrap_or(0)
    }

    pub async fn is_likely_unreachable(&self, peer: &NodeId) -> bool {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::IsLikelyUnreachable { peer: *peer, reply: tx }).await;
        rx.await.unwrap_or(false)
    }

    pub async fn endpoint(&self) -> iroh::Endpoint {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::GetEndpoint { reply: tx }).await;
        rx.await.expect("actor dropped")
    }

    pub async fn secret_seed(&self) -> [u8; 32] {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::GetSecretSeed { reply: tx }).await;
        rx.await.unwrap_or([0u8; 32])
    }

    pub async fn storage(&self) -> Arc<Mutex<Storage>> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::GetStorage { reply: tx }).await;
        rx.await.expect("actor dropped")
    }

    pub async fn blob_store(&self) -> Arc<BlobStore> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::GetBlobStore { reply: tx }).await;
        rx.await.expect("actor dropped")
    }

    pub async fn active_relay_pipes(&self) -> Arc<AtomicU64> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::ActiveRelayPipes { reply: tx }).await;
        rx.await.expect("actor dropped")
    }

    pub async fn can_accept_relay_pipe(&self) -> bool {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::CanAcceptRelayPipe { reply: tx }).await;
        rx.await.unwrap_or(false)
    }

    pub async fn build_anchor_advertised_addr(&self) -> Option<String> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::BuildAnchorAdvertisedAddr { reply: tx }).await;
        rx.await.ok().flatten()
    }

    pub async fn resolve_peer_addr_local(&self, peer: &NodeId) -> Option<iroh::EndpointAddr> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::ResolvePeerAddrLocal { peer: *peer, reply: tx }).await;
        rx.await.ok().flatten()
    }

    pub async fn pick_random_redirect_peer(&self, exclude: &NodeId) -> Option<PeerWithAddress> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::PickRandomRedirectPeer { exclude: *exclude, reply: tx }).await;
        rx.await.ok().flatten()
    }

    pub async fn is_anchor_candidate(&self) -> bool {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::IsAnchorCandidate { reply: tx }).await;
        rx.await.unwrap_or(false)
    }

    pub async fn probe_due(&self) -> bool {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::ProbeDue { reply: tx }).await;
        rx.await.unwrap_or(false)
    }

    pub async fn is_anchor_ref(&self) -> Arc<AtomicBool> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::IsAnchorRef { reply: tx }).await;
        rx.await.expect("actor dropped")
    }

    pub async fn activity_log(&self) -> Arc<std::sync::Mutex<ActivityLog>> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::GetActivityLog { reply: tx }).await;
        rx.await.expect("actor dropped")
    }

    pub async fn score_n2_candidates(&self) -> Vec<(NodeId, f64)> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::ScoreN2Candidates { reply: tx }).await;
        rx.await.unwrap_or_default()
    }

    pub async fn get_peer_observed_addr(&self, peer: &NodeId) -> Option<SocketAddr> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::GetPeerObservedAddr { peer: *peer, reply: tx }).await;
        rx.await.ok().flatten()
    }

    pub async fn get_peer_last_activity(&self, peer: &NodeId) -> Option<Arc<AtomicU64>> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::GetPeerLastActivity { peer: *peer, reply: tx }).await;
        rx.await.ok().flatten()
    }

    // === Mutation operations ===

    pub async fn accept_connection(
        &self,
        conn: iroh::endpoint::Connection,
        remote_node_id: NodeId,
        remote_addr: Option<SocketAddr>,
    ) -> bool {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::AcceptConnection {
            conn,
            remote_node_id,
            remote_addr,
            reply: tx,
        }).await;
        rx.await.unwrap_or(false)
    }

    pub async fn register_connection(
        &self,
        peer_id: NodeId,
        conn: iroh::endpoint::Connection,
        addrs: Vec<SocketAddr>,
        slot_kind: PeerSlotKind,
    ) {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::RegisterConnection {
            peer_id,
            conn,
            addrs,
            slot_kind,
            reply: tx,
        }).await;
        let _ = rx.await;
    }

    pub async fn disconnect_peer(&self, peer: &NodeId) {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::DisconnectPeer { peer: *peer, reply: tx }).await;
        let _ = rx.await;
    }

    pub async fn add_session(
        &self,
        node_id: NodeId,
        conn: iroh::endpoint::Connection,
        reach_method: SessionReachMethod,
        remote_addr: Option<SocketAddr>,
    ) {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::AddSession {
            node_id,
            conn,
            reach_method,
            remote_addr,
            reply: tx,
        }).await;
        let _ = rx.await;
    }

    pub async fn remove_session(&self, peer: &NodeId) {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::RemoveSession { peer: *peer, reply: tx }).await;
        let _ = rx.await;
    }

    pub async fn touch_session(&self, peer: &NodeId) {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::TouchSession { peer: *peer, reply: tx }).await;
        let _ = rx.await;
    }

    pub async fn reap_idle_sessions(&self, idle_timeout_ms: u64) {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::ReapIdleSessions { idle_timeout_ms, reply: tx }).await;
        let _ = rx.await;
    }

    pub fn mark_reachable(&self, peer: &NodeId) {
        let _ = self.tx.try_send(ConnCommand::MarkReachable { peer: *peer });
    }

    pub fn mark_unreachable(&self, peer: &NodeId) {
        let _ = self.tx.try_send(ConnCommand::MarkUnreachable { peer: *peer });
    }

    /// Add a node to the sticky N1 set (reported in N1 share until expiry).
    pub fn add_sticky_n1(&self, peer: &NodeId, duration_ms: u64) {
        let _ = self.tx.try_send(ConnCommand::AddStickyN1 { peer: *peer, duration_ms });
    }

    pub fn set_nat_filtering(&self, filtering: crate::types::NatFiltering) {
        let _ = self.tx.try_send(ConnCommand::SetNatFiltering { filtering });
    }

    pub async fn set_growth_tx(&self, tx: tokio::sync::mpsc::Sender<()>) {
        let _ = self.tx.send(ConnCommand::SetGrowthTx { tx }).await;
    }

    pub async fn set_recovery_tx(&self, tx: tokio::sync::mpsc::Sender<()>) {
        let _ = self.tx.send(ConnCommand::SetRecoveryTx { tx }).await;
    }

    // === Fire-and-forget ===

    pub fn notify_growth(&self) {
        let _ = self.tx.try_send(ConnCommand::NotifyGrowth);
    }

    pub fn notify_recovery(&self) {
        let _ = self.tx.try_send(ConnCommand::NotifyRecovery);
    }

    pub fn log_activity(&self, level: ActivityLevel, cat: ActivityCategory, msg: String, peer: Option<NodeId>) {
        let _ = self.tx.try_send(ConnCommand::LogActivity { level, cat, msg, peer });
    }

    // === Dedup checks ===

    pub async fn check_seen_worm(&self, worm_id: &WormId) -> bool {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::CheckSeenWorm { worm_id: *worm_id, reply: tx }).await;
        rx.await.unwrap_or(false)
    }

    pub async fn check_seen_intro(&self, intro_id: &IntroId) -> bool {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::CheckSeenIntro { intro_id: *intro_id, reply: tx }).await;
        rx.await.unwrap_or(false)
    }

    pub async fn check_seen_probe(&self, probe_id: &[u8; 16]) -> bool {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::CheckSeenProbe { probe_id: *probe_id, reply: tx }).await;
        rx.await.unwrap_or(false)
    }

    pub fn record_probe_success(&self) {
        let _ = self.tx.try_send(ConnCommand::RecordProbeSuccess);
    }

    pub fn record_probe_failure(&self) {
        let _ = self.tx.try_send(ConnCommand::RecordProbeFailure);
    }

    // === Referral management ===

    pub async fn handle_anchor_register(&self, payload: AnchorRegisterPayload) {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::HandleAnchorRegister { payload, reply: tx }).await;
        let _ = rx.await;
    }

    pub async fn pick_referrals(&self, exclude: &NodeId, count: usize) -> Vec<AnchorReferral> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::PickReferrals { exclude: *exclude, count, reply: tx }).await;
        rx.await.unwrap_or_default()
    }

    pub fn mark_referral_disconnected(&self, node_id: &NodeId) {
        let _ = self.tx.try_send(ConnCommand::MarkReferralDisconnected { node_id: *node_id });
    }

    pub fn mark_referral_reconnected(&self, node_id: &NodeId) {
        let _ = self.tx.try_send(ConnCommand::MarkReferralReconnected { node_id: *node_id });
    }

    // === Diff/routing ===

    pub async fn process_routing_diff(
        &self,
        from_peer: &NodeId,
        payload: NodeListUpdatePayload,
    ) -> anyhow::Result<usize> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::ProcessRoutingDiff {
            from_peer: *from_peer,
            payload,
            reply: tx,
        }).await;
        rx.await.map_err(|_| anyhow::anyhow!("actor dropped"))?
    }

    pub async fn find_relays_for(&self, target: &NodeId) -> Vec<(NodeId, u8)> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::FindRelaysFor { target: *target, reply: tx }).await;
        rx.await.unwrap_or_default()
    }

    pub async fn upnp_external_addr(&self) -> Option<SocketAddr> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::GetUpnpExternalAddr { reply: tx }).await;
        rx.await.ok().flatten()
    }

    /// Touch session last_active (fire-and-forget, no-op if not a session peer).
    pub fn touch_session_if_exists(&self, peer: &NodeId) {
        let _ = self.tx.try_send(ConnCommand::TouchSessionIfExists { peer: *peer });
    }

    pub async fn get_diff_data(&self) -> DiffSnapshot {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::GetDiffData { reply: tx }).await;
        rx.await.unwrap_or_else(|_| DiffSnapshot {
            our_node_id: [0u8; 32],
            connections: vec![],
            diff_seq: 0,
            n1_added: vec![],
            n1_removed: vec![],
            n2_added: vec![],
            n2_removed: vec![],
        })
    }

    // === Complex operations ===

    pub async fn rebalance_slots(&self) -> anyhow::Result<Vec<NodeId>> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::RebalanceSlots { reply: tx }).await;
        rx.await.map_err(|_| anyhow::anyhow!("actor dropped"))?
    }

    pub async fn worm_lookup(&self, target: &NodeId) -> anyhow::Result<Option<WormResult>> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::WormLookup { target: *target, reply: tx }).await;
        rx.await.map_err(|_| anyhow::anyhow!("actor dropped"))?
    }

    pub async fn content_search(
        &self,
        target: &NodeId,
        post_id: Option<PostId>,
        blob_id: Option<[u8; 32]>,
    ) -> anyhow::Result<Option<WormResult>> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::ContentSearch {
            target: *target,
            post_id,
            blob_id,
            reply: tx,
        }).await;
        rx.await.map_err(|_| anyhow::anyhow!("actor dropped"))?
    }

    pub async fn post_fetch(
        &self,
        holder: &NodeId,
        post_id: &PostId,
    ) -> anyhow::Result<Option<crate::protocol::SyncPost>> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::PostFetch {
            holder: *holder,
            post_id: *post_id,
            reply: tx,
        }).await;
        rx.await.map_err(|_| anyhow::anyhow!("actor dropped"))?
    }

    pub async fn tcp_punch(
        &self,
        holder: &NodeId,
        browser_ip: String,
        post_id: &PostId,
    ) -> anyhow::Result<Option<String>> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::TcpPunch {
            holder: *holder,
            browser_ip,
            post_id: *post_id,
            reply: tx,
        }).await;
        rx.await.map_err(|_| anyhow::anyhow!("actor dropped"))?
    }

    pub async fn resolve_address(&self, target: &NodeId) -> anyhow::Result<Option<String>> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::ResolveAddress { target: *target, reply: tx }).await;
        rx.await.map_err(|_| anyhow::anyhow!("actor dropped"))?
    }

    pub async fn pull_from_peer(&self, peer: &NodeId) -> anyhow::Result<PullSyncStats> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::PullFromPeer { peer: *peer, reply: tx }).await;
        rx.await.map_err(|_| anyhow::anyhow!("actor dropped"))?
    }

    pub async fn fetch_engagement_from_peer(&self, peer: &NodeId) -> anyhow::Result<usize> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::FetchEngagement { peer: *peer, reply: tx }).await;
        rx.await.map_err(|_| anyhow::anyhow!("actor dropped"))?
    }

    pub async fn initiate_anchor_probe(&self) -> anyhow::Result<bool> {
        let (tx, rx) = oneshot::channel();
        let _ = self.tx.send(ConnCommand::InitiateAnchorProbe { reply: tx }).await;
        rx.await.map_err(|_| anyhow::anyhow!("actor dropped"))?
    }
}

/// The actor task that processes commands by locking the shared ConnectionManager.
pub struct ConnectionActor {
    cm: Arc<Mutex<ConnectionManager>>,
    rx: mpsc::Receiver<ConnCommand>,
}

impl ConnectionActor {
    /// Spawn the actor wrapping a shared Arc<Mutex<CM>>, returning a ConnHandle.
    /// During migration, both the actor and legacy lock-callers share state.
    pub fn spawn_with_arc(cm: Arc<Mutex<ConnectionManager>>) -> ConnHandle {
        let (tx, rx) = mpsc::channel(256);
        let actor = ConnectionActor { cm, rx };
        tokio::spawn(actor.run());
        ConnHandle::new(tx)
    }

    /// Spawn the actor owning the ConnectionManager directly (Phase 5+).
    pub fn spawn(cm: ConnectionManager) -> ConnHandle {
        Self::spawn_with_arc(Arc::new(Mutex::new(cm)))
    }

    async fn run(mut self) {
        while let Some(cmd) = self.rx.recv().await {
            self.handle(cmd).await;
        }
        debug!("ConnectionActor shutting down");
    }

    async fn handle(&mut self, cmd: ConnCommand) {
        match cmd {
            // --- Reads ---
            ConnCommand::IsConnected { peer, reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.is_connected(&peer));
            }
            ConnCommand::ConnectionCount { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.connection_count());
            }
            ConnCommand::ConnectedPeers { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.connected_peers());
            }
            ConnCommand::ConnectionInfo { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.connection_info());
            }
            ConnCommand::GetConnection { peer, reply } => {
                let cm = self.cm.lock().await;
                let conn = cm.connections_ref().get(&peer).map(|pc| pc.connection.clone());
                let _ = reply.send(conn);
            }
            ConnCommand::GetAnyConnection { peer, reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.get_any_connection(&peer));
            }
            ConnCommand::GetConnectionMap { reply } => {
                let cm = self.cm.lock().await;
                let map: Vec<_> = cm.connections_ref().iter().map(|(nid, pc)| {
                    (*nid, pc.connection.clone(), pc.slot_kind, Arc::clone(&pc.last_activity))
                }).collect();
                let _ = reply.send(map);
            }
            ConnCommand::OurNodeId { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(*cm.our_node_id());
            }
            ConnCommand::OurNatProfile { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.our_nat_profile());
            }
            ConnCommand::NatType { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.nat_type());
            }
            ConnCommand::NatMapping { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.nat_mapping());
            }
            ConnCommand::NatFiltering { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.nat_filtering);
            }
            ConnCommand::SessionInfo { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.session_info());
            }
            ConnCommand::HasSession { peer, reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.has_session(&peer));
            }
            ConnCommand::IsConnectedOrSession { peer, reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.is_connected_or_session(&peer));
            }
            ConnCommand::SessionPeerIds { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.session_peer_ids());
            }
            ConnCommand::AvailableLocalSlots { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.available_local_slots());
            }
            ConnCommand::IsLikelyUnreachable { peer, reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.is_likely_unreachable(&peer));
            }
            ConnCommand::GetEndpoint { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.endpoint().clone());
            }
            ConnCommand::GetSecretSeed { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.secret_seed);
            }
            ConnCommand::GetStorage { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.storage_ref());
            }
            ConnCommand::GetBlobStore { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(Arc::clone(&cm.blob_store));
            }
            ConnCommand::ActiveRelayPipes { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(Arc::clone(cm.active_relay_pipes()));
            }
            ConnCommand::CanAcceptRelayPipe { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.can_accept_relay_pipe());
            }
            ConnCommand::BuildAnchorAdvertisedAddr { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.build_anchor_advertised_addr());
            }
            ConnCommand::ResolvePeerAddrLocal { peer, reply } => {
                let cm = self.cm.lock().await;
                let r = cm.resolve_peer_addr_local(&peer).await;
                let _ = reply.send(r);
            }
            ConnCommand::PickRandomRedirectPeer { exclude, reply } => {
                let cm = self.cm.lock().await;
                let r = cm.pick_random_redirect_peer(&exclude).await;
                let _ = reply.send(r);
            }
            ConnCommand::IsAnchorCandidate { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.is_anchor_candidate());
            }
            ConnCommand::ProbeDue { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.probe_due());
            }
            ConnCommand::IsAnchorRef { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(Arc::clone(&cm.is_anchor));
            }
            ConnCommand::GetActivityLog { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(Arc::clone(&cm.activity_log));
            }
            ConnCommand::ScoreN2Candidates { reply } => {
                let cm = self.cm.lock().await;
                let r = cm.score_n2_candidates().await;
                let _ = reply.send(r);
            }
            ConnCommand::GetPeerObservedAddr { peer, reply } => {
                let cm = self.cm.lock().await;
                let addr = cm.connections_ref().get(&peer)
                    .and_then(|pc| pc.remote_addr);
                let _ = reply.send(addr);
            }

            // --- Mutations ---
            ConnCommand::AcceptConnection { conn, remote_node_id, remote_addr, reply } => {
                let mut cm = self.cm.lock().await;
                let r = cm.accept_connection(conn, remote_node_id, remote_addr);
                let _ = reply.send(r);
            }
            ConnCommand::RegisterConnection { peer_id, conn, addrs, slot_kind, reply } => {
                let mut cm = self.cm.lock().await;
                cm.register_connection(peer_id, conn, &addrs, slot_kind).await;
                let _ = reply.send(());
            }
            ConnCommand::DisconnectPeer { peer, reply } => {
                let mut cm = self.cm.lock().await;
                cm.disconnect_peer(&peer).await;
                let _ = reply.send(());
            }
            ConnCommand::AddSession { node_id, conn, reach_method, remote_addr, reply } => {
                let mut cm = self.cm.lock().await;
                cm.add_session(node_id, conn, reach_method, remote_addr);
                let _ = reply.send(());
            }
            ConnCommand::RemoveSession { peer, reply } => {
                let mut cm = self.cm.lock().await;
                cm.remove_session(&peer);
                let _ = reply.send(());
            }
            ConnCommand::TouchSession { peer, reply } => {
                let mut cm = self.cm.lock().await;
                cm.touch_session(&peer);
                let _ = reply.send(());
            }
            ConnCommand::ReapIdleSessions { idle_timeout_ms, reply } => {
                let mut cm = self.cm.lock().await;
                cm.reap_idle_sessions(idle_timeout_ms).await;
                let _ = reply.send(());
            }
            ConnCommand::MarkReachable { peer } => {
                let mut cm = self.cm.lock().await;
                cm.mark_reachable(&peer);
            }
            ConnCommand::MarkUnreachable { peer } => {
                let mut cm = self.cm.lock().await;
                cm.mark_unreachable(&peer);
            }
            ConnCommand::SetNatFiltering { filtering } => {
                let mut cm = self.cm.lock().await;
                cm.nat_filtering = filtering;
            }
            ConnCommand::AddStickyN1 { peer, duration_ms } => {
                let mut cm = self.cm.lock().await;
                let expiry = now_ms() + duration_ms;
                cm.sticky_n1.insert(peer, expiry);
            }
            ConnCommand::SetGrowthTx { tx } => {
                let mut cm = self.cm.lock().await;
                cm.set_growth_tx(tx);
            }
            ConnCommand::SetRecoveryTx { tx } => {
                let mut cm = self.cm.lock().await;
                cm.set_recovery_tx(tx);
            }

            // --- Fire-and-forget ---
            ConnCommand::NotifyGrowth => {
                let cm = self.cm.lock().await;
                cm.notify_growth();
            }
            ConnCommand::NotifyRecovery => {
                let cm = self.cm.lock().await;
                cm.notify_recovery();
            }
            ConnCommand::LogActivity { level, cat, msg, peer } => {
                let cm = self.cm.lock().await;
                cm.log_activity(level, cat, msg, peer);
            }

            // --- Dedup checks ---
            ConnCommand::CheckSeenWorm { worm_id, reply } => {
                let now = now_ms();
                let mut cm = self.cm.lock().await;
                cm.seen_worms.retain(|_, ts| now - *ts < WORM_DEDUP_EXPIRY_MS);
                let seen = cm.seen_worms.contains_key(&worm_id);
                if !seen {
                    cm.seen_worms.insert(worm_id, now);
                }
                let _ = reply.send(seen);
            }
            ConnCommand::CheckSeenIntro { intro_id, reply } => {
                let now = now_ms();
                let mut cm = self.cm.lock().await;
                cm.seen_intros.retain(|_, ts| now - *ts < RELAY_INTRO_DEDUP_EXPIRY_MS);
                let seen = cm.seen_intros.contains_key(&intro_id);
                if !seen {
                    cm.seen_intros.insert(intro_id, now);
                }
                let _ = reply.send(seen);
            }
            ConnCommand::CheckSeenProbe { probe_id, reply } => {
                let now = now_ms();
                let mut cm = self.cm.lock().await;
                cm.seen_probes.retain(|_, ts| now - *ts < 60_000);
                let seen = cm.seen_probes.contains_key(&probe_id);
                if !seen {
                    cm.seen_probes.insert(probe_id, now);
                }
                let _ = reply.send(seen);
            }
            ConnCommand::RecordProbeSuccess => {
                let mut cm = self.cm.lock().await;
                cm.last_probe_success_ms = now_ms();
                cm.probe_failure_streak = 0;
            }
            ConnCommand::RecordProbeFailure => {
                let mut cm = self.cm.lock().await;
                cm.probe_failure_streak = cm.probe_failure_streak.saturating_add(1);
            }

            // --- Referral management ---
            ConnCommand::HandleAnchorRegister { payload, reply } => {
                let mut cm = self.cm.lock().await;
                cm.handle_anchor_register(payload).await;
                let _ = reply.send(());
            }
            ConnCommand::PickReferrals { exclude, count, reply } => {
                let mut cm = self.cm.lock().await;
                let r = cm.pick_referrals(&exclude, count);
                let _ = reply.send(r);
            }
            ConnCommand::MarkReferralDisconnected { node_id } => {
                let mut cm = self.cm.lock().await;
                cm.mark_referral_disconnected(&node_id);
            }
            ConnCommand::MarkReferralReconnected { node_id } => {
                let mut cm = self.cm.lock().await;
                cm.mark_referral_reconnected(&node_id);
            }

            // --- Diff/routing ---
            ConnCommand::ProcessRoutingDiff { from_peer, payload, reply } => {
                let cm = self.cm.lock().await;
                let r = cm.process_routing_diff(&from_peer, payload).await;
                let _ = reply.send(r);
            }
            ConnCommand::GetDiffData { reply } => {
                let mut cm = self.cm.lock().await;
                // Prune expired sticky N1 entries first
                let now = now_ms();
                cm.sticky_n1.retain(|_, expiry| *expiry > now);
                let sticky_peers: Vec<NodeId> = cm.sticky_n1.keys().copied().collect();
                // Compute diff snapshot
                let storage = cm.storage.lock().await;
                let current_n1: HashSet<NodeId> = {
                    let mut set = HashSet::new();
                    for nid in cm.connections_ref().keys() {
                        set.insert(*nid);
                    }
                    if let Ok(routes) = storage.list_social_routes() {
                        for route in &routes {
                            if route.status == crate::types::SocialStatus::Online {
                                set.insert(route.node_id);
                            }
                        }
                    }
                    for nid in &sticky_peers {
                        set.insert(*nid);
                    }
                    set
                };
                let current_n2: HashSet<NodeId> = storage.build_n2_share()
                    .unwrap_or_default()
                    .into_iter()
                    .collect();
                drop(storage);

                let n1_added: Vec<NodeId> = current_n1.difference(&cm.last_n1_set).copied().collect();
                let n1_removed: Vec<NodeId> = cm.last_n1_set.difference(&current_n1).copied().collect();
                let n2_added: Vec<NodeId> = current_n2.difference(&cm.last_n2_set).copied().collect();
                let n2_removed: Vec<NodeId> = cm.last_n2_set.difference(&current_n2).copied().collect();

                cm.last_n1_set = current_n1;
                cm.last_n2_set = current_n2;

                let seq = cm.diff_seq.fetch_add(1, Ordering::Relaxed);

                let conns: Vec<(NodeId, iroh::endpoint::Connection)> = cm.connections_ref()
                    .iter()
                    .map(|(nid, pc)| (*nid, pc.connection.clone()))
                    .collect();

                let _ = reply.send(DiffSnapshot {
                    our_node_id: *cm.our_node_id(),
                    connections: conns,
                    diff_seq: seq,
                    n1_added,
                    n1_removed,
                    n2_added,
                    n2_removed,
                });
            }

            // --- Relay/address lookups ---
            ConnCommand::FindRelaysFor { target, reply } => {
                let cm = self.cm.lock().await;
                let r = cm.find_relays_for(&target).await;
                let _ = reply.send(r);
            }
            ConnCommand::GetUpnpExternalAddr { reply } => {
                let cm = self.cm.lock().await;
                let _ = reply.send(cm.upnp_external_addr);
            }
            ConnCommand::TouchSessionIfExists { peer } => {
                let mut cm = self.cm.lock().await;
                if let Some(session) = cm.sessions.get_mut(&peer) {
                    session.last_active_at = now_ms();
                }
            }

            // --- Complex operations ---
            ConnCommand::RebalanceSlots { reply } => {
                let mut cm = self.cm.lock().await;
                let r = cm.rebalance_slots().await;
                let _ = reply.send(r);
            }
            ConnCommand::WormLookup { target, reply } => {
                let cm = self.cm.lock().await;
                let r = cm.initiate_worm_lookup(&target).await;
                let _ = reply.send(r);
            }
            ConnCommand::ContentSearch { target, post_id, blob_id, reply } => {
                let cm = self.cm.lock().await;
                let r = cm.initiate_content_search(&target, post_id, blob_id).await;
                let _ = reply.send(r);
            }
            ConnCommand::PostFetch { holder, post_id, reply } => {
                let cm = self.cm.lock().await;
                let r = cm.send_post_fetch(&holder, &post_id).await;
                let _ = reply.send(r);
            }
            ConnCommand::TcpPunch { holder, browser_ip, post_id, reply } => {
                let cm = self.cm.lock().await;
                let r = cm.send_tcp_punch(&holder, browser_ip, &post_id).await;
                let _ = reply.send(r);
            }
            ConnCommand::ResolveAddress { target, reply } => {
                let cm = self.cm.lock().await;
                let r = cm.resolve_address(&target).await;
                let _ = reply.send(r);
            }
            ConnCommand::PullFromPeer { peer, reply } => {
                let cm = self.cm.lock().await;
                let r = cm.pull_from_peer(&peer).await;
                let _ = reply.send(r);
            }
            ConnCommand::FetchEngagement { peer, reply } => {
                let cm = self.cm.lock().await;
                let r = cm.fetch_engagement_from_peer(&peer).await;
                let _ = reply.send(r);
            }
            ConnCommand::InitiateAnchorProbe { reply } => {
                let mut cm = self.cm.lock().await;
                let r = cm.initiate_anchor_probe().await;
                let _ = reply.send(r);
            }
            ConnCommand::GetPeerLastActivity { peer, reply } => {
                let cm = self.cm.lock().await;
                let activity = cm.connections_ref().get(&peer)
                    .map(|pc| Arc::clone(&pc.last_activity));
                let _ = reply.send(activity);
            }
        }
    }
}

/// Standalone initial exchange (connector side) — does NOT require conn_mgr lock.
/// Opens a bi-stream, sends our N1/N2/profile/deletes/post_ids, reads theirs.
pub async fn initial_exchange_connect(
    storage: &Arc<Mutex<Storage>>,
    our_node_id: &NodeId,
    conn: &iroh::endpoint::Connection,
    remote_node_id: NodeId,
    anchor_addr: Option<String>,
    our_nat_type: crate::types::NatType,
    our_http_capable: bool,
    our_http_addr: Option<String>,
    our_device_role: Option<crate::types::DeviceRole>,
    our_cache_pressure: Option<u8>,
) -> anyhow::Result<ExchangeResult> {
    let our_payload = {
        let storage = storage.lock().await;
        let n1 = storage.build_n1_share()?;
        let n2 = storage.build_n2_share()?;
        let profile = storage.get_profile(our_node_id)?;
        let deletes = storage.list_delete_records()?;
        let post_ids = storage.list_post_ids()?;
        let peer_addresses = storage.build_peer_addresses_for(our_node_id)?;
        InitialExchangePayload {
            n1_node_ids: n1,
            n2_node_ids: n2,
            profile,
            deletes,
            post_ids,
            peer_addresses,
            anchor_addr,
            your_observed_addr: None, // connector doesn't know remote's observed addr
            nat_type: Some(our_nat_type.to_string()),
            nat_mapping: Some(crate::types::NatProfile::from_nat_type(our_nat_type).mapping.to_string()),
            nat_filtering: Some(crate::types::NatProfile::from_nat_type(our_nat_type).filtering.to_string()),
            http_capable: our_http_capable,
            http_addr: our_http_addr,
            device_role: our_device_role.map(|r| r.as_str().to_string()),
            cache_pressure: our_cache_pressure,
        }
    };

    let (mut send, mut recv) = conn.open_bi().await?;
    write_typed_message(&mut send, MessageType::InitialExchange, &our_payload).await?;
    send.finish()?;

    // 10s timeout: if the remote doesn't respond, bail rather than hanging forever
    let exchange_fut = async {
        let msg_type = read_message_type(&mut recv).await?;
        if msg_type == MessageType::RefuseRedirect {
            let payload: RefuseRedirectPayload = read_payload(&mut recv, MAX_PAYLOAD).await?;
            return Ok(ExchangeResult::Refused { redirect: payload.redirect });
        }
        if msg_type != MessageType::InitialExchange {
            anyhow::bail!("expected InitialExchange, got {:?}", msg_type);
        }
        let their_payload: InitialExchangePayload = read_payload(&mut recv, MAX_PAYLOAD).await?;
        process_exchange_payload(storage, our_node_id, &remote_node_id, &their_payload).await?;
        Ok(ExchangeResult::Accepted)
    };

    match tokio::time::timeout(std::time::Duration::from_secs(10), exchange_fut).await {
        Ok(result) => result,
        Err(_) => {
            warn!(peer = hex::encode(remote_node_id), "Initial exchange timed out (10s)");
            anyhow::bail!("initial exchange timed out");
        }
    }
}

/// Standalone initial exchange (acceptor side) — does NOT require conn_mgr lock.
/// Message type byte already consumed by caller.
pub async fn initial_exchange_accept(
    storage: &Arc<Mutex<Storage>>,
    our_node_id: &NodeId,
    mut send: iroh::endpoint::SendStream,
    mut recv: iroh::endpoint::RecvStream,
    remote_node_id: NodeId,
    anchor_addr: Option<String>,
    remote_addr: Option<SocketAddr>,
    our_nat_type: crate::types::NatType,
    our_http_capable: bool,
    our_http_addr: Option<String>,
    our_device_role: Option<crate::types::DeviceRole>,
    our_cache_pressure: Option<u8>,
) -> anyhow::Result<()> {
    let their_payload: InitialExchangePayload = read_payload(&mut recv, MAX_PAYLOAD).await?;

    let our_payload = {
        let storage = storage.lock().await;
        let n1 = storage.build_n1_share()?;
        let n2 = storage.build_n2_share()?;
        let profile = storage.get_profile(our_node_id)?;
        let deletes = storage.list_delete_records()?;
        let post_ids = storage.list_post_ids()?;
        let peer_addresses = storage.build_peer_addresses_for(our_node_id)?;
        InitialExchangePayload {
            n1_node_ids: n1,
            n2_node_ids: n2,
            profile,
            deletes,
            post_ids,
            peer_addresses,
            anchor_addr,
            your_observed_addr: remote_addr.map(|a| a.to_string()),
            nat_type: Some(our_nat_type.to_string()),
            nat_mapping: Some(crate::types::NatProfile::from_nat_type(our_nat_type).mapping.to_string()),
            nat_filtering: Some(crate::types::NatProfile::from_nat_type(our_nat_type).filtering.to_string()),
            http_capable: our_http_capable,
            http_addr: our_http_addr,
            device_role: our_device_role.map(|r| r.as_str().to_string()),
            cache_pressure: our_cache_pressure,
        }
    };

    write_typed_message(&mut send, MessageType::InitialExchange, &our_payload).await?;
    send.finish()?;

    process_exchange_payload(storage, our_node_id, &remote_node_id, &their_payload).await?;
    Ok(())
}

/// Process the peer's initial exchange payload (shared between connect and accept sides).
async fn process_exchange_payload(
    storage: &Arc<Mutex<Storage>>,
    our_node_id: &NodeId,
    remote_node_id: &NodeId,
    payload: &InitialExchangePayload,
) -> anyhow::Result<()> {
    let storage = storage.lock().await;

    // Filter out our own ID from their N1 before storing as our N2
    let filtered_n1: Vec<NodeId> = payload.n1_node_ids.iter()
        .filter(|nid| *nid != our_node_id)
        .copied()
        .collect();
    storage.set_peer_n1(remote_node_id, &filtered_n1)?;
    debug!(peer = hex::encode(remote_node_id), raw = payload.n1_node_ids.len(), stored = filtered_n1.len(), "Stored peer N1 as our N2 (filtered self)");

    let filtered_n2: Vec<NodeId> = payload.n2_node_ids.iter()
        .filter(|nid| *nid != our_node_id)
        .copied()
        .collect();
    storage.set_peer_n2(remote_node_id, &filtered_n2)?;
    debug!(peer = hex::encode(remote_node_id), raw = payload.n2_node_ids.len(), stored = filtered_n2.len(), "Stored peer N2 as our N3 (filtered self)");

    if let Some(ref profile) = payload.profile {
        let _ = storage.store_profile(profile);
    }

    for dr in &payload.deletes {
        if crypto::verify_delete_signature(&dr.author, &dr.post_id, &dr.signature) {
            let _ = storage.store_delete(dr);
            let _ = storage.apply_delete(dr);
        }
    }

    for pa in &payload.peer_addresses {
        if let Ok(nid) = crate::parse_node_id_hex(&pa.n) {
            let addrs: Vec<std::net::SocketAddr> = pa.a.iter().filter_map(|a| a.parse().ok()).collect();
            if !addrs.is_empty() {
                let _ = storage.upsert_peer(&nid, &addrs, None);
            }
        }
    }

    let our_post_ids: HashSet<[u8; 32]> = storage.list_post_ids()?.into_iter().collect();
    for pid in &payload.post_ids {
        if our_post_ids.contains(pid) {
            let _ = storage.record_replica(pid, remote_node_id);
        }
    }

    // Process anchor's advertised address
    if let Some(ref anchor_addr_str) = payload.anchor_addr {
        if let Ok(sock) = anchor_addr_str.parse::<SocketAddr>() {
            let _ = storage.upsert_known_anchor(remote_node_id, &[sock]);
            let _ = storage.upsert_peer(remote_node_id, &[sock], None);
            info!(peer = hex::encode(remote_node_id), addr = %sock, "Stored anchor's advertised address");
        }
    }

    // Log observed address (STUN-like feedback)
    if let Some(ref observed) = payload.your_observed_addr {
        info!(observed_addr = %observed, reporter = hex::encode(remote_node_id), "Peer reports our address as");
    }

    // Store peer's NAT type
    if let Some(ref nat_str) = payload.nat_type {
        let nat = crate::types::NatType::from_str_label(nat_str);
        let _ = storage.set_peer_nat_type(remote_node_id, nat);
        debug!(peer = hex::encode(remote_node_id), nat_type = %nat, "Stored peer NAT type");
    }

    // Store peer's NAT profile (mapping + filtering) if provided
    if payload.nat_mapping.is_some() || payload.nat_filtering.is_some() {
        let mapping = payload.nat_mapping.as_deref()
            .map(crate::types::NatMapping::from_str_label)
            .unwrap_or(crate::types::NatMapping::Unknown);
        let filtering = payload.nat_filtering.as_deref()
            .map(crate::types::NatFiltering::from_str_label)
            .unwrap_or(crate::types::NatFiltering::Unknown);
        let profile = crate::types::NatProfile::new(mapping, filtering);
        let _ = storage.set_peer_nat_profile(remote_node_id, &profile);
        debug!(peer = hex::encode(remote_node_id), mapping = %mapping, filtering = %filtering, "Stored peer NAT profile");
    }

    // Store peer's HTTP capability
    if payload.http_capable {
        let _ = storage.set_peer_http_info(remote_node_id, true, payload.http_addr.as_deref());
        debug!(peer = hex::encode(remote_node_id), http_addr = ?payload.http_addr, "Stored peer HTTP capability");
    }

    // Store peer's CDN device role and cache pressure
    if payload.device_role.is_some() || payload.cache_pressure.is_some() {
        let _ = storage.set_peer_device_role(
            remote_node_id,
            payload.device_role.as_deref(),
            payload.cache_pressure,
        );
        debug!(
            peer = hex::encode(remote_node_id),
            role = ?payload.device_role,
            pressure = ?payload.cache_pressure,
            "Stored peer CDN role"
        );
    }

    Ok(())
}

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