app/modules/search/api/cursor.py

import base64
import hashlib
import json
import secrets
from typing import Any

from app.modules.search.domain.models import SearchResourceType
CURSOR_VERSION = 1
MAX_SEARCH_OFFSET = 1000


def build_search_cursor_context(payload: dict[str, Any]) -> str:
    encoded = json.dumps(
        payload,
        ensure_ascii=True,
        separators=(",", ":"),
        sort_keys=True,
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()[:24]


def encode_search_cursor(
    resource_type: SearchResourceType,
    context_fingerprint: str,
    offset: int,
) -> str:
    payload = {
        "v": CURSOR_VERSION,
        "r": resource_type.value,
        "c": context_fingerprint,
        "o": offset,
    }
    encoded = base64.urlsafe_b64encode(
        json.dumps(payload, separators=(",", ":"), sort_keys=True).encode("utf-8")
    )
    return encoded.decode("ascii").rstrip("=")


def decode_search_cursor(
    cursor: str,
    resource_type: SearchResourceType,
    context_fingerprint: str,
) -> int:
    if not cursor or len(cursor) > 512:
        raise ValueError("cursor has an invalid length")
    try:
        padding = "=" * (-len(cursor) % 4)
        decoded = base64.urlsafe_b64decode((cursor + padding).encode("ascii"))
        payload: Any = json.loads(decoded.decode("utf-8"))
    except (ValueError, UnicodeError, json.JSONDecodeError) as exc:
        raise ValueError("cursor is not valid") from exc

    if not isinstance(payload, dict) or payload.get("v") != CURSOR_VERSION:
        raise ValueError("cursor version is not supported")
    if payload.get("r") != resource_type.value:
        raise ValueError("cursor belongs to another resource type")

    cursor_context = payload.get("c")
    if not isinstance(cursor_context, str) or not secrets.compare_digest(
        cursor_context, context_fingerprint
    ):
        raise ValueError("cursor belongs to another search context")

    offset = payload.get("o")
    if not isinstance(offset, int) or isinstance(offset, bool):
        raise ValueError("cursor offset is invalid")
    if offset < 1 or offset > MAX_SEARCH_OFFSET:
        raise ValueError("cursor offset is outside the supported range")
    return offset