//! Reproduction for peat-mesh#247: under sustained/concurrent write load a
//! latency-sensitive (high-QoS) document must still converge near the floor
//! and not be head-of-line-blocked behind a backlog of lower-priority
//! documents.
//!
//! ## Status: `#[ignore]`d regression guard (ADR-0013 fix landed)
//!
//! `#[ignore]`d so the networked P2P functional job picks it up
//! (`cargo test --features automerge-backend --tests -- --ignored`). It pins
//! the #247 acceptance criterion now that the doc-level QoS-priority fanout
//! has landed: a Critical-QoS document enqueued **last**, behind the whole
//! Bulk backlog, is fanned out — and applied on the receiver — near rank 0.
//! Before the fix this was rank ≈ NUM_BULK (dead last); a regression that
//! reinstates arrival-order fanout would push the rank back up and fail here.
//!
//! ## What earlier attempts got wrong (kept so it isn't re-litigated)
//!
//! A first cut measured convergence by polling `get()` on the receiver. That
//! number (~80 ms under a 24×64 KiB backlog) was **`get()`-polling
//! starvation**, not pipeline latency: with all backends in one process the
//! runtime is saturated applying the backlog's CPU-heavy FullHistory delta
//! merges, so the polling task can't run — the probe was actually in every
//! receiver's store at ~15–28 ms. Unfaithful metric.
//!
//! A second cut added a frame-level priority lane to the persistent channel
//! and measured the critical doc's **apply rank** on the receiver (via the
//! store's observer stream — emitted in apply order, timing-independent).
//! That revealed the real shape: the critical doc still applied **dead last**,
//! because the serial fanout driver is paced by per-doc prepare cost, so the
//! single QUIC writer drains in lockstep and no queue ever builds for the
//! frame lane to reorder. The head-of-line stall is **upstream of the
//! channel**: a critical doc's fanout doesn't *start* until all prior
//! fanouts finish serially.
//!
//! ## What this measures
//!
//! The **global apply rank** of a single high-QoS (`alerts` → Critical)
//! document among a backlog of `NUM_BULK` lower-QoS (`sitreps` → Bulk)
//! documents on the *receiver*, via `AutomergeStore::subscribe_to_observer_changes`
//! (fires on every applied put, all collections, in apply order). The critical
//! doc is driven **last** — the worst case for any FIFO-ish fanout.
//!
//! - rank ≈ 0 ⇒ the Critical-QoS doc was applied ahead of the Bulk backlog ⇒
//!   converges at the per-document floor regardless of backlog size. The
//!   #247 acceptance shape, achieved by doc-level QoS-priority fanout.
//! - rank ≈ NUM_BULK ⇒ it waited behind the whole backlog ⇒ latency grows
//!   with load (the #247 symptom; current behaviour).

#![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;

/// Lower-QoS bulk backlog collection (`for_collection` → `Bulk`).
const BULK_COLLECTION: &str = "sitreps";
/// Latency-sensitive collection (`for_collection` → `Critical`).
const CRITICAL_COLLECTION: &str = "alerts";
const CRITICAL_ID: &str = "critical-probe";
const NUM_BULK: usize = 24;
const BULK_DOC_BYTES: usize = 64 * 1024;

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: "repro-247".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)
}

fn teach_addr(from: &Arc<AutomergeBackend>, to: &Arc<AutomergeBackend>) {
    let addr_to = to.blob_store().endpoint().addr();
    from.blob_store().memory_lookup().add_endpoint_info(addr_to);
}

fn bulk_fields(seq: usize) -> HashMap<String, serde_json::Value> {
    let mut m = HashMap::new();
    m.insert("seq".to_string(), serde_json::Value::from(seq as u64));
    m.insert(
        "body".to_string(),
        serde_json::Value::String("x".repeat(BULK_DOC_BYTES)),
    );
    m
}

#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
#[ignore = "peat-mesh#247: rank<=3 regression guard for the QoS-priority fanout \
            (ADR-0013), run by the P2P functional job via --ignored."]
