"""Notification governance (PR1): per-category × channel muting + quiet hours,
inbox bucketing, and the preference centre endpoints.

Covers the send-time gate's decision logic (``resolve_gating``), the inbox
bucket classification + filter, the per-bucket unread summary, and the
self-service GET/PUT contract incl. sanitisation.
"""
from __future__ import annotations

from datetime import UTC, datetime

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.identity.person_models import IdentityPerson
from app.notifications.categories import (
    BUCKET_ACTION,
    BUCKET_INFO,
    BUCKET_SOCIAL,
    classify,
)
from app.notifications.models import STATUS_DELIVERED, NotificationOutbox
from app.notifications.preferences_service import (
    NotificationPreferenceService,
    quiet_window_park_until,
)

# ── pure classification (no DB) ─────────────────────────────────────

def test_classify_known_and_fallback() -> None:
    assert classify("cloud.scope_request").key == "cloud"
    assert classify("cloud.scope_request").bucket == BUCKET_ACTION
    assert classify("feed.like.summary").bucket == BUCKET_SOCIAL
    assert classify("hire_day_it_info").bucket == BUCKET_INFO  # lifecycle
    # IT-ops escalations are non-mutable + not employee-facing.
    sysc = classify("scheduled_job.failure")
    assert sysc.key == "system"
    assert sysc.mutable is False
    assert sysc.employee_facing is False
    # Unknown key falls through to a mutable action item (fail-towards-visible).
    other = classify("brand_new_thing_2099")
    assert other.key == "other"
    assert other.bucket == BUCKET_ACTION
    assert other.mutable is True
    # Precedence: a cloud escalation lands in system, not the cloud category.
    assert classify("cloud.grant.manual_revoke_escalated").key == "system"
    # Leaver-SLA escalations use UNDERSCORE keys — they must classify as the
    # non-mutable system category (not the mutable 'other' fallback).
    warn = classify("leaver_blocked_sla_warn")
    assert warn.key == "system"
    assert warn.mutable is False


def test_quiet_window_edges() -> None:
    from zoneinfo import ZoneInfo

    tz = ZoneInfo("Asia/Shanghai")
    # 23:00 CST is inside 22:00→08:00 (wraps midnight) → park to next 08:00.
    now = datetime(2026, 6, 17, 15, 0, tzinfo=UTC)  # 23:00 CST
    assert quiet_window_park_until(now, tz, "22:00", "08:00") is not None
    # noon is outside.
    noon = datetime(2026, 6, 17, 4, 0, tzinfo=UTC)  # 12:00 CST
    assert quiet_window_park_until(noon, tz, "22:00", "08:00") is None
    # zero-length window is disabled.
    assert quiet_window_park_until(now, tz, "09:00", "09:00") is None


# ── send-time gate ──────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_resolve_gating_mutes_only_requested_category(
    db_session: AsyncSession, employee_user,
) -> None:
    user, _ = employee_user
    person = await db_session.get(IdentityPerson, user.id)
    svc = NotificationPreferenceService(db_session)
    await svc.upsert(
        person_id=user.id,
        muted_channels={"cloud": ["feishu"]},
        quiet_hours_enabled=False,
        quiet_hours_start=None,
        quiet_hours_end=None,
    )
    await db_session.commit()

    # cloud.* with feishu muted → feishu dropped, email kept.
    muted, park = await svc.resolve_gating(
        recipient=person,
        template_key="cloud.owner_zero_spend",
        requested_push_channels=["feishu", "email"],
    )
    assert muted == ["feishu"]
    assert park is None

    # A different category is untouched.
    muted2, _ = await svc.resolve_gating(
        recipient=person,
        template_key="asset.borrow_overdue",
        requested_push_channels=["feishu", "email"],
    )
    assert muted2 == []


