chore: sync GovOPlaN module split state

src/govoplan_audit/backend/api/v1/routes.py

@@ -3,20 +3,25 @@ from __future__ import annotations
 from datetime import datetime, timedelta, timezone
 
 from fastapi import APIRouter, Depends, HTTPException, Query, status
-from sqlalchemy import false, func, or_
+from sqlalchemy import and_, false, func, or_
 from sqlalchemy.orm import Session
 
-from govoplan_access.backend.auth.dependencies import ApiPrincipal, has_scope, require_any_scope
+from govoplan_access.auth import ApiPrincipal, has_scope, require_any_scope
 from govoplan_audit.backend.db.models import AuditLog
+from govoplan_core.audit.logging import AUDIT_MODULE_ID, AUDIT_SYSTEM_EVENTS_COLLECTION, AUDIT_TENANT_EVENTS_COLLECTION
 from govoplan_core.core.access import CAPABILITY_ACCESS_ADMINISTRATION, AccessAdministration
+from govoplan_core.core.change_sequence import decode_sequence_watermark, encode_sequence_watermark, max_sequence_id, sequence_entries_since, sequence_watermark_is_expired
+from govoplan_core.core.pagination import KeysetCursorError, decode_keyset_cursor, encode_keyset_cursor, keyset_query_fingerprint
 from govoplan_core.core.runtime import get_registry
 from govoplan_core.db.session import get_session
 from govoplan_tenancy.backend.db.models import Tenant
 
-from .schemas import AuditAdminItem, AuditAdminListResponse, AuditLogItemResponse, AuditLogListResponse
+from .schemas import AuditAdminDeltaResponse, AuditAdminItem, AuditAdminListResponse, AuditLogItemResponse, AuditLogListResponse
 
 router = APIRouter(tags=["audit"])
 
+AUDIT_ADMIN_CURSOR_SCOPE = "audit.admin"
+
 
 def _access_administration() -> AccessAdministration:
     registry = get_registry()
@@ -91,7 +96,7 @@ def _audit_time_filter(raw: str | None):
     try:
         parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
     except ValueError as exc:
-        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Invalid audit date filter.") from exc
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Invalid audit date filter.") from exc
     if parsed.tzinfo is None:
         parsed = parsed.replace(tzinfo=timezone.utc)
     else:
@@ -111,6 +116,176 @@ def _audit_time_filter(raw: str | None):
     return AuditLog.created_at == parsed
 
 
