"""Identity data-quality findings — the continuous reconciliation layer of the Identity Hub. A cron (`identity.data_quality_scan`) sweeps the identity graph for the problems a mature identity source must surface — a departed person who can still log in, a manager pointer to someone who already left, a person stuck failing downstream sync — and upserts one row per problem here. The admin panel reads them; each finding links back to the person's golden record (bidirectional navigation). Two tables: * :class:`IdentityDataQualityFinding` — one row per (check, subject), deduped across runs via ``(finding_type, dedup_key)`` UNIQUE. A re-seen problem bumps ``last_seen_at``; a problem that disappears from a scan is auto-resolved. Mirrors the ``auditor_findings`` lifecycle shape so the open→acknowledged→accepted_risk→resolved vocabulary is consistent across the platform. * :class:`IdentityDataQualityRun` — per-scan history with per-type counts, so the panel can render a trend without parsing logs. """ from __future__ import annotations import uuid from datetime import datetime from typing import Any from sqlalchemy import ( JSON, Boolean, DateTime, ForeignKey, Index, Integer, String, Text, UniqueConstraint, func, ) from sqlalchemy.orm import Mapped, mapped_column from app.core.database import Base from app.core.sql_types import GUID as Uuid # Finding-type catalog (kept here as the single source of truth; the # scanner emits these and the frontend i18n-keys off them). FINDING_TYPES: tuple[str, ...] = ( "orphan_login_identity", # live login binding for a departed/deleted person "stale_tombstone", # HRIS-purged person that can still log in "manager_departed", # active person whose manager has left "missing_manager", # active person with no manager pointer "sync_drift_failed", # person stuck in last_status='failed' on an enabled target # 2026-07-04: Helix employment_status disagrees with SF ground truth — # Helix says active but SF User.status='f' (forfeited); leaver_pending # is excluded (offboarding already in flight; its SLA owns stuck cases). # The P0051 class: a real leaver still counted active in Helix. Detected # by pulling the live SF roster (list_all_users) per HRIS instance — the # daily safety net beside the incremental sync + weekly full sync. "hris_status_divergence", # 2026-09-07: HR's 邮箱名 (SF PerPersonal.customString4) is not an email # name — a display name ("Koji Betsuyaku"), a personal address, or text # Helix cannot use at all. Helix copes (it folds separators, else falls # back to the firstName/lastName derivation) so this never blocks a # hire; the finding exists so the dirty SF field gets CLEANED rather # than silently coped with forever. P01574: the value sat wrong in SF # for 3 months and surfaced only as a nightly "rename this mailbox" # card aimed at IT — the wrong desk for an HRIS data problem. "hris_email_name_malformed", # 2026-07-04: a leaver whose identity-plane session cutoff FAILED — # OffboardingProcess.session_enforcement.google_session.status='failed'. # Pure-DB read of the proof recorded by the offboarding force-logout path. "offboarding_session_not_cut", ) # Severity per type — drives the panel sort + the new-finding notification # threshold (only ``high`` findings page admins). FINDING_SEVERITY: dict[str, str] = { "orphan_login_identity": "high", "stale_tombstone": "high", "manager_departed": "medium", "sync_drift_failed": "medium", "missing_manager": "low", # HR hygiene, not an outage: Helix already copes with the value, so this # must never page anyone (only ``high`` pages). It sits in the panel # until HR fixes the SF field, then auto-resolves on the next scan. "hris_email_name_malformed": "low", # A real leaver still active in Helix (access not cut) — page IT. "hris_status_divergence": "high", # A leaver whose session cutoff failed — page IT. "offboarding_session_not_cut": "high", } class IdentityDataQualityFinding(Base): __tablename__ = "identity_data_quality_findings" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( UniqueConstraint( "finding_type", "dedup_key", name="uq_identity_dq_finding_dedup", ), Index("ix_identity_dq_finding_status_sev", "status", "severity"), Index("ix_identity_dq_finding_type_status", "finding_type", "status"), Index("ix_identity_dq_finding_person", "person_id"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) finding_type: Mapped[str] = mapped_column(String(64), nullable=False) # Stable dedup signature within a finding_type, e.g. the person id, the # external-identity id, or "{target_id}:{person_id}". Upsert key. dedup_key: Mapped[str] = mapped_column(String(255), nullable=False) severity: Mapped[str] = mapped_column( String(16), nullable=False, server_default="medium", ) # The person the finding is about (nullable — an orphan login identity # may point at a deleted person). ON DELETE SET NULL so a hard person # purge doesn't cascade-delete the audit trail of the problem. person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), ) subject_display: Mapped[str | None] = mapped_column(String(255)) title: Mapped[str] = mapped_column(String(512), nullable=False) detail: Mapped[dict[str, Any] | None] = mapped_column(JSON) # open | acknowledged | accepted_risk | resolved status: Mapped[str] = mapped_column( String(16), nullable=False, server_default="open", ) # True when a scan stopped seeing the problem (vs a human decision). auto_resolved: Mapped[bool] = mapped_column( Boolean, nullable=False, server_default="0", ) first_seen_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) last_seen_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) resolved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) decided_by_person_id: Mapped[uuid.UUID | None] = mapped_column( Uuid, ForeignKey("identity_persons.id", ondelete="SET NULL"), ) decided_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) decision_note: Mapped[str | None] = mapped_column(Text) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(), ) class IdentityDataQualityRun(Base): __tablename__ = "identity_data_quality_runs" __mapper_args__ = {"eager_defaults": True} __table_args__ = ( Index("ix_identity_dq_runs_started", "started_at"), ) id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) started_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, ) finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) duration_ms: Mapped[int | None] = mapped_column(Integer) status: Mapped[str] = mapped_column(String(16), nullable=False) trigger_source: Mapped[str] = mapped_column( String(64), nullable=False, server_default="cron", ) # Open-finding totals at end of scan (the panel trend reads these). open_total: Mapped[int | None] = mapped_column(Integer) open_high: Mapped[int | None] = mapped_column(Integer) new_count: Mapped[int | None] = mapped_column(Integer) resolved_count: Mapped[int | None] = mapped_column(Integer) # Per-type open counts, e.g. {"orphan_login_identity": 2, ...}. by_type: Mapped[dict[str, Any] | None] = mapped_column(JSON) error_message: Mapped[str | None] = mapped_column(Text)