"""B-88 (DAT-52, DAT-74, ARC-07, TX-28, BET-11, DAT-16): SQLite transactions are the
ones the code assumes.

Everything here runs on a real file database whose engine went through
`_register_sqlite_pragmas` — an in-memory engine without it would not exercise
`BEGIN`/`BEGIN IMMEDIATE`, `synchronous` or the lock at all — with the lock wait
shortened to 100 ms so a busy database is provoked by simply holding the write
lock from a second session.
"""

from contextlib import asynccontextmanager
from types import SimpleNamespace

import pytest
from fastapi import Response
from cryptography.fernet import Fernet
from httpx import ASGITransport, AsyncClient
from sqlalchemy import func, select, text
from sqlalchemy.exc import OperationalError
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.bets.service import BetError, place_bet
from app.config import settings
from tests.unit.fake_network import FakeNetwork, rpc_error
from app.db import base as db_base
from app.db.base import Base, begin_immediate
from app.db.models import (
    AuditLog,
    PendingTransaction,
    RoundConfig,
    RoundParticipant,
    User,
    UtxoEvent,
    Withdrawal,
)
from app.deposits.service import (
    credit_confirmed_utxos,
    mark_utxos_spent_externally,
    reinstate_reappeared_utxos,
)
from app.wallet.hd import derive_user_address
from app.withdrawals.service import WithdrawalError, request_withdrawal

_FEE_ADDRESS = "plm1q5x25wd6q463mfhckjraaedgjg0lyu73qfcj43n"
_EXTERNAL_ADDRESS = "plm1qqph9qup2mp7w7g5nlsdhdc9m2pp44ampzw0ctx"


@pytest.fixture
async def engine(tmp_path, monkeypatch):
    monkeypatch.setattr(db_base, "_SQLITE_BUSY_TIMEOUT_MS", 100)
    eng = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/b88.db")
    db_base._register_sqlite_pragmas(eng)
    async with eng.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield eng
    await eng.dispose()


@pytest.fixture
def session_factory(engine):
    return async_sessionmaker(engine, expire_on_commit=False)


@pytest.fixture
def keyed(tmp_path, monkeypatch):
    """A real master key, so the HD wallet can derive and sign."""
    monkeypatch.setattr(settings, "master_key_path", str(tmp_path / "master.xprv.enc"))
    monkeypatch.setattr(settings, "xprv_encryption_key", Fernet.generate_key().decode())
    from app.wallet import hd

    hd._account_key = None
    hd.generate_master_key()
    yield
    hd._account_key = None


@pytest.fixture
async def funded(session_factory, keyed):
    """Fee address + one user holding 15 PLM in a single UTXO."""
    async with session_factory() as session:
        session.add(RoundConfig(fee_address=_FEE_ADDRESS))
        user = User(username="u0", password_hash="x", derivation_index=0, address=derive_user_address(0))
        session.add(user)
        await session.commit()
        session.add(UtxoEvent(user_id=user.id, txid="00" * 32, vout=0, amount_sats=1_500_000_000, confirmed_height=100))
        await session.commit()
        user_id = user.id
    yield user_id


@asynccontextmanager
async def _holding_write_lock(session_factory):
    """Another writer holding SQLite's write lock: every begin_immediate elsewhere
    times out after the (shortened) busy_timeout."""
    async with session_factory() as holder:
        await begin_immediate(holder)
        try:
            yield
        finally:
            await holder.rollback()


class _Client:
    """Fake Electrum client. `on_broadcast` runs first (to grab the lock *after* the
    coins have gone out); `fail` makes the broadcast raise."""

    def __init__(self, on_broadcast=None, fail=False):
        self.broadcasted: list[str] = []
        self._on_broadcast = on_broadcast
        self._fail = fail

    async def broadcast(self, raw_tx_hex: str) -> str:
        self.broadcasted.append(raw_tx_hex)
        if self._on_broadcast is not None:
            await self._on_broadcast()
        if self._fail:
            raise rpc_error("mempool conflict")
        return "txid"


# --- the engine setup: ARC-07 / DAT-52 ---------------------------------------------


