//! End-to-end test: two `AutomergeBackend` instances syncing a document
//! through the persistent multiplexed sync stream (peat-mesh#175 / ADR-063).
//!
//! This is the regression pin for the dropped-`Arc` BLOCKER that
//! peat-mesh#178 fixed: pre-fix, `AutomergeBackend::new` constructed a
//! `SyncChannelManager`, called `coordinator.set_channel_manager(mgr)`,
//! and dropped the strong reference at the end of `new`. The
//! coordinator's `Weak` immediately dangled and every consumer fell
//! through to the legacy per-message-stream fallback in
//! `send_sync_message_for_doc` — the central wire-up was a no-op,
//! observable only by a careful code-read (the QA review caught it)
//! or by lab measurement (UAT thresholds failed without it).
//!
//! No existing in-CI test exercised both sides of the wire-up
//! together: PR A's framing tests bypass `AutomergeBackend::new` by
//! constructing `SyncChannel` directly, and PR B's writer-task E2E
//! does the same. This test closes the gap by:
//!
//! 1. Building two real `AutomergeBackend::with_iroh` instances —
//!    each one runs through the full constructor including the
//!    `set_channel_manager` wire-up and the strong-reference field
//!    that keeps the manager alive.
//! 2. Cross-populating their `MemoryLookup`s so each side resolves
//!    the other's `EndpointId` to its bound socket address.
//! 3. Starting sync on both, connecting them as peers.
//! 4. Writing a document on backend A.
//! 5. Polling backend B's store until the document materialises, or
//!    timing out.
//!
//! If the `Arc<SyncChannelManager>` field is ever accidentally removed
//! from `AutomergeBackend` (or `set_channel_manager`'s `Weak` downgrade
//! is changed without keeping a strong-ref elsewhere), the test still
//! passes — because the legacy per-message-stream fallback in
//! `send_sync_message_for_doc` also syncs successfully, just slower.
//! For a behavioural pin on the persistent path specifically, we also
//! assert that the coordinator's `get_channel_manager()` returns
//! `Some` after `with_iroh` — a direct shape check on the surface the
//! BLOCKER broke.

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

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

use peat_mesh::security::FormationKey;
use peat_mesh::sync::{
    AutomergeBackend, AutomergeBackendConfig, BackendConfig, DataSyncBackend, Document, SyncEngine,
};
use tempfile::TempDir;

