//! E-32 push: SPEC §14.4.3 and every step of §14.4.4 in order, each rejection reason of §14.4.6, batch limits and
//! push batching (AC-13.4).

use axum::http::Method;
use serde_json::{Value, json};

use crate::common::*;
use crate::sync_fixtures::*;

/// A change applied to a pushed entity.
type Mutator = Box<dyn Fn(&mut Value)>;

const T1: &str = "2026-10-17T11:00:00.000Z";
const T2: &str = "2026-10-17T11:30:00.000Z";

// ---- whole-request rules ----------------------------------------------------------------------------------------

#[tokio::test]
async fn batch_of_101_is_batch_too_large_and_0_too_and_100_is_fine() {
    let app = app().await;
    let many = |n: usize| {
        (0..n)
            .map(|_| exp(expense(&new_id(), T1)))
            .collect::<Vec<_>>()
    };
    let reply = push(&app, &admin_token(), many(101)).await;
    assert_error(&reply, 422, "batch_too_large");
    assert_eq!(count(&app, "expenses").await, 0);
    let reply = push(&app, &admin_token(), Vec::new()).await;
    assert_error(&reply, 422, "batch_too_large");
    let reply = push(&app, &admin_token(), many(100)).await;
    assert_eq!(reply.status, 200, "{}", reply.text());
    assert_eq!(reply.json["results"].as_array().unwrap().len(), 100);
    assert_eq!(count(&app, "expenses").await, 100);
}

#[tokio::test]
async fn body_over_1_mib_is_payload_too_large() {
    let app = app().await;
    let mut e = expense(&new_id(), T1);
    e["notes"] = json!("x".repeat(1024 * 1024 + 1));
    let reply = push(&app, &admin_token(), vec![exp(e)]).await;
    assert_error(&reply, 413, "payload_too_large");
}

#[tokio::test]
async fn malformed_body_or_envelope_is_invalid_request() {
    let app = app().await;
    let reply = app
        .call(
            Method::POST,
            "/sync",
            &admin_token(),
            Some(json!({"nope": []})),
        )
        .await;
    assert_error(&reply, 400, "invalid_request");
    // Without a usable mutation_id, entity_type or entity.id there is no per-mutation result to give.
    for bad in [
        json!({"mutation_id": "x", "entity_type": "expense", "entity": expense(&new_id(), T1)}),
        json!({"mutation_id": new_id(), "entity_type": "account", "entity": expense(&new_id(), T1)}),
        json!({"mutation_id": new_id(), "entity_type": "expense", "entity": {"id": 5}}),
        json!({"mutation_id": new_id().to_uppercase(), "entity_type": "expense", "entity": expense(&new_id(), T1)}),
    ] {
        let reply = push(&app, &admin_token(), vec![exp(expense(&new_id(), T1)), bad]).await;
        assert_error(&reply, 400, "invalid_request");
    }
    assert_eq!(count(&app, "expenses").await, 0);
}

#[tokio::test]
async fn step2_household_incomplete_fails_the_whole_request() {
    let app = TestApp::admin_only().await;
    let reply = push(&app, &admin_token(), vec![exp(expense(&new_id(), T1))]).await;
    assert_error(&reply, 409, "household_incomplete");
    assert_eq!(count(&app, "sync_mutations").await, 0);
}

#[tokio::test]
async fn results_come_in_request_order_with_exact_key_sets() {
    let app = app().await;
    let ids: Vec<String> = (0..3).map(|_| new_id()).collect();
    let mut bad = expense(&ids[1], T1);
    bad["amount_ore"] = json!(0);
    let mutations = vec![
        exp(expense(&ids[0], T1)),
        exp(bad),
        set(settlement(&ids[2], T1)),
    ];
    let reply = push(&app, &admin_token(), mutations.clone()).await;
    assert_eq!(reply.status, 200);
    assert_eq!(reply.json["server_time"], NOW);
    let results = reply.json["results"].as_array().unwrap();
    for (r, m) in results.iter().zip(&mutations) {
        assert_eq!(r["mutation_id"], m["mutation_id"]);
        assert_eq!(r["entity_id"], m["entity"]["id"]);
        assert_eq!(r["entity_type"], m["entity_type"]);
    }
    let keys: Vec<_> = results[0].as_object().unwrap().keys().cloned().collect();
    assert_eq!(
        keys,
        [
            "entity",
            "entity_id",
            "entity_type",
            "mutation_id",
            "server_version",
            "status"
        ]
    );
    rejected(&results[1], "validation_failed");
    accepted(&results[2]);
}

// ---- step 1: replay ---------------------------------------------------------------------------------------------

