//! The **streaming/delta** Iceberg → FalkorDB path against a **real FalkorDB**.
//!
//! This is the LAW-7 container test for the incremental ingest: skade writes an
//! Iceberg snapshot, the `IcebergSource` + `Listener` ingest it, and we assert
//! the live FalkorDB node count via `GRAPH.QUERY` — then a SECOND skade snapshot
//! (more rows) is written and the **delta** ingest is run, asserting only the
//! new nodes were added (cursor advanced, no full reload).
//!
//! Pure Rust end to end: skade (Iceberg, arrow 57) → knut-bifrost delta scan
//! (manifest diff, arrow 56) → FalkorDbSink (Cypher MERGE) → FalkorDB. No Spark.
//!
//! ## Prerequisites
//! - Podman socket: `systemctl --user start podman.socket`
//! - FalkorDB image (override with `KNUT_FALKORDB_IMAGE`):
//!   `podman build -t knut-falkordb-valkey9 -f containers/falkordb-valkey9/Containerfile .`
//!   (or `KNUT_FALKORDB_IMAGE=docker.io/falkordb/falkordb:latest`)
//!
//! ## Run
//! ```bash
//! cargo test -p knut-bifrost --features skade --test delta_iceberg_falkordb \
//!   -- --ignored --nocapture
//! ```
#![cfg(feature = "skade")]

use std::sync::Arc;
use std::time::Duration;

use knut_bifrost::{
    Listener, MappingSpec, SyncState, TransformRegistry, sink::FalkorDbSink,
    source::IcebergSource,
};
use skade::arrow_array::{Int64Array, RecordBatch as B57, StringArray};
use skade::arrow_schema::{DataType, Field, Schema};
use testcontainers::{GenericImage, ImageExt, core::ContainerPort, runners::AsyncRunner};

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

fn people57(ids: &[i64]) -> B57 {
    let names: Vec<String> = ids.iter().map(|i| format!("person_{i}")).collect();
    B57::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()
}

/// Append `ids` as one new Iceberg snapshot via skade.
async fn append_iceberg(dir: &std::path::Path, ids: &[i64]) {
    let wh = skade::Warehouse::open(dir).await.unwrap();
    let batch = people57(ids);
    let mut table = wh.table_or_create("people", &batch.schema()).await.unwrap();
    table.append(&[batch]).await.unwrap();
    // `wh`/`table` drop here → redb lock released for the IcebergSource.
}

/// Equality-delete `ids` as one new Iceberg `Delete` snapshot via skade.
async fn delete_iceberg(dir: &std::path::Path, ids: &[i64]) {
    let wh = skade::Warehouse::open(dir).await.unwrap();
    let mut table = wh.table("people").await.unwrap();
    let key = B57::try_new(
        Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)])),
        vec![Arc::new(Int64Array::from(ids.to_vec()))],
    )
    .unwrap();
    table.delete_equality(&key, &["id"]).await.unwrap();
}

fn ensure_docker_host() {
    if std::env::var("DOCKER_HOST").is_err() {
        let uid = std::process::Command::new("id")
            .arg("-u")
            .output()
            .ok()
            .and_then(|o| String::from_utf8(o.stdout).ok())
            .and_then(|s| s.trim().parse::<u32>().ok())
            .unwrap_or(1000);
        let sock = format!("/run/user/{uid}/podman/podman.sock");
        if std::path::Path::new(&sock).exists() {
            unsafe { std::env::set_var("DOCKER_HOST", format!("unix://{sock}")) };
        }
    }
}

async fn count(conn: &mut redis::aio::MultiplexedConnection, graph: &str, q: &str) -> i64 {
    fn walk(v: &redis::Value) -> Option<i64> {
        match v {
            redis::Value::Int(i) => Some(*i),
            redis::Value::Array(items) | redis::Value::Set(items) => items.iter().find_map(walk),
            redis::Value::BulkString(b) => String::from_utf8_lossy(b).parse().ok(),
            _ => None,
        }
    }
    let v: redis::Value = redis::cmd("GRAPH.QUERY")
        .arg(graph)
        .arg(q)
        .query_async(conn)
        .await
        .unwrap();
    walk(&v).unwrap_or(-1)
}

