//! End-to-end QUIC sync integration test.
//!
//! Mirrors `e2e_sync.rs` but uses `QuicServer` / `QuicConnection` /
//! `quic_connect` instead of the TCP transport.  The same direction-aware
//! push protocol applies; only the framing layer changes.

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

use clawhdf5_onion::format::NO_PARENT;
use clawhdf5_onion::writer::OnionFile;
use clawsync_onion::{differ::diff_revisions, manifest::ClawSyncManifest, merger::merge_packets};
use clawsync_transport::{
    protocol::SyncMessage,
    quic::{QuicConfig, QuicServer, quic_connect},
};
use tempfile::NamedTempFile;

fn localhost_any() -> SocketAddr {
    SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)
}

const H5_BASE: &[u8] = b"\x89HDF\r\n\x1a\n";

// ─────────────────────────────────────────────────────────────────────────────
// Helpers
// ─────────────────────────────────────────────────────────────────────────────

/// Create an `OnionFile` with `n` committed revisions.
fn make_onion(n: u8) -> (NamedTempFile, OnionFile) {
    let tmp = NamedTempFile::new().unwrap();
    let h5_path = tmp.path().with_extension("h5");
    std::fs::write(&h5_path, H5_BASE).unwrap();
    let mut onion = OnionFile::create(&h5_path, 4096).unwrap();
    for i in 0..n {
        let mut s = onion.begin_session(None).unwrap();
        s.record_page(0, &vec![i; 4096]);
        onion.commit_session(s, Some(&format!("rev {i}"))).unwrap();
    }
    (tmp, onion)
}

/// Build a pair of self-signed QuicConfigs.  The server config uses one cert;
/// the client config trusts that cert.  The client's QuicConfig also needs a
/// valid (but unused) server_config, so we generate a separate cert for it.
fn make_quic_configs() -> (QuicConfig, QuicConfig) {
    fn gen_cert() -> (
        rustls::pki_types::CertificateDer<'static>,
        rustls::pki_types::PrivateKeyDer<'static>,
    ) {
        let cert = rcgen::generate_simple_self_signed(vec!["localhost".to_string()]).unwrap();
        let cert_der = rustls::pki_types::CertificateDer::from(cert.cert.der().to_vec());
        let key_der =
            rustls::pki_types::PrivateKeyDer::try_from(cert.key_pair.serialize_der()).unwrap();
        (cert_der, key_der)
    }

    // Server cert: used for TLS handshake; client must trust it.
    let (srv_cert_der, srv_key_der) = gen_cert();
    let srv_server_config =
        quinn::ServerConfig::with_single_cert(vec![srv_cert_der.clone()], srv_key_der).unwrap();

    let mut roots = rustls::RootCertStore::empty();
    roots.add(srv_cert_der).unwrap();
    let client_config = quinn::ClientConfig::with_root_certificates(Arc::new(roots)).unwrap();

    // Client-side QuicConfig requires a server_config field; generate a fresh
    // cert+key pair so there is no key/cert mismatch.
    let (cli_cert_der, cli_key_der) = gen_cert();
    let cli_server_config =
        quinn::ServerConfig::with_single_cert(vec![cli_cert_der], cli_key_der).unwrap();

    let srv_cfg = QuicConfig {
        server_config: srv_server_config,
        client_config: client_config.clone(),
    };
    let cli_cfg = QuicConfig {
        server_config: cli_server_config,
        client_config,
    };
    (srv_cfg, cli_cfg)
}

// ─────────────────────────────────────────────────────────────────────────────
// Core helper: run a full push from `src` → `dst` over QUIC loopback.
// Returns (packets_transferred, final_dst).
// ─────────────────────────────────────────────────────────────────────────────

