//! Reusable keyset (cursor) pagination.
//!
//! Keyset, not offset: a paged query fetches only `limit + 1` rows via an
//! indexed range scan on `(sort_col, id)`, and derives the next page's cursor
//! from the last row, it never fetches the full set and slices in memory. The
//! cursor is **opaque** (base64) so a consumer's client contract does not change
//! if the encoded sort key later does.
//!
//! When to reach for this vs. offset:
//! - **Keyset (this module)**, forward "load more" / infinite-scroll lists,
//!   large or hot tables, anywhere a full-scan-per-page would be wrong.
//! - **Offset**, a numbered-page admin UI that needs "page 7 of 42" and a
//!   total count (e.g. the admin user list). Offset is correct there; keyset
//!   cannot express page numbers or totals. Don't convert those.
//!
//! Consumer contract: a keyset query MUST have an index on `(<sort_col>, id)` in
//! its sort direction, or the `LIMIT n+1` degrades to sort-the-world. The
//! generic machinery here (cursor encode/decode, page assembly, next-page
//! detection) is written once; each consumer writes ~3 lines of SQL:
//!
//! ```ignore
//! // WHERE <base predicates>
//! //   AND ($2::timestamptz IS NULL OR (created_at, id) < ($2, $3))
//! // ORDER BY created_at DESC, id DESC
//! // LIMIT $1                      -- bind params.fetch_limit()
//! let rows = sqlx::query_as::<_, Row>(SQL)
//!     .bind(params.fetch_limit())
//!     .bind(params.after.as_ref().map(|c| c.ts))
//!     .bind(params.after.as_ref().map(|c| c.id))
//!     .fetch_all(pool).await?;
//! Ok(Page::from_rows(rows, &params, |r| Cursor::new(r.created_at, r.id)))
//! ```

use base64::Engine;
use chrono::{DateTime, Utc};
use serde::Serialize;
use uuid::Uuid;

/// Default page size when a consumer does not specify one.
pub const DEFAULT_PAGE_SIZE: u32 = 50;
/// Upper clamp on page size so a client can't request an unbounded page.
pub const MAX_PAGE_SIZE: u32 = 200;

/// An opaque forward cursor: the `(sort-timestamp, id)` of the last row on a
/// page. Ordering is `(ts, id)` DESC, with `id` as the unique tiebreaker so the
/// keyset is a total order (no row is skipped or repeated across pages when two
/// rows share a timestamp).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Cursor {
    pub ts: DateTime<Utc>,
    pub id: Uuid,
}

impl Cursor {
    pub fn new(ts: DateTime<Utc>, id: Uuid) -> Self {
        Self { ts, id }
    }

    /// Encode as an opaque base64 token: `<unix_millis>:<uuid>`. The internal
    /// format is deliberately not part of any contract, decode tolerates only
    /// what encode produces, and both can change without touching callers.
    pub fn encode(&self) -> String {
        let raw = format!("{}:{}", self.ts.timestamp_millis(), self.id);
        base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(raw)
    }

    // (see `Serialize` impl below: a Cursor serializes AS its opaque token, so a
    // `Page.next` field emits the JSON string a client passes straight back.)

    /// Decode a token produced by [`encode`](Self::encode). Returns `None` for
    /// any malformed / tampered token, a bad cursor is treated as "start from
    /// the beginning", never an error (a client should never be able to 500 the
    /// server by editing a cursor).
    pub fn decode(token: &str) -> Option<Self> {
        let bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
            .decode(token)
            .ok()?;
        let raw = String::from_utf8(bytes).ok()?;
        let (millis, id) = raw.split_once(':')?;
        let ts = DateTime::<Utc>::from_timestamp_millis(millis.parse().ok()?)?;
        Some(Self::new(ts, id.parse().ok()?))
    }
}

impl Serialize for Cursor {
    /// A cursor serializes as its opaque token string, so `Page.next` renders as
    /// the exact value a client hands back as `?after=`.
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        serializer.serialize_str(&self.encode())
    }
}

/// Parameters for a keyset page request: where to start, and how many rows.
#[derive(Debug, Clone)]
pub struct PageParams {
    pub after: Option<Cursor>,
    pub limit: u32,
}

