tablo-core 0.10.0

use toasty::stmt::List;
use topcoat::context::Cx;

use super::*;
use crate::{
    ComputedColumn, TextColumn,
    dom::ignored,
    lens,
    table::{SelectFilter, Sort, TablePage, TableState, TernaryFilter},
    test_support::{User, cx, memory_db},
};

#[derive(Debug, Clone, toasty::Model, crate::ModelColumns)]
pub(super) struct Task {
    #[key]
    #[auto]
    pub(super) id: uuid::Uuid,
    pub(super) title: String,
    pub(super) status: String,
    pub(super) featured: bool,
    pub(super) created_at: jiff::Timestamp,
}

fn status_table() -> Table<Task> {
    Table::<Task>::new(TextColumn::new(lens!(Task.title))).filters(SelectFilter::new(
        Task::fields().status(),
        vec!["published".to_string(), "draft".to_string()],
    ))
}

fn filters_state(pairs: &[(&str, &str)]) -> TableState {
    TableState {
        filters: pairs
            .iter()
            .map(|(k, v)| (k.to_string(), (*v).into()))
            .collect(),
        ..TableState::default()
    }
}

/// A search matches a substring of any searchable column, and a blank term or a table with no
/// searchable column matches nothing.
#[tokio::test]
async fn a_search_ors_its_term_across_the_searchable_columns() {
    let mut db = memory_db(toasty::models!(Task)).await;
    for (title, status) in [
        ("Ada's notes", "draft"),
        ("Plans", "for Ada"),
        ("Other", "draft"),
    ] {
        toasty::create!(Task {
            title,
            status,
            featured: false,
            created_at: jiff::Timestamp::UNIX_EPOCH,
        })
        .exec(&mut db)
        .await
        .unwrap();
    }
    let table = Table::<Task>::new((
        TextColumn::new(lens!(Task.title)).searchable(),
        TextColumn::new(lens!(Task.status)).searchable(),
    ));
    let mut titles: Vec<String> = Task::filter(table.search_expr("Ada").expect("a term searches"))
        .exec(&mut db)
        .await
        .unwrap()
        .into_iter()
        .map(|task| task.title)
        .collect();
    titles.sort();
    assert_eq!(titles, ["Ada's notes", "Plans"], "either column may match");
    assert!(table.search_expr("").is_none());
    assert!(table.search_expr("   ").is_none());
    let unsearchable = Table::<Task>::new(TextColumn::new(lens!(Task.title)));
    assert!(unsearchable.search_expr("Ada").is_none());
}

#[test]
fn page_size_records_a_zero_page_size() {
    let errors = Table::<User>::new(TextColumn::new(lens!(User.name)))
        .page_size(0)
        .declaration_errors();
    assert_eq!(errors, [DeclarationErrorKind::ZeroPageSize]);
}

/// The user names in the order `table` sorts them for `state`.
async fn names_in_order(
    db: &mut toasty::Db,
    table: &Table<User>,
    state: &TableState,
) -> Vec<String> {
    table
        .order_bys_for(state)
        .iter()
        .fold(User::all(), |query, order| query.order_by(order.clone()))
        .exec(db)
        .await
        .unwrap()
        .into_iter()
        .map(|user| user.name)
        .collect()
}

/// The sort parameter orders by its column in its direction; a column the table does not sort
/// falls back to the first sortable column, and a table with none orders by primary key.
#[tokio::test]
async fn the_sort_parameter_orders_rows_and_falls_back_to_a_stable_order() {
    let mut db = memory_db(toasty::models!(User)).await;
    for name in ["Bob", "Ada", "Cara"] {
        toasty::create!(User { name }).exec(&mut db).await.unwrap();
    }
    let sorted = Table::<User>::new(TextColumn::new(lens!(User.name)).sortable());
    let sort = |column: &str, descending| TableState {
        sort: Some(Sort {
            column: column.to_string(),
            descending,
        }),
        ..TableState::default()
    };
    assert_eq!(
        names_in_order(&mut db, &sorted, &sort("name", true)).await,
        ["Cara", "Bob", "Ada"]
    );
    assert_eq!(
        names_in_order(&mut db, &sorted, &TableState::default()).await,
        ["Ada", "Bob", "Cara"],
        "no parameter sorts by the first sortable column"
    );
    assert_eq!(
        names_in_order(&mut db, &sorted, &sort("nope", true)).await,
        ["Ada", "Bob", "Cara"],
        "an unknown column is ignored, direction included"
    );

    let unsorted = Table::<User>::new(TextColumn::new(lens!(User.name)));
    let mut by_key = User::all().exec(&mut db).await.unwrap();
    by_key.sort_by_key(|user| user.id);
    assert_eq!(
        names_in_order(&mut db, &unsorted, &TableState::default()).await,
        by_key.into_iter().map(|user| user.name).collect::<Vec<_>>(),
        "a table with no sortable column orders by primary key"
    );
}

