//! DB-layer contract tests for `db::synckit::log`: the pull side, where a
//! dropped entry is invisible.
//!
//! The change log is the sync engine's source of truth, so the failure this
//! file exists to catch is the silent one: a pull that drops an entry at a page
//! boundary, a cursor that is inclusive at one end and exclusive at the other,
//! or a filter that turns a full drain into a partial one. Every pagination
//! test here asserts the same thing from two directions, that draining the log
//! a page at a time yields exactly what one unpaginated pull yields, because
//! that equality is what a client's convergence depends on and neither half
//! alone would notice a missing row.
//!
//! Pinned here: cursor exclusivity at both ends, the empty-list table filter
//! meaning "nothing" rather than "everything", `since` being inclusive of its
//! own boundary, pull ordering being arrival order and not the client clock,
//! and the two filters composing as AND rather than OR.
//!
//! The write side of the same module is `db_synckit_log_push_layer`, and the
//! audit trail is `db_synckit_security_layer`. Neighbouring ground is
//! deliberately not repeated: `db_synckit_layer` pins append order, per-user
//! scoping, blobs and keys; `db_synckit_groups` pins group membership and the
//! structural isolation of `sync_group_log` from personal pulls.
//!
//! Delete this file and the pagination contract of the sync protocol is
//! unasserted at this layer.

use crate::harness::db::TestDb;
use crate::harness::seed_user;

use makenotwork::db::synckit;
use makenotwork::db::{SyncAppId, SyncDeviceId, UserId};
use uuid::Uuid;

/// One change tuple in the shape `push_sync_changes` and `push_group_changes`
/// both take.
type Change = (
    String,
    String,
    String,
    chrono::DateTime<chrono::Utc>,
    Option<serde_json::Value>,
);

/// Seed a sync app owned by `user`.
async fn seed_app(pool: &sqlx::PgPool, user: UserId, name: &str) -> SyncAppId {
    sqlx::query_scalar::<_, SyncAppId>(
        "INSERT INTO sync_apps (creator_id, name, api_key_hash, api_key_prefix)
         VALUES ($1, $2, $3, $4) RETURNING id",
    )
    .bind(user)
    .bind(name)
    .bind(format!("hash_{name}"))
    .bind(&name[..name.len().min(8)])
    .fetch_one(pool)
    .await
    .expect("seed sync app")
}

/// Seed a device row for a user within an app.
async fn seed_device(
    pool: &sqlx::PgPool,
    app: SyncAppId,
    user: UserId,
    name: &str,
) -> SyncDeviceId {
    sqlx::query_scalar::<_, SyncDeviceId>(
        "INSERT INTO sync_devices (app_id, user_id, device_name, platform)
         VALUES ($1, $2, $3, 'macos') RETURNING id",
    )
    .bind(app)
    .bind(user)
    .bind(name)
    .fetch_one(pool)
    .await
    .expect("seed device")
}

/// A fixed instant offset by whole hours. Fixed rather than `Utc::now()` so a
/// `since` boundary can be asserted for equality without depending on the
/// wall clock.
fn ts(offset_hours: i64) -> chrono::DateTime<chrono::Utc> {
    "2024-03-01T12:00:00Z"
        .parse::<chrono::DateTime<chrono::Utc>>()
        .expect("fixed base timestamp parses")
        + chrono::Duration::hours(offset_hours)
}

/// An INSERT change on `table` for `row`, client-stamped at `ts(offset_hours)`.
fn change_at(table: &str, row: &str, offset_hours: i64) -> Change {
    (
        table.to_string(),
        "INSERT".to_string(),
        row.to_string(),
        ts(offset_hours),
        Some(serde_json::json!({ "row": row })),
    )
}

/// An INSERT change on `table` for `row` at the base instant.
fn change(table: &str, row: &str) -> Change {
    change_at(table, row, 0)
}

/// Push `changes` as one fresh batch, returning the cursor the push reports.
async fn push(
    pool: &sqlx::PgPool,
    app: SyncAppId,
    user: UserId,
    device: SyncDeviceId,
    changes: &[Change],
) -> i64 {
    synckit::push_sync_changes(pool, app, user, device, Uuid::new_v4(), changes)
        .await
        .expect("push sync changes")
}