#[tokio::test]
async fn step1_replay_returns_the_stored_result_unchanged_with_no_other_effect() {
    let app = app().await;
    let id = new_id();
    let m = exp(expense(&id, T1));
    let first = push_one(&app, &admin_token(), m.clone()).await;
    let version = sequence(&app).await;
    let audits = count(&app, "audit_log").await;
    let pushed = pushes(&app).len();

    // The same mutation id with different content (and later in time) replays the first result.
    let mut changed = m.clone();
    changed["entity"]["amount_ore"] = json!(1);
    changed["entity"]["updated_at"] = json!(T2);
    app.set_now("2026-10-18T12:00:00.000Z");
    let again = push_one(&app, &admin_token(), changed).await;
    assert_eq!(again, first);
    assert_eq!(sequence(&app).await, version);
    assert_eq!(count(&app, "audit_log").await, audits);
    assert_eq!(pushes(&app).len(), pushed);
    assert_eq!(count(&app, "sync_mutations").await, 1);
    let amount: i64 = app.scalar_i64("SELECT amount_ore FROM expenses").await;
    assert_eq!(amount, 41200);

    // A rejected result replays too.
    let mut bad = exp(expense(&new_id(), T1));
    bad["entity"]["description"] = json!("");
    let r1 = push_one(&app, &admin_token(), bad.clone()).await;
    rejected(&r1, "validation_failed");
    bad["entity"]["description"] = json!("fixed");
    let r2 = push_one(&app, &admin_token(), bad).await;
    assert_eq!(r1, r2);
}

#[tokio::test]
async fn duplicate_mutation_ids_in_one_request_replay_the_first() {
    let app = app().await;
    let m = exp(expense(&new_id(), T1));
    let reply = push(&app, &admin_token(), vec![m.clone(), m]).await;
    let results = reply.json["results"].as_array().unwrap();
    assert_eq!(results[0], results[1]);
    assert_eq!(count(&app, "expenses").await, 1);
}

// ---- step 3: shape ----------------------------------------------------------------------------------------------

#[tokio::test]
async fn step3_shape_errors_are_rejected_validation_failed_per_mutation() {
    let app = app().await;
    let cases: Vec<(&str, Mutator)> = vec![
        (
            "missing field",
            Box::new(|e| {
                e.as_object_mut().unwrap().remove("description");
            }),
        ),
        (
            "missing deleted_at",
            Box::new(|e| {
                e.as_object_mut().unwrap().remove("deleted_at");
            }),
        ),
        ("wrong type", Box::new(|e| e["amount_ore"] = json!("41200"))),
        (
            "unknown enum",
            Box::new(|e| e["split_mode"] = json!("half")),
        ),
        (
            "not a uuid",
            Box::new(|e| e["account_id"] = json!("vardag")),
        ),
        (
            "uppercase uuid",
            Box::new(|e| e["account_id"] = json!(ACCOUNT_1.to_uppercase())),
        ),
        (
            "split user not a uuid",
            Box::new(|e| e["splits"][0]["user_id"] = json!("jacob")),
        ),
        (
            "timestamp form",
            Box::new(|e| e["updated_at"] = json!("2026-10-17T11:00:00Z")),
        ),
        (
            "deleted_at form",
            Box::new(|e| e["deleted_at"] = json!("yesterday")),
        ),
    ];
    for (name, mutate) in cases {
        let mut e = expense(&new_id(), T1);
        mutate(&mut e);
        let r = push_one(&app, &admin_token(), exp(e)).await;
        assert_eq!(r["reason"], "validation_failed", "{name}");
        rejected(&r, "validation_failed");
        assert_eq!(r["current"], Value::Null, "{name}");
    }
    let mut s = settlement(&new_id(), T1);
    s["allocations"][0]["account_id"] = json!(1);
    rejected(
        &push_one(&app, &admin_token(), set(s)).await,
        "validation_failed",
    );
    assert_eq!(count(&app, "expenses").await, 0);
    assert_eq!(count(&app, "sync_mutations").await, 10);
}

#[tokio::test]
async fn server_owned_fields_are_ignored_if_sent() {
    let app = app().await;
    let mut e = expense(&new_id(), T1);
    e["server_version"] = json!(999_999);
    e["created_by_user_id"] = json!(MEMBER_ID);
    e["recurring_template_id"] = json!(UNKNOWN_ID);
    e["recurring_period"] = json!("2026-10");
    let entity = accepted(&push_one(&app, &admin_token(), exp(e)).await);
    assert_eq!(entity["created_by_user_id"], ADMIN_ID);
    assert_eq!(entity["recurring_template_id"], Value::Null);
    assert_eq!(entity["recurring_period"], Value::Null);
    assert_ne!(entity["server_version"], 999_999);
}

// ---- step 4: clock ----------------------------------------------------------------------------------------------

