//! Publish and discover mesh-llm meshes via Nostr relays.
//!
//! A running mesh publishes a replaceable event (kind 31990, d-tag "mesh-llm")
//! containing its invite token, served models, VRAM, node count, etc.
//! Other nodes can discover available meshes and auto-join.

use anyhow::Result;
use nostr_sdk::prelude::*;
use serde::{Deserialize, Serialize};
use std::time::Duration;

/// NIP-89 "Application Handler" kind — used for service advertisements.
pub const MESH_SERVICE_KIND: u16 = 31990;

/// Default public relays.
pub const DEFAULT_RELAYS: &[&str] = &[
    "wss://relay.damus.io",
    "wss://nos.lol",
    "wss://relay.nostr.band",
    "wss://nostr.land",
    "wss://nostr.wine",
];

pub struct PublishLoopConfig {
    pub relays: Vec<String>,
    pub name: Option<String>,
    pub region: Option<String>,
    pub max_clients: Option<usize>,
    pub interval_secs: u64,
    pub status_tx: Option<tokio::sync::watch::Sender<Option<PublishStateUpdate>>>,
}

/// What we publish about a mesh.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MeshListing {
    /// Base64 invite token (others use this to --join)
    pub invite_token: String,
    /// Models currently loaded and serving inference
    pub serving: Vec<String>,
    /// Models the mesh wants but nobody is serving yet (need more GPUs)
    #[serde(default)]
    pub wanted: Vec<String>,
    /// Models on disk across the mesh (could be loaded if a GPU becomes free)
    #[serde(default)]
    pub on_disk: Vec<String>,
    /// Total VRAM across all GPU nodes (bytes)
    pub total_vram_bytes: u64,
    /// Number of GPU nodes in the mesh
    pub node_count: usize,
    /// Number of connected clients (API-only nodes)
    #[serde(default)]
    pub client_count: usize,
    /// Maximum clients this mesh accepts (0 = unlimited)
    #[serde(default)]
    pub max_clients: usize,
    /// Optional human-readable name for the mesh
    #[serde(skip_serializing_if = "Option::is_none")]
    pub name: Option<String>,
    /// Optional geographic region
    #[serde(skip_serializing_if = "Option::is_none")]
    pub region: Option<String>,
    /// Stable mesh identity — all nodes in the same mesh share this ID.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub mesh_id: Option<String>,
}

/// Discovered mesh from Nostr.
#[derive(Debug, Clone, serde::Serialize)]
pub struct DiscoveredMesh {
    pub listing: MeshListing,
    pub publisher_npub: String,
    pub published_at: u64,
    pub expires_at: Option<u64>,
}

impl std::fmt::Display for DiscoveredMesh {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let vram_gb = self.listing.total_vram_bytes as f64 / 1e9;
        let models = if self.listing.serving.is_empty() {
            "(no models loaded)".to_string()
        } else {
            self.listing.serving.join(", ")
        };
        write!(
            f,
            "{}  {} node(s), {:.0}GB capacity  serving: {}",
            self.listing.name.as_deref().unwrap_or("(unnamed)"),
            self.listing.node_count,
            vram_gb,
            models,
        )?;
        if let Some(ref region) = self.listing.region {
            write!(f, "  region: {}", region)?;
        }
        if !self.listing.wanted.is_empty() {
            write!(f, "  wanted: {}", self.listing.wanted.join(", "))?;
        }
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// Keys — stored in ~/.mesh-llm/nostr.nsec
// ---------------------------------------------------------------------------

fn nostr_key_path() -> Result<std::path::PathBuf> {
    let home =
        dirs::home_dir().ok_or_else(|| anyhow::anyhow!("Cannot determine home directory"))?;
    Ok(home.join(".mesh-llm").join("nostr.nsec"))
}

/// Load or generate a Nostr keypair for publishing.
pub fn load_or_create_keys() -> Result<Keys> {
    load_or_create_keys_at(&nostr_key_path()?)
}

fn load_or_create_keys_at(path: &std::path::Path) -> Result<Keys> {
    if let Some(parent) = path.parent() {
        ensure_private_nostr_dir(parent)?;
    }

    if path.exists() {
        ensure_private_nostr_key_file(path)?;
        let nsec = std::fs::read_to_string(path)?;
        let sk = SecretKey::from_bech32(nsec.trim())?;
        Ok(Keys::new(sk))
    } else {
        let keys = Keys::generate();
        let nsec = keys.secret_key().to_bech32()?;
        crate::crypto::write_keystore_bytes_atomically(path, nsec.as_bytes())?;
        tracing::info!("Generated new Nostr key, saved to {}", path.display());
        Ok(keys)
    }
}

#[cfg(unix)]
fn ensure_private_nostr_dir(dir: &std::path::Path) -> Result<()> {
    use std::os::unix::fs::PermissionsExt;

    std::fs::create_dir_all(dir)?;
    let metadata = std::fs::metadata(dir)?;
    let mut perms = metadata.permissions();
    if perms.mode() & 0o077 != 0 {
        perms.set_mode(0o700);
        std::fs::set_permissions(dir, perms)?;
    }
    Ok(())
}

#[cfg(not(unix))]
fn ensure_private_nostr_dir(dir: &std::path::Path) -> Result<()> {
    std::fs::create_dir_all(dir)?;
    Ok(())
}

#[cfg(unix)]
fn ensure_private_nostr_key_file(path: &std::path::Path) -> Result<()> {
    use std::os::unix::fs::PermissionsExt;

    let metadata = std::fs::symlink_metadata(path)?;
    if !metadata.file_type().is_file() {
        anyhow::bail!("Nostr key path is not a regular file");
    }
    let mut perms = metadata.permissions();
    if perms.mode() & 0o077 != 0 {
        perms.set_mode(0o600);
        std::fs::set_permissions(path, perms)?;
    }
    Ok(())
}

#[cfg(not(unix))]
fn ensure_private_nostr_key_file(_path: &std::path::Path) -> Result<()> {
    Ok(())
}

/// Delete the Nostr key and node identity key.  After rotation the
/// node gets a fresh identity on next start.
pub fn rotate_keys() -> Result<()> {
    let home =
        dirs::home_dir().ok_or_else(|| anyhow::anyhow!("Cannot determine home directory"))?;
    let mesh_dir = home.join(".mesh-llm");

    let nostr_path = nostr_key_path()?;
    if nostr_path.exists() {
        std::fs::remove_file(&nostr_path)?;
        eprintln!("🔑 Deleted {}", nostr_path.display());
    } else {
        eprintln!("No Nostr key to rotate (none exists yet).");
    }

    let node_key_path = mesh_dir.join("key");
    if node_key_path.exists() {
        std::fs::remove_file(&node_key_path)?;
        eprintln!("🔑 Deleted {}", node_key_path.display());
    } else {
        eprintln!("No node key to rotate (none exists yet).");
    }

    eprintln!();
    eprintln!("✅ Keys rotated. New identities will be generated on next start.");
    Ok(())
}

// ---------------------------------------------------------------------------
// Publisher — background task that keeps the listing fresh
// ---------------------------------------------------------------------------

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum PublishStateUpdate {
    Public,
    PublishFailed,
}

fn report_publish_state(
    status_tx: &Option<tokio::sync::watch::Sender<Option<PublishStateUpdate>>>,
    last_reported: &mut Option<PublishStateUpdate>,
    next: PublishStateUpdate,
) {
    if *last_reported == Some(next) {
        return;
    }
    if let Some(tx) = status_tx {
        let _ = tx.send(Some(next));
    }
    *last_reported = Some(next);
}

pub struct Publisher {
    client: Client,
    keys: Keys,
}

impl Publisher {
    pub async fn new(keys: Keys, relays: &[String]) -> Result<Self> {
        let _ = rustls::crypto::ring::default_provider().install_default();
        let client = Client::new(keys.clone());
        for relay in relays {
            client.add_relay(relay).await?;
        }
        client.connect().await;
        Ok(Self { client, keys })
    }

    pub fn npub(&self) -> String {
        self.keys.public_key().to_bech32().unwrap_or_default()
    }

    /// Publish (or replace) the mesh listing. Uses a replaceable event
    /// (kind 31990 + d-tag) so each publisher has exactly one listing.
    pub async fn publish(&self, listing: &MeshListing, ttl_secs: u64) -> Result<()> {
        let expiration = Timestamp::now().as_secs() + ttl_secs;
        let content = serde_json::to_string(listing)?;

        let tags = vec![
            Tag::custom(TagKind::Custom("d".into()), vec!["mesh-llm".to_string()]),
            Tag::custom(TagKind::Custom("k".into()), vec!["mesh-llm".to_string()]),
            Tag::custom(
                TagKind::Custom("expiration".into()),
                vec![expiration.to_string()],
            ),
        ];

        let builder = EventBuilder::new(Kind::Custom(MESH_SERVICE_KIND), content).tags(tags);
        self.client.send_event_builder(builder).await?;
        Ok(())
    }

