"""Owner-facing cloud-efficiency digest (owner_digest + 触达设置 API).

Default-off weekly per-owner aggregate: one card per owner with pending
actionable recommendations (server-level owner ∪ subscription 业主),
respecting triage/snooze, the canary allowlist, active-employee gating,
and per-week dedup. The admin preview shows the blast radius before the
switch is flipped.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta
from decimal import Decimal

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

from app.domains.cmdb.models import (
    CmdbCloudScope,
    CmdbCostFact,
    CmdbServer,
    CmdbServerEfficiency,
    CmdbServerEfficiencyTriage,
)
from app.domains.cmdb.owner_digest import (
    OwnerDigestConfig,
    collect_owner_digest,
    save_owner_digest_config,
    send_owner_digest,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration
from app.notifications.models import NotificationOutbox


async def _integration(db: AsyncSession) -> Integration:
    integ = Integration(
        id=uuid.uuid4(), kind="cloud", provider="aws",
        name="aws-dig", slug=f"aws-{uuid.uuid4().hex[:6]}",
        enabled=True, is_primary=False, capabilities=["servers", "metrics"],
    )
    db.add(integ)
    await db.flush()
    return integ


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


async def _server(db: AsyncSession, integ: Integration, **kw) -> CmdbServer:
    now = datetime.now(UTC)
    d = {
        "id": uuid.uuid4(),
        "instance_id": kw.pop("instance_id", f"i-{uuid.uuid4().hex[:10]}"),
        "name": kw.pop("name", "srv"),
        "status": kw.pop("status", "running"),
        "region": "us-east-1",
        "integration_id": str(integ.id),
        "first_seen_at": now - timedelta(days=100),
        "last_synced_at": now,
    }
    d.update(kw)
    s = CmdbServer(**d)
    db.add(s)
    await db.flush()
    return s


async def _efficiency(db, integ, server, **kw) -> CmdbServerEfficiency:
    d = {
        "id": uuid.uuid4(),
        "server_id": str(server.id),
        "integration_id": str(integ.id),
        "verdict": kw.pop("verdict", "idle"),
        "recommendation": kw.pop("recommendation", "stop"),
        "estimated_monthly_savings_usd": kw.pop("savings", Decimal("80")),
        "computed_at": datetime.now(UTC),
    }
    d.update(kw)
    e = CmdbServerEfficiency(**d)
    db.add(e)
    await db.flush()
    return e


async def _scope_with_attribution(
    db, integ, owner, server, *, environment: str = "dev",
    display_name: str = "dig-sub",
) -> CmdbCloudScope:
    scope = CmdbCloudScope(
        id=uuid.uuid4(), integration_id=str(integ.id),
        scope_type="aws_account", external_id=f"acct-{uuid.uuid4().hex[:8]}",
        display_name=display_name, owner_person_id=str(owner.id),
        environment=environment,
    )
    db.add(scope)
    await db.flush()
    db.add(CmdbCostFact(
        fact_key=uuid.uuid4().hex, date=datetime.now(UTC).date(),
        integration_id=str(integ.id), scope_id=str(scope.id),
        resource_id=server.instance_id, resource_type="Instance",
        service="AmazonEC2", amount_usd=Decimal("3"), currency="USD",
    ))
    await db.flush()
    return scope


async def _digest_outbox_rows(db) -> list[NotificationOutbox]:
    return list(
        (
            await db.execute(
                select(NotificationOutbox).where(
                    NotificationOutbox.related_resource_type
                    == "cloud_efficiency_owner_digest",
                )
            )
        ).scalars()
    )


# ── aggregation ──────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_collect_aggregates_by_effective_owner(db_session: AsyncSession):
    integ = await _integration(db_session)
    alice = await _person(db_session, f"alice-{uuid.uuid4().hex[:6]}@x.com")

    # Two pending machines: one server-owned, one via subscription.
    owned = await _server(db_session, integ, owner_person_id=str(alice.id))
    await _efficiency(db_session, integ, owned, savings=Decimal("100"))
    via_scope = await _server(db_session, integ)
    await _scope_with_attribution(db_session, integ, alice, via_scope)
    await _efficiency(db_session, integ, via_scope, verdict="oversized",
                      recommendation="downsize", savings=Decimal("50"))

    # Snoozed (live) item — must NOT count.
    snoozed = await _server(db_session, integ, owner_person_id=str(alice.id))
    await _efficiency(db_session, integ, snoozed, savings=Decimal("999"))
    db_session.add(CmdbServerEfficiencyTriage(
        server_id=str(snoozed.id), status="snoozed",
        verdict_at_decision="idle",
        snooze_until=datetime.now(UTC) + timedelta(days=10),
    ))

    # Leaver's machine — owner inactive, dropped entirely.
    leaver = await _person(
        db_session, f"gone-{uuid.uuid4().hex[:6]}@x.com",
        employment_status="left",
    )
    l_srv = await _server(db_session, integ, owner_person_id=str(leaver.id))
    await _efficiency(db_session, integ, l_srv, savings=Decimal("77"))
    await db_session.flush()

    entries = await collect_owner_digest(db_session, max_items=5)
    assert len(entries) == 1
    e = entries[0]
    assert e.person_id == str(alice.id)
    assert e.server_count == 2
    assert e.estimated_monthly_savings_usd == pytest.approx(150.0)
    # Sorted by savings desc; scope-attributed item carries the scope name.
    assert e.items[0]["estimated_monthly_savings_usd"] == 100.0
    scope_item = next(i for i in e.items if i["scope_name"])
    assert scope_item["scope_name"] == "dig-sub"


@pytest.mark.asyncio
async def test_collect_env_filter_drops_excluded_subscription_machines(
    db_session: AsyncSession,
):
    """A machine inherits its subscription's environment — with
    exclude_environments=['prod'] the prod-attributed box drops out even
    when it is EXPLICITLY server-owned, while dev-attributed and
    unattributable machines stay."""
    integ = await _integration(db_session)
    alice = await _person(db_session, f"alice-{uuid.uuid4().hex[:6]}@x.com")

    # Server-owned box attributed (via cost fact) to a PROD subscription.
    prod_srv = await _server(
        db_session, integ, owner_person_id=str(alice.id), name="prod-box",
    )
    await _scope_with_attribution(
        db_session, integ, alice, prod_srv,
        environment="prod", display_name="prod-sub",
    )
    await _efficiency(db_session, integ, prod_srv, savings=Decimal("500"))

    # Box inheriting owner from a DEV subscription.
    dev_srv = await _server(db_session, integ, name="dev-box")
    await _scope_with_attribution(
        db_session, integ, alice, dev_srv, environment="dev",
    )
    await _efficiency(db_session, integ, dev_srv, savings=Decimal("40"))

    # Server-owned box with NO subscription attribution at all — unknown
    # environment is NOT "excluded".
    lone_srv = await _server(
        db_session, integ, owner_person_id=str(alice.id), name="lone-box",
    )
    await _efficiency(db_session, integ, lone_srv, savings=Decimal("30"))
    await db_session.flush()

    # No filter: all three count.
    entries = await collect_owner_digest(db_session, max_items=5)
    assert len(entries) == 1 and entries[0].server_count == 3

    # prod excluded: the prod-attributed box is gone, the other two stay.
    entries = await collect_owner_digest(
        db_session, max_items=5, exclude_environments=["prod"],
    )
    assert len(entries) == 1
    assert entries[0].server_count == 2
    names = {i["name"] for i in entries[0].items}
    assert names == {"dev-box", "lone-box"}

    # Excluding every known env still keeps the unattributable box.
    entries = await collect_owner_digest(
        db_session, max_items=5,
        exclude_environments=["dev", "staging", "prod"],
    )
    assert len(entries) == 1
    assert {i["name"] for i in entries[0].items} == {"lone-box"}


@pytest.mark.asyncio
async def test_config_env_filter_persists_and_preview_overrides(
    client: AsyncClient, db_session: AsyncSession, cloud_admin_headers,
):
    integ = await _integration(db_session)
    alice = await _person(db_session, f"alice-{uuid.uuid4().hex[:6]}@x.com")
    prod_srv = await _server(db_session, integ, owner_person_id=str(alice.id))
    await _scope_with_attribution(
        db_session, integ, alice, prod_srv, environment="prod",
    )
    await _efficiency(db_session, integ, prod_srv, savings=Decimal("60"))
    await db_session.commit()

    # Unknown env value → 422 on both PUT and preview override.
    r = await client.put(
        "/api/v1/cmdb/efficiency/owner-digest/config",
        json={"exclude_environments": ["production"]},
        headers=cloud_admin_headers,
    )
    assert r.status_code == 422
    r = await client.get(
        "/api/v1/cmdb/efficiency/owner-digest/preview",
        params={"exclude_environments": "production"},
        headers=cloud_admin_headers,
    )
    assert r.status_code == 422

    # Preview override (not yet saved) filters the prod machine out.
    r = await client.get(
        "/api/v1/cmdb/efficiency/owner-digest/preview",
        params={"exclude_environments": "prod"},
        headers=cloud_admin_headers,
    )
    assert r.status_code == 200
    assert r.json()["data"]["totals"]["recipients"] == 0

    # Persist the filter; the saved config now applies without the param.
    r = await client.put(
        "/api/v1/cmdb/efficiency/owner-digest/config",
        json={"exclude_environments": ["prod"]},
        headers=cloud_admin_headers,
    )
    assert r.status_code == 200
    assert r.json()["data"]["exclude_environments"] == ["prod"]
    r = await client.get(
        "/api/v1/cmdb/efficiency/owner-digest/preview",
        headers=cloud_admin_headers,
    )
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["config"]["exclude_environments"] == ["prod"]
    assert data["totals"]["recipients"] == 0
    # Explicit empty override clears the saved filter for this preview.
    r = await client.get(
        "/api/v1/cmdb/efficiency/owner-digest/preview",
        params={"exclude_environments": ""},
        headers=cloud_admin_headers,
    )
    assert r.json()["data"]["totals"]["recipients"] == 1


@pytest.mark.asyncio
async def test_send_respects_env_filter(client, db_session: AsyncSession):
    integ = await _integration(db_session)
    alice = await _person(db_session, f"alice-{uuid.uuid4().hex[:6]}@x.com")
    prod_srv = await _server(db_session, integ, owner_person_id=str(alice.id))
    await _scope_with_attribution(
        db_session, integ, alice, prod_srv, environment="prod",
    )
    await _efficiency(db_session, integ, prod_srv, savings=Decimal("60"))
    await save_owner_digest_config(
        db_session,
        OwnerDigestConfig(enabled=True, exclude_environments=["prod"]),
    )
    await db_session.commit()

    res = await send_owner_digest(db_session)
    assert res["status"] == "skipped"
    assert "no pending recommendations" in res["reason"]
    assert await _digest_outbox_rows(db_session) == []


# ── $0 / min-savings guard ───────────────────────────────────────────
@pytest.mark.asyncio
async def test_collect_drops_zero_and_unattributed_savings(
    db_session: AsyncSession,
):
    """A recipient whose flagged machines carry $0 / null savings (e.g. an
    AWS/GKE box with no attributable cost) must NEVER be collected — pushing
    a '预计每月可省 $0' card is the bug. The floor is >= 0 so exactly-$0 is
    dropped even at the default threshold."""
    integ = await _integration(db_session)
    zero = await _person(db_session, f"zero-{uuid.uuid4().hex[:6]}@x.com")
    nullp = await _person(db_session, f"null-{uuid.uuid4().hex[:6]}@x.com")
    real = await _person(db_session, f"real-{uuid.uuid4().hex[:6]}@x.com")

    z_srv = await _server(db_session, integ, owner_person_id=str(zero.id))
    await _efficiency(db_session, integ, z_srv, savings=Decimal("0"))
    n_srv = await _server(db_session, integ, owner_person_id=str(nullp.id))
    await _efficiency(db_session, integ, n_srv, savings=None)
    r_srv = await _server(db_session, integ, owner_person_id=str(real.id))
    await _efficiency(db_session, integ, r_srv, savings=Decimal("42"))
    await db_session.flush()

    entries = await collect_owner_digest(db_session, max_items=5)
    assert [e.person_id for e in entries] == [str(real.id)]


@pytest.mark.asyncio
async def test_collect_min_savings_threshold(db_session: AsyncSession):
    """A positive floor drops recipients at or below it."""
    integ = await _integration(db_session)
    small = await _person(db_session, f"small-{uuid.uuid4().hex[:6]}@x.com")
    big = await _person(db_session, f"big-{uuid.uuid4().hex[:6]}@x.com")
    s_srv = await _server(db_session, integ, owner_person_id=str(small.id))
    await _efficiency(db_session, integ, s_srv, savings=Decimal("5"))
    b_srv = await _server(db_session, integ, owner_person_id=str(big.id))
    await _efficiency(db_session, integ, b_srv, savings=Decimal("50"))
    await db_session.flush()

    entries = await collect_owner_digest(
        db_session, max_items=5, min_savings_usd=10.0,
    )
    assert [e.person_id for e in entries] == [str(big.id)]


@pytest.mark.asyncio
async def test_neutralized_stopped_box_excluded(db_session: AsyncSession):
    """An already-stopped idle/oversized box is neutralized to
    recommendation='none' at compute time; it keeps its verdict for
    visibility but must NOT be collected (nagging the owner to stop a box
    that's already stopped is exactly the '$0' embarrassment)."""
    integ = await _integration(db_session)
    alice = await _person(db_session, f"alice-{uuid.uuid4().hex[:6]}@x.com")
    stopped = await _server(
        db_session, integ, owner_person_id=str(alice.id), status="stopped",
    )
    await _efficiency(
        db_session, integ, stopped,
        verdict="idle", recommendation="none", savings=Decimal("0"),
    )
    await db_session.flush()
    entries = await collect_owner_digest(db_session, max_items=5)
    assert entries == []


@pytest.mark.asyncio
async def test_config_min_savings_persists_and_preview_overrides(
    client: AsyncClient, db_session: AsyncSession, cloud_admin_headers,
):
    integ = await _integration(db_session)
    alice = await _person(db_session, f"alice-{uuid.uuid4().hex[:6]}@x.com")
    srv = await _server(db_session, integ, owner_person_id=str(alice.id))
    await _efficiency(db_session, integ, srv, savings=Decimal("5"))
    await db_session.commit()

    # Preview override raises the floor above this recipient → 0 recipients.
    r = await client.get(
        "/api/v1/cmdb/efficiency/owner-digest/preview",
        params={"min_savings": 10},
        headers=cloud_admin_headers,
    )
    assert r.status_code == 200
    assert r.json()["data"]["totals"]["recipients"] == 0

    # Persist the floor; saved config now applies without the param.
    r = await client.put(
        "/api/v1/cmdb/efficiency/owner-digest/config",
        json={"min_savings_usd": 10},
        headers=cloud_admin_headers,
    )
    assert r.status_code == 200
    assert r.json()["data"]["min_savings_usd"] == 10.0
    r = await client.get(
        "/api/v1/cmdb/efficiency/owner-digest/preview",
        headers=cloud_admin_headers,
    )
    assert r.json()["data"]["totals"]["recipients"] == 0


# ── send path ────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_send_disabled_skips(client, db_session: AsyncSession):
    res = await send_owner_digest(db_session)
    assert res["status"] == "skipped"
    assert "disabled" in res["reason"]


@pytest.mark.asyncio
async def test_send_creates_one_card_per_owner_and_dedups(
    client, db_session: AsyncSession,
):
    integ = await _integration(db_session)
    alice = await _person(db_session, f"alice-{uuid.uuid4().hex[:6]}@x.com")
    bob = await _person(db_session, f"bob-{uuid.uuid4().hex[:6]}@x.com")
    for owner, n in ((alice, 2), (bob, 1)):
        for _ in range(n):
            srv = await _server(db_session, integ, owner_person_id=str(owner.id))
            await _efficiency(db_session, integ, srv)
    await save_owner_digest_config(
        db_session, OwnerDigestConfig(enabled=True),
    )
    await db_session.commit()

    res = await send_owner_digest(db_session)
    assert res["status"] == "ok"
    assert res["dispatched"] == 2  # ONE aggregated card per owner
    rows = await _digest_outbox_rows(db_session)
    assert len(rows) == 2
    by_recipient = {str(r.recipient_person_id): r for r in rows}
    assert "2 台" in by_recipient[str(alice.id)].title

    # Re-run inside the dedup window → nothing new.
    res2 = await send_owner_digest(db_session)
    assert res2["status"] == "ok"
    assert res2["dispatched"] == 0
    assert res2["deduped"] == 2
    assert len(await _digest_outbox_rows(db_session)) == 2


@pytest.mark.asyncio
async def test_send_canary_narrows_recipients(client, db_session: AsyncSession):
    integ = await _integration(db_session)
    alice = await _person(db_session, f"alice-{uuid.uuid4().hex[:6]}@x.com")
    bob = await _person(db_session, f"bob-{uuid.uuid4().hex[:6]}@x.com")
    for owner in (alice, bob):
        srv = await _server(db_session, integ, owner_person_id=str(owner.id))
        await _efficiency(db_session, integ, srv)
    await save_owner_digest_config(
        db_session,
        OwnerDigestConfig(enabled=True, canary_person_ids=[str(alice.id)]),
    )
    await db_session.commit()

    res = await send_owner_digest(db_session)
    assert res["status"] == "ok"
    assert res["dispatched"] == 1
    rows = await _digest_outbox_rows(db_session)
    assert len(rows) == 1
    assert str(rows[0].recipient_person_id) == str(alice.id)


# ── admin API ────────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_config_rbac_and_persistence(
    client: AsyncClient, db_session: AsyncSession,
    cloud_admin_headers, employee_headers,
):
    # Employee: no read, no write.
    r = await client.get(
        "/api/v1/cmdb/efficiency/owner-digest/config", headers=employee_headers,
    )
    assert r.status_code == 403
    r = await client.put(
        "/api/v1/cmdb/efficiency/owner-digest/config",
        json={"enabled": True}, headers=employee_headers,
    )
    assert r.status_code == 403

    # Cloud admin: defaults off → flip on with canary.
    r = await client.get(
        "/api/v1/cmdb/efficiency/owner-digest/config",
        headers=cloud_admin_headers,
    )
    assert r.status_code == 200
    assert r.json()["data"]["enabled"] is False

    pid = str(uuid.uuid4())
    r = await client.put(
        "/api/v1/cmdb/efficiency/owner-digest/config",
        json={"enabled": True, "canary_person_ids": [pid]},
        headers=cloud_admin_headers,
    )
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["enabled"] is True and data["canary_person_ids"] == [pid]

    r = await client.get(
        "/api/v1/cmdb/efficiency/owner-digest/config",
        headers=cloud_admin_headers,
    )
    assert r.json()["data"]["enabled"] is True