#[tokio::test]
async fn step4_updated_at_more_than_10_minutes_ahead_is_clock_skew() {
    let app = app().await;
    let r = push_one(
        &app,
        &admin_token(),
        exp(expense(&new_id(), "2026-10-17T12:10:00.001Z")),
    )
    .await;
    rejected(&r, "clock_skew");
    assert_eq!(r["current"], Value::Null);
    let r = push_one(
        &app,
        &admin_token(),
        exp(expense(&new_id(), "2026-10-17T12:10:00.000Z")),
    )
    .await;
    accepted(&r);
    // Clock skew comes before the tombstone and validation checks: the stored row is the current.
    let id = new_id();
    push_one(&app, &admin_token(), exp(expense(&id, T1))).await;
    let mut e = expense(&id, "2026-10-18T12:00:00.000Z");
    e["amount_ore"] = json!(0);
    let r = push_one(&app, &admin_token(), exp(e)).await;
    rejected(&r, "clock_skew");
    assert_eq!(r["current"]["id"], id);
}

// ---- steps 5, 8, 9: load, validate, insert -----------------------------------------------------------------------

#[tokio::test]
async fn step9_create_inserts_with_server_fields_audit_create_and_push() {
    let app = app().await;
    let id = new_id();
    let before = sequence(&app).await;
    let mut pushed = expense(&id, T1);
    pushed["description"] = json!("  ICA Maxi  ");
    pushed["created_at"] = json!("2026-10-17T10:59:00.000Z");
    let r = push_one(&app, &member_token(), exp(pushed)).await;
    let entity = accepted(&r);
    assert_eq!(entity["description"], "ICA Maxi", "stored trimmed");
    assert_eq!(entity["created_by_user_id"], MEMBER_ID);
    assert_eq!(entity["updated_by_user_id"], MEMBER_ID);
    assert_eq!(entity["created_at"], "2026-10-17T10:59:00.000Z");
    assert_eq!(entity["updated_at"], T1);
    assert!(entity["server_version"].as_i64().unwrap() > before);
    let (device, received): (Option<String>, String) = sqlx::query_as(
        "SELECT updated_by_device_id, server_received_at FROM expenses WHERE id = ?",
    )
    .bind(&id)
    .fetch_one(&app.state.db)
    .await
    .unwrap();
    assert_eq!(device.as_deref(), Some(MEMBER_DEVICE_ID));
    assert_eq!(received, NOW);

    let audit = audit_of(&app, &id).await;
    assert_eq!(audit.len(), 1);
    let a = &audit[0];
    assert_eq!((a.action.as_str(), a.source.as_str()), ("create", "app"));
    assert_eq!(a.actor_user_id, MEMBER_ID);
    assert_eq!(a.actor_device_id.as_deref(), Some(MEMBER_DEVICE_ID));
    assert_eq!((a.client_at.as_str(), a.server_at.as_str()), (T1, NOW));
    let fields: Vec<_> = a
        .changes
        .as_array()
        .unwrap()
        .iter()
        .map(|c| c["field"].as_str().unwrap())
        .collect();
    assert_eq!(
        fields,
        [
            "account_id",
            "category_id",
            "payer_user_id",
            "amount_ore",
            "description",
            "notes",
            "expense_date",
            "split_mode",
            "splits"
        ]
    );
    assert!(
        a.changes
            .as_array()
            .unwrap()
            .iter()
            .all(|c| c["old"].is_null())
    );
    assert_eq!(a.changes[4]["new"], "ICA Maxi");
    assert_eq!(
        pushes(&app),
        [(
            "expense_created".to_owned(),
            Some(id.clone()),
            MEMBER_ID.to_owned(),
            1
        )]
    );

    let sid = new_id();
    let entity = accepted(&push_one(&app, &member_token(), set(settlement(&sid, T1))).await);
    assert_eq!(entity["allocations"].as_array().unwrap().len(), 2);
    let audit = audit_of(&app, &sid).await;
    let fields: Vec<_> = audit[0]
        .changes
        .as_array()
        .unwrap()
        .iter()
        .map(|c| c["field"].as_str().unwrap())
        .collect();
    assert_eq!(
        fields,
        [
            "from_user_id",
            "to_user_id",
            "amount_ore",
            "settlement_date",
            "note",
            "allocations"
        ]
    );
    assert_eq!(pushes(&app)[1].0, "settlement_created");
}

/// One field rule of §9.6 / §9.1 / §9.4 broken: validation_failed.
async fn assert_invalid(app: &TestApp, entity_type: &str, entity: Value, why: &str) {
    let r = push_one(app, &admin_token(), mutation(entity_type, entity)).await;
    assert_eq!(r["reason"], "validation_failed", "{why}: {r:#}");
}

