"""Run-failure alert dispatch + dedup for the Identity Sync Center.

Regression for the reported gap: an ``event:hire_confirm`` Google
Workspace dept push 403'd and produced NO alert, because the only
failure-notification path was the cron dispatcher's raise-on-failure —
which event / manual runs never traverse. The fix routes the alert at
:meth:`IdentitySyncService._finalise_run`, the chokepoint every run
(cron / manual / event) passes through, and lets admins configure per-
target recipients.

Pins:

* a ``failed`` run with explicit ``alert_recipient_ids`` notifies exactly
  those people (in-app + Feishu + email outbox row);
* an empty recipient list notifies NOBODY — there is no IT-admin fallback
  (2026-06-19: "不要默认通知管理员", see DECISIONS.md);
* ``alert_enabled=False`` suppresses the alert entirely, even with
  recipients configured;
* a ``partial`` run (some per-person failures) also alerts;
* two consecutive same-signature failures dedup to a single push;
* a changed failure kind re-notifies.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest
import pytest_asyncio
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from app.auth.models import AuthUserRole
from app.domains.identity_sync.alert_notifier import maybe_notify_run_failure
from app.domains.identity_sync.models import IdentitySyncRun, IdentitySyncTarget
from app.identity.person_models import IdentityPerson
from app.notifications.models import NotificationOutbox

pytestmark = pytest.mark.asyncio

_TEMPLATE = "identity_sync.run_failure"


@pytest_asyncio.fixture
async def patched_session_factory(db_engine):
    """Repoint ``async_session_factory`` at the test engine so the
    notifier's own sessions hit the DB the fixture populates."""
    from app.core import database as _db

    original = _db.async_session_factory
    _db.async_session_factory = async_sessionmaker(
        db_engine, class_=AsyncSession, expire_on_commit=False,
    )
    try:
        yield
    finally:
        _db.async_session_factory = original


async def _person(db: AsyncSession, *, email: str) -> IdentityPerson:
    p = IdentityPerson(
        id=uuid.uuid4(),
        work_email=email,
        display_name=email.split("@", 1)[0],
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{email}",
    )
    db.add(p)
    await db.flush()
    return p


async def _admin(db: AsyncSession, *, email: str) -> IdentityPerson:
    from tests.conftest import _get_role_id  # type: ignore[attr-defined]

    p = await _person(db, email=email)
    role_id = await _get_role_id(db, "global_admin")
    db.add(AuthUserRole(user_id=p.id, role_id=role_id))
    await db.flush()
    return p


async def _target(
    db: AsyncSession,
    *,
    slug: str,
    alert_recipient_ids: list[str] | None,
    alert_enabled: bool = True,
) -> IdentitySyncTarget:
    t = IdentitySyncTarget(
        id=uuid.uuid4(),
        name=f"Alert {slug}",
        slug=slug,
        kind="scim",
        enabled=True,
        on_offboard="deactivate",
        alert_enabled=alert_enabled,
        mapping_config={"rules": [
            {"source_template": "{work_email}", "target_field": "userName"},
        ]},
        filter_config={
            "include_offboarded": False, "departments": [],
            "include_subtree": True, "countries": [],
            "employment_statuses": [], "job_title_includes": [],
            "email_suffixes": [], "person_ids": [], "mode": "and",
        },
        schedule_config={"trigger": "manual", "cron_expr": None, "timezone": "UTC"},
        runner_options={},
        alert_recipient_ids=alert_recipient_ids,
    )
    db.add(t)
    await db.flush()
    return t


async def _run(
    db: AsyncSession,
    target_id,
    *,
    status: str,
    error_message: str | None,
    started_at: datetime,
    failed_count: int = 1,
    trigger_source: str = "event:hire_confirm",
) -> IdentitySyncRun:
    r = IdentitySyncRun(
        id=uuid.uuid4(),
        target_id=target_id,
        started_at=started_at,
        finished_at=started_at,
        duration_ms=10,
        status=status,
        trigger_source=trigger_source,
        error_message=error_message,
        failed_count=failed_count,
        persons_in_scope=1,
    )
    db.add(r)
    await db.flush()
    return r


async def _outbox(db: AsyncSession, *, person_id) -> list[NotificationOutbox]:
    rows = (
        await db.execute(
            select(NotificationOutbox).where(
                NotificationOutbox.template_key == _TEMPLATE,
                NotificationOutbox.recipient_person_id == person_id,
            )
        )
    ).scalars().all()
    return list(rows)


_ERR_403 = (
    "HttpError: <HttpError 403 when requesting "
    "https://admin.googleapis.com/admin/directory/v1/users/x returned "
    "'Not Authorized to access this resource/api'>"
)