async def test_every_connection_is_full_sync_wal_with_no_driver_transactions(engine):
    """DAT-52: synchronous=FULL, WAL, busy_timeout — and the driver's implicit BEGIN
    switched off (isolation_level None), so the transaction is the one we emit."""
    async with engine.connect() as conn:
        assert (await conn.exec_driver_sql("PRAGMA synchronous")).scalar() == 2  # FULL
        assert (await conn.exec_driver_sql("PRAGMA journal_mode")).scalar().lower() == "wal"
        assert (await conn.exec_driver_sql("PRAGMA busy_timeout")).scalar() == 100
        raw = await conn.get_raw_connection()
        assert raw.driver_connection.isolation_level is None


async def test_a_read_starts_a_real_transaction_with_a_stable_snapshot(session_factory):
    """DAT-74: with the legacy driver mode a SELECT ran outside any transaction and
    saw whatever had been committed since the previous statement; now the first
    statement opens BEGIN and the reads that follow share one snapshot."""
    async with session_factory() as setup:
        setup.add(User(username="a", password_hash="x", derivation_index=0, address="a"))
        await setup.commit()

    async with session_factory() as reader, session_factory() as writer:
        assert await reader.scalar(select(func.count()).select_from(User)) == 1
        raw = await (await reader.connection()).get_raw_connection()
        assert raw.driver_connection.in_transaction

        writer.add(User(username="b", password_hash="x", derivation_index=1, address="b"))
        await writer.commit()

        assert await reader.scalar(select(func.count()).select_from(User)) == 1  # snapshot held
        await reader.commit()
        assert await reader.scalar(select(func.count()).select_from(User)) == 2


async def test_commit_and_rollback_still_work_without_the_drivers_transactions(session_factory):
    async with session_factory() as session:
        session.add(User(username="kept", password_hash="x", derivation_index=0, address="k"))
        await session.commit()
        session.add(User(username="gone", password_hash="x", derivation_index=1, address="g"))
        await session.flush()
        await session.rollback()
    async with session_factory() as session:
        names = (await session.scalars(select(User.username))).all()
    assert names == ["kept"]


# --- begin_immediate: DAT-74 ------------------------------------------------------


async def test_begin_immediate_takes_the_write_lock_before_any_read(session_factory):
    async with session_factory() as first, session_factory() as second:
        await begin_immediate(first)
        with pytest.raises(OperationalError, match="locked"):
            await begin_immediate(second)
        await first.rollback()
        await begin_immediate(second)  # free again once the holder is done


async def test_a_plain_reader_is_not_blocked_by_the_lock_holder(session_factory):
    """WAL: only writers queue; BEGIN IMMEDIATE must not turn every GET into one."""
    async with session_factory() as holder, session_factory() as reader:
        await begin_immediate(holder)
        assert await reader.scalar(select(func.count()).select_from(User)) == 0


async def test_begin_immediate_closes_the_readonly_transaction_that_preceded_it(session_factory):
    """The callers arrive with the authentication lookup already in a transaction."""
    async with session_factory() as session:
        await session.scalar(select(func.count()).select_from(User))
        assert session.in_transaction()
        await begin_immediate(session)
        assert session.in_transaction()
        raw = await (await session.connection()).get_raw_connection()
        assert raw.driver_connection.in_transaction
        # and it really holds the write lock
        async with session_factory() as other:
            with pytest.raises(OperationalError):
                await begin_immediate(other)


async def test_begin_immediate_keeps_loaded_objects_usable(session_factory, funded):
    async with session_factory() as session:
        user = await session.get(User, funded)
        await begin_immediate(session)
        assert user.username == "u0"  # no lazy load needed after the commit it did


async def test_begin_immediate_refuses_a_session_with_unflushed_changes(session_factory):
    async with session_factory() as session:
        session.add(User(username="x", password_hash="x", derivation_index=0, address="x"))
        with pytest.raises(RuntimeError, match="unflushed"):
            await begin_immediate(session)
        session.expunge_all()

    async with session_factory() as session:
        user = User(username="y", password_hash="x", derivation_index=1, address="y")
        session.add(user)
        await session.commit()
        user.username = "z"  # dirty
        with pytest.raises(RuntimeError, match="unflushed"):
            await begin_immediate(session)
        await session.rollback()


async def test_begin_immediate_is_harmless_on_an_engine_without_the_sqlite_setup():
    """The service tests run on plain in-memory engines: the option is never read."""
    eng = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with eng.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    async with async_sessionmaker(eng, expire_on_commit=False)() as session:
        await begin_immediate(session)
        session.add(User(username="a", password_hash="x", derivation_index=0, address="a"))
        await session.commit()
    await eng.dispose()


