"""Scheduled-job handler for Slack seat sync.

``slack.seat_sync`` pulls every active full member of the PingCAP
Slack workspace once a day and reconciles it with Helix's
``software_seat_allocations`` for the Slack plan:

* **Seat import** — every Slack full member whose email matches an
  ``identity_persons`` row but has no active
  :class:`SoftwareSeatAllocation` on the Slack plan gets one created
  with ``reason='auto-imported from Slack'``. Back-fills seats that
  existed on Slack before Helix tracked them.

* **Drift cleanup** — Helix-side active seats whose person no longer
  appears in the Slack full-member list (IT admin removed them
  manually, or an offboarded person got deactivated) are flipped to
  ``status=revoked`` so the dashboard count matches reality. Seats
  younger than 2 hours are skipped to avoid racing with grants
  that are still in-flight.

Slack's free/Pro/Business+ tiers all support the two read scopes the
bot token needs (``users:read`` + ``users:read.email``), so this
handler works on every plan. The manual_notify invite flow is handled
separately by the provisioning worker + Feishu DM hook — this sync is
purely directory reconciliation.

Idempotent: re-running yields zero diffs in steady state. Per-user
errors are collected into the summary instead of aborting the whole
run — one flaky identity row shouldn't block the other 500.
"""
from __future__ import annotations

import logging
import uuid
from datetime import UTC, datetime
from typing import Any

from sqlalchemy import select
from sqlalchemy.exc import OperationalError
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import format_exception_for_log
from app.scheduled_jobs._seat_sync_helpers import (
    ExternalHolderObservation,
    import_dedup_filter,
    raise_if_total_failure,
    record_seat_reconciliation,
    safe_to_drift_revoke,
    write_drift_event_audit,
    write_sync_audit,
)
from app.scheduled_jobs.registry import register_scheduled_job

logger = logging.getLogger(__name__)