    /// Delete our listing (e.g. on shutdown).
    pub async fn unpublish(&self) -> Result<()> {
        // Fetch our own events
        let filter = Filter::new()
            .kind(Kind::Custom(MESH_SERVICE_KIND))
            .author(self.keys.public_key())
            .limit(10);
        let events = self
            .client
            .fetch_events(filter, Duration::from_secs(5))
            .await?;
        for event in events.iter() {
            let request = EventDeletionRequest::new().id(event.id);
            let _ = self
                .client
                .send_event_builder(EventBuilder::delete(request))
                .await;
        }
        Ok(())
    }
}

/// Background publish loop. Republishes every `interval` seconds using
/// fresh data from the mesh node.
///
/// If `max_clients` is set, delists when that many clients are connected
/// and re-publishes when clients drop below the cap.
pub async fn publish_loop(node: crate::mesh::Node, keys: Keys, config: PublishLoopConfig) {
    let PublishLoopConfig {
        relays,
        name,
        region,
        max_clients,
        interval_secs,
        status_tx,
    } = config;
    let mut last_reported = None;
    let Some(publisher) =
        create_publish_loop_publisher(&keys, &relays, &status_tx, &mut last_reported).await
    else {
        return;
    };

    let npub = publisher.npub();
    log_publish_client_cap(max_clients);

    // Wait for local serving to be ready before first publish (up to 60s).
    wait_for_local_serving_ready(&node).await;
    eprintln!(
        "📡 Publishing mesh to Nostr (npub: {}...{})",
        &npub[..12],
        &npub[npub.len() - 8..]
    );

    let mut delisted = false;

    // Reusable client for solo-convergence discovery checks.
    let disco = DiscoveryClient::new(&relays).await.ok();

    loop {
        let peers = node.peers().await;
        let client_count = peer_client_count(&peers);
        if update_delisted_state(
            &publisher,
            max_clients,
            client_count,
            &mut delisted,
            interval_secs,
        )
        .await
        {
            continue;
        }

        if wait_while_delisted(delisted, interval_secs).await {
            continue;
        }

        if maybe_rejoin_larger_mesh(
            &node,
            &publisher,
            &relays,
            name.as_deref(),
            interval_secs,
            disco.as_ref(),
            &peers,
        )
        .await
        {
            continue;
        }

        let invite_token = node.invite_token();
        let listing = build_publish_listing(
            &node,
            &peers,
            invite_token,
            client_count,
            max_clients,
            name.clone(),
            region.clone(),
        )
        .await;

        publish_current_listing(
            &publisher,
            &listing,
            interval_secs,
            client_count,
            &status_tx,
            &mut last_reported,
        )
        .await;

        tokio::time::sleep(Duration::from_secs(interval_secs)).await;
    }
}

// ---------------------------------------------------------------------------
// Publish watchdog — take over publishing if the original publisher dies
// ---------------------------------------------------------------------------

/// Watch for our mesh's Nostr listing to disappear, then start publishing.
/// Multiple nodes may start publishing simultaneously — that's fine, each
/// publishes with their own key and invite token, giving discoverers
/// multiple entry points to the same mesh.
///
/// Only runs on active (non-client) nodes that joined via `--auto`.
pub async fn publish_watchdog(
    node: crate::mesh::Node,
    relays: Vec<String>,
    mesh_name: Option<String>,
    region: Option<String>,
    check_interval_secs: u64,
    status_tx: Option<tokio::sync::watch::Sender<Option<PublishStateUpdate>>>,
) {
    watchdog_initial_delay().await;

    // Reusable client for repeated discovery checks.
    let disco = DiscoveryClient::new(&relays).await.ok();
    let filter = MeshFilter::default();

    loop {
        match discover(&relays, &filter, disco.as_ref()).await {
            Ok(meshes) => {
                if should_take_over_publish(&node, &meshes).await {
                    if !confirm_missing_listing_after_backoff(
                        &relays,
                        &filter,
                        disco.as_ref(),
                        &node,
                    )
                    .await
                    {
                        tokio::time::sleep(Duration::from_secs(check_interval_secs)).await;
                        continue;
                    }

                    eprintln!("📡 Taking over Nostr publishing for the mesh");
                    let Some(keys) = load_watchdog_publish_keys(check_interval_secs).await else {
                        continue;
                    };
                    publish_loop(
                        node,
                        keys,
                        PublishLoopConfig {
                            relays,
                            name: mesh_name,
                            region,
                            max_clients: None,
                            interval_secs: 60,
                            status_tx,
                        },
                    )
                    .await;
                    return;
                }
            }
            Err(e) => {
                tracing::debug!("Publish watchdog: Nostr check failed: {e}");
            }
        }

        // Check frequently so we catch gaps fast
        let next_check = (rand::random::<u64>() % 15) + 20; // 20-35s
        tokio::time::sleep(Duration::from_secs(next_check)).await;
    }
}

// ---------------------------------------------------------------------------
// Discovery — find meshes on Nostr
// ---------------------------------------------------------------------------

/// Criteria for filtering discovered meshes.
#[derive(Debug, Clone, Default)]
pub struct MeshFilter {
    /// Match meshes by name (case-insensitive exact match)
    pub name: Option<String>,
    /// Match meshes serving (or wanting) this model name (substring match)
    pub model: Option<String>,
    /// Minimum total VRAM in GB
    pub min_vram_gb: Option<f64>,
    /// Geographic region
    pub region: Option<String>,
}

impl MeshFilter {
    pub fn matches(&self, mesh: &DiscoveredMesh) -> bool {
        if let Some(ref name) = self.name {
            match &mesh.listing.name {
                Some(n) if n.eq_ignore_ascii_case(name) => {}
                _ => return false,
            }
        }
        if let Some(ref model) = self.model {
            let model_lower = model.to_lowercase();
            let has_model = mesh
                .listing
                .serving
                .iter()
                .any(|m| m.to_lowercase().contains(&model_lower))
                || mesh
                    .listing
                    .wanted
                    .iter()
                    .any(|m| m.to_lowercase().contains(&model_lower))
                || mesh
                    .listing
                    .on_disk
                    .iter()
                    .any(|m| m.to_lowercase().contains(&model_lower));
            if !has_model {
                return false;
            }
        }
        if let Some(min_gb) = self.min_vram_gb {
            let vram_gb = mesh.listing.total_vram_bytes as f64 / 1e9;
            if vram_gb < min_gb {
                return false;
            }
        }
        if let Some(ref region) = self.region {
            match &mesh.listing.region {
                Some(r) if r.eq_ignore_ascii_case(region) => {}
                _ => return false,
            }
        }
        true
    }
}

/// A reusable read-only Nostr client for discovery.
/// Create once, pass to repeated `discover()` calls to avoid opening
/// new websocket connections and generating throwaway keys every time.
pub struct DiscoveryClient {
    client: Client,
}

impl DiscoveryClient {
    pub async fn new(relays: &[String]) -> Result<Self> {
        let _ = rustls::crypto::ring::default_provider().install_default();
        let keys = Keys::generate();
        let client = Client::new(keys);
        let mut added = 0;
        for relay in relays {
            match client.add_relay(relay).await {
                Ok(_) => added += 1,
                Err(e) => tracing::warn!("Nostr relay {relay}: {e}"),
            }
        }
        if added == 0 {
            anyhow::bail!(
                "Could not connect to any Nostr relay (tried {})",
                relays.len()
            );
        }
        client.connect().await;
        Ok(Self { client })
    }
}

/// Discover meshes from Nostr relays.
///
/// If `cached_client` is provided, reuses its connections.  Otherwise
/// creates (and drops) a one-shot client — fine for the initial
/// `--auto` join but wasteful in tight loops.
pub async fn discover(
    relays: &[String],
    filter: &MeshFilter,
    cached_client: Option<&DiscoveryClient>,
) -> Result<Vec<DiscoveredMesh>> {
    // Build a temporary client only when no cached one is supplied.
    let _tmp;
    let client: &Client = if let Some(cc) = cached_client {
        &cc.client
    } else {
        _tmp = build_discovery_client(relays).await?;
        &_tmp
    };

    let nostr_filter = Filter::new()
        .kind(Kind::Custom(MESH_SERVICE_KIND))
        .custom_tag(
            SingleLetterTag::lowercase(Alphabet::K),
            "mesh-llm".to_string(),
        )
        .limit(100);

    let events = match client
        .fetch_events(nostr_filter, Duration::from_secs(5))
        .await
    {
        Ok(e) => e,
        Err(e) => {
            tracing::warn!("Nostr fetch failed: {e}");
            return Ok(Vec::new()); // No results rather than hard error
        }
    };

    let now = Timestamp::now().as_secs();

    // Dedupe by publisher (keep latest per pubkey, using replaceable event semantics)
    let latest = latest_events_by_pubkey(&events);

    let mut meshes = Vec::new();
    for event in latest.values() {
        let Some(discovered) = parse_discovered_mesh(event, now) else {
            continue;
        };

        if filter.matches(&discovered) {
            meshes.push(discovered);
        }
    }

    // Sort by node count (bigger meshes first), then VRAM
    meshes.sort_by(|a, b| {
        b.listing
            .node_count
            .cmp(&a.listing.node_count)
            .then(b.listing.total_vram_bytes.cmp(&a.listing.total_vram_bytes))
    });

    Ok(meshes)
}

async fn wait_for_local_serving_ready(node: &crate::mesh::Node) {
    for _ in 0..120 {
        if node.is_llama_ready().await {
            return;
        }
        tokio::time::sleep(Duration::from_millis(500)).await;
    }
}

fn peer_client_count(peers: &[crate::mesh::PeerInfo]) -> usize {
    peers
        .iter()
        .filter(|peer| matches!(peer.role, crate::mesh::NodeRole::Client))
        .count()
}

fn non_client_peer_count(peers: &[crate::mesh::PeerInfo]) -> usize {
    peers
        .iter()
        .filter(|peer| !matches!(peer.role, crate::mesh::NodeRole::Client))
        .count()
}

async fn maybe_rejoin_larger_mesh(
    node: &crate::mesh::Node,
    publisher: &Publisher,
    relays: &[String],
    mesh_name: Option<&str>,
    interval_secs: u64,
    disco: Option<&DiscoveryClient>,
    peers: &[crate::mesh::PeerInfo],
) -> bool {
    let Some(my_node_count) = solo_mesh_node_count(peers) else {
        return false;
    };
    let Ok(listings) = discover(relays, &MeshFilter::default(), disco).await else {
        return false;
    };
    let my_npub = publisher.npub();
    let my_mesh_id = node.mesh_id().await;
    let target = pick_larger_mesh_target(
        &listings,
        &my_npub,
        my_mesh_id.as_deref(),
        mesh_name,
        my_node_count,
    );
    let Some(target) = target else {
        return false;
    };
    rejoin_larger_mesh_target(node, publisher, target, my_node_count, interval_secs).await
}

async fn create_publish_loop_publisher(
    keys: &Keys,
    relays: &[String],
    status_tx: &Option<tokio::sync::watch::Sender<Option<PublishStateUpdate>>>,
    last_reported: &mut Option<PublishStateUpdate>,
) -> Option<Publisher> {
    match Publisher::new(keys.clone(), relays).await {
        Ok(publisher) => Some(publisher),
        Err(err) => {
            report_publish_state(status_tx, last_reported, PublishStateUpdate::PublishFailed);
            tracing::error!("Failed to create Nostr publisher: {err}");
            None
        }
    }
}

fn log_publish_client_cap(max_clients: Option<usize>) {
    if let Some(cap) = max_clients {
        eprintln!("   Will delist when {} clients connected", cap);
    }
}

async fn wait_while_delisted(delisted: bool, interval_secs: u64) -> bool {
    if !delisted {
        return false;
    }
    tokio::time::sleep(Duration::from_secs(interval_secs)).await;
    true
}

async fn publish_current_listing(
    publisher: &Publisher,
    listing: &MeshListing,
    interval_secs: u64,
    client_count: usize,
    status_tx: &Option<tokio::sync::watch::Sender<Option<PublishStateUpdate>>>,
    last_reported: &mut Option<PublishStateUpdate>,
) {
    let ttl = interval_secs * 2;
    match publisher.publish(listing, ttl).await {
        Ok(()) => {
            report_publish_state(status_tx, last_reported, PublishStateUpdate::Public);
            tracing::debug!(
                "Published mesh listing ({} models, {} nodes, {} clients)",
                listing.serving.len(),
                listing.node_count,
                client_count
            );
        }
        Err(err) => {
            report_publish_state(status_tx, last_reported, PublishStateUpdate::PublishFailed);
            tracing::warn!("Failed to publish to Nostr: {err}");
        }
    }
}

async fn watchdog_initial_delay() {
    let jitter = (rand::random::<u64>() % 20) + 10;
    tokio::time::sleep(Duration::from_secs(jitter)).await;
}

async fn should_take_over_publish(node: &crate::mesh::Node, meshes: &[DiscoveredMesh]) -> bool {
    let our_peers = node.peers().await;
    let served = node.models_being_served().await;
    let our_mesh_id = node.mesh_id().await;
    !mesh_listing_present(meshes, our_mesh_id.as_deref(), &served)
        && (!our_peers.is_empty() || !served.is_empty())
}

async fn confirm_missing_listing_after_backoff(
    relays: &[String],
    filter: &MeshFilter,
    disco: Option<&DiscoveryClient>,
    node: &crate::mesh::Node,
) -> bool {
    let backoff = (rand::random::<u64>() % 7) + 3;
    eprintln!("📡 Mesh listing missing from Nostr — waiting {backoff}s before taking over...");
    tokio::time::sleep(Duration::from_secs(backoff)).await;

    let Ok(recheck) = discover(relays, filter, disco).await else {
        return true;
    };
    let served = node.models_being_served().await;
    let our_mesh_id = node.mesh_id().await;
    let still_missing = !mesh_listing_present(&recheck, our_mesh_id.as_deref(), &served);
    if !still_missing {
        eprintln!("📡 Someone else took over publishing — standing down");
    }
    still_missing
}

async fn load_watchdog_publish_keys(check_interval_secs: u64) -> Option<Keys> {
    match load_or_create_keys() {
        Ok(keys) => Some(keys),
        Err(err) => {
            tracing::warn!("Failed to load Nostr keys for publish takeover: {err}");
            tokio::time::sleep(Duration::from_secs(check_interval_secs)).await;
            None
        }
    }
}

fn solo_mesh_node_count(peers: &[crate::mesh::PeerInfo]) -> Option<usize> {
    let gpu_peers = non_client_peer_count(peers);
    (gpu_peers == 0).then_some(gpu_peers + 1)
}

async fn rejoin_larger_mesh_target(
    node: &crate::mesh::Node,
    publisher: &Publisher,
    target: &DiscoveredMesh,
    my_node_count: usize,
    interval_secs: u64,
) -> bool {
    eprintln!(
        "📡 Found larger mesh '{}' ({} nodes vs our {}) — rejoining",
        target.listing.name.as_deref().unwrap_or("unnamed"),
        target.listing.node_count,
        my_node_count
    );
    unpublish_before_rejoin(publisher).await;
    if join_larger_mesh(node, target).await.is_err() {
        tokio::time::sleep(Duration::from_secs(interval_secs)).await;
        return true;
    }
    eprintln!("📡 Merged into mesh — resuming publish as member");
    tokio::time::sleep(Duration::from_secs(30)).await;
    true
}

async fn unpublish_before_rejoin(publisher: &Publisher) {
    if let Err(e) = publisher.unpublish().await {
        tracing::warn!("Failed to unpublish solo listing: {e}");
    }
}

async fn join_larger_mesh(node: &crate::mesh::Node, target: &DiscoveredMesh) -> Result<()> {
    node.join(&target.listing.invite_token).await.map_err(|e| {
        tracing::warn!("Merge/rejoin failed: {e}");
        e
    })
}

fn pick_larger_mesh_target<'a>(
    listings: &'a [DiscoveredMesh],
    my_npub: &str,
    my_mesh_id: Option<&str>,
    mesh_name: Option<&str>,
    my_node_count: usize,
) -> Option<&'a DiscoveredMesh> {
    let split_target = my_mesh_id.and_then(|mesh_id| {
        listings.iter().find(|mesh| {
            mesh.listing.mesh_id.as_deref() == Some(mesh_id)
                && mesh.publisher_npub != my_npub
                && mesh.listing.node_count > my_node_count
        })
    });
    split_target.or_else(|| {
        (mesh_name.is_none()).then(|| {
            listings.iter().find(|mesh| {
                mesh.publisher_npub != my_npub
                    && mesh.listing.name.is_none()
                    && mesh.listing.node_count > my_node_count
            })
        })?
    })
}