async def test_failed_run_notifies_configured_recipients(
    db_session: AsyncSession, patched_session_factory,
):
    db = db_session
    recipient = await _person(db, email="oncall@pingcap.com")
    bystander = await _person(db, email="nobody@pingcap.com")
    target = await _target(
        db, slug="alert-explicit", alert_recipient_ids=[str(recipient.id)],
    )
    run = await _run(
        db, target.id, status="failed", error_message=_ERR_403,
        started_at=datetime.now(UTC),
    )
    await db.commit()

    await maybe_notify_run_failure(run_id=run.id)

    async with patched_factory_session(db_session) as fresh:
        assert len(await _outbox(fresh, person_id=recipient.id)) == 1
        assert await _outbox(fresh, person_id=bystander.id) == []


async def test_empty_recipients_notifies_nobody(
    db_session: AsyncSession, patched_session_factory,
):
    """No recipients ⇒ nobody is paged. There is deliberately NO
    IT-admin fallback (2026-06-19: 不要默认通知管理员)."""
    db = db_session
    admin = await _admin(db, email="admin-nofallback@pingcap.com")
    target = await _target(
        db, slug="alert-nofallback", alert_recipient_ids=None,
    )
    run = await _run(
        db, target.id, status="failed", error_message=_ERR_403,
        started_at=datetime.now(UTC),
    )
    await db.commit()

    await maybe_notify_run_failure(run_id=run.id)

    async with patched_factory_session(db_session) as fresh:
        # The admin must NOT be paged — the old fallback is gone.
        assert await _outbox(fresh, person_id=admin.id) == []


async def test_switch_off_suppresses_alert(
    db_session: AsyncSession, patched_session_factory,
):
    """``alert_enabled=False`` mutes the target entirely — even a
    configured recipient gets nothing."""
    db = db_session
    recipient = await _person(db, email="oncall-muted@pingcap.com")
    target = await _target(
        db, slug="alert-muted", alert_recipient_ids=[str(recipient.id)],
        alert_enabled=False,
    )
    run = await _run(
        db, target.id, status="failed", error_message=_ERR_403,
        started_at=datetime.now(UTC),
    )
    await db.commit()

    await maybe_notify_run_failure(run_id=run.id)

    async with patched_factory_session(db_session) as fresh:
        assert await _outbox(fresh, person_id=recipient.id) == []


async def test_partial_run_also_alerts(
    db_session: AsyncSession, patched_session_factory,
):
    db = db_session
    recipient = await _person(db, email="oncall-partial@pingcap.com")
    target = await _target(
        db, slug="alert-partial", alert_recipient_ids=[str(recipient.id)],
    )
    run = await _run(
        db, target.id, status="partial", error_message="2 of 5 failed",
        started_at=datetime.now(UTC), failed_count=2,
    )
    await db.commit()

    await maybe_notify_run_failure(run_id=run.id)

    async with patched_factory_session(db_session) as fresh:
        assert len(await _outbox(fresh, person_id=recipient.id)) == 1


async def test_consecutive_same_signature_dedups(
    db_session: AsyncSession, patched_session_factory,
):
    db = db_session
    recipient = await _person(db, email="oncall-dedup@pingcap.com")
    target = await _target(
        db, slug="alert-dedup", alert_recipient_ids=[str(recipient.id)],
    )
    base = datetime.now(UTC)
    # Prior failure (same error → same signature), already on record.
    await _run(
        db, target.id, status="failed", error_message=_ERR_403,
        started_at=base - timedelta(minutes=5),
    )
    current = await _run(
        db, target.id, status="failed", error_message=_ERR_403,
        started_at=base,
    )
    await db.commit()

    await maybe_notify_run_failure(run_id=current.id)

    async with patched_factory_session(db_session) as fresh:
        # Prior same-signature failure suppresses the repeat push.
        assert await _outbox(fresh, person_id=recipient.id) == []


async def test_changed_signature_renotifies(
    db_session: AsyncSession, patched_session_factory,
):
    db = db_session
    recipient = await _person(db, email="oncall-change@pingcap.com")
    target = await _target(
        db, slug="alert-change", alert_recipient_ids=[str(recipient.id)],
    )
    base = datetime.now(UTC)
    await _run(
        db, target.id, status="failed",
        error_message="HttpError 401 invalid credentials",
        started_at=base - timedelta(minutes=5),
    )
    current = await _run(
        db, target.id, status="failed", error_message=_ERR_403,
        started_at=base,
    )
    await db.commit()

    await maybe_notify_run_failure(run_id=current.id)

    async with patched_factory_session(db_session) as fresh:
        # Different failure kind (401 → 403) is a fresh signal.
        assert len(await _outbox(fresh, person_id=recipient.id)) == 1