@register_scheduled_job(
    key="slack.seat_sync",
    display_name="Slack · 工作区成员同步",
    description=(
        "每天拉取 Slack 工作区的 full member 列表,把匹配到 Helix 员工的"
        "成员镜像到 software_seat_allocations,同时把 Slack 上已不在的"
        "成员对应的 Helix 坐席标为 revoked。Guest 和 Bot 会被忽略。"
        "需要 Bot Token 的 users:read + users:read.email scope,所有 "
        "Slack 档位都支持。"
    ),
)
async def run_slack_seat_sync(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    # Deadlock-victim restart on fresh sessions — same incident class
    # as jira.seat_sync 2026-06-12 (worker restart double-run racing
    # INSERTs). The reconcile is idempotent: import has the 25h-grace
    # dedup filter, revoke has safe_to_drift_revoke.
    from app.scheduled_jobs.handlers import run_with_lock_retry

    return await run_with_lock_retry(
        db,
        lambda session: _run_slack_seat_sync_once(session, config),
        label="slack.seat_sync",
    )


async def _run_slack_seat_sync_once(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from app.core.coordinators.slack_seat_coordinator import (
        get_slack_plan,
        load_slack_driver,
    )
    from app.domains.asset.models import (
        SoftwareExternalHolder,
        SoftwareSeatAllocation,
        SoftwareSeatSyncStat,
    )
    from app.identity.email_lookup import build_person_by_email_index

    driver = await load_slack_driver(db)
    if driver is None:
        return {
            "slack_members_listed": 0,
            "seats_imported": 0,
            "seats_revoked": 0,
            "errors": ["slack integration not configured or not enabled"],
        }

    plan = await get_slack_plan(db)
    if plan is None:
        return {
            "slack_members_listed": 0,
            "seats_imported": 0,
            "seats_revoked": 0,
            "errors": [
                "Slack license plan not found; run migration 056 or "
                "create a plan whose parent catalog name contains 'Slack'."
            ],
        }

    errors: list[str] = []

    # 1. List Slack full members (paginated inside the driver).
    # If the primary list fails (credential rotation, API outage),
    # we MUST raise so failure_notifier fires. Returning early with
    # ``errors=[...]`` lands ``status='ok'`` on the run row and the
    # admin gets no signal — drift cleanup will refuse (good) but the
    # import side silently goes dark for as long as the credential
    # stays broken.
    try:
        members = await driver.list_full_members()
    except Exception as e:  # noqa: BLE001
        raise RuntimeError(
            f"slack.list_full_members failed: {e}"
        ) from e

    # Normalise by email. Slack's ``profile.email`` is the canonical
    # member email; fall back to top-level ``email`` just in case.
    # Track no-email count separately from email-collision dropouts so the
    # reconciliation identity (vendor = matched + external + no_email)
    # holds — see the explicit counters used in the snapshot below.
    members_by_email: dict[str, dict[str, Any]] = {}
    members_no_email = 0
    for m in members:
        profile = m.get("profile") or {}
        email = str(
            profile.get("email") or m.get("email") or "",
        ).strip().lower()
        if email:
            members_by_email[email] = m
        else:
            members_no_email += 1

    # 2. Load identity_persons for matching emails.
    # Uses the alias-aware shared helper so a Slack member whose
    # email is one of the Helix person's mirrored Google Workspace
    # aliases (not the canonical work_email) still resolves.
    # Leavers are excluded so the daily reconcile doesn't re-import
    # an active seat for someone who is already gone on the Helix
    # side but still lingering on the Slack side. ``leaver_pending``
    # people INSIDE their notice period stay in scope (the helper's
    # default ``leaver_notice_grace``) — they are still working and
    # still hold the seat; revoking it the night IT confirms the
    # termination churned the allocation row and duplicated the
    # offboarding cancel event.
    person_by_email = await build_person_by_email_index(
        db, members_by_email.keys(),
        exclude_employment_statuses=("left", "leaver_pending"),
    )

    # 3. Load existing seats in the (plan, user) unique-index scope.
    # Active OR pending OR recently-revoked — see zoom_handlers comment
    # for the rationale (avoids IntegrityError under migration 119's
    # partial unique index when a stuck-pending row exists; ``import_
    # dedup_filter`` adds the 2h "freshly-revoked" arm so the import
    # branch doesn't race the vendor-side revoke task and re-create
    # a seat the admin just revoked).
    now = datetime.now(UTC)
    existing_all = list((await db.execute(
        select(SoftwareSeatAllocation).where(
            SoftwareSeatAllocation.plan_id == plan.id,
            import_dedup_filter(SoftwareSeatAllocation, now=now),
        )
    )).scalars().all())
    existing_by_user_id = {s.user_id: s for s in existing_all}
    existing_seats = [s for s in existing_all if s.status == "active"]

    # 4. Back-fill seats for Slack members who don't have one yet.
    seats_imported = 0
    # ``matched_email_count`` is the snapshot's matched value (one per
    # matched email, NOT per distinct person — see jira_handlers.py for
    # the off-by-N rationale). ``active_slack_user_ids`` below stays as
    # the per-PERSON set for the drift-revoke loop where per-person is
    # the correct semantic.
    matched_email_count = 0
    # External (unmatched) full members persisted for the reconciliation
    # drawer — Slack previously surfaced nothing about this gap at all.
    external_observations: list[ExternalHolderObservation] = []
    for email, _sm in members_by_email.items():
        person = person_by_email.get(email)
        if person is None:
            # Slack member whose email isn't in Helix — external
            # collaborator, ex-employee already deleted. Not an error.
            _profile = _sm.get("profile") or {}
            external_observations.append(ExternalHolderObservation(
                email=email,
                vendor_account_id=_sm.get("id"),
                account_status="active",
                display_name=(
                    _profile.get("real_name")
                    or _profile.get("display_name")
                    or None
                ),
            ))
            continue
        matched_email_count += 1
        if person.id in existing_by_user_id:
            continue  # already mirrored
        try:
            # Per-row SAVEPOINT — one bad INSERT (FK / unique-index
            # collision) rolls back just this row, not the session.
            async with db.begin_nested():
                db.add(SoftwareSeatAllocation(
                    plan_id=plan.id,
                    user_id=person.id,
                    status="active",
                    allocated_at=now,
                    reason="auto-imported from Slack via slack.seat_sync",
                ))
                await db.flush()
            seats_imported += 1
        except OperationalError:
            # Deadlock (1213) killed the whole transaction — propagate
            # so the fresh-session lock-retry wrapper restarts the run.
            raise
        except Exception as e:  # noqa: BLE001
            errors.append(
                f"{email}: import failed: {format_exception_for_log(e)}"
            )
            if len(errors) > 30:
                errors.append(
                    f"...truncated (total emails={len(members_by_email)})",
                )
                break

    # 5. Drift cleanup — Helix active seats whose Slack presence is gone.
    #
    # SAFETY GUARD via shared helper: refuses drift cleanup when the
    # vendor's matched-user count drops below 50% of Helix's active
    # seat count. Catches both (a) empty-tenant outages and (b) the
    # subtler "transient paginator truncation returns 5/500 users"
    # failure mode where a binary 0-vs-non-zero check passes but the
    # subsequent revoke loop nukes 495 paid seats.
    seats_revoked = 0
    active_slack_user_ids: set = set()
    for email in members_by_email:
        p = person_by_email.get(email)
        if p is not None:
            active_slack_user_ids.add(p.id)

    safe, reason = safe_to_drift_revoke(
        matched_count=len(members_by_email),
        existing_count=len(existing_seats),
        label="Slack seat sync",
        want_count=len(active_slack_user_ids),
    )
    if not safe:
        errors.append(reason or "drift cleanup refused")
    else:
        for seat in existing_seats:
            if seat.user_id in active_slack_user_ids:
                continue
            allocated = seat.allocated_at
            if allocated is not None:
                if allocated.tzinfo is None:
                    allocated = allocated.replace(tzinfo=UTC)
                if (now - allocated).total_seconds() < 7200:
                    # Too fresh — might be a grant still in flight.
                    continue
            # rollback expires modified instances; capture first
            seat_id, seat_user_id, seat_plan_id = (
                seat.id, seat.user_id, seat.plan_id,
            )
            try:
                # Per-seat SAVEPOINT — one bad flush must not poison
                # the session and take the rest of the run with it.
                async with db.begin_nested():
                    seat.status = "revoked"
                    seat.revoked_at = now
                    seat.reason = (
                        (seat.reason or "")
                        + "\n[auto-revoked by slack.seat_sync: "
                        "user no longer a full member on Slack]"
                    ).strip()
                    await write_drift_event_audit(
                        db,
                        job_key="slack.seat_sync",
                        seat_id=seat_id,
                        user_id=seat_user_id,
                        plan_id=seat_plan_id,
                        event="revoked",
                        detail={
                            "reason": "user no longer a full member on Slack",
                        },
                    )
                    await db.flush()
                seats_revoked += 1
            except OperationalError:
                # Whole-transaction rollback (deadlock) — see import loop.
                raise
            except Exception as e:  # noqa: BLE001
                errors.append(
                    f"seat {seat_id}: drift revoke failed: "
                    f"{format_exception_for_log(e)}"
                )

    await record_seat_reconciliation(
        db,
        holder_model=SoftwareExternalHolder,
        stat_model=SoftwareSeatSyncStat,
        integration_id=uuid.UUID(driver.integration_id),
        software_id=plan.software_id,
        plan_id=plan.id,
        provider="slack",
        product_key=None,
        site=None,
        # Post-email-dedupe billable identity count, NOT raw len(members) —
        # otherwise a Slack workspace where two accounts share an email
        # (rare; old SSO migration artefacts) breaks the reconciliation
        # identity. ``members_no_email`` covers Slack bots / shared-mailbox
        # accounts with no email at all.
        vendor_license_holders=len(members_by_email) + members_no_email,
        # Per-email match count (NOT per-person) so an employee with two
        # Slack accounts via aliased emails doesn't collapse to one in
        # the snapshot — same off-by-N pattern documented in
        # jira_handlers.py around the licensed_user_ids set.
        holders_matched=matched_email_count,
        holders_no_email=members_no_email,
        externals=external_observations,
        now=now,
    )

    summary = {
        "slack_members_listed": len(members),
        "full_members_with_email": len(members_by_email),
        "seats_imported": seats_imported,
        "seats_revoked": seats_revoked,
        "external_unmatched": len(external_observations),
        "errors": errors,
    }
    await write_sync_audit(db, job_key="slack.seat_sync", summary=summary)
    await db.commit()

    raise_if_total_failure(
        job_key="slack.seat_sync",
        summary=summary,
        progress_keys=("seats_imported", "seats_revoked"),
    )
    return summary