async fn build_publish_listing(
    node: &crate::mesh::Node,
    peers: &[crate::mesh::PeerInfo],
    invite_token: String,
    client_count: usize,
    max_clients: Option<usize>,
    name: Option<String>,
    region: Option<String>,
) -> MeshListing {
    let serving = collect_actually_serving_models(node, peers).await;
    let served_set: std::collections::HashSet<&str> = serving.iter().map(String::as_str).collect();
    let wanted = collect_wanted_models(node, &served_set).await;
    let on_disk = collect_available_models(node, peers, &served_set).await;
    let total_vram_bytes = peers
        .iter()
        .filter(|peer| !matches!(peer.role, crate::mesh::NodeRole::Client))
        .map(|peer| peer.vram_bytes)
        .sum::<u64>()
        + node.vram_bytes();
    let node_count = non_client_peer_count(peers) + 1;
    MeshListing {
        invite_token,
        serving,
        wanted,
        on_disk,
        total_vram_bytes,
        node_count,
        client_count,
        max_clients: max_clients.unwrap_or(0),
        name,
        region,
        mesh_id: node.mesh_id().await,
    }
}

async fn collect_actually_serving_models(
    node: &crate::mesh::Node,
    peers: &[crate::mesh::PeerInfo],
) -> Vec<String> {
    let mut serving = Vec::new();
    if matches!(node.role().await, crate::mesh::NodeRole::Host { .. }) {
        extend_unique(&mut serving, node.hosted_models().await);
    }
    for peer in peers {
        if matches!(peer.role, crate::mesh::NodeRole::Host { .. }) {
            extend_unique(&mut serving, peer.routable_models());
        }
    }
    serving
}

async fn collect_wanted_models(
    node: &crate::mesh::Node,
    served_set: &std::collections::HashSet<&str>,
) -> Vec<String> {
    let mut wanted = Vec::new();
    for model in node.active_demand().await.keys() {
        if !served_set.contains(model.as_str()) && !wanted.contains(model) {
            wanted.push(model.clone());
        }
    }
    wanted
}

async fn collect_available_models(
    node: &crate::mesh::Node,
    peers: &[crate::mesh::PeerInfo],
    served_set: &std::collections::HashSet<&str>,
) -> Vec<String> {
    let mut available = Vec::new();
    extend_unique_filtered(&mut available, node.available_models().await, served_set);
    for peer in peers {
        extend_unique_filtered(&mut available, peer.available_models.clone(), served_set);
    }
    available
}

fn extend_unique(into: &mut Vec<String>, values: Vec<String>) {
    for value in values {
        if !into.contains(&value) {
            into.push(value);
        }
    }
}

fn extend_unique_filtered(
    into: &mut Vec<String>,
    values: Vec<String>,
    served_set: &std::collections::HashSet<&str>,
) {
    for value in values {
        if !served_set.contains(value.as_str()) && !into.contains(&value) {
            into.push(value);
        }
    }
}