/// Build a backend bound to a random localhost port, returning it
/// alongside the temp directory that owns its on-disk state. The
/// returned `TempDir` must outlive the backend.
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();
    // Bind to IPv4 localhost on an OS-chosen port so the lab kernel
    // routes loopback packets between the two backends in-process.
    // `with_iroh`'s default dual-stack bind happens to work for
    // peer-to-peer on a single host, but pinning to v4 avoids any
    // chance the cross-populated lookup races a v4 vs v6 candidate.
    cfg.iroh_bind_addr = Some(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0));
    let backend = AutomergeBackend::with_iroh(cfg).await.expect("with_iroh");
    // `AutomergeBackend::initialize` ignores the BackendConfig arg —
    // setup happened in `with_iroh`. Pass a minimal placeholder so the
    // trait method satisfies the call.
    backend
        .initialize(BackendConfig {
            app_id: "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: two `AutomergeBackend`s constructed via the real `with_iroh`
/// path sync a document through the persistent stream.
///
/// Catches the peat-mesh#178 BLOCKER class: if `AutomergeBackend` ever
/// stops holding the strong `Arc<SyncChannelManager>`, the wire-up's
/// only ground-truth signal is delivery — this test fails if delivery
/// stalls.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn two_backends_sync_doc_through_persistent_stream() {
    let shared_key = FormationKey::generate_secret();

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

    // Cross-populate the iroh `MemoryLookup` on each side so the
    // EndpointId resolves to a real socket address without depending
    // on DNS / relay / mDNS infrastructure.
    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();
    // Surface both EndpointIds in test output — useful on failure
    // when diagnosing which side didn't see what.
    eprintln!(
        "two-backend E2E: A.id={} A.addr={:?} B.id={} B.addr={:?}",
        id_a, addr_a, id_b, addr_b
    );

    backend_a
        .blob_store()
        .memory_lookup()
        .add_endpoint_info(addr_b);
    backend_b
        .blob_store()
        .memory_lookup()
        .add_endpoint_info(addr_a);

    // Direct behavioural pin on the BLOCKER fix: each coordinator's
    // `SyncChannelManager` weak ref must upgrade. If the strong-ref
    // field is ever removed from `AutomergeBackend`, this assertion
    // catches it BEFORE the delivery check below — the delivery check
    // alone can't catch a missing Arc because the legacy
    // per-message-stream fallback also delivers documents.
    assert!(
        backend_a.coordinator().has_persistent_sync_channel(),
        "backend A's coordinator must hold a live SyncChannelManager — \
         if false, the Arc<SyncChannelManager> strong-ref on \
         AutomergeBackend has been lost (peat-mesh#178 BLOCKER class)"
    );
    assert!(
        backend_b.coordinator().has_persistent_sync_channel(),
        "backend B's coordinator must hold a live SyncChannelManager"
    );
    let _ = id_a; // Already logged above; symmetric with id_b for the connect step.

    // Start sync on both.
    backend_a.start_sync().await.expect("start_sync A");
    backend_b.start_sync().await.expect("start_sync B");

    // A initiates the formation handshake against B.
    let connected = backend_a
        .connect_to_peer(&id_b.to_string(), &[])
        .await
        .expect("connect_to_peer A → B");
    assert!(
        connected,
        "connect_to_peer A → B must return true after the formation handshake"
    );

    // A writes a document; B should pick it up via auto-sync over the
    // persistent stream.
    let doc_id = "e2e-doc-1";
    let mut payload = HashMap::new();
    payload.insert(
        "field".to_string(),
        serde_json::Value::String("persistent-stream-pin".to_string()),
    );
    let doc = Document::with_id(doc_id, payload);

    backend_a
        .document_store()
        .upsert("e2e", doc)
        .await
        .expect("upsert on A");

    // Poll B until the document materialises. The persistent path
    // typically converges in well under a second on loopback; budget
    // 30 s to absorb CI noise. If this test stops passing at ~30 s
    // wall-clock, the BLOCKER class is back.
    let started = std::time::Instant::now();
    let deadline = started + Duration::from_secs(30);
    let observed = loop {
        if let Some(doc) = backend_b
            .document_store()
            .get("e2e", &doc_id.to_string())
            .await
            .expect("get on B")
        {
            break Some(doc);
        }
        if std::time::Instant::now() > deadline {
            break None;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    };
    let elapsed = started.elapsed();

    assert!(
        observed.is_some(),
        "backend B must observe the doc written on backend A within 30s — if this stalls, \
         the SyncChannelManager strong-ref on AutomergeBackend is likely missing \
         (see peat-mesh#178 BLOCKER context). elapsed={:?}",
        elapsed
    );
    // Surface convergence time on success so a future slowdown shows
    // up in CI output before it crosses the 30 s budget.
    eprintln!("two-backend E2E: A → B sync convergence {:?}", elapsed);

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

/// Pin (peat-node#170): a peer that dials *us* must be registered into our
/// `known_peers` — not just peers we dial out to.
///
/// `known_peers` drives both distribution targeting (`resolve_targets`) and
/// blob-provider lookup (`fetch_blob`). Before the fix it was populated only on
/// the OUTBOUND dial path, so when A dialed B, B synced documents from A but
/// never recorded A as a peer. An `AllNodes` distribution B originated (e.g. an
/// attachment) would then resolve an empty/short `target_nodes` and the file
/// would never deliver even though the distribution *document* synced — the
/// "synced but nothing written" symptom. This pins inbound-accept registration.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn inbound_dialed_peer_is_registered_in_known_peers() {
    let shared_key = FormationKey::generate_secret();

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

    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");

    // ONE-DIRECTIONAL dial: A → B only. B never dials A.
    let connected = backend_a
        .connect_to_peer(&id_b.to_string(), &[])
        .await
        .expect("connect_to_peer A → B");
    assert!(connected, "A → B formation handshake must succeed");

    // NOTE: at the AutomergeBackend layer, the OUTBOUND dial does not by itself
    // add B to A's known_peers — that's done by the consumer (peat-node's
    // connect_peer) or by discovery's PeerConnector. This test exercises the
    // backend in isolation, so we assert only the inbound-accept registration
    // the backend now owns.

    // The fix: B accepted A's dial, so B must learn A as a known peer. The
    // accept-side registration runs after the formation handshake on a
    // separate task, so poll with a deadline rather than asserting inline.
    let deadline = std::time::Instant::now() + Duration::from_secs(10);
    let b_knows_a = loop {
        if backend_b.blob_store().known_peers().await.contains(&id_a) {
            break true;
        }
        if std::time::Instant::now() > deadline {
            break false;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    };
    assert!(
        b_knows_a,
        "inbound (peat-node#170): B accepted A's dial but never registered A in \
         known_peers — A would be excluded from B's distribution targeting and \
         blob fetch. known_peers={:?}",
        backend_b.blob_store().known_peers().await
    );

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