async fn run_quic_push(src: OnionFile, mut dst: OnionFile) -> (u64, OnionFile) {
    let (srv_cfg, cli_cfg) = make_quic_configs();

    let server = QuicServer::bind(localhost_any(), srv_cfg).await.unwrap();
    let server_addr = server.local_addr;

    let src_manifest = ClawSyncManifest::from_onion("src", &src, H5_BASE);

    // ── Server task ──────────────────────────────────────────────────────────
    let server_handle = tokio::spawn(async move {
        let conn = server.accept().await.unwrap();
        let mut packets_received: u64 = 0;

        loop {
            let msg = conn.recv().await.unwrap();
            match msg {
                SyncMessage::ManifestRequest { .. } => {
                    let dst_manifest = ClawSyncManifest::from_onion("dst", &dst, H5_BASE);
                    conn.send(&SyncMessage::ManifestResponse {
                        manifest: dst_manifest,
                    })
                    .await
                    .unwrap();
                }

                SyncMessage::LayerPacket { packet } => {
                    let rev = packet.revision;
                    merge_packets(&mut dst, vec![packet], false).unwrap();
                    packets_received += 1;
                    conn.send(&SyncMessage::Ack { revision: rev })
                        .await
                        .unwrap();
                }

                SyncMessage::SyncComplete { .. } => {
                    dst.flush().unwrap();
                    conn.close();
                    break;
                }

                other => panic!("server: unexpected {other:?}"),
            }
        }

        (packets_received, dst)
    });

    // ── Client logic ─────────────────────────────────────────────────────────
    let client = quic_connect(server_addr, "localhost", cli_cfg)
        .await
        .unwrap();

    client
        .send(&SyncMessage::ManifestRequest {
            agent_id: "src".to_string(),
            head_revision: src_manifest.head_revision,
            revision_count: src_manifest.revision_count,
        })
        .await
        .unwrap();

    let remote_head = match client.recv().await.unwrap() {
        SyncMessage::ManifestResponse { manifest } => {
            if manifest.revision_count == 0 {
                NO_PARENT
            } else {
                manifest.head_revision
            }
        }
        other => panic!("client: expected ManifestResponse, got {other:?}"),
    };

    let packets = diff_revisions(&src, remote_head).unwrap();
    let total_sent = packets.len() as u64;

    for packet in packets {
        let rev = packet.revision;
        client
            .send(&SyncMessage::LayerPacket { packet })
            .await
            .unwrap();
        match client.recv().await.unwrap() {
            SyncMessage::Ack { revision } => assert_eq!(revision, rev),
            other => panic!("client: expected Ack, got {other:?}"),
        }
    }

    client
        .send(&SyncMessage::SyncComplete {
            revisions_transferred: total_sent,
            bytes_transferred: 0,
        })
        .await
        .unwrap();

    let (server_received, dst_final) = server_handle.await.unwrap();
    assert_eq!(server_received, total_sent);

    (total_sent, dst_final)
}

// ─────────────────────────────────────────────────────────────────────────────
// Tests
// ─────────────────────────────────────────────────────────────────────────────

/// Push 5 revisions to an empty destination over QUIC — all 5 must arrive.
#[tokio::test]
async fn quic_push_all_to_empty() {
    let (_src_tmp, src) = make_onion(5);
    let (_dst_tmp, dst) = make_onion(0);

    let (transferred, dst_final) = run_quic_push(src, dst).await;

    assert_eq!(transferred, 5);
    assert_eq!(dst_final.revision_count(), 5);
}

/// Push when already in sync — zero packets over QUIC.
#[tokio::test]
async fn quic_push_noop_when_in_sync() {
    let (_src_tmp, src) = make_onion(3);
    let (_dst_tmp, dst) = make_onion(3);

    let (transferred, dst_final) = run_quic_push(src, dst).await;

    assert_eq!(transferred, 0);
    assert_eq!(dst_final.revision_count(), 3);
}

