"""Runner base class + per-person outcome dataclass.

Concrete runners ( ``ScimRunner``, ``GoogleSchemaRunner`` ) implement
the wire-protocol methods. Each runner declares which operations it
supports via class attributes ``supports_deactivate`` and
``supports_delete``; the orchestrator inspects those flags before
calling and short-circuits to ``status='excluded'`` (no API call,
state row marked) when an operation isn't supported.

Methods:

* :meth:`upsert` (REQUIRED, abstract) — create or update one person.
* :meth:`deactivate` — flip downstream user to inactive. Implement
  iff the class sets ``supports_deactivate=True`` (default). Base
  class raises NotImplementedError so a missing override fails
  loudly rather than returning a misleading success.
* :meth:`delete` — hard-delete or clear-fields. Same contract as
  ``deactivate`` — gated by ``supports_delete``.
* :meth:`describe` (REQUIRED, abstract) — short label for log lines.
* :meth:`aclose` — optional cleanup hook for HTTP sessions.

Runners do NOT touch IdentitySyncState — the orchestrator does. This
split lets runners be pure HTTP clients with no DB dependency.
"""
from __future__ import annotations

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any

# ``PersonOutcome.detail['blocked_reason']`` values that describe an
# EXPECTED, self-clearing state rather than something a human must fix.
# The blocked-backlog notifier skips these: a per-run create budget that
# deliberately spreads a backlog across several sweeps is working as
# designed, and paging IT for it every night would train them to ignore
# the alert that does matter.
#
# Membership is asserted against the runner-side constants in tests, so a
# renamed reason can't silently start paging (or silently stop).
SELF_DRAINING_BLOCKED_REASONS = frozenset({
    "zoom_create_budget_exhausted",
})


@dataclass(slots=True)
class PersonOutcome:
    """Result of a single per-person operation.

    ``status`` mirrors the IdentitySyncState.last_status enum but
    additionally carries ``unchanged`` (would have skipped at hash
    layer; runners don't usually return this — the orchestrator does
    the skip — but it's here for tests / single-person force paths).
    """
    status: str  # success | skipped | failed | deactivated | deleted | unchanged | excluded
    external_id: str | None = None
    error: str | None = None
    # Optional debug payload — kept off the per-person row but echoed
    # into the run summary for the most recent failures.
    detail: dict[str, Any] | None = None
    # Observability (recorded verbatim on the per-person ledger row):
    #   ``request_body`` — the ACTUAL wire body sent to the vendor for this
    #     person (final shape after every transform: department_ids, orders,
    #     e164 mobile, custom_attrs, …). NOT the mapping-rendered payload —
    #     that hid the real department_ids and made 41041 undiagnosable.
    #   ``response`` — the vendor's response envelope (code / msg / id), or
    #     the parsed error on failure. So the ledger shows exactly what we
    #     sent AND what came back, per person, without a live re-probe.
    # A runner that touches the vendor SHOULD populate both; a hash-skip /
    # gate path leaves them None (nothing was sent).
    request_body: dict[str, Any] | None = None
    response: dict[str, Any] | None = None
    # Per-slot outcomes (PR2 resilient writes): when a write degrades by
    # dropping an un-appliable slot (a dept the vendor rejects, a bad
    # mobile, …), each entry records {slot, status, code, detail} so the
    # ledger/UI shows field-level truth instead of one opaque row status.
    slot_outcomes: list[dict[str, Any]] | None = None


