Add: pagination at global search

app/modules/search/api/cursor.py

@@ -0,0 +1,67 @@
+import base64
+import hashlib
+import json
+import secrets
+from typing import Any
+
+from app.modules.search.domain.models import SearchResourceType
+from app.shared.nlp.preprocessing.text import prepare_for_embedding
+
+
+CURSOR_VERSION = 1
+MAX_SEARCH_OFFSET = 1000
+
+
+def encode_search_cursor(
+    resource_type: SearchResourceType,
+    normalized_query: str,
+    offset: int,
+) -> str:
+    payload = {
+        "v": CURSOR_VERSION,
+        "r": resource_type.value,
+        "q": _query_fingerprint(normalized_query),
+        "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,
+    query: 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")
+
+    expected_query = _query_fingerprint(prepare_for_embedding(query))
+    cursor_query = payload.get("q")
+    if not isinstance(cursor_query, str) or not secrets.compare_digest(
+        cursor_query, expected_query
+    ):
+        raise ValueError("cursor belongs to another query")
+
+    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
+
+
+def _query_fingerprint(normalized_query: str) -> str:
+    return hashlib.sha256(normalized_query.encode("utf-8")).hexdigest()[:24]

app/modules/search/api/router.py

@@ -3,6 +3,7 @@ import secrets
 from fastapi import APIRouter, Depends, Header, HTTPException
 
 from app.modules.search.api.dependencies import get_search_all_use_case
+from app.modules.search.api.cursor import decode_search_cursor
 from app.modules.search.api.schemas import (
     GlobalSearchRequest,
     GlobalSearchResponse,
@@ -40,12 +41,24 @@ async def search_all(
         if payload.resource_types
         else ALL_SEARCH_RESOURCE_TYPES
     )
+    try:
+        offsets = {
+            resource_type: decode_search_cursor(
+                cursor=cursor,
+                resource_type=resource_type,
+                query=payload.query,
+            )
+            for resource_type, cursor in payload.cursors.items()
+        }
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
     result = await use_case.execute(
         query=payload.query,
         resource_types=resource_types,
         per_type_limit=payload.per_type_limit,
         top_limit=payload.top_limit,
         requester_id=str(payload.requester_id) if payload.requester_id else None,
+        offsets=offsets,
     )
     return result_to_schema(result)
 

app/modules/search/api/schemas.py

@@ -1,8 +1,9 @@
 from typing import Any
 from uuid import UUID
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
+from app.modules.search.api.cursor import MAX_SEARCH_OFFSET, encode_search_cursor
 from app.modules.search.domain.models import SearchAllResult, SearchHit, SearchResourceType
 
 
@@ -14,6 +15,20 @@ class GlobalSearchRequest(BaseModel):
     per_type_limit: int = Field(default=5, ge=1, le=20)
     top_limit: int = Field(default=10, ge=1, le=50)
     requester_id: UUID | None = None
+    cursors: dict[SearchResourceType, str] = Field(default_factory=dict)
+
+    @model_validator(mode="after")
+    def validate_cursors(self) -> "GlobalSearchRequest":
+        if not self.cursors:
+            return self
+        if not self.resource_types:
+            raise ValueError("resource_types is required when cursors are provided")
+        requested = set(self.resource_types)
+        unexpected = set(self.cursors) - requested
+        if unexpected:
+            names = ", ".join(sorted(resource_type.value for resource_type in unexpected))
+            raise ValueError(f"cursor resources were not requested: {names}")
+        return self
 
 
 class SearchHitSchema(BaseModel):
@@ -34,11 +49,19 @@ class GlobalSearchMetadataSchema(BaseModel):
     embedding_computed_once: bool
 
 
+class SearchSectionPaginationSchema(BaseModel):
+    page_size: int
+    returned_count: int
+    has_more: bool
+    next_cursor: str | None = None
+
+
 class GlobalSearchResponse(BaseModel):
     query: str
     normalized_query: str
     top_results: list[SearchHitSchema]
     sections: dict[str, list[SearchHitSchema]]
+    pagination: dict[str, SearchSectionPaginationSchema]
     metadata: GlobalSearchMetadataSchema
 
 
@@ -51,6 +74,28 @@ def result_to_schema(result: SearchAllResult) -> GlobalSearchResponse:
             resource_type.value: [_hit_to_schema(hit) for hit in hits]
             for resource_type, hits in result.sections.items()
         },
+        pagination={
+            resource_type.value: SearchSectionPaginationSchema(
+                page_size=page.page_size,
+                returned_count=page.returned_count,
+                has_more=(
+                    page.has_more
+                    and page.next_offset is not None
+                    and page.next_offset <= MAX_SEARCH_OFFSET
+                ),
+                next_cursor=(
+                    encode_search_cursor(
+                        resource_type,
+                        result.normalized_query,
+                        page.next_offset,
+                    )
+                    if page.next_offset is not None
+                    and page.next_offset <= MAX_SEARCH_OFFSET
+                    else None
+                ),
+            )
+            for resource_type, page in result.pagination.items()
+        },
         metadata=GlobalSearchMetadataSchema(
             strategy="parallel_hybrid_fasttext_full_text_rrf",
             queried_resources=list(result.sections),

app/modules/search/application/ports/search_provider.py

@@ -11,6 +11,7 @@ class SearchProvider(Protocol):
         query: str,
         embedding: list[float],
         limit: int,
+        offset: int,
         requester_id: str | None,
     ) -> Sequence[SearchHit]:
         """Return already-ranked candidates for exactly one resource type."""

app/modules/search/application/use_cases/search_all.py

@@ -7,6 +7,7 @@ from app.modules.search.domain.models import (
     SearchAllResult,
     SearchHit,
     SearchResourceType,
+    SearchSectionPagination,
 )
 from app.shared.logging.config import get_logger
 from app.shared.nlp.embeddings.base import EmbeddingProvider
@@ -31,9 +32,11 @@ class SearchAllUseCase:
         per_type_limit: int = 5,
         top_limit: int = 10,
         requester_id: str | None = None,
+        offsets: dict[SearchResourceType, int] | None = None,
     ) -> SearchAllResult:
         normalized_query = prepare_for_embedding(query)
         embedding = self._embedding_provider.embed_text(normalized_query)
