perf(cases): paginate authorized records in SQL

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

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@govoplan/cases-webui",
-  "version": "0.1.24",
+  "version": "0.1.25",
   "private": true,
   "type": "module",
   "main": "webui/src/index.ts",

pyproject.toml

@@ -4,12 +4,12 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "govoplan-cases"
-version = "0.1.24"
+version = "0.1.25"
 description = "GovOPlaN administrative case context module."
 readme = "README.md"
 requires-python = ">=3.12"
 authors = [{ name = "GovOPlaN" }]
-dependencies = ["govoplan-core>=0.1.30"]
+dependencies = ["govoplan-core>=0.1.46"]
 
 [tool.setuptools.packages.find]
 where = ["src"]

src/govoplan_cases/backend/manifest.py

@@ -76,7 +76,7 @@ from govoplan_core.db.base import Base
 
 
 MODULE_ID = "cases"
-MODULE_VERSION = "0.1.24"
+MODULE_VERSION = "0.1.25"
 READ_SCOPE = "cases:case:read"
 CREATE_SCOPE = "cases:case:create"
 UPDATE_SCOPE = "cases:case:update"
@@ -823,6 +823,38 @@ manifest = ModuleManifest(
                 "outcome": "The eAkte preserves an exact case snapshot reference while Cases retains authority.",
             },
         ),