# --- place_bet: BET-11, DAT-16, TX-28 ---------------------------------------------


async def _bet_state(session_factory, user_id):
    async with session_factory() as session:
        participants = (await session.scalars(select(RoundParticipant))).all()
        pendings = (await session.scalars(select(PendingTransaction))).all()
        utxo = await session.scalar(select(UtxoEvent).where(UtxoEvent.user_id == user_id))
        return participants, pendings, utxo


async def test_place_bet_on_a_real_database_commits_and_promotes(session_factory, funded):
    client = _Client()
    async with session_factory() as session:
        user = await session.get(User, funded)
        participant = await place_bet(session, client, user)
    assert participant.status == "broadcast"
    participants, pendings, utxo = await _bet_state(session_factory, funded)
    assert [p.status for p in participants] == ["broadcast"]
    assert [p.status for p in pendings] == ["pending"]
    assert utxo.spent_txid == participants[0].bet_txid
    assert len(client.broadcasted) == 1


async def test_place_bet_answers_service_busy_when_the_lock_is_not_granted(session_factory, funded):
    """BET-11 / api.md `service_busy`: refused before anything is built or sent."""
    client = _Client()
    async with _holding_write_lock(session_factory):
        async with session_factory() as session:
            user = await session.get(User, funded)
            with pytest.raises(BetError) as excinfo:
                await place_bet(session, client, user)
    assert excinfo.value.code == "service_busy"
    assert client.broadcasted == []
    participants, pendings, utxo = await _bet_state(session_factory, funded)
    assert participants == [] and pendings == []
    assert utxo.spent_txid is None


async def test_place_bet_can_be_retried_after_a_busy_refusal(session_factory, funded):
    client = _Client()
    async with _holding_write_lock(session_factory):
        async with session_factory() as session:
            user = await session.get(User, funded)
            with pytest.raises(BetError):
                await place_bet(session, client, user)
    async with session_factory() as session:
        user = await session.get(User, funded)
        participant = await place_bet(session, client, user)
    assert participant.status == "broadcast"


async def test_place_bet_maps_a_lock_timeout_at_the_phase_one_commit_to_service_busy_not_round_closing(
    session_factory, funded
):
    """DAT-16: an OperationalError at the compare-and-set / commit is the lock wait
    expiring, never a snapshot conflict (SQLite has none) — so not `round_closing`.
    Rolled back, nothing sent."""
    client = _Client()
    async with session_factory() as session:
        user = await session.get(User, funded)
        real_commit = session.commit
        calls = {"n": 0}

        async def flaky_commit():
            calls["n"] += 1
            if calls["n"] == 1:  # the phase-1 commit (begin_immediate's own commit is a no-op here)
                raise OperationalError("COMMIT", {}, Exception("database is locked"))
            await real_commit()

        session.commit = flaky_commit
        with pytest.raises(BetError) as excinfo:
            await place_bet(session, client, user)
    assert excinfo.value.code == "service_busy"
    assert client.broadcasted == []
    participants, pendings, utxo = await _bet_state(session_factory, funded)
    assert participants == [] and pendings == [] and utxo.spent_txid is None


async def test_place_bet_after_a_broadcast_never_undoes_it_when_the_lock_is_lost(session_factory, funded):
    """TX-28 / DAT-74: the tx is on the network, so a busy database afterwards leaves
    the rows `building` for the reconciler instead of failing or releasing anything;
    the caller still gets the bet, with the status the database holds."""
    holder_cm = _holding_write_lock(session_factory)

    async def grab_lock():
        await holder_cm.__aenter__()

    client = _Client(on_broadcast=grab_lock)
    try:
        async with session_factory() as session:
            user = await session.get(User, funded)
            participant = await place_bet(session, client, user)
    finally:
        await holder_cm.__aexit__(None, None, None)

    assert participant.status == "building"
    assert len(client.broadcasted) == 1
    participants, pendings, utxo = await _bet_state(session_factory, funded)
    assert [p.status for p in participants] == ["building"]
    assert [p.status for p in pendings] == ["building"]
    assert utxo.spent_txid == participants[0].bet_txid  # still reserved
    async with session_factory() as session:
        events = (await session.scalars(select(AuditLog.event_type))).all()
    assert "bet_placed" not in events and "bet_broadcast_failed" not in events


