"""Assistant channel: event parsing, release gate, identity and dedup.

These are the invariants that decide whether the channel is safe to point at
500 employees, so each one is asserted on behaviour rather than on structure:
the gate actually keeps the bot silent, an unknown sender actually gets
guidance instead of someone else's data, and a Feishu redelivery actually
produces one reply rather than two.
"""
from __future__ import annotations

from datetime import UTC, datetime

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.assistant.feishu_events import handle_message_event
from app.assistant.settings import SETTING_TURNS_PER_HOUR
from app.audit.models import AuditLog
from app.identity.person_models import IdentityExternalIdentity
from app.integrations.models import PlatformSetting
from tests.test_assistant.conftest import (
    OPEN_ID,
    UNION_ID,
    make_feishu_person,
    message_event,
    set_gate,
)

pytestmark = pytest.mark.asyncio


# ── release gate ────────────────────────────────────────────────────


async def test_gate_closed_stays_completely_silent(
    db_session: AsyncSession, feishu_person, sent,
):
    outcome = await handle_message_event(
        db_session, message_event(), trace_id="t-gate",
    )
    assert outcome == "disabled"
    assert sent == []


async def test_gate_requires_exact_true(
    db_session: AsyncSession, feishu_person, sent,
):
    """A JSON string "true" must not release the product."""
    await set_gate(db_session, "true")
    outcome = await handle_message_event(
        db_session, message_event(), trace_id="t-gate-str",
    )
    assert outcome == "disabled"
    assert sent == []


# ── sender identity ─────────────────────────────────────────────────


async def test_known_sender_gets_answered(
    db_session: AsyncSession, enabled_gate, feishu_person, sent, kernel,
):
    outcome = await handle_message_event(
        db_session, message_event(), trace_id="t-ok",
    )
    assert outcome == "answered"
    assert len(sent) == 1
    assert sent[0]["open_id"] == OPEN_ID
    assert sent[0]["text"] == "你的电脑保修到 2027-03-01。"


async def test_unknown_sender_gets_binding_guidance_not_silence(
    db_session: AsyncSession, enabled_gate, sent,
):
    outcome = await handle_message_event(
        db_session,
        message_event(union_id="on_nobody", open_id="ou_nobody"),
        trace_id="t-unbound",
    )
    assert outcome == "unbound"
    assert len(sent) == 1
    assert "helix.pingcap.net" in sent[0]["text"]


async def test_contact_sync_identity_is_not_a_login_identity(
    db_session: AsyncSession, enabled_gate, sent,
):
    """A directory-sync row must not identify a chat sender."""
    await make_feishu_person(
        db_session,
        email="contact.only@test.com",
        name="Contact Only",
        is_login_identity=False,
    )

    outcome = await handle_message_event(
        db_session, message_event(), trace_id="t-contact",
    )
    assert outcome == "unbound"


async def test_soft_deleted_identity_is_not_accepted(
    db_session: AsyncSession, enabled_gate, feishu_person, sent,
):
    row = (
        await db_session.execute(
            select(IdentityExternalIdentity).where(
                IdentityExternalIdentity.provider_subject == UNION_ID,
            ),
        )
    ).scalars().one()
    row.deleted_at = datetime.now(UTC)
    await db_session.commit()

    outcome = await handle_message_event(
        db_session, message_event(), trace_id="t-deleted",
    )
    assert outcome == "unbound"


async def test_departed_employee_is_refused(
    db_session: AsyncSession, enabled_gate, sent,
):
    await make_feishu_person(
        db_session, email="gone@test.com", employment_status="left",
    )

    outcome = await handle_message_event(
        db_session, message_event(), trace_id="t-left",
    )
    assert outcome == "departed"
    assert len(sent) == 1


async def test_leaver_pending_keeps_read_access(
    db_session: AsyncSession, enabled_gate, sent, kernel,
):
    """Someone working out their notice needs their own state most of all."""
    await make_feishu_person(
        db_session, email="notice@test.com", employment_status="leaver_pending",
    )

    outcome = await handle_message_event(
        db_session, message_event(), trace_id="t-pending",
    )
    assert outcome == "answered"


# ── redelivery, unsupported input, budget ───────────────────────────


async def test_feishu_redelivery_answers_once(
    db_session: AsyncSession, enabled_gate, feishu_person, sent, kernel,
):
    payload = message_event(event_id="evt-dup")
    first = await handle_message_event(db_session, payload, trace_id="t-1")
    second = await handle_message_event(db_session, payload, trace_id="t-2")
    assert first == "answered"
    assert second == "duplicate"
    assert len(sent) == 1


async def test_unsupported_message_gets_a_next_step(
    db_session: AsyncSession, enabled_gate, feishu_person, sent,
):
    outcome = await handle_message_event(
        db_session,
        message_event(message_type="image", content="{}", event_id="evt-img"),
        trace_id="t-img",
    )
    assert outcome == "unsupported"
    assert len(sent) == 1


async def test_a_rate_limited_sender_is_told_once_not_left_in_silence(
    db_session: AsyncSession, enabled_gate, feishu_person, sent, kernel,
):
    """Silence reads as "the bot is broken" and misdirects support."""
    db_session.add(PlatformSetting(key=SETTING_TURNS_PER_HOUR, value=5))
    await db_session.commit()

    for i in range(8):
        await handle_message_event(
            db_session, message_event(event_id=f"evt-n-{i}"), trace_id=f"t-{i}",
        )

    notices = [m for m in sent if "IT" in m["text"] or "IT 求助" in m["text"]]
    assert len(notices) == 1, "told exactly once per window"


async def test_sender_ids_that_disagree_are_refused(
    db_session: AsyncSession, enabled_gate, feishu_person, sent, kernel,
):
    """The answer is computed for union_id but delivered to open_id.

    If they name different people, the reply would carry one employee's data to
    another's inbox. A disagreement is refused rather than resolved by
    precedence.
    """
    await make_feishu_person(
        db_session, subject="ou_attacker_open", email="attacker@test.com",
    )

    outcome = await handle_message_event(
        db_session,
        message_event(union_id=UNION_ID, open_id="ou_attacker_open"),
        trace_id="t-mismatch",
    )

    assert outcome == "mismatch"
    assert kernel["calls"] == []


async def test_hourly_budget_stops_a_runaway_sender(
    db_session: AsyncSession, enabled_gate, feishu_person, sent, kernel,
):
    db_session.add(PlatformSetting(key=SETTING_TURNS_PER_HOUR, value=5))
    await db_session.commit()

    outcomes = [
        await handle_message_event(
            db_session, message_event(event_id=f"evt-rate-{i}"), trace_id=f"t-{i}",
        )
        for i in range(7)
    ]
    assert outcomes[:5] == ["answered"] * 5
    assert outcomes[5:] == ["rate_limited", "rate_limited"]
    # Five answers plus ONE notice — the ceiling is announced, not repeated.
    assert len(sent) == 6


# ── audit ───────────────────────────────────────────────────────────


async def test_every_handled_message_is_audited(
    db_session: AsyncSession, enabled_gate, feishu_person, sent, kernel,
):
    await handle_message_event(
        db_session, message_event(event_id="evt-audit"), trace_id="t-audit",
    )
    rows = (
        await db_session.execute(
            select(AuditLog).where(AuditLog.action == "assistant.message.received"),
        )
    ).scalars().all()
    assert len(rows) == 1
    assert rows[0].actor_user_id == feishu_person.id
    assert rows[0].detail["outcome"] == "answered"
    assert rows[0].detail["channel"] == "feishu"
    # Employee free text must not accumulate in the audit spine.
    assert "text" not in rows[0].detail