//! Provider-gossip end-to-end pin (peat-node#170, slice 1).
//!
//! Proves the `peat/blob-announce/1` ALPN carries "who holds blob X" between
//! real iroh endpoints and lands it in the receiver's provider index — the
//! discovery half of multi-hop blob delivery. A node announces a holding to a
//! peer; the peer's `fetch_blob` provider lookup (`peers_with_blob`) must then
//! return the announced holder, even though no blob transfer occurred.
//!
//! Runs real QUIC over loopback, so (like the other two-backend e2e tests) it
//! is gated to `automerge-backend` and exercised on the self-hosted CI runner;
//! it is contention-flaky on shared hosted runners.

#![cfg(feature = "automerge-backend")]

use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::Arc;
use std::time::Duration;

use peat_mesh::security::FormationKey;
use peat_mesh::storage::blob_announce::{self, BlobAnnounce, DEFAULT_ANNOUNCE_TTL};
use peat_mesh::storage::blob_traits::{BlobHash, BlobMetadata, BlobStore};
use peat_mesh::storage::BlobStoreExt;
use peat_mesh::sync::{
    AutomergeBackend, AutomergeBackendConfig, BackendConfig, DataSyncBackend, SyncEngine,
};
use tempfile::TempDir;

async fn build_backend(formation_id: &str, shared_key: &str) -> (Arc<AutomergeBackend>, TempDir) {
    let dir = TempDir::new().expect("tempdir");
    let mut cfg = AutomergeBackendConfig::default();
    cfg.data_dir = dir.path().to_path_buf();
    cfg.formation_id = formation_id.to_string();
    cfg.base64_shared_key = shared_key.to_string();
    cfg.iroh_bind_addr = Some(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0));
    let backend = AutomergeBackend::with_iroh(cfg).await.expect("with_iroh");
    backend
        .initialize(BackendConfig {
            app_id: "gossip-e2e".to_string(),
            persistence_dir: dir.path().to_path_buf(),
            shared_key: None,
            transport: peat_mesh::sync::TransportConfig::default(),
            extra: std::collections::HashMap::new(),
        })
        .await
        .expect("initialize");
    (backend, dir)
}

/// Pin: an announce that A holds blob X reaches B's provider index over the
/// dedicated ALPN.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn provider_announce_reaches_peer_index() {
    let shared_key = FormationKey::generate_secret();

    let (backend_a, _dir_a) = build_backend("gossip-formation", &shared_key).await;
    let (backend_b, _dir_b) = build_backend("gossip-formation", &shared_key).await;

    // Cross-populate iroh address lookup so the announce connection resolves
    // without DNS / relay / mDNS.
    let addr_a = backend_a.blob_store().endpoint().addr();
    let addr_b = backend_b.blob_store().endpoint().addr();
    let id_a = backend_a.blob_store().endpoint().id();
    let id_b = backend_b.blob_store().endpoint().id();
    backend_a
        .blob_store()
        .memory_lookup()
        .add_endpoint_info(addr_b);
    backend_b
        .blob_store()
        .memory_lookup()
        .add_endpoint_info(addr_a);

    backend_a.start_sync().await.expect("start_sync A");
    backend_b.start_sync().await.expect("start_sync B");

    // A announces that it holds a blob (no transfer — discovery only). A
    // well-formed (64 hex char / BLAKE3-shaped) hash; apply_announce drops
    // malformed ones.
    let hash = BlobHash::from_hex(&"a1b2c3d4".repeat(8));
    let announce = BlobAnnounce {
        holder: id_a,
        addrs: backend_a
            .blob_store()
            .bound_addr_string()
            .into_iter()
            .collect(),
        hashes: vec![hash.clone()],
        ttl: DEFAULT_ANNOUNCE_TTL,
    };
    blob_announce::send_announce(backend_a.blob_store().endpoint(), id_b, &announce)
        .await
        .expect("A → B announce send");

    // B's provider index must learn that A holds the blob. The handler applies
    // on a task after the ack, so poll with a deadline.
    let deadline = std::time::Instant::now() + Duration::from_secs(10);
    let learned = loop {
        if backend_b
            .blob_store()
            .peers_with_blob(&hash)
            .await
            .contains(&id_a)
        {
            break true;
        }
        if std::time::Instant::now() > deadline {
            break false;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    };
    assert!(
        learned,
        "B must learn (via blob-announce gossip) that A holds the blob — \
         peers_with_blob={:?}, expected to contain A={}",
        backend_b.blob_store().peers_with_blob(&hash).await,
        id_a.fmt_short()
    );

    backend_a.shutdown().await.expect("shutdown A");
    backend_b.shutdown().await.expect("shutdown B");
}