#[tokio::test]
async fn step8_expense_rules_of_9_6_and_9_1() {
    let app = app().await;
    let e = || expense(&new_id(), T1);
    let with = |f: &dyn Fn(&mut Value)| {
        let mut v = e();
        f(&mut v);
        v
    };
    let cases: Vec<(&str, Value)> = vec![
        (
            "amount 0",
            with(&|v| {
                v["amount_ore"] = json!(0);
            }),
        ),
        (
            "amount over max",
            with(&|v| {
                v["amount_ore"] = json!(1_000_000_001);
                v["splits"][0]["share_ore"] = json!(500_000_001);
                v["splits"][1]["share_ore"] = json!(500_000_000);
            }),
        ),
        (
            "empty description",
            with(&|v| {
                v["description"] = json!("   ");
            }),
        ),
        (
            "description 101",
            with(&|v| {
                v["description"] = json!("é".repeat(101));
            }),
        ),
        (
            "notes 1001",
            with(&|v| {
                v["notes"] = json!("n".repeat(1001));
            }),
        ),
        (
            "notes control char",
            with(&|v| {
                v["notes"] = json!("a\tb");
            }),
        ),
        (
            "date before 2000",
            with(&|v| {
                v["expense_date"] = json!("1999-12-31");
            }),
        ),
        (
            "date over a year ahead",
            with(&|v| {
                v["expense_date"] = json!("2027-10-18");
            }),
        ),
        (
            "date invalid",
            with(&|v| {
                v["expense_date"] = json!("2026-02-30");
            }),
        ),
        (
            "share mismatch",
            with(&|v| {
                v["splits"][0]["share_ore"] = json!(20000);
                v["splits"][1]["share_ore"] = json!(21200);
            }),
        ),
        (
            "bp sum",
            with(&|v| {
                v["splits"][0]["share_bp"] = json!(6000);
            }),
        ),
        (
            "bp out of range",
            with(&|v| {
                v["splits"][0]["share_bp"] = json!(-1);
                v["splits"][1]["share_bp"] = json!(10001);
            }),
        ),
        (
            "bp missing in ratio mode",
            with(&|v| {
                v["split_mode"] = json!("ratio");
                v["splits"][0]["share_bp"] = json!(null);
            }),
        ),
        (
            "exact with bp",
            with(&|v| {
                v["split_mode"] = json!("exact");
            }),
        ),
        (
            "exact sum",
            with(&|v| {
                v["split_mode"] = json!("exact");
                v["splits"][0]["share_bp"] = json!(null);
                v["splits"][1]["share_bp"] = json!(null);
                v["splits"][1]["share_ore"] = json!(1);
            }),
        ),
        (
            "full with 50/50",
            with(&|v| {
                v["split_mode"] = json!("full");
            }),
        ),
        (
            "unknown split user",
            with(&|v| {
                v["splits"][1]["user_id"] = json!(UNKNOWN_ID);
            }),
        ),
        (
            "one split",
            with(&|v| {
                v["splits"].as_array_mut().unwrap().pop();
            }),
        ),
        (
            "three splits",
            with(&|v| {
                let first = v["splits"][0].clone();
                v["splits"].as_array_mut().unwrap().push(first);
            }),
        ),
        (
            "payer not in household",
            with(&|v| {
                v["payer_user_id"] = json!(UNKNOWN_ID);
            }),
        ),
    ];
    for (why, entity) in cases {
        assert_invalid(&app, "expense", entity, why).await;
    }
    assert_eq!(count(&app, "expenses").await, 0);
    assert_eq!(count(&app, "audit_log").await, 0);
    assert!(pushes(&app).is_empty());
}

#[tokio::test]
async fn step8_valid_edges_are_accepted() {
    let app = app().await;
    let ok = |v: Value| async { accepted(&push_one(&app, &admin_token(), exp(v)).await) };
    let mut v = expense(&new_id(), T1);
    v["expense_date"] = json!("2027-10-17"); // today (Stockholm) + 365
    v["description"] = json!("é".repeat(100));
    v["notes"] = json!("rad ett\nrad två");
    ok(v).await;
    let mut v = expense(&new_id(), T1);
    v["expense_date"] = json!("2000-01-01");
    ok(v).await;
    // Mode default: the bp are not compared with the account's current default (here 60/40 vs the account's 50/50).
    let mut v = expense(&new_id(), T1);
    v["splits"] = json!([
        {"user_id": ADMIN_ID, "share_ore": 24720, "share_bp": 6000},
        {"user_id": MEMBER_ID, "share_ore": 16480, "share_bp": 4000}
    ]);
    ok(v).await;
    // Ratio rule: the payer absorbs the remainder; exact; full; splits in any order.
    let mut v = expense(&new_id(), T1);
    v["amount_ore"] = json!(1001);
    v["split_mode"] = json!("ratio");
    v["splits"] = json!([
        {"user_id": MEMBER_ID, "share_ore": 500, "share_bp": 5000},
        {"user_id": ADMIN_ID, "share_ore": 501, "share_bp": 5000}
    ]);
    let entity = ok(v).await;
    assert_eq!(
        entity["splits"][0]["user_id"], ADMIN_ID,
        "sorted by user_id"
    );
    let mut v = expense(&new_id(), T1);
    v["split_mode"] = json!("exact");
    v["splits"] = json!([
        {"user_id": ADMIN_ID, "share_ore": 41000, "share_bp": null},
        {"user_id": MEMBER_ID, "share_ore": 200, "share_bp": null}
    ]);
    ok(v).await;
    let mut v = expense(&new_id(), T1);
    v["split_mode"] = json!("full");
    v["splits"] = json!([
        {"user_id": ADMIN_ID, "share_ore": 0, "share_bp": 0},
        {"user_id": MEMBER_ID, "share_ore": 41200, "share_bp": 10000}
    ]);
    ok(v).await;
}