fn mesh_listing_present(
    meshes: &[DiscoveredMesh],
    mesh_id: Option<&str>,
    served: &[String],
) -> bool {
    if let Some(mesh_id) = mesh_id {
        return meshes
            .iter()
            .any(|mesh| mesh.listing.mesh_id.as_deref() == Some(mesh_id));
    }
    !served.is_empty()
        && meshes.iter().any(|mesh| {
            served
                .iter()
                .any(|model| mesh.listing.serving.contains(model))
        })
}

async fn update_delisted_state(
    publisher: &Publisher,
    max_clients: Option<usize>,
    client_count: usize,
    delisted: &mut bool,
    interval_secs: u64,
) -> bool {
    let Some(cap) = max_clients else {
        return false;
    };
    if client_count >= cap && !*delisted {
        if let Err(e) = publisher.unpublish().await {
            tracing::warn!("Failed to unpublish from Nostr: {e}");
        }
        eprintln!(
            "📡 Delisted from Nostr ({} clients, cap is {})",
            client_count, cap
        );
        *delisted = true;
        tokio::time::sleep(Duration::from_secs(interval_secs)).await;
        return true;
    }
    if client_count < cap && *delisted {
        eprintln!(
            "📡 Re-publishing to Nostr ({} clients, cap is {})",
            client_count, cap
        );
        *delisted = false;
    }
    false
}

async fn build_discovery_client(relays: &[String]) -> Result<Client> {
    let _ = rustls::crypto::ring::default_provider().install_default();
    let keys = Keys::generate();
    let client = Client::new(keys);
    let mut added = 0;
    for relay in relays {
        match client.add_relay(relay).await {
            Ok(_) => added += 1,
            Err(e) => tracing::warn!("Nostr relay {relay}: {e}"),
        }
    }
    if added == 0 {
        anyhow::bail!(
            "Could not connect to any Nostr relay (tried {})",
            relays.len()
        );
    }
    client.connect().await;
    Ok(client)
}

fn latest_events_by_pubkey<'a>(events: &'a Events) -> std::collections::HashMap<String, &'a Event> {
    let mut latest: std::collections::HashMap<String, &'a Event> = std::collections::HashMap::new();
    for event in events.iter() {
        let pubkey = event.pubkey.to_hex();
        match latest.get(&pubkey) {
            Some(existing) if event.created_at.as_secs() <= existing.created_at.as_secs() => {}
            _ => {
                latest.insert(pubkey, event);
            }
        }
    }
    latest
}

fn parse_discovered_mesh(event: &Event, now: u64) -> Option<DiscoveredMesh> {
    let expires_at = event
        .tags
        .iter()
        .find(|t| t.as_slice().first().map(|s| s.as_str()) == Some("expiration"))
        .and_then(|t| t.as_slice().get(1))
        .and_then(|s| s.parse::<u64>().ok());
    if expires_at.is_some_and(|exp| exp < now) {
        return None;
    }

    let listing: MeshListing = match serde_json::from_str(&event.content) {
        Ok(listing) => listing,
        Err(err) => {
            tracing::warn!(
                "Skipping Nostr listing from {}: bad JSON: {err}",
                event.pubkey.to_bech32().unwrap_or_default()
            );
            return None;
        }
    };
    if let Err(err) = crate::mesh::Node::decode_invite_token(&listing.invite_token) {
        tracing::warn!(
            "Skipping Nostr listing from {}: {err}",
            event.pubkey.to_bech32().unwrap_or_default()
        );
        return None;
    }

    Some(DiscoveredMesh {
        listing,
        publisher_npub: event.pubkey.to_bech32().unwrap_or_default(),
        published_at: event.created_at.as_secs(),
        expires_at,
    })
}

// ---------------------------------------------------------------------------
// Smart auto-join: score meshes, detect staleness, prefer geo match
// ---------------------------------------------------------------------------

/// Is this mesh eligible for `--auto` when the user did not specify `--mesh-name`?
///
/// `--auto` joins the default community mesh. Eligible listings are:
///   - unnamed (the implicit default), or
///   - the blessed community name "mesh-llm".
///
/// Any other named mesh is still publicly discoverable on Nostr, but it is
/// not the default — the user must opt in by name via `--mesh-name`.
pub fn is_auto_eligible(mesh: &DiscoveredMesh) -> bool {
    match mesh.listing.name.as_deref() {
        None => true,
        Some(name) => name.eq_ignore_ascii_case("mesh-llm"),
    }
}

/// Score a mesh for auto-join. Higher = better.
/// Considers region match, capacity, and model availability.
/// Freshness is mostly irrelevant since Nostr listings expire at 120s (TTL=2×60s),
/// so anything we see from discover() is already reasonably fresh.
pub fn score_mesh(mesh: &DiscoveredMesh, _now_secs: u64, last_mesh_id: Option<&str>) -> i64 {
    let mut score: i64 = 100; // base score — if we can see it, it's alive

    // The canonical community mesh is an unnamed listing (`name: None`) — that
    // is what you get by default when you don't pass `--mesh-name`, and it's
    // what the public relay shows today. Give it a bonus so it ranks above
    // anything else in `--auto`. The literal name "mesh-llm" is treated as a
    // defensive alias for the same thing: nothing in the wild publishes with
    // that name right now, but older docs and test runs may, and if one ever
    // appears it should rank alongside unnamed rather than below it.
    //
    // Other named meshes are excluded from `--auto` entirely by
    // `is_auto_eligible`, so they don't get a score adjustment here — when
    // the user targets one via `--mesh-name`, the raw score is what matters
    // and any bonus or penalty would skew ranking.
    match mesh.listing.name.as_deref() {
        None => score += 300,
        Some(n) if n.eq_ignore_ascii_case("mesh-llm") => score += 300,
        Some(_) => {}
    }

    // Sticky preference: strong bonus for the mesh we were last on
    if let (Some(last_id), Some(mesh_id)) = (last_mesh_id, &mesh.listing.mesh_id) {
        if last_id == mesh_id {
            score += 500; // strong preference, not infinite — dead/degraded mesh loses on other factors
        }
    }

    // Capacity: prefer meshes that aren't full
    if mesh.listing.max_clients > 0 {
        if mesh.listing.client_count >= mesh.listing.max_clients {
            score -= 1000; // full — don't join
        } else {
            let headroom = mesh.listing.max_clients - mesh.listing.client_count;
            score += (headroom as i64).min(20); // some capacity bonus
        }
    }

    // Size: prefer meshes with more nodes (more resilient)
    score += (mesh.listing.node_count as i64).min(10) * 5;

    // Models: prefer meshes with more warm models
    score += (mesh.listing.serving.len() as i64) * 10;

    // Wanted models: mesh needs help — bonus if we'd be useful
    score += (mesh.listing.wanted.len() as i64) * 15;

    score
}

/// Decision from smart auto-join.
#[derive(Debug)]
pub enum AutoDecision {
    /// Ranked list of meshes to try joining (best first)
    Join {
        candidates: Vec<(String, DiscoveredMesh)>,
    },
    /// No suitable mesh found — start a new one with these models
    StartNew { models: Vec<String> },
}

/// Pick meshes to join, ranked by score, or decide to start a new one.
///
/// - Scores all discovered meshes (freshness, region, capacity)
/// - Filters out stale/full meshes
/// - Returns all viable candidates ranked by score so the caller
///   can probe each in order and fall back to the next on failure
pub fn smart_auto(
    meshes: &[DiscoveredMesh],
    my_vram_gb: f64,
    target_name: Option<&str>,
) -> AutoDecision {
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs();

    let last_mesh_id = crate::mesh::load_last_mesh_id();

    // If target name is set, only consider meshes with that exact name.
    // Otherwise `--auto` considers only the community mesh: unnamed listings
    // plus the blessed name "mesh-llm". Other named meshes are still publicly
    // discoverable on Nostr but must be opted into by name via `--mesh-name`.
    let candidates: Vec<&DiscoveredMesh> = if let Some(target) = target_name {
        meshes
            .iter()
            .filter(|m| {
                m.listing
                    .name
                    .as_ref()
                    .map(|n| n.eq_ignore_ascii_case(target))
                    .unwrap_or(false)
            })
            .collect()
    } else {
        meshes.iter().filter(|m| is_auto_eligible(m)).collect()
    };

    // Score and rank
    let mut scored: Vec<(&DiscoveredMesh, i64)> = candidates
        .iter()
        .map(|m| (*m, score_mesh(m, now, last_mesh_id.as_deref())))
        .collect();
    scored.sort_by_key(|entry| std::cmp::Reverse(entry.1));

    // Collect viable candidates.
    // If the user specified --mesh-name, take all candidates (they already
    // filtered by name above — the user explicitly asked for this mesh).
    // Otherwise, require positive score to filter out stale meshes.
    let viable: Vec<(String, DiscoveredMesh)> = scored
        .iter()
        .filter(|(_, score)| target_name.is_some() || *score > 0)
        .map(|(m, _)| (m.listing.invite_token.clone(), (*m).clone()))
        .collect();

    if !viable.is_empty() {
        return AutoDecision::Join { candidates: viable };
    }

    // No suitable mesh — recommend models for a new one based on VRAM
    let models = default_models_for_vram(my_vram_gb);
    AutoDecision::StartNew { models }
}

/// Model tiers by VRAM requirement (approximate loaded size × 1.1 headroom).
/// Model tiers for auto-selection, ordered largest-first.
/// min_vram = file_size * 1.1 rounded up. Prefer Qwen3 over 2.5 at same tier.
/// Parse a size string like "2.5GB" to GB as f64.
fn parse_size_gb(s: &str) -> f64 {
    s.trim_end_matches("GB").parse::<f64>().unwrap_or(0.0)
}

