# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/across-what-keyset-cursor-and-limit-ceiling-boundaries-must-an-instance-paginate-collection-queries-to-prevent-offset-scan-degradation-and-unbounded-result-oom.md
# Remnant Warrant:    https://corpusfetch.com/c/across-what-keyset-cursor-and-limit-ceiling-boundaries-must-an-instance-paginate-collection-queries-to-prevent-offset-scan-degradation-and-unbounded-result-oom.warrant.json
# Markdown SHA-256:   4deb613608acea9109022797c8e14e44bbde517a8c79e5521f40fb3281114676
# Code AST SHA-256:   1293378c9ef2c33b4ecc0f712fd1bfd56ae92a6a6e3cf6cfe669ca0a9f406c21
# Verified Classes:   KeysetPage
# Defensive Guards:   6 fail-closed invariants (zero external deps)

import base64
import json
import sqlite3
from dataclasses import dataclass
from typing import Any, Generic, Sequence, TypeVar

T = TypeVar("T")

MAX_API_PAGE_LIMIT = 100
DEFAULT_API_PAGE_LIMIT = 25


@dataclass(frozen=True)
class KeysetPage(Generic[T]):
    items: list[T]
    next_cursor: str | None
    has_more: bool


def clamp_page_limit(
    requested_limit: int | None,
    *,
    default_limit: int = DEFAULT_API_PAGE_LIMIT,
    max_limit: int = MAX_API_PAGE_LIMIT,
) -> int:
    '''
    Clamps a caller-supplied pagination limit into `[1, max_limit]`.
    Rejects booleans and negative/zero bounds.
    '''
    if requested_limit is None:
        return default_limit
    if isinstance(requested_limit, bool) or not isinstance(requested_limit, int):
        raise TypeError("Pagination limit must be an integer")
    if requested_limit < 1:
        raise ValueError("Pagination limit must be >= 1")
    return min(requested_limit, max_limit)


def encode_keyset_cursor(sort_value: str | int, tiebreaker_id: str | int) -> str:
    '''Encodes a compound `(sort_value, tiebreaker_id)` position into an opaque URL-safe cursor.'''
    payload = json.dumps({"s": sort_value, "id": tiebreaker_id}, separators=(",", ":"))
    return base64.urlsafe_b64encode(payload.encode("utf-8")).rstrip(b"=").decode("ascii")


def decode_keyset_cursor(cursor_str: str) -> tuple[str | int, str | int]:
    '''Decodes and validates an opaque keyset cursor into `(sort_value, tiebreaker_id)`.'''
    if not cursor_str or len(cursor_str) > 512:
        raise ValueError("Invalid or oversized pagination cursor")
    try:
        padding = "=" * ((4 - len(cursor_str) % 4) % 4)
        raw = base64.urlsafe_b64decode((cursor_str + padding).encode("ascii"))
        data = json.loads(raw.decode("utf-8"))
        sort_val = data["s"]
        tie_id = data["id"]
        if isinstance(sort_val, bool) or isinstance(tie_id, bool):
            raise ValueError("Boolean cursor values are forbidden")
        if not isinstance(sort_val, (str, int)) or not isinstance(tie_id, (str, int)):
            raise ValueError("Cursor components must be str or int")
        return sort_val, tie_id
    except Exception as exc:
        raise ValueError(f"Malformed pagination cursor: {cursor_str!r}") from exc


def fetch_events_keyset_page(
    conn: sqlite3.Connection,
    *,
    limit: int | None = None,
    cursor: str | None = None,
) -> KeysetPage[dict[str, Any]]:
    '''
    Executes O(log N) keyset pagination over `events(created_at_ms, id)` using
    a composite index `CREATE INDEX idx_events_sort ON events(created_at_ms ASC, id ASC)`.
    Fetches `safe_limit + 1` rows to compute `has_more` without running COUNT(*).
    '''
    safe_limit = clamp_page_limit(limit)
    fetch_count = safe_limit + 1

    if cursor is None:
        rows = conn.execute(
            "SELECT id, created_at_ms, payload "
            "FROM events "
            "ORDER BY created_at_ms ASC, id ASC "
            "LIMIT ?",
            (fetch_count,),
        ).fetchall()
    else:
        last_ts, last_id = decode_keyset_cursor(cursor)
        # Gate: Explicit tuple comparison seeks directly in the composite B-tree index
        rows = conn.execute(
            "SELECT id, created_at_ms, payload "
            "FROM events "
            "WHERE (created_at_ms > ?) OR (created_at_ms = ? AND id > ?) "
            "ORDER BY created_at_ms ASC, id ASC "
            "LIMIT ?",
            (last_ts, last_ts, last_id, fetch_count),
        ).fetchall()

    has_more = len(rows) > safe_limit
    page_rows = rows[:safe_limit]

    items = [
        {"id": r[0], "created_at_ms": r[1], "payload": r[2]}
        for r in page_rows
    ]

    next_cursor: str | None = None
    if has_more and items:
        tail = items[-1]
        next_cursor = encode_keyset_cursor(tail["created_at_ms"], tail["id"])

    return KeysetPage(items=items, next_cursor=next_cursor, has_more=has_more)