#[tokio::test]
async fn step8_settlement_rules_of_9_4() {
    let app = app().await;
    let s = || settlement(&new_id(), T1);
    let with = |f: &dyn Fn(&mut Value)| {
        let mut v = s();
        f(&mut v);
        v
    };
    let cases: Vec<(&str, Value)> = vec![
        (
            "same user",
            with(&|v| {
                v["to_user_id"] = json!(MEMBER_ID);
            }),
        ),
        (
            "unknown user",
            with(&|v| {
                v["to_user_id"] = json!(UNKNOWN_ID);
            }),
        ),
        (
            "amount 0",
            with(&|v| {
                v["amount_ore"] = json!(0);
            }),
        ),
        (
            "no allocations",
            with(&|v| {
                v["allocations"] = json!([]);
            }),
        ),
        (
            "allocation 0",
            with(&|v| {
                v["allocations"][0]["amount_ore"] = json!(0);
                v["allocations"][1]["amount_ore"] = json!(50000);
            }),
        ),
        (
            "sum mismatch",
            with(&|v| {
                v["allocations"][0]["amount_ore"] = json!(1);
            }),
        ),
        (
            "duplicate account",
            with(&|v| {
                v["allocations"][1]["account_id"] = json!(ACCOUNT_1);
            }),
        ),
        (
            "51 allocations",
            with(&|v| {
                v["amount_ore"] = json!(51);
                v["allocations"] = Value::Array(
                    (0..51)
                        .map(|_| json!({"account_id": new_id(), "amount_ore": 1}))
                        .collect(),
                );
            }),
        ),
        (
            "note 201",
            with(&|v| {
                v["note"] = json!("n".repeat(201));
            }),
        ),
        (
            "date",
            with(&|v| {
                v["settlement_date"] = json!("16/10/2026");
            }),
        ),
    ];
    for (why, entity) in cases {
        assert_invalid(&app, "settlement", entity, why).await;
    }
    assert_eq!(count(&app, "settlements").await, 0);
}

#[tokio::test]
async fn step8_deleted_references_are_reference_not_found_archived_are_fine() {
    let app = app().await;
    tombstone(&app, "accounts", ACCOUNT_2).await;
    tombstone(&app, "categories", HOME).await;
    let mut e = expense(&new_id(), T1);
    e["account_id"] = json!(ACCOUNT_2);
    let r = push_one(&app, &admin_token(), exp(e)).await;
    rejected(&r, "reference_not_found");
    assert_eq!(r["message"], "account_id");
    let mut e = expense(&new_id(), T1);
    e["account_id"] = json!(UNKNOWN_ID);
    rejected(
        &push_one(&app, &admin_token(), exp(e)).await,
        "reference_not_found",
    );
    let mut e = expense(&new_id(), T1);
    e["category_id"] = json!(HOME);
    let r = push_one(&app, &admin_token(), exp(e)).await;
    rejected(&r, "reference_not_found");
    assert_eq!(r["message"], "category_id");
    let r = push_one(&app, &admin_token(), set(settlement(&new_id(), T1))).await;
    rejected(&r, "reference_not_found");
    assert_eq!(r["message"], "allocations");

    // §9.10: archived accounts and categories accept writes.
    sqlx::query("UPDATE accounts SET archived_at = ? WHERE id = ?")
        .bind(NOW)
        .bind(ACCOUNT_1)
        .execute(&app.state.db)
        .await
        .unwrap();
    sqlx::query("UPDATE categories SET archived_at = ? WHERE id = ?")
        .bind(NOW)
        .bind(GROCERIES)
        .execute(&app.state.db)
        .await
        .unwrap();
    accepted(&push_one(&app, &admin_token(), exp(expense(&new_id(), T1))).await);
}

#[tokio::test]
async fn step8_rejection_of_an_existing_row_carries_it_as_current() {
    let app = app().await;
    let id = new_id();
    let stored = accepted(&push_one(&app, &admin_token(), exp(expense(&id, T1))).await);
    let mut e = expense(&id, T2);
    e["amount_ore"] = json!(0);
    let r = push_one(&app, &admin_token(), exp(e)).await;
    rejected(&r, "validation_failed");
    assert_eq!(r["current"], stored);
}

// ---- step 6: stored tombstone -----------------------------------------------------------------------------------