async def test_place_bet_refused_broadcast_with_a_busy_database_leaves_the_rows_building(session_factory, funded):
    """A refused broadcast whose release cannot be written stays `building` (the
    reconciler decides), and the caller still gets `broadcast_failed`."""
    holder_cm = _holding_write_lock(session_factory)

    async def grab_lock():
        await holder_cm.__aenter__()

    client = _Client(on_broadcast=grab_lock, fail=True)
    try:
        async with session_factory() as session:
            user = await session.get(User, funded)
            with pytest.raises(BetError) as excinfo:
                await place_bet(session, client, user, network=FakeNetwork(known=False))
    finally:
        await holder_cm.__aexit__(None, None, None)

    assert excinfo.value.code == "broadcast_failed"
    participants, pendings, utxo = await _bet_state(session_factory, funded)
    assert [p.status for p in participants] == ["building"]
    assert [p.status for p in pendings] == ["building"]
    assert utxo.spent_txid == participants[0].bet_txid


async def test_place_bet_refused_broadcast_still_releases_when_the_database_is_free(session_factory, funded):
    client = _Client(fail=True)
    async with session_factory() as session:
        user = await session.get(User, funded)
        with pytest.raises(BetError) as excinfo:
            await place_bet(session, client, user, network=FakeNetwork(known=False))
    assert excinfo.value.code == "broadcast_failed"
    participants, pendings, utxo = await _bet_state(session_factory, funded)
    assert participants == [] and pendings == [] and utxo.spent_txid is None


async def test_concurrent_bets_of_one_user_cannot_both_select_the_same_utxo(session_factory, funded):
    """DAT-74: with the lock taken before the reads, two place_bet calls for the same
    user (which the per-user lock normally forbids) cannot both read the UTXO as free."""
    import asyncio

    client = _Client()

    async def attempt():
        async with session_factory() as session:
            user = await session.get(User, funded)
            return await place_bet(session, client, user)

    results = await asyncio.gather(attempt(), attempt(), return_exceptions=True)
    ok = [r for r in results if not isinstance(r, Exception)]
    assert len(ok) == 1
    failures = [r for r in results if isinstance(r, Exception)]
    assert len(failures) == 1 and isinstance(failures[0], BetError)
    assert len(client.broadcasted) == 1


# --- request_withdrawal ----------------------------------------------------------


async def _withdrawal_state(session_factory, user_id):
    async with session_factory() as session:
        return (
            (await session.scalars(select(Withdrawal))).all(),
            (await session.scalars(select(PendingTransaction))).all(),
            await session.scalar(select(UtxoEvent).where(UtxoEvent.user_id == user_id)),
        )


async def test_withdrawal_on_a_real_database_commits_and_promotes(session_factory, funded):
    client = _Client()
    async with session_factory() as session:
        user = await session.get(User, funded)
        withdrawal = await request_withdrawal(session, client, user, _EXTERNAL_ADDRESS, 1_000_000_000)
    assert withdrawal.status == "broadcast"
    withdrawals, pendings, utxo = await _withdrawal_state(session_factory, funded)
    assert [w.status for w in withdrawals] == ["broadcast"]
    assert [p.status for p in pendings] == ["pending"]
    assert utxo.spent_txid == withdrawals[0].txid


async def test_withdrawal_answers_service_busy_when_the_lock_is_not_granted(session_factory, funded):
    client = _Client()
    async with _holding_write_lock(session_factory):
        async with session_factory() as session:
            user = await session.get(User, funded)
            with pytest.raises(WithdrawalError) as excinfo:
                await request_withdrawal(session, client, user, _EXTERNAL_ADDRESS, 1_000_000_000)
    assert excinfo.value.code == "service_busy"
    assert client.broadcasted == []
    withdrawals, pendings, utxo = await _withdrawal_state(session_factory, funded)
    assert withdrawals == [] and pendings == [] and utxo.spent_txid is None


async def test_withdrawal_validation_errors_do_not_wait_for_the_lock(session_factory, funded):
    """The checks that need no database answer immediately, lock or not."""
    client = _Client()
    async with _holding_write_lock(session_factory):
        async with session_factory() as session:
            user = await session.get(User, funded)
            with pytest.raises(WithdrawalError) as excinfo:
                await request_withdrawal(session, client, user, "not-an-address", 1_000_000_000)
    assert excinfo.value.code == "invalid_address"


