"""Tests for the software-seat reconciliation backbone.

Covers the shared persistence helper every ``*.seat_sync`` handler funnels
through (``record_seat_reconciliation``) + the two aggregation functions
the HTTP / MCP surfaces read (``_compute_seat_reconciliation_overview`` /
``_compute_seat_reconciliation_detail``). Real DB (the recon helper relies
on MySQL ``ON DUPLICATE KEY UPDATE``), no external mocks — the data flows
purely through the new tables.

Verifies the "vendor bills 72 but Helix shows 69, the other 3 are external"
shape end-to-end:
  ① a sync records external holders + a per-plan snapshot;
  ② re-running with a holder gone flips it to ``resolved`` (not deleted)
     and keeps ``first_observed_at`` while bumping ``last_observed_at``;
  ③ the overview aggregates vendor/occupied/external/scheduled-cancel
     per software from live tables;
  ④ the detail lists the external holders (with vendor deep-link) and the
     live scheduled-cancel ("下个月不占") seats.
"""
from __future__ import annotations

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

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

from app.domains.asset.models import (
    AssetSoftwareCatalog,
    SoftwareExternalHolder,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
    SoftwareSeatSyncStat,
)
from app.domains.asset.router import (
    _compute_seat_reconciliation_detail,
    _compute_seat_reconciliation_overview,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration
from app.scheduled_jobs._seat_sync_helpers import (
    ExternalHolderObservation,
    record_seat_reconciliation,
)


async def _integration(db: AsyncSession) -> Integration:
    integ = Integration(
        id=uuid.uuid4(),
        kind="software_provisioning",
        provider="jira",
        name="recon-test",
        slug=f"recon-{uuid.uuid4().hex[:6]}",
        enabled=True,
        is_primary=True,
        config_public={},
    )
    db.add(integ)
    await db.flush()
    return integ


async def _plan(
    db: AsyncSession, *, catalog_name: str = "Jira Software",
    plan_name: str = "Full member",
) -> SoftwareLicensePlan:
    sw = AssetSoftwareCatalog(
        id=uuid.uuid4(), name=catalog_name, asset_type="SaaS",
        status="active", offboarding_policy="auto_revoke",
    )
    db.add(sw)
    await db.flush()
    plan = SoftwareLicensePlan(
        id=uuid.uuid4(), software_id=sw.id, name=plan_name,
        price=Decimal("7.50"), billing_cycle="monthly", status="active",
        max_seats=None,
    )
    db.add(plan)
    await db.flush()
    return plan


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


# ── ① first sync records external holders + snapshot ────────────────


@pytest.mark.asyncio
async def test_record_persists_externals_and_stat(db_session: AsyncSession):
    integ = await _integration(db_session)
    plan = await _plan(db_session)
    now = datetime.now(UTC)

    n = await record_seat_reconciliation(
        db_session,
        holder_model=SoftwareExternalHolder,
        stat_model=SoftwareSeatSyncStat,
        integration_id=integ.id,
        software_id=plan.software_id,
        plan_id=plan.id,
        provider="jira",
        product_key="jira-software",
        site="tidb.atlassian.net",
        # Mirrors prod JSM shape: vendor=72 = matched=69 + external=3 + 0.
        # The 3 distinct external observations + 1 duplicate (3rd contractor)
        # exercise both buckets-add-up AND in-run email dedup.
        vendor_license_holders=72,
        holders_matched=69,
        holders_no_email=0,
        externals=[
            ExternalHolderObservation(
                email="contractor@partner.com",
                vendor_account_id="acc-1",
                account_status="active",
            ),
            ExternalHolderObservation(
                email="svc-bot@pingcap.com",
                vendor_account_id="acc-2",
                account_status="active",
            ),
            ExternalHolderObservation(
                email="consultant@vendor.com",
                vendor_account_id="acc-3",
                account_status="active",
            ),
            # Duplicate email (same plan, another site) is deduped.
            ExternalHolderObservation(email="contractor@partner.com"),
        ],
        now=now,
    )
    await db_session.commit()

    assert n == 3  # 4 raw observations → 3 distinct emails

    holders = (await db_session.execute(
        select(SoftwareExternalHolder)
        .where(SoftwareExternalHolder.plan_id == plan.id),
    )).scalars().all()
    assert len(holders) == 3
    assert all(not h.resolved for h in holders)
    assert {h.email for h in holders} == {
        "contractor@partner.com",
        "svc-bot@pingcap.com",
        "consultant@vendor.com",
    }
    # All four row-persisted observations land with the default
    # ``category='external'`` discriminator — see also the dedicated
    # ``test_all_four_categories_persist`` test below for the other
    # three categories (leaver_residual / vendor_inactive / off_paid_site).
    assert all(h.category == "external" for h in holders)

    stat = (await db_session.execute(
        select(SoftwareSeatSyncStat)
        .where(SoftwareSeatSyncStat.plan_id == plan.id),
    )).scalar_one()
    assert stat.vendor_license_holders == 72
    assert stat.holders_matched == 69
    assert stat.holders_external == 3
    assert stat.holders_no_email == 0
    # Identity 72 = 69 + 3 + 0 + 0 + 0 holds (off_paid_site excluded,
    # but here it's also 0 — see test_off_paid_site_bucket for the
    # case where vendor_license_holders < total holders).
    assert stat.holders_leaver_residual == 0
    assert stat.holders_vendor_inactive == 0
    assert stat.holders_off_paid_site == 0


# ── ② absent holder resolves; first_observed_at preserved ───────────


@pytest.mark.asyncio
async def test_absent_holder_resolves_and_preserves_first_seen(
    db_session: AsyncSession,
):
    integ = await _integration(db_session)
    plan = await _plan(db_session)
    t0 = datetime.now(UTC) - timedelta(days=1)

    await record_seat_reconciliation(
        db_session,
        holder_model=SoftwareExternalHolder,
        stat_model=SoftwareSeatSyncStat,
        integration_id=integ.id,
        software_id=plan.software_id,
        plan_id=plan.id,
        provider="jira",
        product_key="jira-software",
        site=None,
        vendor_license_holders=10,
        holders_matched=8,
        holders_no_email=0,
        externals=[
            ExternalHolderObservation(email="a@partner.com"),
            ExternalHolderObservation(email="b@partner.com"),
        ],
        now=t0,
    )
    await db_session.commit()
    first_seen = {
        h.email: h.first_observed_at
        for h in (await db_session.execute(
            select(SoftwareExternalHolder)
            .where(SoftwareExternalHolder.plan_id == plan.id),
        )).scalars().all()
    }

    # Second run: ``a`` is gone (matched an employee / left vendor), ``b``
    # still external.
    t1 = datetime.now(UTC)
    await record_seat_reconciliation(
        db_session,
        holder_model=SoftwareExternalHolder,
        stat_model=SoftwareSeatSyncStat,
        integration_id=integ.id,
        software_id=plan.software_id,
        plan_id=plan.id,
        provider="jira",
        product_key="jira-software",
        site=None,
        vendor_license_holders=10,
        holders_matched=9,
        holders_no_email=0,
        externals=[ExternalHolderObservation(email="b@partner.com")],
        now=t1,
    )
    await db_session.commit()

    rows = {
        h.email: h
        for h in (await db_session.execute(
            select(SoftwareExternalHolder)
            .where(SoftwareExternalHolder.plan_id == plan.id),
        )).scalars().all()
    }
    assert rows["a@partner.com"].resolved
    assert rows["a@partner.com"].resolved_at is not None
    assert not rows["b@partner.com"].resolved
    # first_observed_at is preserved on the still-present holder.
    assert rows["b@partner.com"].first_observed_at == first_seen["b@partner.com"]

    stat = (await db_session.execute(
        select(SoftwareSeatSyncStat)
        .where(SoftwareSeatSyncStat.plan_id == plan.id),
    )).scalar_one()
    assert stat.holders_external == 1
    assert stat.holders_matched == 9


# ── ③ + ④ overview + detail aggregation ─────────────────────────────


@pytest.mark.asyncio
async def test_overview_and_detail_aggregation(db_session: AsyncSession):
    integ = await _integration(db_session)
    plan = await _plan(db_session, catalog_name="Zoom", plan_name="Licensed")
    now = datetime.now(UTC)

    # Two occupied employee seats — one scheduled to cancel next period.
    p1 = await _person(db_session, "alice@pingcap.com")
    p2 = await _person(db_session, "bob@pingcap.com")
    db_session.add(SoftwareSeatAllocation(
        plan_id=plan.id, user_id=p1.id, status="active", allocated_at=now,
    ))
    db_session.add(SoftwareSeatAllocation(
        plan_id=plan.id, user_id=p2.id, status="active", allocated_at=now,
        cancel_at_period_end=now + timedelta(days=20),
        cancel_requested_at=now, cancel_requested_by=p2.id,
        cancel_reason="not needed next month",
    ))
    await db_session.flush()

    await record_seat_reconciliation(
        db_session,
        holder_model=SoftwareExternalHolder,
        stat_model=SoftwareSeatSyncStat,
        integration_id=integ.id,
        software_id=plan.software_id,
        plan_id=plan.id,
        provider="zoom",
        product_key="zoom-licensed",
        site=None,
        vendor_license_holders=3,
        holders_matched=2,
        holders_no_email=0,
        externals=[
            ExternalHolderObservation(
                email="ext@partner.com", vendor_account_id="zoom-9",
                account_status="active",
            ),
        ],
        now=now,
    )
    await db_session.commit()

    overview = await _compute_seat_reconciliation_overview(db_session)
    row = next(r for r in overview if r["software_id"] == str(plan.software_id))
    assert row["vendor_license_holders"] == 3
    assert row["occupied_seats"] == 2
    assert row["external_unmatched"] == 1
    assert row["scheduled_cancel"] == 1
    assert row["provider"] == "zoom"
    assert row["software_name"] == "Zoom"

    detail = await _compute_seat_reconciliation_detail(
        db_session, plan.software_id, page=1, page_size=50,
        include_resolved=False,
    )
    assert detail["summary"]["vendor_license_holders"] == 3
    assert detail["summary"]["external_unmatched"] == 1
    assert detail["summary"]["occupied_seats"] == 2
    assert detail["summary"]["scheduled_cancel"] == 1
    assert detail["external_holders"]["total"] == 1
    holder = detail["external_holders"]["items"][0]
    assert holder["email"] == "ext@partner.com"
    # Zoom external holder gets a vendor-admin deep-link from its id.
    assert holder["vendor_admin_url"] == "https://zoom.us/user/zoom-9/profile"
    assert len(detail["scheduled_cancels"]) == 1
    sc = detail["scheduled_cancels"][0]
    assert sc["user_email"] == "bob@pingcap.com"
    assert sc["cancel_requested_by_name"] == p2.display_name
    assert len(detail["per_plan"]) == 1


# ── ⑤ all four row-persisted categories land with correct discriminator ──


@pytest.mark.asyncio
async def test_all_four_categories_persist_with_correct_category_field(
    db_session: AsyncSession,
):
    """Every row-persisted anomaly bucket (``external`` / ``leaver_residual`` /
    ``vendor_inactive`` / ``off_paid_site``) must round-trip the
    ``category`` discriminator + (when present) the ``person_id`` FK,
    and the migration-217 ``ix_software_external_holder_plan_category``
    index must return exactly that category's rows on filter.

    Pins three things at once:
      1. The handler's per-observation ``category`` field is honoured by
         the helper's ``mysql_insert(...).values(rows)`` write — a bug
         that defaulted every row to ``'external'`` would silently
         erase the new tabs from the drawer.
      2. The ``person_id`` FK is populated for matched-Helix-person
         categories (leaver_residual / vendor_inactive / off_paid_site)
         and NULL for ``external`` (definitionally no Helix match).
      3. The per-category index is queryable — the drawer's per-tab
         load relies on ``WHERE plan_id=X AND category=Y`` returning a
         single category's rows in a single index seek.
    """
    integ = await _integration(db_session)
    plan = await _plan(db_session)
    now = datetime.now(UTC)

    leaver = await _person(db_session, "leaver@pingcap.com")
    vendor_inactive_person = await _person(
        db_session, "deactivated@pingcap.com",
    )
    off_site_person = await _person(db_session, "sandbox-only@pingcap.com")

    await record_seat_reconciliation(
        db_session,
        holder_model=SoftwareExternalHolder,
        stat_model=SoftwareSeatSyncStat,
        integration_id=integ.id,
        software_id=plan.software_id,
        plan_id=plan.id,
        provider="jira",
        product_key="jira-software",
        site="tidb.atlassian.net",
        # Identity: 4 = 1 matched + 1 ext + 1 leaver + 1 inactive.
        # The off_paid_site holder is recorded as an observation but
        # is EXCLUDED from vendor_license_holders — Atlassian doesn't
        # bill for sandbox / personal sites, so the headline must not
        # count them either.
        vendor_license_holders=4,
        holders_matched=1,
        holders_no_email=0,
        holders_leaver_residual=1,
        holders_vendor_inactive=1,
        holders_off_paid_site=1,
        externals=[
            ExternalHolderObservation(
                email="contractor@partner.com",
                vendor_account_id="acc-ext",
                account_status="active",
                category="external",
                # person_id stays None — there's no Helix match for an
                # external holder by definition.
            ),
            ExternalHolderObservation(
                email="leaver@pingcap.com",
                vendor_account_id="acc-lv",
                account_status="active",
                category="leaver_residual",
                person_id=leaver.id,
            ),
            ExternalHolderObservation(
                email="deactivated@pingcap.com",
                vendor_account_id="acc-inact",
                account_status="inactive",
                category="vendor_inactive",
                person_id=vendor_inactive_person.id,
            ),
            ExternalHolderObservation(
                email="sandbox-only@pingcap.com",
                vendor_account_id="acc-sb",
                account_status="active",
                category="off_paid_site",
                person_id=off_site_person.id,
                site="tidb-sandbox.atlassian.net",
            ),
        ],
        now=now,
    )
    await db_session.commit()

    # ── (1) every row carries its declared category ──
    rows_by_email = {
        h.email: h
        for h in (await db_session.execute(
            select(SoftwareExternalHolder)
            .where(SoftwareExternalHolder.plan_id == plan.id),
        )).scalars().all()
    }
    assert rows_by_email["contractor@partner.com"].category == "external"
    assert rows_by_email["leaver@pingcap.com"].category == "leaver_residual"
    assert (
        rows_by_email["deactivated@pingcap.com"].category == "vendor_inactive"
    )
    assert (
        rows_by_email["sandbox-only@pingcap.com"].category == "off_paid_site"
    )

    # ── (2) person_id FK populated for matched-person categories ──
    assert rows_by_email["contractor@partner.com"].person_id is None
    assert rows_by_email["leaver@pingcap.com"].person_id == leaver.id
    assert (
        rows_by_email["deactivated@pingcap.com"].person_id
        == vendor_inactive_person.id
    )
    assert (
        rows_by_email["sandbox-only@pingcap.com"].person_id
        == off_site_person.id
    )
    # Per-observation site (off_paid_site holder's actual non-paid host)
    # overrides the per-call site arg so the drawer can show "where".
    assert (
        rows_by_email["sandbox-only@pingcap.com"].site
        == "tidb-sandbox.atlassian.net"
    )

    # ── (3) per-category index lookup returns just that category ──
    for cat in ("external", "leaver_residual", "vendor_inactive", "off_paid_site"):
        cat_rows = (await db_session.execute(
            select(SoftwareExternalHolder).where(
                SoftwareExternalHolder.plan_id == plan.id,
                SoftwareExternalHolder.category == cat,
            ),
        )).scalars().all()
        assert len(cat_rows) == 1, (
            f"category={cat!r} per-plan index lookup returned "
            f"{len(cat_rows)} rows, expected 1"
        )
        assert cat_rows[0].category == cat

    # ── (4) reverse lookup by person_id (the offboarding-view path) ──
    leaver_residuals = (await db_session.execute(
        select(SoftwareExternalHolder).where(
            SoftwareExternalHolder.person_id == leaver.id,
        ),
    )).scalars().all()
    assert len(leaver_residuals) == 1
    assert leaver_residuals[0].category == "leaver_residual"

    # ── (5) headline stat row mirrors the per-category counts ──
    stat = (await db_session.execute(
        select(SoftwareSeatSyncStat)
        .where(SoftwareSeatSyncStat.plan_id == plan.id),
    )).scalar_one()
    assert stat.vendor_license_holders == 4
    assert stat.holders_matched == 1
    assert stat.holders_external == 1
    assert stat.holders_leaver_residual == 1
    assert stat.holders_vendor_inactive == 1
    # off_paid_site recorded but NOT counted in vendor_license_holders.
    assert stat.holders_off_paid_site == 1
    assert stat.holders_no_email == 0


# ── ⑥ #1665: transaction-aborting DB error must re-raise, not be swallowed ──


class _RaisingExecuteSession:
    """Thin proxy over an AsyncSession that raises a chosen error on the
    first ``execute`` call, then delegates. Used to inject a failure inside
    ``record_seat_reconciliation``'s SAVEPOINT block."""

    def __init__(self, real, exc):
        self._real = real
        self._exc = exc
        self._fired = False

    async def execute(self, *a, **kw):
        if not self._fired:
            self._fired = True
            raise self._exc
        return await self._real.execute(*a, **kw)

    def begin_nested(self):
        return self._real.begin_nested()

    def __getattr__(self, name):
        return getattr(self._real, name)


def _integrity_error(errno):
    from sqlalchemy.exc import IntegrityError
    orig = Exception("db failure")
    orig.args = (errno, "err")
    return IntegrityError("stmt", {}, orig)


@pytest.mark.asyncio
async def test_recon_reraises_transaction_aborting_error(db_session: AsyncSession):
    """A lock-wait/deadlock errno (1213) aborts the whole tx server-side; the
    SAVEPOINT can't rescue it. Swallowing it would leave the handler's session
    pending-rollback so its later commit silently fails — it must re-raise."""
    integ = await _integration(db_session)
    plan = await _plan(db_session)
    await db_session.flush()

    proxy = _RaisingExecuteSession(db_session, _integrity_error(1213))
    with pytest.raises(Exception) as ei:
        await record_seat_reconciliation(
            proxy,
            holder_model=SoftwareExternalHolder,
            stat_model=SoftwareSeatSyncStat,
            integration_id=integ.id,
            software_id=plan.software_id,
            plan_id=plan.id,
            provider="jira",
            product_key="jira-software",
            site=None,
            vendor_license_holders=1,
            holders_matched=0,
            holders_no_email=0,
            externals=[ExternalHolderObservation(email="x@partner.com")],
            now=datetime.now(UTC),
        )
    from app.core.exceptions import is_transaction_aborting_db_error
    assert is_transaction_aborting_db_error(ei.value)
    await db_session.rollback()


@pytest.mark.asyncio
async def test_recon_swallows_non_aborting_snapshot_error(db_session: AsyncSession):
    """A non-transaction-aborting statement error (e.g. a plain unique-collision
    errno 1062) stays best-effort: the SAVEPOINT rolls back just the snapshot
    block, the helper logs + returns 0, and the caller session stays usable."""
    integ = await _integration(db_session)
    plan = await _plan(db_session)
    await db_session.flush()

    proxy = _RaisingExecuteSession(db_session, _integrity_error(1062))
    n = await record_seat_reconciliation(
        proxy,
        holder_model=SoftwareExternalHolder,
        stat_model=SoftwareSeatSyncStat,
        integration_id=integ.id,
        software_id=plan.software_id,
        plan_id=plan.id,
        provider="jira",
        product_key="jira-software",
        site=None,
        vendor_license_holders=1,
        holders_matched=0,
        holders_no_email=0,
        externals=[ExternalHolderObservation(email="y@partner.com")],
        now=datetime.now(UTC),
    )
    assert n == 0  # swallowed → reporting-only miss
    # Session still usable — a follow-up write commits cleanly.
    await db_session.commit()