use super::{
    AedbConfig, AedbInstance, ColumnDef, ColumnType, CommitFinality, ConsistencyMode, DdlOperation,
    DurabilityMode, Expr, Mutation, Query, Row, Value, create_table,
};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tempfile::tempdir;

#[tokio::test]
async fn commit_with_visible_finality_can_return_before_durable_head_in_batch_mode() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        durability_mode: DurabilityMode::Batch,
        batch_interval_ms: 60_000,
        batch_max_bytes: usize::MAX,
        ..AedbConfig::default()
    };
    let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
    db.create_project("p").await.expect("project");

    let result = db
        .commit_with_finality(
            Mutation::KvSet {
                project_id: "p".into(),
                scope_id: "app".into(),
                key: b"fast-visible".to_vec(),
                value: b"v".to_vec(),
            },
            CommitFinality::Visible,
        )
        .await
        .expect("commit");

    assert!(
        result.durable_head_seq < result.commit_seq,
        "visible finality should not require durable head in batch mode"
    );
}

#[tokio::test]
async fn commit_with_durable_finality_waits_until_durable_head_catches_up() {
    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        durability_mode: DurabilityMode::Batch,
        batch_interval_ms: 60_000,
        batch_max_bytes: usize::MAX,
        ..AedbConfig::default()
    };
    let db = Arc::new(AedbInstance::open_anonymous(config, dir.path()).expect("open"));
    db.create_project("p").await.expect("project");

    let fsync_db = Arc::clone(&db);
    let fsync_task = tokio::spawn(async move {
        tokio::time::sleep(Duration::from_millis(20)).await;
        fsync_db.force_fsync().await.expect("force fsync");
    });

    let started = Instant::now();
    let result = db
        .commit_with_finality(
            Mutation::KvSet {
                project_id: "p".into(),
                scope_id: "app".into(),
                key: b"fast-durable".to_vec(),
                value: b"v".to_vec(),
            },
            CommitFinality::Durable,
        )
        .await
        .expect("commit");
    fsync_task.await.expect("join fsync");

    assert!(
        started.elapsed() >= Duration::from_millis(15),
        "durable finality should wait for WAL durability in batch mode"
    );
    assert!(
        result.durable_head_seq >= result.commit_seq,
        "durable finality must report durable head at or beyond commit sequence"
    );
}

