//! Lifecycle test matrix: **create file → detect → ingest → teardown**, with
//! no container and no human. Each case stages parquet drops in a fresh temp
//! directory (setup), runs the bifröst `Listener` with a recording sink, and
//! asserts what was detected and ingested; the tempdir drop is teardown.
//!
//! The final test asserts on the **structured tracing events** the library
//! emits (`file_detected`, `table_synced`, `round_complete`) — proving the log
//! stream is a stable contract a test or bencher can drive headlessly.

use std::sync::{Arc, Mutex};

use arrow_array::{Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
use async_trait::async_trait;
use knut_bifrost::{
    GraphMutation, Listener, MappingSpec, SinkReport, SyncMode, SyncState, TransformRegistry,
    sink::GraphSink, source::FileSource,
};

/// Emit a functional-status row into the nornir test matrix. With the
/// `testmatrix` feature ON it records `ok`/`detail` for `check`; with the
/// feature OFF it is a no-op (and `nornir-testmatrix` is not built into this
/// test target). Component is the integration suite that ran.
#[cfg(feature = "testmatrix")]
fn fs(check: &str, ok: bool, detail: &str) {
    nornir_testmatrix::functional_status("knut-bifrost-lifecycle", check, ok, detail);
}
#[cfg(not(feature = "testmatrix"))]
fn fs(_check: &str, _ok: bool, _detail: &str) {}

const MAPPING: &str = r#"
version: 1
tables:
  - table: people
    mode: MODE
    pipeline:
      - vertex:
          label: Person
          id_column: id
          properties: [name]
"#;

fn people_batch(ids: &[i64]) -> RecordBatch {
    let names: Vec<String> = ids.iter().map(|i| format!("person_{i}")).collect();
    RecordBatch::try_new(
        Arc::new(Schema::new(vec![
            Field::new("id", DataType::Int64, false),
            Field::new("name", DataType::Utf8, false),
        ])),
        vec![
            Arc::new(Int64Array::from(ids.to_vec())),
            Arc::new(StringArray::from(names)),
        ],
    )
    .unwrap()
}

/// Stage one parquet chunk under `<root>/people/<name>`.
fn drop_chunk(root: &std::path::Path, name: &str, ids: &[i64]) {
    let dir = root.join("people");
    std::fs::create_dir_all(&dir).unwrap();
    graphar::io::write_chunk(
        dir.join(name),
        std::slice::from_ref(&people_batch(ids)),
        &graphar::FileType::Parquet,
    )
    .unwrap();
}

/// A sink that records every mutation it receives — the "ingested" side.
#[derive(Clone, Default)]
struct RecordingSink {
    vertices: Arc<Mutex<usize>>,
    edges: Arc<Mutex<usize>>,
}

#[async_trait]
impl GraphSink for RecordingSink {
    async fn apply(&mut self, mutation: &GraphMutation) -> knut_bifrost::Result<SinkReport> {
        let v: usize = mutation.vertices.iter().map(|s| s.batch.num_rows()).sum();
        let e: usize = mutation.edges.iter().map(|s| s.batch.num_rows()).sum();
        *self.vertices.lock().unwrap() += v;
        *self.edges.lock().unwrap() += e;
        let d: usize = mutation.deletes.iter().map(|s| s.batch.num_rows()).sum();
        Ok(SinkReport {
            vertices_written: v,
            edges_written: e,
            deleted: d,
        })
    }
}

fn mapping_for(mode: SyncMode) -> MappingSpec {
    let m = match mode {
        SyncMode::Full => "full",
        SyncMode::Incremental => "incremental",
        SyncMode::Cdc => "cdc",
    };
    MappingSpec::from_yaml(&MAPPING.replace("MODE", m)).unwrap()
}

/// Build a listener over `root` with a fresh recording sink + state file.
fn listener(
    root: &std::path::Path,
    mode: SyncMode,
) -> (Listener<FileSource, RecordingSink>, RecordingSink) {
    let sink = RecordingSink::default();
    let listener = Listener::new(
        mapping_for(mode),
        TransformRegistry::new(),
        FileSource::new(root),
        sink.clone(),
        SyncState::open(root.join(".state.json")).unwrap(),
    );
    (listener, sink)
}

// ── the matrix ──────────────────────────────────────────────────────────────

#[tokio::test]
async fn incremental_single_drop_then_idle() {
    let dir = tempfile::tempdir().unwrap();
    drop_chunk(dir.path(), "chunk0.parquet", &[1, 2, 3]);
    let (mut l, sink) = listener(dir.path(), SyncMode::Incremental);

    let round = l.sync_once().await.unwrap();
    assert_eq!(round.synced.len(), 1, "the drop was detected + ingested");
    assert_eq!(round.synced[0].1, 3, "3 rows");
    assert_eq!(*sink.vertices.lock().unwrap(), 3, "3 vertices ingested");

    // Idle: no new file → nothing re-ingested.
    let idle = l.sync_once().await.unwrap();
    assert!(idle.synced.is_empty());
    let v = *sink.vertices.lock().unwrap();
    assert_eq!(v, 3, "no double-ingest");
    fs(
        "incremental_single_drop_then_idle",
        round.synced.len() == 1 && round.synced[0].1 == 3 && v == 3 && idle.synced.is_empty(),
        &format!("1 drop → 3 rows / {v} vertices, then idle (no double-ingest)"),
    );
}

#[tokio::test]
async fn incremental_second_drop_is_delta_only() {
    let dir = tempfile::tempdir().unwrap();
    drop_chunk(dir.path(), "chunk0.parquet", &[1, 2]);
    let (mut l, sink) = listener(dir.path(), SyncMode::Incremental);
    l.sync_once().await.unwrap();
    assert_eq!(*sink.vertices.lock().unwrap(), 2);

    // A second file appears → only its rows flow.
    drop_chunk(dir.path(), "chunk1.parquet", &[3, 4, 5]);
    let round = l.sync_once().await.unwrap();
    assert_eq!(round.synced[0].1, 3, "only the new file's 3 rows");
    let v = *sink.vertices.lock().unwrap();
    assert_eq!(v, 5, "2 + 3 total");
    fs(
        "incremental_second_drop_is_delta_only",
        round.synced[0].1 == 3 && v == 5,
        &format!("second drop shipped only its 3 rows; {v} vertices total"),
    );
}

#[tokio::test]
async fn full_mode_remerges_every_round() {
    let dir = tempfile::tempdir().unwrap();
    drop_chunk(dir.path(), "chunk0.parquet", &[1, 2]);
    drop_chunk(dir.path(), "chunk1.parquet", &[3, 4]);
    let (mut l, sink) = listener(dir.path(), SyncMode::Full);

    let round = l.sync_once().await.unwrap();
    assert_eq!(round.synced[0].1, 4, "both files read");
    // Full re-reads every round (idempotent MERGE downstream).
    let round2 = l.sync_once().await.unwrap();
    assert_eq!(round2.synced[0].1, 4, "full re-reads");
    let v = *sink.vertices.lock().unwrap();
    assert_eq!(v, 8, "4 + 4 (MERGE dedupes in a real sink)");
    fs(
        "full_mode_remerges_every_round",
        round.synced[0].1 == 4 && round2.synced[0].1 == 4 && v == 8,
        &format!("full mode re-read 4 rows each round → {v} vertices"),
    );
}

#[tokio::test]
async fn empty_directory_is_idle() {
    let dir = tempfile::tempdir().unwrap();
    let (mut l, sink) = listener(dir.path(), SyncMode::Incremental);
    let round = l.sync_once().await.unwrap();
    assert!(round.synced.is_empty() && round.failed.is_empty());
    let v = *sink.vertices.lock().unwrap();
    assert_eq!(v, 0);
    fs(
        "empty_directory_is_idle",
        round.synced.is_empty() && round.failed.is_empty() && v == 0,
        "empty dir → idle round, 0 vertices",
    );
}

#[tokio::test]
async fn multi_file_first_round_reads_all() {
    let dir = tempfile::tempdir().unwrap();
    for (i, name) in ["a.parquet", "b.parquet", "c.parquet"].iter().enumerate() {
        drop_chunk(dir.path(), name, &[(i * 10) as i64, (i * 10 + 1) as i64]);
    }
    let (mut l, sink) = listener(dir.path(), SyncMode::Incremental);
    let round = l.sync_once().await.unwrap();
    assert_eq!(round.synced[0].1, 6, "3 files × 2 rows");
    let v = *sink.vertices.lock().unwrap();
    assert_eq!(v, 6);
    fs(
        "multi_file_first_round_reads_all",
        round.synced[0].1 == 6 && v == 6,
        &format!("first round read all 3 files → {v} vertices"),
    );
}

// ── AIP-82: the listener auto-emits asset events after a round ──────────────

#[tokio::test]
async fn listener_emits_asset_events_after_round() {
    use knut_bifrost::{AssetEmitter, AssetEvent};

    #[derive(Clone, Default)]
    struct RecEmitter {
        seen: Arc<Mutex<Vec<AssetEvent>>>,
    }
    #[async_trait]
    impl AssetEmitter for RecEmitter {
        async fn emit(&self, events: &[AssetEvent]) -> knut_bifrost::Result<()> {
            self.seen.lock().unwrap().extend_from_slice(events);
            Ok(())
        }
    }

    let dir = tempfile::tempdir().unwrap();
    drop_chunk(dir.path(), "chunk0.parquet", &[1, 2, 3]);
    let emitter = RecEmitter::default();
    let (l, _sink) = listener(dir.path(), SyncMode::Incremental);
    let mut l = l.with_asset_emitter(Arc::new(emitter.clone()), "graph://demo/");

    // First round ingests + emits one asset event for the `people` table.
    l.sync_once().await.unwrap();
    {
        let seen = emitter.seen.lock().unwrap();
        assert_eq!(seen.len(), 1, "one asset event per ingested table");
        assert_eq!(seen[0].uri, "graph://demo/people");
        assert_eq!(seen[0].extra["rows"], 3);
    }

    // Idle round: nothing ingested → nothing emitted.
    l.sync_once().await.unwrap();
    let n = emitter.seen.lock().unwrap().len();
    assert_eq!(n, 1, "idle round emits no asset events");
    fs(
        "listener_emits_asset_events_after_round",
        n == 1,
        &format!("{n} AIP-82 asset event after ingest; idle round emits none"),
    );
}

// ── the log contract (drive the real CLI, assert its JSON) ──────────────────

/// Drive the `bifrost` binary headlessly — create a file drop, run one
/// `--dry-run` round with `--json-logs`, and assert the structured lifecycle
/// events appear in order on stdout. This is the contract a test or bencher
/// keys off: no database, no human, no in-process subscriber races.
///
/// Requires `--features cli` (so the `bifrost` bin is built); gated to avoid a
/// missing-binary error on the default test run.
#[cfg(feature = "cli")]
#[test]
fn bifrost_cli_emits_lifecycle_json() {
    use std::process::Command;

    let dir = tempfile::tempdir().unwrap();
    drop_chunk(dir.path(), "chunk0.parquet", &[1, 2, 3]);
    let mapping = dir.path().join("mapping.yml");
    std::fs::write(&mapping, MAPPING.replace("MODE", "incremental")).unwrap();

    let out = Command::new(env!("CARGO_BIN_EXE_bifrost"))
        .args(["--json-logs", "run"])
        .arg("--mapping")
        .arg(&mapping)
        .arg("--files")
        .arg(dir.path())
        .args(["--dry-run", "--once"])
        .output()
        .expect("run bifrost");
    assert!(
        out.status.success(),
        "bifrost exited non-zero: {:?}",
        out.status
    );

    let logs = String::from_utf8(out.stdout).unwrap();
    let events: Vec<String> = logs
        .lines()
        .filter_map(|line| serde_json::from_str::<serde_json::Value>(line).ok())
        .filter_map(|v| v["fields"]["event"].as_str().map(String::from))
        .collect();

    for expected in [
        "file_detected",
        "files_read",
        "dry_run_mutation",
        "table_synced",
        "round_complete",
    ] {
        assert!(
            events.contains(&expected.to_string()),
            "missing event {expected}; got {events:?}"
        );
    }
    let pos = |e: &str| events.iter().position(|x| x == e).unwrap();
    let ordered =
        pos("file_detected") < pos("table_synced") && pos("table_synced") < pos("round_complete");
    assert!(pos("file_detected") < pos("table_synced"), "detect precedes ingest");
    assert!(pos("table_synced") < pos("round_complete"), "ingest precedes round end");
    fs(
        "bifrost_cli_emits_lifecycle_json",
        out.status.success() && ordered,
        "CLI --json-logs emits file_detected→table_synced→round_complete in order",
    );
}