async def test_standing_partial_dedups_across_many_interleaved_event_runs(
    db_session: AsyncSession, patched_session_factory,
):
    """High-cadence regression: a standing nightly ``partial`` must dedup
    against the PREVIOUS night's partial even when far more than the old
    ``LIMIT 25`` clean single-person ``event`` runs interleave between them.

    The old count window lost the prior alerting sweep behind the event
    traffic → re-paged the same partial every night (storm). The time window
    finds it regardless of interleave count → suppressed.
    """
    db = db_session
    recipient = await _person(db, email="oncall-hicadence@pingcap.com")
    target = await _target(
        db, slug="alert-hicadence", alert_recipient_ids=[str(recipient.id)],
    )
    base = datetime.now(UTC)
    # Previous night's partial — same signature, ~23h ago (inside the window).
    await _run(
        db, target.id, status="partial", error_message=_ERR_403,
        started_at=base - timedelta(hours=23), trigger_source="scheduler",
    )
    # 30 clean single-person event pushes since — more than the old LIMIT 25.
    for i in range(30):
        await _run(
            db, target.id, status="ok", error_message=None,
            started_at=base - timedelta(hours=23) + timedelta(minutes=5 * (i + 1)),
            failed_count=0, trigger_source="event:hire_confirm",
        )
    current = await _run(
        db, target.id, status="partial", error_message=_ERR_403,
        started_at=base, trigger_source="scheduler",
    )
    await db.commit()

    await maybe_notify_run_failure(run_id=current.id)

    async with patched_factory_session(db_session) as fresh:
        # Standing same-signature partial → deduped despite the event flood.
        assert await _outbox(fresh, person_id=recipient.id) == []


async def test_recovery_via_full_sweep_then_refailure_renotifies(
    db_session: AsyncSession, patched_session_factory,
):
    """A successful FULL sweep between two same-signature failures is a
    genuine recovery — the re-failure is a new incident and must re-page,
    even inside the dedup window. (Clean ``event`` runs would NOT count as
    recovery; a full ``scheduler``/``manual`` ``ok`` does.)"""
    db = db_session
    recipient = await _person(db, email="oncall-recover@pingcap.com")
    target = await _target(
        db, slug="alert-recover", alert_recipient_ids=[str(recipient.id)],
    )
    base = datetime.now(UTC)
    await _run(
        db, target.id, status="failed", error_message=_ERR_403,
        started_at=base - timedelta(hours=2), trigger_source="scheduler",
    )
    # Full sweep went green — the standing failure cleared.
    await _run(
        db, target.id, status="ok", error_message=None,
        started_at=base - timedelta(hours=1), failed_count=0,
        trigger_source="scheduler",
    )
    current = await _run(
        db, target.id, status="failed", error_message=_ERR_403,
        started_at=base, trigger_source="scheduler",
    )
    await db.commit()

    await maybe_notify_run_failure(run_id=current.id)

    async with patched_factory_session(db_session) as fresh:
        assert len(await _outbox(fresh, person_id=recipient.id)) == 1


async def test_same_signature_outside_window_renotifies(
    db_session: AsyncSession, patched_session_factory,
):
    """Low-cadence regression: a same-signature failure whose last
    occurrence is older than the dedup window is a fresh regression and must
    re-page (the old ``LIMIT 25`` held ~25 days of daily-cron history and
    wrongly suppressed it)."""
    db = db_session
    recipient = await _person(db, email="oncall-stale@pingcap.com")
    target = await _target(
        db, slug="alert-stale", alert_recipient_ids=[str(recipient.id)],
    )
    base = datetime.now(UTC)
    # Last same-signature failure is well outside the 26h window.
    await _run(
        db, target.id, status="failed", error_message=_ERR_403,
        started_at=base - timedelta(hours=30), trigger_source="scheduler",
    )
    current = await _run(
        db, target.id, status="failed", error_message=_ERR_403,
        started_at=base, trigger_source="scheduler",
    )
    await db.commit()

    await maybe_notify_run_failure(run_id=current.id)

    async with patched_factory_session(db_session) as fresh:
        assert len(await _outbox(fresh, person_id=recipient.id)) == 1


def patched_factory_session(_db_session):
    """Open a fresh session on the patched (test-engine) factory to read
    post-commit state — reusing ``db_session`` after the notifier's
    foreign commit risks a stale-snapshot greenlet error."""
    from app.core import database as _db

    return _db.async_session_factory()