//! Push and pull: the changelog round-trip, its encryption, pagination, and the
//! guards that stop a push without a master key or a session.

use crate::common::*;

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

/// One Insert the way a caller builds it, with a zero clock: these tests assert
/// on what crosses the wire, never on HLC ordering.
fn insert(table: &str, row_id: &str, data: serde_json::Value) -> ChangeEntry {
    ChangeEntry {
        table: table.into(),
        op: ChangeOp::Insert,
        row_id: row_id.into(),
        timestamp: Utc::now(),
        hlc: Hlc::zero(DeviceId::nil()),
        data: Some(data),
        extra: serde_json::Map::default(),
    }
}

// ── Push / Pull with encryption ──

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

    let (client, _key) = kit.keyed();
    let device_id = DeviceId::new(Uuid::new_v4());
    let cursor = client
        .push(
            device_id,
            vec![insert("tasks", "row-1", json!({"title": "Secret task"}))],
        )
        .await
        .unwrap();

    assert_eq!(cursor, 1);

    // Verify the request body was sent with encrypted data (not plaintext)
    let body = kit.body(PUSH_PATH).await;
    let wire_data = body["changes"][0]["data"].as_str().unwrap();
    assert!(
        !wire_data.contains("Secret task"),
        "Plaintext should not appear on the wire"
    );
}

#[tokio::test]
async fn pull_decrypts_data() {
    let kit = MockKit::start().await;
    let (client, key) = kit.keyed();

    // Encrypt a value to simulate what the server would return
    let plaintext = json!({"title": "Decrypted task"});
    let encrypted = synckit_client::crypto::encrypt_json(&plaintext, &key).unwrap();

    let device_id = DeviceId::new(Uuid::new_v4());
    kit.post(PULL_PATH)
        .json(json!({
            "changes": [{
                "seq": 1,
                "device_id": device_id,
                "table": "tasks",
                "op": "INSERT",
                "row_id": "row-1",
                "timestamp": "2025-06-01T12:00:00Z",
                "data": encrypted,
            }],
            "cursor": 1,
            "has_more": false,
        }))
        .await;

    let (changes, cursor, has_more) = client.pull(device_id, 0).await.unwrap();
    assert_eq!(changes.len(), 1);
    assert_eq!(cursor, 1);
    assert!(!has_more);
    assert_eq!(changes[0].data.as_ref().unwrap(), &plaintext);
}

#[tokio::test]
async fn push_retries_on_503() {
    let kit = MockKit::start().await;
    kit.post(PUSH_PATH).code(503).once().empty().await;
    kit.post(PUSH_PATH).json(json!({"cursor": 5})).await;

    let (client, _key) = kit.keyed();
    let cursor = client
        .push(DeviceId::new(Uuid::new_v4()), vec![])
        .await
        .unwrap();
    assert_eq!(cursor, 5);
}

#[tokio::test]
async fn push_fails_immediately_on_401() {
    let kit = MockKit::start().await;
    kit.post(PUSH_PATH)
        .code(401)
        .exactly(1)
        .text("Unauthorized")
        .await;

    let (client, _key) = kit.keyed();
    let err = client
        .push(DeviceId::new(Uuid::new_v4()), vec![])
        .await
        .unwrap_err();
    assert!(matches!(err, SyncKitError::Server { status: 401, .. }));
}

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

    // First pull: has_more = true
    kit.post(PULL_PATH)
        .once()
        .json(json!({
            "changes": [],
            "cursor": 50,
            "has_more": true,
        }))
        .await;

    let (changes, cursor, has_more) = client.pull(device_id, 0).await.unwrap();
    assert!(changes.is_empty());
    assert_eq!(cursor, 50);
    assert!(has_more);

    // Second pull from cursor 50: has_more = false
    kit.post(PULL_PATH)
        .json(json!({
            "changes": [],
            "cursor": 100,
            "has_more": false,
        }))
        .await;

    let (_, cursor2, has_more2) = client.pull(device_id, 50).await.unwrap();
    assert_eq!(cursor2, 100);
    assert!(!has_more2);
}