@pytest.mark.asyncio
async def test_resolve_gating_never_mutes_non_mutable_category(
    db_session: AsyncSession, employee_user,
) -> None:
    """Even if a stale client crams a non-mutable category into the map, the
    gate refuses to mute it (and upsert drops it on the way in)."""
    user, _ = employee_user
    person = await db_session.get(IdentityPerson, user.id)
    svc = NotificationPreferenceService(db_session)
    row = await svc.upsert(
        person_id=user.id,
        muted_channels={"system": ["feishu"], "security": ["email"]},
        quiet_hours_enabled=False,
        quiet_hours_start=None,
        quiet_hours_end=None,
    )
    await db_session.commit()
    # Non-mutable categories were dropped by the sanitiser.
    assert "system" not in row.muted_channels
    assert "security" not in row.muted_channels
    muted, _ = await svc.resolve_gating(
        recipient=person,
        template_key="scheduled_job.failure",
        requested_push_channels=["feishu", "email"],
    )
    assert muted == []


@pytest.mark.asyncio
async def test_resolve_gating_quiet_hours_parks_mutable_only(
    db_session: AsyncSession, employee_user,
) -> None:
    user, _ = employee_user
    person = await db_session.get(IdentityPerson, user.id)
    person.country = "cn"  # Asia/Shanghai, deterministic window math
    svc = NotificationPreferenceService(db_session)
    await svc.upsert(
        person_id=user.id,
        muted_channels={},
        quiet_hours_enabled=True,
        quiet_hours_start="09:00",
        quiet_hours_end="17:00",
    )
    await db_session.commit()

    inside = datetime(2026, 6, 17, 4, 0, tzinfo=UTC)   # 12:00 CST → inside
    outside = datetime(2026, 6, 17, 14, 0, tzinfo=UTC)  # 22:00 CST → outside

    _, park = await svc.resolve_gating(
        recipient=person, template_key="asset.borrow_overdue",
        requested_push_channels=["feishu"], now_utc=inside,
    )
    assert park is not None

    _, park_out = await svc.resolve_gating(
        recipient=person, template_key="asset.borrow_overdue",
        requested_push_channels=["feishu"], now_utc=outside,
    )
    assert park_out is None

    # Non-mutable category never parks even inside the window.
    _, park_sys = await svc.resolve_gating(
        recipient=person, template_key="scheduled_job.failure",
        requested_push_channels=["feishu"], now_utc=inside,
    )
    assert park_sys is None


# ── inbox bucketing + endpoints ─────────────────────────────────────

async def _seed(db, rid, title, template_key, read=False):
    row = NotificationOutbox(
        recipient_person_id=rid, title=title, body_text="body",
        channels=["email"], status=STATUS_DELIVERED, template_key=template_key,
        read_at=datetime.now(UTC) if read else None,
    )
    db.add(row)
    await db.commit()
    return row


@pytest.mark.asyncio
async def test_social_push_suppressed_when_muted(
    db_session: AsyncSession, employee_user, monkeypatch,
) -> None:
    """Muting the social category must stop the Feishu ping on the
    lightweight feed path too — while still keeping the inbox row."""
    from app.notifications.service import (
        NotificationEnvelope,
        NotificationService,
    )

    user, _ = employee_user
    person = await db_session.get(IdentityPerson, user.id)
    pref = NotificationPreferenceService(db_session)
    await pref.upsert(
        person_id=user.id,
        muted_channels={"social": ["feishu"]},
        quiet_hours_enabled=False,
        quiet_hours_start=None,
        quiet_hours_end=None,
    )
    await db_session.commit()

    calls: list = []

    async def fake_send_feishu(self, env):  # noqa: ANN001, ARG001
        calls.append(env.template_key)
        return (True, None, None)

    monkeypatch.setattr(NotificationService, "_send_feishu", fake_send_feishu)

    svc = NotificationService(db_session)
    env = NotificationEnvelope(
        recipient=person, title="liked your post", body_text="x",
        channels=["feishu"], template_key="feed.like.summary",
    )
    await svc.push_inbox_and_feishu(env)
    await db_session.commit()
    assert calls == []  # muted → no Feishu push

    # The inbox row is still written (data never lost, only the ping quiets).
    rows = (await db_session.execute(
        NotificationOutbox.__table__.select().where(
            NotificationOutbox.recipient_person_id == user.id,
        )
    )).all()
    assert len(rows) == 1

    # Un-mute → the push fires.
    await pref.upsert(
        person_id=user.id, muted_channels={}, quiet_hours_enabled=False,
        quiet_hours_start=None, quiet_hours_end=None,
    )
    await db_session.commit()
    await svc.push_inbox_and_feishu(env)
    await db_session.commit()
    assert calls == ["feed.like.summary"]


