autumn-web 0.3.0

//! Standardized pagination primitives.
//!
//! Autumn ships two complementary flavours of pagination:
//!
//! 1. **Offset pagination** ([`PageRequest`] / [`Page<T>`]) — classic
//!    `?page=N&size=M` with metadata (total elements, total pages).
//!    Best for stable, browse-style UIs.
//! 2. **Cursor pagination** ([`CursorRequest`] / [`CursorPage<T>`]) —
//!    keyset/seek pagination with an opaque `next_cursor` token. Best
//!    for real-time feeds and infinite scroll: O(1) page depth and
//!    zero duplicates under concurrent inserts.
//!
//! # Quick start (offset)
//!
//! Paginating a handler takes three lines: run the count query, run the page
//! query, and wrap the result in a [`Page`].
//!
//! ```rust,ignore
//! use autumn_web::prelude::*;
//! use autumn_web::pagination::{Page, PageRequest};
//! use diesel::prelude::*;
//! use diesel_async::RunQueryDsl;
//!
//! #[get("/api/posts")]
//! async fn list(page: PageRequest, mut db: Db) -> AutumnResult<Json<Page<Post>>> {
//!     let total: i64 = posts::table.count().get_result(&mut db).await?;
//!     let items: Vec<Post> = posts::table
//!         .limit(page.limit()).offset(page.offset())
//!         .select(Post::as_select())
//!         .load(&mut db).await?;
//!     Ok(Json(Page::new(items, total, &page)))
//! }
//! ```
//!
//! # Quick start (cursor)
//!
//! Cursor pagination is keyset pagination: filter by a stable, deterministic
//! sort key (with a unique tie-breaker like `id`), fetch `limit + 1` rows,
//! and let [`CursorPage::from_overfetched`] derive the `next_cursor` from
//! the boundary row.
//!
//! ```rust,ignore
//! use autumn_web::prelude::*;
//! use autumn_web::pagination::{CursorPage, CursorRequest};
//! use chrono::{DateTime, Utc};
//! use diesel::prelude::*;
//! use diesel_async::RunQueryDsl;
//! use serde::{Deserialize, Serialize};
//!
//! // Sort key — created_at + id is the stable, deterministic tie-breaker.
//! #[derive(Serialize, Deserialize)]
//! struct PostCursor { created_at: DateTime<Utc>, id: i64 }
//!
//! #[get("/api/feed")]
//! async fn feed(cur: CursorRequest, mut db: Db) -> AutumnResult<Json<CursorPage<Post>>> {
//!     let mut q = posts::table.into_boxed();
//!     if let Some(c) = cur.decode::<PostCursor>() {
//!         q = q.filter(
//!             posts::created_at.lt(c.created_at)
//!                 .or(posts::created_at.eq(c.created_at).and(posts::id.lt(c.id))),
//!         );
//!     }
//!     let items: Vec<Post> = q
//!         .order((posts::created_at.desc(), posts::id.desc()))
//!         .limit(cur.fetch_limit())
//!         .select(Post::as_select())
//!         .load(&mut db).await?;
//!     Ok(Json(CursorPage::from_overfetched(items, &cur, |p| {
//!         PostCursor { created_at: p.created_at, id: p.id }
//!     })))
//! }
//! ```
//!
//! # Query contract
//!
//! Offset pagination uses two query parameters:
//!
//! | Parameter | Meaning | Default | Clamped to |
//! |-----------|---------|---------|------------|
//! | `page` | 1-based page index | `1` | `>= 1` |
//! | `size` | Items per page | [`DEFAULT_PAGE_SIZE`] | <code>1..=[`MAX_PAGE_SIZE`]</code> |
//!
//! Cursor pagination uses:
//!
//! | Parameter | Meaning | Default | Clamped to |
//! |-----------|---------|---------|------------|
//! | `cursor` | Opaque token from a prior `next_cursor` (omit for first page) | `None` | — |
//! | `size` | Items per page | [`DEFAULT_PAGE_SIZE`] | <code>1..=[`MAX_PAGE_SIZE`]</code> |
//!
//! Requests like `?size=0`, `?size=9999`, `?page=0`, or even `?page=abc`
//! are silently coerced to the valid range rather than rejected — bad
//! pagination parameters should not 400. Unparseable or tampered cursors
//! decode to `None` (i.e. fall back to the first page) for the same reason.
//!
//! # Signed cursors (optional)
//!
//! Plain cursors are *opaque but unsigned* — the same model used by
//! Stripe, GitHub, and Relay. Forging one is equivalent to seeking to
//! an arbitrary offset, which clients can already do with `?page=N`,
//! so for sort-key-only cursors (timestamp + id) signing adds no real
//! protection.
//!
//! However, if a handler ever encodes anything *sensitive to tampering*
//! into the cursor payload — a tenant id, a user scope, anything the
//! handler relies on to filter results — switch to the signed API:
//!
//! - [`Cursor::encode_signed`] / [`Cursor::decode_signed`]
//! - [`CursorRequest::decode_signed`]
//! - [`CursorPage::from_overfetched_signed`]
//!
//! All three take a key as `&[u8]`. Tokens are signed with HMAC-SHA256
//! and verified in constant time; tampered or unsigned tokens decode
//! to `None`.
//!
//! # Response shape
//!
//! [`Page<T>`] serializes as:
//!
//! ```json
//! {
//!   "content": [ ... ],
//!   "page": 1,
//!   "size": 20,
//!   "total_elements": 137,
//!   "total_pages": 7,
//!   "has_next": true,
//!   "has_previous": false
//! }
//! ```
//!
//! [`CursorPage<T>`] serializes as:
//!
//! ```json
//! {
//!   "content": [ ... ],
//!   "size": 20,
//!   "next_cursor": "eyJpZCI6MTIzfQ",
//!   "has_next": true
//! }
//! ```

use axum::extract::FromRequestParts;
use axum::http::request::Parts;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};

/// Default number of items per page when no `size` is provided.
pub const DEFAULT_PAGE_SIZE: u32 = 20;

/// Hard upper bound on `size` — prevents clients from requesting huge
/// pages that could OOM the server or overwhelm the database.
pub const MAX_PAGE_SIZE: u32 = 100;

// ── PageRequest ─────────────────────────────────────────────────────

/// Pagination parameters parsed from the query string.
///
/// Use as a handler extractor to receive `?page=N&size=M`. Both fields
/// are optional; missing values fall back to [`DEFAULT_PAGE_SIZE`] and
/// page `1`. Out-of-range *and unparseable* values are clamped rather
/// than rejected: `page < 1` becomes `1`, `size` is clamped to
/// <code>1..=[`MAX_PAGE_SIZE`]</code>, and inputs like `?page=abc` are
/// silently ignored. A list endpoint should never 400 because of a
/// malformed pager.
///
/// # Examples
///
/// ```rust,no_run
/// use autumn_web::prelude::*;
/// use autumn_web::pagination::PageRequest;
///
/// #[get("/api/items")]
/// async fn list(page: PageRequest) -> String {
///     format!("page {} (limit {}, offset {})", page.page(), page.limit(), page.offset())
/// }
/// ```
#[derive(Debug, Clone, Copy, Default, Deserialize)]
pub struct PageRequest {
    #[serde(default)]
    page: Option<u32>,
    #[serde(default)]
    size: Option<u32>,
}

