//! Inject-and-assert tests for the pluggable [`skade::ObjectStore`] and the
//! warehouse running over it. No live MinIO required — these exercise the
//! always-on `MemoryStore` and (under `--features rustfs`) the embedded local
//! store, and round-trip a real Iceberg table through the trait.

use std::sync::Arc;

use bytes::Bytes;
use skade::arrow_array::{Int64Array, RecordBatch, StringArray};
use skade::arrow_schema::{DataType, Field, Schema};
use skade::object_store::{MemoryStore, ObjectStore};
#[cfg(feature = "rustfs")]
use skade::object_store::ObjectStoreConfig;
use skade::Warehouse;

/// The core contract: put → get → list → exists → delete round-trips with the
/// EXACT bytes and key set we injected.
async fn roundtrip_contract(store: Arc<dyn ObjectStore>) {
    // Inject three objects with distinct, non-trivial bodies.
    let a = Bytes::from_static(b"alpha-body-0123456789");
    let b = Bytes::from(vec![7u8; 4096]);
    let c = Bytes::from_static(b"");

    store.put("ns/a.txt", a.clone()).await.unwrap();
    store.put("ns/sub/b.bin", b.clone()).await.unwrap();
    store.put("other/c", c.clone()).await.unwrap();

    // get returns exactly what we put.
    assert_eq!(store.get("ns/a.txt").await.unwrap(), a, "a body mismatch");
    assert_eq!(store.get("ns/sub/b.bin").await.unwrap(), b, "b body mismatch");
    assert_eq!(store.get("other/c").await.unwrap(), c, "empty body mismatch");

    // get_range slices exactly.
    assert_eq!(
        store.get_range("ns/a.txt", 6..11).await.unwrap().as_ref(),
        b"body-",
        "range slice wrong"
    );

    // exists is precise.
    assert!(store.exists("ns/a.txt").await.unwrap());
    assert!(!store.exists("ns/missing").await.unwrap());

    // list by prefix returns exactly the matching keys.
    let mut under_ns = store.list("ns/").await.unwrap();
    under_ns.sort();
    assert_eq!(under_ns, vec!["ns/a.txt".to_string(), "ns/sub/b.bin".to_string()]);

    let all = store.list("").await.unwrap();
    assert_eq!(all.len(), 3, "expected 3 objects total, got {all:?}");

    // size reflects the injected length.
    assert_eq!(store.size("ns/sub/b.bin").await.unwrap(), Some(4096));
    assert_eq!(store.size("ns/missing").await.unwrap(), None);

    // delete removes exactly one; a missing delete is a no-op.
    store.delete("ns/a.txt").await.unwrap();
    assert!(!store.exists("ns/a.txt").await.unwrap());
    store.delete("ns/a.txt").await.unwrap(); // idempotent
    assert_eq!(store.list("").await.unwrap().len(), 2);
}

#[tokio::test]
async fn memory_store_roundtrips() {
    roundtrip_contract(Arc::new(MemoryStore::new())).await;
}

#[cfg(feature = "rustfs")]
#[tokio::test]
async fn local_fs_store_roundtrips() {
    let dir = tempfile::tempdir().unwrap();
    let store = skade::object_store::LocalFsStore::new(dir.path()).unwrap();
    roundtrip_contract(Arc::new(store)).await;
}

fn sample_schema() -> Schema {
    Schema::new(vec![
        Field::new("id", DataType::Int64, false),
        Field::new("name", DataType::Utf8, false),
    ])
}

fn sample_batch(schema: &Schema, ids: Vec<i64>, names: Vec<&str>) -> RecordBatch {
    RecordBatch::try_new(
        Arc::new(schema.clone()),
        vec![
            Arc::new(Int64Array::from(ids)),
            Arc::new(StringArray::from(names)),
        ],
    )
    .unwrap()
}

/// The warehouse writes AND reads a real Iceberg table through the trait: an
/// in-process `MemoryStore` holds every data/metadata blob, and the bytes we
/// appended come back. This proves the `ObjectStore` → iceberg `Storage` bridge
/// is wired end-to-end (no local filesystem warehouse tree involved).
#[tokio::test]
async fn warehouse_writes_and_reads_through_memory_store() {
    let dir = tempfile::tempdir().unwrap();
    let db = dir.path().join("catalog.redb");
    let store = Arc::new(MemoryStore::new());

    let wh = Warehouse::open_with_store(&db, "s3://lake/warehouse", store.clone())
        .await
        .unwrap();

    let schema = sample_schema();
    let mut t = wh.table_or_create("events", &schema).await.unwrap();
    t.append(&[sample_batch(&schema, vec![1, 2, 3], vec!["a", "b", "c"])])
        .await
        .unwrap();

    // The data + metadata actually landed in the injected store.
    assert!(!store.is_empty(), "no blobs written to the object store");
    let keys = store.list("").await.unwrap();
    assert!(
        keys.iter().any(|k| k.ends_with(".parquet")),
        "no parquet data file in store; keys={keys:?}"
    );
    assert!(
        keys.iter().any(|k| k.contains("metadata")),
        "no metadata blob in store; keys={keys:?}"
    );

    // Read back: exact rows out.
    let back = t.read().await.unwrap();
    let rows: usize = back.iter().map(|b| b.num_rows()).sum();
    assert_eq!(rows, 3, "expected 3 rows back through the trait");
}

/// Same end-to-end Iceberg round-trip but via a serializable
/// [`ObjectStoreConfig`] (the config path Njord uses from settings), proving the
/// config-built backend honors the configured warehouse URI / store.
#[cfg(feature = "rustfs")]
#[tokio::test]
async fn warehouse_roundtrips_through_local_fs_config() {
    let cat_dir = tempfile::tempdir().unwrap();
    let store_dir = tempfile::tempdir().unwrap();
    let db = cat_dir.path().join("catalog.redb");

    let cfg = ObjectStoreConfig::LocalFs {
        root: store_dir.path().to_path_buf(),
    };
    let wh = Warehouse::open_with_object_store_config(
        &db,
        format!("file://{}", store_dir.path().display()),
        cfg,
    )
    .await
    .unwrap();

    let schema = sample_schema();
    let mut t = wh.table_or_create("logs", &schema).await.unwrap();
    t.append(&[sample_batch(&schema, vec![10, 20], vec!["x", "y"])])
        .await
        .unwrap();

    // Blobs landed under the configured store root (endpoint/root honored).
    let mut found_parquet = false;
    for entry in walkdir(store_dir.path()) {
        if entry.extension().map(|e| e == "parquet").unwrap_or(false) {
            found_parquet = true;
        }
    }
    assert!(found_parquet, "no parquet under the configured store root");

    let back = t.read().await.unwrap();
    let rows: usize = back.iter().map(|b| b.num_rows()).sum();
    assert_eq!(rows, 2);
}

/// Minimal recursive dir walk (avoid a walkdir dep for one test).
#[cfg(feature = "rustfs")]
fn walkdir(root: &std::path::Path) -> Vec<std::path::PathBuf> {
    let mut out = Vec::new();
    let mut stack = vec![root.to_path_buf()];
    while let Some(d) = stack.pop() {
        if let Ok(rd) = std::fs::read_dir(&d) {
            for e in rd.flatten() {
                let p = e.path();
                if p.is_dir() {
                    stack.push(p);
                } else {
                    out.push(p);
                }
            }
        }
    }
    out
}
