//! Multi-process coordination tests for `KvWorkLog`.
//!
//! Two `KvWorkLog` instances share the same `Arc<dyn KvStore>` (simulating
//! two processes sharing a NATS-KV namespace). All assertions verify that
//! operations on instance A are immediately visible to instance B — exercising
//! spec § 1.10 (NATS-KV multi-process WorkLog).

use klieo_bus_memory::MemoryBus;
use klieo_ops::worklog::{KvWorkLog, WorkItem, WorkLog, WorkLogError, WorkStatus};

fn item(title: &str) -> WorkItem {
    WorkItem::new(title, serde_json::json!({}), None, vec![])
}

/// Instance A plans an item; instance B can retrieve it via `get()`.
#[tokio::test]
async fn plan_on_a_visible_on_b() {
    let bus = MemoryBus::new();
    let a = KvWorkLog::new(bus.kv.clone());
    let b = KvWorkLog::new(bus.kv.clone());

    let id = a.plan(item("shared")).await.expect("plan");
    let fetched = b.get(id).await.expect("b must see the item planned by a");
    assert_eq!(fetched.title, "shared");
    assert_eq!(fetched.status, WorkStatus::Ready);
}

/// Instance A creates edge a→b; instance B then attempts b→a.
/// Instance B's `depend()` must detect the cycle via KV-stored adjacency.
#[tokio::test]
async fn cycle_detection_across_instances() {
    let bus = MemoryBus::new();
    let a = KvWorkLog::new(bus.kv.clone());
    let b = KvWorkLog::new(bus.kv.clone());

    let id_a = a.plan(item("node_a")).await.expect("plan a");
    let id_b = a.plan(item("node_b")).await.expect("plan b");

    // A establishes node_b depends on node_a.
    a.depend(id_b.clone(), id_a.clone())
        .await
        .expect("a->b edge");

    // B attempts to add the reverse edge — must be rejected as a cycle.
    let err = b
        .depend(id_a.clone(), id_b.clone())
        .await
        .expect_err("b must detect the cycle");

    assert!(
        matches!(err, WorkLogError::CycleDetected { .. }),
        "expected CycleDetected, got {err:?}"
    );
}

/// Instance A creates parent + child (child depends on parent).
/// Instance A transitions parent to Done.
/// Instance B reads child status and sees Ready (cascade propagated via KV).
#[tokio::test]
async fn dependency_propagation_across_instances() {
    let bus = MemoryBus::new();
    let a = KvWorkLog::new(bus.kv.clone());
    let b = KvWorkLog::new(bus.kv.clone());

    let parent_id = a.plan(item("parent")).await.expect("plan parent");
    let child_item = WorkItem::new(
        "child",
        serde_json::json!({}),
        None,
        vec![parent_id.clone()],
    );
    let child_id = a.plan(child_item).await.expect("plan child");

    // Verify child starts Planned from B's perspective.
    let child_before = b.get(child_id.clone()).await.expect("b must see child");
    assert_eq!(child_before.status, WorkStatus::Planned);

    // A marks parent Done → cascade should promote child to Ready in KV.
    a.transition(parent_id, WorkStatus::Done)
        .await
        .expect("parent done");

    // B reads child status — must be Ready now.
    let child_after = b
        .get(child_id)
        .await
        .expect("b must see child after cascade");
    assert_eq!(
        child_after.status,
        WorkStatus::Ready,
        "child should be Ready after parent Done cascade"
    );
}

/// Instance A plans items; instance B can list them by status.
#[tokio::test]
async fn list_by_status_across_instances() {
    let bus = MemoryBus::new();
    let a = KvWorkLog::new(bus.kv.clone());
    let b = KvWorkLog::new(bus.kv.clone());

    for i in 0..3 {
        a.plan(item(&format!("item_{i}"))).await.expect("plan item");
    }

    let ready = b.list_by_status(WorkStatus::Ready, 10).await;
    assert_eq!(ready.len(), 3, "b must see all 3 Ready items planned by a");
}