/// Walk the personal log `limit` entries at a time the way a client does,
/// carrying the last returned seq forward as the next cursor. Returns the
/// row ids in the order they were handed over.
///
/// The iteration guard is what makes a non-terminating cursor fail as a test
/// rather than hang the suite.
async fn drain(
    pool: &sqlx::PgPool,
    app: SyncAppId,
    user: UserId,
    limit: i64,
    tables: Option<&[String]>,
    since: Option<chrono::DateTime<chrono::Utc>>,
) -> Vec<String> {
    let mut seen: Vec<String> = Vec::new();
    let mut cursor = 0i64;
    let mut pages = 0;
    loop {
        let page =
            synckit::pull_sync_changes_filtered(pool, app, user, cursor, limit, tables, since)
                .await
                .expect("paginated pull");
        if page.is_empty() {
            break;
        }
        assert!(
            i64::try_from(page.len()).expect("page length fits i64") <= limit,
            "a page must never exceed the limit it was asked for: got {} for limit {limit}",
            page.len()
        );
        cursor = page.last().expect("non-empty page has a last entry").seq;
        seen.extend(page.into_iter().map(|e| e.row_id));
        pages += 1;
        assert!(
            pages <= 64,
            "the cursor did not terminate after 64 pages of limit {limit}; seen so far: {seen:?}"
        );
    }
    seen
}

fn names(v: &[String]) -> Vec<&str> {
    v.iter().map(String::as_str).collect()
}
// ── log: pagination and cursor termination ──────────────────────────────────

#[tokio::test]
async fn a_paginated_drain_yields_exactly_what_one_unpaginated_pull_yields() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sklogp_drain").await;
    let app = seed_app(&db.pool, user, "logdrain").await;
    let device = seed_device(&db.pool, app, user, "laptop").await;

    // Seven entries across three batches of unequal size, so the batch
    // boundaries do not line up with any of the page sizes below.
    push(
        &db.pool,
        app,
        user,
        device,
        &[change("tasks", "r0"), change("tasks", "r1")],
    )
    .await;
    push(
        &db.pool,
        app,
        user,
        device,
        &[
            change("tasks", "r2"),
            change("tasks", "r3"),
            change("tasks", "r4"),
        ],
    )
    .await;
    push(
        &db.pool,
        app,
        user,
        device,
        &[change("tasks", "r5"), change("tasks", "r6")],
    )
    .await;

    let whole = synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 1000, None, None)
        .await
        .unwrap();
    let whole_rows: Vec<String> = whole.iter().map(|e| e.row_id.clone()).collect();
    assert_eq!(
        names(&whole_rows),
        ["r0", "r1", "r2", "r3", "r4", "r5", "r6"],
        "one unpaginated pull is the reference: {whole_rows:?}"
    );

    // 2 and 3 leave a short final page; 7 is an exact multiple, which is the
    // case where a client only stops because the pull after the last full page
    // comes back empty.
    for limit in [2i64, 3, 7] {
        let paged = drain(&db.pool, app, user, limit, None, None).await;
        assert_eq!(
            paged, whole_rows,
            "draining {limit} at a time must lose and duplicate nothing: {paged:?}"
        );
    }
}

