# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/across-what-composite-tie-breaker-and-stable-cursor-gates-must-an-instance-merge-distributed-queries-to-prevent-pagination-drift-and-record-duplication.md
# Remnant Warrant:    https://corpusfetch.com/c/across-what-composite-tie-breaker-and-stable-cursor-gates-must-an-instance-merge-distributed-queries-to-prevent-pagination-drift-and-record-duplication.warrant.json
# Markdown SHA-256:   6be9c06c39938a00a9913b94a4586ba82e37fa3558c41c96594caf0e4773adde
# Code AST SHA-256:   348b1705bb162331d5c54e801e985e78e8e5a7178e94883333777375c6889105
# Verified Classes:   KeysetPageResult, KeysetPaginationEngine
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import base64
import json
import sqlite3
from dataclasses import dataclass
from typing import Any


@dataclass(frozen=True)
class KeysetPageResult:
    items: list[dict[str, Any]]
    next_cursor: str | None
    has_more: bool


class KeysetPaginationEngine:
    '''
    Implements deterministic keyset cursor pagination with composite tie-breakers:
    1. Enforces `ORDER BY sort_col DESC, id DESC` (or `ASC, ASC`).
    2. Encodes composite boundary tuples `(sort_val, id)` into opaque base64 cursors.
    3. Executes $O(1)$ index range lookups via tuple comparison predicates.
    '''

    def __init__(self, conn: sqlite3.Connection) -> None:
        self.conn = conn
        self._init_schema()

    def _init_schema(self) -> None:
        with self.conn:
            self.conn.execute(
                "CREATE TABLE IF NOT EXISTS paginated_events ("
                "  id INTEGER PRIMARY KEY AUTOINCREMENT,"
                "  created_at TEXT NOT NULL,"
                "  payload TEXT NOT NULL"
                ")"
            )
            self.conn.execute(
                "CREATE INDEX IF NOT EXISTS idx_paginated_composite "
                "ON paginated_events (created_at DESC, id DESC)"
            )

    @staticmethod
    def encode_cursor(sort_val: Any, record_id: int) -> str:
        raw = json.dumps([sort_val, record_id], separators=(",", ":"))
        return base64.urlsafe_b64encode(raw.encode("utf-8")).decode("utf-8")

    @staticmethod
    def decode_cursor(cursor_str: str) -> tuple[Any, int] | None:
        if not cursor_str:
            return None
        try:
            raw = base64.urlsafe_b64decode(cursor_str.encode("utf-8")).decode("utf-8")
            data = json.loads(raw)
            if isinstance(data, list) and len(data) == 2:
                return (data[0], int(data[1]))
        except Exception:
            pass
        return None

    def fetch_page(
        self,
        *,
        limit: int = 20,
        cursor: str | None = None,
        sort_desc: bool = True,
    ) -> KeysetPageResult:
        '''
        Gate 3: Queries table using a composite tuple keyset cursor
        (`WHERE (created_at, id) < (?, ?)` for DESC).
        '''
        fetch_limit = max(1, limit) + 1  # fetch one extra to determine `has_more`
        decoded = self.decode_cursor(cursor)

        direction = "<" if sort_desc else ">"
        order_dir = "DESC" if sort_desc else "ASC"

        query = [
            "SELECT id, created_at, payload FROM paginated_events"
        ]
        params: list[Any] = []

        if decoded is not None:
            last_val, last_id = decoded
            query.append(f"WHERE (created_at, id) {direction} (?, ?)")
            params.extend([last_val, last_id])

        query.append(f"ORDER BY created_at {order_dir}, id {order_dir}")
        query.append("LIMIT ?")
        params.append(fetch_limit)

        cursor_obj = self.conn.execute(" ".join(query), params)
        rows = cursor_obj.fetchall()

        items: list[dict[str, Any]] = []
        for r in rows:
            items.append({
                "id": int(r[0]),
                "created_at": r[1],
                "payload": r[2],
            })

        has_more = len(items) > limit
        if has_more:
            items.pop()  # remove extra buffer item

        next_cursor: str | None = None
        if has_more and items:
            last_item = items[-1]
            next_cursor = self.encode_cursor(last_item["created_at"], last_item["id"])

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

    @classmethod
    def verify_boundary_invariants(cls) -> bool:
        encoded = cls.encode_cursor("2026-10-20T12:00:00Z", 42)
        ts, tie_id = cls.decode_cursor(encoded)
        if ts != "2026-10-20T12:00:00Z" or tie_id != 42:
            raise ValueError("Composite cursor round-trip invariant violated")
        assert ts == "2026-10-20T12:00:00Z" and tie_id == 42
        return True