// ── Empty changelog push ──

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

    let (client, _key) = kit.keyed();
    let cursor = client
        .push(DeviceId::new(Uuid::new_v4()), vec![])
        .await
        .unwrap();
    assert_eq!(cursor, 0);
}

// ── Large payload handling ──

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

    let (client, _key) = kit.keyed();

    // Create 1000+ change entries
    let changes: Vec<ChangeEntry> = (0..1100)
        .map(|i| {
            insert(
                "bulk_table",
                &format!("row-{i}"),
                json!({"index": i, "value": format!("data-{i}")}),
            )
        })
        .collect();

    let cursor = client
        .push(DeviceId::new(Uuid::new_v4()), changes)
        .await
        .unwrap();
    assert_eq!(cursor, 1000);
}

// ── Push without master key ──

#[tokio::test]
async fn push_with_data_fails_without_master_key() {
    let kit = MockKit::start().await;
    let client = kit.authed(); // No master key

    let err = client
        .push(
            DeviceId::new(Uuid::new_v4()),
            vec![insert("tasks", "r1", json!({"title": "test"}))],
        )
        .await
        .unwrap_err();
    assert!(
        matches!(err, SyncKitError::NoMasterKey),
        "Push with data should fail without master key: {err:?}"
    );
}

#[tokio::test]
async fn push_delete_requires_master_key() {
    // Deletes used to push without a key (no payload to encrypt). With HLC, a
    // Delete now seals its clock into an encrypted envelope, so the master key is
    // required for every push, Deletes included.
    let kit = MockKit::start().await;
    kit.post(PUSH_PATH).json(json!({"cursor": 1})).await;

    let client = kit.authed(); // No master key loaded.

    let changes = vec![ChangeEntry {
        table: "tasks".into(),
        op: ChangeOp::Delete,
        row_id: "r1".into(),
        timestamp: Utc::now(),
        hlc: Hlc::zero(DeviceId::nil()),
        data: None,
        extra: serde_json::Map::default(),
    }];

    let err = client
        .push(DeviceId::new(Uuid::new_v4()), changes)
        .await
        .unwrap_err();
    assert!(
        matches!(err, SyncKitError::NoMasterKey),
        "Delete now seals an HLC envelope and needs the key: {err:?}"
    );
}

// ── Double-push same data ──

#[tokio::test]
async fn double_push_same_data_both_succeed() {
    let kit = MockKit::start().await;

    // Server returns incrementing cursors
    kit.post(PUSH_PATH).once().json(json!({"cursor": 1})).await;
    kit.post(PUSH_PATH).json(json!({"cursor": 2})).await;

    let (client, _key) = kit.keyed();
    let entry = insert("tasks", "same-row", json!({"title": "duplicate push test"}));

    let cursor1 = client
        .push(DeviceId::new(Uuid::new_v4()), vec![entry.clone()])
        .await
        .unwrap();
    let cursor2 = client
        .push(DeviceId::new(Uuid::new_v4()), vec![entry])
        .await
        .unwrap();

    assert_eq!(cursor1, 1);
    assert_eq!(cursor2, 2);
}

// ── Pull without auth ──

#[tokio::test]
async fn pull_without_auth_returns_not_authenticated() {
    let kit = MockKit::start().await;
    let err = kit
        .client()
        .pull(DeviceId::new(Uuid::new_v4()), 0)
        .await
        .unwrap_err();
    assert!(matches!(err, SyncKitError::NotAuthenticated));
}

