//! Integration test for the `Sync::stop` ↔ NORM multicast contract.
//!
//! Two same-host Ditto SDK instances run a NORM multicast transport on a
//! unique loopback port. The test exercises four phases of the contract:
//!
//! 1. **Sanity** — while both peers have sync active, documents and attachments inserted on peer A
//!    arrive on peer B over NORM multicast.
//! 2. **Partition** — after `peer_b.sync().stop()`, documents and attachments inserted on peer A do
//!    **not** arrive on peer B within a generous window.
//! 3. **Resume (peer A → peer B)** — after `peer_b.sync().start()`, the items inserted during the
//!    partition window arrive on peer B.
//! 4. **Resume (peer B → peer A)** — peer B inserts a *new* doc and a *new* (unique-byte)
//!    attachment after restart; both must reach peer A — including the attachment bytes, verified
//!    via `fetch_attachment`. Exercises the attachment-bridge resubscribe path.
//!
//! Gated on `__ditto_multicast_beta` because NORM
//! multicast is feature-gated through the stack:
//! `dittolive-ditto` → `dittolive-ditto-sys` → `dittoffi` → `ditto-peer`
//! → `ditto-sync-docs`. Marked `#[ignore]` because the test requires
//! multicast on the loopback interface, which is not universally
//! available in sandboxed CI environments. Opt-in:
//!
//! ```text
//! cargo test --features __ditto_multicast_beta \
//!     --test norm_multicast_sync_stop -- --ignored
//! ```

#![cfg(feature = "__ditto_multicast_beta")]

use std::{
    collections::HashMap,
    sync::{
        atomic::{AtomicU16, AtomicUsize, Ordering},
        Arc,
    },
    time::Duration,
};

use anyhow::Error;
use dittolive_ditto::{
    error::DittoError, fs::TempRoot, prelude::*, store::attachment::DittoAttachmentFetchEvent,
    transport::TransportConfig, DatabaseId,
};
use serde_json::json;
use tokio::sync::oneshot;

mod common;

fn loopback_interface_name() -> &'static str {
    if cfg!(target_os = "macos") {
        "lo0"
    } else {
        "lo"
    }
}

fn unique_multicast_port() -> u16 {
    // Above NORM's default (6003) and other commonly-used multicast
    // ports to avoid collisions on the developer's loopback.
    static PORT: AtomicU16 = AtomicU16::new(27_000);
    PORT.fetch_add(1, Ordering::Relaxed)
}

/// Build the multicast-only transport config shared by both peers: every other
/// transport is disabled so NORM multicast is the *only* path between the two
/// peers, pinned to a per-test loopback group/port. `enabled` toggles the NORM
/// multicast transport itself — passing `false` produces a config with no
/// enabled transport at all, which fully partitions the peer.
fn norm_multicast_tc(port: u16, enabled: bool) -> TransportConfig {
    let mut tc = TransportConfig::new();
    // Everything off except NORM multicast so the only path between the two
    // peers is the one we're testing.
    tc.peer_to_peer.bluetooth_le.enabled = false;
    tc.peer_to_peer.lan.enabled = false;
    tc.peer_to_peer.awdl.enabled = false;
    tc.peer_to_peer.wifi_aware.enabled = false;
    tc.listen.tcp.enabled = false;
    tc.listen.http.enabled = false;

    tc.peer_to_peer.multicast.enabled = enabled;
    // Pin to a unique-per-test loopback group/port pair. Bind explicitly to
    // the loopback interface so the multicast traffic never leaves the host.
    tc.peer_to_peer.multicast.group_address = "239.255.42.42".to_string();
    tc.peer_to_peer.multicast.port = port;
    tc.peer_to_peer.multicast.interface = Some(loopback_interface_name().to_string());
    tc
}

