"""飞书后台待同步 —— the one hand-off step Helix structurally cannot do.

Feishu's approval v4 definition API is create / get / subscribe. There
is no update verb and no write for 流程管理员, so the moment an owner
presses 「转交」 the register and the console disagree, and only a human
in the console can end that disagreement.

What these tests pin down is not "we detected a mismatch" — that is one
comparison — but the three silences the predicate must NOT break, and
the stamp lifecycle that decides whether IT hears about a real backlog
once, forever, or never again.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest

from app.domains.approval_registry import constants as const
from app.domains.approval_registry import service as svc
from app.domains.approval_registry.serializers import (
    feishu_admin_drift,
    serialize_flow,
    serialize_published,
)
from app.identity.person_models import IdentityPerson


async def _person(db_session, email, *, name="P", status="active"):
    person = IdentityPerson(
        id=uuid.uuid4(), work_email=email, display_name=name,
        employment_status=status, source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:10]}",
    )
    db_session.add(person)
    await db_session.flush()
    await db_session.commit()
    return person


async def _flow(db_session, name, *, owner=None, status=const.STATUS_ACTIVE):
    flow = await svc.create_flow(
        db_session,
        approval_code=str(uuid.uuid4()).upper(),
        name=name,
        owner_person_id=owner.id if owner else None,
        status=status,
    )
    await db_session.commit()
    return flow


def _admins(*people, synced=True):
    """The shape the nightly sweep writes into ``feishu_admins``."""
    return [
        {
            "user_id": f"fs-{p.id.hex[:8]}",
            "name": p.display_name,
            "email": p.work_email,
            "person_id": str(p.id),
            "active": p.employment_status == "active",
        }
        for p in people
    ]


async def _set_admins(db_session, flow, admins, *, synced_at="now"):
    flow.feishu_admins = admins
    if synced_at == "now":
        flow.feishu_admins_synced_at = datetime.now(UTC).replace(tzinfo=None)
    else:
        flow.feishu_admins_synced_at = synced_at
    await db_session.commit()
    # Re-read through the domain's own loader: ``ApprovalFlow.owner`` is
    # ``lazy="raise"``, so a row handed back from ``create_flow`` cannot
    # be serialized. Reloading here rather than in each test keeps every
    # assertion working against the shape the API actually returns.
    return await svc.get_flow(db_session, flow.id)


# ── The predicate, and the three silences it must keep ──────────────


@pytest.mark.asyncio
async def test_transfer_leaves_the_console_pointing_at_the_predecessor(
    db_session,
):
    """The whole reason this exists: 「转交」 cannot touch Feishu.

    Nothing here mocks a failed API call, because there is no call to
    fail — the console keeps naming the previous owner and the platform
    has no verb that would change that.
    """
    predecessor = await _person(db_session, f"a-{uuid.uuid4().hex[:6]}@t.com")
    successor = await _person(db_session, f"b-{uuid.uuid4().hex[:6]}@t.com")
    flow = await _flow(db_session, "New Cloud Region Request", owner=predecessor)
    await _set_admins(db_session, flow, _admins(predecessor))
    assert feishu_admin_drift(flow) is False

    await svc.set_owner(
        db_session, flow, owner_person_id=successor.id, actor=predecessor,
        announce=const.HANDOVER_TRANSFER,
    )
    await db_session.commit()

    assert feishu_admin_drift(flow) is True
    drifting = await svc.flows_with_feishu_admin_drift(db_session)
    assert [f.id for f in drifting] == [flow.id]


@pytest.mark.asyncio
async def test_no_admin_list_is_not_drift(db_session):
    """An empty admin list is silence, and silence is never disagreement.

    Two states reach here and both must stay quiet. A row nobody has
    ever asked the console about (sweeps before migration 559 omitted
    ``with_admin_id``) has an empty list — reading that as drift would
    have lit up most of the register on the day this shipped, the exact
    mistake 559 was written to undo, in the opposite direction. And a
    definition Feishu genuinely reports with no 流程管理员 cannot
    disagree with us either.

    They are the same assertion because they are the same row: the
    sweep's only writer stamps ``feishu_admins`` and
    ``feishu_admins_synced_at`` together, so 「没问过」 and 「问了，没有」
    are indistinguishable in the data — which is why the predicate keys
    on the list and not on the timestamp.
    """
    owner = await _person(db_session, f"c-{uuid.uuid4().hex[:6]}@t.com")
    flow = await _flow(db_session, "从没问过管理员的", owner=owner)
    assert flow.feishu_admins_synced_at is None
    assert flow.feishu_admins is None
    assert feishu_admin_drift(flow) is False

    # Asked, and the console names nobody: the same silence.
    await _set_admins(db_session, flow, [])
    assert feishu_admin_drift(flow) is False
    assert await svc.flows_with_feishu_admin_drift(db_session) == []


@pytest.mark.asyncio
async def test_ownerless_flow_is_not_drift(db_session):
    """「没人认领」 is its own, louder read, and its fix is to name
    somebody — not to go edit the console."""
    admin = await _person(db_session, f"d-{uuid.uuid4().hex[:6]}@t.com")
    flow = await _flow(db_session, "没有负责人的")
    await _set_admins(db_session, flow, _admins(admin))
    assert flow.owner_person_id is None
    assert feishu_admin_drift(flow) is False


@pytest.mark.asyncio
async def test_admin_who_is_not_a_helix_person_cannot_match(db_session):
    """An admin known only by name (contractor, deleted account) resolves
    to no ``person_id``, so it can never equal an owner id.

    The flow still drifts — a name we cannot resolve is not evidence
    that the right person is listed — but the entry is excluded from the
    comparison rather than crashing it.
    """
    owner = await _person(db_session, f"e-{uuid.uuid4().hex[:6]}@t.com")
    flow = await _flow(db_session, "管理员是外部人的", owner=owner)
    await _set_admins(db_session, flow, [
        {"name": "某外包", "email": None, "person_id": None, "active": True},
    ])
    # No admin resolves to anybody, so there is nothing to disagree with.
    assert feishu_admin_drift(flow) is False

    # Add a resolvable admin who is somebody else: now it disagrees.
    other = await _person(db_session, f"f-{uuid.uuid4().hex[:6]}@t.com")
    await _set_admins(db_session, flow, [
        {"name": "某外包", "email": None, "person_id": None, "active": True},
        *_admins(other),
    ])
    assert feishu_admin_drift(flow) is True


@pytest.mark.asyncio
async def test_owner_among_several_admins_is_not_drift(db_session):
    """Seven of this tenant's flows name three or more admins. Being one
    of them is agreement — the console does not have to name ONLY the
    owner for the successor to hold the rights it grants."""
    owner = await _person(db_session, f"g-{uuid.uuid4().hex[:6]}@t.com")
    other1 = await _person(db_session, f"h-{uuid.uuid4().hex[:6]}@t.com")
    other2 = await _person(db_session, f"i-{uuid.uuid4().hex[:6]}@t.com")
    flow = await _flow(db_session, "三个管理员的", owner=owner)
    await _set_admins(db_session, flow, _admins(other1, owner, other2))
    assert feishu_admin_drift(flow) is False


@pytest.mark.asyncio
async def test_departed_admin_still_counts_as_the_console_answer(db_session):
    """A departed 流程管理员 is who the console still points at, so a
    flow whose owner is someone else drifts — and one still 「owned」 by
    the departed admin does not.

    ``active`` is published for the reader's judgement, never used to
    silently drop the entry: the console does not know the person left,
    and 「后台点的是一个已离职的人」 is a fact the drift row has to be
    able to state.
    """
    gone = await _person(
        db_session, f"j-{uuid.uuid4().hex[:6]}@t.com", status="left",
    )
    live = await _person(db_session, f"k-{uuid.uuid4().hex[:6]}@t.com")
    flow = await _flow(db_session, "管理员已离职的", owner=live)
    flow = await _set_admins(db_session, flow, _admins(gone))
    assert feishu_admin_drift(flow) is True
    assert serialize_flow(flow)["feishu_admins"][0]["active"] is False


@pytest.mark.asyncio
async def test_retired_flow_is_not_in_the_worklist(db_session):
    """A switched-off flow is not somebody's console chore.

    Asserted against a ``retired`` row that is NOT soft-deleted, which
    is the only version of this test that proves anything. Going through
    ``retire_flow`` stamps ``deleted_at`` as well, and the base query
    filters on that — so a retire-then-assert test passes just as
    happily with the status check deleted. ``service.retire_flow``'s own
    docstring names this trap ("the soft-delete filter alone would hide
    a missing status check"); a mutation run confirmed the softer
    version of this test could not see the filter disappear.
    """
    owner = await _person(db_session, f"l-{uuid.uuid4().hex[:6]}@t.com")
    admin = await _person(db_session, f"m-{uuid.uuid4().hex[:6]}@t.com")
    live = await _flow(db_session, "在用的", owner=owner)
    await _set_admins(db_session, live, _admins(admin))
    retired = await _flow(
        db_session, "已下线的", owner=owner, status=const.STATUS_RETIRED,
    )
    await _set_admins(db_session, retired, _admins(admin))

    drifting = await svc.flows_with_feishu_admin_drift(db_session)
    assert [f.id for f in drifting] == [live.id]

    # And the row-level predicate still says yes about it: the
    # disagreement is real, it just is not work.
    assert feishu_admin_drift(retired) is True


# ── The wire shapes ─────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_drift_is_admin_only_and_the_summary_agrees_with_the_list(
    db_session,
):
    """Two invariants in one test because they are the same bug twice.

    The KPI-vs-filter divergence this domain already shipped once (a
    tile counting ``owner_source='inferred'`` while its filter opened a
    shorter list) is prevented by both sides calling one predicate — so
    the count and the list are asserted equal, not merely both non-zero.

    And the all-staff payload must not carry it: telling every employee
    that the console names somebody else invites them to go ask that
    person, who is precisely who the flow no longer routes to.
    """
    owner = await _person(db_session, f"n-{uuid.uuid4().hex[:6]}@t.com")
    admin = await _person(db_session, f"o-{uuid.uuid4().hex[:6]}@t.com")
    drifting = await _flow(db_session, "对不上的", owner=owner)
    drifting = await _set_admins(db_session, drifting, _admins(admin))
    agreeing = await _flow(db_session, "对得上的", owner=owner)
    agreeing = await _set_admins(db_session, agreeing, _admins(owner))

    summary = await svc.coverage_summary(db_session)
    rows = await svc.flows_with_feishu_admin_drift(db_session)
    assert summary["feishu_admin_drift"] == len(rows) == 1

    assert serialize_flow(drifting)["feishu_admin_drift"] is True
    assert serialize_flow(agreeing)["feishu_admin_drift"] is False
    assert "feishu_admin_drift" not in serialize_published(drifting)


# ── The stamp lifecycle ─────────────────────────────────────────────


@pytest.mark.asyncio
async def test_healed_rows_give_their_stamp_back(db_session):
    """The clearing pass, and why it is not tidiness.

    A stamp is a statement about one specific disagreement. Left behind
    after the console was fixed, it would make the SAME flow breaking
    again next month invisible: the digest would skip it as 「already
    told them」 about a finding that had been resolved and re-created in
    between.
    """
    owner = await _person(db_session, f"p-{uuid.uuid4().hex[:6]}@t.com")
    admin = await _person(db_session, f"q-{uuid.uuid4().hex[:6]}@t.com")
    flow = await _flow(db_session, "被修好的", owner=owner)
    await _set_admins(db_session, flow, _admins(admin))

    stamped = datetime.now(UTC).replace(tzinfo=None) - timedelta(days=3)
    flow.feishu_admin_drift_notified_at = stamped
    await db_session.commit()

    # Still drifting → the stamp is left alone.
    drift_ids = {f.id for f in await svc.flows_with_feishu_admin_drift(db_session)}
    assert flow.id in drift_ids
    assert await svc.flows_with_stale_drift_stamp(db_session, drift_ids) == []

    # Somebody fixes the console; the nightly sweep re-reads it.
    await _set_admins(db_session, flow, _admins(owner))
    drift_ids = {f.id for f in await svc.flows_with_feishu_admin_drift(db_session)}
    assert flow.id not in drift_ids
    stale = await svc.flows_with_stale_drift_stamp(db_session, drift_ids)
    assert [f.id for f in stale] == [flow.id]


@pytest.mark.asyncio
async def test_renag_window_gates_the_digest(db_session):
    """Told-recently rows are not re-raised; told-long-ago ones are.

    Once-and-forget turns a real backlog into a card scrolled past in
    June; daily trains IT to mute the channel. The window is the whole
    behaviour, so it is asserted from both sides of the cutoff.
    """
    owner = await _person(db_session, f"r-{uuid.uuid4().hex[:6]}@t.com")
    admin = await _person(db_session, f"s-{uuid.uuid4().hex[:6]}@t.com")
    fresh = await _flow(db_session, "刚说过的", owner=owner)
    await _set_admins(db_session, fresh, _admins(admin))
    old = await _flow(db_session, "很久没说的", owner=owner)
    await _set_admins(db_session, old, _admins(admin))
    never = await _flow(db_session, "从没说过的", owner=owner)
    await _set_admins(db_session, never, _admins(admin))

    now = datetime.now(UTC).replace(tzinfo=None)
    fresh.feishu_admin_drift_notified_at = now - timedelta(days=1)
    old.feishu_admin_drift_notified_at = (
        now - timedelta(days=const.DRIFT_RENAG_DAYS + 1)
    )
    await db_session.commit()

    cutoff = now - timedelta(days=const.DRIFT_RENAG_DAYS)
    due = {
        f.id for f in await svc.flows_with_feishu_admin_drift(db_session)
        if f.feishu_admin_drift_notified_at is None
        or f.feishu_admin_drift_notified_at < cutoff
    }
    assert due == {old.id, never.id}


# ── The pushes: what goes out, and what deliberately does not ───────


@pytest.fixture
def sent_envelopes(monkeypatch):
    """Capture what the domain actually hands the notification service.

    Patched at ``NotificationService.push_inbox_and_feishu`` rather than
    at the domain's own notify functions, so the assertions see the real
    envelope — template key, recipient, context, buttons — that the
    domain built. A stub one level higher would let a card with the
    wrong template key or no recipients pass.
    """
    from app.notifications.service import NotificationService

    captured = []

    async def _capture(self, env, **kwargs):
        captured.append(env)
        return None

    monkeypatch.setattr(
        NotificationService, "push_inbox_and_feishu", _capture,
    )
    return captured


async def _operator(db_session, email):
    """An active ``global_admin`` — the only recipient set these pushes
    have (``notify._registry_maintainers``)."""
    from app.auth.models import AuthUserRole
    from tests.conftest import _get_role_id

    person = await _person(db_session, email, name="IT Operator")
    role_id = await _get_role_id(db_session, "global_admin")
    db_session.add(AuthUserRole(user_id=person.id, role_id=role_id))
    await db_session.flush()
    await db_session.commit()
    return person


@pytest.mark.asyncio
async def test_zero_drift_sends_nothing(db_session, sent_envelopes):
    """No disagreement, no card — not an 「一切正常」 card either.

    A daily job that reports its own success is a daily job people
    filter into a folder, and then the one morning it has something to
    say it is filtered too. Asserted rather than assumed: the digest's
    early return is the only thing standing between this cron and 365
    pushes a year saying nothing happened.
    """
    from app.scheduled_jobs.approval_registry_handlers import (
        run_feishu_admin_drift_digest,
    )

    await _operator(db_session, f"op-{uuid.uuid4().hex[:6]}@t.com")
    owner = await _person(db_session, f"t-{uuid.uuid4().hex[:6]}@t.com")
    agreeing = await _flow(db_session, "对得上的", owner=owner)
    await _set_admins(db_session, agreeing, _admins(owner))

    # Cleared after setup, not asserted from empty: naming an owner
    # sends that person their own 「你现在负责这条」 card, which is
    # correct and unrelated. What is under test is what the CRON sends.
    sent_envelopes.clear()
    result = await run_feishu_admin_drift_digest(db_session, {})
    assert result["drifting"] == 0
    assert result["raised"] == 0
    assert result["operators_notified"] == 0
    # The handler's own early return, distinct from the notifier's guard
    # above: it is what puts a readable reason in `last_result_summary`
    # so the cron console can tell 「跑了，两边一致」 from 「跑了，什么都
    # 没算出来」. Asserted because a mutation run showed the return could
    # be deleted with every other test still green.
    assert result["reason"] == "register and Feishu console agree"
    assert sent_envelopes == []


@pytest.mark.asyncio
async def test_the_notifier_itself_refuses_an_empty_list(
    db_session, sent_envelopes,
):
    """The 「0 不要发」 guard, pinned where it actually lives.

    Written after a mutation run: deleting the digest handler's
    ``if not due`` early return did NOT turn any test red, because the
    notifier's own ``if not flows`` guard caught the empty case — and
    deleting THAT one was equally invisible, because the handler
    returned first. Two guards each masking the other's removal is
    coverage that only looks like coverage, so the notifier is now
    exercised directly. Without a guard here the loop reaches
    ``_registry_maintainers`` and posts a card reading 「0 条流程的管理员
    和台账对不上」 to every operator, every morning.
    """
    from app.domains.approval_registry import notify

    await _operator(db_session, f"op-{uuid.uuid4().hex[:6]}@t.com")
    sent_envelopes.clear()
    assert await notify.notify_feishu_admin_drift(db_session, []) == 0
    assert sent_envelopes == []


@pytest.mark.asyncio
async def test_drift_with_nothing_new_also_sends_nothing(
    db_session, sent_envelopes,
):
    """Drift that IT was told about yesterday is not today's news.

    The count is non-zero here, which is exactly the case a naive
    ``if drifting: push`` would get wrong — the page keeps showing the
    backlog while the channel stays quiet until the 14-day re-nag.
    """
    from app.scheduled_jobs.approval_registry_handlers import (
        run_feishu_admin_drift_digest,
    )

    await _operator(db_session, f"op-{uuid.uuid4().hex[:6]}@t.com")
    owner = await _person(db_session, f"u-{uuid.uuid4().hex[:6]}@t.com")
    admin = await _person(db_session, f"v-{uuid.uuid4().hex[:6]}@t.com")
    flow = await _flow(db_session, "昨天已经说过的", owner=owner)
    flow = await _set_admins(db_session, flow, _admins(admin))
    flow.feishu_admin_drift_notified_at = (
        datetime.now(UTC).replace(tzinfo=None) - timedelta(days=1)
    )
    await db_session.commit()

    sent_envelopes.clear()
    result = await run_feishu_admin_drift_digest(db_session, {})
    assert result["drifting"] == 1
    assert result["raised"] == 0
    assert result["reason"] == "nothing new to raise"
    assert sent_envelopes == []


@pytest.mark.asyncio
async def test_transfer_alerts_it_immediately(db_session, sent_envelopes):
    """A person pressing 「转交」 reaches IT the same second.

    The digest was built first and on purpose (one offboarding
    hand-over moves every flow a leaver owned inside one transaction —
    per-event would be nine cards about one console screen). This push
    coexists with it because its seam is narrow: only
    ``announce=HANDOVER_TRANSFER``, which is passed from the two paths
    where a human pressed the button.

    Two cards are expected, and both matter: the successor's
    「你现在负责这条」 and IT's 「后台还没改」. Asserted by template key,
    because 「发了两张卡」 is also what sending the same card twice looks
    like.
    """
    operator = await _operator(db_session, f"op-{uuid.uuid4().hex[:6]}@t.com")
    predecessor = await _person(db_session, f"w-{uuid.uuid4().hex[:6]}@t.com")
    successor = await _person(db_session, f"x-{uuid.uuid4().hex[:6]}@t.com")
    flow = await _flow(db_session, "New Cloud Region Request", owner=predecessor)
    flow = await _set_admins(db_session, flow, _admins(predecessor))

    await svc.set_owner(
        db_session, flow, owner_person_id=successor.id, actor=predecessor,
        announce=const.HANDOVER_TRANSFER,
    )
    await db_session.commit()

    keys = [e.template_key for e in sent_envelopes]
    assert "approval_registry.transfer_console_gap" in keys
    gap = next(
        e for e in sent_envelopes
        if e.template_key == "approval_registry.transfer_console_gap"
    )
    assert gap.recipient.id == operator.id
    # Both names on the card: 「这条对不上」 alone would send the reader
    # back to compare the console by hand.
    assert gap.context["successor_name"] == successor.display_name
    assert predecessor.display_name in gap.context["console_admins"]
    assert gap.url == "/admin/approval-flows?tab=feishu-drift"


@pytest.mark.asyncio
async def test_immediate_alert_claims_the_digest_mention(
    db_session, sent_envelopes,
):
    """Told now, so not told again at 09:00 tomorrow.

    Without the stamp the two pushes stack: the alert fires on the
    transfer and the digest repeats the same flow the next morning,
    which teaches IT that the digest says things they have already
    handled. The row is still on the page — it is still drifting — it
    just is not re-announced.
    """
    from app.scheduled_jobs.approval_registry_handlers import (
        run_feishu_admin_drift_digest,
    )

    await _operator(db_session, f"op-{uuid.uuid4().hex[:6]}@t.com")
    predecessor = await _person(db_session, f"y-{uuid.uuid4().hex[:6]}@t.com")
    successor = await _person(db_session, f"z-{uuid.uuid4().hex[:6]}@t.com")
    flow = await _flow(db_session, "转交过的", owner=predecessor)
    flow = await _set_admins(db_session, flow, _admins(predecessor))
    await svc.set_owner(
        db_session, flow, owner_person_id=successor.id, actor=predecessor,
        announce=const.HANDOVER_TRANSFER,
    )
    await db_session.commit()
    assert flow.feishu_admin_drift_notified_at is not None

    sent_envelopes.clear()
    result = await run_feishu_admin_drift_digest(db_session, {})
    assert result["drifting"] == 1, "still on the page"
    assert result["raised"] == 0, "but not announced twice"
    assert sent_envelopes == []


@pytest.mark.asyncio
async def test_transfer_to_a_console_admin_alerts_nobody(
    db_session, sent_envelopes,
):
    """No gap, no alert.

    A flow whose console entry already names the successor needs no
    console edit. Pushing 「有人转交了」 about it would be an alert with
    nothing behind it, which is the fastest way to get the real ones
    muted.
    """
    await _operator(db_session, f"op-{uuid.uuid4().hex[:6]}@t.com")
    predecessor = await _person(db_session, f"aa-{uuid.uuid4().hex[:6]}@t.com")
    successor = await _person(db_session, f"ab-{uuid.uuid4().hex[:6]}@t.com")
    flow = await _flow(db_session, "两人都是管理员的", owner=predecessor)
    flow = await _set_admins(db_session, flow, _admins(predecessor, successor))

    await svc.set_owner(
        db_session, flow, owner_person_id=successor.id, actor=predecessor,
        announce=const.HANDOVER_TRANSFER,
    )
    await db_session.commit()

    keys = [e.template_key for e in sent_envelopes]
    assert "approval_registry.transfer_console_gap" not in keys
    assert flow.feishu_admin_drift_notified_at is None


@pytest.mark.asyncio
async def test_offboarding_handover_stays_in_the_digest(
    db_session, sent_envelopes,
):
    """The storm guard, asserted rather than trusted.

    An offboarding hand-over moves every flow a leaver owned in one
    transaction. It reaches ``set_owner`` with ``announce=None`` (the
    coordinator sends its own leaver-worded card), so no per-flow IT
    alert fires and the whole batch surfaces as one digest line
    tomorrow. If a future refactor made the immediate alert fire on any
    ownership change, this is the test that goes red.
    """
    leaver = await _person(db_session, f"ac-{uuid.uuid4().hex[:6]}@t.com")
    successor = await _person(db_session, f"ad-{uuid.uuid4().hex[:6]}@t.com")
    await _operator(db_session, f"op-{uuid.uuid4().hex[:6]}@t.com")
    flows = []
    for i in range(3):
        f = await _flow(db_session, f"离职者名下 {i}", owner=leaver)
        flows.append(await _set_admins(db_session, f, _admins(leaver)))

    for f in flows:
        await svc.set_owner(
            db_session, f, owner_person_id=successor.id, actor=leaver,
            announce=None,
        )
    await db_session.commit()

    keys = [e.template_key for e in sent_envelopes]
    assert "approval_registry.transfer_console_gap" not in keys
    assert all(f.feishu_admin_drift_notified_at is None for f in flows)
    # All three ARE drifting — they are just one digest line, not three
    # cards.
    assert len(await svc.flows_with_feishu_admin_drift(db_session)) == 3