#[tokio::test]
async fn step6_tombstone_accepts_a_repeated_delete_and_rejects_anything_else_as_deleted() {
    let app = app().await;
    let id = new_id();
    push_one(&app, &admin_token(), exp(expense(&id, T1))).await;
    let mut del = expense(&id, T2);
    del["deleted_at"] = json!(T2);
    let tomb = accepted(&push_one(&app, &admin_token(), exp(del.clone())).await);
    let version = sequence(&app).await;
    let audits = count(&app, "audit_log").await;
    let pushed = pushes(&app).len();

    // Idempotent delete: accepted, nothing changes.
    let mut del2 = del.clone();
    del2["deleted_at"] = json!("2026-10-17T11:45:00.000Z");
    del2["updated_at"] = json!("2026-10-17T11:45:00.000Z");
    let r = push_one(&app, &member_token(), exp(del2)).await;
    assert_eq!(accepted(&r), tomb);
    assert_eq!(sequence(&app).await, version);
    assert_eq!(count(&app, "audit_log").await, audits);
    assert_eq!(pushes(&app).len(), pushed);

    // An edit of the tombstone, even a newer one: deleted, with the tombstone as current.
    let mut edit = expense(&id, "2026-10-17T11:59:00.000Z");
    edit["amount_ore"] = json!(100);
    edit["splits"][0]["share_ore"] = json!(50);
    edit["splits"][1]["share_ore"] = json!(50);
    let r = push_one(&app, &member_token(), exp(edit)).await;
    rejected(&r, "deleted");
    assert_eq!(r["current"], tomb);
    assert_eq!(sequence(&app).await, version);
}

// ---- step 7: pushed delete -------------------------------------------------------------------------------------

#[tokio::test]
async fn step7_delete_of_a_stored_row_always_wins_and_keeps_other_fields() {
    let app = app().await;
    let id = new_id();
    let stored = accepted(&push_one(&app, &admin_token(), exp(expense(&id, T2))).await);
    // An older delete with different content: still applied; other fields keep the stored values.
    let mut del = expense(&id, T1);
    del["deleted_at"] = json!("2026-10-17T11:10:00.000Z");
    del["amount_ore"] = json!(999);
    del["description"] = json!("changed");
    let entity = accepted(&push_one(&app, &member_token(), exp(del)).await);
    assert_eq!(entity["deleted_at"], "2026-10-17T11:10:00.000Z");
    assert_eq!(entity["updated_at"], T2, "the later of stored and pushed");
    assert_eq!(entity["amount_ore"], 41200);
    assert_eq!(entity["description"], "ICA Maxi");
    assert_eq!(entity["updated_by_user_id"], MEMBER_ID);
    assert_eq!(entity["created_by_user_id"], ADMIN_ID);
    assert!(
        entity["server_version"].as_i64().unwrap() > stored["server_version"].as_i64().unwrap()
    );

    let audit = audit_of(&app, &id).await;
    assert_eq!(audit.len(), 2);
    assert_eq!(audit[1].action, "delete");
    assert_eq!(audit[1].client_at, "2026-10-17T11:10:00.000Z");
    assert_eq!(
        audit[1].changes,
        json!([{"field": "deleted_at", "old": null, "new": "2026-10-17T11:10:00.000Z"}])
    );
    assert_eq!(
        pushes(&app)[1],
        (
            "expense_deleted".to_owned(),
            Some(id),
            MEMBER_ID.to_owned(),
            1
        )
    );
    // A newer pushed updated_at is taken.
    let sid = new_id();
    push_one(&app, &admin_token(), set(settlement(&sid, T1))).await;
    let mut del = settlement(&sid, T2);
    del["deleted_at"] = json!(T2);
    let entity = accepted(&push_one(&app, &admin_token(), set(del)).await);
    assert_eq!(entity["updated_at"], T2);
    assert_eq!(pushes(&app).last().unwrap().0, "settlement_deleted");
}

#[tokio::test]
async fn step7_delete_of_a_never_synced_row_stores_a_tombstone_without_audit_or_push() {
    let app = app().await;
    let id = new_id();
    let mut del = expense(&id, T1);
    del["deleted_at"] = json!(T1);
    let entity = accepted(&push_one(&app, &admin_token(), exp(del)).await);
    assert_eq!(entity["deleted_at"], T1);
    assert_eq!(entity["created_by_user_id"], ADMIN_ID);
    assert!(audit_of(&app, &id).await.is_empty());
    assert!(pushes(&app).is_empty());
    // References are still checked.
    tombstone(&app, "accounts", ACCOUNT_2).await;
    let mut del = expense(&new_id(), T1);
    del["deleted_at"] = json!(T1);
    del["account_id"] = json!(ACCOUNT_2);
    let r = push_one(&app, &admin_token(), exp(del)).await;
    rejected(&r, "reference_not_found");
    assert_eq!(r["current"], Value::Null);
}

// ---- step 10: stale ---------------------------------------------------------------------------------------------