async fn open_norm_peer(app_id: DatabaseId, port: u16) -> Result<(TempRoot, Ditto), DittoError> {
    let root = TempRoot::new();
    let connect = DittoConfigConnect::SmallPeersOnly { private_key: None };
    let mut config =
        DittoConfig::new(app_id.to_string(), connect).with_persistence_directory(root.root_path());
    // Required for two same-host peers (especially two in one process) to see
    // each other's multicast traffic. Defaults to false because production
    // peers on different hosts don't need it.
    config
        .system_parameters
        .insert("multicast_loopback".into(), serde_json::Value::Bool(true));
    let ditto = Ditto::open(config).await?;
    ditto.set_license_from_env("DITTO_LICENSE")?;

    ditto.set_transport_config(norm_multicast_tc(port, true));
    Ok((root, ditto))
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "requires NORM multicast on loopback; opt-in via --ignored"]
async fn sync_stop_partitions_norm_multicast_for_docs_and_attachments() -> Result<(), Error> {
    let app_id = DatabaseId::generate();
    let port = unique_multicast_port();

    // TempRoot must outlive Ditto (LIFO drop order), so destructure with
    // _root first per the SDK testing convention.
    let (_root_a, peer_a) = open_norm_peer(app_id.clone(), port).await?;
    let (_root_b, peer_b) = open_norm_peer(app_id, port).await?;

    peer_a.sync().start()?;
    peer_b.sync().start()?;

    // Subscribe both peers to both collections so multicast relays them in
    // both directions. Phases 1-3 exercise peer_a → peer_b; phase 4
    // exercises peer_b → peer_a.
    let _doc_subscription_a = peer_a
        .sync()
        .register_subscription("SELECT * FROM test_docs")?;
    let _att_subscription_a = peer_a
        .sync()
        .register_subscription("SELECT * FROM COLLECTION test_attachments (att ATTACHMENT)")?;
    let _doc_subscription_b = peer_b
        .sync()
        .register_subscription("SELECT * FROM test_docs")?;
    let _att_subscription_b = peer_b
        .sync()
        .register_subscription("SELECT * FROM COLLECTION test_attachments (att ATTACHMENT)")?;

    // Observers update an atomic cell with each new item-count snapshot.
    // Cumulative count (not delta), so reading the cell at any later time
    // gives the latest-observed value — perfect for asserting both
    // "eventually reached N" and "stayed at N for the partition window".
    let (doc_count_a, _doc_observer_a) =
        register_count_observer(&peer_a, "SELECT * FROM test_docs")?;
    let (att_count_a, _att_observer_a) = register_count_observer(
        &peer_a,
        "SELECT * FROM COLLECTION test_attachments (att ATTACHMENT)",
    )?;
    let (doc_count_b, _doc_observer_b) =
        register_count_observer(&peer_b, "SELECT * FROM test_docs")?;
    let (att_count_b, _att_observer_b) = register_count_observer(
        &peer_b,
        "SELECT * FROM COLLECTION test_attachments (att ATTACHMENT)",
    )?;

    // -------- Phase 1: sync active on both -> arrivals are expected. --------
    insert_doc(&peer_a, 1).await?;
    insert_attachment_doc(&peer_a, 1).await?;

    assert!(
        wait_until(&doc_count_b, |n| n >= 1, Duration::from_secs(20)).await,
        "Phase 1: peer_b should receive the doc inserted on peer_a while sync is active",
    );
    assert!(
        wait_until(&att_count_b, |n| n >= 1, Duration::from_secs(20)).await,
        "Phase 1: peer_b should receive the attachment doc inserted on peer_a while sync is active",
    );

    // -------- Phase 2: stop sync on peer_b -> nothing must arrive ----------
    // (and peer_b must not originate, either — bridge watcher must
    // have been torn down with the session).
    peer_b.sync().stop();
    assert!(
        !peer_b.sync().is_active(),
        "peer_b sync should report inactive after stop()",
    );

    // 2a. peer_a inserts a doc and an attachment while peer_b is
    // partitioned. peer_b's receive path must be torn down (no
    // arrivals at peer_b).
    insert_doc(&peer_a, 2).await?;
    insert_attachment_doc(&peer_a, 2).await?;

    // Long enough that, if NORM weren't actually torn down, peer_a's
    // inserts would have arrived at peer_b. peer_b's observers must
    // still report 1 — the Phase-1 doc / attachment received pre-stop.
    tokio::time::sleep(Duration::from_secs(5)).await;
    assert_eq!(
        doc_count_b.load(Ordering::SeqCst),
        1,
        "Phase 2a: peer_b must not receive the doc inserted while its sync was stopped",
    );
    assert_eq!(
        att_count_b.load(Ordering::SeqCst),
        1,
        "Phase 2a: peer_b must not receive the attachment doc inserted while its sync was stopped",
    );

    // 2b. peer_b originates a phase-22 attachment (unique id) while
    // stopped. If the bridge watcher leaked past `Sync::stop`, peer_a
    // would see the new attachment-doc row. Baseline captured after
    // the 2a sleep so peer_a's observer has caught its self-insert.
    let att_a_baseline_before_b_insert = att_count_a.load(Ordering::SeqCst);
    insert_attachment_doc(&peer_b, 22).await?;
    tokio::time::sleep(Duration::from_secs(5)).await;
    assert_eq!(
        att_count_a.load(Ordering::SeqCst),
        att_a_baseline_before_b_insert,
        "Phase 2b: peer_a must not receive peer_b's phase-22 attachment — peer_b's bridge watcher \
         must be torn down so locally-completed attachments do not broadcast while sync is stopped",
    );

    // -------- Phase 3: restart sync on peer_b -> arrivals resume (a→b). ------
    peer_b.sync().start()?;
    assert!(
        peer_b.sync().is_active(),
        "peer_b sync should report active after start()",
    );

    assert!(
        wait_until(&doc_count_b, |n| n >= 2, Duration::from_secs(20)).await,
        "Phase 3: peer_b should receive the Phase-2 doc after sync.start() resumes the transport",
    );
    // Going into Phase 3, peer_b's att_count is 2 (Phase-1 from peer_a
    // pre-stop, plus the local Phase-22 from 2b). Resume must bring in
    // peer_a's Phase-2 attachment, taking the count to 3.
    assert!(
        wait_until(&att_count_b, |n| n >= 3, Duration::from_secs(20)).await,
        "Phase 3: peer_b should receive the Phase-2 attachment doc after sync.start() resumes the \
         transport",
    );

    // -------- Phase 4: peer_b originates after restart -> peer_a receives. --
    //
    // After a stop/start cycle peer_b must be able to originate new
    // multicast attachments — exercising the bridge built after the
    // restart. Assertions are relative to peer_a's pre-Phase-4
    // baseline.
    let doc_baseline_a = doc_count_a.load(Ordering::SeqCst);
    let att_baseline_a = att_count_a.load(Ordering::SeqCst);

    insert_doc(&peer_b, 4).await?;
    insert_attachment_doc(&peer_b, 4).await?;

    assert!(
        wait_until(
            &doc_count_a,
            |n| n > doc_baseline_a,
            Duration::from_secs(20),
        )
        .await,
        "Phase 4: peer_a should receive the doc peer_b inserted after its sync restarted",
    );
    assert!(
        wait_until(
            &att_count_a,
            |n| n > att_baseline_a,
            Duration::from_secs(20),
        )
        .await,
        "Phase 4: peer_a should see the attachment doc-row peer_b inserted after its sync \
         restarted (diffcaster doc-replication path)",
    );

    // Doc-rows arrive via diffcaster; the attachment bytes themselves
    // must arrive via the multicast attachment bridge. Without the
    // bridge's resubscribe path the MATA frame never goes on the wire
    // after a `Sync::stop` / `Sync::start` cycle, peer_a's blob storage
    // never receives the bytes, and `fetch_attachment` never reports
    // `Completed`.
    assert!(
        peer_a_received_attachment_bytes_for_phase(&peer_a, 4, Duration::from_secs(15)).await?,
        "Phase 4: peer_a's `fetch_attachment` for peer_b's restart-originated attachment must \
         complete (attachment bridge resubscribe path)",
    );

    Ok(())
}