#[tokio::test]
#[ignore = "long-running finality latency profile"]
async fn finality_profile_visible_vs_durable_low_latency_mode() {
    async fn run_profile(
        config: AedbConfig,
        finality: CommitFinality,
        ops: usize,
    ) -> (u64, u64, u64, crate::OperationalMetrics) {
        let dir = tempdir().expect("temp");
        let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
        db.create_project("p").await.expect("project");
        let started = Instant::now();
        let mut lat_sum = 0u128;
        let mut lat_max = 0u64;
        for i in 0..ops {
            let op_started = Instant::now();
            db.commit_with_finality(
                Mutation::KvSet {
                    project_id: "p".into(),
                    scope_id: "app".into(),
                    key: format!("finality:{finality:?}:{i}").into_bytes(),
                    value: i.to_be_bytes().to_vec(),
                },
                finality,
            )
            .await
            .expect("commit with finality");
            let us = op_started.elapsed().as_micros() as u64;
            lat_sum = lat_sum.saturating_add(us as u128);
            lat_max = lat_max.max(us);
        }
        db.force_fsync().await.expect("flush");
        let elapsed = started.elapsed().as_secs_f64().max(0.001);
        let tps = (ops as f64 / elapsed) as u64;
        let avg_us = (lat_sum / ops.max(1) as u128) as u64;
        let op = db.operational_metrics().await;
        (tps, avg_us, lat_max, op)
    }

    let ops = std::env::var("AEDB_FINALITY_PROFILE_OPS")
        .ok()
        .and_then(|v| v.parse::<usize>().ok())
        .unwrap_or(600)
        .max(200);

    let mut low_latency_no_coalesce = AedbConfig::low_latency([1u8; 32]);
    low_latency_no_coalesce.durable_ack_coalescing_enabled = false;
    low_latency_no_coalesce.durable_ack_coalesce_window_us = 0;
    let low_latency_coalesce = AedbConfig::low_latency([1u8; 32]);

    let (visible_tps, visible_avg_us, visible_max_us, visible_op) = run_profile(
        low_latency_no_coalesce.clone(),
        CommitFinality::Visible,
        ops,
    )
    .await;
    let (durable_base_tps, durable_base_avg_us, durable_base_max_us, durable_base_op) =
        run_profile(low_latency_no_coalesce, CommitFinality::Durable, ops).await;
    let (durable_tps, durable_avg_us, durable_max_us, durable_op) =
        run_profile(low_latency_coalesce, CommitFinality::Durable, ops).await;

    eprintln!(
        "finality_profile: ops={} visible_tps={} durable_base_tps={} durable_coalesced_tps={} visible_avg_us={} durable_base_avg_us={} durable_coalesced_avg_us={} visible_max_us={} durable_base_max_us={} durable_coalesced_max_us={} visible_durable_wait_ops={} durable_base_wait_ops={} durable_coalesced_wait_ops={} visible_avg_durable_wait_us={} durable_base_avg_durable_wait_us={} durable_coalesced_avg_durable_wait_us={} visible_wal_sync_ops={} durable_base_wal_sync_ops={} durable_coalesced_wal_sync_ops={} visible_avg_wal_sync_us={} durable_base_avg_wal_sync_us={} durable_coalesced_avg_wal_sync_us={} visible_avg_wal_append_us={} durable_base_avg_wal_append_us={} durable_coalesced_avg_wal_append_us={}",
        ops,
        visible_tps,
        durable_base_tps,
        durable_tps,
        visible_avg_us,
        durable_base_avg_us,
        durable_avg_us,
        visible_max_us,
        durable_base_max_us,
        durable_max_us,
        visible_op.durable_wait_ops,
        durable_base_op.durable_wait_ops,
        durable_op.durable_wait_ops,
        visible_op.avg_durable_wait_micros,
        durable_base_op.avg_durable_wait_micros,
        durable_op.avg_durable_wait_micros,
        visible_op.wal_sync_ops,
        durable_base_op.wal_sync_ops,
        durable_op.wal_sync_ops,
        visible_op.avg_wal_sync_micros,
        durable_base_op.avg_wal_sync_micros,
        durable_op.avg_wal_sync_micros,
        visible_op.avg_wal_append_micros,
        durable_base_op.avg_wal_append_micros,
        durable_op.avg_wal_append_micros
    );

    assert_eq!(
        visible_op.queue_full_rejections, 0,
        "visible finality profile should not saturate queue"
    );
    assert_eq!(
        durable_base_op.queue_full_rejections, 0,
        "durable baseline profile should not saturate queue"
    );
    assert_eq!(
        durable_op.queue_full_rejections, 0,
        "durable coalesced profile should not saturate queue"
    );
    assert_eq!(
        visible_op.timeout_rejections, 0,
        "visible finality profile should not timeout"
    );
    assert_eq!(
        durable_base_op.timeout_rejections, 0,
        "durable baseline profile should not timeout"
    );
    assert_eq!(
        durable_op.timeout_rejections, 0,
        "durable coalesced profile should not timeout"
    );
    assert_eq!(
        visible_op.durable_wait_ops, 0,
        "visible finality profile should not accumulate durable wait operations"
    );
    assert!(
        durable_op.durable_wait_ops > 0,
        "durable finality profile should accumulate durable wait operations"
    );
    assert!(
        durable_tps >= durable_base_tps.saturating_div(2),
        "coalesced durable finality regressed severely: base={durable_base_tps} coalesced={durable_tps}"
    );
    assert!(
        durable_tps <= visible_tps.saturating_mul(2),
        "durable finality profile produced implausible TPS vs visible: visible={visible_tps} durable={durable_tps}"
    );
}