/// Partial delta over QUIC: N=8, K=3 → 5 packets.
#[tokio::test]
async fn quic_push_partial_delta() {
    const K: u8 = 3;
    const N: u8 = 8;

    let (_src_tmp, src) = make_onion(N);
    let (_dst_tmp, dst) = make_onion(K);

    let (transferred, dst_final) = run_quic_push(src, dst).await;

    assert_eq!(transferred, (N - K) as u64);
    assert_eq!(dst_final.revision_count(), N as u64);
}

/// BLAKE3 hashes must survive the QUIC wire intact.
#[tokio::test]
async fn quic_revision_hashes_match_after_sync() {
    let (_src_tmp, src) = make_onion(4);
    let (_dst_tmp, dst) = make_onion(0);

    let src_revisions = src.list_revisions();
    let (_, dst_final) = run_quic_push(src, dst).await;
    let dst_revisions = dst_final.list_revisions();

    assert_eq!(src_revisions.len(), dst_revisions.len());
    for (s, d) in src_revisions.iter().zip(dst_revisions.iter()) {
        assert_eq!(
            s.blake3_hex, d.blake3_hex,
            "rev {} hash mismatch",
            s.revision
        );
        assert_eq!(s.annotation, d.annotation);
    }
}

/// Large push over QUIC: 20→7 = 13 packets.
#[tokio::test]
async fn quic_large_push_correct_count() {
    const K: u8 = 7;
    const N: u8 = 20;

    let (_src_tmp, src) = make_onion(N);
    let (_dst_tmp, dst) = make_onion(K);

    let (transferred, dst_final) = run_quic_push(src, dst).await;

    assert_eq!(transferred, (N - K) as u64);
    assert_eq!(dst_final.revision_count(), N as u64);
}

// ─────────────────────────────────────────────────────────────────────────────
// Pull helper: server has `src`; client starts with `dst` and pulls from it.
// Returns (packets_transferred, final_client_onion).
// ─────────────────────────────────────────────────────────────────────────────