/// Look up the phase-tagged attachment doc on `peer`, then drive
/// `fetch_attachment` for it and return whether `Completed` arrived within
/// `timeout`. If the bytes never made it into `peer`'s blob storage, the
/// fetch hangs and we time out → `false`.
async fn peer_a_received_attachment_bytes_for_phase(
    peer: &Ditto,
    phase: u32,
    timeout: Duration,
) -> Result<bool, Error> {
    // Filter by phase so the assertion is focused on a single
    // attachment doc-row even when the peer has locally authored
    // others.
    let result = peer
        .store()
        .execute((
            "SELECT * FROM COLLECTION test_attachments (att ATTACHMENT) WHERE phase = :phase",
            json!({ "phase": phase }),
        ))
        .await?;
    if result.item_count() == 0 {
        return Ok(false);
    }
    let value = result.get_item(0).unwrap().value();
    let token = value
        .get("att")
        .and_then(|v| v.as_object())
        .ok_or_else(|| anyhow::anyhow!("attachment token missing on phase-{phase} doc"))?;

    let (tx, rx) = oneshot::channel::<()>();
    let tx = std::sync::Mutex::new(Some(tx));
    let fetcher = peer.store().fetch_attachment(token, move |event| {
        if matches!(event, DittoAttachmentFetchEvent::Completed { .. }) {
            if let Some(tx) = tx.lock().unwrap().take() {
                let _ = tx.send(());
            }
        }
    })?;

    // `tokio::time::timeout(..).await` returns
    // `Result<Result<(), oneshot::RecvError>, Elapsed>`. Both layers
    // must be `Ok`: the outer to confirm we didn't time out, the inner
    // to confirm the oneshot actually received a value (the fetcher
    // closure dropping without firing `Completed` resolves the inner
    // future to `Err(RecvError)` — failure, not success).
    let completed = matches!(tokio::time::timeout(timeout, rx).await, Ok(Ok(())));
    drop(fetcher);
    Ok(completed)
}