/// Build model tiers from the catalog, sorted largest first.
/// Each entry is (model_ref, min_vram_gb) where min_vram = file_size * 1.1.
/// Excludes draft models (< 1GB).
fn model_tiers() -> Vec<(String, f64)> {
    let _ = crate::models::remote_catalog::ensure_catalog();
    let mut tiers: Vec<_> = crate::models::remote_catalog::loaded_models()
        .unwrap_or_default()
        .into_iter()
        .filter_map(|m| {
            let size = m.size.as_deref()?;
            if parse_size_gb(size) < 1.0 {
                return None;
            }
            (
                crate::models::remote_catalog_model_ref(&m),
                parse_size_gb(size) * 1.1,
            )
                .into()
        })
        .collect();
    tiers.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
    tiers
}

/// Pick the model to SERVE for `--auto` based on VRAM.
/// Returns a single-element vec (the model this node should load).
///
/// One model per node. Biggest model that fits with 15% KV cache headroom.
///
/// Tiers:
///   <8GB:    Qwen3-4B (2.5G)
///   8-24GB:  Qwen3-8B (5G)
///   24-50GB: Qwen3.5-27B (17G) — vision + text
///   50-63GB: GLM-4.7-Flash (18G) — fast, tool calling
///   63-179GB: Qwen3-Coder-Next (48G) — frontier coder ~85B
///   179GB+:  MiniMax-M2.5 (138G) — flagship
pub fn auto_model_pack(vram_gb: f64) -> Vec<String> {
    let local_models = crate::models::scan_local_models();
    let tiers = model_tiers();

    // Helper: check if a model is on disk
    let on_disk = |name: &str| local_models.contains(&name.to_string());
    // Helper: model size from tiers
    let size_of = |name: &str| -> f64 {
        tiers
            .iter()
            .find(|(n, _)| *n == name)
            .map(|(_, s)| *s)
            .unwrap_or(0.0)
    };
    let usable = vram_gb * 0.85; // 15% headroom for KV cache

    // Opinionated packs — each is (generalist, optional specialist(s))
    // The order within a tier prefers: on-disk first, then opinionated default.
    struct Pack {
        min_vram: f64,
        models: Vec<String>,
    }
    let catalog_ref = |name: &str| {
        crate::models::find_remote_catalog_model_exact(name)
            .map(|model| crate::models::remote_catalog_model_ref(&model))
            .unwrap_or_else(|| name.to_string())
    };
    let packs: Vec<Pack> = vec![
        // One model per tier. Node serves one model at a time.
        Pack {
            min_vram: 179.0,
            models: vec![catalog_ref("MiniMax-M2.5-Q4_K_M")],
        },
        Pack {
            min_vram: 63.0,
            models: vec![catalog_ref("Qwen3-Coder-Next-Q4_K_M")],
        },
        Pack {
            min_vram: 50.0,
            models: vec![catalog_ref("GLM-4.7-Flash-Q4_K_M")],
        },
        Pack {
            min_vram: 24.0,
            models: vec![catalog_ref("Qwen3.5-27B-Q4_K_M")],
        },
        Pack {
            min_vram: 8.0,
            models: vec![catalog_ref("Qwen3-8B-Q4_K_M")],
        },
        Pack {
            min_vram: 0.0,
            models: vec![catalog_ref("Qwen3-4B-Q4_K_M")],
        },
    ];

    // Find the best pack that fits
    for pack in packs {
        if vram_gb < pack.min_vram {
            continue;
        }
        // Check all models in the pack actually fit within usable VRAM
        let total: f64 = pack.models.iter().map(|m| size_of(m)).sum();
        if total <= usable {
            return pack.models.clone();
        }
    }

    // Fallback: largest single model that fits, prefer on-disk
    let on_disk_fit = tiers
        .iter()
        .find(|(name, min_vram)| *min_vram <= usable && on_disk(name));
    let any_fit = tiers.iter().find(|(_, min_vram)| *min_vram <= usable);

    let primary = on_disk_fit
        .or(any_fit)
        .map(|(name, _)| name.to_string())
        .unwrap_or_else(|| catalog_ref("Qwen3-4B-Q4_K_M"));

    vec![primary]
}

/// Models to advertise as "wanted" for demand seeding.
/// These tell other nodes what the mesh could use, covering every VRAM tier.
/// NOT served by this node — just demand hints for the mesh.
pub fn demand_seed_models() -> Vec<String> {
    [
        "Qwen3-Coder-Next-Q4_K_M",
        "Qwen3.5-27B-Q4_K_M",
        "GLM-4.7-Flash-Q4_K_M",
        "Qwen3-8B-Q4_K_M",
        "Qwen3-4B-Q4_K_M",
        "Qwen3-0.6B-Q4_K_M",
    ]
    .into_iter()
    .map(|name| {
        crate::models::find_remote_catalog_model_exact(name)
            .map(|model| crate::models::remote_catalog_model_ref(&model))
            .unwrap_or_else(|| name.to_string())
    })
    .collect()
}

/// Legacy wrapper — returns serving models + demand seeds combined.
/// Used by `smart_auto` for the StartNew decision.
pub fn default_models_for_vram(vram_gb: f64) -> Vec<String> {
    let mut models = auto_model_pack(vram_gb);
    for m in demand_seed_models() {
        if !models.contains(&m) {
            models.push(m);
        }
    }
    models
}

#[cfg(test)]
mod auto_pack_tests {
    use super::*;

    fn catalog_ref(name: &str) -> String {
        crate::models::find_remote_catalog_model_exact(name)
            .map(|model| crate::models::remote_catalog_model_ref(&model))
            .unwrap_or_else(|| name.to_string())
    }

    #[test]
    fn pack_4gb_starter() {
        let pack = auto_model_pack(4.0);
        assert_eq!(pack, vec![catalog_ref("Qwen3-4B-Q4_K_M")]);
    }

    #[test]
    fn pack_8gb_single_model() {
        let pack = auto_model_pack(8.0);
        assert_eq!(pack, vec![catalog_ref("Qwen3-8B-Q4_K_M")]);
    }

    #[test]
    fn pack_16gb_single() {
        let pack = auto_model_pack(16.0);
        assert_eq!(pack, vec![catalog_ref("Qwen3-8B-Q4_K_M")]);
    }

    #[test]
    fn pack_24gb_vision() {
        let pack = auto_model_pack(24.0);
        assert_eq!(pack, vec![catalog_ref("Qwen3.5-27B-Q4_K_M")]);
    }

    #[test]
    fn pack_50gb_glm_flash() {
        let pack = auto_model_pack(50.0);
        assert_eq!(pack, vec![catalog_ref("GLM-4.7-Flash-Q4_K_M")]);
    }

    #[test]
    fn pack_63gb_frontier_coder() {
        let pack = auto_model_pack(63.0);
        assert_eq!(pack, vec![catalog_ref("Qwen3-Coder-Next-Q4_K_M")]);
    }

    #[test]
    fn pack_85gb_frontier_coder() {
        let pack = auto_model_pack(85.0);
        assert_eq!(pack, vec![catalog_ref("Qwen3-Coder-Next-Q4_K_M")]);
    }

    #[test]
    fn pack_206gb_minimax() {
        let pack = auto_model_pack(206.0);
        assert_eq!(pack, vec![catalog_ref("MiniMax-M2.5-Q4_K_M")]);
    }

    #[test]
    fn pack_between_tiers_falls_through() {
        // 40GB: below 50GB tier, falls to 24GB tier (Qwen3.5-27B)
        let pack = auto_model_pack(40.0);
        assert_eq!(pack, vec![catalog_ref("Qwen3.5-27B-Q4_K_M")]);
    }

    #[test]
    fn demand_seeds_are_separate() {
        let seeds = demand_seed_models();
        assert!(seeds.len() >= 4);
        assert!(seeds.contains(&catalog_ref("Qwen3-0.6B-Q4_K_M")));
        assert!(seeds.contains(&catalog_ref("Qwen3-Coder-Next-Q4_K_M")));
    }

    #[test]
    fn default_models_includes_both() {
        let all = default_models_for_vram(30.0);
        let pack = auto_model_pack(30.0);
        let seeds = demand_seed_models();
        // Pack models come first
        for m in &pack {
            assert!(
                all.contains(m),
                "pack model {m} missing from default_models"
            );
        }
        // Seeds are also present
        for m in &seeds {
            assert!(
                all.contains(m),
                "seed model {m} missing from default_models"
            );
        }
        // No duplicates
        let mut deduped = all.clone();
        deduped.sort();
        deduped.dedup();
        assert_eq!(all.len(), deduped.len());
    }
}

// ---------------------------------------------------------------------------
// Unit tests: score_mesh, smart_auto, MeshFilter
// ---------------------------------------------------------------------------
#[cfg(test)]
mod scoring_tests {
    use super::*;

    fn make_mesh(
        name: Option<&str>,
        mesh_id: Option<&str>,
        serving: &[&str],
        node_count: usize,
        vram: u64,
        clients: usize,
        max_clients: usize,
    ) -> DiscoveredMesh {
        DiscoveredMesh {
            listing: MeshListing {
                invite_token: format!("invite-{}", mesh_id.unwrap_or("test")),
                serving: serving.iter().map(|s| s.to_string()).collect(),
                wanted: vec![],
                on_disk: vec![],
                total_vram_bytes: vram,
                node_count,
                client_count: clients,
                max_clients,
                name: name.map(|s| s.to_string()),
                region: None,
                mesh_id: mesh_id.map(|s| s.to_string()),
            },
            publisher_npub: format!("npub-{}", mesh_id.unwrap_or("test")),
            published_at: 1000,
            expires_at: Some(2000),
        }
    }