impl PageRequest {
    /// Construct a [`PageRequest`] explicitly. Values are clamped to the
    /// valid ranges defined by [`DEFAULT_PAGE_SIZE`] / [`MAX_PAGE_SIZE`].
    #[must_use]
    pub const fn new(page: u32, size: u32) -> Self {
        Self {
            page: Some(page),
            size: Some(size),
        }
    }

    /// Resolved 1-based page number. `0` or missing is coerced to `1`.
    #[must_use]
    pub const fn page(&self) -> u32 {
        match self.page {
            Some(p) if p >= 1 => p,
            _ => 1,
        }
    }

    /// Resolved page size, clamped to <code>1..=[`MAX_PAGE_SIZE`]</code>.
    #[must_use]
    pub const fn size(&self) -> u32 {
        match self.size {
            Some(0) | None => DEFAULT_PAGE_SIZE,
            Some(s) if s > MAX_PAGE_SIZE => MAX_PAGE_SIZE,
            Some(s) => s,
        }
    }

    /// `LIMIT` value for a Diesel or raw SQL query (`== size()`).
    #[must_use]
    pub const fn limit(&self) -> i64 {
        self.size() as i64
    }

    /// `OFFSET` value for a Diesel or raw SQL query.
    #[must_use]
    pub const fn offset(&self) -> i64 {
        ((self.page() - 1) as i64) * (self.size() as i64)
    }
}

impl<S> FromRequestParts<S> for PageRequest
where
    S: Send + Sync,
{
    type Rejection = std::convert::Infallible;

    async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
        // Manual parse rather than `Query::<Self>::from_request_parts` so
        // that unparseable values (`?page=abc`, `?size=`, duplicate keys,
        // percent-encoding errors) fall back to defaults instead of
        // rejecting the whole request with a 400.
        Ok(parts.uri.query().map_or_else(Self::default, parse_query))
    }
}

/// Best-effort parse of a URL-encoded query string into a [`PageRequest`].
/// Unknown keys, malformed values, and percent-decoding failures are
/// silently ignored. Later occurrences of `page`/`size` win, matching the
/// behaviour of `serde_urlencoded`.
fn parse_query(query: &str) -> PageRequest {
    let mut req = PageRequest::default();
    for pair in query.split('&').filter(|s| !s.is_empty()) {
        let (raw_key, raw_value) = pair.split_once('=').unwrap_or((pair, ""));
        let Ok(key) = percent_decode(raw_key) else {
            continue;
        };
        let Ok(value) = percent_decode(raw_value) else {
            continue;
        };
        match key.as_str() {
            "page" => {
                if let Ok(n) = value.parse::<u32>() {
                    req.page = Some(n);
                }
            }
            "size" => {
                if let Ok(n) = value.parse::<u32>() {
                    req.size = Some(n);
                }
            }
            _ => {}
        }
    }
    req
}

/// Decode a single URL-encoded segment. `+` is treated as a space (the
/// `application/x-www-form-urlencoded` convention used by
/// `serde_urlencoded` and browsers).
fn percent_decode(input: &str) -> Result<String, std::str::Utf8Error> {
    let bytes = input.as_bytes();
    let mut out: Vec<u8> = Vec::with_capacity(bytes.len());
    let mut i = 0;
    while i < bytes.len() {
        match bytes[i] {
            b'+' => {
                out.push(b' ');
                i += 1;
            }
            b'%' if i + 2 < bytes.len() => {
                let hi = (bytes[i + 1] as char).to_digit(16);
                let lo = (bytes[i + 2] as char).to_digit(16);
                if let (Some(h), Some(l)) = (hi, lo) {
                    // h and l are both `0..=15` from `to_digit(16)`, so
                    // `(h << 4) | l` fits in a u8 by construction.
                    out.push(u8::try_from((h << 4) | l).unwrap_or(0));
                    i += 3;
                } else {
                    out.push(bytes[i]);
                    i += 1;
                }
            }
            b => {
                out.push(b);
                i += 1;
            }
        }
    }
    std::str::from_utf8(&out).map(ToOwned::to_owned)
}

// ── Page<T> ─────────────────────────────────────────────────────────

/// Paginated response wrapper with navigation metadata.
///
/// `Page` serializes to JSON for API responses and exposes the fields a
/// Maud template needs to render pager links (previous/next, page index,
/// total pages).
///
/// Construct one with [`Page::new`] after running your count + page
/// queries, or with [`Page::empty`] when you have no data to return.
///
/// # JSON shape
///
/// ```json
/// {
///   "content": [ /* T items */ ],
///   "page": 1,
///   "size": 20,
///   "total_elements": 137,
///   "total_pages": 7,
///   "has_next": true,
///   "has_previous": false
/// }
/// ```
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Page<T> {
    /// The items on this page.
    pub content: Vec<T>,
    /// Current 1-based page index.
    pub page: u32,
    /// Page size used to produce `content`.
    pub size: u32,
    /// Total number of items across every page.
    pub total_elements: u64,
    /// Total number of pages (`ceil(total_elements / size)`), minimum `1`.
    pub total_pages: u32,
    /// Whether there is a page after this one.
    pub has_next: bool,
    /// Whether there is a page before this one.
    pub has_previous: bool,
}

impl<T> Page<T> {
    /// Build a page from the materialized `items` and the total row
    /// count returned by the database.
    ///
    /// `total` is accepted as `i64` to match Diesel's
    /// `COUNT(*)` result type; values below zero are treated as zero.
    #[must_use]
    pub fn new(items: Vec<T>, total: i64, request: &PageRequest) -> Self {
        let size = request.size();
        let page = request.page();
        let total_elements = u64::try_from(total).unwrap_or(0);

        // ceil(total / size), minimum 1 so an empty result still
        // reports `total_pages = 1` — callers don't have to branch on
        // "no rows" when rendering a pager.
        let total_pages = if total_elements == 0 {
            1
        } else {
            // size() is always >= 1, so this division is safe.
            u32::try_from(total_elements.div_ceil(u64::from(size))).unwrap_or(u32::MAX)
        };

        Self {
            content: items,
            page,
            size,
            total_elements,
            total_pages,
            has_next: page < total_pages,
            has_previous: page > 1,
        }
    }

    /// Build an empty page using the caller's request parameters.
    ///
    /// Useful when a filter short-circuits before hitting the database.
    #[must_use]
    pub fn empty(request: &PageRequest) -> Self {
        Self::new(Vec::new(), 0, request)
    }

    /// Transform the content while preserving pagination metadata.
    ///
    /// Typical use: converting database rows into DTOs for JSON output
    /// without re-running the count query.
    pub fn map<U, F: FnMut(T) -> U>(self, f: F) -> Page<U> {
        Page {
            content: self.content.into_iter().map(f).collect(),
            page: self.page,
            size: self.size,
            total_elements: self.total_elements,
            total_pages: self.total_pages,
            has_next: self.has_next,
            has_previous: self.has_previous,
        }
    }
}

// ── Cursor encoding ─────────────────────────────────────────────────
//
// Cursor tokens are base64url-encoded JSON. base64url is used (rather
// than the standard alphabet) so that tokens are safe to embed in a
// URL without percent-encoding, and padding (`=`) is omitted to keep
// them tidy. Tokens are *opaque* to clients — encoding the structure
// keeps callers from forging cursors but, more importantly, lets the
// server change the schema without breaking the wire contract.

const BASE64URL_ALPHABET: &[u8; 64] =
    b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";

