"""Point-in-time holding ledger — the two tables that used to lose history.

``auth_user_roles`` revokes by DELETE and ``user_permissions`` resets its
row in place under ``UNIQUE(user_id, permission_id)``, so before migration
483 neither could answer "did X hold this on date D". These tests assert
the ledger written by ``app.core.access_cycle_listener`` answers it, and —
more importantly — that a *closed* holding stops answering yes.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

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

from app.auth.models import AuthRole, AuthUserRole, AuthUserRoleCycle
from app.core.access_cycle_listener import register_listener
from app.domains.asset.models import AssetSoftwareCatalog
from app.domains.permission.models import (
    PERM_ACTIVE,
    PERM_PENDING,
    PERM_REVOKED,
    SoftwarePermission,
    UserPermission,
    UserPermissionCycle,
)

# The listener is attached in ``app.main``'s lifespan, which only runs for
# tests that take the ``client`` fixture. These tests drive the ORM
# directly, so attach it here; registration is idempotent and global.
register_listener()


async def _mk_permission(db: AsyncSession) -> SoftwarePermission:
    software = AssetSoftwareCatalog(name=f"Ledger SW {uuid.uuid4().hex[:6]}")
    db.add(software)
    await db.flush()
    perm = SoftwarePermission(
        software_id=software.id,
        permission_key=f"perm_{uuid.uuid4().hex[:8]}",
        display_name="Ledger test permission",
    )
    db.add(perm)
    await db.flush()
    return perm


async def _perm_cycles(
    db: AsyncSession, user_permission_id: uuid.UUID,
) -> list[UserPermissionCycle]:
    # The listener writes through Core DML on the flush's connection, so
    # ledger rows already in this session's identity map are stale after
    # a later transition. Real readers open their own session;
    # ``populate_existing`` refreshes just these rows (``expire_all()``
    # would also expire the test's person fixtures and trigger a lazy
    # refresh outside the greenlet).
    rows = await db.execute(
        select(UserPermissionCycle)
        .where(UserPermissionCycle.user_permission_id == user_permission_id)
        .order_by(UserPermissionCycle.opened_at)
        .execution_options(populate_existing=True)
    )
    return list(rows.scalars().all())


async def _role_cycles(
    db: AsyncSession, user_id: uuid.UUID, role_id: uuid.UUID,
) -> list[AuthUserRoleCycle]:
    rows = await db.execute(
        select(AuthUserRoleCycle)
        .where(
            AuthUserRoleCycle.user_id == user_id,
            AuthUserRoleCycle.role_id == role_id,
        )
        .order_by(AuthUserRoleCycle.granted_at)
        .execution_options(populate_existing=True)
    )
    return list(rows.scalars().all())


async def _held_permission_at(
    db: AsyncSession, permission_id: uuid.UUID, at: datetime,
) -> set[uuid.UUID]:
    """The evidence query this whole ledger exists to make possible."""
    rows = await db.execute(
        select(UserPermissionCycle.user_id).where(
            UserPermissionCycle.permission_id == permission_id,
            UserPermissionCycle.granted_at.is_not(None),
            UserPermissionCycle.granted_at <= at,
            (UserPermissionCycle.closed_at.is_(None))
            | (UserPermissionCycle.closed_at > at),
        )
    )
    return {row[0] for row in rows.all()}


@pytest.mark.asyncio
async def test_permission_pending_row_is_not_held_access(db_session, employee_user):
    """A request that never activated must never appear in a roster.

    The cycle opens on the PENDING row (so the request is auditable), but
    ``granted_at`` stays NULL until the provisioning task confirms the
    vendor-side grant. Counting a PENDING row as access is the failure
    mode that made the pre-2026-05-06 UI claim people had permissions the
    driver never granted.
    """
    employee, _ = employee_user
    perm = await _mk_permission(db_session)
    row = UserPermission(
        user_id=employee.id, permission_id=perm.id, status=PERM_PENDING,
    )
    db_session.add(row)
    await db_session.flush()

    cycles = await _perm_cycles(db_session, row.id)
    assert len(cycles) == 1
    assert cycles[0].open_status == PERM_PENDING
    assert cycles[0].granted_at is None
    assert cycles[0].closed_at is None

    now = datetime.now(UTC)
    assert await _held_permission_at(db_session, perm.id, now) == set()


@pytest.mark.asyncio
async def test_permission_grant_then_revoke_closes_the_interval(
    db_session, employee_user,
):
    employee, _ = employee_user
    perm = await _mk_permission(db_session)
    row = UserPermission(
        user_id=employee.id, permission_id=perm.id, status=PERM_PENDING,
    )
    db_session.add(row)
    await db_session.flush()

    granted_at = datetime.now(UTC) - timedelta(days=10)
    row.status = PERM_ACTIVE
    row.granted_at = granted_at
    await db_session.flush()

    during = granted_at + timedelta(days=1)
    assert employee.id in await _held_permission_at(db_session, perm.id, during)

    revoked_at = datetime.now(UTC) - timedelta(days=2)
    row.status = PERM_REVOKED
    row.revoked_at = revoked_at
    await db_session.flush()

    cycles = await _perm_cycles(db_session, row.id)
    assert len(cycles) == 1
    assert cycles[0].granted_at is not None
    assert cycles[0].closed_at is not None
    assert cycles[0].close_status == PERM_REVOKED

    # Held during the window, not held after it — the whole point.
    assert employee.id in await _held_permission_at(db_session, perm.id, during)
    assert employee.id not in await _held_permission_at(
        db_session, perm.id, datetime.now(UTC),
    )


@pytest.mark.asyncio
async def test_permission_resurrection_keeps_the_earlier_interval(
    db_session, employee_user,
):
    """The gap this ledger was built for.

    ``request_permission`` / ``admin_grant`` reuse the terminal row and
    overwrite ``granted_at`` — so after a re-request the live row shows
    only the new holding and the earlier one is gone. The ledger must
    keep both, with the first one closed.
    """
    employee, _ = employee_user
    perm = await _mk_permission(db_session)
    row = UserPermission(
        user_id=employee.id, permission_id=perm.id, status=PERM_PENDING,
    )
    db_session.add(row)
    await db_session.flush()

    first_granted = datetime.now(UTC) - timedelta(days=60)
    row.status = PERM_ACTIVE
    row.granted_at = first_granted
    await db_session.flush()

    first_revoked = datetime.now(UTC) - timedelta(days=40)
    row.status = PERM_REVOKED
    row.revoked_at = first_revoked
    await db_session.flush()

    # Re-request: the service resets the SAME row in place.
    row.status = PERM_PENDING
    row.granted_at = None
    row.revoked_at = None
    await db_session.flush()

    second_granted = datetime.now(UTC) - timedelta(days=5)
    row.status = PERM_ACTIVE
    row.granted_at = second_granted
    await db_session.flush()

    cycles = await _perm_cycles(db_session, row.id)
    assert len(cycles) == 2, "the earlier holding must survive the reset"
    first, second = cycles
    assert first.closed_at is not None
    assert second.closed_at is None

    inside_first = first_granted + timedelta(days=1)
    in_the_gap = first_revoked + timedelta(days=1)
    inside_second = second_granted + timedelta(days=1)

    assert employee.id in await _held_permission_at(db_session, perm.id, inside_first)
    assert employee.id not in await _held_permission_at(db_session, perm.id, in_the_gap)
    assert employee.id in await _held_permission_at(db_session, perm.id, inside_second)


@pytest.mark.asyncio
async def test_role_grant_and_orm_revoke_bound_the_interval(
    db_session, employee_user,
):
    employee, _ = employee_user
    role_id = (
        await db_session.execute(
            select(AuthRole.id).where(AuthRole.name == "auditor")
        )
    ).scalar_one()

    link = AuthUserRole(user_id=employee.id, role_id=role_id)
    db_session.add(link)
    await db_session.flush()

    cycles = await _role_cycles(db_session, employee.id, role_id)
    assert len(cycles) == 1
    assert cycles[0].revoked_at is None
    # Snapshot, not a live join — a later rename must not rewrite history.
    assert cycles[0].role_name == "auditor"

    await db_session.delete(link)
    await db_session.flush()

    cycles = await _role_cycles(db_session, employee.id, role_id)
    assert len(cycles) == 1
    assert cycles[0].revoked_at is not None, (
        "revoking a role deletes the auth_user_roles row; the ledger "
        "interval is the only surviving evidence it was ever held"
    )


@pytest.mark.asyncio
async def test_role_regrant_opens_a_second_interval(db_session, employee_user):
    employee, _ = employee_user
    role_id = (
        await db_session.execute(
            select(AuthRole.id).where(AuthRole.name == "auditor")
        )
    ).scalar_one()

    first = AuthUserRole(user_id=employee.id, role_id=role_id)
    db_session.add(first)
    await db_session.flush()
    await db_session.delete(first)
    await db_session.flush()

    # A fresh grant of the same role — the live table cannot tell this
    # apart from "held continuously since today".
    second = AuthUserRole(
        user_id=employee.id,
        role_id=role_id,
        granted_at=datetime.now(UTC) + timedelta(seconds=1),
    )
    db_session.add(second)
    await db_session.flush()

    cycles = await _role_cycles(db_session, employee.id, role_id)
    assert len(cycles) == 2
    assert cycles[0].revoked_at is not None
    assert cycles[1].revoked_at is None


@pytest.mark.asyncio
async def test_rolled_back_grant_leaves_no_cycle(db_session, employee_user):
    """A ledger row must never claim a grant the transaction discarded.

    The listener writes on the flush's own connection precisely so this
    holds; an independent connection would leave a ghost interval behind
    and inflate every roster that intersects it.
    """
    employee, _ = employee_user
    # Read the id up front: a rollback expires every ORM object in the
    # session, and touching an attribute afterwards lazy-loads outside
    # the greenlet.
    person_id = employee.id
    role_id = (
        await db_session.execute(
            select(AuthRole.id).where(AuthRole.name == "auditor")
        )
    ).scalar_one()

    # SAVEPOINT rather than an outer rollback so the discarded grant is
    # isolated from the fixtures this session created. ``after_soft_rollback``
    # is the hook that clears the listener's queue on this path.
    async with db_session.begin_nested() as savepoint:
        db_session.add(AuthUserRole(user_id=person_id, role_id=role_id))
        await db_session.flush()
        await savepoint.rollback()

    assert await _role_cycles(db_session, person_id, role_id) == []

    # And the queue must not survive to contaminate the NEXT flush — the
    # ghost-row class the identity audit listener was rewritten to stop.
    other_perm = await _mk_permission(db_session)
    row = UserPermission(
        user_id=person_id, permission_id=other_perm.id, status=PERM_PENDING,
    )
    db_session.add(row)
    await db_session.flush()
    assert await _role_cycles(db_session, person_id, role_id) == []


@pytest.mark.asyncio
async def test_close_and_open_in_one_flush_leaves_exactly_one_open(
    db_session, employee_user,
):
    """The resurrection reset closes and opens inside a SINGLE flush.

    This is why the listener issues Core DML in collection order rather
    than ``session.add()``: a deferred INSERT would land *after* the
    close statement ran, so the close would miss the cycle it was
    supposed to bound and the new interval would stay open forever —
    reading as "still holds it" for every date thereafter.
    """
    employee, _ = employee_user
    perm = await _mk_permission(db_session)
    row = UserPermission(
        user_id=employee.id, permission_id=perm.id, status=PERM_ACTIVE,
        granted_at=datetime.now(UTC) - timedelta(days=3),
    )
    db_session.add(row)
    await db_session.flush()

    row.status = PERM_REVOKED
    row.revoked_at = datetime.now(UTC)
    await db_session.flush()

    # One flush that both closes the lingering interval and opens a new
    # one on the same row.
    row.status = PERM_PENDING
    row.granted_at = None
    row.revoked_at = None
    await db_session.flush()

    cycles = await _perm_cycles(db_session, row.id)
    open_cycles = [c for c in cycles if c.closed_at is None]
    assert len(open_cycles) == 1, (
        "close must be applied before the new interval is inserted"
    )
    assert open_cycles[0].granted_at is None