+def _audit_delta_collections(effective_scope: str) -> tuple[str, ...]:
+    if effective_scope == "system":
+        return (AUDIT_SYSTEM_EVENTS_COLLECTION,)
+    if effective_scope == "tenant":
+        return (AUDIT_TENANT_EVENTS_COLLECTION,)
+    return (AUDIT_TENANT_EVENTS_COLLECTION, AUDIT_SYSTEM_EVENTS_COLLECTION)
+
+
+def _audit_delta_watermark(session: Session, *, effective_scope: str, tenant_id: str | None) -> str:
+    return encode_sequence_watermark(
+        max_sequence_id(
+            session,
+            tenant_id=tenant_id if effective_scope == "tenant" else None,
+            module_id=AUDIT_MODULE_ID,
+            collections=_audit_delta_collections(effective_scope),
+        )
+    )
+
+
+def _audit_delta_entries(
+    session: Session,
+    *,
+    effective_scope: str,
+    tenant_id: str | None,
+    since: str,
+    limit: int,
+):
+    try:
+        since_sequence = decode_sequence_watermark(since)
+    except ValueError as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
+    scoped_tenant_id = tenant_id if effective_scope == "tenant" else None
+    collections = _audit_delta_collections(effective_scope)
+    if sequence_watermark_is_expired(
+        session,
+        since=since_sequence,
+        tenant_id=scoped_tenant_id,
+        module_id=AUDIT_MODULE_ID,
+        collections=collections,
+    ):
+        return None, False
+    entries_plus_one = sequence_entries_since(
+        session,
+        since=since_sequence,
+        tenant_id=scoped_tenant_id,
+        module_id=AUDIT_MODULE_ID,
+        collections=collections,
+        limit=limit + 1,
+    )
+    has_more = len(entries_plus_one) > limit
+    return entries_plus_one[:limit], has_more
+
+
+def _audit_delta_response_watermark(
+    session: Session,
+    *,
+    effective_scope: str,
+    tenant_id: str | None,
+    entries,
+    has_more: bool,
+) -> str:
+    return encode_sequence_watermark(entries[-1].id) if has_more and entries else _audit_delta_watermark(session, effective_scope=effective_scope, tenant_id=tenant_id)
+
+
+def _audit_items(session: Session, rows: list[AuditLog], access_admin: AccessAdministration) -> list[AuditAdminItem]:
+    actor_email_by_user_id = access_admin.actor_email_by_user_id(session, {row.user_id for row in rows if row.user_id})
+    return [
+        AuditAdminItem(
+            id=row.id,
+            scope=row.scope,
+            tenant_id=row.tenant_id,
+            actor_email=actor_email_by_user_id.get(row.user_id) if row.user_id else None,
+            action=row.action,
+            object_type=row.object_type,
+            object_id=row.object_id,
+            details=row.details or {},
+            created_at=row.created_at,
+        )
+        for row in rows
+    ]
+
+
+def _audit_filter_params(
+    *,
+    filter_time: str | None,
+    filter_actor: str | None,
+    filter_action: str | None,
+    filter_object: str | None,
+    filter_tenant: str | None,
+) -> dict[str, str]:
+    return {
+        "time": filter_time or "",
+        "actor": filter_actor or "",
+        "action": filter_action or "",
+        "object": filter_object or "",
+        "tenant": filter_tenant or "",
+    }
+
+
+def _audit_cursor_fingerprint(
+    *,
+    effective_scope: str,
+    tenant_id: str | None,
+    page_size: int,
+    sort_by: str,
+    sort_direction: str,
+    filters: dict[str, str],
+) -> str:
+    return keyset_query_fingerprint(
+        AUDIT_ADMIN_CURSOR_SCOPE,
+        {
+            "scope": effective_scope,
+            "tenant_id": tenant_id or "",
+            "page_size": page_size,
+            "sort_by": sort_by,
+            "sort_direction": sort_direction,
+            "filters": filters,
+        },
+    )
+
+
+def _audit_sort_value(row: AuditLog, sort_by: str):
+    if sort_by == "time":
+        return row.created_at
+    if sort_by == "actor":
+        return row.user_id or "System"
+    if sort_by == "action":
+        return row.action
+    if sort_by == "object":
+        return f"{row.object_type or ''} {row.object_id or ''}"
+    if sort_by == "tenant":
+        return row.tenant_id or ""
+    raise KeysetCursorError("Unsupported audit sort column")
+
+
+def _audit_cursor_for_row(row: AuditLog, *, sort_by: str, sort_direction: str, fingerprint: str) -> str:
+    return encode_keyset_cursor(
+        AUDIT_ADMIN_CURSOR_SCOPE,
+        fingerprint=fingerprint,
+        values={
+            "id": row.id,
+            "sort_by": sort_by,
+            "sort_direction": sort_direction,
+            "sort_value": _audit_sort_value(row, sort_by),
+        },
+    )
+
+
+def _audit_decode_sort_value(sort_by: str, value):
+    if sort_by == "time":
+        if not isinstance(value, str):
+            raise KeysetCursorError("Invalid pagination cursor")
+        try:
+            return datetime.fromisoformat(value.replace("Z", "+00:00"))
+        except ValueError as exc:
+            raise KeysetCursorError("Invalid pagination cursor") from exc
+    if value is None:
+        return ""
+    return str(value)
+
+
+def _audit_cursor_condition(sort_column, *, sort_by: str, sort_direction: str, cursor_values: dict[str, object]):
+    cursor_id = cursor_values.get("id")
+    if not isinstance(cursor_id, str) or not cursor_id:
+        raise KeysetCursorError("Invalid pagination cursor")
+    sort_value = _audit_decode_sort_value(sort_by, cursor_values.get("sort_value"))
+    primary_after = sort_column > sort_value if sort_direction == "asc" else sort_column < sort_value
+    return or_(primary_after, and_(sort_column == sort_value, AuditLog.id < cursor_id))
+
+
 @router.get("/admin/audit", response_model=AuditAdminListResponse)
 def list_admin_audit(
     tenant_id: str | None = Query(default=None),
@@ -120,6 +295,7 @@ def list_admin_audit(
     offset: int = Query(default=0, ge=0),
     page: int | None = Query(default=None, ge=1),
     page_size: int | None = Query(default=None, ge=1, le=500),
+    cursor: str | None = Query(default=None),
     sort_by: str = Query(default="time"),
     sort_direction: str = Query(default="desc"),
     filter_time: str | None = Query(default=None),
@@ -132,13 +308,14 @@ def list_admin_audit(
 ):
     effective_scope = audit_scope or ("all" if all_tenants else "tenant")
     if effective_scope not in {"tenant", "system", "all"}:
-        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Audit scope must be tenant, system or all.")
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Audit scope must be tenant, system or all.")
     if sort_by not in {"time", "actor", "action", "object", "tenant"}:
-        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Unsupported audit sort column.")
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Unsupported audit sort column.")
     if sort_direction not in {"asc", "desc"}:
-        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Audit sort direction must be asc or desc.")
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Audit sort direction must be asc or desc.")
 
     query = session.query(AuditLog)
+    resolved_tenant_id: str | None = None
     if effective_scope != "all":
         query = query.filter(AuditLog.scope == effective_scope)
     if effective_scope == "system":
@@ -151,6 +328,7 @@ def list_admin_audit(
         if not has_scope(principal, "audit:read"):
             raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: audit:read")
         tenant = _resolve_tenant(session, principal, tenant_id)
+        resolved_tenant_id = tenant.id
         query = query.filter(AuditLog.tenant_id == tenant.id)
 
     object_text = func.coalesce(AuditLog.object_type, "") + " " + func.coalesce(AuditLog.object_id, "")
@@ -166,15 +344,134 @@ def list_admin_audit(
         if condition is not None:
             query = query.filter(condition)
 
+    sort_columns = {
+        "time": AuditLog.created_at,
+        "actor": func.coalesce(AuditLog.user_id, "System"),
+        "action": AuditLog.action,
+        "object": object_text,
+        "tenant": func.coalesce(AuditLog.tenant_id, ""),
+    }
+    sort_column = sort_columns[sort_by]
+    order = sort_column.asc() if sort_direction == "asc" else sort_column.desc()
+    ordered_query = query.order_by(order, AuditLog.id.desc())
     total = query.count()
     effective_page_size = page_size or limit
     pages = max(1, (total + effective_page_size - 1) // effective_page_size)
-    if page is not None or page_size is not None:
-        effective_page = min(page or 1, pages)
-        effective_offset = (effective_page - 1) * effective_page_size
+    filters = _audit_filter_params(
+        filter_time=filter_time,
+        filter_actor=filter_actor,
+        filter_action=filter_action,
+        filter_object=filter_object,
+        filter_tenant=filter_tenant,
+    )
+    fingerprint = _audit_cursor_fingerprint(
+        effective_scope=effective_scope,
+        tenant_id=resolved_tenant_id,
+        page_size=effective_page_size,
+        sort_by=sort_by,
+        sort_direction=sort_direction,
+        filters=filters,
+    )
+
+    start_cursor: str | None = None
+    if cursor:
+        try:
+            cursor_values = decode_keyset_cursor(AUDIT_ADMIN_CURSOR_SCOPE, cursor, fingerprint=fingerprint)
+            if cursor_values is None:
+                raise KeysetCursorError("Invalid pagination cursor")
+            page_query = query.filter(_audit_cursor_condition(sort_column, sort_by=sort_by, sort_direction=sort_direction, cursor_values=cursor_values))
+        except KeysetCursorError as exc:
+            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
+        effective_page = page or (offset // effective_page_size + 1)
+        effective_offset = 0
+        start_cursor = cursor
     else:
-        effective_page = offset // effective_page_size + 1
-        effective_offset = offset
+        if page is not None or page_size is not None:
+            effective_page = min(page or 1, pages)
+            effective_offset = (effective_page - 1) * effective_page_size
+        else:
+            effective_page = offset // effective_page_size + 1
+            effective_offset = offset
+        page_query = query
+        if effective_offset > 0:
+            previous_row = ordered_query.offset(effective_offset - 1).limit(1).first()
+            if previous_row is not None:
+                start_cursor = _audit_cursor_for_row(previous_row, sort_by=sort_by, sort_direction=sort_direction, fingerprint=fingerprint)
+
+    rows_plus_one = page_query.order_by(order, AuditLog.id.desc()).offset(effective_offset).limit(effective_page_size + 1).all()
+    rows = rows_plus_one[:effective_page_size]
+    next_cursor = (
+        _audit_cursor_for_row(rows[-1], sort_by=sort_by, sort_direction=sort_direction, fingerprint=fingerprint)
+        if len(rows_plus_one) > effective_page_size and rows else None
+    )
+
+    return AuditAdminListResponse(
+        total=total,
+        page=effective_page,
+        page_size=effective_page_size,
+        pages=pages,
+        cursor=start_cursor,
+        next_cursor=next_cursor,
+        items=_audit_items(session, rows, access_admin),
+    )
+
+
+@router.get("/admin/audit/delta", response_model=AuditAdminDeltaResponse)
+def list_admin_audit_delta(
+    tenant_id: str | None = Query(default=None),
+    all_tenants: bool = Query(default=False),
+    audit_scope: str | None = Query(default=None, alias="scope"),
+    limit: int = Query(default=100, ge=1, le=500),
+    page_size: int | None = Query(default=None, ge=1, le=500),
+    cursor: str | None = Query(default=None),
+    sort_by: str = Query(default="time"),
+    sort_direction: str = Query(default="desc"),
+    filter_time: str | None = Query(default=None),
+    filter_actor: str | None = Query(default=None),
+    filter_action: str | None = Query(default=None),
+    filter_object: str | None = Query(default=None),
+    filter_tenant: str | None = Query(default=None),
+    since: str | None = Query(default=None),
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_any_scope("audit:read", "system:audit:read")),
+):
+    effective_scope = audit_scope or ("all" if all_tenants else "tenant")
+    if effective_scope not in {"tenant", "system", "all"}:
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Audit scope must be tenant, system or all.")
+    if sort_by not in {"time", "actor", "action", "object", "tenant"}:
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Unsupported audit sort column.")
+    if sort_direction not in {"asc", "desc"}:
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Audit sort direction must be asc or desc.")
+
+    query = session.query(AuditLog)
+    resolved_tenant_id: str | None = None
+    if effective_scope != "all":
+        query = query.filter(AuditLog.scope == effective_scope)
+    if effective_scope == "system":
+        if not has_scope(principal, "system:audit:read"):
+            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: system:audit:read")
+    elif effective_scope == "all" or all_tenants:
+        if not has_scope(principal, "system:audit:read"):
+            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: system:audit:read")
+    else:
+        if not has_scope(principal, "audit:read"):
+            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: audit:read")
+        tenant = _resolve_tenant(session, principal, tenant_id)
+        resolved_tenant_id = tenant.id
+        query = query.filter(AuditLog.tenant_id == tenant.id)
+
+    object_text = func.coalesce(AuditLog.object_type, "") + " " + func.coalesce(AuditLog.object_id, "")
+    access_admin = _access_administration()
+    filters = [
+        _audit_time_filter(filter_time),
+        _audit_actor_filter(access_admin, session, filter_actor),
+        _audit_text_filter(AuditLog.action, filter_action),
+        _audit_text_filter(object_text, filter_object),
+        _audit_text_filter(AuditLog.tenant_id, filter_tenant),
+    ]
+    for condition in filters:
+        if condition is not None:
+            query = query.filter(condition)
 
     sort_columns = {
         "time": AuditLog.created_at,
@@ -185,28 +482,108 @@ def list_admin_audit(
     }
     sort_column = sort_columns[sort_by]
     order = sort_column.asc() if sort_direction == "asc" else sort_column.desc()
-    rows = query.order_by(order, AuditLog.id.desc()).offset(effective_offset).limit(effective_page_size).all()
-    actor_email_by_user_id = access_admin.actor_email_by_user_id(session, {row.user_id for row in rows if row.user_id})
+    total = query.count()
+    effective_page_size = page_size or limit
+    pages = max(1, (total + effective_page_size - 1) // effective_page_size)
+    filters = _audit_filter_params(
+        filter_time=filter_time,
+        filter_actor=filter_actor,
+        filter_action=filter_action,
+        filter_object=filter_object,
+        filter_tenant=filter_tenant,
+    )
+    fingerprint = _audit_cursor_fingerprint(
+        effective_scope=effective_scope,
+        tenant_id=resolved_tenant_id,
+        page_size=effective_page_size,
+        sort_by=sort_by,
+        sort_direction=sort_direction,
+        filters=filters,
+    )
+    start_cursor: str | None = None
+    page_query = query
+    if cursor:
+        try:
+            cursor_values = decode_keyset_cursor(AUDIT_ADMIN_CURSOR_SCOPE, cursor, fingerprint=fingerprint)
+            if cursor_values is None:
+                raise KeysetCursorError("Invalid pagination cursor")
+            page_query = query.filter(_audit_cursor_condition(sort_column, sort_by=sort_by, sort_direction=sort_direction, cursor_values=cursor_values))
+            start_cursor = cursor
+        except KeysetCursorError as exc:
+            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
 
-    return AuditAdminListResponse(
+    if since is None:
+        rows_plus_one = page_query.order_by(order, AuditLog.id.desc()).limit(effective_page_size + 1).all()
+        rows = rows_plus_one[:effective_page_size]
+        next_cursor = (
+            _audit_cursor_for_row(rows[-1], sort_by=sort_by, sort_direction=sort_direction, fingerprint=fingerprint)
+            if len(rows_plus_one) > effective_page_size and rows else None
+        )
+        return AuditAdminDeltaResponse(
+            total=total,
+            page=1,
+            page_size=effective_page_size,
+            pages=pages,
+            cursor=start_cursor,
+            next_cursor=next_cursor,
+            items=_audit_items(session, rows, access_admin),
+            deleted=[],
+            watermark=_audit_delta_watermark(session, effective_scope=effective_scope, tenant_id=resolved_tenant_id),
+            has_more=False,
+            full=True,
+        )
+
+    entries, has_more = _audit_delta_entries(
+        session,
+        effective_scope=effective_scope,
+        tenant_id=resolved_tenant_id,
+        since=since,
+        limit=effective_page_size,
+    )
+    if entries is None:
+        rows_plus_one = page_query.order_by(order, AuditLog.id.desc()).limit(effective_page_size + 1).all()
+        rows = rows_plus_one[:effective_page_size]
+        next_cursor = (
+            _audit_cursor_for_row(rows[-1], sort_by=sort_by, sort_direction=sort_direction, fingerprint=fingerprint)
+            if len(rows_plus_one) > effective_page_size and rows else None
+        )
+        return AuditAdminDeltaResponse(
+            total=total,
+            page=1,
+            page_size=effective_page_size,
+            pages=pages,
+            cursor=start_cursor,
+            next_cursor=next_cursor,
+            items=_audit_items(session, rows, access_admin),
+            deleted=[],
+            watermark=_audit_delta_watermark(session, effective_scope=effective_scope, tenant_id=resolved_tenant_id),
+            has_more=False,
+            full=True,
+        )
+
+    changed_ids = [entry.resource_id for entry in entries if entry.resource_type == "audit_log"]
+    rows = (
+        page_query.filter(AuditLog.id.in_(changed_ids)).order_by(order, AuditLog.id.desc()).limit(effective_page_size).all()
+        if changed_ids else []
+    )
+    return AuditAdminDeltaResponse(
         total=total,
-        page=effective_page,
+        page=1,
         page_size=effective_page_size,
         pages=pages,
-        items=[
-            AuditAdminItem(
-                id=row.id,
-                scope=row.scope,
-                tenant_id=row.tenant_id,
-                actor_email=actor_email_by_user_id.get(row.user_id) if row.user_id else None,
-                action=row.action,
-                object_type=row.object_type,
-                object_id=row.object_id,
-                details=row.details or {},
-                created_at=row.created_at,
-            )
-            for row in rows
-        ],
+        cursor=start_cursor,
+        next_cursor=None,
+        items=_audit_items(session, rows, access_admin),
+        deleted=[],
+        watermark=_audit_delta_response_watermark(
+            session,
+            effective_scope=effective_scope,
+            tenant_id=resolved_tenant_id,
+            entries=entries,
+            has_more=has_more,
+        ),
+        has_more=has_more,
+        full=False,
     )
 
 

src/govoplan_audit/backend/api/v1/schemas.py

@@ -5,6 +5,8 @@ from typing import Any, Literal
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator
 
+from govoplan_core.api.v1.schemas import DeltaDeletedItem
+
 
 class AuditAdminItem(BaseModel):
     id: str
@@ -24,6 +26,15 @@ class AuditAdminListResponse(BaseModel):
     page: int = 1
     page_size: int = 100
     pages: int = 1
+    cursor: str | None = None
+    next_cursor: str | None = None
+
+
+class AuditAdminDeltaResponse(AuditAdminListResponse):
+    deleted: list[DeltaDeletedItem] = Field(default_factory=list)
+    watermark: str | None = None
+    has_more: bool = False
+    full: bool = False
 
 
 class AuditLogItemResponse(BaseModel):

src/govoplan_audit/backend/db/models.py

@@ -22,9 +22,9 @@ class AuditLog(Base, TimestampMixin):
 
     id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
     scope: Mapped[str] = mapped_column(String(20), default="tenant", nullable=False, index=True)
-    tenant_id: Mapped[str | None] = mapped_column(ForeignKey("tenants.id", ondelete="CASCADE"), nullable=True, index=True)
-    user_id: Mapped[str | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True)
-    api_key_id: Mapped[str | None] = mapped_column(ForeignKey("api_keys.id", ondelete="SET NULL"), nullable=True, index=True)
+    tenant_id: Mapped[str | None] = mapped_column(ForeignKey("tenancy_tenants.id", ondelete="CASCADE"), nullable=True, index=True)
+    user_id: Mapped[str | None] = mapped_column(ForeignKey("access_users.id", ondelete="SET NULL"), nullable=True, index=True)
+    api_key_id: Mapped[str | None] = mapped_column(ForeignKey("access_api_keys.id", ondelete="SET NULL"), nullable=True, index=True)
     action: Mapped[str] = mapped_column(String(100), nullable=False, index=True)
     object_type: Mapped[str | None] = mapped_column(String(100), index=True)
     object_id: Mapped[str | None] = mapped_column(String(100), index=True)