@dataclass(slots=True)
class RunSummary:
    """Aggregate counters returned to the service after a sweep.

    Person-keyed counters (``persons_in_scope`` + ``created`` /
    ``updated`` / ``skipped`` / ``deactivated`` / ``deleted`` /
    ``excluded`` / ``failed``) cover Phase B (and every non-Feishu
    kind, which is single-phase).

    Dept-keyed counters (``dept_*``) are populated by Phase A of
    ``feishu_contact`` and stay zero for every other kind. Surfacing
    them in the same struct lets one IdentitySyncRun row carry both
    halves of a combined sweep without splitting into two runs.
    """
    persons_in_scope: int = 0
    # Of ``persons_in_scope``, how many were pulled in by the sweep
    # rather than matching the filter directly (e.g. CN managers of
    # foreign reports for a foreigners-only Navan target). Subset of
    # ``persons_in_scope``, not additive.
    auto_included: int = 0
    # Number of first-time notifications enqueued during this sweep
    # (subset of ``auto_included`` — only fires when the upsert
    # succeeded and ``notified_at`` was previously NULL).
    auto_included_notified: int = 0
    created: int = 0
    updated: int = 0
    skipped: int = 0
    # Of ``skipped``, how many came back FROM the runner — i.e. the runner
    # made a vendor round-trip, the vendor ANSWERED, and the answer was
    # "nothing to change here" (google_schema's GET-then-diff, zoom's
    # already-active user, scim's domain-not-allowed refusal, …). The rest
    # of ``skipped`` is the local payload-hash short-circuit, which issues
    # no HTTP at all.
    #
    # The distinction is load-bearing for the run-status split: a
    # vendor-verified skip is proof the target is UP (that is exactly what
    # a hash-skip is not — see ``test_credential_outage_run_status``).
    # Subset of ``skipped``, not additive.
    verified_skipped: int = 0
    deactivated: int = 0
    deleted: int = 0
    excluded: int = 0
    failed: int = 0
    # Persons the vendor rejected for a NON-transient, data-dependent
    # reason (missing required field, unmapped value_map code, resigned
    # vendor account, dept not in the active org tree, …). Distinct from
    # ``failed`` (retry-worthy) and ``excluded`` (policy no-op): a
    # ``blocked`` row is terminal-until-the-underlying-data-changes — it
    # persists its hash so subsequent sweeps re-affirm it cheaply WITHOUT
    # re-burning a vendor call, and auto-retries the moment the rendered
    # payload changes (i.e. the data is fixed). Keeps the run GREEN (a
    # handful of people awaiting a data/config fix is not a sync failure)
    # while surfacing an actionable, deduplicated backlog to the admin.
    blocked: int = 0
    # STRICT value_map misses seen this sweep: {rule_target_field: {code, ...}}
    # → surfaced as a "new downstream code with no value_map entry" signal
    # so a brand-new HRIS code (new legal entity, new employee-type) is a
    # one-time config task, not a silent field-drop that freezes vendor data.
    value_map_misses: dict[str, set] = field(default_factory=dict)
    # Phase A (dept sweep, feishu_contact only) counters.
    dept_in_scope: int = 0
    dept_created: int = 0
    dept_updated: int = 0
    dept_soft_deleted: int = 0
    dept_failed: int = 0
    # Org nodes deliberately NOT projected because their subtree holds
    # no in-scope (post-filter) member — empty / out-of-scope shells.
    dept_skipped: int = 0
    errors: list[str] = field(default_factory=list)
    extra: dict[str, Any] = field(default_factory=dict)

    def as_dict(self) -> dict[str, Any]:
        return {
            "persons_in_scope": self.persons_in_scope,
            "auto_included": self.auto_included,
            "auto_included_notified": self.auto_included_notified,
            "created": self.created,
            "updated": self.updated,
            "skipped": self.skipped,
            "verified_skipped": self.verified_skipped,
            "deactivated": self.deactivated,
            "deleted": self.deleted,
            "excluded": self.excluded,
            "failed": self.failed,
            "blocked": self.blocked,
            "value_map_misses": {
                k: sorted(v) for k, v in self.value_map_misses.items() if v
            },
            "dept_in_scope": self.dept_in_scope,
            "dept_created": self.dept_created,
            "dept_updated": self.dept_updated,
            "dept_soft_deleted": self.dept_soft_deleted,
            "dept_failed": self.dept_failed,
            "dept_skipped": self.dept_skipped,
            "errors": self.errors[:50],
            "errors_truncated": len(self.errors) > 50,
            **self.extra,
        }