+        effective_offsets = offsets or {}
         providers = [
             self._providers[resource_type]
             for resource_type in resource_types
@@ -44,7 +47,8 @@ class SearchAllUseCase:
                 provider.search(
                     query=normalized_query,
                     embedding=embedding,
-                    limit=per_type_limit,
+                    limit=per_type_limit + 1,
+                    offset=effective_offsets.get(provider.resource_type, 0),
                     requester_id=requester_id,
                 )
                 for provider in providers
@@ -55,6 +59,7 @@ class SearchAllUseCase:
         sections: dict[SearchResourceType, list[SearchHit]] = {
             resource_type: [] for resource_type in resource_types
         }
+        pagination: dict[SearchResourceType, SearchSectionPagination] = {}
         failures: dict[SearchResourceType, str] = {}
         for provider, outcome in zip(providers, outcomes):
             if isinstance(outcome, BaseException):
@@ -64,10 +69,23 @@ class SearchAllUseCase:
                     type(outcome).__name__,
                 )
                 failures[provider.resource_type] = type(outcome).__name__
+                pagination[provider.resource_type] = SearchSectionPagination(
+                    page_size=per_type_limit,
+                    returned_count=0,
+                    has_more=False,
+                )
                 continue
-            sections[provider.resource_type] = sorted(
-                list(outcome), key=lambda hit: hit.score, reverse=True
-            )[:per_type_limit]
+            ranked = sorted(list(outcome), key=lambda hit: hit.score, reverse=True)
+            page_hits = ranked[:per_type_limit]
+            has_more = len(ranked) > per_type_limit
+            current_offset = effective_offsets.get(provider.resource_type, 0)
+            sections[provider.resource_type] = page_hits
+            pagination[provider.resource_type] = SearchSectionPagination(
+                page_size=per_type_limit,
+                returned_count=len(page_hits),
+                has_more=has_more,
+                next_offset=(current_offset + len(page_hits) if has_more else None),
+            )
 
         top_results = _diversified_top_results(sections, top_limit)
         return SearchAllResult(
@@ -75,6 +93,7 @@ class SearchAllUseCase:
             normalized_query=normalized_query,
             top_results=top_results,
             sections=sections,
+            pagination=pagination,
             failed_resources=failures,
         )
 