#[tokio::test]
async fn commit_success_is_observable_at_its_commit_seq() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");

    let result = db
        .commit(Mutation::KvSet {
            project_id: "p".into(),
            scope_id: "app".into(),
            key: b"inclusion-proof".to_vec(),
            value: b"ok".to_vec(),
        })
        .await
        .expect("commit");

    let at_seq = db
        .kv_get_no_auth(
            "p",
            "app",
            b"inclusion-proof",
            ConsistencyMode::AtSeq(result.commit_seq),
        )
        .await
        .expect("kv_get at seq")
        .expect("value present at commit seq");
    assert_eq!(at_seq.value, b"ok".to_vec());
}

#[tokio::test]
async fn subscribe_commits_delivers_delta_after_commit() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");

    db.create_project("p").await.expect("project");
    // Subscribe AFTER setup so we observe only the deltas under test.
    let mut rx = db.subscribe_commits();

    db.commit(Mutation::KvSet {
        project_id: "p".into(),
        scope_id: "app".into(),
        key: b"hello".to_vec(),
        value: b"world".to_vec(),
    })
    .await
    .expect("commit");

    // Drain until we find our KvSet — internal subsystems may also produce
    // bookkeeping deltas, but ours must arrive within the timeout window.
    let mut saw_kv_set = false;
    let mut last_seq = 0u64;
    let deadline = std::time::Instant::now() + Duration::from_secs(2);
    while std::time::Instant::now() < deadline {
        match tokio::time::timeout(Duration::from_millis(200), rx.recv()).await {
            Ok(Ok(delta)) => {
                last_seq = last_seq.max(delta.seq);
                if delta
                    .mutations
                    .iter()
                    .any(|m| matches!(m, Mutation::KvSet { key, .. } if key == b"hello"))
                {
                    saw_kv_set = true;
                    break;
                }
            }
            Ok(Err(_)) => break,
            Err(_) => break,
        }
    }
    assert!(last_seq > 0, "broadcast delivered at least one delta");
    assert!(
        saw_kv_set,
        "broadcast delta must include the committed KvSet mutation"
    );

    db.shutdown().await.expect("shutdown");
}

#[tokio::test]
async fn subscribe_commits_lagged_subscriber_can_resume() {
    let config = AedbConfig {
        commit_broadcast_capacity: 2,
        ..AedbConfig::default()
    };
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");

    let mut rx = db.subscribe_commits();
    db.create_project("p").await.expect("project");

    for i in 0..8u8 {
        db.commit(Mutation::KvSet {
            project_id: "p".into(),
            scope_id: "app".into(),
            key: vec![i],
            value: vec![i],
        })
        .await
        .expect("commit");
    }

    let mut saw_lagged = false;
    let mut delivered = 0usize;
    let deadline = std::time::Instant::now() + Duration::from_secs(2);
    while std::time::Instant::now() < deadline {
        match tokio::time::timeout(Duration::from_millis(200), rx.recv()).await {
            Ok(Ok(_delta)) => {
                delivered += 1;
                if saw_lagged && delivered >= 2 {
                    break;
                }
            }
            Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(_))) => {
                saw_lagged = true;
            }
            Ok(Err(tokio::sync::broadcast::error::RecvError::Closed)) => break,
            Err(_) => break,
        }
    }
    assert!(
        saw_lagged,
        "slow subscriber must observe RecvError::Lagged with capacity=2 + burst"
    );

    db.commit(Mutation::KvSet {
        project_id: "p".into(),
        scope_id: "app".into(),
        key: vec![99],
        value: vec![99],
    })
    .await
    .expect("post-burst commit");

    let post = tokio::time::timeout(Duration::from_secs(1), rx.recv())
        .await
        .expect("post-burst delivery within timeout")
        .expect("post-burst delta");
    let resumed = post.mutations.iter().any(|m| {
        matches!(
            m,
            Mutation::KvSet { key, .. } if key == &[99u8]
        )
    });
    assert!(resumed, "subscriber must resume after Lagged error");

    db.shutdown().await.expect("shutdown");
}