async def test_withdrawal_after_a_broadcast_never_undoes_it_when_the_lock_is_lost(session_factory, funded):
    holder_cm = _holding_write_lock(session_factory)

    async def grab_lock():
        await holder_cm.__aenter__()

    client = _Client(on_broadcast=grab_lock)
    try:
        async with session_factory() as session:
            user = await session.get(User, funded)
            withdrawal = await request_withdrawal(session, client, user, _EXTERNAL_ADDRESS, 1_000_000_000)
    finally:
        await holder_cm.__aexit__(None, None, None)

    assert withdrawal.status == "building"
    withdrawals, pendings, utxo = await _withdrawal_state(session_factory, funded)
    assert [w.status for w in withdrawals] == ["building"]
    assert [p.status for p in pendings] == ["building"]
    assert utxo.spent_txid == withdrawals[0].txid


async def test_withdrawal_refused_broadcast_with_a_busy_database_leaves_the_rows_building(session_factory, funded):
    holder_cm = _holding_write_lock(session_factory)

    async def grab_lock():
        await holder_cm.__aenter__()

    client = _Client(on_broadcast=grab_lock, fail=True)
    try:
        async with session_factory() as session:
            user = await session.get(User, funded)
            with pytest.raises(WithdrawalError) as excinfo:
                await request_withdrawal(
                    session, client, user, _EXTERNAL_ADDRESS, 1_000_000_000, network=FakeNetwork(known=False)
                )
    finally:
        await holder_cm.__aexit__(None, None, None)

    assert excinfo.value.code == "broadcast_failed"
    withdrawals, pendings, utxo = await _withdrawal_state(session_factory, funded)
    assert [w.status for w in withdrawals] == ["building"]
    assert [p.status for p in pendings] == ["building"]
    assert utxo.spent_txid == withdrawals[0].txid


async def test_withdrawal_refused_broadcast_still_releases_when_the_database_is_free(session_factory, funded):
    client = _Client(fail=True)
    async with session_factory() as session:
        user = await session.get(User, funded)
        with pytest.raises(WithdrawalError):
            await request_withdrawal(
                session, client, user, _EXTERNAL_ADDRESS, 1_000_000_000, network=FakeNetwork(known=False)
            )
    withdrawals, pendings, utxo = await _withdrawal_state(session_factory, funded)
    assert [w.status for w in withdrawals] == ["failed"]
    assert [p.status for p in pendings] == ["failed"]
    assert utxo.spent_txid is None


# --- the HTTP mapping: api.md `service_busy` -> 503 ---------------------------------


def _fake_request(client):
    listener = SimpleNamespace(client=client)

    class _Locks:
        @asynccontextmanager
        async def acquire(self, user_id):
            yield

    return SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(electrum_listener=listener, user_locks=_Locks())))


async def test_bet_route_answers_503_for_service_busy(session_factory, funded):
    from fastapi import HTTPException

    from app.api.routes.bets import create_bet

    async with _holding_write_lock(session_factory):
        async with session_factory() as session:
            user = await session.get(User, funded)
            with pytest.raises(HTTPException) as excinfo:
                await create_bet(_fake_request(_Client()), Response(), user=user, session=session)
    assert excinfo.value.status_code == 503
    assert excinfo.value.detail["code"] == "service_busy"


async def test_withdrawal_route_answers_503_for_service_busy(session_factory, funded):
    from fastapi import HTTPException

    from app.api.routes.withdrawals import WithdrawalRequest, create_withdrawal

    body = WithdrawalRequest(external_address=_EXTERNAL_ADDRESS, amount_sats=1_000_000_000)
    async with _holding_write_lock(session_factory):
        async with session_factory() as session:
            user = await session.get(User, funded)
            with pytest.raises(HTTPException) as excinfo:
                await create_withdrawal(body, _fake_request(_Client()), Response(), user=user, session=session)
    assert excinfo.value.status_code == 503
    assert excinfo.value.detail["code"] == "service_busy"


# --- registration ----------------------------------------------------------------