+        DocumentationTopic(
+            id="cases.authorized-pagination",
+            title="Authorized case pages and exact totals",
+            summary="Filter current case access in the database before counting and paging.",
+            body=(
+                "Case lists use one current tenant/access/purpose predicate for both exact totals and pages of at most "
+                "200 records. Restricted access requires an active matching subject grant, sufficient permission, and "
+                "an exact declared purpose; general administrator scopes do not bypass this rule. Hidden cases never "
+                "consume page slots or enter totals. The database evaluates grant existence without loading all tenant "
+                "case IDs or grants into the application. Pages retain recorded-time and stable case-ID ordering; each "
+                "request rechecks current grants and is not a snapshot across concurrent changes. Exact JSON permission "
+                "and purpose predicates support SQLite and PostgreSQL, reject malformed element types, and fail closed "
+                "on unsupported database dialects. Historical reads still use current access."
+            ),
+            layer="always", documentation_types=("user", "admin"),
+            audience=("user", "case_manager", "module_admin"), order=17,
+            translations={"de": {
+                "title": "Berechtigte Vorgangsseiten und exakte Gesamtzahlen",
+                "summary": "Aktuellen Vorgangszugriff vor Zählung und Seitenauswahl in der Datenbank filtern.",
+                "body": (
+                    "Vorgangslisten verwenden denselben aktuellen Mandanten-/Zugriffs-/Zweckfilter für exakte Gesamtzahlen "
+                    "und Seiten mit höchstens 200 Datensätzen. Eingeschränkter Zugriff verlangt eine aktive passende "
+                    "Subjektfreigabe, ausreichende Rechte und einen exakt angegebenen Zweck; allgemeine Administratorrechte "
+                    "umgehen diese Regel nicht. Verborgene Vorgänge belegen keine Seitenplätze und zählen nicht mit. "
+                    "Die Datenbank prüft Freigaben, ohne alle Vorgangskennungen oder Freigaben des Mandanten in die Anwendung "
+                    "zu laden. Die Sortierung nach Erfassungszeit und stabiler Vorgangskennung bleibt bestehen; jede Anfrage "
+                    "prüft aktuelle Freigaben erneut und ist kein Snapshot über parallele Änderungen. Exakte JSON-Rechte- "
+                    "und Zweckfilter unterstützen SQLite und PostgreSQL, verwerfen fehlerhafte Elementtypen und lehnen "
+                    "nicht unterstützte Datenbankdialekte sicher ab. Historische Abrufe verwenden weiterhin aktuelle Rechte."
+                ),
+            }},
+        ),
         DocumentationTopic(
             id="cases.governance.purpose-bound-access",
             title="Purpose-bound access to restricted cases",

src/govoplan_cases/backend/service.py

@@ -8,9 +8,10 @@ import json
 from typing import Any
 import uuid
 
-from sqlalchemy import func
+from sqlalchemy import and_, exists, func, or_
 from sqlalchemy.orm import Session
 
+from govoplan_core.core.principal_helpers import principal_user_first_actor as _principal_actor
 from govoplan_core.core.events import (
     EventActorRef,
     EventObjectRef,
@@ -24,6 +25,7 @@ from govoplan_core.core.institutional import (
     InstitutionalReference,
 )
 from govoplan_core.security.module_permissions import scopes_grant_compatible
+from govoplan_core.db.json_predicates import json_array_contains_string
 from govoplan_cases.backend.db.models import (
     CaseAccessGrant,
     CaseIdentity,
@@ -538,13 +540,7 @@ def list_cases(
         CaseRecordRevision.tenant_id == tenant_id,
         CaseRecordRevision.superseded_at.is_(None),
     )
-    eligible = _eligible_case_ids(
-        session,
-        principal,
-        permission="read",
-        purpose=purpose,
-    )
-    statement = statement.filter(CaseRecordRevision.case_id.in_(eligible))
+    statement = statement.filter(_case_access_predicate(principal, permission="read", purpose=purpose))
     if status_keys:
         statement = statement.filter(
             CaseRecordRevision.status_key.in_(tuple(status_keys))
@@ -707,42 +703,30 @@ def can_access_case(
     )
 
 
-def _eligible_case_ids(
-    session: Session,
+def _case_access_predicate(
     principal: object,
     *,
     permission: str,
     purpose: str | None,
-) -> tuple[str, ...]:
-    tenant_id = _principal_tenant(principal)
-    current = session.query(
-        CaseRecordRevision.case_id,
-        CaseRecordRevision.access_mode,
-    ).filter(
-        CaseRecordRevision.tenant_id == tenant_id,
-        CaseRecordRevision.superseded_at.is_(None),
-    ).all()
-    eligible = {
-        case_id for case_id, access_mode in current if access_mode == "tenant"
-    }
+):
+    """Current tenant/purpose/grant policy, without materializing tenant IDs."""
+    tenant_visible = CaseRecordRevision.access_mode == "tenant"
     declared_purpose = str(purpose or "").strip()
     if not declared_purpose:
-        return tuple(eligible)
+        return tenant_visible
     subjects = _principal_subjects(principal)
     if not subjects:
-        return tuple(eligible)
-    grants = session.query(CaseAccessGrant).filter(
-        CaseAccessGrant.tenant_id == tenant_id,
+        return tenant_visible
+    allowed_permissions = ("admin", "read", "update", "share") if permission == "read" else ("admin", permission)
+    grant_matches = exists().where(
+        CaseAccessGrant.tenant_id == CaseRecordRevision.tenant_id,
+        CaseAccessGrant.case_id == CaseRecordRevision.case_id,
         CaseAccessGrant.active.is_(True),
-    ).all()
-    eligible.update(
-        grant.case_id
-        for grant in grants
-        if (grant.subject_kind, grant.subject_id) in subjects
-        and _access_permissions_allow(tuple(grant.permissions or ()), permission)
-        and declared_purpose in tuple(grant.allowed_purposes or ())
+        or_(*(and_(CaseAccessGrant.subject_kind == kind, CaseAccessGrant.subject_id == subject_id) for kind, subject_id in subjects)),
+        or_(*(json_array_contains_string(CaseAccessGrant.permissions, value) for value in allowed_permissions)),
+        json_array_contains_string(CaseAccessGrant.allowed_purposes, declared_purpose),
     )
-    return tuple(eligible)
+    return or_(tenant_visible, grant_matches)
 
 
 def _sync_access_grants(
@@ -1366,20 +1350,6 @@ def _principal_tenant(principal: object) -> str:
     return tenant_id
 
 
-def _principal_actor(principal: object) -> str | None:
-    user = getattr(principal, "user", None)
-    for value in (
-        getattr(user, "id", None),
-        getattr(principal, "account_id", None),
-        getattr(principal, "identity_id", None),
-        getattr(principal, "membership_id", None),
-    ):
-        candidate = str(value or "").strip()
-        if candidate:
-            return candidate
-    return None
-
-
 def _session(value: object) -> Session:
     if not hasattr(value, "query"):
         raise InstitutionalContextError("Case registry requires a database session.")

tests/test_case_list_queries.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+from dataclasses import replace
+import unittest
+
+from sqlalchemy import event, select
+from sqlalchemy.dialects import postgresql
+
+import test_case_lifecycle as fixture
+from govoplan_cases.backend import service
+from govoplan_cases.backend.db.models import CaseAccessGrant, CaseRecordRevision
+
+
+class CaseListQueryTests(unittest.TestCase):
+    def setUp(self):
+        self.fixture = fixture.CaseLifecycleTests()
+        self.fixture.setUp()
+
+    def tearDown(self):
+        self.fixture.tearDown()
+
+    def seed(self, count=50, *, restricted=False):
+        for index in range(count):
+            original = fixture.record()
+            reference = replace(original.reference, object_id=f"case-{index:03}")
+            record = replace(
+                original, reference=reference, case_number=f"CASE-{index:03}",
+                context=replace(original.context, case_ref=reference),
+                access_mode="restricted" if restricted else "tenant",
+            )
+            service.create_case(self.fixture.session, self.fixture.principal, record=record, idempotency_key=f"fixture-{index}")
+            self.fixture.session.add(CaseAccessGrant(
+                tenant_id="tenant-1", case_id=reference.object_id, subject_kind="account", subject_id="other-account",
+                permissions=["read"], allowed_purposes=["cases.casework"], source="manual", active=True, source_revision=1,
+            ))
+        self.fixture.session.commit()
+        self.fixture.session.expunge_all()
+
+    def test_filtered_empty_page_does_not_materialize_grants_or_case_ids(self):
+        self.seed()
+        loaded = []
+        queries = []
+
+        def row_loaded(target, context):
+            loaded.append(target.id)
+
+        def executed(conn, cursor, statement, parameters, context, executemany):
+            if statement.lstrip().upper().startswith("SELECT"):
+                queries.append((statement, len(parameters)))
+
+        event.listen(CaseAccessGrant, "load", row_loaded)
+        event.listen(self.fixture.engine, "before_cursor_execute", executed)
+        try:
+            result = service.list_cases(
+                self.fixture.session, self.fixture.principal,
+                query="no-fixture-matches-this", purpose="cases.casework", limit=1,
+            )
+            self.assertEqual(((), 0), result)
+            self.assertEqual([], loaded)
+            self.assertEqual(2, len(queries))
+            self.assertTrue(all(count < 30 for _, count in queries))
+            self.assertIn("LIMIT", queries[-1][0])
+        finally:
+            event.remove(CaseAccessGrant, "load", row_loaded)
+            event.remove(self.fixture.engine, "before_cursor_execute", executed)
+
+    def test_current_grants_exact_purpose_and_permissions_govern_total(self):
+        self.seed(4, restricted=True)
+        reader = fixture.Principal(account_id="other-account")
+        session = self.fixture.session
+        self.assertEqual(((), 0), service.list_cases(session, reader, limit=1))
+        self.assertEqual(((), 0), service.list_cases(session, reader, purpose="cases.case", limit=1))
+        page, total = service.list_cases(session, reader, purpose="cases.casework", offset=1, limit=1)
+        self.assertEqual(4, total)
+        self.assertEqual("case-001", page[0].reference.object_id)
+        grants = session.query(CaseAccessGrant).filter(CaseAccessGrant.subject_id == "other-account").order_by(CaseAccessGrant.case_id).all()
+        grants[0].active = False
+        grants[1].permissions = ["reader"]
+        grants[2].allowed_purposes = ["cases.casework.extra"]
+        grants[3].permissions = ["update"]
+        session.commit()
+        self.assertEqual(1, service.list_cases(session, reader, purpose="cases.casework", limit=1)[1])
+        grants[3].allowed_purposes = {"not_an_array": "cases.casework"}
+        session.commit()
+        self.assertEqual(((), 0), service.list_cases(session, reader, purpose="cases.casework", limit=1))
+        self.assertEqual(((), 0), service.list_cases(session, fixture.Principal(tenant_id="tenant-2", account_id="other-account"), purpose="cases.casework", limit=1))
+
+    def test_access_predicate_compiles_for_postgresql_without_case_id_lists(self):
+        compiled = select(CaseRecordRevision.case_id).where(
+            service._case_access_predicate(self.fixture.principal, permission="read", purpose="cases.casework"),
+        ).compile(dialect=postgresql.dialect())
+        self.assertIn("EXISTS", str(compiled))
+        self.assertIn("json_array_elements", str(compiled))
+        self.assertNotIn("cases.casework", str(compiled))

webui/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@govoplan/cases-webui",
-  "version": "0.1.24",
+  "version": "0.1.25",
   "private": true,
   "type": "module",
   "main": "src/index.ts",