"""Orphan-account business logic. Discovery + drift cleanup live in the scheduled-job handler — this service is the *write* side: assign owners, set purpose, list / search, and the callable used by ``OffboardingService._scan_blockers`` to find orphans owned by a given person. """ from __future__ import annotations import logging from datetime import datetime, timedelta from typing import Any from uuid import UUID from sqlalchemy import case, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import AppException, NotFoundError from app.core.txn import best_effort from app.domains.google_workspace.models import GoogleOrphanAccount from app.identity.person_models import IdentityPerson logger = logging.getLogger(__name__) # Leaver-remnant fast-path epsilon for ``soft_delete_stale``: a row # observed by the CURRENT sync tick carries ``last_seen_at == now``, # but MySQL DATETIME truncates microseconds, so a strict ``< now`` # comparison would sweep rows the tick just stamped. One hour is far # above truncation error and far below the daily cron cadence. _LEAVER_SWEEP_EPSILON = timedelta(hours=1) def _serialize( row: GoogleOrphanAccount, *, owner: IdentityPerson | None = None, former_employee: bool = False, ) -> dict[str, Any]: return { "id": str(row.id), "primary_email": row.primary_email, "google_user_id": row.google_user_id, "display_name": row.display_name, "org_unit_path": row.org_unit_path, "suspended": row.suspended, "is_admin": row.is_admin, "creation_time": row.creation_time.isoformat() if row.creation_time else None, "last_login_time": ( row.last_login_time.isoformat() if row.last_login_time else None ), "owner_user_id": str(row.owner_user_id) if row.owner_user_id else None, "owner_display_name": owner.display_name if owner else None, "owner_work_email": owner.work_email if owner else None, "owner_employment_status": ( owner.employment_status if owner else None ), "purpose": row.purpose, # True when ``primary_email`` matches an identity_persons row # with ``employment_status='left'`` — the account is a leaver # remnant that fell into the orphan inventory because its # human left, not a deliberately shared mailbox. "former_employee": former_employee, "first_seen_at": row.first_seen_at.isoformat(), "last_seen_at": row.last_seen_at.isoformat(), "deleted_at": row.deleted_at.isoformat() if row.deleted_at else None, } class OrphanAccountService: def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) async def _left_person_emails( self, emails: list[str], ) -> set[str]: """Subset of ``emails`` that belong to a person with ``employment_status='left'`` (case-insensitive). Orphan ``primary_email`` is stored lowercase; person ``work_email`` casing is HRIS-controlled, so lower both sides.""" wanted = {e.lower() for e in emails if e} if not wanted: return set() rows = (await self.db.execute( select(func.lower(IdentityPerson.work_email)).where( func.lower(IdentityPerson.work_email).in_(wanted), IdentityPerson.employment_status == "left", ), )).scalars().all() return set(rows) async def list( self, *, unassigned_only: bool = False, owner_user_id: UUID | None = None, search: str | None = None, include_suspended: bool = True, include_deleted: bool = False, limit: int | None = 1000, offset: int = 0, ) -> list[dict[str, Any]]: """List orphans with eager-loaded owner. ``limit=None`` removes the row cap entirely — used by the CSV export, which is a compliance inventory and must not silently truncate. The paginated listing endpoint keeps a generous default ceiling. Filters compose: ``unassigned_only`` returns rows with NULL owner; ``owner_user_id`` returns rows owned by exactly that person (used by the leaver-blocker UI); ``search`` matches primary_email / display_name; ``include_suspended=False`` drops suspended rows from the listing (admin UI default); ``include_deleted=True`` surfaces rows the seat sync has marked as deleted-upstream (default hides them — owner + audit history are preserved on the row). """ stmt = select(GoogleOrphanAccount) if unassigned_only: stmt = stmt.where(GoogleOrphanAccount.owner_user_id.is_(None)) if owner_user_id is not None: stmt = stmt.where(GoogleOrphanAccount.owner_user_id == owner_user_id) if not include_suspended: stmt = stmt.where(GoogleOrphanAccount.suspended.is_(False)) if not include_deleted: stmt = stmt.where(GoogleOrphanAccount.deleted_at.is_(None)) if search: term = f"%{search.strip().lower()}%" stmt = stmt.where(or_( GoogleOrphanAccount.primary_email.ilike(term), GoogleOrphanAccount.display_name.ilike(term), )) # Unassigned first so admins triage them; then alphabetical. # MySQL has no NULLS FIRST/LAST so we lift it into a CASE # expression that ranks NULL=0, NOT NULL=1. stmt = stmt.order_by( case((GoogleOrphanAccount.owner_user_id.is_(None), 0), else_=1).asc(), GoogleOrphanAccount.primary_email.asc(), ) # Bound the scan: this was the one list endpoint with no cap. The # orphan set is low-hundreds today, so a generous default ceiling is # a defensive guard rather than a behaviour change for real data. # ``limit=None`` (CSV export) intentionally skips the cap so the # compliance inventory is never silently truncated. if limit is not None: stmt = stmt.limit(max(1, limit)) stmt = stmt.offset(max(0, offset)) rows = list((await self.db.execute(stmt)).scalars().all()) owner_ids = {r.owner_user_id for r in rows if r.owner_user_id} owners: dict[UUID, IdentityPerson] = {} if owner_ids: owner_rows = (await self.db.execute( select(IdentityPerson).where( IdentityPerson.id.in_(owner_ids), ), )).scalars().all() owners = {p.id: p for p in owner_rows} left_emails = await self._left_person_emails( [r.primary_email for r in rows], ) return [ _serialize( r, owner=owners.get(r.owner_user_id) if r.owner_user_id else None, former_employee=r.primary_email.lower() in left_emails, ) for r in rows ] async def list_orphans_owned_by( self, person_id: UUID, ) -> list[GoogleOrphanAccount]: """Return the raw rows owned by a person — used by the offboarding scanner to build blocker rows. Kept as ORM rows so the caller has the full record handy for the blocker label. Filters ``deleted_at IS NULL`` — an orphan that's already gone from Google upstream doesn't need a hand-over blocker on the leaver. Otherwise the leaver would be stuck on a phantom row that no operator can actually re-assign in the Google console. """ rows = (await self.db.execute( select(GoogleOrphanAccount).where( GoogleOrphanAccount.owner_user_id == person_id, GoogleOrphanAccount.deleted_at.is_(None), ), )).scalars().all() return list(rows) async def assign_owner( self, orphan_id: UUID, new_owner_id: UUID, *, purpose: str | None = None, trace_id: str, actor_user_id: UUID, ) -> dict[str, Any]: """Set / replace the responsible person for an orphan account. ``new_owner_id`` MUST resolve to an active person — assigning ownership to a leaver would just queue another blocker on them, so we reject up-front. ``purpose`` is optional; if provided it's stored verbatim, if omitted the existing value is left alone (this method is also called by the leaver- blocker handover where purpose hasn't changed). """ row = await self.db.get(GoogleOrphanAccount, orphan_id) if row is None: raise NotFoundError("GoogleOrphanAccount", str(orphan_id)) new_owner = await self.db.get(IdentityPerson, new_owner_id) if new_owner is None or new_owner.deleted_at is not None: raise AppException( code="OWNER_NOT_FOUND", message="新负责人不存在", status_code=400, ) if new_owner.employment_status != "active": raise AppException( code="OWNER_INACTIVE", message=( f"新负责人 {new_owner.display_name or new_owner.work_email} " f"当前状态为 {new_owner.employment_status}," "不能作为共用账号负责人" ), status_code=400, ) previous_owner_id = row.owner_user_id row.owner_user_id = new_owner_id if purpose is not None: row.purpose = purpose.strip() or None await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="google_orphan.owner_assigned", status="success", resource_type="google_orphan_account", resource_id=str(row.id), detail={ "primary_email": row.primary_email, "previous_owner_id": ( str(previous_owner_id) if previous_owner_id else None ), "new_owner_id": str(new_owner_id), }, ) left_emails = await self._left_person_emails([row.primary_email]) return _serialize( row, owner=new_owner, former_employee=row.primary_email.lower() in left_emails, ) async def update_purpose( self, orphan_id: UUID, purpose: str | None, *, trace_id: str, actor_user_id: UUID, ) -> dict[str, Any]: row = await self.db.get(GoogleOrphanAccount, orphan_id) if row is None: raise NotFoundError("GoogleOrphanAccount", str(orphan_id)) row.purpose = (purpose or "").strip() or None await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="google_orphan.purpose_updated", status="success", resource_type="google_orphan_account", resource_id=str(row.id), detail={"primary_email": row.primary_email}, ) owner = ( await self.db.get(IdentityPerson, row.owner_user_id) if row.owner_user_id else None ) left_emails = await self._left_person_emails([row.primary_email]) return _serialize( row, owner=owner, former_employee=row.primary_email.lower() in left_emails, ) async def upsert_from_google( self, *, primary_email: str, google_user_id: str | None, display_name: str | None, org_unit_path: str | None, suspended: bool, is_admin: bool, creation_time: datetime | None, last_login_time: datetime | None, observed_at: datetime, ) -> tuple[GoogleOrphanAccount, bool]: """Insert-or-update from the seat-sync job. Returns ``(row, created)``. ``last_seen_at`` is always bumped to ``observed_at`` so the UI can display "last observed". ``owner_user_id`` is NEVER touched here — owner assignment is an operator action, not something we infer from Google data. Revive on observation: if a previously soft-deleted row reappears in the Google tenant list (account restored, or a brand-new account happened to land on the same primary_email), ``deleted_at`` is cleared. Owner + purpose stay intact so the admin's earlier audit trail isn't lost. """ row = (await self.db.execute( select(GoogleOrphanAccount).where( GoogleOrphanAccount.primary_email == primary_email, ), )).scalar_one_or_none() created = False if row is None: row = GoogleOrphanAccount( primary_email=primary_email, first_seen_at=observed_at, ) self.db.add(row) created = True row.google_user_id = google_user_id or row.google_user_id row.display_name = display_name row.org_unit_path = org_unit_path row.suspended = suspended row.is_admin = is_admin if creation_time is not None: row.creation_time = creation_time if last_login_time is not None: row.last_login_time = last_login_time row.last_seen_at = observed_at # Round-27 pre-launch (2026-05-25): un-soft-delete on # re-observation MUST emit an audit row, mirroring the cmdb # sync ``cmdb.sync.resurrected`` invariant # (``feedback_on_duplicate_key_preserve_consistency`` + # ``feedback_no_speculative_backfill``). Without this, a row # previously suppressed by ``soft_delete_stale`` (sweep # tombstone) silently re-appears in admin UI with no forensic # trail — operators can't tell "account was restored" from # "sweep was wrong". The audit is best-effort: never block the # sync on audit-write failure. prior_deleted_at = row.deleted_at if prior_deleted_at is not None: row.deleted_at = None async with best_effort( self.db, label="orphan_service best-effort step", log=logger, ) as be_360_step: from app.audit.service import AuditService # noqa: PLC0415 audit_svc = AuditService(self.db) await audit_svc.log( trace_id="google.seat_sync.orphan.resurrected", actor_type="system", actor_user_id=None, action="google.orphan.resurrected", status="success", resource_type="google_orphan_account", resource_id=str(row.id) if row.id else None, detail={ "primary_email": primary_email, "prior_deleted_at": prior_deleted_at.isoformat(), "observed_at": observed_at.isoformat(), "google_user_id": google_user_id, }, ) if not be_360_step.ok: import logging # noqa: PLC0415 logging.getLogger(__name__).warning( "google.orphan_resurrected audit write failed for " "primary_email=%s: %s", primary_email, be_360_step.reason(), ) return row, created async def soft_delete_stale( self, *, now: datetime, grace_days: int, ) -> list[GoogleOrphanAccount]: """Stamp ``deleted_at`` on rows whose ``last_seen_at`` is older than ``now - grace_days``. Called from ``google.seat_sync`` after Phase C finishes so only rows the current tick failed to observe are eligible. ``grace_days <= 0`` disables the soft-delete sweep entirely (operator escape hatch — same convention as Feishu's ``drift_grace_days``). Leaver-remnant fast path (owner-reported 2026-06-11, ``chujie@pingcap.com``): an orphan whose ``primary_email`` matches a person with ``employment_status='left'`` is not a shared mailbox — it's the leaver's own account that fell into the inventory because the human left. Once the admin deletes it in the Google console there is no reason to keep showing it for another ``grace_days`` days, so those rows are swept as soon as a successful tick fails to observe them. Safe because ``list_all_users`` fail-stops on any page error and Phase C/D are skipped entirely on an empty tenant list — a successful run is always a complete roster, never a partial one. Returns the rows just stamped so the caller can include their emails in the run summary / audit. Idempotent: rows already soft-deleted (``deleted_at IS NOT NULL``) are skipped. """ if grace_days <= 0: return [] cutoff = now - timedelta(days=grace_days) stamped: dict[UUID, GoogleOrphanAccount] = {} rows = (await self.db.execute( select(GoogleOrphanAccount).where( GoogleOrphanAccount.last_seen_at < cutoff, GoogleOrphanAccount.deleted_at.is_(None), ), )).scalars().all() for r in rows: r.deleted_at = now stamped[r.id] = r leaver_rows = (await self.db.execute( select(GoogleOrphanAccount) .join( IdentityPerson, func.lower(IdentityPerson.work_email) == GoogleOrphanAccount.primary_email, ) .where( IdentityPerson.employment_status == "left", GoogleOrphanAccount.deleted_at.is_(None), GoogleOrphanAccount.last_seen_at < now - _LEAVER_SWEEP_EPSILON, ), )).scalars().all() fast_swept: list[GoogleOrphanAccount] = [] for r in leaver_rows: if r.id in stamped: continue r.deleted_at = now stamped[r.id] = r fast_swept.append(r) if fast_swept: logger.info( "google.seat_sync: leaver-remnant fast sweep " "tombstoned %d orphan(s) without waiting for the " "%d-day grace: %s", len(fast_swept), grace_days, ", ".join(r.primary_email for r in fast_swept[:20]), ) if stamped: await self.db.flush() return list(stamped.values())