@pytest.fixture
async def auth_client(session_factory, funded, monkeypatch):
    from fastapi import FastAPI

    from app.auth.routes import router
    from app.db.session import get_session

    monkeypatch.setattr(settings, "jwt_secret", "x" * 32)
    app = FastAPI()
    app.include_router(router)

    async def _session():
        async with session_factory() as session:
            yield session

    app.dependency_overrides[get_session] = _session
    app.state.electrum_listener = SimpleNamespace(address_for_new_user=lambda *a, **k: None)
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://t") as ac:
        yield ac


async def test_concurrent_registrations_get_distinct_derivation_indexes(auth_client, session_factory):
    """DAT-74: the max(derivation_index) read and the insert are one transaction under
    the write lock, so two simultaneous signups cannot pick the same index."""
    import asyncio

    responses = await asyncio.gather(
        *(auth_client.post("/auth/register", json={"username": f"racer{i}", "password": "long-enough-1"}) for i in range(4))
    )
    assert [r.status_code for r in responses] == [201] * 4
    async with session_factory() as session:
        indexes = (await session.scalars(select(User.derivation_index))).all()
    assert len(indexes) == len(set(indexes)) == 5  # the funded fixture user plus four


async def test_registration_answers_503_service_busy_when_the_lock_is_not_granted(auth_client, session_factory):
    async with _holding_write_lock(session_factory):
        resp = await auth_client.post("/auth/register", json={"username": "late", "password": "long-enough-1"})
    assert resp.status_code == 503
    assert resp.json()["detail"]["code"] == "service_busy"
    async with session_factory() as session:
        assert await session.scalar(select(User).where(User.username == "late")) is None


# --- deposits --------------------------------------------------------------------


async def test_credit_confirmed_utxos_needs_the_write_lock(session_factory, funded):
    entries = [{"tx_hash": "ab" * 32, "tx_pos": 0, "height": 100, "value": 5_000_000}]
    async with _holding_write_lock(session_factory):
        async with session_factory() as session:
            with pytest.raises(OperationalError):
                await credit_confirmed_utxos(session, funded, entries)
    async with session_factory() as session:
        assert await credit_confirmed_utxos(session, funded, entries) == 1


async def test_mark_utxos_spent_externally_needs_the_write_lock(session_factory, funded):
    async with session_factory() as session:
        utxo_id = (await session.scalar(select(UtxoEvent.id).where(UtxoEvent.user_id == funded)))
    async with _holding_write_lock(session_factory):
        async with session_factory() as session:
            with pytest.raises(OperationalError):
                await mark_utxos_spent_externally(session, funded, [utxo_id])
    async with session_factory() as session:
        assert await mark_utxos_spent_externally(session, funded, [utxo_id]) == 1


async def test_reinstate_takes_the_lock_only_when_there_is_something_to_reinstate(session_factory, funded):
    """It runs on every refresh: with nothing flagged it must not queue for the lock."""
    async with session_factory() as session:
        utxo = await session.scalar(select(UtxoEvent).where(UtxoEvent.user_id == funded))
        utxo.spent_txid = "external-spend"
        await session.commit()
        entry = {"tx_hash": utxo.txid, "tx_pos": utxo.vout, "height": 100, "value": utxo.amount_sats}

    async with _holding_write_lock(session_factory):
        async with session_factory() as session:
            # nothing flagged reappears -> no lock needed, no error
            assert await reinstate_reappeared_utxos(session, funded, []) == 0
        async with session_factory() as session:
            with pytest.raises(OperationalError):
                await reinstate_reappeared_utxos(session, funded, [entry])

    async with session_factory() as session:
        assert await reinstate_reappeared_utxos(session, funded, [entry]) == 1
        utxo = await session.scalar(select(UtxoEvent).where(UtxoEvent.user_id == funded))
        assert utxo.spent_txid is None


# --- background writers: scheduler, poller, reconciler, RBF -------------------------


async def test_scheduler_tick_state_transition_needs_the_write_lock_and_is_retried(session_factory, funded):
    """DAT-74: a state transition under a busy database raises out of the tick (run()
    logs it and tries again five seconds later) instead of half-writing."""
    from datetime import datetime, timedelta, timezone

    from app.db.models import Round
    from app.rounds.scheduler import RoundScheduler

    async with session_factory() as session:
        session.add(Round(status="open", opened_at=datetime.now(timezone.utc) - timedelta(seconds=99999), duration_seconds=1))
        await session.commit()

    scheduler = RoundScheduler(session_factory, SimpleNamespace(client=object(), tip_height=1, tip_header_hex="00"))
    async with _holding_write_lock(session_factory):
        with pytest.raises(OperationalError):
            await scheduler._tick()
    async with session_factory() as session:
        assert (await session.scalar(select(Round.status))) == "open"  # untouched

    await scheduler._tick()  # lock free: the same tick now goes through
    async with session_factory() as session:
        assert (await session.scalar(select(Round.status))) == "closed"  # no participants