    #[test]
    fn score_unnamed_community_mesh_bonus() {
        // Unnamed is the canonical community mesh and should get the bonus.
        let mesh = make_mesh(
            None,
            Some("abc"),
            &["Qwen3-8B-Q4_K_M"],
            3,
            48_000_000_000,
            1,
            10,
        );
        let score = score_mesh(&mesh, 1500, None);
        // base(100) + community(300) + headroom + nodes(15) + models(10)
        assert!(
            score > 400,
            "unnamed community mesh should score high, got {score}"
        );
    }

    #[test]
    fn score_mesh_llm_alias_matches_unnamed() {
        // "mesh-llm" is a defensive alias for the community mesh and must
        // score identically to an unnamed listing with equivalent stats.
        let unnamed = make_mesh(None, Some("u"), &["m1"], 2, 24_000_000_000, 0, 0);
        let alias = make_mesh(
            Some("mesh-llm"),
            Some("a"),
            &["m1"],
            2,
            24_000_000_000,
            0,
            0,
        );
        assert_eq!(
            score_mesh(&unnamed, 1500, None),
            score_mesh(&alias, 1500, None)
        );
    }

    #[test]
    fn public_listing_json_does_not_expose_control_endpoint_data() {
        let control_endpoint =
            "control://mesh-llm-control/1?token=owner-only-endpoint-never-public";
        let mesh = make_mesh(
            Some("mesh-llm"),
            Some("public-mesh"),
            &["Qwen3-8B-Q4_K_M"],
            2,
            24_000_000_000,
            0,
            0,
        );

        let listing_json = serde_json::to_string(&mesh.listing).expect("listing must serialize");
        let discovered_json = serde_json::to_string(&mesh).expect("discovered mesh must serialize");

        assert!(
            !listing_json.contains(control_endpoint),
            "published MeshListing JSON must not leak control endpoint data"
        );
        assert!(
            !discovered_json.contains(control_endpoint),
            "discovered public listing JSON must not leak control endpoint data"
        );
        assert!(
            !listing_json.contains("mesh-llm-control/1"),
            "published MeshListing JSON must not mention the owner-control ALPN"
        );
        assert!(
            !discovered_json.contains("owner-only-endpoint"),
            "public discovery JSON must not expose owner-only endpoint tokens"
        );
    }

    #[test]
    fn score_other_named_mesh_no_community_bonus() {
        // Non-community named meshes are excluded from --auto entirely by
        // `is_auto_eligible`; within `score_mesh` they simply don't get the
        // community bonus. When the user targets one via --mesh-name, the
        // raw score is what's used to rank.
        let mesh = make_mesh(
            Some("bobs-cluster"),
            Some("xyz"),
            &["Qwen3-8B-Q4_K_M"],
            3,
            48_000_000_000,
            0,
            0,
        );
        let score = score_mesh(&mesh, 1500, None);
        // base(100) + nodes(15) + models(10) — no community bonus, no penalty
        assert!(
            score < 300,
            "non-community named mesh should not get community bonus, got {score}"
        );
        assert!(score > 0, "named mesh with real nodes should be positive");
    }

    #[test]
    fn other_named_mesh_not_auto_eligible() {
        let bobs = make_mesh(Some("bobs-cluster"), Some("x"), &[], 1, 0, 0, 0);
        let community = make_mesh(Some("mesh-llm"), Some("c"), &[], 1, 0, 0, 0);
        let community_caps = make_mesh(Some("MESH-LLM"), Some("c2"), &[], 1, 0, 0, 0);
        let unnamed = make_mesh(None, Some("u"), &[], 1, 0, 0, 0);
        assert!(!is_auto_eligible(&bobs));
        assert!(is_auto_eligible(&community));
        assert!(is_auto_eligible(&community_caps));
        assert!(is_auto_eligible(&unnamed));
    }

    #[test]
    fn score_full_mesh_penalty() {
        let mesh = make_mesh(
            None,
            Some("full"),
            &["Qwen3-8B-Q4_K_M"],
            2,
            16_000_000_000,
            5,
            5,
        );
        let score = score_mesh(&mesh, 1500, None);
        assert!(score < 0, "full mesh should score negative, got {score}");
    }

    #[test]
    fn score_sticky_mesh_bonus() {
        let mesh = make_mesh(
            None,
            Some("my-mesh"),
            &["Qwen3-8B-Q4_K_M"],
            2,
            16_000_000_000,
            0,
            0,
        );
        let score_sticky = score_mesh(&mesh, 1500, Some("my-mesh"));
        let score_fresh = score_mesh(&mesh, 1500, None);
        assert!(
            score_sticky > score_fresh + 400,
            "sticky bonus should be large, sticky={score_sticky} fresh={score_fresh}"
        );
    }

    #[test]
    fn score_more_nodes_better() {
        let small = make_mesh(
            None,
            Some("s"),
            &["Qwen3-8B-Q4_K_M"],
            1,
            8_000_000_000,
            0,
            0,
        );
        let big = make_mesh(
            None,
            Some("b"),
            &["Qwen3-8B-Q4_K_M"],
            5,
            40_000_000_000,
            0,
            0,
        );
        assert!(score_mesh(&big, 1500, None) > score_mesh(&small, 1500, None));
    }

    #[test]
    fn score_more_models_better() {
        let one = make_mesh(
            None,
            Some("1"),
            &["Qwen3-8B-Q4_K_M"],
            2,
            16_000_000_000,
            0,
            0,
        );
        let two = make_mesh(
            None,
            Some("2"),
            &["Qwen3-8B-Q4_K_M", "Qwen3-32B-Q4_K_M"],
            2,
            40_000_000_000,
            0,
            0,
        );
        assert!(score_mesh(&two, 1500, None) > score_mesh(&one, 1500, None));
    }
}

#[cfg(test)]
mod filter_tests {
    use super::*;

    fn make_mesh_for_filter(
        serving: &[&str],
        wanted: &[&str],
        on_disk: &[&str],
        vram: u64,
        region: Option<&str>,
    ) -> DiscoveredMesh {
        make_mesh_for_filter_named(serving, wanted, on_disk, vram, region, None)
    }

    fn make_mesh_for_filter_named(
        serving: &[&str],
        wanted: &[&str],
        on_disk: &[&str],
        vram: u64,
        region: Option<&str>,
        name: Option<&str>,
    ) -> DiscoveredMesh {
        DiscoveredMesh {
            listing: MeshListing {
                invite_token: "tok".into(),
                serving: serving.iter().map(|s| s.to_string()).collect(),
                wanted: wanted.iter().map(|s| s.to_string()).collect(),
                on_disk: on_disk.iter().map(|s| s.to_string()).collect(),
                total_vram_bytes: vram,
                node_count: 1,
                client_count: 0,
                max_clients: 0,
                name: name.map(|s| s.to_string()),
                region: region.map(|s| s.to_string()),
                mesh_id: None,
            },
            publisher_npub: "npub-test".into(),
            published_at: 1000,
            expires_at: Some(2000),
        }
    }

    #[test]
    fn filter_default_matches_all() {
        let m = make_mesh_for_filter(&["Qwen3-8B-Q4_K_M"], &[], &[], 8_000_000_000, None);
        assert!(MeshFilter::default().matches(&m));
    }

    #[test]
    fn filter_model_serving() {
        let m = make_mesh_for_filter(&["Qwen3-8B-Q4_K_M"], &[], &[], 8_000_000_000, None);
        let f = MeshFilter {
            model: Some("qwen3-8b".into()),
            ..Default::default()
        };
        assert!(f.matches(&m));
    }

    #[test]
    fn filter_model_wanted() {
        let m = make_mesh_for_filter(&[], &["Qwen3-32B-Q4_K_M"], &[], 8_000_000_000, None);
        let f = MeshFilter {
            model: Some("32b".into()),
            ..Default::default()
        };
        assert!(f.matches(&m));
    }

    #[test]
    fn filter_model_on_disk() {
        let m = make_mesh_for_filter(&[], &[], &["MiniMax-M2.5-Q4_K_M"], 8_000_000_000, None);
        let f = MeshFilter {
            model: Some("minimax".into()),
            ..Default::default()
        };
        assert!(f.matches(&m));
    }

    #[test]
    fn filter_model_no_match() {
        let m = make_mesh_for_filter(&["Qwen3-8B-Q4_K_M"], &[], &[], 8_000_000_000, None);
        let f = MeshFilter {
            model: Some("llama".into()),
            ..Default::default()
        };
        assert!(!f.matches(&m));
    }

    #[test]
    fn filter_min_vram() {
        let m = make_mesh_for_filter(&[], &[], &[], 8_000_000_000, None);
        let pass = MeshFilter {
            min_vram_gb: Some(5.0),
            ..Default::default()
        };
        let fail = MeshFilter {
            min_vram_gb: Some(16.0),
            ..Default::default()
        };
        assert!(pass.matches(&m));
        assert!(!fail.matches(&m));
    }

    #[test]
    fn filter_region() {
        let m = make_mesh_for_filter(&[], &[], &[], 8_000_000_000, Some("us-east"));
        let pass = MeshFilter {
            region: Some("us-east".into()),
            ..Default::default()
        };
        let fail = MeshFilter {
            region: Some("eu-west".into()),
            ..Default::default()
        };
        assert!(pass.matches(&m));
        assert!(!fail.matches(&m));
    }