/// The titles in the order `table` sorts the tasks for `state`.
pub(super) async fn titles_in_order(
    db: &mut toasty::Db,
    table: &Table<Task>,
    state: &TableState,
) -> Vec<String> {
    table
        .order_bys_for(state)
        .iter()
        .fold(Task::all(), |query, order| query.order_by(order.clone()))
        .exec(db)
        .await
        .unwrap()
        .into_iter()
        .map(|task| task.title)
        .collect()
}

/// An offset page orders by its sort, then by primary key, so rows sharing a sort value keep one
/// order across pages; a table ordered by its key alone does not repeat it.
#[test]
fn an_offset_page_breaks_sort_ties_by_primary_key() {
    let key = crate::toasty_compat::pk::pk_order_bys::<User>();
    let sorted = Table::<User>::new(TextColumn::new(lens!(User.name)).sortable());
    let state = TableState::default();
    let mut expected = sorted.order_bys_for(&state);
    expected.extend(key.clone());
    assert_eq!(sorted.offset_order_bys_for(&state), expected);

    let unsorted = Table::<User>::new(TextColumn::new(lens!(User.name)));
    assert_eq!(unsorted.offset_order_bys_for(&state), key);
}

#[tokio::test]
async fn table_page_round_trips_real_cursors() {
    let mut db = memory_db(toasty::models!(User)).await;
    for name in ["Ada", "Bob", "Cara"] {
        toasty::create!(User { name }).exec(&mut db).await.unwrap();
    }
    let cx = Cx::default().with(db);
    let users_table = Table::<User>::new(TextColumn::new(lens!(User.name)).sortable());
    let mut db = crate::db::db(&cx).unwrap();

    // Page 1 of 1-per-page.
    let page1 = users_table
        .order_bys_for(&TableState::default())
        .iter()
        .fold(User::all(), |q, ord| q.order_by(ord.clone()))
        .paginate(1)
        .exec(&mut db)
        .await
        .unwrap();
    let tp1 = TablePage::from_toasty_page(page1).unwrap();
    assert_eq!(tp1.rows.len(), 1);
    assert_eq!(tp1.rows[0].name, "Ada");
    let cursor = tp1.next_cursor.expect("full page has a next cursor");

    let tp1_decoded = crate::toasty_compat::cursor::decode(&cursor).unwrap();
    let page2 = User::all()
        .order_by(User::fields().name().asc())
        .paginate(1)
        .after(tp1_decoded)
        .exec(&mut db)
        .await
        .unwrap();
    let tp2 = TablePage::from_toasty_page(page2).unwrap();
    assert_eq!(tp2.rows[0].name, "Bob", "cursor must resume after Ada");
}

#[test]
fn unapplied_filters_flags_unknown_keys_and_rejected_values() {
    let tbl = status_table();
    assert!(
        tbl.unapplied_filters(&cx(), &filters_state(&[]), &[])
            .is_empty()
    );
    assert!(
        tbl.unapplied_filters(&cx(), &filters_state(&[("status", "published")]), &[])
            .is_empty(),
        "valid filter must apply"
    );
    assert_eq!(
        tbl.unapplied_filters(&cx(), &filters_state(&[("stauts", "published")]), &[]),
        vec![("f.stauts=published".to_string(), ignored::UNKNOWN)]
    );
    assert_eq!(
        tbl.unapplied_filters(&cx(), &filters_state(&[("status", "Published")]), &[]),
        vec![("f.status=Published".to_string(), ignored::INVALID)]
    );
}

