kvitt/server/crates/kvitt-server/tests/sync_api/pull.rs

//! E-31 pull (SPEC §14.4.2): cursor rules, paging, ordering, tombstones, children with their parent, the epoch.

use axum::http::Method;
use serde_json::Value;

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

const T1: &str = "2026-10-17T11:00:00.000Z";
const ARRAYS: [&str; 7] = [
    "users",
    "categories",
    "accounts",
    "recurring_templates",
    "expenses",
    "settlements",
    "audit_entries",
];

fn versions(page: &Value) -> Vec<i64> {
    let mut all: Vec<i64> = ARRAYS
        .iter()
        .flat_map(|a| {
            page[*a]
                .as_array()
                .unwrap()
                .iter()
                .map(|r| r["server_version"].as_i64().unwrap())
        })
        .collect();
    all.sort_unstable();
    all
}

#[tokio::test]
async fn since_greater_than_the_sequence_is_cursor_ahead_equal_is_an_empty_page() {
    let app = app().await;
    let current = sequence(&app).await;
    let reply = pull(&app, &admin_token(), current + 1, None).await;
    assert_error(&reply, 409, "cursor_ahead");
    let reply = pull(&app, &admin_token(), current, None).await;
    assert_eq!(reply.status, 200);
    assert_like_example(
        &reply.json,
        "sync/pull-empty-page.json",
        &["epoch", "since", "next_since", "server_time"],
    );
    assert_eq!(reply.json["next_since"], current);
    assert_eq!(reply.json["has_more"], false);
}

#[tokio::test]
async fn query_parameters_are_checked() {
    let app = app().await;
    for path in [
        "/sync",
        "/sync?since=abc",
        "/sync?since=-1",
        "/sync?since=0&limit=0",
        "/sync?since=0&limit=1001",
        "/sync?since=0&limit=x",
    ] {
        let reply = app.call(Method::GET, path, &admin_token(), None).await;
        assert_error(&reply, 400, "invalid_request");
    }
    assert_eq!(pull(&app, &admin_token(), 0, Some(1000)).await.status, 200);
    assert_eq!(pull(&app, &admin_token(), 0, Some(1)).await.status, 200);
}

#[tokio::test]
async fn the_first_page_has_everything_in_version_order_with_the_epoch_and_server_time() {
    let app = app().await;
    let id = new_id();
    push_one(&app, &admin_token(), exp(expense(&id, T1))).await;
    push_one(&app, &admin_token(), set(settlement(&new_id(), T1))).await;
    let reply = pull(&app, &member_token(), 0, None).await;
    assert_eq!(reply.status, 200);
    let page = &reply.json;
    let epoch: String = sqlx::query_scalar("SELECT value FROM meta WHERE key = 'sync_epoch'")
        .fetch_one(&app.state.db)
        .await
        .unwrap();
    assert_eq!(page["epoch"], epoch);
    assert_eq!(page["since"], 0);
    assert_eq!(page["server_time"], NOW);
    assert_eq!(page["has_more"], false);
    assert_eq!(page["next_since"], sequence(&app).await);
    let counts: Vec<usize> = ARRAYS
        .iter()
        .map(|a| page[*a].as_array().unwrap().len())
        .collect();
    // 2 users, 14 default categories, 2 accounts, no templates, 1 expense, 1 settlement, 2 audit entries.
    assert_eq!(counts, [2, 14, 2, 0, 1, 1, 2]);
    for a in ARRAYS {
        let v: Vec<i64> = page[a]
            .as_array()
            .unwrap()
            .iter()
            .map(|r| r["server_version"].as_i64().unwrap())
            .collect();
        let mut sorted = v.clone();
        sorted.sort_unstable();
        assert_eq!(v, sorted, "{a} sorted by server_version");
    }
    // Children come with the parent.
    assert_eq!(page["expenses"][0]["splits"].as_array().unwrap().len(), 2);
    assert_eq!(
        page["accounts"][0]["default_shares"]
            .as_array()
            .unwrap()
            .len(),
        2
    );
    assert_eq!(page["audit_entries"][0]["entity_id"], id);
    let keys: Vec<_> = page["audit_entries"][0]
        .as_object()
        .unwrap()
        .keys()
        .cloned()
        .collect();
    assert_eq!(
        keys,
        [
            "action",
            "actor_user_id",
            "changes",
            "client_at",
            "entity_id",
            "entity_type",
            "id",
            "server_at",
            "server_version",
            "source"
        ]
    );
}