@pytest.mark.asyncio
async def test_preview_shows_blast_radius(
    client: AsyncClient, db_session: AsyncSession, cloud_admin_headers,
):
    integ = await _integration(db_session)
    alice = await _person(db_session, f"alice-{uuid.uuid4().hex[:6]}@x.com")
    srv = await _server(db_session, integ, owner_person_id=str(alice.id))
    await _efficiency(db_session, integ, srv, savings=Decimal("60"))
    await db_session.commit()

    r = await client.get(
        "/api/v1/cmdb/efficiency/owner-digest/preview",
        headers=cloud_admin_headers,
    )
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["totals"]["recipients"] == 1
    assert data["totals"]["would_send"] == 1
    assert data["totals"]["servers"] == 1
    rec = data["recipients"][0]
    assert rec["person_id"] == str(alice.id)
    assert rec["items"][0]["estimated_monthly_savings_usd"] == 60.0
    # No outbox rows were created — preview is a dry run.
    assert await _digest_outbox_rows(db_session) == []


@pytest.mark.asyncio
async def test_history_returns_recent_runs(
    client: AsyncClient, db_session: AsyncSession,
    cloud_admin_headers, employee_headers,
):
    """The history endpoint surfaces the cron's real run log so IT can
    confirm the weekly push actually fired (proof the preview can't give)."""
    from app.scheduled_jobs.models import ScheduledJob, ScheduledJobRun

    job = ScheduledJob(
        id=uuid.uuid4(),
        name="cmdb.efficiency_owner_digest",
        display_name="Owner digest",
        handler_key="cmdb.efficiency_owner_digest",
        trigger_type="cron",
        cron_expr="15 2 * * 1",
        timezone="UTC",
        enabled=True,
        last_status="ok",
        last_run_at=datetime.now(UTC),
        run_count=2,
    )
    db_session.add(job)
    await db_session.flush()
    # One successful send + one silent skip — both must render.
    db_session.add(ScheduledJobRun(
        id=uuid.uuid4(), job_id=job.id,
        started_at=datetime.now(UTC) - timedelta(days=7),
        finished_at=datetime.now(UTC) - timedelta(days=7),
        status="ok", triggered_by="scheduler",
        result_summary={
            "status": "ok", "recipients": 3, "dispatched": 3,
            "deduped": 0, "failed": 0, "exclude_environments": ["prod"],
        },
    ))
    db_session.add(ScheduledJobRun(
        id=uuid.uuid4(), job_id=job.id,
        started_at=datetime.now(UTC) - timedelta(days=14),
        finished_at=datetime.now(UTC) - timedelta(days=14),
        status="skipped", triggered_by="scheduler",
        result_summary={"status": "skipped", "reason": "owner digest disabled"},
    ))
    await save_owner_digest_config(db_session, OwnerDigestConfig(enabled=True))
    await db_session.commit()

    # Employee has no cloud-read scope → 403.
    r = await client.get(
        "/api/v1/cmdb/efficiency/owner-digest/history", headers=employee_headers,
    )
    assert r.status_code == 403

    r = await client.get(
        "/api/v1/cmdb/efficiency/owner-digest/history", headers=cloud_admin_headers,
    )
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["enabled"] is True
    assert data["job"]["run_count"] == 2
    assert data["job"]["cron_expr"] == "15 2 * * 1"
    # Newest first.
    runs = data["runs"]
    assert len(runs) == 2
    assert runs[0]["status"] == "ok"
    assert runs[0]["dispatched"] == 3
    assert runs[0]["exclude_environments"] == ["prod"]
    assert runs[0]["skipped_reason"] is None
    assert runs[1]["status"] == "skipped"
    assert runs[1]["skipped_reason"] == "owner digest disabled"
    assert runs[1]["dispatched"] is None