fn register_count_observer(
    peer: &Ditto,
    query: &str,
) -> Result<(Arc<AtomicUsize>, Arc<dittolive_ditto::store::StoreObserver>), DittoError> {
    let count = Arc::new(AtomicUsize::new(0));
    let count_inner = Arc::clone(&count);
    let observer = peer.store().register_observer(query, move |result| {
        count_inner.store(result.item_count(), Ordering::SeqCst);
    })?;
    Ok((count, observer))
}

async fn insert_doc(peer: &Ditto, phase: u32) -> Result<(), Error> {
    peer.store()
        .execute((
            "INSERT INTO test_docs DOCUMENTS (:doc)",
            json!({ "doc": { "phase": phase } }),
        ))
        .await?;
    Ok(())
}

async fn insert_attachment_doc(peer: &Ditto, phase: u32) -> Result<(), Error> {
    // Phase-specific bytes so each phase produces a unique
    // content-addressed attachment id. Required for Phase 4 to actually
    // exercise the bridge: identical bytes would be deduped by the
    // bridge's `broadcast_ids` set and never re-broadcast.
    let bytes: Vec<u8> = format!("phase-{phase}-attachment-payload-")
        .repeat(64)
        .into_bytes();
    let metadata: HashMap<String, String> = [("phase".to_string(), phase.to_string())]
        .into_iter()
        .collect();
    let attachment = peer
        .store()
        .new_attachment_from_bytes(&bytes, metadata)
        .await?;
    peer.store()
        .execute((
            "INSERT INTO COLLECTION test_attachments (att ATTACHMENT) DOCUMENTS (:doc)",
            json!({ "doc": { "phase": phase, "att": &attachment } }),
        ))
        .await?;
    Ok(())
}