#[tokio::test]
async fn paging_walks_every_row_once_parents_with_children_count_as_one() {
    let app = app().await;
    for _ in 0..9 {
        push_one(&app, &admin_token(), exp(expense(&new_id(), T1))).await;
    }
    // 2 users + 14 categories + 2 accounts + 9 expenses + 9 audit entries = 36 rows.
    let mut since = 0;
    let mut seen = Vec::new();
    let mut pages = 0;
    loop {
        let reply = pull(&app, &admin_token(), since, Some(5)).await;
        let page = &reply.json;
        let v = versions(page);
        assert!(v.len() <= 5);
        assert_eq!(page["since"], since);
        assert_eq!(page["next_since"], v.last().copied().unwrap_or(since));
        assert!(v.iter().all(|x| *x > since));
        seen.extend(v);
        since = page["next_since"].as_i64().unwrap();
        pages += 1;
        if page["has_more"] == false {
            break;
        }
    }
    assert_eq!(pages, 8);
    assert_eq!(seen.len(), 36);
    let mut dedup = seen.clone();
    dedup.dedup();
    assert_eq!(dedup, seen);
    // The final empty page.
    let reply = pull(&app, &admin_token(), since, Some(5)).await;
    assert_eq!(reply.json["next_since"], since);
    assert_eq!(reply.json["has_more"], false);
    // has_more is false when the page ends exactly on the last row.
    let reply = pull(&app, &admin_token(), 0, Some(36)).await;
    assert_eq!(reply.json["has_more"], false);
    let reply = pull(&app, &admin_token(), 0, Some(35)).await;
    assert_eq!(reply.json["has_more"], true);
}

#[tokio::test]
async fn the_default_limit_is_500() {
    let app = app().await;
    for _ in 0..500 {
        app.insert_expense(ACCOUNT_1, GROCERIES, 100, ADMIN_ID, (50, 50), false)
            .await;
    }
    let reply = pull(&app, &admin_token(), 0, None).await;
    assert_eq!(versions(&reply.json).len(), 500);
    assert_eq!(reply.json["has_more"], true);
}

#[tokio::test]
async fn tombstones_are_pulled_and_updated_rows_move_to_their_new_version() {
    let app = app().await;
    let id = new_id();
    push_one(&app, &admin_token(), exp(expense(&id, T1))).await;
    let cursor = sequence(&app).await;
    let mut del = expense(&id, "2026-10-17T11:30:00.000Z");
    del["deleted_at"] = serde_json::json!("2026-10-17T11:30:00.000Z");
    push_one(&app, &admin_token(), exp(del)).await;
    let page = pull(&app, &member_token(), cursor, None).await.json;
    assert_eq!(page["expenses"].as_array().unwrap().len(), 1);
    assert_eq!(
        page["expenses"][0]["deleted_at"],
        "2026-10-17T11:30:00.000Z"
    );
    assert_eq!(page["audit_entries"][0]["action"], "delete");
    // A full pull holds the row once, at its new version.
    let page = pull(&app, &member_token(), 0, None).await.json;
    assert_eq!(page["expenses"].as_array().unwrap().len(), 1);
}

#[tokio::test]
async fn an_epoch_change_is_visible_in_pull() {
    let app = app().await;
    let before = pull(&app, &admin_token(), 0, Some(1)).await.json["epoch"].clone();
    let new = kvitt_server::cli::rotate_epoch::rotate(&app.state.db)
        .await
        .unwrap();
    let after = pull(&app, &admin_token(), 0, Some(1)).await.json["epoch"].clone();
    assert_ne!(before, after);
    assert_eq!(after, new);
    assert_eq!(new.len(), 32);
    assert!(
        new.chars()
            .all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase())
    );
}