perf(tickets): paginate scoped records in SQL

Release v0.1.23. Coordinated integrity review: GovOPlaN/govoplan-core#298.

src/govoplan_tickets/backend/manifest.py

@@ -64,7 +64,7 @@ from govoplan_tickets.backend.service import (
 
 MODULE_ID = "tickets"
 MODULE_NAME = "Tickets"
-MODULE_VERSION = "0.1.22"
+MODULE_VERSION = "0.1.23"
 WRITE_SCOPE = LEGACY_WRITE_SCOPE
 OPTIONAL_DEPENDENCIES = (
     "cases",
@@ -172,6 +172,36 @@ ROLE_TEMPLATES = (
 )
 
 DOCUMENTATION = (
+    DocumentationTopic(
+        id="tickets.authorized-pagination",
+        title="Authorized ticket pages and totals",
+        summary="Ticket pages and their exact totals use the same current authorization filter.",
+        body=(
+            "Ticket lists apply tenant, deletion, search, queue, status, and current read-access filters in the database "
+            "before counting or selecting a page. Elevated ticket scopes retain their existing access; other readers see "
+            "tenant-visible tickets and tickets linked to their current subjects or actor identities. Hidden tickets do "
+            "not enter totals or consume page slots. Pages contain at most 200 records and use service target, priority, "
+            "updated time, then stable ticket ID ordering. Each request evaluates current access again; paging is not an "
+            "immutable snapshot across concurrent edits. Exact JSON subject matching supports SQLite and PostgreSQL and "
+            "does not coerce numeric or boolean identifiers; unsupported database dialects fail closed."
+        ),
+        layer="always", documentation_types=("user", "admin"),
+        audience=("user", "operator", "tenant_admin"),
+        translations={"de": {
+            "title": "Berechtigte Ticketseiten und Gesamtzahlen",
+            "summary": "Ticketseiten und ihre exakten Gesamtzahlen verwenden denselben aktuellen Berechtigungsfilter.",
+            "body": (
+                "Ticketlisten wenden Mandanten-, Lösch-, Such-, Warteschlangen-, Status- und aktuelle Lesefilter in der "
+                "Datenbank vor Zählung und Seitenauswahl an. Erweiterte Ticketrechte behalten ihren bisherigen Zugriff; "
+                "andere Lesende sehen mandantenweit sichtbare Tickets und Tickets mit Bezug zu ihren aktuellen Subjekten "
+                "oder Akteurskennungen. Verborgene Tickets zählen nicht mit und belegen keine Seitenplätze. Seiten enthalten "
+                "höchstens 200 Datensätze, sortiert nach Serviceziel, Priorität, Änderungszeit und stabiler Ticketkennung. "
+                "Jede Anfrage prüft den aktuellen Zugriff erneut; Seitenabrufe bilden bei parallelen Änderungen keinen "
+                "unveränderlichen Snapshot. Exakte JSON-Subjektvergleiche unterstützen SQLite und PostgreSQL ohne Umwandlung "
+                "numerischer oder boolescher Kennungen; nicht unterstützte Datenbankdialekte werden sicher abgelehnt."
+            ),
+        }},
+    ),
     DocumentationTopic(
         id="tickets.module-boundary",
         title="Tickets module boundary",

src/govoplan_tickets/backend/service.py

@@ -6,8 +6,10 @@ import hashlib
 import json
 from typing import Any
 
+from sqlalchemy import false, func, or_, true
 from sqlalchemy.orm import Session
 
+from govoplan_core.core.principal_helpers import principal_actor_ids as _principal_actor_ids
 from govoplan_core.core.events import (
     EventActorRef,
     EventObjectRef,
@@ -22,6 +24,7 @@ from govoplan_core.core.tickets import (
     ticket_routing_provider,
 )
 from govoplan_core.security.module_permissions import scopes_grant_compatible
+from govoplan_core.db.json_predicates import json_array_contains_object_strings, json_object_matches_strings
 from govoplan_tickets.backend.db.models import (
     Ticket,
     TicketComment,
@@ -163,14 +166,15 @@ def list_tickets(
     clean_query = query.strip().casefold()
     if clean_query:
         statement = statement.filter(Ticket.search_text.contains(clean_query))
-    candidates = statement.order_by(
+    statement = statement.filter(_ticket_read_predicate(principal))
+    total = int(statement.with_entities(func.count()).scalar() or 0)
+    selected = statement.order_by(
         Ticket.service_target_at.asc().nullslast(),
         Ticket.priority.desc(),
         Ticket.updated_at.desc(),
-    ).all()
-    accessible = tuple(row for row in candidates if _can_read_row(principal, row))
-    selected = accessible[offset : offset + limit]
-    return tuple(_record(row) for row in selected), len(accessible)
+        Ticket.id.asc(),
+    ).offset(offset).limit(limit).all()
+    return tuple(_record(row) for row in selected), total
 
 
 def triage_ticket(
@@ -926,6 +930,26 @@ def _required_ticket(session: Session, principal: object, *, ticket_id: str, loc
     return row
 
 
+def _ticket_read_predicate(principal: object):
+    """Owner policy shared by SQL count/page; never page before authorization."""
+    if not _has_scope(principal, READ_SCOPE):
+        return false()
+    if _has_any_scope(principal, TRIAGE_SCOPE, ASSIGN_SCOPE, RESOLVE_SCOPE, ADMIN_SCOPE, LEGACY_WRITE_SCOPE):
+        return true()
+    conditions = [Ticket.visibility == "tenant"]
+    actors = _principal_actor_ids(principal)
+    if actors:
+        conditions.append(Ticket.created_by.in_(actors))
+    for kind, subject_id in _principal_subjects(principal):
+        fields = {"kind": kind, "id": subject_id}
+        conditions.extend(
+            json_object_matches_strings(column, fields)
+            for column in (Ticket.assignee, Ticket.reporter, Ticket.requester)
+        )
+        conditions.append(json_array_contains_object_strings(Ticket.participants, fields))
+    return or_(*conditions)
+
+
 def _can_read_row(principal: object, row: Ticket) -> bool:
     if row.tenant_id != _principal_tenant(principal) or not _has_scope(principal, READ_SCOPE):
         return False
@@ -1020,22 +1044,6 @@ def _principal_subjects(principal: object) -> tuple[tuple[str, str], ...]:
     return tuple(dict.fromkeys(values))
 
 
-def _principal_actor_ids(principal: object) -> tuple[str, ...]:
-    user = getattr(principal, "user", None)
-    return tuple(
-        dict.fromkeys(
-            str(value)
-            for value in (
-                getattr(principal, "account_id", None),
-                getattr(principal, "identity_id", None),
-                getattr(principal, "membership_id", None),
-                getattr(user, "id", None),
-            )
-            if str(value or "").strip()
-        )
-    )
-
-
 def _principal_actor(principal: object) -> str | None:
     values = _principal_actor_ids(principal)
     return values[0] if values else None