/// Pin (QA #3): the multi-hop re-broadcast path. A line A–B–C where A and C are
/// NOT directly connected. A announces a holding to B; C — never contacted by
/// A — must learn it via B's re-broadcast. Also pins the `HoldingsOnly` trust
/// path (C receives a relayed announce whose holder ≠ sender) and termination
/// (C has no onward peer to relay to).
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn relay_announce_reaches_non_adjacent_peer() {
    let shared_key = FormationKey::generate_secret();

    let (backend_a, _dir_a) = build_backend("relay-formation", &shared_key).await;
    let (backend_b, _dir_b) = build_backend("relay-formation", &shared_key).await;
    let (backend_c, _dir_c) = build_backend("relay-formation", &shared_key).await;

    let addr_a = backend_a.blob_store().endpoint().addr();
    let addr_b = backend_b.blob_store().endpoint().addr();
    let addr_c = backend_c.blob_store().endpoint().addr();
    let id_a = backend_a.blob_store().endpoint().id();
    let id_c = backend_c.blob_store().endpoint().id();

    // Connectivity: A↔B and B↔C resolvable; A and C never learn each other.
    backend_a
        .blob_store()
        .memory_lookup()
        .add_endpoint_info(addr_b.clone());
    backend_b
        .blob_store()
        .memory_lookup()
        .add_endpoint_info(addr_a);
    backend_b
        .blob_store()
        .memory_lookup()
        .add_endpoint_info(addr_c);
    backend_c
        .blob_store()
        .memory_lookup()
        .add_endpoint_info(addr_b);

    // B must have C as a known peer to re-broadcast to it (re-broadcast targets
    // known_peers). At the backend layer this is set explicitly.
    backend_b.blob_store().add_peer(id_c).await;

    for b in [&backend_a, &backend_b, &backend_c] {
        b.start_sync().await.expect("start_sync");
    }

    // A originates a first-party announce (holder == A) to B only.
    let hash = BlobHash::from_hex(&"0badc0de".repeat(8));
    let announce = BlobAnnounce {
        holder: id_a,
        addrs: backend_a
            .blob_store()
            .bound_addr_string()
            .into_iter()
            .collect(),
        hashes: vec![hash.clone()],
        ttl: DEFAULT_ANNOUNCE_TTL,
    };
    let id_b = backend_b.blob_store().endpoint().id();
    blob_announce::send_announce(backend_a.blob_store().endpoint(), id_b, &announce)
        .await
        .expect("A → B announce");

    // C must learn (via B's re-broadcast) that A holds the blob, despite never
    // being contacted by A directly.
    let deadline = std::time::Instant::now() + Duration::from_secs(15);
    let learned = loop {
        if backend_c
            .blob_store()
            .peers_with_blob(&hash)
            .await
            .contains(&id_a)
        {
            break true;
        }
        if std::time::Instant::now() > deadline {
            break false;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    };
    assert!(
        learned,
        "C must learn A holds the blob via B's re-broadcast (multi-hop). \
         C.peers_with_blob={:?}, expected A={}",
        backend_c.blob_store().peers_with_blob(&hash).await,
        id_a.fmt_short()
    );

    backend_a.shutdown().await.expect("shutdown A");
    backend_b.shutdown().await.expect("shutdown B");
    backend_c.shutdown().await.expect("shutdown C");
}

/// Pin (QA #2): the fetch-on-acquire integration. B fetches a real blob from A;
/// B's `fetch_blob` success path must fire `announce_local_holding`, so B's
/// other peer C learns that B now holds the blob — the mechanism that carries a
/// blob outward hop by hop. Exercises real blob transfer, not a synthetic
/// announce.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn fetch_then_announce_reaches_peer() {
    let shared_key = FormationKey::generate_secret();

    let (backend_a, _dir_a) = build_backend("fetch-formation", &shared_key).await;
    let (backend_b, _dir_b) = build_backend("fetch-formation", &shared_key).await;
    let (backend_c, _dir_c) = build_backend("fetch-formation", &shared_key).await;

    let addr_a = backend_a.blob_store().endpoint().addr();
    let addr_b = backend_b.blob_store().endpoint().addr();
    let addr_c = backend_c.blob_store().endpoint().addr();
    let id_a = backend_a.blob_store().endpoint().id();
    let id_b = backend_b.blob_store().endpoint().id();
    let id_c = backend_c.blob_store().endpoint().id();

    // A↔B (B fetches from A and announces back), B↔C (B announces onward to C).
    backend_a
        .blob_store()
        .memory_lookup()
        .add_endpoint_info(addr_b.clone());
    backend_b
        .blob_store()
        .memory_lookup()
        .add_endpoint_info(addr_a);
    backend_b
        .blob_store()
        .memory_lookup()
        .add_endpoint_info(addr_c);
    backend_c
        .blob_store()
        .memory_lookup()
        .add_endpoint_info(addr_b);

    // B must know A (to fetch from it) and C (to announce onward).
    backend_b.blob_store().add_peer(id_a).await;
    backend_b.blob_store().add_peer(id_c).await;

    for b in [&backend_a, &backend_b, &backend_c] {
        b.start_sync().await.expect("start_sync");
    }

    // A creates a real blob.
    let token = backend_a
        .blob_store()
        .create_blob_from_bytes(b"multi-hop attachment payload", BlobMetadata::default())
        .await
        .expect("A create_blob_from_bytes");

    // B fetches it from A (real transfer). Success must trigger the announce.
    backend_b
        .blob_store()
        .fetch_blob_simple(&token)
        .await
        .expect("B fetch_blob from A");

    // C must learn that B holds the blob — i.e. fetch_blob fired
    // announce_local_holding and it reached C.
    let hash = token.hash.clone();
    let deadline = std::time::Instant::now() + Duration::from_secs(20);
    let learned = loop {
        if backend_c
            .blob_store()
            .peers_with_blob(&hash)
            .await
            .contains(&id_b)
        {
            break true;
        }
        if std::time::Instant::now() > deadline {
            break false;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    };
    assert!(
        learned,
        "C must learn B holds the blob after B fetched it (fetch → \
         announce_local_holding → C). C.peers_with_blob={:?}, expected B={}",
        backend_c.blob_store().peers_with_blob(&hash).await,
        id_b.fmt_short()
    );
    let _ = id_c;

    backend_a.shutdown().await.expect("shutdown A");
    backend_b.shutdown().await.expect("shutdown B");
    backend_c.shutdown().await.expect("shutdown C");
}