class SyncRunner(ABC):
    """Per-kind wire-protocol adapter.

    Runners are instantiated per-target by the service layer and
    discarded after the run — they may hold per-instance state (HTTP
    session) but should not cache cross-run state.

    Capability flags:

    * :attr:`supports_deactivate` — when False, the orchestrator
      MUST NOT call ``deactivate()``; it short-circuits to
      ``status='excluded'`` for that person without an API call.
      Used by ``GoogleSchemaRunner`` where Google account suspension
      is a separate domain (``google_workspace.account_provisioner``)
      and "deactivate the schema field projection" is meaningless —
      the right behavior is "stop touching this person's record."
    * :attr:`supports_delete` — same, for ``delete()``. ``False`` means
      the runner doesn't expose a delete primitive; the orchestrator
      short-circuits the person to ``status='excluded'`` with NO
      vendor call (it does NOT downgrade to deactivate), and
      ``_validate_on_offboard`` rejects ``on_offboard=delete`` at
      target save time for such kinds so the silent-skip can't be
      configured in the first place.
    """

    supports_deactivate: bool = True
    supports_delete: bool = True

    @abstractmethod
    async def upsert(
        self,
        *,
        person_email: str,
        payload: dict[str, Any],
        prior_external_id: str | None,
    ) -> PersonOutcome:
        """Create or update a downstream user."""

    async def deactivate(
        self,
        *,
        person_email: str,
        prior_external_id: str | None,
    ) -> PersonOutcome:
        """Mark a downstream user inactive without deleting.

        Subclasses that set ``supports_deactivate = False`` MUST NOT
        override this method — the orchestrator never calls it on
        them. Subclasses that set it ``True`` (default) MUST override
        with a real implementation; the base raises NotImplementedError
        so a missing override fails loudly instead of silently
        returning a misleading success.
        """
        raise NotImplementedError(
            f"{type(self).__name__}.deactivate() not implemented; either "
            "implement it or set supports_deactivate=False on the class."
        )

    async def delete(
        self,
        *,
        person_email: str,
        prior_external_id: str | None,
    ) -> PersonOutcome:
        """Hard-delete a downstream user.

        Same loud-failure contract as ``deactivate``: subclasses with
        ``supports_delete=True`` must implement; the orchestrator
        never calls this on classes that set the flag to False.
        """
        raise NotImplementedError(
            f"{type(self).__name__}.delete() not implemented; either "
            "implement it or set supports_delete=False on the class."
        )

    @abstractmethod
    def describe(self) -> str:
        """Short human label for logs (target name not included)."""

    async def aclose(self) -> None:
        """Optional: release any per-runner resources (HTTP session)."""
        return None


@dataclass(slots=True)
class DeptOutcome:
    """Result of a single per-department operation.

    Mirrors :class:`PersonOutcome` but for dept-keyed runners. The
    dept-sweep orchestrator (``service._sweep_departments``) walks this
    in topological order and writes state to ``identity_sync_department_state``.
    """
    status: str  # success | skipped | failed | soft_deleted | unchanged
    external_id: str | None = None
    error: str | None = None
    detail: dict[str, Any] | None = None


class DeptSyncRunner(ABC):
    """Per-kind dept-shaped wire-protocol adapter.

    Departments are pushed in topological order (parent before child),
    so the orchestrator hands the runner one dept at a time and pre-
    sorts the input. The runner does NOT topo-sort itself — that's the
    orchestrator's job because it owns the IdentityOrgNode snapshot.

    Capabilities mirror SyncRunner:

    * :attr:`supports_soft_delete` — when True, the orchestrator may
      call :meth:`soft_delete` for depts that have disappeared from
      HRIS. When False, the orchestrator short-circuits to
      ``status='excluded'`` without an API call.

    Methods:

    * :meth:`upsert_dept` (REQUIRED) — create or update one dept.
    * :meth:`soft_delete` — mark a dept inactive without removing
      historical references. Default raises NotImplementedError so a
      missing override fails loudly. Subclasses with
      ``supports_soft_delete = False`` MUST NOT call this method.
    * :meth:`describe` (REQUIRED) — short label for log lines.
    * :meth:`aclose` — optional resource cleanup.
    """

    supports_soft_delete: bool = True

    @abstractmethod
    async def upsert_dept(
        self,
        *,
        dept_code: str,
        dept_name: str,
        parent_dept_code: str | None,
        leader_email: str | None,
        dept_order: int | None,
        prior_external_id: str | None,
    ) -> DeptOutcome:
        """Create or update one department.

        Takes raw kwargs instead of a mapping-rule-rendered payload
        because dept fields are admin-invariant (Feishu's
        ``code/name/parent/leader/order`` slots are fixed). The
        ``feishu_contact`` orchestrator's Phase A passes these
        directly from the IdentityOrgNode model — no mapping_config
        round-trip for dept rows.
        """

    async def soft_delete(
        self,
        *,
        dept_code: str,
        prior_external_id: str | None,
    ) -> DeptOutcome:
        """Mark a department inactive (vendor-specific semantics).

        For Feishu: ``PATCH /contact/v3/departments/{id}`` with
        ``status.is_deleted=true``. Hard delete is intentionally NOT
        exposed by this runner family — see ADR in
        ``docs/architecture/DECISIONS.md``.
        """
        raise NotImplementedError(
            f"{type(self).__name__}.soft_delete() not implemented; either "
            "implement it or set supports_soft_delete=False on the class."
        )

    @abstractmethod
    def describe(self) -> str:
        """Short human label for logs (target name not included)."""

    async def aclose(self) -> None:
        return None