#[tokio::test]
async fn the_pull_cursor_is_exclusive_at_both_ends() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sklogp_cursor").await;
    let app = seed_app(&db.pool, user, "logcursor").await;
    let device = seed_device(&db.pool, app, user, "laptop").await;

    push(
        &db.pool,
        app,
        user,
        device,
        &[
            change("tasks", "r0"),
            change("tasks", "r1"),
            change("tasks", "r2"),
            change("tasks", "r3"),
            change("tasks", "r4"),
        ],
    )
    .await;

    let all = synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 100, None, None)
        .await
        .unwrap();
    assert_eq!(all.len(), 5, "five entries were pushed: {all:?}");

    // Resuming at the seq of r1 must hand back r2 onward. An inclusive `>=`
    // would re-deliver r1 and return four entries, which is why the count and
    // the first row id are both asserted.
    let after_second =
        synckit::pull_sync_changes_filtered(&db.pool, app, user, all[1].seq, 100, None, None)
            .await
            .unwrap();
    let rows: Vec<&str> = after_second.iter().map(|e| e.row_id.as_str()).collect();
    assert_eq!(
        rows,
        ["r2", "r3", "r4"],
        "a cursor is the last entry already seen, not the next one to send: {rows:?}"
    );

    // The cursor a client holds after a full drain returns nothing, which is
    // the only reason the drain loop above terminates.
    let at_end = synckit::pull_sync_changes_filtered(
        &db.pool,
        app,
        user,
        all.last().unwrap().seq,
        100,
        None,
        None,
    )
    .await
    .unwrap();
    assert!(
        at_end.is_empty(),
        "the final cursor must drain empty, got {at_end:?}"
    );

    // A cursor past the end (a log truncated behind a client, or a client that
    // held a cursor from another app) is empty rather than an error or a rewind.
    let past_end = synckit::pull_sync_changes_filtered(
        &db.pool,
        app,
        user,
        all.last().unwrap().seq + 5_000,
        100,
        None,
        None,
    )
    .await
    .unwrap();
    assert!(
        past_end.is_empty(),
        "a cursor beyond the highest seq returns nothing, got {past_end:?}"
    );
}

#[tokio::test]
async fn a_limit_bounds_one_page_and_a_zero_limit_returns_no_page_at_all() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sklogp_limit").await;
    let app = seed_app(&db.pool, user, "loglimit").await;
    let device = seed_device(&db.pool, app, user, "laptop").await;

    push(
        &db.pool,
        app,
        user,
        device,
        &[
            change("tasks", "r0"),
            change("tasks", "r1"),
            change("tasks", "r2"),
            change("tasks", "r3"),
            change("tasks", "r4"),
        ],
    )
    .await;

    let page = synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 3, None, None)
        .await
        .unwrap();
    let rows: Vec<&str> = page.iter().map(|e| e.row_id.as_str()).collect();
    assert_eq!(
        rows,
        ["r0", "r1", "r2"],
        "a limit takes the lowest seqs, not an arbitrary three: {rows:?}"
    );

    // A limit larger than the log is not an error and does not pad.
    let over = synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 500, None, None)
        .await
        .unwrap();
    assert_eq!(over.len(), 5, "a generous limit returns all five: {over:?}");

    // Pinned because it is a live hazard for a caller that computes a page
    // size: a zero limit is an empty page, indistinguishable from "drained",
    // so a client that ever asks for zero silently stops syncing.
    let none = synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 0, None, None)
        .await
        .unwrap();
    assert!(
        none.is_empty(),
        "a zero limit reads as a finished drain, got {none:?}"
    );
}

#[tokio::test]
async fn the_unfiltered_pull_matches_the_filtered_pull_with_no_filters() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sklogp_parity").await;
    let app = seed_app(&db.pool, user, "logparity").await;
    let device = seed_device(&db.pool, app, user, "laptop").await;

    push(
        &db.pool,
        app,
        user,
        device,
        &[
            change("tasks", "r0"),
            change("notes", "r1"),
            change("tasks", "r2"),
            change("tags", "r3"),
        ],
    )
    .await;

    let legacy = synckit::pull_sync_changes(&db.pool, app, user, 0, 100)
        .await
        .unwrap();
    let filtered = synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 100, None, None)
        .await
        .unwrap();

    let legacy_rows: Vec<&str> = legacy.iter().map(|e| e.row_id.as_str()).collect();
    let filtered_rows: Vec<&str> = filtered.iter().map(|e| e.row_id.as_str()).collect();
    assert_eq!(
        legacy_rows, filtered_rows,
        "passing no filters is documented as identical to the older pull: {legacy_rows:?} vs {filtered_rows:?}"
    );
    assert_eq!(
        legacy.iter().map(|e| e.seq).collect::<Vec<_>>(),
        filtered.iter().map(|e| e.seq).collect::<Vec<_>>(),
        "and hands back the same cursors"
    );
    assert_eq!(legacy_rows, ["r0", "r1", "r2", "r3"]);
}