#[tokio::test]
async fn step10_equal_or_older_updated_at_is_stale_ties_go_to_the_server() {
    let app = app().await;
    let id = new_id();
    let stored = accepted(&push_one(&app, &admin_token(), exp(expense(&id, T2))).await);
    for at in [T2, T1] {
        let mut e = expense(&id, at);
        e["description"] = json!("Other");
        let r = push_one(&app, &member_token(), exp(e)).await;
        rejected(&r, "stale");
        assert_eq!(r["message"], "a newer version exists");
        assert_eq!(r["current"], stored);
    }
    assert_eq!(audit_of(&app, &id).await.len(), 1);
}

// ---- step 11: update --------------------------------------------------------------------------------------------

#[tokio::test]
async fn step11_newer_update_writes_the_diff_and_keeps_server_owned_fields() {
    let app = app().await;
    // A generated expense: recurring fields must survive an edit.
    let id = new_id();
    accepted(&push_one(&app, &admin_token(), exp(expense(&id, T1))).await);
    sqlx::query("INSERT INTO recurring_templates (id, account_id, category_id, amount_ore, description, notes, payer_user_id, split_mode, day_of_month, start_period, created_by_user_id, updated_by_user_id, created_at, updated_at, server_version) VALUES (?, ?, ?, 41200, 'ICA', '', ?, 'default', 1, '2026-10', ?, ?, ?, ?, 900000)")
        .bind(UNKNOWN_ID).bind(ACCOUNT_1).bind(GROCERIES).bind(ADMIN_ID).bind(ADMIN_ID).bind(ADMIN_ID).bind(NOW).bind(NOW)
        .execute(&app.state.db).await.unwrap();
    sqlx::query(
        "UPDATE expenses SET recurring_template_id = ?, recurring_period = '2026-10' WHERE id = ?",
    )
    .bind(UNKNOWN_ID)
    .bind(&id)
    .execute(&app.state.db)
    .await
    .unwrap();
    let mut e = expense(&id, T2);
    e["amount_ore"] = json!(41300);
    e["splits"][0]["share_ore"] = json!(20650);
    e["splits"][1]["share_ore"] = json!(20650);
    e["notes"] = json!("pant");
    e["created_at"] = json!("2026-01-01T00:00:00.000Z");
    let entity = accepted(&push_one(&app, &member_token(), exp(e)).await);
    assert_eq!(entity["amount_ore"], 41300);
    assert_eq!(entity["updated_at"], T2);
    assert_eq!(entity["updated_by_user_id"], MEMBER_ID);
    assert_eq!(entity["created_by_user_id"], ADMIN_ID);
    assert_eq!(entity["recurring_template_id"], UNKNOWN_ID);
    assert_eq!(entity["recurring_period"], "2026-10");
    assert_eq!(
        entity["created_at"], T1,
        "created_at keeps the stored value"
    );
    let audit = audit_of(&app, &id).await;
    assert_eq!(audit[1].action, "update");
    assert_eq!(audit[1].client_at, T2);
    assert_eq!(
        audit[1].changes,
        json!([
            {"field": "amount_ore", "old": 41200, "new": 41300},
            {"field": "notes", "old": "", "new": "pant"},
            {"field": "splits",
             "old": [{"user_id": ADMIN_ID, "share_ore": 20600, "share_bp": 5000}, {"user_id": MEMBER_ID, "share_ore": 20600, "share_bp": 5000}],
             "new": [{"user_id": ADMIN_ID, "share_ore": 20650, "share_bp": 5000}, {"user_id": MEMBER_ID, "share_ore": 20650, "share_bp": 5000}]}
        ])
    );
    assert_eq!(
        pushes(&app)[1],
        (
            "expense_updated".to_owned(),
            Some(id),
            MEMBER_ID.to_owned(),
            1
        )
    );

    // Settlement update with new allocations replaces the children.
    let sid = new_id();
    push_one(&app, &admin_token(), set(settlement(&sid, T1))).await;
    let mut s = settlement(&sid, T2);
    s["allocations"] = json!([{"account_id": ACCOUNT_2, "amount_ore": 50000}]);
    let entity = accepted(&push_one(&app, &admin_token(), set(s)).await);
    assert_eq!(
        entity["allocations"],
        json!([{"account_id": ACCOUNT_2, "amount_ore": 50000}])
    );
    let n: i64 = app
        .scalar_i64("SELECT count(*) FROM settlement_allocations")
        .await;
    assert_eq!(n, 1);
    assert_eq!(
        audit_of(&app, &sid).await[1].changes[0]["field"],
        "allocations"
    );
}

#[tokio::test]
async fn step11_empty_diff_changes_nothing_not_even_updated_at() {
    let app = app().await;
    let id = new_id();
    let stored = accepted(&push_one(&app, &admin_token(), exp(expense(&id, T1))).await);
    let version = sequence(&app).await;
    // Newer updated_at, a different created_at and splits in another order: no diff field differs.
    let mut e = expense(&id, T2);
    e["splits"].as_array_mut().unwrap().reverse();
    e["description"] = json!(" ICA Maxi ");
    let r = push_one(&app, &member_token(), exp(e)).await;
    assert_eq!(accepted(&r), stored);
    assert_eq!(sequence(&app).await, version);
    assert_eq!(audit_of(&app, &id).await.len(), 1);
    assert_eq!(pushes(&app).len(), 1);
}