async fn critical_doc_is_not_head_of_line_blocked_by_bulk_backlog() {
    let shared_key = FormationKey::generate_secret();
    let (source, _ds) = build_backend("repro-247", &shared_key).await;
    let (leaf, _dl) = build_backend("repro-247", &shared_key).await;

    teach_addr(&source, &leaf);
    teach_addr(&leaf, &source);

    // Global apply-order signal on the receiver: fires on every applied put
    // (all collections), in apply order. Subscribe before any traffic.
    let mut applied_rx = leaf.store().subscribe_to_observer_changes();

    // Connect source → leaf. We deliberately do NOT `start_sync` the source:
    // its auto-pusher would fan out on local writes with nondeterministic
    // ordering, racing the explicit serial drive below.
    let id_leaf = leaf.blob_store().endpoint().id();
    let connected = source
        .connect_to_peer(&id_leaf.to_string(), &[])
        .await
        .expect("connect source → leaf");
    assert!(connected, "connect_to_peer must succeed");

    // Stage all documents in the source store (local writes; with no
    // auto-pusher running these do not fan out on their own).
    for i in 0..NUM_BULK {
        source
            .document_store()
            .upsert(
                BULK_COLLECTION,
                Document::with_id(format!("bulk-{i:03}"), bulk_fields(i)),
            )
            .await
            .expect("stage bulk");
    }
    let mut crit_fields = HashMap::new();
    crit_fields.insert("urgent".to_string(), serde_json::Value::Bool(true));
    source
        .document_store()
        .upsert(
            CRITICAL_COLLECTION,
            Document::with_id(CRITICAL_ID, crit_fields),
        )
        .await
        .expect("stage critical");

    // Start sync on the source — this spawns the QoS-priority fanout worker
    // (ADR-0013) that drains the queue. The staged writes above happened
    // BEFORE start_sync subscribed to change events, so the auto-pusher does
    // not see them; we drive the queue explicitly below to control ordering.
    source.start_sync().await.expect("start_sync source");

    // Enqueue all (Bulk-QoS) docs FIRST, then the (Critical-QoS) doc LAST —
    // the worst case for a FIFO fanout. `enqueue_fanout` is non-blocking, so
    // all NUM_BULK + 1 entries are in the priority queue effectively at once
    // and the single worker drains them highest-QoS-first: the Critical doc
    // must be fanned (and therefore applied on the receiver) ahead of the
    // Bulk backlog despite being enqueued last.
    let coordinator = source.coordinator();
    for i in 0..NUM_BULK {
        coordinator.enqueue_fanout(&format!("{BULK_COLLECTION}:bulk-{i:03}"));
    }
    coordinator.enqueue_fanout(&format!("{CRITICAL_COLLECTION}:{CRITICAL_ID}"));

    // Collect the receiver's apply order until all NUM_BULK + 1 docs applied
    // (or a deadline). Record the rank at which the critical key appears.
    let critical_key = format!("{CRITICAL_COLLECTION}:{CRITICAL_ID}");
    let total = NUM_BULK + 1;
    let mut applied: Vec<String> = Vec::with_capacity(total);
    let mut critical_rank: Option<usize> = None;
    let deadline = tokio::time::Instant::now() + Duration::from_secs(60);
    while applied.len() < total {
        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
        if remaining.is_zero() {
            break;
        }
        match tokio::time::timeout(remaining, applied_rx.recv()).await {
            Ok(Ok(key)) => {
                if key == critical_key && critical_rank.is_none() {
                    critical_rank = Some(applied.len());
                }
                applied.push(key);
            }
            Ok(Err(_)) => break, // lagged/closed
            Err(_) => break,     // deadline
        }
    }

    let rank = critical_rank.unwrap_or_else(|| {
        panic!(
            "critical doc never applied on receiver; applied {}/{}: {:?}",
            applied.len(),
            total,
            applied
        )
    });
    eprintln!(
        "repro-247: critical ({CRITICAL_COLLECTION}) applied at rank {rank}/{total} \
         on receiver (0 == first), ahead of {NUM_BULK} bulk ({BULK_COLLECTION}) docs"
    );

    // Everything converges...
    assert_eq!(
        applied.len(),
        total,
        "all {total} documents must converge on the receiver"
    );

    // ...and the regression guard (ADR-0013): the Critical-QoS doc, though
    // enqueued LAST behind the whole Bulk backlog, is fanned out — and applied
    // on the receiver — near the front. Slack of 3 covers the at-most-one
    // in-flight bulk fanout the worker has already started when the queue
    // fills, plus loopback scheduling jitter. Pre-fix this was rank ≈ NUM_BULK.
    assert!(
        rank <= 3,
        "critical doc was head-of-line-blocked: applied at rank {rank}/{total} \
         (expected ≤ 3). The QoS-priority fanout (ADR-0013) must drain the \
         Critical-QoS doc ahead of the Bulk backlog."
    );

    source.shutdown().await.expect("shutdown source");
    leaf.shutdown().await.expect("shutdown leaf");
}