    #[test]
    fn filter_region_case_insensitive() {
        let m = make_mesh_for_filter(&[], &[], &[], 8_000_000_000, Some("US-East"));
        let f = MeshFilter {
            region: Some("us-east".into()),
            ..Default::default()
        };
        assert!(f.matches(&m));
    }

    #[test]
    fn filter_combined() {
        let m = make_mesh_for_filter(
            &["Qwen3-8B-Q4_K_M"],
            &[],
            &[],
            16_000_000_000,
            Some("us-east"),
        );
        let pass = MeshFilter {
            model: Some("qwen3".into()),
            min_vram_gb: Some(10.0),
            region: Some("us-east".into()),
            ..Default::default()
        };
        let fail_model = MeshFilter {
            model: Some("llama".into()),
            min_vram_gb: Some(10.0),
            region: Some("us-east".into()),
            ..Default::default()
        };
        assert!(pass.matches(&m));
        assert!(!fail_model.matches(&m));
    }

    #[test]
    fn filter_name_exact() {
        let m = make_mesh_for_filter_named(&[], &[], &[], 8_000_000_000, None, Some("poker-night"));
        let f = MeshFilter {
            name: Some("poker-night".into()),
            ..Default::default()
        };
        assert!(f.matches(&m));
    }

    #[test]
    fn filter_name_case_insensitive() {
        let m = make_mesh_for_filter_named(&[], &[], &[], 8_000_000_000, None, Some("Poker-Night"));
        let f = MeshFilter {
            name: Some("poker-night".into()),
            ..Default::default()
        };
        assert!(f.matches(&m));
    }

    #[test]
    fn filter_name_no_match() {
        let m = make_mesh_for_filter_named(&[], &[], &[], 8_000_000_000, None, Some("other-mesh"));
        let f = MeshFilter {
            name: Some("poker-night".into()),
            ..Default::default()
        };
        assert!(!f.matches(&m));
    }

    #[test]
    fn filter_name_mesh_unnamed() {
        // Mesh has no name — filter by name should not match.
        let m = make_mesh_for_filter_named(&[], &[], &[], 8_000_000_000, None, None);
        let f = MeshFilter {
            name: Some("poker-night".into()),
            ..Default::default()
        };
        assert!(!f.matches(&m));
    }

    #[test]
    fn filter_name_none_matches_all() {
        // No name filter — matches meshes with and without names.
        let named =
            make_mesh_for_filter_named(&[], &[], &[], 8_000_000_000, None, Some("poker-night"));
        let unnamed = make_mesh_for_filter_named(&[], &[], &[], 8_000_000_000, None, None);
        let f = MeshFilter::default();
        assert!(f.matches(&named));
        assert!(f.matches(&unnamed));
    }
}

#[cfg(test)]
mod smart_auto_tests {
    use super::*;

    fn make_mesh(
        name: Option<&str>,
        mesh_id: &str,
        serving: &[&str],
        node_count: usize,
        vram: u64,
        clients: usize,
        max_clients: usize,
    ) -> DiscoveredMesh {
        DiscoveredMesh {
            listing: MeshListing {
                invite_token: format!("invite-{mesh_id}"),
                serving: serving.iter().map(|s| s.to_string()).collect(),
                wanted: vec![],
                on_disk: vec![],
                total_vram_bytes: vram,
                node_count,
                client_count: clients,
                max_clients,
                name: name.map(|s| s.to_string()),
                region: None,
                mesh_id: Some(mesh_id.to_string()),
            },
            publisher_npub: format!("npub-{mesh_id}"),
            published_at: 1000,
            expires_at: Some(2000),
        }
    }

    #[test]
    fn smart_auto_both_community_aliases_eligible() {
        // Both unnamed listings and the "mesh-llm" alias are eligible for
        // `--auto` and score equally on name. With other factors equal they
        // tie at the same score; what matters here is that both appear as
        // candidates and that `"mesh-llm"` is no longer penalised relative
        // to unnamed.
        let meshes = vec![
            make_mesh(None, "ccc", &["Qwen3-8B-Q4_K_M"], 2, 24_000_000_000, 0, 0),
            make_mesh(
                Some("mesh-llm"),
                "aaa",
                &["Qwen3-8B-Q4_K_M"],
                2,
                24_000_000_000,
                0,
                0,
            ),
        ];
        let now = 1500;
        let unnamed_score = score_mesh(&meshes[0], now, None);
        let alias_score = score_mesh(&meshes[1], now, None);
        assert_eq!(
            unnamed_score, alias_score,
            "None and 'mesh-llm' should score equally as community aliases",
        );
        match smart_auto(&meshes, 8.0, None) {
            AutoDecision::Join { candidates } => {
                assert_eq!(candidates.len(), 2);
            }
            AutoDecision::StartNew { .. } => panic!("should join, not start new"),
        }
    }

    #[test]
    fn smart_auto_excludes_other_named_meshes() {
        // Without --mesh-name, --auto must only consider the community mesh
        // (unnamed or name == "mesh-llm"). Other named meshes — even though
        // they are publicly discoverable on Nostr — should never appear as
        // candidates unless the user targets them by name.
        let meshes = vec![
            make_mesh(
                Some("bobs-cluster"),
                "bbb",
                &["Qwen3-8B-Q4_K_M"],
                5,
                80_000_000_000,
                0,
                0,
            ),
            make_mesh(
                Some("alice-cluster"),
                "aac",
                &["Qwen3-8B-Q4_K_M"],
                3,
                24_000_000_000,
                0,
                0,
            ),
        ];
        match smart_auto(&meshes, 8.0, None) {
            AutoDecision::Join { .. } => {
                panic!("other named meshes must not be joined by --auto")
            }
            AutoDecision::StartNew { models } => {
                assert!(!models.is_empty());
            }
        }
    }

    #[test]
    fn smart_auto_larger_unnamed_beats_smaller_alias() {
        // Both unnamed and "mesh-llm" are eligible with the same name bonus.
        // With capacity as the tiebreaker, the larger unnamed mesh wins —
        // which is what we want, since unnamed is the canonical identity
        // of the public community mesh.
        let meshes = vec![
            make_mesh(
                None,
                "unnamed-1",
                &["Qwen3-8B-Q4_K_M"],
                5,
                40_000_000_000,
                0,
                0,
            ),
            make_mesh(
                Some("mesh-llm"),
                "alias-1",
                &["Qwen3-8B-Q4_K_M"],
                2,
                16_000_000_000,
                0,
                0,
            ),
        ];
        match smart_auto(&meshes, 8.0, None) {
            AutoDecision::Join { candidates } => {
                assert_eq!(candidates.len(), 2);
                assert_eq!(candidates[0].0, "invite-unnamed-1");
            }
            AutoDecision::StartNew { .. } => panic!("should join"),
        }
    }

    #[test]
    fn smart_auto_filters_full_mesh() {
        let meshes = vec![make_mesh(
            None,
            "full",
            &["Qwen3-8B-Q4_K_M"],
            2,
            16_000_000_000,
            10,
            10,
        )];
        match smart_auto(&meshes, 8.0, None) {
            AutoDecision::Join { candidates } => {
                // Full mesh should still appear (score might be negative but target_name is None
                // so it filters on score > 0)
                assert!(candidates.is_empty(), "full mesh should be filtered out");
            }
            AutoDecision::StartNew { models } => {
                assert!(!models.is_empty());
            }
        }
    }

    #[test]
    fn smart_auto_target_name_filters() {
        let meshes = vec![
            make_mesh(
                Some("mesh-llm"),
                "aaa",
                &["Qwen3-8B-Q4_K_M"],
                3,
                48_000_000_000,
                1,
                10,
            ),
            make_mesh(
                Some("private"),
                "bbb",
                &["Qwen3-32B-Q4_K_M"],
                2,
                40_000_000_000,
                0,
                0,
            ),
        ];
        match smart_auto(&meshes, 8.0, Some("private")) {
            AutoDecision::Join { candidates } => {
                assert!(!candidates.is_empty());
                // Only "private" mesh should match
                for (token, _) in &candidates {
                    assert_eq!(token, "invite-bbb");
                }
            }
            AutoDecision::StartNew { .. } => panic!("should find the named mesh"),
        }
    }

    #[test]
    fn smart_auto_empty_starts_new() {
        match smart_auto(&[], 24.0, None) {
            AutoDecision::StartNew { models } => {
                assert!(!models.is_empty());
            }
            AutoDecision::Join { .. } => panic!("no meshes should mean start new"),
        }
    }

    #[test]
    fn smart_auto_sticky_preference() {
        // Save a fake last-mesh
        let dir = dirs::home_dir().unwrap().join(".mesh-llm");
        let path = dir.join("last-mesh");
        let had_file = path.exists();
        let old_content = if had_file {
            std::fs::read_to_string(&path).ok()
        } else {
            None
        };

        // Write our test mesh_id
        std::fs::create_dir_all(&dir).ok();
        std::fs::write(&path, "sticky-mesh").ok();

        let meshes = vec![
            make_mesh(None, "other", &["Qwen3-8B-Q4_K_M"], 3, 24_000_000_000, 0, 0),
            make_mesh(
                None,
                "sticky-mesh",
                &["Qwen3-8B-Q4_K_M"],
                2,
                16_000_000_000,
                0,
                0,
            ),
        ];
        let result = smart_auto(&meshes, 8.0, None);

        // Restore
        if let Some(old) = old_content {
            std::fs::write(&path, old).ok();
        } else if had_file {
            // shouldn't happen but be safe
        } else {
            std::fs::remove_file(&path).ok();
        }

        match result {
            AutoDecision::Join { candidates } => {
                assert!(!candidates.is_empty());
                // Sticky mesh should be first despite fewer nodes
                assert_eq!(candidates[0].0, "invite-sticky-mesh");
            }
            AutoDecision::StartNew { .. } => panic!("should join"),
        }
    }
}