// Throughput twin (rows/s for incremental-vs-full ingest) lives in the nornir
// bencher `knut.delta_ingest` (xtask/examples/nornir-bench.rs); this stays the
// downscaled fast correctness check (tiny N=3 / M=2 / one delete) of the same path.
#[tokio::test]
#[ignore = "needs podman + a FalkorDB image; run with --ignored"]
async fn delta_iceberg_streams_only_new_rows_into_falkordb() {
    ensure_docker_host();

    // ── setup: FalkorDB container ───────────────────────────────────────────
    let image = std::env::var("KNUT_FALKORDB_IMAGE")
        .unwrap_or_else(|_| "localhost/knut-falkordb-valkey9:latest".into());
    let (name, tag) = image.rsplit_once(':').unwrap_or((image.as_str(), "latest"));
    let falkor = GenericImage::new(name, tag)
        .with_exposed_port(ContainerPort::Tcp(6379))
        .with_startup_timeout(Duration::from_secs(120))
        .start()
        .await
        .expect("start falkordb (build the image first, or set KNUT_FALKORDB_IMAGE)");
    let port = falkor.get_host_port_ipv4(6379).await.unwrap();
    let url = format!("redis://127.0.0.1:{port}");
    let graph = "delta";

    let client = redis::Client::open(url.clone()).unwrap();
    let mut conn = client.get_multiplexed_async_connection().await.unwrap();

    // ── the Iceberg warehouse (skade) + the streaming listener ──────────────
    let warehouse = tempfile::tempdir().unwrap();
    let state = tempfile::tempdir().unwrap();

    // ── snapshot A: N=3 rows → ingest → assert 3 nodes ──────────────────────
    append_iceberg(warehouse.path(), &[1, 2, 3]).await;

    let mut listener = Listener::new(
        MappingSpec::from_yaml(MAPPING).unwrap(),
        TransformRegistry::new(),
        IcebergSource::new(warehouse.path()),
        FalkorDbSink::connect(&url, graph).await.unwrap(),
        SyncState::open(state.path().join("state.json")).unwrap(),
    );

    let round = listener.sync_once().await.unwrap();
    assert_eq!(round.failed, vec![], "snapshot A round failures");
    assert_eq!(round.synced.len(), 1, "table synced");
    assert_eq!(round.synced[0].1, 3, "N=3 rows shipped from snapshot A");
    assert_eq!(
        count(&mut conn, graph, "MATCH (n:Person) RETURN count(n)").await,
        3,
        "N=3 Person nodes after snapshot A",
    );

    // ── idle: no new snapshot → nothing shipped, count unchanged ────────────
    let idle = listener.sync_once().await.unwrap();
    assert!(idle.synced.is_empty(), "no new snapshot → idle round");
    assert_eq!(
        count(&mut conn, graph, "MATCH (n:Person) RETURN count(n)").await,
        3,
    );

    // ── snapshot B: M=2 MORE rows → DELTA ingest → assert only +2 (total 5) ──
    append_iceberg(warehouse.path(), &[4, 5]).await;

    let round = listener.sync_once().await.unwrap();
    assert_eq!(round.failed, vec![], "snapshot B round failures");
    assert_eq!(
        round.synced[0].1, 2,
        "DELTA shipped only M=2 new rows (not a full reload of 5)",
    );
    assert_eq!(
        count(&mut conn, graph, "MATCH (n:Person) RETURN count(n)").await,
        5,
        "N+M = 5 Person nodes after the delta (cursor advanced, no reload)",
    );

    // ── the new ids 4 and 5 are present (the delta actually injected them) ───
    assert_eq!(
        count(
            &mut conn,
            graph,
            "MATCH (n:Person) WHERE n.id IN [4,5] RETURN count(n)"
        )
        .await,
        2,
        "the two delta nodes (ids 4,5) are in the graph",
    );

    // ── snapshot C: a CDC equality-DELETE of id=2 → DELTA removes only it ────
    delete_iceberg(warehouse.path(), &[2]).await;

    let round = listener.sync_once().await.unwrap();
    assert_eq!(round.failed, vec![], "snapshot C (delete) round failures");
    assert_eq!(round.synced.len(), 1, "the delete is one synced round");
    assert_eq!(
        round.synced[0].2.deleted, 1,
        "the CDC delta deleted exactly one node",
    );
    assert_eq!(
        count(&mut conn, graph, "MATCH (n:Person) RETURN count(n)").await,
        4,
        "5 − 1 = 4 nodes after the equality-delete of id=2 (pure-Rust CDC delete)",
    );
    assert_eq!(
        count(
            &mut conn,
            graph,
            "MATCH (n:Person) WHERE n.id = 2 RETURN count(n)"
        )
        .await,
        0,
        "id=2 is gone; the other nodes are untouched",
    );

    // teardown: container + tempdirs drop here.
}