app/modules/search/domain/models.py

@@ -27,10 +27,19 @@ class SearchHit:
     metadata: dict[str, Any] = field(default_factory=dict)
 
 
+@dataclass(frozen=True)
+class SearchSectionPagination:
+    page_size: int
+    returned_count: int
+    has_more: bool
+    next_offset: int | None = None
+
+
 @dataclass(frozen=True)
 class SearchAllResult:
     query: str
     normalized_query: str
     top_results: list[SearchHit]
     sections: dict[SearchResourceType, list[SearchHit]]
+    pagination: dict[SearchResourceType, SearchSectionPagination]
     failed_resources: dict[SearchResourceType, str]

app/modules/search/infrastructure/mock_provider.py

@@ -36,6 +36,7 @@ class MockHybridSearchProvider(SearchProvider):
         query: str,
         embedding: list[float],
         limit: int,
+        offset: int,
         requester_id: str | None,
     ) -> Sequence[SearchHit]:
         query_terms = set(prepare_for_embedding(query).split())
@@ -67,7 +68,8 @@ class MockHybridSearchProvider(SearchProvider):
                     metadata=metadata,
                 )
             )
-        return sorted(hits, key=lambda hit: hit.score, reverse=True)[:limit]
+        ranked = sorted(hits, key=lambda hit: hit.score, reverse=True)
+        return ranked[offset : offset + limit]
 
 
 def get_mock_search_documents() -> dict[SearchResourceType, list[MockSearchDocument]]:

app/modules/search/infrastructure/pgvector_provider.py

@@ -21,15 +21,19 @@ class PgvectorPlaceSearchProvider(SearchProvider):
         query: str,
         embedding: list[float],
         limit: int,
+        offset: int,
         requester_id: str | None,
     ) -> Sequence[SearchHit]:
         del query, requester_id
         matches = await self._vector_client.match_places(
             embedding=embedding,
             filters={"is_active": True},
-            limit=limit,
+            limit=offset + limit,
         )
-        return [_to_search_hit(self.resource_type, match) for match in matches]
+        return [
+            _to_search_hit(self.resource_type, match)
+            for match in matches[offset : offset + limit]
+        ]
 
 
 class PgvectorHybridSearchProvider(SearchProvider):
@@ -46,6 +50,7 @@ class PgvectorHybridSearchProvider(SearchProvider):
         query: str,
         embedding: list[float],
         limit: int,
+        offset: int,
         requester_id: str | None,
     ) -> Sequence[SearchHit]:
         filters = {"is_active": True}
@@ -56,9 +61,12 @@ class PgvectorHybridSearchProvider(SearchProvider):
             query_text=query,
             embedding=embedding,
             filters=filters,
-            limit=limit,
+            limit=offset + limit,
         )
-        return [_to_search_hit(self.resource_type, match) for match in matches]
+        return [
+            _to_search_hit(self.resource_type, match)
+            for match in matches[offset : offset + limit]
+        ]
 
 
 def _to_search_hit(resource_type: SearchResourceType, match: VectorMatch) -> SearchHit:

docs/api_endpoints.md

@@ -447,9 +447,14 @@ No recibe query parameters. Toda la entrada se envia en el body.
 | `per_type_limit` | `integer` | No | `5` | Entre 1 y 20 |
 | `top_limit` | `integer` | No | `10` | Entre 1 y 50 |
 | `requester_id` | `UUID \| null` | No | `null` | Activa busqueda privada |
+| `cursors` | `object` | No | `{}` | Cursor opaco por tipo de recurso |
 
 No se aceptan campos adicionales.
 
+En la primera solicitud se omite `cursors`. La respuesta incluye un bloque
+`pagination` independiente para cada recurso consultado. Cuando `has_more` es `true`,
+`next_cursor` se envia en la siguiente solicitud usando la misma consulta.
+
 #### Busqueda publica
 
 No requiere `Authorization`.
@@ -522,6 +527,26 @@ debe obtenerlo de la sesion autenticada y construir la llamada interna.
     "events": [],
     "users": []
   },