fn base64url_encode(input: &[u8]) -> String {
    let mut out = String::with_capacity(input.len().div_ceil(3) * 4);
    let mut chunks = input.chunks_exact(3);
    for chunk in &mut chunks {
        let n = (u32::from(chunk[0]) << 16) | (u32::from(chunk[1]) << 8) | u32::from(chunk[2]);
        out.push(BASE64URL_ALPHABET[((n >> 18) & 0x3F) as usize] as char);
        out.push(BASE64URL_ALPHABET[((n >> 12) & 0x3F) as usize] as char);
        out.push(BASE64URL_ALPHABET[((n >> 6) & 0x3F) as usize] as char);
        out.push(BASE64URL_ALPHABET[(n & 0x3F) as usize] as char);
    }
    let rem = chunks.remainder();
    match rem.len() {
        0 => {}
        1 => {
            let n = u32::from(rem[0]) << 16;
            out.push(BASE64URL_ALPHABET[((n >> 18) & 0x3F) as usize] as char);
            out.push(BASE64URL_ALPHABET[((n >> 12) & 0x3F) as usize] as char);
        }
        2 => {
            let n = (u32::from(rem[0]) << 16) | (u32::from(rem[1]) << 8);
            out.push(BASE64URL_ALPHABET[((n >> 18) & 0x3F) as usize] as char);
            out.push(BASE64URL_ALPHABET[((n >> 12) & 0x3F) as usize] as char);
            out.push(BASE64URL_ALPHABET[((n >> 6) & 0x3F) as usize] as char);
        }
        _ => unreachable!("chunks_exact remainder is < 3 by construction"),
    }
    out
}

fn base64url_decode(input: &str) -> Option<Vec<u8>> {
    let bytes = input.as_bytes();
    let value = |b: u8| -> Option<u32> {
        match b {
            b'A'..=b'Z' => Some(u32::from(b - b'A')),
            b'a'..=b'z' => Some(u32::from(b - b'a') + 26),
            b'0'..=b'9' => Some(u32::from(b - b'0') + 52),
            b'-' => Some(62),
            b'_' => Some(63),
            _ => None,
        }
    };
    let mut out = Vec::with_capacity(bytes.len() * 3 / 4);
    let mut i = 0;
    while i + 4 <= bytes.len() {
        let n = (value(bytes[i])? << 18)
            | (value(bytes[i + 1])? << 12)
            | (value(bytes[i + 2])? << 6)
            | value(bytes[i + 3])?;
        out.push(u8::try_from((n >> 16) & 0xFF).ok()?);
        out.push(u8::try_from((n >> 8) & 0xFF).ok()?);
        out.push(u8::try_from(n & 0xFF).ok()?);
        i += 4;
    }
    match bytes.len() - i {
        0 => {}
        1 => return None, // a single trailing char is not a valid base64 group
        2 => {
            let n = (value(bytes[i])? << 18) | (value(bytes[i + 1])? << 12);
            out.push(u8::try_from((n >> 16) & 0xFF).ok()?);
        }
        3 => {
            let n = (value(bytes[i])? << 18)
                | (value(bytes[i + 1])? << 12)
                | (value(bytes[i + 2])? << 6);
            out.push(u8::try_from((n >> 16) & 0xFF).ok()?);
            out.push(u8::try_from((n >> 8) & 0xFF).ok()?);
        }
        _ => unreachable!("bytes.len() - i is < 4 by the loop condition"),
    }
    Some(out)
}

/// Opaque cursor tokens for cursor-based pagination.
///
/// A cursor wraps a serializable sort-key payload (typically a struct
/// containing a timestamp and a unique tie-breaker like `id`) into a
/// URL-safe string. The on-the-wire format is base64url-encoded JSON;
/// callers should treat tokens as opaque.
///
/// # Examples
///
/// ```rust
/// use autumn_web::pagination::Cursor;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Serialize, Deserialize, PartialEq, Debug)]
/// struct Key { id: i64 }
///
/// let token = Cursor::encode(&Key { id: 42 }).unwrap();
/// let decoded: Key = Cursor::decode(&token).unwrap();
/// assert_eq!(decoded, Key { id: 42 });
/// ```
pub struct Cursor;

impl Cursor {
    /// Encode a serializable value as an opaque URL-safe cursor token.
    ///
    /// # Errors
    ///
    /// Returns `serde_json::Error` if the value cannot be serialized.
    pub fn encode<T: Serialize>(value: &T) -> Result<String, serde_json::Error> {
        Ok(base64url_encode(&serde_json::to_vec(value)?))
    }

    /// Decode an opaque cursor token back into a typed value.
    ///
    /// Returns `None` for any malformed token — invalid base64, invalid
    /// UTF-8, or JSON that doesn't match `T`. A list endpoint should
    /// silently fall back to the first page rather than 400 on a
    /// tampered or stale cursor, matching the forgiving behaviour of
    /// the `?page=` / `?size=` parser.
    #[must_use]
    pub fn decode<T: DeserializeOwned>(token: &str) -> Option<T> {
        let bytes = base64url_decode(token)?;
        serde_json::from_slice(&bytes).ok()
    }

    /// Encode a value as a *signed* cursor token using HMAC-SHA256.
    ///
    /// Use this when the cursor payload encodes anything sensitive to
    /// tampering — for example a tenant boundary, a user id, or any
    /// scope a handler relies on to filter results. Without signing,
    /// a client could edit the JSON and re-encode it. Plain
    /// [`Cursor::encode`] is fine for cursors that only carry sort-key
    /// values (timestamps, primary keys), since forging one is
    /// equivalent to seeking to an arbitrary offset.
    ///
    /// The token format is `<base64url(json)>.<base64url(hmac)>`.
    /// The signature covers exactly the payload bytes; the encoded
    /// payload itself is not encrypted (cursors are not secrets).
    ///
    /// # Errors
    ///
    /// Returns `serde_json::Error` if the value cannot be serialized.
    pub fn encode_signed<T: Serialize>(value: &T, key: &[u8]) -> Result<String, serde_json::Error> {
        let payload = serde_json::to_vec(value)?;
        let payload_b64 = base64url_encode(&payload);
        let mac = hmac_sha256(key, payload_b64.as_bytes());
        let sig_b64 = base64url_encode(&mac);
        Ok(format!("{payload_b64}.{sig_b64}"))
    }

    /// Decode a signed cursor token, verifying its HMAC-SHA256 signature.
    ///
    /// Returns `None` for any of: malformed structure, malformed
    /// base64, signature mismatch, JSON that doesn't match `T`. The
    /// signature is verified in constant time. A handler that uses
    /// this should treat `None` the same way it treats no cursor
    /// (fall back to first page) rather than returning an error —
    /// the goal is to ignore tampered cursors, not to surface them
    /// as user-facing failures.
    #[must_use]
    pub fn decode_signed<T: DeserializeOwned>(token: &str, key: &[u8]) -> Option<T> {
        let (payload_b64, sig_b64) = token.split_once('.')?;
        let expected_sig = base64url_decode(sig_b64)?;
        let actual_sig = hmac_sha256(key, payload_b64.as_bytes());
        if !constant_time_eq(&expected_sig, &actual_sig) {
            return None;
        }
        let payload = base64url_decode(payload_b64)?;
        serde_json::from_slice(&payload).ok()
    }
}

fn hmac_sha256(key: &[u8], message: &[u8]) -> [u8; 32] {
    use hmac::{Hmac, Mac};
    use sha2::Sha256;
    // `Hmac::new_from_slice` accepts any key length.
    let mut mac = <Hmac<Sha256> as Mac>::new_from_slice(key).expect("HMAC accepts any key length");
    mac.update(message);
    mac.finalize().into_bytes().into()
}

fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
    use subtle::ConstantTimeEq;
    a.ct_eq(b).into()
}

// ── CursorRequest ───────────────────────────────────────────────────

/// Cursor pagination parameters parsed from the query string.
///
/// Use as a handler extractor to receive `?cursor=<token>&size=<n>`.
/// Both fields are optional: missing `cursor` means "give me the
/// first page", missing `size` means [`DEFAULT_PAGE_SIZE`]. The same
/// forgiving coercion as [`PageRequest`] applies — `?size=0` falls
/// back to the default, `?size=9999` is clamped to [`MAX_PAGE_SIZE`],
/// and unparseable values are silently ignored. A malformed cursor is
/// treated as no cursor.
///
/// # Examples
///
/// ```rust,no_run
/// use autumn_web::prelude::*;
/// use autumn_web::pagination::CursorRequest;
///
/// #[get("/api/feed")]
/// async fn feed(cur: CursorRequest) -> String {
///     format!("size={}, cursor={:?}", cur.size(), cur.cursor())
/// }
/// ```
#[derive(Debug, Clone, Default, Deserialize)]
pub struct CursorRequest {
    #[serde(default)]
    cursor: Option<String>,
    #[serde(default)]
    size: Option<u32>,
}

impl CursorRequest {
    /// Construct a [`CursorRequest`] explicitly. Useful in tests.
    #[must_use]
    pub const fn new(cursor: Option<String>, size: u32) -> Self {
        Self {
            cursor,
            size: Some(size),
        }
    }

    /// The raw, opaque cursor token, if any.
    #[must_use]
    pub fn cursor(&self) -> Option<&str> {
        self.cursor.as_deref()
    }

    /// Decode the cursor into a typed sort-key value.
    ///
    /// Returns `None` if the cursor is missing or unparseable. Use
    /// this in handlers to add the keyset filter to the query.
    #[must_use]
    pub fn decode<T: DeserializeOwned>(&self) -> Option<T> {
        Cursor::decode(self.cursor.as_deref()?)
    }

    /// Decode a *signed* cursor, verifying its HMAC-SHA256 signature
    /// against `key`. See [`Cursor::decode_signed`] for the threat
    /// model and when to use signed vs. unsigned cursors.
    ///
    /// Returns `None` if the cursor is missing, malformed, or has an
    /// invalid signature.
    #[must_use]
    pub fn decode_signed<T: DeserializeOwned>(&self, key: &[u8]) -> Option<T> {
        Cursor::decode_signed(self.cursor.as_deref()?, key)
    }

    /// Resolved page size, clamped to <code>1..=[`MAX_PAGE_SIZE`]</code>.
    #[must_use]
    pub const fn size(&self) -> u32 {
        match self.size {
            Some(0) | None => DEFAULT_PAGE_SIZE,
            Some(s) if s > MAX_PAGE_SIZE => MAX_PAGE_SIZE,
            Some(s) => s,
        }
    }

    /// `LIMIT` value matching the requested page size.
    #[must_use]
    pub const fn limit(&self) -> i64 {
        self.size() as i64
    }

    /// `LIMIT + 1` — fetch one extra row so the handler can detect
    /// whether a next page exists without an extra query.
    ///
    /// This is the value to pass to Diesel's `.limit(...)` when using
    /// [`CursorPage::from_overfetched`].
    #[must_use]
    pub const fn fetch_limit(&self) -> i64 {
        self.size() as i64 + 1
    }
}

impl<S> FromRequestParts<S> for CursorRequest
where
    S: Send + Sync,
{
    type Rejection = std::convert::Infallible;

    async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
        Ok(parts
            .uri
            .query()
            .map_or_else(Self::default, parse_cursor_query))
    }
}

/// Best-effort parse of a URL-encoded query string into a [`CursorRequest`].
/// Same coercion rules as [`parse_query`]: unknown keys, malformed values,
/// and percent-decoding failures are silently ignored. Later occurrences of
/// `cursor`/`size` win.
fn parse_cursor_query(query: &str) -> CursorRequest {
    let mut req = CursorRequest::default();
    for pair in query.split('&').filter(|s| !s.is_empty()) {
        let (raw_key, raw_value) = pair.split_once('=').unwrap_or((pair, ""));
        let Ok(key) = percent_decode(raw_key) else {
            continue;
        };
        let Ok(value) = percent_decode(raw_value) else {
            continue;
        };
        match key.as_str() {
            "cursor" if !value.is_empty() => {
                req.cursor = Some(value);
            }
            "size" => {
                if let Ok(n) = value.parse::<u32>() {
                    req.size = Some(n);
                }
            }
            _ => {}
        }
    }
    req
}

// ── CursorPage<T> ───────────────────────────────────────────────────

/// Paginated response wrapper for cursor-based pagination.
///
/// `CursorPage` serializes to JSON for API responses and is the
/// counterpart to [`Page<T>`] for keyset/seek pagination. Construct
/// one with [`CursorPage::from_overfetched`] after running a single
/// `LIMIT n+1` query.
///
/// # JSON shape
///
/// ```json
/// {
///   "content": [ /* T items */ ],
///   "size": 20,
///   "next_cursor": "eyJpZCI6MTIzfQ",
///   "has_next": true
/// }
/// ```
///
/// `next_cursor` is `null` on the last page (`has_next == false`).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CursorPage<T> {
    /// The items on this page.
    pub content: Vec<T>,
    /// Page size used to produce `content`.
    pub size: u32,
    /// Opaque cursor for the next page, or `None` if this is the last page.
    pub next_cursor: Option<String>,
    /// Whether there is a page after this one.
    pub has_next: bool,
}

impl<T> CursorPage<T> {
    /// Build a page from an over-fetched result set.
    ///
    /// The caller fetches `request.fetch_limit()` rows (one more than
    /// the page size). If that returned the extra row, this method
    /// truncates `content` back to the page size, marks `has_next`
    /// true, and derives `next_cursor` from the *last kept* item via
    /// `cursor_fn`. If fewer rows came back, this is the last page
    /// and `next_cursor` is `None`.
    ///
    /// `cursor_fn` is called at most once.
    ///
    /// # Errors / panics
    ///
    /// If `cursor_fn`'s value fails to serialize (extremely unlikely
    /// for the simple sort-key structs typically used here),
    /// `next_cursor` is set to `None` and `has_next` to `false`. The
    /// page is still returned — the alternative would be a 500 on a
    /// successful query, which is the wrong tradeoff for a list
    /// endpoint.
    #[must_use]
    pub fn from_overfetched<K, F>(items: Vec<T>, request: &CursorRequest, cursor_fn: F) -> Self
    where
        K: Serialize,
        F: FnOnce(&T) -> K,
    {
        Self::from_overfetched_inner(items, request, cursor_fn, |k| Cursor::encode(&k).ok())
    }

    /// Variant of [`Self::from_overfetched`] that signs `next_cursor`
    /// with HMAC-SHA256 using `key`.
    ///
    /// Use this when the cursor payload encodes anything sensitive to
    /// tampering (tenant ids, user scopes). For sort-key-only cursors
    /// (timestamp + id), plain [`Self::from_overfetched`] is fine —
    /// forging an unsigned cursor is equivalent to seeking to an
    /// arbitrary offset, which clients can already do with `?page=N`.
    ///
    /// The corresponding extractor side calls
    /// [`CursorRequest::decode_signed`] with the same key.
    #[must_use]
    pub fn from_overfetched_signed<K, F>(
        items: Vec<T>,
        request: &CursorRequest,
        key: &[u8],
        cursor_fn: F,
    ) -> Self
    where
        K: Serialize,
        F: FnOnce(&T) -> K,
    {
        Self::from_overfetched_inner(items, request, cursor_fn, |k| {
            Cursor::encode_signed(&k, key).ok()
        })
    }