@pytest.mark.asyncio
async def test_inbox_bucket_filter_and_row_fields(
    client, employee_user, db_session,
) -> None:
    user, token = employee_user
    headers = {"Authorization": f"Bearer {token}"}
    await _seed(db_session, user.id, "asset", "asset.borrow_overdue")   # action
    await _seed(db_session, user.id, "welcome", "hire_day_it_info")     # info
    await _seed(db_session, user.id, "like", "feed.like.summary")       # social

    r = await client.get(
        "/api/v1/notifications/inbox?bucket=action", headers=headers,
    )
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["meta"]["total"] == 1
    assert body["data"][0]["category"] == "assets"
    assert body["data"][0]["bucket"] == BUCKET_ACTION

    r = await client.get(
        "/api/v1/notifications/inbox?bucket=social", headers=headers,
    )
    assert r.json()["meta"]["total"] == 1
    assert r.json()["data"][0]["bucket"] == BUCKET_SOCIAL

    r = await client.get(
        "/api/v1/notifications/inbox?bucket=info", headers=headers,
    )
    assert r.json()["meta"]["total"] == 1


@pytest.mark.asyncio
async def test_inbox_summary_unread_by_bucket(
    client, employee_user, db_session,
) -> None:
    user, token = employee_user
    headers = {"Authorization": f"Bearer {token}"}
    await _seed(db_session, user.id, "a", "asset.borrow_overdue")
    await _seed(db_session, user.id, "b", "feed.like.summary")
    await _seed(db_session, user.id, "c", "hire_day_it_info", read=True)

    r = await client.get(
        "/api/v1/notifications/inbox/summary", headers=headers,
    )
    data = r.json()["data"]
    assert data["unread_by_bucket"]["action"] == 1
    assert data["unread_by_bucket"]["social"] == 1
    assert data["unread_by_bucket"]["info"] == 0  # the info row was read


@pytest.mark.asyncio
async def test_preferences_endpoint_roundtrip_and_sanitisation(
    client, employee_user,
) -> None:
    _, token = employee_user
    headers = {"Authorization": f"Bearer {token}"}

    # Default (no row) returns the full catalog with all-on.
    r = await client.get("/api/v1/notifications/preferences", headers=headers)
    assert r.status_code == 200, r.text
    data = r.json()["data"]
    assert data["quiet_hours_enabled"] is False
    cat_keys = {c["key"] for c in data["categories"]}
    assert "social" in cat_keys and "system" not in cat_keys

    # Save with junk that must be sanitised away.
    r = await client.put(
        "/api/v1/notifications/preferences",
        headers=headers,
        json={
            "muted_channels": {
                "social": ["feishu", "email", "sms_bogus"],
                "system": ["feishu"],          # non-mutable → dropped
                "bogus_cat": ["feishu"],       # unknown → dropped
            },
            "quiet_hours_enabled": True,
            "quiet_hours_start": "22:00",
            "quiet_hours_end": "08:00",
        },
    )
    assert r.status_code == 200, r.text
    saved = r.json()["data"]
    assert saved["muted_channels"] == {"social": ["email", "feishu"]}
    assert saved["quiet_hours_enabled"] is True
    assert saved["quiet_hours_start"] == "22:00"

    # Unparseable (but length-valid) time → service forces quiet hours off.
    r = await client.put(
        "/api/v1/notifications/preferences",
        headers=headers,
        json={
            "muted_channels": {},
            "quiet_hours_enabled": True,
            "quiet_hours_start": "99:99",  # 5 chars, but not a real time
            "quiet_hours_end": "08:00",
        },
    )
    assert r.status_code == 200, r.text
    assert r.json()["data"]["quiet_hours_enabled"] is False