// ── Encryption roundtrip through push/pull (end-to-end) ──

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

    let device_id = DeviceId::new(Uuid::new_v4());
    let original_data = json!({
        "title": "End-to-end test",
        "tags": ["e2e", "encryption"],
        "nested": {"key": "value"},
        "count": 42
    });

    kit.post(PUSH_PATH).json(json!({"cursor": 1})).await;

    client
        .push(
            device_id,
            vec![insert("tasks", "e2e-row", original_data.clone())],
        )
        .await
        .unwrap();

    // Extract the encrypted data that was sent to the server
    let push_body = kit.body(PUSH_PATH).await;
    let wire_entry = &push_body["changes"][0];

    // Feed encrypted data back through pull
    kit.post(PULL_PATH)
        .json(json!({
            "changes": [{
                "seq": 1,
                "device_id": device_id,
                "table": wire_entry["table"],
                "op": wire_entry["op"],
                "row_id": wire_entry["row_id"],
                "timestamp": wire_entry["timestamp"],
                "data": wire_entry["data"],
            }],
            "cursor": 1,
            "has_more": false,
        }))
        .await;

    let (changes, _, _) = client.pull(device_id, 0).await.unwrap();
    assert_eq!(changes.len(), 1);
    assert_eq!(
        changes[0].data.as_ref().unwrap(),
        &original_data,
        "Data must survive push encryption + pull decryption"
    );
}

// ── Pagination is a transport detail, not a content one ──

/// Build `n` encrypted changes with distinguishable payloads, starting at
/// `first_seq`. Returns the wire JSON and the plaintexts they should decrypt to.
fn encrypted_changes(
    key: &[u8; 32],
    device_id: DeviceId,
    first_seq: i64,
    n: i64,
) -> (Vec<serde_json::Value>, Vec<serde_json::Value>) {
    let mut wire = Vec::new();
    let mut plain = Vec::new();
    for i in 0..n {
        let seq = first_seq + i;
        let payload = json!({"title": format!("task {seq}"), "n": seq});
        let encrypted = synckit_client::crypto::encrypt_json(&payload, key).unwrap();
        wire.push(json!({
            "seq": seq,
            "device_id": device_id,
            "table": "tasks",
            "op": "INSERT",
            "row_id": format!("row-{seq}"),
            "timestamp": "2025-06-01T12:00:00Z",
            "data": encrypted,
        }));
        plain.push(payload);
    }
    (wire, plain)
}

/// **Metamorphic relation:** the same six changes delivered as one page and as
/// three pages must decrypt to the same sequence. Pagination is a property of
/// the transport, so anything it changes about the content is a bug.
///
/// The existing `pull_with_has_more_pagination` asserts the cursor plumbing
/// against empty `changes` arrays, so it cannot see a page boundary that drops,
/// duplicates or reorders a row. This relates two runs instead of judging one,
/// which needs no expected-value table (Chen et al. 1998).
#[tokio::test]
async fn a_paginated_pull_yields_what_an_unpaginated_pull_yields() {
    let key = synckit_client::crypto::generate_master_key();
    let device_id = DeviceId::new(Uuid::new_v4());
    const TOTAL: i64 = 6;

    // Run A: one page.
    let unpaginated = {
        let kit = MockKit::start().await;
        let client = kit.authed();
        client.set_master_key_raw(key);
        let (wire, _) = encrypted_changes(&key, device_id, 1, TOTAL);
        kit.post(PULL_PATH)
            .json(json!({
                "changes": wire,
                "cursor": TOTAL,
                "has_more": false,
            }))
            .await;

        let (changes, cursor, has_more) = client.pull(device_id, 0).await.unwrap();
        assert_eq!(cursor, TOTAL);
        assert!(!has_more);
        changes
    };

    // Run B: the same changes, three pages of two, drained the way a caller
    // drains them.
    let paginated = {
        let kit = MockKit::start().await;
        let client = kit.authed();
        client.set_master_key_raw(key);
        for page in 0..3i64 {
            let first = page * 2 + 1;
            let (wire, _) = encrypted_changes(&key, device_id, first, 2);
            let cursor = first + 1;
            kit.post(PULL_PATH)
                .once()
                .json(json!({
                    "changes": wire,
                    "cursor": cursor,
                    "has_more": page < 2,
                }))
                .await;
        }

        let mut collected = Vec::new();
        let mut cursor = 0i64;
        loop {
            let (changes, next, has_more) = client.pull(device_id, cursor).await.unwrap();
            collected.extend(changes);
            cursor = next;
            if !has_more {
                break;
            }
        }
        assert_eq!(cursor, TOTAL, "the drain should end on the same cursor");
        collected
    };

    assert_eq!(
        unpaginated.len(),
        TOTAL as usize,
        "the single-page run delivered nothing, so the comparison below is vacuous"
    );
    assert_eq!(
        unpaginated.len(),
        paginated.len(),
        "pagination changed how many changes arrived: {} vs {}",
        unpaginated.len(),
        paginated.len()
    );
    for (a, b) in unpaginated.iter().zip(paginated.iter()) {
        assert_eq!(a.row_id, b.row_id, "pagination reordered or dropped a row");
        assert_eq!(a.data, b.data, "pagination changed a decrypted payload");
        assert_eq!(a.table, b.table);
    }
}