    fn from_overfetched_inner<K, F, E>(
        mut items: Vec<T>,
        request: &CursorRequest,
        cursor_fn: F,
        encode: E,
    ) -> Self
    where
        K: Serialize,
        F: FnOnce(&T) -> K,
        E: FnOnce(K) -> Option<String>,
    {
        let size = request.size();
        let limit = size as usize;
        let has_next = items.len() > limit;
        if has_next {
            items.truncate(limit);
        }
        let next_cursor = if has_next {
            // The boundary row is the last one we kept. Encoding from
            // it (rather than the popped row) means the next query
            // can use a strict inequality and still see every row,
            // even under concurrent inserts that land between pages.
            items.last().map(cursor_fn).and_then(encode)
        } else {
            None
        };
        // If encoding failed for some reason, don't claim a next page
        // we can't actually serve.
        let has_next = has_next && next_cursor.is_some();
        Self {
            content: items,
            size,
            next_cursor,
            has_next,
        }
    }

    /// Build an empty page using the caller's request parameters.
    ///
    /// Useful when a filter short-circuits before hitting the database.
    #[must_use]
    pub const fn empty(request: &CursorRequest) -> Self {
        Self {
            content: Vec::new(),
            size: request.size(),
            next_cursor: None,
            has_next: false,
        }
    }

    /// Transform the content while preserving pagination metadata.
    pub fn map<U, F: FnMut(T) -> U>(self, f: F) -> CursorPage<U> {
        CursorPage {
            content: self.content.into_iter().map(f).collect(),
            size: self.size,
            next_cursor: self.next_cursor,
            has_next: self.has_next,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use axum::Router;
    use axum::body::Body;
    use axum::http::{Request, StatusCode};
    use axum::routing::get;
    use tower::ServiceExt;

    // ── PageRequest coercion ────────────────────────────────────

    #[test]
    fn defaults_when_nothing_provided() {
        let r = PageRequest::default();
        assert_eq!(r.page(), 1);
        assert_eq!(r.size(), DEFAULT_PAGE_SIZE);
        assert_eq!(r.limit(), i64::from(DEFAULT_PAGE_SIZE));
        assert_eq!(r.offset(), 0);
    }

    #[test]
    fn page_zero_is_coerced_to_one() {
        let r = PageRequest::new(0, 10);
        assert_eq!(r.page(), 1);
        assert_eq!(r.offset(), 0);
    }

    #[test]
    fn size_is_clamped_to_max() {
        let r = PageRequest::new(1, 9_999);
        assert_eq!(r.size(), MAX_PAGE_SIZE);
        assert_eq!(r.limit(), i64::from(MAX_PAGE_SIZE));
    }

    #[test]
    fn size_zero_falls_back_to_default() {
        let r = PageRequest::new(3, 0);
        assert_eq!(r.size(), DEFAULT_PAGE_SIZE);
    }

    #[test]
    fn offset_matches_page_and_size() {
        let r = PageRequest::new(3, 25);
        assert_eq!(r.offset(), 50);
        assert_eq!(r.limit(), 25);
    }

    // ── Page metadata ──────────────────────────────────────────

    #[test]
    fn empty_page_has_one_total_page() {
        let req = PageRequest::new(5, 50);
        let page: Page<i32> = Page::empty(&req);
        assert_eq!(page.page, 5);
        assert_eq!(page.size, 50);
        assert_eq!(page.total_elements, 0);
        assert_eq!(page.total_pages, 1);
        assert!(!page.has_next);
        assert!(page.has_previous);
        assert!(page.content.is_empty());
    }

    #[test]
    fn metadata_reflects_middle_page() {
        let req = PageRequest::new(3, 20);
        let page = Page::new(vec![1_i32; 20], 137, &req);
        assert_eq!(page.page, 3);
        assert_eq!(page.size, 20);
        assert_eq!(page.total_elements, 137);
        assert_eq!(page.total_pages, 7); // ceil(137/20) == 7
        assert!(page.has_next);
        assert!(page.has_previous);
    }

    #[test]
    fn metadata_reflects_last_page() {
        let req = PageRequest::new(7, 20);
        let page = Page::new(vec![1_i32; 17], 137, &req);
        assert_eq!(page.total_pages, 7);
        assert!(!page.has_next);
        assert!(page.has_previous);
    }

    #[test]
    fn negative_total_is_treated_as_zero() {
        let page: Page<i32> = Page::new(vec![], -1, &PageRequest::default());
        assert_eq!(page.total_elements, 0);
        assert_eq!(page.total_pages, 1);
    }

    #[test]
    fn map_preserves_metadata() {
        let req = PageRequest::new(2, 10);
        let page = Page::new(vec![1_i32, 2, 3], 25, &req);
        let mapped = page.map(|n| n.to_string());
        assert_eq!(mapped.page, 2);
        assert_eq!(mapped.size, 10);
        assert_eq!(mapped.total_elements, 25);
        assert_eq!(mapped.total_pages, 3);
        assert!(mapped.has_next);
        assert!(mapped.has_previous);
        assert_eq!(mapped.content, vec!["1", "2", "3"]);
    }

    // ── percent_decode ─────────────────────────────────────────

    #[test]
    fn percent_decode_happy_path() {
        assert_eq!(percent_decode("hello%20world").unwrap(), "hello world");
        assert_eq!(percent_decode("a%2Bb").unwrap(), "a+b");
        assert_eq!(percent_decode("%32").unwrap(), "2");
    }

    #[test]
    fn percent_decode_plus_to_space() {
        assert_eq!(percent_decode("hello+world").unwrap(), "hello world");
    }

    #[test]
    fn percent_decode_invalid_hex_is_ignored() {
        assert_eq!(percent_decode("abc%3Gdef").unwrap(), "abc%3Gdef");
        assert_eq!(percent_decode("%%%").unwrap(), "%%%");
    }

    #[test]
    fn percent_decode_incomplete_at_eof() {
        assert_eq!(percent_decode("abc%").unwrap(), "abc%");
        assert_eq!(percent_decode("abc%2").unwrap(), "abc%2");
    }

    // ── JSON serialization ─────────────────────────────────────

    #[test]
    fn page_serializes_to_expected_shape() {
        let req = PageRequest::new(2, 10);
        let page = Page::new(vec!["a", "b"], 25, &req);
        let json = serde_json::to_value(&page).unwrap();
        assert_eq!(json["page"], 2);
        assert_eq!(json["size"], 10);
        assert_eq!(json["total_elements"], 25);
        assert_eq!(json["total_pages"], 3);
        assert_eq!(json["has_next"], true);
        assert_eq!(json["has_previous"], true);
        assert_eq!(json["content"], serde_json::json!(["a", "b"]));
    }

    // ── Extractor tests ────────────────────────────────────────

    async fn echo(page: PageRequest) -> String {
        format!("{}:{}:{}", page.page(), page.size(), page.offset())
    }

    async fn fetch(uri: &str) -> (StatusCode, String) {
        let app = Router::new().route("/items", get(echo));
        let res = app
            .oneshot(Request::builder().uri(uri).body(Body::empty()).unwrap())
            .await
            .unwrap();
        let status = res.status();
        let bytes = axum::body::to_bytes(res.into_body(), usize::MAX)
            .await
            .unwrap();
        (status, String::from_utf8(bytes.to_vec()).unwrap())
    }

    #[tokio::test]
    async fn extractor_uses_defaults_when_query_missing() {
        let (status, body) = fetch("/items").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body, format!("1:{DEFAULT_PAGE_SIZE}:0"));
    }

    #[tokio::test]
    async fn extractor_parses_page_and_size() {
        let (status, body) = fetch("/items?page=4&size=25").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body, "4:25:75");
    }