def _payout_scheduler(session_factory, client):
    from app.rounds.scheduler import RoundScheduler
    from tests.unit.test_scheduler import FakePayoutListener

    listener = FakePayoutListener(client)
    # B-82: the network phase of a payout is the quorum's verification of the round's coins,
    # so that is where a test injects "the lock is lost while the network is being used".
    listener.on_verify = getattr(client, "_hook_listunspent", None)
    return RoundScheduler(session_factory, listener)


class _PayoutClient:
    def __init__(self, entries, hook_listunspent=None, hook_broadcast=None):
        self._entries = entries
        self._hook_listunspent = hook_listunspent
        self._hook_broadcast = hook_broadcast
        self.broadcasted: list[str] = []

    async def listunspent(self, scripthash):
        if self._hook_listunspent:
            await self._hook_listunspent()
        return self._entries

    async def broadcast(self, raw_tx_hex):
        self.broadcasted.append(raw_tx_hex)
        if self._hook_broadcast:
            await self._hook_broadcast()
        return "txid"


async def _seed_payout(session_factory):
    from tests.unit.test_scheduler import _POOL_AMOUNT_SATS, _seed_paying_out_round

    await _seed_paying_out_round(session_factory)
    return [{"tx_hash": "33" * 32, "tx_pos": 0, "height": 10, "value": _POOL_AMOUNT_SATS + 100_000}]


async def test_trigger_payout_defers_when_the_database_is_busy_before_reading(session_factory, keyed):
    from app.db.models import Round

    entries = await _seed_payout(session_factory)
    client = _PayoutClient(entries)
    async with _holding_write_lock(session_factory):
        await _payout_scheduler(session_factory, client)._trigger_payout(1)
    assert client.broadcasted == []
    async with session_factory() as session:
        assert (await session.scalars(select(PendingTransaction))).all() == []
        assert (await session.get(Round, 1)).payout_txid is None


async def test_trigger_payout_does_not_broadcast_when_it_cannot_persist_the_intent(session_factory, keyed):
    """The lock is lost between building and persisting: nothing may be sent."""
    entries = await _seed_payout(session_factory)
    holder_cm = _holding_write_lock(session_factory)

    async def grab():
        await holder_cm.__aenter__()

    client = _PayoutClient(entries, hook_listunspent=grab)
    try:
        await _payout_scheduler(session_factory, client)._trigger_payout(1)
    finally:
        await holder_cm.__aexit__(None, None, None)
    assert client.broadcasted == []
    async with session_factory() as session:
        assert (await session.scalars(select(PendingTransaction))).all() == []


async def test_trigger_payout_leaves_the_row_building_when_the_lock_is_lost_after_the_broadcast(
    session_factory, keyed
):
    entries = await _seed_payout(session_factory)
    holder_cm = _holding_write_lock(session_factory)

    async def grab():
        await holder_cm.__aenter__()

    client = _PayoutClient(entries, hook_broadcast=grab)
    try:
        await _payout_scheduler(session_factory, client)._trigger_payout(1)
    finally:
        await holder_cm.__aexit__(None, None, None)
    assert len(client.broadcasted) == 1
    async with session_factory() as session:
        pending = (await session.scalars(select(PendingTransaction))).one()
        assert pending.status == "building"  # the reconciler promotes it (TX-28)
        events = (await session.scalars(select(AuditLog.event_type))).all()
        assert "payout_sent" not in events


async def _seed_pending_withdrawal(session_factory, funded, status="pending", **kw):
    from datetime import datetime, timedelta, timezone

    async with session_factory() as session:
        wd = Withdrawal(user_id=funded, external_address=_EXTERNAL_ADDRESS, amount_requested_sats=1, amount_sent_sats=1, txid="abcd", status="broadcast")
        session.add(wd)
        await session.flush()
        row = PendingTransaction(
            kind="withdrawal",
            withdrawal_id=wd.id,
            user_id=funded,
            current_txid="abcd",
            fee_rate_sat_vb=1,
            raw_tx_hex="00",
            status=status,
            broadcast_at=datetime.now(timezone.utc) - timedelta(days=1),
            **kw,
        )
        session.add(row)
        await session.commit()
        return row.id