/// Poll the atomic cell until `predicate(count)` is true, or the timeout
/// elapses. Returns whether the predicate was ever satisfied.
async fn wait_until(
    cell: &AtomicUsize,
    predicate: impl Fn(usize) -> bool,
    timeout: Duration,
) -> bool {
    let deadline = tokio::time::Instant::now() + timeout;
    loop {
        if predicate(cell.load(Ordering::SeqCst)) {
            return true;
        }
        if tokio::time::Instant::now() >= deadline {
            return false;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
}

/// End-to-end proof that toggling `peer_to_peer.multicast.enabled` via
/// `set_transport_config` — *without* stopping sync — starts and halts real
/// cross-peer data sync.
///
/// Two same-host peers whose ONLY transport is NORM multicast, so disabling
/// multicast on peer B fully partitions it even though its sync session stays
/// active the whole time (asserted via `sync().is_active()`).
///
/// Companion to `sync_stop_partitions_norm_multicast_for_docs_and_attachments`,
/// which drives the same partition/resume via `sync().stop()` / `start()`.
/// `#[ignore]` for the same reason: it needs real NORM multicast on the
/// loopback interface, which isn't universally available in sandboxed CI. Run:
///
/// ```text
/// cargo test --features __ditto_multicast_beta \
///     --test norm_multicast_sync_stop -- --ignored
/// ```
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "requires NORM multicast on loopback; opt-in via --ignored"]
async fn config_toggle_starts_and_stops_norm_multicast_sync() -> Result<(), Error> {
    let app_id = DatabaseId::generate();
    let port = unique_multicast_port();

    let (_root_a, peer_a) = open_norm_peer(app_id.clone(), port).await?;
    let (_root_b, peer_b) = open_norm_peer(app_id, port).await?;

    peer_a.sync().start()?;
    peer_b.sync().start()?;

    let _sub_a = peer_a
        .sync()
        .register_subscription("SELECT * FROM test_docs")?;
    let _sub_b = peer_b
        .sync()
        .register_subscription("SELECT * FROM test_docs")?;
    let (doc_count_b, _doc_observer_b) =
        register_count_observer(&peer_b, "SELECT * FROM test_docs")?;

    // -------- Phase 1: multicast on, both sync-active -> B receives. --------
    insert_doc(&peer_a, 1).await?;
    assert!(
        wait_until(&doc_count_b, |n| n >= 1, Duration::from_secs(20)).await,
        "Phase 1: peer_b should receive peer_a's doc over NORM multicast while both are active",
    );

    // -------- Phase 2: disable multicast on B via config (sync stays active). --
    // multicast is B's only transport, so this fully partitions B without ever
    // touching its sync session. `set_transport_config` tears the NORM engine
    // down synchronously (the spawner's stop closure runs within the call), so
    // doc 2 below is inserted strictly after B has detached.
    peer_b.set_transport_config(norm_multicast_tc(port, false));
    assert!(
        peer_b.sync().is_active(),
        "Phase 2: disabling multicast via set_transport_config must NOT stop sync",
    );

    insert_doc(&peer_a, 2).await?;
    // Wait the same window Phase 1 treats as a valid delivery (20s) so a
    // slow-but-working delivery can't let this negative assertion pass
    // vacuously: if multicast weren't actually torn down, doc 2 would have
    // arrived within this window. doc_count_b must stay at 1.
    tokio::time::sleep(Duration::from_secs(20)).await;
    assert_eq!(
        doc_count_b.load(Ordering::SeqCst),
        1,
        "Phase 2: with multicast disabled via config, peer_b must not receive the doc inserted on \
         peer_a — even though peer_b's sync is still active",
    );

    // -------- Phase 3: re-enable multicast on B via config -> doc 2 arrives. --
    peer_b.set_transport_config(norm_multicast_tc(port, true));
    assert!(
        peer_b.sync().is_active(),
        "Phase 3: re-enabling multicast via config keeps sync active",
    );
    assert!(
        wait_until(&doc_count_b, |n| n >= 2, Duration::from_secs(20)).await,
        "Phase 3: re-enabling multicast via config must resume sync and deliver the doc inserted \
         during the partition",
    );

    Ok(())
}