#[test]
fn unapplied_filters_flags_dropped_filters() {
    let tbl = status_table();
    let too_many = std::iter::once("f.status=published".to_string())
        .chain((0..crate::table::state::MAX_FILTERS).map(|i| format!("f.k{i}=v")))
        .collect::<Vec<_>>()
        .join("&");
    let too_long = format!(
        "f.status={}",
        "a".repeat(crate::table::state::MAX_FILTER_LEN + 1)
    );
    for query in [too_many.as_str(), too_long.as_str(), "filters=status:draft"] {
        let state = TableState::from_query(query);
        assert!(
            tbl.unapplied_filters(&cx(), &state, &[])
                .iter()
                .any(|(_, kind)| *kind == ignored::DROPPED),
            "the dropped filters of {query:.40} must be reported"
        );
    }
}

#[test]
fn ternary_all_is_a_neutral_noop_not_an_invalid_value() {
    let tbl = Table::<Task>::new(TextColumn::new(lens!(Task.title)))
        .filters(TernaryFilter::new(Task::fields().featured()));
    let state = filters_state(&[("featured", "all")]);
    assert!(
        tbl.filter_expr(&cx(), &state, &[]).is_none(),
        "all must select no predicate"
    );
    assert!(
        tbl.unapplied_filters(&cx(), &state, &[]).is_empty(),
        "all must not be flagged, got {:?}",
        tbl.unapplied_filters(&cx(), &state, &[])
    );
    // Genuine garbage still flags.
    assert_eq!(
        tbl.unapplied_filters(&cx(), &filters_state(&[("featured", "maybe")]), &[]),
        vec![("f.featured=maybe".to_string(), ignored::INVALID)]
    );
}

/// A request whose every filter is ignored lists the table unfiltered, which the list's banner
/// says; one that names a valid filter beside them still filters.
#[test]
fn ignored_filters_alone_leave_the_table_unfiltered() {
    let tbl = status_table();
    assert!(
        tbl.filter_expr(&cx(), &filters_state(&[("status", "typo")]), &[])
            .is_none()
    );
    assert!(
        tbl.filter_expr(
            &cx(),
            &filters_state(&[("status", "published"), ("bogus", "x")]),
            &[]
        )
        .is_some()
    );
}

struct NoColumns;

impl<M> IntoColumns<M> for NoColumns {
    fn into_columns(self) -> Vec<crate::column::BoxColumn<M>> {
        Vec::new()
    }
}

#[test]
fn empty_column_set_is_misdeclared() {
    let errors = Table::<User>::new(NoColumns).declaration_errors();
    assert_eq!(errors, [DeclarationErrorKind::NoColumns]);
}

#[test]
fn duplicate_column_name_is_misdeclared_on_field_computed_collision() {
    let errors = Table::<Task>::new((
        TextColumn::new(lens!(Task.status)).sortable(),
        ComputedColumn::new("Status", |t: &Task| t.status.clone()),
    ))
    .declaration_errors();
    assert!(
        errors
            .iter()
            .any(|error| matches!(error, DeclarationErrorKind::DuplicateColumn { .. })),
        "{errors:?}"
    );
}

#[test]
fn duplicate_column_name_is_misdeclared_on_case_only_computed_collision() {
    let errors = Table::<User>::new((
        ComputedColumn::new("Status", |u: &User| u.name.clone()),
        ComputedColumn::new("STATUS", |u: &User| u.name.clone()),
    ))
    .declaration_errors();
    assert!(
        errors
            .iter()
            .any(|error| matches!(error, DeclarationErrorKind::DuplicateColumn { .. })),
        "{errors:?}"
    );
}

#[test]
fn duplicate_column_name_is_misdeclared_on_duplicate_field() {
    let errors = Table::<User>::new((
        TextColumn::new(lens!(User.name)),
        TextColumn::new(lens!(User.name)),
    ))
    .declaration_errors();
    assert!(
        errors
            .iter()
            .any(|error| matches!(error, DeclarationErrorKind::DuplicateColumn { .. })),
        "{errors:?}"
    );
}