// ── The three pull variants ──
//
// `pull_inner` is covered through the base `pull` above (cursor advance,
// has_more, decryption, pagination). What each wrapper adds on top is the shape
// of the request body it posts and the shape of what it hands back, so that is
// what these pin.

/// One encrypted change on the wire, as the server would return it.
fn served_change(
    seq: i64,
    device_id: DeviceId,
    row_id: &str,
    key: &[u8; 32],
    plaintext: &serde_json::Value,
) -> serde_json::Value {
    let encrypted = synckit_client::crypto::encrypt_json(plaintext, key).unwrap();
    json!({
        "seq": seq,
        "device_id": device_id,
        "table": "tasks",
        "op": "INSERT",
        "row_id": row_id,
        "timestamp": "2025-06-01T12:00:00Z",
        "data": encrypted,
    })
}

#[tokio::test]
async fn pull_filtered_puts_tables_and_since_in_the_body() {
    let kit = MockKit::start().await;
    let (client, key) = kit.keyed();
    let device_id = DeviceId::new(Uuid::new_v4());
    let plaintext = json!({"title": "filtered"});

    kit.post(PULL_PATH)
        .json(json!({
            "changes": [served_change(7, device_id, "row-1", &key, &plaintext)],
            "cursor": 7,
            "has_more": true,
        }))
        .await;

    let since = "2025-05-01T00:00:00Z"
        .parse::<chrono::DateTime<Utc>>()
        .unwrap();
    let filter = synckit_client::PullFilter {
        tables: Some(vec!["tasks".to_string(), "notes".to_string()]),
        since: Some(since),
    };
    let (changes, cursor, has_more) = client.pull_filtered(device_id, 3, filter).await.unwrap();

    assert_eq!(cursor, 7);
    assert!(has_more);
    assert_eq!(changes.len(), 1);
    assert_eq!(changes[0].row_id, "row-1");
    assert_eq!(changes[0].data.as_ref().unwrap(), &plaintext);

    let body = kit.body(PULL_PATH).await;
    assert_eq!(body["cursor"], 3);
    assert_eq!(body["device_id"], json!(device_id));
    assert_eq!(body["tables"], json!(["tasks", "notes"]));
    assert_eq!(
        body["since"]
            .as_str()
            .unwrap()
            .parse::<chrono::DateTime<Utc>>()
            .unwrap(),
        since
    );
}

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

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

    client
        .pull_filtered(device_id, 0, synckit_client::PullFilter::default())
        .await
        .unwrap();

    let body = kit.body(PULL_PATH).await;
    assert!(
        body.get("tables").is_none(),
        "an empty table list is not sent"
    );
    assert!(body.get("since").is_none(), "an absent since is not sent");
}

#[tokio::test]
async fn pull_rich_keeps_device_id_and_seq() {
    let kit = MockKit::start().await;
    let (client, key) = kit.keyed();
    let origin = DeviceId::new(Uuid::new_v4());
    let me = DeviceId::new(Uuid::new_v4());
    let first = json!({"title": "one"});
    let second = json!({"title": "two"});

    kit.post(PULL_PATH)
        .json(json!({
            "changes": [
                served_change(41, origin, "row-a", &key, &first),
                served_change(42, origin, "row-b", &key, &second),
            ],
            "cursor": 42,
            "has_more": false,
        }))
        .await;

    let (changes, cursor, has_more) = client.pull_rich(me, 40).await.unwrap();

    assert_eq!(cursor, 42);
    assert!(!has_more);
    assert_eq!(changes.len(), 2);
    assert_eq!(changes[0].seq, 41);
    assert_eq!(changes[1].seq, 42);
    assert_eq!(changes[0].device_id, origin);
    assert_eq!(changes[1].device_id, origin);
    assert_ne!(
        changes[0].device_id, me,
        "the wrapper carries the originating device, not the puller"
    );
    assert_eq!(changes[0].entry.row_id, "row-a");
    assert_eq!(changes[0].entry.data.as_ref().unwrap(), &first);
    assert_eq!(changes[1].entry.data.as_ref().unwrap(), &second);

    let body = kit.body(PULL_PATH).await;
    assert_eq!(body["cursor"], 40);
    assert_eq!(body["device_id"], json!(me));
    assert!(body.get("tables").is_none(), "pull_rich sends no filter");
}