#[cfg(test)]
mod rotate_key_tests {
    use super::*;
    use serial_test::serial;
    use std::fs;

    // rotate_keys uses hardcoded paths (~/.mesh-llm/), so we test the logic
    // by verifying files are created/deleted in the real location.
    // This is safe because rotate_keys only deletes key and nostr.nsec.
    //
    // Both scenarios (keys present + keys missing) run in a single test to
    // avoid a race — Rust runs tests in parallel and both would touch the
    // same files.

    #[test]
    #[serial]
    fn rotate_deletes_both_keys_and_handles_missing() {
        let dir = dirs::home_dir().unwrap().join(".mesh-llm");
        fs::create_dir_all(&dir).ok();

        let key_path = dir.join("key");
        let nsec_path = dir.join("nostr.nsec");

        // Save originals so we can restore after the test.
        let orig_key = if key_path.exists() {
            Some(fs::read(&key_path).unwrap())
        } else {
            None
        };
        let orig_nsec = if nsec_path.exists() {
            Some(fs::read(&nsec_path).unwrap())
        } else {
            None
        };

        // --- Scenario 1: both keys exist → rotate deletes them ---
        fs::write(&key_path, b"test-node-key").unwrap();
        fs::write(&nsec_path, b"test-nostr-nsec").unwrap();

        let result = rotate_keys();
        assert!(result.is_ok(), "rotate should succeed when keys exist");
        assert!(!key_path.exists(), "node key should be deleted");
        assert!(!nsec_path.exists(), "nostr key should be deleted");

        // --- Scenario 2: no keys on disk → rotate still succeeds ---
        // (files were just deleted above, so the directory is clean)
        let result = rotate_keys();
        assert!(result.is_ok(), "rotate should succeed even with no keys");

        // Restore originals.
        if let Some(k) = orig_key {
            fs::write(&key_path, k).ok();
        }
        if let Some(n) = orig_nsec {
            fs::write(&nsec_path, n).ok();
        }
    }
}

#[cfg(test)]
mod key_file_tests {
    use super::*;
    use std::path::PathBuf;
    use std::time::{SystemTime, UNIX_EPOCH};

    fn temp_key_path(prefix: &str) -> PathBuf {
        let unique = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        std::env::temp_dir()
            .join(format!("{prefix}-{unique}"))
            .join("nostr.nsec")
    }

    #[test]
    fn load_or_create_keys_at_round_trips() {
        let path = temp_key_path("mesh-llm-nostr-key");
        let first = load_or_create_keys_at(&path).unwrap();
        let second = load_or_create_keys_at(&path).unwrap();
        assert_eq!(
            first.secret_key().to_bech32().unwrap(),
            second.secret_key().to_bech32().unwrap()
        );
        let _ = std::fs::remove_dir_all(path.parent().unwrap());
    }

    #[cfg(unix)]
    #[test]
    fn load_or_create_keys_at_hardens_existing_permissions() {
        use std::os::unix::fs::PermissionsExt;

        let path = temp_key_path("mesh-llm-nostr-key-perms");
        let dir = path.parent().unwrap();
        std::fs::create_dir_all(dir).unwrap();
        std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o755)).unwrap();

        let keys = Keys::generate();
        let nsec = keys.secret_key().to_bech32().unwrap();
        std::fs::write(&path, &nsec).unwrap();
        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();

        let loaded = load_or_create_keys_at(&path).unwrap();
        assert_eq!(
            loaded.secret_key().to_bech32().unwrap(),
            keys.secret_key().to_bech32().unwrap()
        );
        assert_eq!(
            std::fs::metadata(dir).unwrap().permissions().mode() & 0o777,
            0o700
        );
        assert_eq!(
            std::fs::metadata(&path).unwrap().permissions().mode() & 0o777,
            0o600
        );

        let _ = std::fs::remove_dir_all(dir);
    }

    #[cfg(unix)]
    #[test]
    fn load_or_create_keys_at_rejects_symlink_key() {
        use std::os::unix::fs::PermissionsExt;

        let path = temp_key_path("mesh-llm-nostr-key-symlink");
        let dir = path.parent().unwrap();
        std::fs::create_dir_all(dir).unwrap();
        let real_file = dir.join("nostr.real");
        let keys = Keys::generate();
        let nsec = keys.secret_key().to_bech32().unwrap();
        std::fs::write(&real_file, &nsec).unwrap();
        std::fs::set_permissions(&real_file, std::fs::Permissions::from_mode(0o600)).unwrap();
        std::os::unix::fs::symlink(&real_file, &path).unwrap();

        let result = load_or_create_keys_at(&path);
        assert!(result.is_err(), "expected error for symlinked nostr key");

        let _ = std::fs::remove_dir_all(dir);
    }
}

// ---------------------------------------------------------------------------
// Integration test — publish/discover against real Nostr relays
// ---------------------------------------------------------------------------
#[cfg(test)]
mod integration_tests {
    use super::*;
    use base64::Engine;

    fn fake_invite_token() -> String {
        // Build a syntactically valid invite token by encoding a minimal
        // EndpointAddr JSON. We use Node::invite_token indirectly by just
        // crafting the JSON that decode_invite_token expects.
        use std::sync::atomic::{AtomicU64, Ordering};
        static COUNTER: AtomicU64 = AtomicU64::new(1);
        let n = COUNTER.fetch_add(1, Ordering::Relaxed);
        // EndpointAddr serialises as {"id":"<base32>","addrs":[]}
        // We need a valid 32-byte public key. Use a deterministic one.
        let mut seed = [0u8; 32];
        seed[..8].copy_from_slice(&n.to_le_bytes());
        let key = iroh::SecretKey::from_bytes(&seed);
        let addr = iroh::EndpointAddr {
            id: iroh::EndpointId::from(key.public()),
            addrs: Default::default(),
        };
        let json = serde_json::to_vec(&addr).expect("serialize");
        base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(json)
    }

    /// End-to-end: two publishers advertise the same mesh, a reusable
    /// DiscoveryClient finds both listings, and fields round-trip correctly.
    /// Covers publish, discover, multi-publisher, and client reuse in one test.
    #[tokio::test]
    #[ignore = "requires live public Nostr relays"]
    async fn publish_discover_round_trip() {
        let relays: Vec<String> = DEFAULT_RELAYS.iter().map(|s| s.to_string()).collect();
        let mesh_name = format!("mesh-llm-test-{}", rand::random::<u32>());
        let mesh_id = format!("test-id-{}", rand::random::<u32>());

        // Publisher A
        let keys_a = Keys::generate();
        let pub_a = Publisher::new(keys_a.clone(), &relays)
            .await
            .expect("pub_a");
        let token_a = fake_invite_token();
        let token_b = fake_invite_token();
        let listing_a = MeshListing {
            invite_token: token_a.clone(),
            serving: vec!["Qwen3-8B-Q4_K_M".into()],
            wanted: vec![],
            on_disk: vec![],
            total_vram_bytes: 16_000_000_000,
            node_count: 2,
            client_count: 0,
            max_clients: 0,
            name: Some(mesh_name.clone()),
            region: Some("test-region".into()),
            mesh_id: Some(mesh_id.clone()),
        };
        pub_a.publish(&listing_a, 120).await.expect("publish A");

        // Publisher B — same mesh, different invite token
        let keys_b = Keys::generate();
        let pub_b = Publisher::new(keys_b.clone(), &relays)
            .await
            .expect("pub_b");
        let mut listing_b = listing_a.clone();
        listing_b.invite_token = token_b.clone();
        pub_b.publish(&listing_b, 120).await.expect("publish B");

        tokio::time::sleep(Duration::from_secs(3)).await;

        // Discover with reusable client (tests DiscoveryClient + discover)
        let dc = DiscoveryClient::new(&relays).await.expect("dc");
        let meshes = discover(&relays, &MeshFilter::default(), Some(&dc))
            .await
            .expect("discover");

        let found: Vec<_> = meshes
            .iter()
            .filter(|m| m.listing.mesh_id.as_deref() == Some(mesh_id.as_str()))
            .collect();
        assert!(
            found.len() >= 2,
            "should find both publishers for mesh_id={mesh_id}, found {}",
            found.len()
        );

        // Verify fields round-tripped
        let m = &found[0];
        assert_eq!(m.listing.name.as_deref(), Some(mesh_name.as_str()));
        assert_eq!(m.listing.serving, vec!["Qwen3-8B-Q4_K_M"]);
        assert_eq!(m.listing.node_count, 2);
        assert_eq!(m.listing.total_vram_bytes, 16_000_000_000);

        // Both invite tokens present
        let tokens: Vec<_> = found
            .iter()
            .map(|m| m.listing.invite_token.as_str())
            .collect();
        assert!(
            tokens.contains(&token_a.as_str()),
            "missing token_a in {tokens:?}"
        );
        assert!(
            tokens.contains(&token_b.as_str()),
            "missing token_b in {tokens:?}"
        );

        // Second discover with same client still works
        let r2 = discover(&relays, &MeshFilter::default(), Some(&dc))
            .await
            .expect("second discover");
        let found2: Vec<_> = r2
            .iter()
            .filter(|m| m.listing.mesh_id.as_deref() == Some(mesh_id.as_str()))
            .collect();
        assert!(found2.len() >= 2, "reused client should still find both");

        // Cleanup
        pub_a.unpublish().await.ok();
        pub_b.unpublish().await.ok();
    }
}
