//! SC-003 (append-only / no-upsert) and SC-004 (delete_orphans warn-once)
//! integration tests for ClickhouseSink::write_document_impl.
//!
//! Skip-if-no-DSN. Mirrors the shape of clickhouse_sink_create_table.rs.

use chunkshop::backends::ClickhouseBackend;
use chunkshop::chunker::Chunk;
use chunkshop::config::TargetConfig;
use chunkshop::sinks::ClickhouseSink;
use serde_json::json;
use tracing_test::traced_test;

const DSN_ENV: &str = "CHUNKSHOP_TEST_DSN_CH";

fn skip_if_no_dsn() -> Option<()> {
    if std::env::var(DSN_ENV).is_err() {
        eprintln!("skipping: {DSN_ENV} not set");
        return None;
    }
    Some(())
}

fn make_cfg(
    db: &str,
    table: &str,
    mode: &str,
    delete_orphans: bool,
) -> chunkshop::config::ClickhouseTargetConfig {
    let yaml = format!(
        "type: clickhouse\ndsn_env: {DSN_ENV}\ndatabase: {db}\ntable: {table}\nmode: {mode}\nhnsw: false\ndelete_orphans: {delete_orphans}"
    );
    let raw: serde_yaml_ng::Value = serde_yaml_ng::from_str(&yaml).unwrap();
    let target: TargetConfig = serde_yaml_ng::from_value(raw).unwrap();
    let TargetConfig::Clickhouse(t) = target else {
        unreachable!("expected Clickhouse variant");
    };
    t
}

fn make_chunks(doc_id: &str, n: usize) -> (Vec<Chunk>, Vec<Vec<f32>>, Vec<Vec<String>>) {
    let chunks: Vec<Chunk> = (0..n)
        .map(|i| Chunk {
            doc_id: doc_id.to_string(),
            seq_num: i,
            original_content: format!("orig {i}"),
            embedded_content: format!("emb {i}"),
            metadata: json!({}),
        })
        .collect();
    let embs: Vec<Vec<f32>> = (0..n).map(|i| vec![i as f32; 4]).collect();
    let tags: Vec<Vec<String>> = (0..n).map(|_| vec![]).collect();
    (chunks, embs, tags)
}

async fn count_rows(backend: &ClickhouseBackend, db: &str, table: &str) -> anyhow::Result<u64> {
    #[derive(clickhouse::Row, serde::Deserialize)]
    struct C {
        c: u64,
    }
    let client = backend.client().await?;
    let q = format!("SELECT count() AS c FROM `{db}`.`{table}`");
    let mut cur = client.query(&q).fetch::<C>()?;
    let r = cur
        .next()
        .await?
        .ok_or_else(|| anyhow::anyhow!("count() returned no rows"))?;
    Ok(r.c)
}

async fn drop_db(backend: &ClickhouseBackend, db: &str) -> anyhow::Result<()> {
    let client = backend.client().await?;
    client
        .query(&format!("DROP DATABASE IF EXISTS `{db}` SYNC"))
        .execute()
        .await?;
    Ok(())
}

/// SC-003: ClickHouse sink is append-only. Re-ingesting the same
/// `(doc_id, seq_num)` twice produces duplicate rows on the default
/// `MergeTree` engine — there's no upsert. Users opt into
/// `ReplacingMergeTree(created_at)` for lazy dedup at merge time.
#[tokio::test]
async fn reingest_produces_duplicate_rows_on_default_engine() -> anyhow::Result<()> {
    if skip_if_no_dsn().is_none() {
        return Ok(());
    }
    let db = "chunkshop_r4_append_only";
    let backend = ClickhouseBackend::new(DSN_ENV.to_string());
    drop_db(&backend, db).await?;

    let cfg = make_cfg(db, "chunks", "overwrite", false);
    let sink = ClickhouseSink::new(cfg, ClickhouseBackend::new(DSN_ENV.to_string()), 4);
    sink.create_table_impl().await?;

    let (chunks, embs, tags) = make_chunks("doc-1", 3);
    sink.write_document_impl("doc-1", &chunks, &embs, &tags)
        .await?;
    sink.write_document_impl("doc-1", &chunks, &embs, &tags)
        .await?;

    let n = count_rows(&backend, db, "chunks").await?;
    assert_eq!(
        n, 6,
        "SC-003: re-ingest must produce duplicate rows on default engine; got {n}"
    );

    drop_db(&backend, db).await?;
    Ok(())
}

