//! The create→detect→ingest→teardown lifecycle against a **real FalkorDB**.
//!
//! Same shape as `tests/lifecycle.rs`, but the recording sink is replaced by a
//! live `FalkorDbSink` backed by a FalkorDB container (testcontainers handles
//! setup + teardown). Each round stages a parquet drop and then asserts the
//! actual node count in FalkorDB via `GRAPH.QUERY`.
//!
//! ## 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 set `KNUT_FALKORDB_IMAGE=docker.io/falkordb/falkordb:latest`)
//!
//! ## Run
//! ```bash
//! cargo test -p knut-bifrost --test lifecycle_falkordb -- --ignored --nocapture
//! ```

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

use arrow_array::{Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
use knut_bifrost::{
    Listener, MappingSpec, SyncState, TableState, TransformRegistry, sink::FalkorDbSink,
    source::FileSource,
};
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 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()
}

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();
}

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)
}

#[tokio::test]
#[ignore = "needs podman + a FalkorDB image; run with --ignored"]
async fn lifecycle_ingests_into_real_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 = "lifecycle";

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

    // ── a listener over a watched dir → real FalkorDB sink ──────────────────
    let drops = tempfile::tempdir().unwrap();
    let state = tempfile::tempdir().unwrap();
    let mut listener = Listener::new(
        MappingSpec::from_yaml(MAPPING).unwrap(),
        TransformRegistry::new(),
        FileSource::new(drops.path()),
        FalkorDbSink::connect(&url, graph).await.unwrap(),
        SyncState::open(state.path().join("state.json")).unwrap(),
    );

    // ── round 1: create a file → detect → ingest ────────────────────────────
    drop_chunk(drops.path(), "chunk0.parquet", &[1, 2, 3]);
    let round = listener.sync_once().await.unwrap();
    assert_eq!(round.failed, vec![], "round 1 failures");
    assert_eq!(round.synced.len(), 1);
    assert_eq!(round.synced[0].1, 3, "3 rows shipped");
    assert_eq!(
        count(&mut conn, graph, "MATCH (n:Person) RETURN count(n)").await,
        3,
        "3 Person nodes in FalkorDB",
    );

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

    // ── round 2: a second drop → incremental delta ──────────────────────────
    drop_chunk(drops.path(), "chunk1.parquet", &[4, 5]);
    let round = listener.sync_once().await.unwrap();
    assert_eq!(round.failed, vec![], "round 2 failures");
    assert_eq!(round.synced[0].1, 2, "only the new file's 2 rows");
    assert_eq!(
        count(&mut conn, graph, "MATCH (n:Person) RETURN count(n)").await,
        5,
        "5 Person nodes after the incremental drop",
    );

    // ── re-ingesting the same id MERGEs (idempotent), not duplicates ────────
    drop_chunk(drops.path(), "chunk2.parquet", &[5, 6]);
    listener.sync_once().await.unwrap();
    assert_eq!(
        count(&mut conn, graph, "MATCH (n:Person) RETURN count(n)").await,
        6,
        "id 5 re-merged (not duplicated); only 6 added",
    );

    // teardown: container + tempdirs drop here.
}

#[tokio::test]
#[ignore = "needs podman + a Valkey/FalkorDB image; run with --ignored"]
async fn valkey_state_ledger_is_shared_across_instances() {
    ensure_docker_host();

    // The FalkorDB image is a Valkey 9 server — use it as the ledger store.
    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 valkey = GenericImage::new(name, tag)
        .with_exposed_port(ContainerPort::Tcp(6379))
        .with_startup_timeout(Duration::from_secs(120))
        .start()
        .await
        .expect("start valkey/falkordb");
    let port = valkey.get_host_port_ipv4(6379).await.unwrap();
    let url = format!("redis://127.0.0.1:{port}");
    let key = "bifrost:test:state";

    // Instance 1 commits some progress.
    let mut s1 = SyncState::open_valkey(&url, key).await.unwrap();
    assert_eq!(
        s1.get("lake.users"),
        TableState::default(),
        "fresh ledger is empty"
    );
    s1.advance(
        "lake.users",
        TableState {
            snapshot_id: Some(7),
            done_files: vec!["chunk0.parquet".into(), "chunk1.parquet".into()],
            ..Default::default()
        },
    )
    .await
    .unwrap();

    // A *separate* instance opening the same hash sees the committed state —
    // the shared ledger survives multi-instance (unlike the file backend).
    let s2 = SyncState::open_valkey(&url, key).await.unwrap();
    let st = s2.get("lake.users");
    assert_eq!(st.snapshot_id, Some(7));
    assert_eq!(st.done_files, vec!["chunk0.parquet", "chunk1.parquet"]);
    assert!(st.synced_at.is_some(), "advance stamped synced_at");
}
