"""Bounded, stable listing for role-change lifecycle events. HTTP and MCP deliberately share this builder so their filters, pagination, and grant-count semantics cannot drift. The cursor is HMAC-signed because it contains an event identifier and the fixed time window used for the scan. """ from __future__ import annotations import base64 import binascii import hashlib import hmac import json from datetime import UTC, datetime, timedelta from typing import Any from uuid import UUID from sqlalchemy import and_, case, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.config import settings from app.core.coordinators.role_change_coordinator import ( REVIEW_STATUS_PENDING, ROLE_CHANGE_EVENT_TYPE, ) from app.core.exceptions import ValidationError from app.domains.cmdb.models import CmdbCloudAccessGrant from app.domains.hr.models import HrLifecycleEvent, HrRoleChangeReview from app.hr_lifecycle.mover_review_service import classify_move from app.identity.person_models import IdentityOrgMembership, IdentityPerson _CURSOR_VERSION = 1 _CURSOR_MAX_AGE = timedelta(hours=1) def _normalized_query(q: str | None) -> str: return q.strip().lower() if q else "" def _filter_fingerprint( *, days: int, only_pending: bool, q: str, page_size: int, cursor_subject: str, ) -> str: payload = json.dumps( { "days": days, "only_pending": only_pending, "page_size": page_size, "q": q, # A cursor is a continuation capability over admin-only HR data. # Bind it to the authenticated principal so copying it to another # otherwise-authorised account cannot continue that traversal. "subject": cursor_subject, }, sort_keys=True, separators=(",", ":"), ) return hashlib.sha256(payload.encode("utf-8")).hexdigest() def _cursor_signature(encoded_payload: str) -> str: return hmac.new( settings.JWT_SECRET_KEY.encode("utf-8"), b"role-change-list-v1\0" + encoded_payload.encode("ascii"), hashlib.sha256, ).hexdigest()[:32] def _encode_cursor( *, snapshot_at: datetime, window_start: datetime, after_created_at: datetime, after_id: UUID, page: int, filter_fingerprint: str, ) -> str: payload = json.dumps( { "v": _CURSOR_VERSION, "s": snapshot_at.isoformat(), "w": window_start.isoformat(), "a": after_created_at.isoformat(), "i": str(after_id), "p": page, "f": filter_fingerprint, }, sort_keys=True, separators=(",", ":"), ) encoded = base64.urlsafe_b64encode(payload.encode("utf-8")).decode("ascii") return f"{encoded}.{_cursor_signature(encoded)}" def _aware_datetime(value: object) -> datetime: parsed = datetime.fromisoformat(str(value)) if parsed.tzinfo is None or parsed.utcoffset() is None: raise ValueError("cursor datetime must be timezone-aware") return parsed.astimezone(UTC) def _as_utc(value: datetime) -> datetime: """Normalise driver-returned DATETIME values before cursor encoding. MySQL/TiDB commonly return a naive value even for ``timezone=True`` ORM columns. The database stores UTC, so attaching UTC here keeps cursors portable across the production driver and the timezone-aware test path. """ if value.tzinfo is None or value.utcoffset() is None: return value.replace(tzinfo=UTC) return value.astimezone(UTC) def _decode_cursor( raw: str, *, expected_filter_fingerprint: str, ) -> tuple[datetime, datetime, datetime, UUID, int]: try: encoded, signature = raw.rsplit(".", 1) if not hmac.compare_digest(signature, _cursor_signature(encoded)): raise ValueError("bad cursor signature") payload = json.loads( base64.urlsafe_b64decode(encoded.encode("ascii")).decode("utf-8"), ) if payload.get("v") != _CURSOR_VERSION: raise ValueError("unsupported cursor version") if payload.get("f") != expected_filter_fingerprint: raise ValueError("cursor filters do not match") snapshot_at = _aware_datetime(payload["s"]) window_start = _aware_datetime(payload["w"]) after_created_at = _aware_datetime(payload["a"]) after_id = UUID(str(payload["i"])) page = int(payload["p"]) now = datetime.now(UTC) if ( page < 2 or window_start > snapshot_at or not window_start <= after_created_at < snapshot_at or snapshot_at > now + timedelta(minutes=1) or now - snapshot_at > _CURSOR_MAX_AGE ): raise ValueError("invalid cursor bounds") except (ValueError, KeyError, TypeError, json.JSONDecodeError, binascii.Error) as exc: raise ValidationError("cursor is not valid") from exc return snapshot_at, window_start, after_created_at, after_id, page def _axis( before: Any, after: Any, *, labels: dict[str, str] | None = None, ) -> dict[str, Any]: """One before/after pair, resolved and pre-compared. ``changed`` is computed here rather than in the browser because the comparison has to be made on the RAW values (a department code, a person id) while the page shows the LABELS — and two different codes can resolve to the same display string when the org tree has been renamed. A frontend comparing what it renders would call that "no change" and hide the only axis that moved. """ lookup = labels or {} def _label(value: Any) -> str | None: if value in (None, ""): return None return lookup.get(str(value)) or str(value) return { "before": _label(before), "after": _label(after), "changed": (before or None) != (after or None), } def _serialize_row( event: HrLifecycleEvent, person: IdentityPerson, pending_count: int, total_count: int, review: HrRoleChangeReview | None = None, *, dept_labels: dict[str, str] | None = None, person_labels: dict[str, str] | None = None, ) -> dict[str, Any]: """One row of the console. The raw ``before`` / ``after`` blobs are still returned — MCP callers and the forensic timeline read them — but the console now gets the move already resolved into three named axes. It used to render ``JSON.stringify(before)`` into the two widest columns on the page, so the operator saw a department CODE and a bare manager UUID and could not tell who had moved where. Eight of the nine cards sent on 2026-09-05 differed ONLY in the manager axis, which was in that blob and unreadable. ``review`` is the packet for this move, when one exists. A title-only change deliberately has none (access follows the org and the reporting line, not the title string), and the page says so rather than drawing an empty progress bar. """ detail = event.detail or {} before = detail.get("before") or {} after = detail.get("after") or {} dept_labels = dict(dept_labels or {}) # A packet stores the org path as it was AT THE TIME. Prefer it over # today's org tree: a reviewer opening this months later should see # where the person actually moved from, not what that node is called # now (and the old node may not exist any more at all). if review is not None: if review.before_dept_code and review.before_dept_path: dept_labels[review.before_dept_code] = review.before_dept_path if review.after_dept_code and review.after_dept_path: dept_labels[review.after_dept_code] = review.after_dept_path # Explicitly ``Any``-valued: three axis dicts plus a nullable string # ``kind``, and an inferred value type would reject whichever was # added last. move: dict[str, Any] = { "department": _axis( before.get("dept_code"), after.get("dept_code"), labels=dept_labels, ), "manager": _axis( before.get("manager_person_id"), after.get("manager_person_id"), labels=person_labels, ), "job_title": _axis(before.get("job_title"), after.get("job_title")), } # 🔴 Delegated, not re-implemented. The precedence ("a move that # changes both department and manager IS a department change") lives # in ``classify_move``, which is also what decides whether a packet # gets built at all. A second copy here would let the console label # a move one way while the packet was built — or not built — on the # other, and the two would drift the first time either is touched. move["kind"] = classify_move(before, after) review_block: dict[str, Any] | None = None if review is not None: reviewer_id = ( str(review.reviewer_person_id) if review.reviewer_person_id else None ) review_block = { "id": str(review.id), "status": review.status, "reviewable_items": review.reviewable_items, "decided_items": review.decided_items, "total_items": review.total_items, # The two numbers that must never be conflated: how much the # person HOLDS vs how much this move calls into question. "delta_items": review.delta_items, "baseline_source": review.baseline_source, "reviewer": ( { "id": reviewer_id, "display_name": (person_labels or {}).get(reviewer_id or ""), } if reviewer_id else None ), # Whether anybody was actually told, and if not, whether that # was a decision. An admin looking at an untouched packet # needs to know which of the two it is. "notified_at": ( review.notified_at.isoformat() if review.notified_at else None ), "notify_suppressed_at": ( review.notify_suppressed_at.isoformat() if review.notify_suppressed_at else None ), "notify_suppressed_reason": review.notify_suppressed_reason, } return { "event_id": str(event.id), "person": { "id": str(person.id), "display_name": person.display_name, "work_email": person.work_email, "employee_no": person.employee_no, "employment_status": person.employment_status, "department": person.primary_department, "job_title": person.job_title, }, "effective_date": event.effective_date.isoformat(), "created_at": event.created_at.isoformat(), # Raw blobs kept for MCP / forensic callers. The console reads # ``move`` instead. "before": detail.get("before"), "after": detail.get("after"), "move": move, "review": review_block, "pending_count": int(pending_count), "total_count": int(total_count), } async def _resolve_labels( db: AsyncSession, rows: list[tuple[Any, ...]], ) -> tuple[dict[str, str], dict[str, str]]: """Batch-resolve department codes and person ids for one page. Two queries for the whole page, never one per row: this endpoint's contract is a bounded number of SELECTs regardless of row count, and resolving a manager name inside the serializer would quietly turn a 20-row page into 40 round trips. """ dept_codes: set[str] = set() person_ids: set[str] = set() for event, _person, _pending, _total, review, _cursor_at in rows: detail = event.detail or {} for side in ("before", "after"): blob = detail.get(side) or {} code = blob.get("dept_code") if code: dept_codes.add(str(code)) mgr = blob.get("manager_person_id") if mgr: person_ids.add(str(mgr)) if review is not None and review.reviewer_person_id: person_ids.add(str(review.reviewer_person_id)) dept_labels: dict[str, str] = {} if dept_codes: # ``org_path`` reads far better than the leaf name on a move # between two same-named leaves under different parents, which # is common in this org tree. for code, path, name in ( await db.execute( select( IdentityOrgMembership.org_node_source_id, IdentityOrgMembership.org_path, IdentityOrgMembership.org_node_name, ).where( IdentityOrgMembership.org_node_source_id.in_(dept_codes), ).distinct() ) ).all(): if code and code not in dept_labels: label = path or name if label: dept_labels[str(code)] = str(label)[:512] person_labels: dict[str, str] = {} if person_ids: # ``person_ids`` are strings out of a JSON blob; the column is a # GUID type, so bind UUID objects and skip anything unparseable # rather than letting one malformed id fail the whole page. parsed: list[UUID] = [] for raw in person_ids: try: parsed.append(UUID(raw)) except (TypeError, ValueError): continue if parsed: for pid, name in ( await db.execute( select(IdentityPerson.id, IdentityPerson.display_name) .where(IdentityPerson.id.in_(parsed)) ) ).all(): person_labels[str(pid)] = name or "" return dept_labels, person_labels async def build_role_change_list( db: AsyncSession, *, days: int = 30, page_size: int = 20, only_pending: bool = False, q: str | None = None, cursor: str | None = None, cursor_subject: str, ) -> dict[str, Any]: """Return one bounded page ordered by ``created_at DESC, id DESC``. The cursor fixes the scan window and advances strictly after the last row returned. A role-change inserted between page requests therefore appears on a fresh traversal, never in the middle of the current traversal where it could shift an OFFSET and duplicate or hide an existing row. """ if not 1 <= days <= 180: raise ValidationError("days must be between 1 and 180") if not 1 <= page_size <= 100: raise ValidationError("page_size must be between 1 and 100") if q is not None and len(q) > 100: raise ValidationError("q must be at most 100 characters") if cursor is not None and len(cursor) > 2000: raise ValidationError("cursor must be at most 2000 characters") if not cursor_subject: raise ValidationError("cursor subject is required") needle = _normalized_query(q) fingerprint = _filter_fingerprint( days=days, only_pending=only_pending, q=needle, page_size=page_size, cursor_subject=cursor_subject, ) if cursor: snapshot_at, window_start, after_at, after_id, page = _decode_cursor( cursor, expected_filter_fingerprint=fingerprint, ) else: # Migration 447 makes the ordering column DATETIME(6), matching this # application-clock precision. The exclusive boundary admits every # event committed before the request while excluding later inserts, # including inserts in the same wall-clock second. snapshot_at = datetime.now(UTC) window_start = snapshot_at - timedelta(days=days) after_at = None after_id = None page = 1 grant_counts = ( select( CmdbCloudAccessGrant.person_id.label("person_id"), func.count(CmdbCloudAccessGrant.id).label("total_count"), func.sum( case( ( CmdbCloudAccessGrant.review_status == REVIEW_STATUS_PENDING, 1, ), else_=0, ), ).label("pending_count"), ) .group_by(CmdbCloudAccessGrant.person_id) .subquery() ) person_grant_join = func.binary(grant_counts.c.person_id) == func.binary(IdentityPerson.id) pending_count = func.coalesce(grant_counts.c.pending_count, 0) total_count = func.coalesce(grant_counts.c.total_count, 0) filters = [ HrLifecycleEvent.event_type == ROLE_CHANGE_EVENT_TYPE, HrLifecycleEvent.created_at >= window_start, HrLifecycleEvent.created_at < snapshot_at, ] if needle: filters.append( or_( func.lower(IdentityPerson.display_name).contains( needle, autoescape=True, ), func.lower(IdentityPerson.work_email).contains( needle, autoescape=True, ), ), ) if only_pending: filters.append(pending_count > 0) total = int( await db.scalar( select(func.count(HrLifecycleEvent.id)) .select_from(HrLifecycleEvent) .join(IdentityPerson, IdentityPerson.id == HrLifecycleEvent.person_id) .outerjoin(grant_counts, person_grant_join) .where(*filters), ) or 0 ) page_filters = list(filters) if after_at is not None and after_id is not None: page_filters.append( or_( HrLifecycleEvent.created_at < after_at, and_( HrLifecycleEvent.created_at == after_at, # Production stores UUIDs as dashed CHAR(36), while # SQLAlchemy's generic ``Uuid`` comparator may bind the # RHS as 32 undashed hex characters. Comparing those # representations makes the page tail compare less than # itself, duplicating one row on every page. Bind a # plain dashed string and compare bytes, matching the # table's actual storage and deterministic ORDER BY. func.binary(HrLifecycleEvent.id) < func.binary(str(after_id)), ), ), ) # LEFT JOIN the packet on its natural key. Unique on # ``(person_id, effective_date)``, so this can never multiply a row — # which is why it is safe to add here and NOT to the COUNT query # above, whose result must stay provably identical to before. review_join = ( HrRoleChangeReview.person_id == HrLifecycleEvent.person_id ) & ( HrRoleChangeReview.effective_date == HrLifecycleEvent.effective_date ) result = await db.execute( select( HrLifecycleEvent, IdentityPerson, pending_count.label("pending_count"), total_count.label("total_count"), HrRoleChangeReview, # Read the ordering key as an independent SQL scalar. The ORM # identity map may hold a just-inserted Python datetime with # microseconds even though the production DATETIME column stores # seconds. Building a cursor from that cached value would make # the database row compare less than its own cursor and repeat # the page tail. This scalar always reflects stored precision. HrLifecycleEvent.created_at.label("cursor_created_at"), ) .join(IdentityPerson, IdentityPerson.id == HrLifecycleEvent.person_id) .outerjoin(grant_counts, person_grant_join) .outerjoin(HrRoleChangeReview, review_join) .where(*page_filters) .order_by( HrLifecycleEvent.created_at.desc(), HrLifecycleEvent.id.desc(), ) .limit(page_size + 1), ) rows = list(result.all()) has_more = len(rows) > page_size visible = rows[:page_size] next_cursor = None if has_more and visible: tail, _person, _pending, _total, _review, tail_created_at = visible[-1] next_cursor = _encode_cursor( snapshot_at=snapshot_at, window_start=window_start, after_created_at=_as_utc(tail_created_at), after_id=tail.id, page=page + 1, filter_fingerprint=fingerprint, ) dept_labels, person_labels = await _resolve_labels(db, visible) return { "rows": [ _serialize_row( event, person, pending, grant_total, review, dept_labels=dept_labels, person_labels=person_labels, ) for event, person, pending, grant_total, review, _cursor_created_at in visible ], "window_days": days, "page": page, "page_size": page_size, "total": total, "next_cursor": next_cursor, } __all__ = ["build_role_change_list"]