+  "pagination": {
+    "clubs": {
+      "page_size": 5,
+      "returned_count": 1,
+      "has_more": false,
+      "next_cursor": null
+    },
+    "events": {
+      "page_size": 5,
+      "returned_count": 0,
+      "has_more": false,
+      "next_cursor": null
+    },
+    "users": {
+      "page_size": 5,
+      "returned_count": 0,
+      "has_more": false,
+      "next_cursor": null
+    }
+  },
   "metadata": {
     "strategy": "parallel_hybrid_fasttext_full_text_rrf",
     "queried_resources": ["clubs", "events", "users"],
@@ -534,6 +559,44 @@ debe obtenerlo de la sesion autenticada y construir la llamada interna.
 `top_results` es una seleccion global diversificada. `sections` conserva los resultados
 separados por tipo de recurso.
 
+#### Solicitar la siguiente pagina
+
+Cada recurso se pagina de manera independiente. Para cargar mas lugares, la interfaz
+debe reutilizar exactamente `query`, declarar `resource_types: ["places"]` y enviar el
+cursor recibido en `pagination.places.next_cursor`:
+
+```json
+{
+  "query": "cafeteria tranquila",
+  "resource_types": ["places"],
+  "per_type_limit": 5,
+  "top_limit": 5,
+  "cursors": {
+    "places": "CURSOR_DEVUELTO_POR_LA_PAGINA_ANTERIOR"
+  }
+}
+```
+
+La respuesta de cada seccion contiene:
+
+| Campo | Descripcion |
+| --- | --- |
+| `page_size` | Limite solicitado para esa pagina |
+| `returned_count` | Cantidad realmente devuelta |
+| `has_more` | Indica si existe al menos otra pagina |
+| `next_cursor` | Cursor opaco de continuacion o `null` si termino |
+
+Reglas de los cursores:
+
+- estan asociados al tipo de recurso y a la consulta normalizada;
+- no deben interpretarse ni construirse en la app cliente;
+- no pueden reutilizarse con otra consulta o con otro recurso;
+- cuando se envia `cursors`, `resource_types` es obligatorio y debe contener sus claves;
+- se pueden pedir varias continuaciones en una llamada enviando un cursor por recurso.
+
+El contrato SQL utiliza el ID externo como desempate estable cuando dos resultados tienen
+el mismo puntaje. Esto evita cambios arbitrarios de orden entre paginas consecutivas.
+
 Si falla un proveedor individual, los demas pueden responder normalmente. El recurso
 fallido aparece en `metadata.failed_resources`.
 

sql/aws_pgvector_contract.sql

@@ -122,7 +122,7 @@ AS $$
           (filters ? 'place_ids') IS FALSE
           OR p.external_id IN (SELECT jsonb_array_elements_text(filters->'place_ids'))
       )
-    ORDER BY p.embedding <=> query_embedding
+    ORDER BY p.embedding <=> query_embedding, p.external_id ASC
     LIMIT match_count;
 $$;
 
@@ -342,9 +342,11 @@ BEGIN
             SELECT
                 e.external_id,
                 1 - (e.embedding <=> $1) AS semantic_score,
-                row_number() OVER (ORDER BY e.embedding <=> $1) AS vector_rank
+                row_number() OVER (
+                    ORDER BY e.embedding <=> $1, e.external_id ASC
+                ) AS vector_rank
             FROM eligible e
-            ORDER BY e.embedding <=> $1
+            ORDER BY e.embedding <=> $1, e.external_id ASC
             LIMIT GREATEST($3 * 3, 30)
         ),
         lexical_results AS (
@@ -352,12 +354,12 @@ BEGIN
                 e.external_id,
                 ts_rank_cd(e.textsearch, query.value) AS lexical_score,
                 row_number() OVER (
-                    ORDER BY ts_rank_cd(e.textsearch, query.value) DESC
+                    ORDER BY ts_rank_cd(e.textsearch, query.value) DESC, e.external_id ASC
                 ) AS lexical_rank
             FROM eligible e
             CROSS JOIN websearch_to_tsquery('simple', $2) AS query(value)
             WHERE e.textsearch @@ query.value
-            ORDER BY lexical_score DESC
+            ORDER BY lexical_score DESC, e.external_id ASC
             LIMIT GREATEST($3 * 3, 30)
         ),
         fused AS (
@@ -392,7 +394,7 @@ BEGIN
             f.lexical_score::double precision
         FROM fused f
         JOIN eligible e USING (external_id)
-        ORDER BY score DESC, f.semantic_score DESC NULLS LAST
+        ORDER BY score DESC, f.semantic_score DESC NULLS LAST, e.external_id ASC
         LIMIT GREATEST($3, 1)
     $query$, target_table)
     USING p_query_embedding, p_query_text, p_match_count, p_filters;