#[test]
fn duplicate_filter_name_is_misdeclared_on_duplicate_field() {
    let errors = Table::<Task>::new(TextColumn::new(lens!(Task.title)))
        .filters((
            SelectFilter::new(Task::fields().status(), vec!["published".to_string()]),
            SelectFilter::new(Task::fields().status(), vec!["draft".to_string()]),
        ))
        .declaration_errors();
    assert!(
        errors
            .iter()
            .any(|error| matches!(error, DeclarationErrorKind::DuplicateFilter { .. })),
        "{errors:?}"
    );
}

#[test]
fn a_filter_name_holding_a_dot_is_misdeclared() {
    let errors = Table::<Task>::new(TextColumn::new(lens!(Task.title)))
        .filters((QueryFilter::new("by.status", "Status")
            .option("Published", Task::fields().status().eq("published")),))
        .declaration_errors();
    assert_eq!(
        errors,
        [DeclarationErrorKind::DottedFilterName {
            name: "by.status".to_string()
        }]
    );
}

async fn seeded_users(names: &[&str]) -> topcoat::context::Cx {
    let mut db = memory_db(toasty::models!(User)).await;
    for name in names {
        toasty::create!(User {
            name: name.to_string()
        })
        .exec(&mut db)
        .await
        .unwrap();
    }
    Cx::default().with(db)
}

/// A cursor-paged table of users, `per_page` to a page: the cursor walk's subject.
fn paged_users_table(per_page: usize) -> Table<User> {
    Table::<User>::new(TextColumn::new(lens!(User.name)).sortable())
        .page_size(per_page)
        .pagination(crate::table::Pagination::Cursor)
}

#[tokio::test]
async fn full_walk_reaches_every_row_exactly_once_without_phantoms() {
    let cx = seeded_users(&["u01", "u02", "u03", "u04", "u05"]).await;
    let tbl = paged_users_table(2);
    let query = || toasty::stmt::Query::<List<User>>::all();
    // Forward walk from the first page to exhaustion.
    let mut seen = Vec::new();
    let mut state = TableState::default();
    let mut last = TablePage::load(&cx, &tbl, query(), &state).await.unwrap();
    assert!(last.prev_cursor.is_none(), "first page has no prev");
    loop {
        seen.extend(last.rows.iter().map(|u| u.name.clone()));
        match last.next_cursor.clone() {
            Some(cursor) => {
                state = TableState {
                    cursor: Some(crate::table::Cursor::After(cursor)),
                    ..TableState::default()
                };
                last = TablePage::load(&cx, &tbl, query(), &state).await.unwrap();
            }
            None => break,
        }
    }
    assert_eq!(seen, vec!["u01", "u02", "u03", "u04", "u05"]);
    // Backward walk from the terminal page to the first.
    let mut back = vec![last.rows.iter().map(|u| u.name.clone()).collect::<Vec<_>>()];
    while let Some(cursor) = last.prev_cursor.clone() {
        state = TableState {
            cursor: Some(crate::table::Cursor::Before(cursor)),
            ..TableState::default()
        };
        last = TablePage::load(&cx, &tbl, query(), &state).await.unwrap();
        back.push(last.rows.iter().map(|u| u.name.clone()).collect::<Vec<_>>());
    }
    back.reverse();
    assert_eq!(
        back,
        vec![
            vec!["u01".to_string(), "u02".to_string()],
            vec!["u03".to_string(), "u04".to_string()],
            vec!["u05".to_string()],
        ]
    );
    assert!(last.prev_cursor.is_none(), "first page has no prev");
}

#[tokio::test]
async fn exact_boundary_pages_carry_exact_cursors() {
    let cx = seeded_users(&["u01", "u02", "u03", "u04"]).await;
    let tbl = paged_users_table(2);
    let query = || toasty::stmt::Query::<List<User>>::all();
    let first = TablePage::load(&cx, &tbl, query(), &TableState::default())
        .await
        .unwrap();
    assert_eq!(first.rows.len(), 2);
    let cursor = first.next_cursor.clone().expect("page 1 of 2 has a next");
    let state = TableState {
        cursor: Some(crate::table::Cursor::After(cursor)),
        ..TableState::default()
    };
    let second = TablePage::load(&cx, &tbl, query(), &state).await.unwrap();
    assert_eq!(
        second
            .rows
            .iter()
            .map(|u| u.name.clone())
            .collect::<Vec<_>>(),
        vec!["u03".to_string(), "u04".to_string()]
    );
    assert!(
        second.next_cursor.is_none(),
        "terminal full page must not offer a next page"
    );
    assert!(
        second.prev_cursor.is_some(),
        "second page must offer a prev page"
    );
}

