"""回执:审批过了,卡片和铃铛都得知道。

用户的原话:「helix 的推送没有任何回执的能力,这个审批过了,还是去审批」。

这个文件钉住的是四条不变量,每一条都对应一种「看起来能用、实际在骗人」的写法:

1. **记录先落地,卡片后重画。** 反过来会得到最坏的结果:飞书卡片变成绿色
   「已批准」,而铃铛里那条还写着「去审批」—— 两个面对着同一件事说不同的话。
   所以卡片改失败时,结案记录必须**仍然**是对的。
2. **永不抛异常。** 调用点是领域流程的终态迁移(批准/驳回/关单),权威状态已经
   写完了;一次回执失败绝不能回滚一次真实的批准。
3. **幂等。** 只结 `settled_at IS NULL` 的行 —— 对账 cron 会对同一件事反复调用。
4. **红点跟着变。** `read_at`(我看过了)和 `settled_at`(那件事完了)是两件事:
   一条从没被点开的通知,它说的事情完全可能已经批准。
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.identity.person_models import IdentityPerson
from app.notifications import card_design, settlement
from app.notifications.inbox_service import InboxService
from app.notifications.models import NotificationOutbox
from tests.support.card_shape import (
    assert_card_schema,
    button_labels,
    header_badges,
    markdown_of,
)

RESOURCE = "cmdb_cloud_grant_renewal_request"


async def _person(db: AsyncSession, *, name: str, lang: str = "zh") -> IdentityPerson:
    person = IdentityPerson(
        id=uuid.uuid4(),
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
        display_name=name,
        work_email=f"s-{uuid.uuid4().hex[:8]}@pingcap.com",
        employment_status="active",
        preferred_language=lang,
    )
    db.add(person)
    await db.flush()
    return person


async def _row(
    db: AsyncSession,
    *,
    recipient: IdentityPerson,
    resource_id: uuid.UUID,
    message_id: str | None = "om_test",
    template_key: str | None = "approval.pending",
) -> NotificationOutbox:
    row = NotificationOutbox(
        recipient_person_id=recipient.id,
        recipient_email=recipient.work_email,
        template_key=template_key,
        title="施宏伟 申请续期 PingCAP SCC 的 1 项权限",
        body_text="请审批。",
        url="/cloud/scopes/abc?renewals=1",
        channels=["feishu", "email"],
        status="delivered",
        delivered_channels=["feishu", "email"],
        feishu_message_id=message_id,
        related_resource_type=RESOURCE,
        related_resource_id=str(resource_id),
    )
    db.add(row)
    await db.flush()
    return row


class _FakeClient:
    """记下每一次卡片改写;可按剧本失败。"""

    def __init__(self, *, fail: bool = False) -> None:
        self.fail = fail
        self.patched: list[tuple[str, dict]] = []

    async def update_card_message(self, *, message_id: str, card: dict) -> None:
        if self.fail:
            raise RuntimeError("feishu says no")
        self.patched.append((message_id, card))


@pytest.fixture
def fake_feishu(monkeypatch):
    """把飞书这一侧整体换成替身。

    换的是 ``settlement`` 真正用到的那三样:集成配置、客户端、图标解析。**不**换
    `settle` 自己 —— 被测的就是它。
    """
    client = _FakeClient()

    async def _cfg(**_kw):
        return {"app_id": "cli_test", "app_secret": "s"}

    monkeypatch.setattr(settlement, "get_feishu_config", _cfg)
    import app.approval.feishu_client as fc
    monkeypatch.setattr(fc, "FeishuApprovalClient", lambda **_kw: client)
    from app.notifications import card_icon

    async def _icon(*_a, **_kw):
        return None

    monkeypatch.setattr(card_icon, "resolve_icon_img_key", _icon)
    return client


# ── 一、记录先落地,卡片后重画 ──────────────────────────────────

@pytest.mark.asyncio
async def test_settling_stamps_the_row_and_repaints_the_card(
    db_session: AsyncSession, fake_feishu: _FakeClient,
) -> None:
    approver = await _person(db_session, name="倪捷")
    actor = await _person(db_session, name="冯帅")
    rid = uuid.uuid4()
    row = await _row(db_session, recipient=approver, resource_id=rid)

    count = await settlement.settle(
        db_session,
        resource_type=RESOURCE,
        resource_id=rid,
        outcome=card_design.STATUS_APPROVED,
        actor_person_id=actor.id,
    )

    assert count == 1
    await db_session.refresh(row)
    assert row.settled_at is not None
    assert row.settled_outcome == card_design.STATUS_APPROVED
    assert str(row.settled_by_person_id) == str(actor.id)
    assert len(fake_feishu.patched) == 1
    message_id, card = fake_feishu.patched[0]
    assert message_id == "om_test"
    assert_card_schema(card)


@pytest.mark.asyncio
async def test_the_record_survives_a_card_that_cannot_be_repainted(
    db_session: AsyncSession, fake_feishu: _FakeClient,
) -> None:
    """顺序不变量。卡片改不了(消息太老/网络抖)时,结案记录必须**仍然**落地
    —— 否则铃铛继续把它当待办,而员工手上那张卡也没变,两边一起错。"""
    fake_feishu.fail = True
    approver = await _person(db_session, name="倪捷")
    rid = uuid.uuid4()
    row = await _row(db_session, recipient=approver, resource_id=rid)

    count = await settlement.settle(
        db_session, resource_type=RESOURCE, resource_id=rid,
        outcome=card_design.STATUS_REJECTED,
    )

    assert count == 1, "卡片失败把结案也一起丢了"
    await db_session.refresh(row)
    assert row.settled_outcome == card_design.STATUS_REJECTED


@pytest.mark.asyncio
async def test_a_row_with_no_card_is_still_settled(
    db_session: AsyncSession, fake_feishu: _FakeClient,
) -> None:
    """只走了邮件的行没有 message_id —— 它照样要结案,因为铃铛读的是同一张表。"""
    approver = await _person(db_session, name="倪捷")
    rid = uuid.uuid4()
    row = await _row(db_session, recipient=approver, resource_id=rid, message_id=None)

    assert await settlement.settle(
        db_session, resource_type=RESOURCE, resource_id=rid,
        outcome=card_design.STATUS_APPROVED,
    ) == 1
    await db_session.refresh(row)
    assert row.settled_at is not None
    assert fake_feishu.patched == []


# ── 二、幂等 ─────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_settling_twice_changes_nothing_the_second_time(
    db_session: AsyncSession, fake_feishu: _FakeClient,
) -> None:
    """对账 cron 会对同一件事反复调用,只有第一次该做实事。"""
    approver = await _person(db_session, name="倪捷")
    first = await _person(db_session, name="第一个处理人")
    second = await _person(db_session, name="后来那个人")
    rid = uuid.uuid4()
    row = await _row(db_session, recipient=approver, resource_id=rid)

    assert await settlement.settle(
        db_session, resource_type=RESOURCE, resource_id=rid,
        outcome=card_design.STATUS_APPROVED, actor_person_id=first.id,
    ) == 1
    assert await settlement.settle(
        db_session, resource_type=RESOURCE, resource_id=rid,
        outcome=card_design.STATUS_REJECTED, actor_person_id=second.id,
    ) == 0

    await db_session.refresh(row)
    assert row.settled_outcome == card_design.STATUS_APPROVED, "结果被改写了"
    assert str(row.settled_by_person_id) == str(first.id), "处理人被换成后来那个人"
    assert len(fake_feishu.patched) == 1, "卡片被改了第二次"


@pytest.mark.asyncio
async def test_an_unrelated_resource_is_untouched(
    db_session: AsyncSession, fake_feishu: _FakeClient,
) -> None:
    approver = await _person(db_session, name="倪捷")
    mine, theirs = uuid.uuid4(), uuid.uuid4()
    other = await _row(db_session, recipient=approver, resource_id=theirs)
    await _row(db_session, recipient=approver, resource_id=mine)

    await settlement.settle(
        db_session, resource_type=RESOURCE, resource_id=mine,
        outcome=card_design.STATUS_APPROVED,
    )
    await db_session.refresh(other)
    assert other.settled_at is None


# ── 三、永不抛异常,且拒绝非终态 ────────────────────────────────

@pytest.mark.asyncio
async def test_a_non_terminal_outcome_is_refused_not_applied(
    db_session: AsyncSession, fake_feishu: _FakeClient,
) -> None:
    """`pending` 不是结案。一个拼错的值会把卡片改成中性态并吃掉按钮,比不改更糟。"""
    approver = await _person(db_session, name="倪捷")
    rid = uuid.uuid4()
    row = await _row(db_session, recipient=approver, resource_id=rid)

    assert await settlement.settle(
        db_session, resource_type=RESOURCE, resource_id=rid,
        outcome=card_design.STATUS_PENDING,
    ) == 0
    await db_session.refresh(row)
    assert row.settled_at is None
    assert fake_feishu.patched == []


@pytest.mark.asyncio
async def test_nothing_to_settle_is_not_an_error(
    db_session: AsyncSession, fake_feishu: _FakeClient,
) -> None:
    assert await settlement.settle(
        db_session, resource_type=RESOURCE, resource_id=uuid.uuid4(),
        outcome=card_design.STATUS_APPROVED,
    ) == 0


@pytest.mark.asyncio
async def test_a_broken_feishu_integration_does_not_raise(
    db_session: AsyncSession, monkeypatch,
) -> None:
    """集成没配 / 读配置就炸 —— 结案照做,卡片留在原样。"""
    async def _boom(**_kw):
        raise RuntimeError("integration table unreadable")

    monkeypatch.setattr(settlement, "get_feishu_config", _boom)
    approver = await _person(db_session, name="倪捷")
    rid = uuid.uuid4()
    row = await _row(db_session, recipient=approver, resource_id=rid)

    assert await settlement.settle(
        db_session, resource_type=RESOURCE, resource_id=rid,
        outcome=card_design.STATUS_APPROVED,
    ) == 1
    await db_session.refresh(row)
    assert row.settled_at is not None


# ── 四、回执卡长什么样 ──────────────────────────────────────────

@pytest.mark.asyncio
async def test_the_receipt_card_drops_the_action_and_says_who_did_it(
    db_session: AsyncSession, fake_feishu: _FakeClient,
) -> None:
    """原来的「去审批」必须消失 —— 它邀请的是一次会被拒绝的点击。"""
    approver = await _person(db_session, name="倪捷")
    actor = await _person(db_session, name="冯帅")
    rid = uuid.uuid4()
    await _row(db_session, recipient=approver, resource_id=rid)

    await settlement.settle(
        db_session, resource_type=RESOURCE, resource_id=rid,
        outcome=card_design.STATUS_APPROVED, actor_person_id=actor.id,
        note={"zh": "续期至 12-06。", "en": "Renewed to 12-06.", "ja": "12-06 まで更新。"},
    )
    _mid, card = fake_feishu.patched[0]
    assert card["header"]["template"] == "green"
    assert header_badges(card) == ["已批准"]
    body = markdown_of(card)
    assert "冯帅" in body
    assert "续期至 12-06。" in body
    assert "去审批" not in str(card)
    assert button_labels(card) == ["查看详情"]


@pytest.mark.asyncio
async def test_the_receipt_speaks_the_readers_language(
    db_session: AsyncSession, fake_feishu: _FakeClient,
) -> None:
    """一件事的卡片会发给不共享语言的多个人,「已批准」要用读者自己的语言说。"""
    zh = await _person(db_session, name="倪捷", lang="zh")
    en = await _person(db_session, name="Alice", lang="en_US")
    rid = uuid.uuid4()
    await _row(db_session, recipient=zh, resource_id=rid, message_id="om_zh")
    await _row(db_session, recipient=en, resource_id=rid, message_id="om_en")

    await settlement.settle(
        db_session, resource_type=RESOURCE, resource_id=rid,
        outcome=card_design.STATUS_APPROVED,
        note={"zh": "续期至 12-06。", "en": "Renewed to 12-06.", "ja": "12-06 まで。"},
    )
    by_id = dict(fake_feishu.patched)
    assert header_badges(by_id["om_zh"]) == ["已批准"]
    assert header_badges(by_id["om_en"]) == ["Approved"]
    assert "Renewed to 12-06." in markdown_of(by_id["om_en"])
    assert button_labels(by_id["om_en"]) == ["View details"]


@pytest.mark.asyncio
async def test_an_expired_receipt_names_no_actor(
    db_session: AsyncSession, fake_feishu: _FakeClient,
) -> None:
    """超期关单没有处理人 —— 不能渲染成空白,也不能假装是谁做的。"""
    approver = await _person(db_session, name="倪捷")
    rid = uuid.uuid4()
    await _row(db_session, recipient=approver, resource_id=rid)

    await settlement.settle(
        db_session, resource_type=RESOURCE, resource_id=rid,
        outcome=card_design.STATUS_EXPIRED,
    )
    _mid, card = fake_feishu.patched[0]
    assert header_badges(card) == ["已过期"]
    assert "系统自动" in markdown_of(card)


def test_the_headline_is_written_for_each_language() -> None:
    when = datetime(2026, 9, 5, 15, 20, tzinfo=UTC)
    zh = settlement.receipt_headline(
        outcome=card_design.STATUS_APPROVED, actor_name="张三", when=when, lang="zh",
    )
    en = settlement.receipt_headline(
        outcome=card_design.STATUS_APPROVED, actor_name="张三", when=when, lang="en",
    )
    ja = settlement.receipt_headline(
        outcome=card_design.STATUS_APPROVED, actor_name="张三", when=when, lang="ja",
    )
    assert zh == "由 张三 于 09-05 15:20 已批准。"
    assert en == "Approved by 张三 at 09-05 15:20."
    assert "承認されました" in ja
    # 三句都不是把另一种语言的语序硬套过来的
    assert len({zh, en, ja}) == 3


# ── 五、红点 ─────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_a_settled_row_stops_counting_toward_the_red_dot(
    db_session: AsyncSession, fake_feishu: _FakeClient,
) -> None:
    """`read_at` 和 `settled_at` 是两件事:一条从没被点开的通知,它说的事情
    完全可能已经批准 —— 那时它不该再让红点亮着。"""
    approver = await _person(db_session, name="倪捷")
    rid = uuid.uuid4()
    row = await _row(db_session, recipient=approver, resource_id=rid)
    inbox = InboxService(db_session)

    before = await inbox.summary_for_recipient(recipient_id=approver.id)
    assert before["unread_actionable"] == 1

    await settlement.settle(
        db_session, resource_type=RESOURCE, resource_id=rid,
        outcome=card_design.STATUS_APPROVED,
    )
    after = await inbox.summary_for_recipient(recipient_id=approver.id)
    assert after["unread_actionable"] == 0
    # 分组视图必须和红点用**同一个**判据 —— 两处不一致的后果是用户可见的:
    # 红点说 0,而「待处理」那一组仍写着 1,同一个面板自己跟自己矛盾。
    # (复审发现:第一版只在 unread_actionable 上排除了已结案行。)
    assert after["unread_by_bucket"]["action"] == 0, (
        "分组视图仍把已结案的行算作待处理 —— 和红点矛盾"
    )
    # 但它仍然是一条**未读**通知 —— 员工确实没看过,列表头和「全部标为已读」
    # 该继续看见它。
    assert after["unread"] == 1
    await db_session.refresh(row)
    assert row.read_at is None


@pytest.mark.asyncio
async def test_the_inbox_row_carries_the_outcome_to_the_client(
    db_session: AsyncSession, fake_feishu: _FakeClient,
) -> None:
    approver = await _person(db_session, name="倪捷")
    rid = uuid.uuid4()
    await _row(db_session, recipient=approver, resource_id=rid)
    await settlement.settle(
        db_session, resource_type=RESOURCE, resource_id=rid,
        outcome=card_design.STATUS_APPROVED,
    )

    listed = await InboxService(db_session).list_for_recipient(
        recipient_id=approver.id, page=1, page_size=10,
    )
    row_payload = next(
        r for r in listed["items"] if r["related_resource_id"] == str(rid)
    )
    assert row_payload["settled_outcome"] == card_design.STATUS_APPROVED
    assert row_payload["settled_at"] is not None


@pytest.mark.asyncio
async def test_settlement_is_visible_without_the_row_being_read(
    db_session: AsyncSession, fake_feishu: _FakeClient,
) -> None:
    """回执不该顺手把行标成已读 —— 那会掩盖「员工其实从没看过这条」。"""
    approver = await _person(db_session, name="倪捷")
    rid = uuid.uuid4()
    row = await _row(db_session, recipient=approver, resource_id=rid)
    await settlement.settle(
        db_session, resource_type=RESOURCE, resource_id=rid,
        outcome=card_design.STATUS_DONE,
    )
    await db_session.refresh(row)
    assert row.read_at is None
    assert row.settled_at is not None