"""Real PostgreSQL regressions for shared session state replacing Redis."""

import asyncio
from datetime import UTC, datetime, timedelta
from unittest.mock import MagicMock

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

from app.core import state_store as store
from app.models.expiring_state import ExpiringState


async def _expire_all(db_session: AsyncSession) -> None:
    await db_session.execute(
        update(ExpiringState).values(expires_at=datetime.now(UTC) - timedelta(seconds=1))
    )
    await db_session.commit()


async def test_refresh_rotation_logout_and_hashed_storage(db_session: AsyncSession) -> None:
    await store.store_refresh_token("user", "first-secret", 3600)
    assert await store.is_refresh_token_valid("user", "first-secret")
    assert not await store.is_refresh_token_valid("other-user", "first-secret")
    entry = await db_session.scalar(select(ExpiringState))
    assert entry is not None and "first-secret" not in str(entry.value)
    await store.store_refresh_token("user", "second-secret", 3600)
    assert not await store.is_refresh_token_valid("user", "first-secret")
    assert await store.is_refresh_token_valid("user", "second-secret")
    await store.revoke_refresh_token("user")
    assert not await store.is_refresh_token_valid("user", "second-secret")


async def test_expiry_is_enforced_before_cleanup(db_session: AsyncSession) -> None:
    await store.store_refresh_token("user", "refresh", 300)
    await store.blacklist_access_token("access", 300)
    await store.store_client_token("link", {"target_id": "report"}, 300)
    await store.store_dashboard_cache("branch-1", {"total": 1}, 300)
    await store.store_password_reset_otp("a@example.com", "123456", "user", 300)
    await _expire_all(db_session)
    assert not await store.is_refresh_token_valid("user", "refresh")
    assert not await store.is_access_token_blacklisted("access")
    assert await store.get_client_token_payload("link") is None
    assert await store.get_dashboard_cache("branch-1") is None
    assert (await store.verify_password_reset_otp("a@example.com", "123456")).status == "expired"
    assert await db_session.scalar(select(ExpiringState.key).limit(1)) is not None
    await store.cleanup_expired_state()
    assert await db_session.scalar(select(ExpiringState.key).limit(1)) is None


async def test_revoked_access_token_is_shared_and_specific() -> None:
    await store.blacklist_access_token("revoked")
    assert await store.is_access_token_blacklisted("revoked")
    assert not await store.is_access_token_blacklisted("active")


async def test_client_links_preserve_payload_and_can_be_revoked(db_session: AsyncSession) -> None:
    payload = {"target_id": "report-123", "type": "report", "lab_id": "branch-1"}
    await store.store_client_token("private-link", payload)
    assert await store.get_client_token_payload("private-link") == payload
    assert await store.get_client_token_payload("other-link") is None
    entry = await db_session.scalar(select(ExpiringState))
    assert entry is not None and "private-link" not in entry.key
    await store.revoke_client_token("private-link")
    assert await store.get_client_token_payload("private-link") is None


async def test_dashboard_cache_keys_remain_isolated() -> None:
    await store.store_dashboard_cache("branch-1", {"total": 500})
    await store.store_dashboard_cache("branch-2", {"total": 20})
    assert await store.get_dashboard_cache("branch-1") == {"total": 500}
    assert await store.get_dashboard_cache("branch-2") == {"total": 20}


async def test_password_code_is_hashed_normalized_and_single_use(db_session: AsyncSession) -> None:
    await store.store_password_reset_otp(" User@example.com ", "654321", "user-1")
    entry = await db_session.scalar(select(ExpiringState))
    assert entry is not None and "654321" not in str(entry.value)
    result = await store.verify_password_reset_otp("user@EXAMPLE.com", "654321")
    assert result.status == "valid" and result.user_id == "user-1"
    assert (await store.verify_password_reset_otp("user@example.com", "654321")).status == "expired"


async def test_failed_code_attempts_do_not_extend_expiry(db_session: AsyncSession) -> None:
    await store.store_password_reset_otp("user@example.com", "654321", "user")
    before = await db_session.scalar(select(ExpiringState.expires_at))
    for remaining in (4, 3, 2, 1):
        result = await store.verify_password_reset_otp("user@example.com", "000000")
        assert result.status == "invalid" and result.remaining_attempts == remaining
    after = await db_session.scalar(select(ExpiringState.expires_at))
    assert before == after
    assert (await store.verify_password_reset_otp("user@example.com", "000000")).status == "exhausted"
    assert (await store.verify_password_reset_otp("user@example.com", "654321")).status == "expired"


async def test_concurrent_code_consumption_has_one_winner() -> None:
    await store.store_password_reset_otp("user@example.com", "123456", "user")
    results = await asyncio.gather(
        *[store.verify_password_reset_otp("user@example.com", "123456") for _ in range(8)]
    )
    assert sum(result.status == "valid" for result in results) == 1


async def test_concurrent_wrong_codes_exhaust_shared_limit() -> None:
    await store.store_password_reset_otp("user@example.com", "123456", "user")
    results = await asyncio.gather(
        *[store.verify_password_reset_otp("user@example.com", "000000") for _ in range(8)]
    )
    assert sum(result.status == "invalid" for result in results) == 4
    assert sum(result.status == "exhausted" for result in results) == 1
    assert (await store.verify_password_reset_otp("user@example.com", "123456")).status == "expired"


async def test_concurrent_rate_limit_is_atomic_and_resets_after_expiry(db_session: AsyncSession) -> None:
    allowed = await asyncio.gather(
        *[store.check_forgot_password_rate_limit("USER@example.com", max_requests=5) for _ in range(12)]
    )
    assert sum(allowed) == 5
    assert not await store.check_forgot_password_rate_limit(" user@example.com ")
    assert await store.check_forgot_password_rate_limit("other@example.com")
    await _expire_all(db_session)
    assert await store.check_forgot_password_rate_limit("user@example.com")


async def test_security_state_fails_closed_when_database_is_down(monkeypatch: pytest.MonkeyPatch) -> None:
    unavailable = MagicMock()
    unavailable.side_effect = RuntimeError("database unavailable")
    unavailable.begin.side_effect = RuntimeError("database unavailable")
    monkeypatch.setattr(store, "AsyncSessionLocal", unavailable)
    with pytest.raises(RuntimeError, match="database unavailable"):
        await store.is_access_token_blacklisted("token")
    with pytest.raises(RuntimeError, match="database unavailable"):
        await store.blacklist_access_token("token")
    with pytest.raises(RuntimeError, match="database unavailable"):
        await store.check_forgot_password_rate_limit("user@example.com")
    # Dashboard cache remains optional; it cannot authorize access.
    assert await store.get_dashboard_cache("branch") is None