/// Counts sqlite driver executions inside the `gh172-budget` marker span.
struct BudgetState {
    count: std::sync::atomic::AtomicUsize,
    next_span: std::sync::atomic::AtomicU64,
}

/// Marker span attributing driver execs to the budget measurement.
const BUDGET_SPAN: &str = "gh172-budget";

thread_local! {
    static BUDGET_MARKERS: std::cell::RefCell<std::collections::HashSet<u64>> =
        std::cell::RefCell::new(std::collections::HashSet::new());
    static BUDGET_DEPTH: std::cell::Cell<u32> = const { std::cell::Cell::new(0) };
}

struct BudgetVisitor {
    driver: Option<String>,
    message: String,
}

impl tracing::field::Visit for BudgetVisitor {
    fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
        if field.name() == "driver" {
            self.driver = Some(value.to_string());
        }
        self.record_debug(field, &format_args!("{value}"));
    }

    fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
        if field.name() == "message" {
            self.message = format!("{value:?}");
        }
    }
}

impl tracing::Subscriber for BudgetState {
    fn enabled(&self, metadata: &tracing::Metadata<'_>) -> bool {
        metadata.target().starts_with("toasty_driver_sqlite")
            || (metadata.is_span() && metadata.name() == BUDGET_SPAN)
    }

    fn new_span(&self, span: &tracing::span::Attributes<'_>) -> tracing::span::Id {
        let id = self
            .next_span
            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        if span.metadata().name() == BUDGET_SPAN {
            BUDGET_MARKERS.with(|markers| {
                markers.borrow_mut().insert(id);
            });
        }
        tracing::span::Id::from_u64(id)
    }