/// SC-004: `delete_orphans: true` on ClickHouse warns exactly once per process
/// and is otherwise a no-op. The OnceLock in the sink module enforces the
/// at-most-one invariant; this test verifies that AT LEAST one warn event
/// fires inside the test's tracing span when sinks are constructed with
/// `delete_orphans=true`.
///
/// We use `tracing_test`'s injected `logs_contain` helper (function form
/// auto-generated by `#[traced_test]`, scoped to this test's span name) — the
/// same convention used by `tests/semantic_warning.rs`.
#[traced_test]
#[tokio::test]
async fn delete_orphans_warns_exactly_once() -> anyhow::Result<()> {
    if skip_if_no_dsn().is_none() {
        return Ok(());
    }

    // Construct two sinks with delete_orphans=true. Both pass through the
    // warn-once gate; the OnceLock guarantees AT MOST one warn event fires
    // per process. We assert AT LEAST one fires (the gate fires on the first
    // construction inside this test's span).
    let cfg1 = make_cfg("chunkshop_r4_warn_once_a", "chunks", "overwrite", true);
    let _sink1 = ClickhouseSink::new(cfg1, ClickhouseBackend::new(DSN_ENV.to_string()), 4);
    let cfg2 = make_cfg("chunkshop_r4_warn_once_b", "chunks", "overwrite", true);
    let _sink2 = ClickhouseSink::new(cfg2, ClickhouseBackend::new(DSN_ENV.to_string()), 4);

    assert!(
        logs_contain("delete_orphans=true on ClickHouse"),
        "expected at least one warn mentioning delete_orphans=true on ClickHouse"
    );

    Ok(())
}

#[tokio::test]
async fn query_top_k_returns_nearest_chunks() -> anyhow::Result<()> {
    if skip_if_no_dsn().is_none() {
        return Ok(());
    }
    let db = "chunkshop_r4_top_k";
    let backend = ClickhouseBackend::new(DSN_ENV.to_string());
    drop_db(&backend, db).await?;

    let cfg = make_cfg(db, "chunks", "overwrite", false);
    let sink = ClickhouseSink::new(cfg, ClickhouseBackend::new(DSN_ENV.to_string()), 4);
    sink.create_table_impl().await?;

    // Insert 3 chunks with distinct vectors. Query for [1,0,0,0], expect doc-1::0 first.
    let chunks = vec![
        Chunk { doc_id: "doc-1".into(), seq_num: 0, original_content: "a".into(), embedded_content: "a".into(), metadata: json!({}) },
        Chunk { doc_id: "doc-2".into(), seq_num: 0, original_content: "b".into(), embedded_content: "b".into(), metadata: json!({}) },
        Chunk { doc_id: "doc-3".into(), seq_num: 0, original_content: "c".into(), embedded_content: "c".into(), metadata: json!({}) },
    ];
    let embs = vec![
        vec![1.0_f32, 0.0, 0.0, 0.0],
        vec![0.0_f32, 1.0, 0.0, 0.0],
        vec![0.0_f32, 0.0, 1.0, 0.0],
    ];
    let tags = vec![vec![], vec![], vec![]];
    sink.write_document_impl("doc-1", &chunks[..1], &embs[..1], &tags[..1]).await?;
    sink.write_document_impl("doc-2", &chunks[1..2], &embs[1..2], &tags[1..2]).await?;
    sink.write_document_impl("doc-3", &chunks[2..3], &embs[2..3], &tags[2..3]).await?;

    let hits = sink.query_top_k_impl(&[1.0, 0.0, 0.0, 0.0], 3).await?;
    assert_eq!(hits.len(), 3);
    assert_eq!(hits[0].0, "doc-1", "expected doc-1 first; got {hits:?}");

    let n = sink.count_docs_impl().await?;
    assert_eq!(n, 3);

    drop_db(&backend, db).await?;
    Ok(())
}