#[tokio::test]
async fn pull_order_is_arrival_order_and_not_the_client_clock() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sklogp_order").await;
    let app = seed_app(&db.pool, user, "logorder").await;
    let device = seed_device(&db.pool, app, user, "laptop").await;

    // Client timestamps descend as the entries arrive, so ordering by
    // client_timestamp would reverse this list. A device with a skewed clock
    // is the real case: its entries must still replay in the order the server
    // accepted them.
    push(
        &db.pool,
        app,
        user,
        device,
        &[
            change_at("tasks", "arrived-first", 9),
            change_at("tasks", "arrived-second", 5),
            change_at("tasks", "arrived-third", 1),
        ],
    )
    .await;

    let entries = synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 100, None, None)
        .await
        .unwrap();
    let rows: Vec<&str> = entries.iter().map(|e| e.row_id.as_str()).collect();
    assert_eq!(
        rows,
        ["arrived-first", "arrived-second", "arrived-third"],
        "seq order, not client_timestamp order: {rows:?}"
    );
    assert!(
        entries.windows(2).all(|w| w[0].seq < w[1].seq),
        "and seq is strictly increasing: {:?}",
        entries.iter().map(|e| e.seq).collect::<Vec<_>>()
    );
}
// ── log: filters ────────────────────────────────────────────────────────────

#[tokio::test]
async fn the_table_filter_selects_the_named_tables_and_an_empty_list_selects_none() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sklogp_tables").await;
    let app = seed_app(&db.pool, user, "logtables").await;
    let device = seed_device(&db.pool, app, user, "laptop").await;

    push(
        &db.pool,
        app,
        user,
        device,
        &[
            change("tasks", "t0"),
            change("notes", "n0"),
            change("tasks", "t1"),
            change("tags", "g0"),
            change("notes", "n1"),
            change("tasks", "t2"),
        ],
    )
    .await;

    let tasks = vec!["tasks".to_string()];
    let picked = synckit::pull_sync_changes_filtered(
        &db.pool,
        app,
        user,
        0,
        100,
        Some(tasks.as_slice()),
        None,
    )
    .await
    .unwrap();
    let rows: Vec<&str> = picked.iter().map(|e| e.row_id.as_str()).collect();
    assert_eq!(
        rows,
        ["t0", "t1", "t2"],
        "one named table yields three of the six: {rows:?}"
    );

    // Two names, so a filter that only ever honoured the first element would
    // return three here instead of four.
    let two = vec!["tasks".to_string(), "tags".to_string()];
    let picked_two = synckit::pull_sync_changes_filtered(
        &db.pool,
        app,
        user,
        0,
        100,
        Some(two.as_slice()),
        None,
    )
    .await
    .unwrap();
    let rows_two: Vec<&str> = picked_two.iter().map(|e| e.row_id.as_str()).collect();
    assert_eq!(
        rows_two,
        ["t0", "t1", "g0", "t2"],
        "every named table is honoured, in seq order: {rows_two:?}"
    );

    // An empty list is not NULL, so it means "no table matches", not "no
    // filter". A caller that builds the list from a user's selection and lets
    // it come back empty gets nothing, not everything.
    let empty: Vec<String> = Vec::new();
    let none = synckit::pull_sync_changes_filtered(
        &db.pool,
        app,
        user,
        0,
        100,
        Some(empty.as_slice()),
        None,
    )
    .await
    .unwrap();
    assert!(
        none.is_empty(),
        "an empty table list selects nothing, got {none:?}"
    );

    // And a name nobody pushed is empty rather than a wildcard.
    let absent = vec!["ledger".to_string()];
    let missing = synckit::pull_sync_changes_filtered(
        &db.pool,
        app,
        user,
        0,
        100,
        Some(absent.as_slice()),
        None,
    )
    .await
    .unwrap();
    assert!(
        missing.is_empty(),
        "an unknown table selects nothing, got {missing:?}"
    );
}