    #[tokio::test]
    async fn extractor_clamps_size_over_max() {
        let (status, body) = fetch("/items?page=1&size=5000").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body, format!("1:{MAX_PAGE_SIZE}:0"));
    }

    #[tokio::test]
    async fn extractor_coerces_page_zero_to_one() {
        let (status, body) = fetch("/items?page=0&size=10").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body, "1:10:0");
    }

    // ── Malformed input handling ───────────────────────────────
    //
    // A list endpoint should never 400 because of a malformed pager.
    // These cases used to reject through `Query::from_request_parts` —
    // they now fall back to defaults.

    #[tokio::test]
    async fn extractor_ignores_non_numeric_page() {
        let (status, body) = fetch("/items?page=abc&size=10").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body, "1:10:0");
    }

    #[tokio::test]
    async fn extractor_ignores_empty_size() {
        let (status, body) = fetch("/items?page=2&size=").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body, format!("2:{DEFAULT_PAGE_SIZE}:{DEFAULT_PAGE_SIZE}"));
    }

    #[tokio::test]
    async fn extractor_uses_last_value_on_duplicate_keys() {
        let (status, body) = fetch("/items?page=1&page=4&size=5").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body, "4:5:15");
    }

    #[tokio::test]
    async fn extractor_ignores_unknown_keys() {
        let (status, body) = fetch("/items?sort=name&page=2&size=10").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body, "2:10:10");
    }

    #[tokio::test]
    async fn extractor_handles_percent_encoded_values() {
        // `%32` decodes to `2`
        let (status, body) = fetch("/items?page=%32&size=10").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body, "2:10:10");
    }

    #[tokio::test]
    async fn extractor_handles_negative_page_value() {
        // `-1` is not a valid u32 — fall back to the default page.
        let (status, body) = fetch("/items?page=-1&size=10").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body, "1:10:0");
    }

    // ── Cursor: base64url round-trip ───────────────────────────

    #[test]
    fn base64url_encode_known_vectors() {
        // RFC 4648 §10 vectors, with `=` padding stripped (we use base64url-no-pad).
        assert_eq!(base64url_encode(b""), "");
        assert_eq!(base64url_encode(b"f"), "Zg");
        assert_eq!(base64url_encode(b"fo"), "Zm8");
        assert_eq!(base64url_encode(b"foo"), "Zm9v");
        assert_eq!(base64url_encode(b"foob"), "Zm9vYg");
        assert_eq!(base64url_encode(b"fooba"), "Zm9vYmE");
        assert_eq!(base64url_encode(b"foobar"), "Zm9vYmFy");
    }

    #[test]
    fn base64url_uses_url_safe_alphabet() {
        // Bytes 0xFB, 0xEF would produce `+` and `/` in standard base64.
        // The url-safe alphabet uses `-` and `_` instead.
        let encoded = base64url_encode(&[0xFB, 0xEF, 0xFF]);
        assert!(!encoded.contains('+'));
        assert!(!encoded.contains('/'));
        assert!(encoded.contains('-') || encoded.contains('_'));
    }

    #[test]
    fn base64url_round_trip_arbitrary_bytes() {
        for len in 0_u8..=32 {
            let input: Vec<u8> = (0..len).map(|i| i.wrapping_mul(37)).collect();
            let encoded = base64url_encode(&input);
            let decoded = base64url_decode(&encoded).unwrap();
            assert_eq!(decoded, input, "round-trip failed at len {len}");
        }
    }

    #[test]
    fn base64url_decode_rejects_invalid_chars() {
        assert_eq!(base64url_decode("!!!!"), None);
        assert_eq!(base64url_decode("AAAA="), None); // padding not accepted
        assert_eq!(base64url_decode("AAA+"), None); // standard-alphabet char
    }

    #[test]
    fn base64url_decode_rejects_one_trailing_char() {
        // A single base64 char carries only 6 bits — not enough for a byte.
        assert_eq!(base64url_decode("A"), None);
        assert_eq!(base64url_decode("ZmA"), Some(vec![0x66, 0x60]));
    }

    // ── Cursor: encode/decode ──────────────────────────────────

    #[derive(Serialize, Deserialize, PartialEq, Debug)]
    struct PostKey {
        created_at: String,
        id: i64,
    }

    #[test]
    fn cursor_round_trip_preserves_payload() {
        let key = PostKey {
            created_at: "2026-04-27T12:00:00Z".to_string(),
            id: 12_345,
        };
        let token = Cursor::encode(&key).unwrap();
        let decoded: PostKey = Cursor::decode(&token).unwrap();
        assert_eq!(decoded, key);
    }

    #[test]
    fn cursor_token_is_url_safe() {
        // Pick a payload that contains JSON characters which would
        // otherwise need percent-encoding (`{`, `}`, `:`, `"`).
        let key = PostKey {
            created_at: "2026-04-27T12:00:00Z".to_string(),
            id: 1,
        };
        let token = Cursor::encode(&key).unwrap();
        // Only chars from the base64url alphabet — no `+`, `/`, `=`, `{`, `:`.
        assert!(
            token
                .bytes()
                .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_')
        );
    }

    #[test]
    fn cursor_decode_returns_none_for_garbage() {
        // Not base64url at all.
        let decoded: Option<PostKey> = Cursor::decode("!!!not a token!!!");
        assert!(decoded.is_none());
    }

    #[test]
    fn cursor_decode_returns_none_for_wrong_schema() {
        // Valid base64url, valid JSON, but doesn't match the target type.
        let other = serde_json::json!({"unrelated": "value"});
        let token = Cursor::encode(&other).unwrap();
        let decoded: Option<PostKey> = Cursor::decode(&token);
        assert!(decoded.is_none());
    }

    // ── CursorRequest coercion ─────────────────────────────────

    #[test]
    fn cursor_request_defaults_when_empty() {
        let r = CursorRequest::default();
        assert!(r.cursor().is_none());
        assert_eq!(r.size(), DEFAULT_PAGE_SIZE);
        assert_eq!(r.limit(), i64::from(DEFAULT_PAGE_SIZE));
        assert_eq!(r.fetch_limit(), i64::from(DEFAULT_PAGE_SIZE) + 1);
    }

    #[test]
    fn cursor_request_clamps_size_to_max() {
        let r = CursorRequest::new(None, 9_999);
        assert_eq!(r.size(), MAX_PAGE_SIZE);
        assert_eq!(r.fetch_limit(), i64::from(MAX_PAGE_SIZE) + 1);
    }

    #[test]
    fn cursor_request_zero_size_falls_back_to_default() {
        let r = CursorRequest::new(None, 0);
        assert_eq!(r.size(), DEFAULT_PAGE_SIZE);
    }

    #[test]
    fn cursor_request_decode_helper_returns_none_when_missing() {
        let r = CursorRequest::default();
        let decoded: Option<PostKey> = r.decode();
        assert!(decoded.is_none());
    }

    #[test]
    fn cursor_request_decode_helper_round_trips() {
        let key = PostKey {
            created_at: "2026-04-27T00:00:00Z".to_string(),
            id: 7,
        };
        let token = Cursor::encode(&key).unwrap();
        let r = CursorRequest::new(Some(token), 10);
        let decoded: PostKey = r.decode().unwrap();
        assert_eq!(decoded, key);
    }

    // ── CursorPage from_overfetched ────────────────────────────

    #[test]
    fn cursor_page_signals_no_next_when_under_limit() {
        let req = CursorRequest::new(None, 5);
        let items = vec![1_i32, 2, 3]; // fewer than size
        let page = CursorPage::from_overfetched(items, &req, |&n| serde_json::json!({"id": n}));
        assert_eq!(page.content, vec![1, 2, 3]);
        assert!(!page.has_next);
        assert!(page.next_cursor.is_none());
        assert_eq!(page.size, 5);
    }

    #[test]
    fn cursor_page_signals_no_next_at_exact_limit() {
        let req = CursorRequest::new(None, 3);
        // Caller fetched limit+1 = 4, but only got 3 — last page.
        let items = vec![1_i32, 2, 3];
        let page = CursorPage::from_overfetched(items, &req, |&n| serde_json::json!({"id": n}));
        assert_eq!(page.content.len(), 3);
        assert!(!page.has_next);
        assert!(page.next_cursor.is_none());
    }

    #[test]
    fn cursor_page_truncates_overflow_and_emits_next_cursor() {
        let req = CursorRequest::new(None, 3);
        // Caller fetched limit+1 = 4 rows.
        let items = vec![1_i32, 2, 3, 4];
        let page = CursorPage::from_overfetched(items, &req, |&n| serde_json::json!({"id": n}));
        // Only `size` items kept.
        assert_eq!(page.content, vec![1, 2, 3]);
        assert!(page.has_next);
        let token = page.next_cursor.as_ref().expect("next cursor present");
        // Cursor encodes the *last kept* row (id=3), not the popped row (id=4).
        // This matters: the next page filters with a strict inequality
        // against this id, which keeps zero-duplicate behaviour even
        // if a new row gets inserted between page boundaries.
        let decoded: serde_json::Value = Cursor::decode(token).unwrap();
        assert_eq!(decoded, serde_json::json!({"id": 3}));
    }

    #[test]
    fn cursor_page_empty_helper() {
        let req = CursorRequest::new(None, 10);
        let page: CursorPage<i32> = CursorPage::empty(&req);
        assert!(page.content.is_empty());
        assert_eq!(page.size, 10);
        assert!(!page.has_next);
        assert!(page.next_cursor.is_none());
    }

    #[test]
    fn cursor_page_map_preserves_metadata() {
        let req = CursorRequest::new(None, 2);
        let items = vec![1_i32, 2, 3]; // overfetch by 1
        let page = CursorPage::from_overfetched(items, &req, |&n| serde_json::json!({"id": n}));
        let mapped = page.map(|n| n.to_string());
        assert_eq!(mapped.content, vec!["1", "2"]);
        assert!(mapped.has_next);
        assert!(mapped.next_cursor.is_some());
        assert_eq!(mapped.size, 2);
    }

    #[test]
    fn cursor_page_serializes_to_expected_shape() {
        let req = CursorRequest::new(None, 2);
        let items = vec!["a", "b", "c"];
        let page = CursorPage::from_overfetched(items, &req, |s| serde_json::json!({"key": s}));
        let json = serde_json::to_value(&page).unwrap();
        assert_eq!(json["size"], 2);
        assert_eq!(json["has_next"], true);
        assert!(json["next_cursor"].is_string());
        assert_eq!(json["content"], serde_json::json!(["a", "b"]));
    }

    #[test]
    fn cursor_page_last_page_serializes_null_cursor() {
        let req = CursorRequest::new(None, 5);
        let items = vec!["only"];
        let page = CursorPage::from_overfetched(items, &req, |s| serde_json::json!({"key": s}));
        let json = serde_json::to_value(&page).unwrap();
        assert_eq!(json["has_next"], false);
        assert!(json["next_cursor"].is_null());
    }

    // ── CursorRequest extractor ────────────────────────────────

    async fn cursor_echo(req: CursorRequest) -> String {
        format!(
            "{}|{}|{}",
            req.cursor().unwrap_or("-"),
            req.size(),
            req.fetch_limit(),
        )
    }

    async fn fetch_cursor(uri: &str) -> (StatusCode, String) {
        let app = Router::new().route("/feed", get(cursor_echo));
        let res = app
            .oneshot(Request::builder().uri(uri).body(Body::empty()).unwrap())
            .await
            .unwrap();
        let status = res.status();
        let bytes = axum::body::to_bytes(res.into_body(), usize::MAX)
            .await
            .unwrap();
        (status, String::from_utf8(bytes.to_vec()).unwrap())
    }

    #[tokio::test]
    async fn cursor_extractor_uses_defaults_when_query_missing() {
        let (status, body) = fetch_cursor("/feed").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(
            body,
            format!("-|{DEFAULT_PAGE_SIZE}|{}", DEFAULT_PAGE_SIZE + 1)
        );
    }

    #[tokio::test]
    async fn cursor_extractor_parses_cursor_and_size() {
        let (status, body) = fetch_cursor("/feed?cursor=abc123&size=5").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body, "abc123|5|6");
    }

    #[tokio::test]
    async fn cursor_extractor_clamps_size_over_max() {
        let (status, body) = fetch_cursor("/feed?cursor=t&size=9999").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body, format!("t|{MAX_PAGE_SIZE}|{}", MAX_PAGE_SIZE + 1));
    }

    #[tokio::test]
    async fn cursor_extractor_ignores_empty_cursor() {
        // Empty `cursor=` should be treated as "no cursor", not as `Some("")`.
        let (status, body) = fetch_cursor("/feed?cursor=&size=10").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body, "-|10|11");
    }

    #[tokio::test]
    async fn cursor_extractor_does_not_400_on_malformed_size() {
        let (status, body) = fetch_cursor("/feed?cursor=t&size=abc").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(
            body,
            format!("t|{DEFAULT_PAGE_SIZE}|{}", DEFAULT_PAGE_SIZE + 1)
        );
    }

    #[tokio::test]
    async fn cursor_extractor_handles_percent_encoded_token() {
        // base64url tokens never need percent-encoding, but a paranoid
        // client might do it anyway. `%2D` decodes to `-`.
        let (status, body) = fetch_cursor("/feed?cursor=ab%2Dcd&size=2").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body, "ab-cd|2|3");
    }

    // ── Concurrent-insert simulation ───────────────────────────
    //
    // The story's success metric is "zero duplicate items during
    // concurrent inserts". This test simulates a feed where a new row
    // arrives between the first and second page request, and verifies
    // the keyset filter still returns every original row exactly once.

    #[test]
    fn concurrent_inserts_do_not_cause_duplicates() {
        // Sort by (created_at desc, id desc) — this is the recommended
        // tie-breaker pattern from the docs.
        #[derive(Clone, Debug, PartialEq, Eq)]
        struct Row {
            id: i64,
            created_at: i64,
        }
        #[derive(Serialize, Deserialize)]
        struct Key {
            created_at: i64,
            id: i64,
        }

        let mut table: Vec<Row> = (1..=5)
            .map(|id| Row {
                id,
                created_at: 1_000 - id, // older as id grows
            })
            .collect();
        // Sort newest-first so id=1 is first.
        table.sort_by_key(|r| std::cmp::Reverse((r.created_at, r.id)));

        // First request: no cursor, size=2.
        let req1 = CursorRequest::new(None, 2);
        let fetch1 = usize::try_from(req1.fetch_limit()).unwrap();
        let fetched1: Vec<Row> = table.iter().take(fetch1).cloned().collect();
        let page1 = CursorPage::from_overfetched(fetched1, &req1, |r| Key {
            created_at: r.created_at,
            id: r.id,
        });
        let cursor1 = page1.next_cursor.clone().expect("page 1 has next");
        assert_eq!(page1.content.len(), 2);

        // ── Concurrent insert lands BEFORE the next request ──────
        // A new row is inserted with the highest created_at, so it
        // would appear on a fresh page 1 — but our cursor pagination
        // is keyset-based, so the second request must skip it.
        table.insert(
            0,
            Row {
                id: 99,
                created_at: 9_999,
            },
        );

        // Second request: cursor from page 1, size=2.
        let req2 = CursorRequest::new(Some(cursor1), 2);
        let key: Key = req2.decode().unwrap();
        let fetch2 = usize::try_from(req2.fetch_limit()).unwrap();

        // Apply the keyset filter: rows that come *after* the cursor
        // in the (created_at desc, id desc) ordering, then take the
        // overfetch window.
        let fetched2: Vec<Row> = table
            .iter()
            .filter(|r| {
                r.created_at < key.created_at || (r.created_at == key.created_at && r.id < key.id)
            })
            .take(fetch2)
            .cloned()
            .collect();
        let page2 = CursorPage::from_overfetched(fetched2, &req2, |r| Key {
            created_at: r.created_at,
            id: r.id,
        });

        // Combine the two pages. No row should appear twice, and the
        // newly-inserted id=99 must NOT show up (the user already
        // scrolled past where it would have appeared).
        let mut all: Vec<Row> = page1.content;
        all.extend(page2.content);
        let mut ids: Vec<i64> = all.iter().map(|r| r.id).collect();
        let original_len = ids.len();
        ids.sort_unstable();
        ids.dedup();
        assert_eq!(ids.len(), original_len, "no duplicates across pages");
        assert!(
            !all.iter().any(|r| r.id == 99),
            "concurrently-inserted row not duplicated"
        );
    }

    // ── Signed cursors ─────────────────────────────────────────

    const TEST_KEY: &[u8] = b"test-signing-key-do-not-use-in-prod";

    #[derive(Serialize, Deserialize, PartialEq, Debug)]
    struct ScopedCursor {
        tenant_id: i64,
        cursor_id: i64,
    }

    #[test]
    fn signed_cursor_round_trip() {
        let payload = ScopedCursor {
            tenant_id: 42,
            cursor_id: 7,
        };
        let token = Cursor::encode_signed(&payload, TEST_KEY).unwrap();
        // Token shape: <payload>.<sig>
        assert!(token.contains('.'));
        let decoded: ScopedCursor = Cursor::decode_signed(&token, TEST_KEY).unwrap();
        assert_eq!(decoded, payload);
    }

    #[test]
    fn signed_cursor_rejects_tampered_payload() {
        let payload = ScopedCursor {
            tenant_id: 42,
            cursor_id: 7,
        };
        let token = Cursor::encode_signed(&payload, TEST_KEY).unwrap();
        // Forge: re-encode a payload with tenant_id=99 but reuse the
        // original signature.
        let forged_payload = ScopedCursor {
            tenant_id: 99,
            cursor_id: 7,
        };
        let forged_b64 = base64url_encode(&serde_json::to_vec(&forged_payload).unwrap());
        let (_, sig_b64) = token.split_once('.').unwrap();
        let forged_token = format!("{forged_b64}.{sig_b64}");
        let decoded: Option<ScopedCursor> = Cursor::decode_signed(&forged_token, TEST_KEY);
        assert!(decoded.is_none(), "tampered cursor must not verify");
    }

    #[test]
    fn signed_cursor_rejects_wrong_key() {
        let payload = ScopedCursor {
            tenant_id: 42,
            cursor_id: 7,
        };
        let token = Cursor::encode_signed(&payload, TEST_KEY).unwrap();
        let decoded: Option<ScopedCursor> = Cursor::decode_signed(&token, b"different-key");
        assert!(decoded.is_none());
    }

    #[test]
    fn signed_cursor_rejects_unsigned_token() {
        // A plain (unsigned) token should not verify against the
        // signed decoder — even though it's structurally valid JSON.
        let payload = ScopedCursor {
            tenant_id: 42,
            cursor_id: 7,
        };
        let unsigned = Cursor::encode(&payload).unwrap();
        let decoded: Option<ScopedCursor> = Cursor::decode_signed(&unsigned, TEST_KEY);
        assert!(
            decoded.is_none(),
            "unsigned token must not pass signed verification"
        );
    }

    #[test]
    fn signed_cursor_rejects_missing_signature_segment() {
        // Token without a `.` separator.
        let decoded: Option<ScopedCursor> = Cursor::decode_signed("just-some-bytes", TEST_KEY);
        assert!(decoded.is_none());
    }

    #[test]
    fn signed_cursor_rejects_garbage() {
        let decoded: Option<ScopedCursor> = Cursor::decode_signed("!!!.!!!", TEST_KEY);
        assert!(decoded.is_none());
    }

    #[test]
    fn cursor_request_decode_signed_returns_none_when_missing() {
        let r = CursorRequest::default();
        let decoded: Option<ScopedCursor> = r.decode_signed(TEST_KEY);
        assert!(decoded.is_none());
    }

    #[test]
    fn cursor_request_decode_signed_round_trips() {
        let payload = ScopedCursor {
            tenant_id: 42,
            cursor_id: 7,
        };
        let token = Cursor::encode_signed(&payload, TEST_KEY).unwrap();
        let r = CursorRequest::new(Some(token), 10);
        let decoded: ScopedCursor = r.decode_signed(TEST_KEY).unwrap();
        assert_eq!(decoded, payload);
    }

    #[test]
    fn cursor_page_from_overfetched_signed_emits_signed_token() {
        let req = CursorRequest::new(None, 2);
        let items = vec![1_i32, 2, 3]; // overfetch by 1
        let page = CursorPage::from_overfetched_signed(items, &req, TEST_KEY, |&n| ScopedCursor {
            tenant_id: 42,
            cursor_id: i64::from(n),
        });
        assert!(page.has_next);
        let token = page.next_cursor.as_ref().unwrap();
        assert!(token.contains('.'), "signed token format is payload.sig");
        // Round-trip through the signed decoder.
        let key: ScopedCursor = Cursor::decode_signed(token, TEST_KEY).unwrap();
        assert_eq!(key.cursor_id, 2); // boundary = last kept item
        // Plain decoder must NOT happen to extract a structurally
        // valid value, because the token contains the signature suffix.
        let mishandled: Option<ScopedCursor> = Cursor::decode(token);
        assert!(mishandled.is_none());
    }

    #[test]
    fn signed_cursor_signature_is_constant_time_compared() {
        // Smoke test: same key, same input → identical sig. Different
        // key → different sig. (Constant-time-ness itself is not
        // observable from this test; we're just exercising the path.)
        let p = ScopedCursor {
            tenant_id: 1,
            cursor_id: 1,
        };
        let a = Cursor::encode_signed(&p, b"k1").unwrap();
        let b = Cursor::encode_signed(&p, b"k1").unwrap();
        let c = Cursor::encode_signed(&p, b"k2").unwrap();
        assert_eq!(a, b);
        assert_ne!(a, c);
    }
}