    fn record(&self, _span: &tracing::span::Id, _values: &tracing::span::Record<'_>) {}

    fn record_follows_from(&self, _span: &tracing::span::Id, _follows: &tracing::span::Id) {}

    fn event(&self, event: &tracing::Event<'_>) {
        let in_scope = BUDGET_DEPTH.with(|depth| depth.get() > 0);
        if !in_scope {
            return;
        }
        let mut visitor = BudgetVisitor {
            driver: None,
            message: String::new(),
        };
        event.record(&mut visitor);
        if visitor.driver.as_deref() == Some("sqlite") && visitor.message.contains("driver exec") {
            self.count.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        }
    }

    fn enter(&self, span: &tracing::span::Id) {
        let is_marker = BUDGET_MARKERS.with(|markers| markers.borrow().contains(&span.into_u64()));
        if is_marker {
            BUDGET_DEPTH.with(|depth| depth.set(depth.get() + 1));
        }
    }

    fn exit(&self, span: &tracing::span::Id) {
        let is_marker = BUDGET_MARKERS.with(|markers| markers.borrow().contains(&span.into_u64()));
        if is_marker {
            BUDGET_DEPTH.with(|depth| depth.set(depth.get().saturating_sub(1)));
        }
    }
}

static BUDGET_INSTALL: std::sync::OnceLock<std::sync::Arc<BudgetState>> =
    std::sync::OnceLock::new();

/// Install the budget counter as the process-global default.
fn install_budget_counter() -> std::sync::Arc<BudgetState> {
    BUDGET_INSTALL
        .get_or_init(|| {
            let state = std::sync::Arc::new(BudgetState {
                count: std::sync::atomic::AtomicUsize::new(0),
                next_span: std::sync::atomic::AtomicU64::new(1),
            });
            tracing::subscriber::set_global_default(state.clone())
                .expect("budget counter installs once");
            tracing::callsite::rebuild_interest_cache();
            state
        })
        .clone()
}

#[tokio::test(flavor = "current_thread")]
async fn full_page_costs_main_plus_single_direction_probe() {
    use std::sync::atomic::Ordering;
    let cx = seeded_users(&["u01", "u02", "u03", "u04"]).await;
    let tbl = paged_users_table(2);
    let budget = install_budget_counter();
    let _scope = tracing::info_span!(BUDGET_SPAN).entered();
    let count_around = |reset: bool| {
        if reset {
            budget.count.store(0, Ordering::SeqCst);
        }
        budget.count.load(Ordering::SeqCst)
    };
    let ordered = || User::all().order_by(User::fields().name().asc());
    let mut db = crate::db::db(&cx).unwrap();
    count_around(true);
    let bare_main = ordered().paginate(2).exec(&mut db).await.unwrap();
    let bare_main_cost = count_around(false);
    count_around(true);
    let probe_cursor = bare_main.next_cursor.clone().unwrap();
    ordered()
        .paginate(1)
        .after(probe_cursor)
        .exec(&mut db)
        .await
        .unwrap();
    let bare_probe_cost = count_around(false);
    assert!(
        bare_main_cost > 0 && bare_probe_cost > 0,
        "the counter must observe driver execs, got main={bare_main_cost} probe={bare_probe_cost}"
    );
    // Full first page: main + exactly one next probe.
    count_around(true);
    let first = TablePage::load(
        &cx,
        &tbl,
        toasty::stmt::Query::<List<User>>::all(),
        &TableState::default(),
    )
    .await
    .unwrap();
    assert_eq!(
        count_around(false),
        bare_main_cost + bare_probe_cost,
        "full page must cost exactly main + one next probe"
    );
    assert_eq!(first.rows.len(), 2);
    // Short terminal page: main alone, no probe.
    let tbl3 = paged_users_table(3);
    count_around(true);
    let head = TablePage::load(
        &cx,
        &tbl3,
        toasty::stmt::Query::<List<User>>::all(),
        &TableState::default(),
    )
    .await
    .unwrap();
    assert_eq!(head.rows.len(), 3);
    let tail_state = TableState {
        cursor: head.next_cursor.clone().map(crate::table::Cursor::After),
        ..TableState::default()
    };
    count_around(true);
    let tail = TablePage::load(
        &cx,
        &tbl3,
        toasty::stmt::Query::<List<User>>::all(),
        &tail_state,
    )
    .await
    .unwrap();
    assert_eq!(tail.rows.len(), 1);
    assert!(tail.next_cursor.is_none());
    let short_cost = count_around(false);
    count_around(true);
    ordered().paginate(3).exec(&mut db).await.unwrap();
    let bare_short_cost = count_around(false);
    assert_eq!(
        short_cost, bare_short_cost,
        "short page must cost exactly one bare fetch (no probe)"
    );
    let p1 = TablePage::load(
        &cx,
        &tbl,
        toasty::stmt::Query::<List<User>>::all(),
        &TableState::default(),
    )
    .await
    .unwrap();
    let p2_state = TableState {
        cursor: p1.next_cursor.clone().map(crate::table::Cursor::After),
        ..TableState::default()
    };
    let p2 = TablePage::load(
        &cx,
        &tbl,
        toasty::stmt::Query::<List<User>>::all(),
        &p2_state,
    )
    .await
    .unwrap();
    assert_eq!(p2.rows.len(), 2);
    let back_to_first = TableState {
        cursor: p2.prev_cursor.clone().map(crate::table::Cursor::Before),
        ..TableState::default()
    };
    count_around(true);
    let first_again = TablePage::load(
        &cx,
        &tbl,
        toasty::stmt::Query::<List<User>>::all(),
        &back_to_first,
    )
    .await
    .unwrap();
    assert_eq!(
        first_again
            .rows
            .iter()
            .map(|u| u.name.clone())
            .collect::<Vec<_>>(),
        vec!["u01".to_string(), "u02".to_string()]
    );
    assert!(
        first_again.prev_cursor.is_none(),
        "backward landing on the first page must hide the phantom prev"
    );
    assert_eq!(
        count_around(false),
        bare_main_cost + bare_probe_cost,
        "backward landing must cost exactly main + one prev probe"
    );
}

#[tokio::test]
async fn stale_cursor_is_marked_for_retry() {
    use toasty::stmt::Value;
    use toasty_core::stmt::ValueRecord;

    let mut db = memory_db(toasty::models!(Task)).await;
    for title in ["Alpha", "Bravo", "Charlie", "Delta"] {
        toasty::create!(Task {
            title: title.to_string(),
            status: "draft".to_string(),
            featured: false,
            created_at: "2024-01-01T00:00:00Z".parse::<jiff::Timestamp>().unwrap(),
        })
        .exec(&mut db)
        .await
        .unwrap();
    }
    let cx = topcoat::context::Cx::default().with(db);
    let table = || {
        Table::<Task>::new(TextColumn::new(lens!(Task.title)).sortable())
            .page_size(2)
            .pagination(crate::table::Pagination::Cursor)
    };
    // The query orders by `title` then the PK to break ties, so a cursor
    // with three fields has one too many.
    let wide = crate::toasty_compat::cursor::encode(&Value::Record(ValueRecord::from_vec(vec![
        Value::String("Alpha".to_string()),
        Value::String("x".to_string()),
        Value::I64(1),
    ])))
    .unwrap();
    let state = TableState {
        cursor: Some(crate::table::Cursor::After(wide)),
        ..TableState::default()
    };
    let error = TablePage::load(
        &cx,
        &table(),
        toasty::stmt::Query::<List<Task>>::all(),
        &state,
    )
    .await
    .expect_err("a cursor with too many fields must fail the load");
    assert!(
        crate::error::TabloError::is_cursor(&error),
        "a rejected cursor must carry the cursor marker, got {error}"
    );
    assert!(
        matches!(
            crate::error::TabloError::of(&error),
            Some(crate::error::TabloError::CursorRejected(_))
        ),
        "the refusal is not a decode failure, got {error}"
    );

    // A cursor cut from this query's own ordering round-trips.
    let first = TablePage::load(
        &cx,
        &table(),
        toasty::stmt::Query::<List<Task>>::all(),
        &TableState::default(),
    )
    .await
    .unwrap();
    let state = TableState {
        cursor: first.next_cursor.clone().map(crate::table::Cursor::After),
        ..TableState::default()
    };
    assert!(
        TablePage::load(
            &cx,
            &table(),
            toasty::stmt::Query::<List<Task>>::all(),
            &state
        )
        .await
        .is_ok(),
        "a matching cursor must keep loading"
    );
}

/// The rows' names and the position of the page `query` names on a numbered table of
/// `per_page` users.
async fn numbered_page(
    cx: &topcoat::context::Cx,
    table: &Table<User>,
    query: &str,
) -> (Vec<String>, crate::table::PagePosition) {
    let page = TablePage::load(
        cx,
        table,
        toasty::stmt::Query::<List<User>>::all(),
        &TableState::from_query(query),
    )
    .await
    .expect("the page loads");
    let names = page.rows.iter().map(|user| user.name.clone()).collect();
    (
        names,
        page.position.expect("a numbered page knows its position"),
    )
}

/// A numbered table counts its result set, pages by offset in the sorted order, and reads a page
/// past the last as the last.
#[tokio::test]
async fn numbered_pages_count_and_clamp() {
    let cx = seeded_users(&["u01", "u02", "u03", "u04", "u05"]).await;
    let table = Table::<User>::new(TextColumn::new(lens!(User.name)).sortable()).page_size(2);

    let (rows, position) = numbered_page(&cx, &table, "").await;
    assert_eq!(rows, ["u01", "u02"]);
    assert_eq!((position.page, position.total), (1, Some(5)));
    assert_eq!(position.last_page(), Some(3));
    assert!(position.has_next);

    let (rows, position) = numbered_page(&cx, &table, "page=3").await;
    assert_eq!(rows, ["u05"]);
    assert!(!position.has_next);

    let (rows, position) = numbered_page(&cx, &table, "page=9").await;
    assert_eq!((rows, position.page), (vec!["u05".to_string()], 3));

    let (rows, _) = numbered_page(&cx, &table, "page=2&sort=name&dir=desc").await;
    assert_eq!(rows, ["u03", "u02"]);

    // The count follows the search, so the pages do too.
    let searchable =
        Table::<User>::new(TextColumn::new(lens!(User.name)).searchable().sortable()).page_size(2);
    let (rows, position) = numbered_page(&cx, &searchable, "q=u0&page=2").await;
    assert_eq!((rows.len(), position.total), (2, Some(5)));
    let (_, position) = numbered_page(&cx, &searchable, "q=u05").await;
    assert_eq!((position.total, position.has_next), (Some(1), false));
}

/// An empty result set is one empty page, not a page zero.
#[tokio::test]
async fn an_empty_numbered_table_is_page_one() {
    let cx = seeded_users(&[]).await;
    let table = Table::<User>::new(TextColumn::new(lens!(User.name)).sortable());
    let (rows, position) = numbered_page(&cx, &table, "page=4").await;
    assert!(rows.is_empty());
    assert_eq!(
        (position.page, position.total, position.has_next),
        (1, Some(0), false)
    );
    assert_eq!(position.last_page(), Some(1));
}

/// A simple table counts nothing and learns of a next page from one extra row.
#[tokio::test]
async fn simple_pages_look_one_row_ahead() {
    let cx = seeded_users(&["u01", "u02", "u03", "u04"]).await;
    let table = Table::<User>::new(TextColumn::new(lens!(User.name)).sortable())
        .page_size(2)
        .pagination(crate::table::Pagination::Simple);
    let (rows, position) = numbered_page(&cx, &table, "").await;
    assert_eq!(rows, ["u01", "u02"]);
    assert_eq!((position.total, position.has_next), (None, true));
    let (rows, position) = numbered_page(&cx, &table, "page=2").await;
    assert_eq!(rows, ["u03", "u04"]);
    assert!(!position.has_next, "an exactly full last page has no next");
}

/// `?per_page=` sets the page size only to a size the table offers.
#[tokio::test]
async fn per_page_reads_only_an_offered_size() {
    let cx = seeded_users(&["u01", "u02", "u03", "u04", "u05"]).await;
    let table = Table::<User>::new(TextColumn::new(lens!(User.name)).sortable())
        .page_size(2)
        .per_page_options([4]);
    let (rows, position) = numbered_page(&cx, &table, "per_page=4").await;
    assert_eq!((rows.len(), position.per_page), (4, 4));
    let (rows, _) = numbered_page(&cx, &table, "per_page=3").await;
    assert_eq!(rows.len(), 2, "an unoffered size reads as the table's own");
    let fixed = table.clone().per_page_options([]);
    let (rows, _) = numbered_page(&cx, &fixed, "per_page=4").await;
    assert_eq!(rows.len(), 2, "a table offering no choice keeps its size");
}

/// The list's sort outranks an ordering the resource's query carries, on every pagination: Toasty
/// appends an `order_by`, so the query's ordering would otherwise come first.
#[tokio::test]
async fn the_sort_outranks_the_resource_querys_ordering() {
    let cx = seeded_users(&["u01", "u02", "u03"]).await;
    let table = Table::<User>::new(TextColumn::new(lens!(User.name)).sortable());
    for pagination in [
        crate::table::Pagination::Cursor,
        crate::table::Pagination::Pages,
        crate::table::Pagination::Simple,
    ] {
        let page = TablePage::load(
            &cx,
            &table.clone().pagination(pagination),
            toasty::stmt::Query::<List<User>>::all().order_by(User::fields().name().desc()),
            &TableState::from_query("sort=name"),
        )
        .await
        .expect("the page loads");
        let names: Vec<_> = page.rows.iter().map(|user| user.name.as_str()).collect();
        assert_eq!(names, ["u01", "u02", "u03"], "{pagination:?}");
    }
}

/// A state naming page 0, which `TableState`'s public field allows, loads the first page.
#[tokio::test]
async fn page_zero_loads_the_first_page() {
    let cx = seeded_users(&["u01", "u02"]).await;
    let table = Table::<User>::new(TextColumn::new(lens!(User.name)).sortable());
    for pagination in [
        crate::table::Pagination::Pages,
        crate::table::Pagination::Simple,
    ] {
        let state = TableState {
            page: Some(0),
            ..TableState::from_query("sort=name")
        };
        let page = TablePage::load(
            &cx,
            &table.clone().pagination(pagination),
            toasty::stmt::Query::<List<User>>::all(),
            &state,
        )
        .await
        .expect("the page loads");
        let names: Vec<_> = page.rows.iter().map(|user| user.name.as_str()).collect();
        assert_eq!(names, ["u01", "u02"], "{pagination:?}");
    }
}