#[tokio::test]
async fn the_since_filter_includes_an_entry_stamped_exactly_at_the_boundary() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sklogp_since").await;
    let app = seed_app(&db.pool, user, "logsince").await;
    let device = seed_device(&db.pool, app, user, "laptop").await;

    push(
        &db.pool,
        app,
        user,
        device,
        &[
            change_at("tasks", "before", -2),
            change_at("tasks", "exactly", 0),
            change_at("tasks", "after", 2),
        ],
    )
    .await;

    // ts(0) is the stamp of "exactly", so the boundary row separates `>=` from
    // `>`; "before" separates `>=` from an unfiltered pull.
    let from_boundary =
        synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 100, None, Some(ts(0)))
            .await
            .unwrap();
    let rows: Vec<&str> = from_boundary.iter().map(|e| e.row_id.as_str()).collect();
    assert_eq!(
        rows,
        ["exactly", "after"],
        "since is inclusive of its own instant and excludes what came before: {rows:?}"
    );

    let from_later =
        synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 100, None, Some(ts(3)))
            .await
            .unwrap();
    assert!(
        from_later.is_empty(),
        "a since past every entry returns nothing, got {from_later:?}"
    );

    let from_earlier =
        synckit::pull_sync_changes_filtered(&db.pool, app, user, 0, 100, None, Some(ts(-5)))
            .await
            .unwrap();
    assert_eq!(
        from_earlier.len(),
        3,
        "a since before every entry returns all three: {from_earlier:?}"
    );
}

#[tokio::test]
async fn the_table_and_since_filters_compose_as_and_not_or() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sklogp_compose").await;
    let app = seed_app(&db.pool, user, "logcompose").await;
    let device = seed_device(&db.pool, app, user, "laptop").await;

    // One row in each quadrant of (right table, right time), so an OR would
    // return three and an ignored filter would return two or four.
    push(
        &db.pool,
        app,
        user,
        device,
        &[
            change_at("tasks", "old-task", -4),
            change_at("tasks", "new-task", 4),
            change_at("notes", "old-note", -4),
            change_at("notes", "new-note", 4),
        ],
    )
    .await;

    let tasks = vec!["tasks".to_string()];
    let both = synckit::pull_sync_changes_filtered(
        &db.pool,
        app,
        user,
        0,
        100,
        Some(tasks.as_slice()),
        Some(ts(0)),
    )
    .await
    .unwrap();
    let rows: Vec<&str> = both.iter().map(|e| e.row_id.as_str()).collect();
    assert_eq!(
        rows,
        ["new-task"],
        "both filters must hold at once: {rows:?}"
    );
}

#[tokio::test]
async fn a_filtered_paginated_drain_matches_the_filtered_unpaginated_pull() {
    let db = TestDb::new().await;
    let user = seed_user(&db.pool, "sklogp_fdrain").await;
    let app = seed_app(&db.pool, user, "logfdrain").await;
    let device = seed_device(&db.pool, app, user, "laptop").await;

    // The matching entries are deliberately non-contiguous in seq, so a page
    // boundary always falls on a skipped row. This is where a filtered pull
    // loses data if the cursor is advanced by anything other than the seq of
    // the last entry actually returned.
    push(
        &db.pool,
        app,
        user,
        device,
        &[
            change("tasks", "t0"),
            change("notes", "n0"),
            change("tasks", "t1"),
            change("notes", "n1"),
            change("tasks", "t2"),
            change("notes", "n2"),
            change("tasks", "t3"),
        ],
    )
    .await;

    let tasks = vec!["tasks".to_string()];
    let whole = synckit::pull_sync_changes_filtered(
        &db.pool,
        app,
        user,
        0,
        1000,
        Some(tasks.as_slice()),
        None,
    )
    .await
    .unwrap();
    let whole_rows: Vec<String> = whole.iter().map(|e| e.row_id.clone()).collect();
    assert_eq!(
        names(&whole_rows),
        ["t0", "t1", "t2", "t3"],
        "the filtered reference is the four task rows: {whole_rows:?}"
    );

    for limit in [2i64, 3, 4] {
        let paged = drain(&db.pool, app, user, limit, Some(tasks.as_slice()), None).await;
        assert_eq!(
            paged, whole_rows,
            "a filtered drain at {limit} per page must equal the whole filtered pull: {paged:?}"
        );
    }
}