//! Unit 2 — bounded inbound poller wiring.
//!
//! Asserts the poll tick surfaces new rows past the watermark exactly once,
//! advances the watermark monotonically, seeds the watermark to MAX(ROWID) on
//! a fresh boot (so pre-existing text never replays), and is a no-op when the
//! feature is OFF. Drives `poll_once` with injected `read_max`/`read_new`
//! closures over a synthetic source — no chat.db, no macOS dependency.

mod messaging_harness;
use messaging_harness::{inbound, Harness, PAIRED};

use std::sync::atomic::{AtomicI64, Ordering};
use std::sync::{Arc, Mutex};

/// A synthetic chat.db: an ordered row log the test appends to between ticks,
/// plus the `read_max`/`read_new` closures the poller injects.
#[derive(Default)]
struct FakeDb {
    rows: Mutex<Vec<car_ffi_common::integrations::InboundMessage>>,
}

impl FakeDb {
    fn push(&self, handle: &str, body: &str, rowid: i64) {
        self.rows.lock().unwrap().push(inbound(handle, body, rowid));
    }
    fn max_rowid(&self) -> i64 {
        self.rows
            .lock()
            .unwrap()
            .iter()
            .map(|r| r.rowid)
            .max()
            .unwrap_or(0)
    }
    fn rows_after(&self, min: i64) -> Vec<car_ffi_common::integrations::InboundMessage> {
        self.rows
            .lock()
            .unwrap()
            .iter()
            .filter(|r| r.rowid > min)
            .cloned()
            .collect()
    }
}

#[tokio::test]
async fn fresh_boot_seeds_watermark_no_replay() {
    let h = Harness::enabled_paired();
    let db = Arc::new(FakeDb::default());
    // A pre-existing row exists at boot.
    db.push(PAIRED, "approve", 10);
    let _id = h.seed_fire_and_return("x").await;

    let db_max = db.clone();
    let db_new = db.clone();
    let n = h
        .orch
        .poll_once(
            move || Ok(db_max.max_rowid()),
            move |min| Ok(db_new.rows_after(min)),
        )
        .await
        .unwrap();

    // Fresh boot seeds to MAX(ROWID)=10 and surfaces NOTHING (no replay).
    assert_eq!(n, 0, "fresh boot returns zero rows (seed only)");
    assert_eq!(
        h.orch.parse_call_count(),
        0,
        "the pre-existing row is never replayed as new"
    );
}

#[tokio::test]
async fn new_rows_surfaced_exactly_once_watermark_monotonic() {
    let h = Harness::enabled_paired();
    let id = h.seed_fire_and_return("send wire").await;
    h.orch.observe_and_notify().await; // get a code so the reply can resolve
    let code = messaging_harness::extract_code(&h.spy.calls()[0].1).unwrap();

    let db = Arc::new(FakeDb::default());
    let db_max = db.clone();
    let db_new = db.clone();
    let read_max = move || Ok(db_max.max_rowid());
    let read_new = move |min: i64| Ok(db_new.rows_after(min));

    // Tick 1: fresh boot (empty DB), seeds watermark to 0.
    let n0 = h.orch.poll_once(&read_max, &read_new).await.unwrap();
    assert_eq!(n0, 0);

    // A new reply arrives between ticks.
    db.push(PAIRED, &format!("{code} approve"), 1);

    // Tick 2: surfaces the new row exactly once → resolves the approval.
    let n1 = h.orch.poll_once(&read_max, &read_new).await.unwrap();
    assert_eq!(n1, 1, "the new row is surfaced exactly once");
    assert_eq!(h.orch.parse_call_count(), 1, "parsed exactly once");
    assert!(h.is_resolved(&id).await, "the reply resolved its approval");

    // Tick 3: no new rows → zero (the watermark advanced past rowid 1).
    let n2 = h.orch.poll_once(&read_max, &read_new).await.unwrap();
    assert_eq!(n2, 0, "no replay — the watermark advanced");
    assert_eq!(
        h.orch.parse_call_count(),
        1,
        "still exactly one parse total (no replay of the seen row)"
    );
}

#[tokio::test]
async fn watermark_advances_and_persists_across_reconstruction() {
    // The watermark is durable: a second orchestrator over the same base dir
    // does not replay rows the first one already advanced past.
    let h = Harness::enabled_paired();
    let db = Arc::new(FakeDb::default());
    db.push(PAIRED, "hello", 5); // pre-existing
    let db_max = db.clone();
    let db_new = db.clone();

    // Fresh boot seeds to 5.
    h.orch
        .poll_once(
            {
                let d = db_max.clone();
                move || Ok(d.max_rowid())
            },
            {
                let d = db_new.clone();
                move |min| Ok(d.rows_after(min))
            },
        )
        .await
        .unwrap();

    // New row at 6.
    db.push(PAIRED, "approve", 6);
    h.orch
        .poll_once(
            {
                let d = db_max.clone();
                move || Ok(d.max_rowid())
            },
            {
                let d = db_new.clone();
                move |min| Ok(d.rows_after(min))
            },
        )
        .await
        .unwrap();
    assert_eq!(h.orch.parse_call_count(), 1, "only the post-seed row parsed");

    // The watermark file now reflects rowid 6 — load it directly to prove it
    // advanced monotonically (5 → 6).
    let wm = car_ffi_common::integrations::Watermark::load(std::path::Path::new(
        h.watermark_dir(),
    ))
    .unwrap()
    .expect("watermark persisted");
    assert_eq!(wm.last_rowid, 6, "watermark advanced to the latest seen rowid");
}

#[tokio::test]
async fn feature_off_poll_is_noop() {
    let h = Harness::disabled();
    let calls = Arc::new(AtomicI64::new(0));
    let calls2 = calls.clone();
    let n = h
        .orch
        .poll_once(
            move || {
                calls.fetch_add(1, Ordering::SeqCst);
                Ok(0)
            },
            move |_min| Ok(vec![]),
        )
        .await
        .unwrap();
    assert_eq!(n, 0, "feature OFF ⇒ poll surfaces nothing");
    assert_eq!(
        calls2.load(Ordering::SeqCst),
        0,
        "feature OFF ⇒ no chat.db read (read_max never called)"
    );
}