// ---- step 12: sync_mutations ------------------------------------------------------------------------------------

#[tokio::test]
async fn step12_every_result_is_recorded_with_device_and_time() {
    let app = app().await;
    let ok = exp(expense(&new_id(), T1));
    let mut bad = exp(expense(&new_id(), T1));
    bad["entity"]["amount_ore"] = json!(-5);
    let reply = push(&app, &member_token(), vec![ok.clone(), bad.clone()]).await;
    let rows: Vec<(String, String, String, String, String, String)> = sqlx::query_as(
        "SELECT mutation_id, device_id, entity_type, entity_id, result, received_at FROM sync_mutations ORDER BY rowid",
    )
    .fetch_all(&app.state.db)
    .await
    .unwrap();
    assert_eq!(rows.len(), 2);
    for (row, (m, result)) in rows.iter().zip(
        [ok, bad]
            .iter()
            .zip(reply.json["results"].as_array().unwrap()),
    ) {
        assert_eq!(row.0, m["mutation_id"].as_str().unwrap());
        assert_eq!(row.1, MEMBER_DEVICE_ID);
        assert_eq!(row.2, "expense");
        assert_eq!(row.3, m["entity"]["id"].as_str().unwrap());
        assert_eq!(serde_json::from_str::<Value>(&row.4).unwrap(), *result);
        assert_eq!(row.5, NOW);
    }
}

// ---- pushes after commit, sync_batch -----------------------------------------------------------------------------

#[tokio::test]
async fn more_than_3_notifiable_changes_send_one_sync_batch_3_send_each() {
    let app = app().await;
    let three: Vec<Value> = (0..3).map(|_| exp(expense(&new_id(), T1))).collect();
    push(&app, &admin_token(), three).await;
    let p = pushes(&app);
    assert_eq!(p.len(), 3);
    assert!(p.iter().all(|m| m.0 == "expense_created" && m.3 == 1));

    // Four notifiable changes plus a rejection and a no-op (not notifiable): one sync_batch with count 4.
    let id = new_id();
    push_one(&app, &admin_token(), exp(expense(&id, T1))).await;
    let mut bad = expense(&new_id(), T1);
    bad["amount_ore"] = json!(0);
    let mut batch: Vec<Value> = (0..3).map(|_| exp(expense(&new_id(), T1))).collect();
    batch.push(set(settlement(&new_id(), T1)));
    batch.push(exp(bad));
    batch.push(exp(expense(&id, T2)));
    push(&app, &member_token(), batch).await;
    let p = pushes(&app);
    assert_eq!(p.len(), 5);
    assert_eq!(
        p[4],
        ("sync_batch".to_owned(), None, MEMBER_ID.to_owned(), 4)
    );
}

// ---- the contract examples ---------------------------------------------------------------------------------------

/// The `contracts/examples/sync/push-*` scenario: the stored rows of the examples, then each request, compared with
/// its response by value (ids, versions and times of rows created here by type only).
#[tokio::test]
async fn push_examples_reproduce_their_responses() {
    let app = app().await;
    app.set_now("2026-10-17T08:00:01.000Z");
    // Fika (exact) and the settlement stored before, as in the pull-page example.
    let fika = example("sync/push-request-delete.json")["mutations"][0]["entity"].clone();
    let mut stored = fika.clone();
    stored["updated_at"] = stored["created_at"].clone();
    stored["deleted_at"] = Value::Null;
    accepted(&push_one(&app, &member_token(), exp(stored)).await);
    let reply = app
        .call(
            Method::POST,
            "/sync",
            &admin_token(),
            Some(example("sync/push-request-delete.json")),
        )
        .await;
    assert_like_example(
        &reply.json,
        "sync/push-response-delete.json",
        &["server_version"],
    );

    app.set_now(NOW);
    let mut s = example("sync/push-response.json")["results"][1]["current"].clone();
    for key in ["created_by_user_id", "updated_by_user_id", "server_version"] {
        s.as_object_mut().unwrap().remove(key);
    }
    accepted(&push_one(&app, &member_token(), set(s)).await);
    let reply = app
        .call(
            Method::POST,
            "/sync",
            &admin_token(),
            Some(example("sync/push-request.json")),
        )
        .await;
    assert_like_example(&reply.json, "sync/push-response.json", &["server_version"]);

    tombstone(&app, "accounts", ACCOUNT_2).await;
    let reply = app
        .call(
            Method::POST,
            "/sync",
            &admin_token(),
            Some(example("sync/push-request-rejections.json")),
        )
        .await;
    assert_like_example(
        &reply.json,
        "sync/push-response-rejections.json",
        &["server_version"],
    );
}