impl PageParams {
    /// Build from a raw client cursor token + optional page size. A malformed
    /// cursor silently starts from the beginning; the size is clamped to
    /// `[1, MAX_PAGE_SIZE]`.
    pub fn new(after_token: Option<&str>, limit: Option<u32>) -> Self {
        Self {
            after: after_token.and_then(Cursor::decode),
            limit: limit.unwrap_or(DEFAULT_PAGE_SIZE).clamp(1, MAX_PAGE_SIZE),
        }
    }

    /// The SQL `LIMIT` to bind: one more than the page size, so a returned extra
    /// row signals "there is a next page" without a second COUNT query.
    pub fn fetch_limit(&self) -> i64 {
        i64::from(self.limit) + 1
    }
}

/// One page of results plus the cursor to fetch the next page (`None` = last).
#[derive(Debug, Serialize)]
pub struct Page<T> {
    pub items: Vec<T>,
    pub next: Option<Cursor>,
}

impl<T> Page<T> {
    /// Assemble a page from up to `limit + 1` fetched rows (ordered by the same
    /// key the query used). If the extra row is present there is a next page:
    /// drop it and take the next cursor from the last *kept* row. `cursor_of`
    /// extracts the `(ts, id)` sort key from a row.
    pub fn from_rows(
        mut rows: Vec<T>,
        params: &PageParams,
        cursor_of: impl Fn(&T) -> Cursor,
    ) -> Self {
        let has_more = rows.len() as u32 > params.limit;
        rows.truncate(params.limit as usize);
        let next = if has_more {
            rows.last().map(&cursor_of)
        } else {
            None
        };
        Page { items: rows, next }
    }

    /// The opaque next-page token for a client, if any.
    pub fn next_token(&self) -> Option<String> {
        self.next.map(|c| c.encode())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn ts(millis: i64) -> DateTime<Utc> {
        DateTime::<Utc>::from_timestamp_millis(millis).unwrap()
    }

    #[test]
    fn cursor_round_trips_through_opaque_token() {
        let c = Cursor::new(ts(1_700_000_000_123), Uuid::from_u128(42));
        let token = c.encode();
        // Opaque: not the raw value.
        assert_ne!(token, "1700000000123:00000000-0000-0000-0000-00000000002a");
        assert_eq!(Cursor::decode(&token), Some(c));
    }

    #[test]
    fn malformed_cursor_decodes_to_none_not_error() {
        assert_eq!(Cursor::decode("not-base64!!!"), None);
        assert_eq!(Cursor::decode(""), None);
        // Valid base64 but wrong shape.
        let garbage = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode("no-colon");
        assert_eq!(Cursor::decode(&garbage), None);
    }

    #[test]
    fn params_clamp_and_default_page_size() {
        assert_eq!(PageParams::new(None, None).limit, DEFAULT_PAGE_SIZE);
        assert_eq!(PageParams::new(None, Some(0)).limit, 1);
        assert_eq!(PageParams::new(None, Some(10_000)).limit, MAX_PAGE_SIZE);
        assert_eq!(PageParams::new(None, Some(25)).fetch_limit(), 26);
    }

    #[test]
    fn bad_cursor_token_starts_from_beginning() {
        let p = PageParams::new(Some("tampered"), Some(20));
        assert!(
            p.after.is_none(),
            "a bad cursor must not error, just restart"
        );
    }

    #[test]
    fn from_rows_detects_next_page_and_trims_extra() {
        let params = PageParams::new(None, Some(3)); // fetch_limit = 4
        // 4 rows returned for a page size of 3 => there IS a next page.
        let rows: Vec<(i64, Uuid)> = (0..4).map(|i| (i, Uuid::from_u128(i as u128))).collect();
        let page = Page::from_rows(rows, &params, |(m, id)| Cursor::new(ts(*m), *id));
        assert_eq!(page.items.len(), 3, "the extra probe row is trimmed");
        // Next cursor is the last KEPT row (index 2), not the dropped probe row.
        assert_eq!(page.next, Some(Cursor::new(ts(2), Uuid::from_u128(2))));
    }

    #[test]
    fn from_rows_last_page_has_no_next() {
        let params = PageParams::new(None, Some(5)); // fetch_limit = 6
        // Only 3 rows for a page size of 5 => last page.
        let rows: Vec<(i64, Uuid)> = (0..3).map(|i| (i, Uuid::from_u128(i as u128))).collect();
        let page = Page::from_rows(rows, &params, |(m, id)| Cursor::new(ts(*m), *id));
        assert_eq!(page.items.len(), 3);
        assert_eq!(page.next, None);
        assert_eq!(page.next_token(), None);
    }
}