async def test_confirmation_poller_writes_under_the_lock(session_factory, funded):
    from app.tx.confirmation import poll_once
    import app.withdrawals.confirmation  # noqa: F401  (registers the handler)

    row_id = await _seed_pending_withdrawal(session_factory, funded)

    class Client:
        async def get_history(self, scripthash):
            return [{"tx_hash": "abcd", "height": 7}]

    async with _holding_write_lock(session_factory):
        with pytest.raises(OperationalError):
            await poll_once(session_factory, Client(), FakeNetwork())
    async with session_factory() as session:
        assert (await session.get(PendingTransaction, row_id)).status == "pending"

    assert await poll_once(session_factory, Client(), FakeNetwork()) == 1
    async with session_factory() as session:
        assert (await session.get(PendingTransaction, row_id)).status == "confirmed"
        assert (await session.scalar(select(Withdrawal.status))) == "confirmed"


async def test_reconciler_resolves_under_the_lock(session_factory, funded):
    from app.tx.reconcile import reconcile_once

    row_id = await _seed_pending_withdrawal(session_factory, funded, status="building")

    class Client:
        async def get_history(self, scripthash):
            return [{"tx_hash": "abcd", "height": 0}]

    async with _holding_write_lock(session_factory):
        with pytest.raises(OperationalError):
            await reconcile_once(session_factory, Client())
    async with session_factory() as session:
        assert (await session.get(PendingTransaction, row_id)).status == "building"

    assert await reconcile_once(session_factory, Client()) == 1
    async with session_factory() as session:
        assert (await session.get(PendingTransaction, row_id)).status == "pending"


async def test_bump_fee_persists_under_the_lock(session_factory, funded):
    """The persist phase of an RBF bump is a money-moving transaction like any other
    (DAT-74): under a busy database it raises out of bump_fee rather than committing
    a half-applied retarget."""
    from datetime import datetime, timedelta, timezone

    from embit import script
    from embit.transaction import Transaction, TransactionInput, TransactionOutput

    from app.tx.broadcast import bump_fee
    from app.wallet.hd import derive_user_key
    from app.wallet.psbt_builder import Utxo, build_signed_transaction

    signer = derive_user_key(0)
    built = build_signed_transaction(
        signing_key=signer,
        from_script=script.p2wpkh(signer.to_public()),
        utxos=[Utxo("11" * 32, 0, 150_000_000)],
        to_address=_EXTERNAL_ADDRESS,
        amount_sats=10_000_000,
        change_address=derive_user_address(0),
        fee_rate_sat_vb=1,
    )
    async with session_factory() as session:
        session.add(
            PendingTransaction(
                kind="bet",
                user_id=funded,
                current_txid=built.txid,
                fee_rate_sat_vb=1,
                raw_tx_hex=built.raw_hex,
                status="pending",
                broadcast_at=datetime.now(timezone.utc) - timedelta(seconds=1000),
            )
        )
        await session.commit()

    class Client:
        broadcasted: list = []

        async def get_transaction(self, txid, verbose=False):
            tx = Transaction(
                vin=[TransactionInput(b"\x00" * 32, 0)],
                vout=[TransactionOutput(150_000_000, script.Script(b"\x00\x14" + b"\x00" * 20))],
            )
            return tx.serialize().hex()

        async def broadcast(self, raw_hex):
            self.broadcasted.append(raw_hex)
            return "x"

    holder_cm = _holding_write_lock(session_factory)

    class HookedClient(Client):
        async def broadcast(self, raw_hex):
            await super().broadcast(raw_hex)
            await holder_cm.__aenter__()  # the lock is taken right after the send

    client = HookedClient()
    try:
        with pytest.raises(OperationalError):
            await bump_fee(session_factory, client, 1)
    finally:
        await holder_cm.__aexit__(None, None, None)
    async with session_factory() as session:
        row = await session.get(PendingTransaction, 1)
        assert row.current_txid == built.txid  # nothing half-applied
    assert len(client.broadcasted) == 1