#[tokio::test]
async fn pull_filtered_rich_carries_both_the_filter_and_the_metadata() {
    let kit = MockKit::start().await;
    let (client, key) = kit.keyed();
    let origin = DeviceId::new(Uuid::new_v4());
    let me = DeviceId::new(Uuid::new_v4());
    let plaintext = json!({"title": "both"});

    kit.post(PULL_PATH)
        .json(json!({
            "changes": [served_change(9, origin, "row-c", &key, &plaintext)],
            "cursor": 9,
            "has_more": false,
        }))
        .await;

    let since = "2025-04-02T03:04:05Z"
        .parse::<chrono::DateTime<Utc>>()
        .unwrap();
    let filter = synckit_client::PullFilter {
        tables: Some(vec!["tasks".to_string()]),
        since: Some(since),
    };
    let (changes, cursor, has_more) = client.pull_filtered_rich(me, 8, filter).await.unwrap();

    assert_eq!(cursor, 9);
    assert!(!has_more);
    assert_eq!(changes.len(), 1);
    assert_eq!(changes[0].seq, 9);
    assert_eq!(changes[0].device_id, origin);
    assert_eq!(changes[0].entry.row_id, "row-c");
    assert_eq!(changes[0].entry.data.as_ref().unwrap(), &plaintext);

    let body = kit.body(PULL_PATH).await;
    assert_eq!(body["cursor"], 8);
    assert_eq!(body["device_id"], json!(me));
    assert_eq!(body["tables"], json!(["tasks"]));
    assert_eq!(
        body["since"]
            .as_str()
            .unwrap()
            .parse::<chrono::DateTime<Utc>>()
            .unwrap(),
        since
    );
}

#[tokio::test]
async fn pull_rich_drains_a_second_page() {
    let kit = MockKit::start().await;
    let (client, key) = kit.keyed();
    let origin = DeviceId::new(Uuid::new_v4());
    let me = DeviceId::new(Uuid::new_v4());

    kit.post(PULL_PATH)
        .once()
        .json(json!({
            "changes": [served_change(1, origin, "p1", &key, &json!({"n": 1}))],
            "cursor": 1,
            "has_more": true,
        }))
        .await;
    kit.post(PULL_PATH)
        .json(json!({
            "changes": [served_change(2, origin, "p2", &key, &json!({"n": 2}))],
            "cursor": 2,
            "has_more": false,
        }))
        .await;

    let mut cursor = 0;
    let mut collected: Vec<synckit_client::PulledChange> = Vec::new();
    loop {
        let (changes, next, has_more) = client.pull_rich(me, cursor).await.unwrap();
        collected.extend(changes);
        cursor = next;
        if !has_more {
            break;
        }
    }

    assert_eq!(cursor, 2);
    assert_eq!(collected.len(), 2, "the drain needed both round trips");
    assert_eq!(collected[0].seq, 1);
    assert_eq!(collected[1].seq, 2);
    assert_eq!(collected[0].entry.row_id, "p1");
    assert_eq!(collected[1].entry.row_id, "p2");
    assert_eq!(kit.hits(PULL_PATH).await, 2);

    // The second request resumes from the cursor the first returned.
    let bodies = kit.bodies("POST", PULL_PATH).await;
    assert_eq!(bodies[0]["cursor"], 0);
    assert_eq!(bodies[1]["cursor"], 1);
}