tests/test_global_search.py

@@ -56,3 +56,70 @@ def test_requester_scoped_search_rejects_untrusted_callers() -> None:
         },
     )
     assert response.status_code == 403
+
+
+def test_global_search_paginates_each_resource_with_opaque_cursor() -> None:
+    client = TestClient(create_app())
+    first_response = client.post(
+        "/search",
+        json={
+            "query": "lugar tranquilo",
+            "resource_types": ["places"],
+            "per_type_limit": 2,
+            "top_limit": 2,
+        },
+    )
+    assert first_response.status_code == 200
+    first = first_response.json()
+    first_ids = [hit["id"] for hit in first["sections"]["places"]]
+    page = first["pagination"]["places"]
+    assert len(first_ids) == 2
+    assert page == {
+        "page_size": 2,
+        "returned_count": 2,
+        "has_more": True,
+        "next_cursor": page["next_cursor"],
+    }
+    assert isinstance(page["next_cursor"], str)
+
+    second_response = client.post(
+        "/search",
+        json={
+            "query": "lugar tranquilo",
+            "resource_types": ["places"],
+            "per_type_limit": 2,
+            "top_limit": 2,
+            "cursors": {"places": page["next_cursor"]},
+        },
+    )
+    assert second_response.status_code == 200
+    second = second_response.json()
+    second_ids = [hit["id"] for hit in second["sections"]["places"]]
+    assert len(second_ids) == 2
+    assert set(first_ids).isdisjoint(second_ids)
+
+
+def test_search_cursor_cannot_be_reused_for_another_query() -> None:
+    client = TestClient(create_app())
+    first = client.post(
+        "/search",
+        json={
+            "query": "lugar tranquilo",
+            "resource_types": ["places"],
+            "per_type_limit": 1,
+        },
+    ).json()
+    cursor = first["pagination"]["places"]["next_cursor"]
+
+    response = client.post(
+        "/search",
+        json={
+            "query": "evento deportivo",
+            "resource_types": ["places"],
+            "per_type_limit": 1,
+            "cursors": {"places": cursor},
+        },
+    )
+
+    assert response.status_code == 422
+    assert response.json()["detail"] == "cursor belongs to another query"

tests/test_pgvector_search_provider.py

@@ -26,8 +26,14 @@ class RecordingVectorClient:
                 score=0.82,
                 metadata={"name": "Cafe Central", "category": "cafe"},
                 document="cafe tranquilo para trabajar",
-            )
-        ]
+            ),
+            VectorMatch(
+                id="place-2",
+                score=0.75,
+                metadata={"name": "Cafe Sur", "category": "cafe"},
+                document="cafe para conversar",
+            ),
+        ][:limit]
 
     async def search_resource_embeddings(self, **kwargs: object) -> list[VectorMatch]:
         raise AssertionError("Places must not use the hybrid RRF search function")
@@ -41,7 +47,8 @@ async def test_places_use_same_cosine_match_function_as_recommendations() -> Non
     hits = await provider.search(
         query="cafe tranquilo",
         embedding=[0.1, 0.2, 0.3],
-        limit=5,
+        limit=1,
+        offset=1,
         requester_id=None,
     )
 
@@ -49,10 +56,11 @@ async def test_places_use_same_cosine_match_function_as_recommendations() -> Non
         {
             "embedding": [0.1, 0.2, 0.3],
             "filters": {"is_active": True},
-            "limit": 5,
+            "limit": 2,
         }
     ]
+    assert hits[0].id == "place-2"
     assert hits[0].resource_type == SearchResourceType.PLACES
-    assert hits[0].score == 0.82
-    assert hits[0].semantic_score == 0.82
+    assert hits[0].score == 0.75
+    assert hits[0].semantic_score == 0.75
     assert hits[0].lexical_score is None