/// #3: with `row_change_deltas_enabled`, each broadcast `CommitDelta` carries
/// resolved row-level changes — insert vs update distinguished, deletes resolved
/// from a predicate — so a subscription layer can diff without re-querying.
#[tokio::test]
async fn commit_delta_carries_resolved_row_changes_when_enabled() {
    use crate::commit::row_change::RowChangeKind;

    let dir = tempdir().expect("temp");
    let config = AedbConfig {
        row_change_deltas_enabled: true,
        ..AedbConfig::default()
    };
    let db = AedbInstance::open_anonymous(config, dir.path()).expect("open");
    db.create_project("arcana").await.expect("project");
    db.create_scope("arcana", "app").await.expect("scope");
    create_table(
        &db,
        "arcana",
        "app",
        "entities",
        vec![
            ColumnDef {
                name: "instance_id".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
            ColumnDef {
                name: "entity_id".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
            ColumnDef {
                name: "component_name".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
            ColumnDef {
                name: "data".into(),
                col_type: ColumnType::Json,
                nullable: false,
            },
        ],
        vec!["instance_id", "entity_id", "component_name"],
    )
    .await;

    let upsert = |eid: &str, comp: &str, data: &str| Mutation::Upsert {
        project_id: "arcana".into(),
        scope_id: "app".into(),
        table_name: "entities".into(),
        primary_key: vec![
            Value::Text("i1".into()),
            Value::Text(eid.into()),
            Value::Text(comp.into()),
        ],
        row: Row::from_values(vec![
            Value::Text("i1".into()),
            Value::Text(eid.into()),
            Value::Text(comp.into()),
            Value::Json(data.into()),
        ]),
    };

    // Subscribe after DDL so the first delivered delta is our first upsert.
    let mut rx = db.subscribe_commits();

    // New row → Insert, with the new values attached.
    db.commit(upsert("e1", "Pos", r#"{"x":1}"#))
        .await
        .expect("insert");
    let d = rx.recv().await.expect("delta");
    assert_eq!(d.row_changes.len(), 1);
    assert_eq!(d.row_changes[0].kind, RowChangeKind::Insert);
    assert_eq!(d.row_changes[0].table_name, "entities");
    assert!(matches!(&d.row_changes[0].new_row, Some(r) if r.values.len() == 4));

    // Same PK again → Update.
    db.commit(upsert("e1", "Pos", r#"{"x":2}"#))
        .await
        .expect("update");
    let d = rx.recv().await.expect("delta");
    assert_eq!(d.row_changes.len(), 1);
    assert_eq!(d.row_changes[0].kind, RowChangeKind::Update);

    // A second component so the despawn resolves two rows.
    db.commit(upsert("e1", "Vel", r#"{"dx":3}"#))
        .await
        .expect("insert2");
    let _ = rx.recv().await.expect("delta");

    // Despawn via a PK-prefix predicate → both component rows resolved as Delete.
    db.commit(Mutation::DeleteWhere {
        project_id: "arcana".into(),
        scope_id: "app".into(),
        table_name: "entities".into(),
        predicate: Expr::Eq("instance_id".into(), Value::Text("i1".into()))
            .and(Expr::Eq("entity_id".into(), Value::Text("e1".into()))),
        limit: None,
    })
    .await
    .expect("despawn");
    let d = rx.recv().await.expect("delta");
    assert_eq!(d.row_changes.len(), 2, "both components resolved");
    assert!(
        d.row_changes
            .iter()
            .all(|c| c.kind == RowChangeKind::Delete)
    );
    assert!(d.row_changes.iter().all(|c| c.new_row.is_none()));
}

/// #3: when the feature is off (default), deltas carry no row-level changes —
/// zero overhead, unchanged behavior.
#[tokio::test]
async fn commit_delta_row_changes_empty_by_default() {
    let dir = tempdir().expect("temp");
    let db = AedbInstance::open_anonymous(AedbConfig::default(), dir.path()).expect("open");
    db.create_project("p").await.expect("project");
    db.create_scope("p", "app").await.expect("scope");
    create_table(
        &db,
        "p",
        "app",
        "t",
        vec![
            ColumnDef {
                name: "id".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
            ColumnDef {
                name: "v".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
        ],
        vec!["id"],
    )
    .await;
    let mut rx = db.subscribe_commits();
    db.commit(Mutation::Upsert {
        project_id: "p".into(),
        scope_id: "app".into(),
        table_name: "t".into(),
        primary_key: vec![Value::Integer(1)],
        row: Row::from_values(vec![Value::Integer(1), Value::Integer(9)]),
    })
    .await
    .expect("upsert");
    let d = rx.recv().await.expect("delta");
    assert!(d.row_changes.is_empty());
}

// --- read-your-writes contract ---------------------------------------------------------------
//
// `commit()` resolves only after the epoch loop has applied the write to the visible state and
// bumped the snapshot generation (executor loop: apply → bump generation under the state lock →
// ack), so an `AtLatest` read issued after the ack MUST observe the write. Consumers build
// worker loops on this (e.g. Arcana's task queue scans for rows a just-acked commit produced);
// if a cache or pipeline change ever lets an ack overtake visibility, work gets silently
// deferred or dropped. These tests pin the contract empirically, single-writer and under
// concurrency, on both immediate and batch/coalesced durability profiles.

async fn open_ryw_instance(config: AedbConfig, dir: &std::path::Path) -> Arc<AedbInstance> {
    let db = Arc::new(AedbInstance::open_anonymous(config, dir).expect("open"));
    db.create_project("p").await.expect("project");
    db.create_scope("p", "s").await.expect("scope");
    db.commit(Mutation::Ddl(DdlOperation::CreateTable {
        project_id: "p".into(),
        scope_id: "s".into(),
        table_name: "ryw".into(),
        owner_id: None,
        if_not_exists: true,
        columns: vec![
            ColumnDef {
                name: "k".into(),
                col_type: ColumnType::Text,
                nullable: false,
            },
            ColumnDef {
                name: "v".into(),
                col_type: ColumnType::Integer,
                nullable: false,
            },
        ],
        primary_key: vec!["k".into()],
    }))
    .await
    .expect("create table");
    db
}

async fn assert_write_visible_after_ack(db: &AedbInstance, key: String, i: i64) {
    let result = db
        .commit(Mutation::Upsert {
            project_id: "p".into(),
            scope_id: "s".into(),
            table_name: "ryw".into(),
            primary_key: vec![Value::Text(key.clone().into())],
            row: Row::from_values(vec![Value::Text(key.clone().into()), Value::Integer(i)]),
        })
        .await
        .expect("commit");

    // The published read view must already include our commit...
    let probe = db
        .snapshot_probe(ConsistencyMode::AtLatest)
        .await
        .expect("probe");
    assert!(
        probe >= result.commit_seq,
        "AtLatest view seq {probe} lags acked commit_seq {} (read-your-writes violated)",
        result.commit_seq
    );

    // ...and an AtLatest query must actually see the row value we just wrote.
    let rows = db
        .query(
            "p",
            "s",
            Query::select(&["v"])
                .from("ryw")
                .where_(Expr::Eq("k".into(), Value::Text(key.clone().into())))
                .limit(1),
        )
        .await
        .expect("query")
        .rows;
    let got = rows.first().and_then(|r| match r.values.first() {
        Some(Value::Integer(n)) => Some(*n),
        _ => None,
    });
    assert_eq!(
        got,
        Some(i),
        "AtLatest query after ack of key {key} iteration {i} did not see the write"
    );
}

#[tokio::test]
async fn commit_ack_implies_read_your_writes() {
    for config in [AedbConfig::default(), AedbConfig::low_latency([5u8; 32])] {
        let dir = tempdir().expect("temp");
        let db = open_ryw_instance(config, dir.path()).await;
        for i in 0..300i64 {
            assert_write_visible_after_ack(&db, "solo".to_string(), i).await;
        }
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn concurrent_commit_acks_imply_read_your_writes() {
    let coalesced = AedbConfig {
        durable_ack_coalescing_enabled: true,
        ..AedbConfig::default()
    };
    for config in [AedbConfig::default(), coalesced] {
        let dir = tempdir().expect("temp");
        let db = open_ryw_instance(config, dir.path()).await;
        let mut handles = Vec::new();
        for w in 0..8 {
            let db = Arc::clone(&db);
            handles.push(tokio::spawn(async move {
                for i in 0..100i64 {
                    assert_write_visible_after_ack(&db, format!("w{w}"), i).await;
                }
            }));
        }
        for h in handles {
            h.await.expect("worker");
        }
    }
}
