//! Concurrent use of one client: interleaved push/pull, parallel reads of the
//! session and key state, and the stress cases. These assert the absence of
//! panics and data corruption, not a particular interleaving.

use crate::common::*;

const PUSH_PATH: &str = "/api/v1/sync/push";
const PULL_PATH: &str = "/api/v1/sync/pull";

// ── Concurrent access ──

#[tokio::test]
async fn concurrent_push_pull_no_panics() {
    let kit = MockKit::start().await;
    kit.post(PUSH_PATH).json(json!({"cursor": 1})).await;
    kit.post(PULL_PATH)
        .json(json!({
            "changes": [],
            "cursor": 0,
            "has_more": false,
        }))
        .await;

    let device_id = DeviceId::new(Uuid::new_v4());
    let (client, _key) = kit.keyed();
    let client = Arc::new(client);

    let mut handles = Vec::new();
    for _ in 0..4 {
        let c = Arc::clone(&client);
        let did = device_id;
        handles.push(tokio::spawn(async move {
            let _ = c.push(did, vec![]).await;
            let _ = c.pull(did, 0).await;
        }));
    }

    for h in handles {
        h.await.unwrap(); // No panics
    }
}

// ── Concurrent operations ──

#[tokio::test]
async fn concurrent_push_operations_no_data_corruption() {
    let kit = MockKit::start().await;
    kit.post(PUSH_PATH).json(json!({"cursor": 1})).await;

    let (client, _key) = kit.keyed();
    let client = Arc::new(client);

    let mut handles = Vec::new();
    for i in 0..8 {
        let c = Arc::clone(&client);
        handles.push(tokio::spawn(async move {
            let device_id = DeviceId::new(Uuid::new_v4());
            let entry = ChangeEntry {
                table: format!("table_{i}"),
                op: ChangeOp::Insert,
                row_id: format!("row_{i}"),
                timestamp: Utc::now(),
                hlc: Hlc::zero(DeviceId::nil()),
                data: Some(json!({"index": i})),
                extra: serde_json::Map::default(),
            };
            c.push(device_id, vec![entry]).await
        }));
    }

    for h in handles {
        let result = h.await.unwrap();
        assert!(result.is_ok(), "Concurrent push should succeed: {result:?}");
    }
}

#[tokio::test]
async fn concurrent_push_and_pull_interleaved() {
    let kit = MockKit::start().await;
    let device_id = DeviceId::new(Uuid::new_v4());

    kit.post(PUSH_PATH).json(json!({"cursor": 10})).await;
    kit.post(PULL_PATH)
        .json(json!({
            "changes": [],
            "cursor": 10,
            "has_more": false,
        }))
        .await;

    let (client, _key) = kit.keyed();
    let client = Arc::new(client);

    let mut handles = Vec::new();
    for i in 0..4 {
        let c = Arc::clone(&client);
        let did = device_id;
        handles.push(tokio::spawn(async move {
            // Alternate push and pull
            if i % 2 == 0 {
                c.push(did, vec![]).await.map(|_| ())
            } else {
                c.pull(did, 0).await.map(|_| ())
            }
        }));
    }

    for h in handles {
        let result = h.await.unwrap();
        assert!(
            result.is_ok(),
            "Interleaved push/pull should succeed: {result:?}"
        );
    }
}

// ── Concurrency stress tests ──

#[tokio::test]
async fn concurrent_session_info_reads() {
    let kit = MockKit::start().await;
    let client = Arc::new(kit.authed());

    let mut handles = Vec::new();
    for _ in 0..50 {
        let c = Arc::clone(&client);
        handles.push(tokio::spawn(async move { c.session_info() }));
    }

    for h in handles {
        let info = h.await.unwrap();
        assert!(
            info.is_some(),
            "All concurrent reads should see the session"
        );
    }
}

#[tokio::test]
async fn concurrent_has_master_key_reads() {
    let kit = MockKit::start().await;
    let (client, _key) = kit.keyed();
    let client = Arc::new(client);

    let mut handles = Vec::new();
    for _ in 0..50 {
        let c = Arc::clone(&client);
        handles.push(tokio::spawn(async move { c.has_master_key() }));
    }

    for h in handles {
        let has_key = h.await.unwrap();
        assert!(has_key, "All concurrent reads should see the master key");
    }
}

#[tokio::test]
async fn concurrent_status_checks() {
    let kit = MockKit::start().await;
    kit.get("/api/v1/sync/status")
        .json(json!({"total_changes": 5, "latest_cursor": 3}))
        .await;

    let client = Arc::new(kit.authed());

    let mut handles = Vec::new();
    for _ in 0..20 {
        let c = Arc::clone(&client);
        handles.push(tokio::spawn(async move { c.status().await }));
    }

    for h in handles {
        let result = h.await.unwrap();
        assert!(
            result.is_ok(),
            "All concurrent status checks should succeed: {result:?}"
        );
        assert_eq!(result.unwrap().total_changes, 5);
    }
}

#[tokio::test]
async fn concurrent_push_100_entries_each() {
    let kit = MockKit::start().await;
    kit.post(PUSH_PATH).json(json!({"cursor": 1})).await;

    let (client, _key) = kit.keyed();
    let client = Arc::new(client);

    let mut handles = Vec::new();
    for batch in 0..4 {
        let c = Arc::clone(&client);
        handles.push(tokio::spawn(async move {
            let changes: Vec<ChangeEntry> = (0..100)
                .map(|i| ChangeEntry {
                    table: format!("batch_{batch}"),
                    op: ChangeOp::Insert,
                    row_id: format!("row_{i}"),
                    timestamp: Utc::now(),
                    hlc: Hlc::zero(DeviceId::nil()),
                    data: Some(json!({"index": i})),
                    extra: serde_json::Map::default(),
                })
                .collect();
            c.push(DeviceId::new(Uuid::new_v4()), changes).await
        }));
    }

    for h in handles {
        let result = h.await.unwrap();
        assert!(
            result.is_ok(),
            "Concurrent 100-entry push should succeed: {result:?}"
        );
    }
}