async fn run_quic_pull(src: OnionFile, mut dst: OnionFile) -> (u64, OnionFile) {
    let (srv_cfg, cli_cfg) = make_quic_configs();

    let server = QuicServer::bind(localhost_any(), srv_cfg).await.unwrap();
    let server_addr = server.local_addr;

    // ── Server task ──────────────────────────────────────────────────────────
    let server_handle = tokio::spawn(async move {
        let conn = server.accept().await.unwrap();

        let client_rev_count = match conn.recv().await.unwrap() {
            SyncMessage::ManifestRequest { revision_count, .. } => revision_count,
            other => panic!("server: expected ManifestRequest, got {other:?}"),
        };

        let srv_manifest = ClawSyncManifest::from_onion("srv", &src, H5_BASE);
        conn.send(&SyncMessage::ManifestResponse {
            manifest: srv_manifest,
        })
        .await
        .unwrap();

        if src.revision_count() > client_rev_count {
            let remote_head = if client_rev_count == 0 {
                NO_PARENT
            } else {
                client_rev_count - 1
            };
            let packets = diff_revisions(&src, remote_head).unwrap();
            let total = packets.len() as u64;
            let mut bytes_sent = 0u64;
            for packet in packets {
                bytes_sent += packet.page_data_size() as u64;
                conn.send(&SyncMessage::LayerPacket { packet })
                    .await
                    .unwrap();
                match conn.recv().await.unwrap() {
                    SyncMessage::Ack { .. } => {}
                    other => panic!("server: expected Ack, got {other:?}"),
                }
            }
            conn.send(&SyncMessage::SyncComplete {
                revisions_transferred: total,
                bytes_transferred: bytes_sent,
            })
            .await
            .unwrap();
        } else {
            match conn.recv().await.unwrap() {
                SyncMessage::SyncComplete { .. } => {}
                other => panic!("server: expected SyncComplete, got {other:?}"),
            }
        }
        // Quinn resets pending streams when a Connection is dropped.  Give the
        // client a moment to accept the final SyncComplete stream before this
        // task (and thus `conn`) is dropped.  This mirrors the pattern used in
        // the quic_ack_roundtrip unit test.
        tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
    });

    // ── Client logic ─────────────────────────────────────────────────────────
    let client = quic_connect(server_addr, "localhost", cli_cfg)
        .await
        .unwrap();
    let local_rev_count = dst.revision_count();

    client
        .send(&SyncMessage::ManifestRequest {
            agent_id: "client".to_string(),
            head_revision: local_rev_count.saturating_sub(1),
            revision_count: local_rev_count,
        })
        .await
        .unwrap();

    let server_rev_count = match client.recv().await.unwrap() {
        SyncMessage::ManifestResponse { manifest } => manifest.revision_count,
        other => panic!("client: expected ManifestResponse, got {other:?}"),
    };

    if server_rev_count <= local_rev_count {
        client
            .send(&SyncMessage::SyncComplete {
                revisions_transferred: 0,
                bytes_transferred: 0,
            })
            .await
            .unwrap();
        server_handle.await.unwrap();
        return (0, dst);
    }

    let mut packets = Vec::new();
    loop {
        match client.recv().await.unwrap() {
            SyncMessage::LayerPacket { packet } => {
                let rev = packet.revision;
                client
                    .send(&SyncMessage::Ack { revision: rev })
                    .await
                    .unwrap();
                packets.push(packet);
            }
            SyncMessage::SyncComplete {
                revisions_transferred,
                ..
            } => {
                merge_packets(&mut dst, packets, true).unwrap();
                server_handle.await.unwrap();
                return (revisions_transferred, dst);
            }
            other => panic!("client: unexpected {other:?}"),
        }
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// Pull tests
// ─────────────────────────────────────────────────────────────────────────────

/// Pull all revisions from a full QUIC server to an empty client.
#[tokio::test]
async fn quic_pull_all_from_server() {
    let (_srv_tmp, srv) = make_onion(5);
    let (_cli_tmp, cli) = make_onion(0);

    let (transferred, cli_final) = run_quic_pull(srv, cli).await;

    assert_eq!(transferred, 5);
    assert_eq!(cli_final.revision_count(), 5);
}

/// Pull when already in sync over QUIC — zero packets.
#[tokio::test]
async fn quic_pull_noop_when_already_in_sync() {
    let (_srv_tmp, srv) = make_onion(4);
    let (_cli_tmp, cli) = make_onion(4);

    let (transferred, cli_final) = run_quic_pull(srv, cli).await;

    assert_eq!(transferred, 0);
    assert_eq!(cli_final.revision_count(), 4);
}

/// Partial pull over QUIC: server has N, client has K → N-K packets.
#[tokio::test]
async fn quic_pull_partial_delta() {
    const K: u8 = 2;
    const N: u8 = 7;

    let (_srv_tmp, srv) = make_onion(N);
    let (_cli_tmp, cli) = make_onion(K);

    let (transferred, cli_final) = run_quic_pull(srv, cli).await;

    assert_eq!(transferred, (N - K) as u64);
    assert_eq!(cli_final.revision_count(), N as u64);
}

/// BLAKE3 hashes must survive the QUIC pull wire path intact.
#[tokio::test]
async fn quic_pull_revision_hashes_match() {
    let (_srv_tmp, srv) = make_onion(4);
    let (_cli_tmp, cli) = make_onion(0);

    let srv_revisions = srv.list_revisions();
    let (_, cli_final) = run_quic_pull(srv, cli).await;
    let cli_revisions = cli_final.list_revisions();

    assert_eq!(srv_revisions.len(), cli_revisions.len());
    for (s, c) in srv_revisions.iter().zip(cli_revisions.iter()) {
        assert_eq!(
            s.blake3_hex, c.blake3_hex,
            "rev {} hash mismatch",
            s.revision
        );
        assert_eq!(s.annotation, c.annotation);
    }
}