"""B-96: the smaller state inconsistencies — WDR-15, BET-07 (pending-covered bet), TX-16, TX-18,
TX-27, TX-29/AUD-46, DEP-15/DEP-16 and BET-02/WDR-01."""

import asyncio
import json
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace

import pytest
from sqlalchemy import select

from app.bets.service import BetError, place_bet
from app.db.models import AuditLog, PendingTransaction, Round, RoundParticipant, User, UtxoEvent, Withdrawal
from app.rounds.events import broadcaster
from app.rounds.scheduler import CLOSING_STALL_THRESHOLD_SECONDS, RoundScheduler
from app.tx.reconcile import reconcile_once
from app.withdrawals.service import WithdrawalError, request_withdrawal
from tests.unit.fake_network import FakeNetwork
from tests.unit.test_reconcile import KnownTxClient, _seed_bet
from tests.unit.test_reconcile import session_factory as reconcile_session_factory  # noqa: F401  (fixture)
from tests.unit.test_scheduler import FakeListener, _seed_closing_round
from tests.unit.test_scheduler import session_factory as scheduler_session_factory  # noqa: F401  (fixture)
from tests.unit.test_withdrawals import (
    BET_AMOUNT_SATS,
    EXTERNAL_ADDRESS,
    FakeElectrumClient,
    _make_funded_user,
)
from tests.unit.test_withdrawals import session_factory as wallet_session_factory  # noqa: F401  (fixture)


# --- WDR-15 and the pending-covered bet ---------------------------------------------


async def test_a_reduced_full_balance_withdrawal_is_checked_against_the_minimum(wallet_session_factory):
    """WDR-15: moving the minimum from a one-coin balance leaves a dust limit behind, which puts
    the amount that actually goes out under the minimum."""
    session_factory = wallet_session_factory
    user_id = await _make_funded_user(session_factory, 50, BET_AMOUNT_SATS)
    client = FakeElectrumClient()
    async with session_factory() as session:
        user = await session.get(User, user_id)
        with pytest.raises(WithdrawalError) as excinfo:
            await request_withdrawal(session, client, user, EXTERNAL_ADDRESS, BET_AMOUNT_SATS)
    assert excinfo.value.code == "amount_below_minimum"
    assert excinfo.value.params == {"minimum_sats": BET_AMOUNT_SATS}
    assert client.broadcasted == []


async def test_a_bet_covered_only_by_unconfirmed_change_says_wait_not_insufficient(wallet_session_factory):
    """BET-07 / DEP-13: the balance on screen covers the bet, the confirmed one does not."""
    session_factory = wallet_session_factory
    user_id = await _make_funded_user(session_factory, 51, 3_000_000_000)
    async with session_factory() as session:
        user = await session.get(User, user_id)
        await request_withdrawal(session, FakeElectrumClient(), user, EXTERNAL_ADDRESS, 1_200_000_000)
    async with session_factory() as session:
        user = await session.get(User, user_id)
        assert user.cached_balance_sats == 0
        with pytest.raises(BetError) as excinfo:
            await place_bet(session, FakeElectrumClient(), user)
    assert excinfo.value.code == "balance_pending_confirmation"
    assert excinfo.value.params["pending_sats"] > 0


async def test_a_bet_with_no_money_at_all_is_still_insufficient(wallet_session_factory):
    session_factory = wallet_session_factory
    user_id = await _make_funded_user(session_factory, 52, 100_000_000)
    async with session_factory() as session:
        user = await session.get(User, user_id)
        with pytest.raises(BetError) as excinfo:
            await place_bet(session, FakeElectrumClient(), user)
    assert excinfo.value.code == "insufficient_balance"


# --- TX-16 / TX-18 / TX-27 -----------------------------------------------------------


async def _events(session_factory):
    async with session_factory() as session:
        return [e.event_type for e in (await session.scalars(select(AuditLog).order_by(AuditLog.id))).all()]


async def test_a_recovered_bet_gets_the_event_its_send_would_have_written(reconcile_session_factory):
    """TX-16 / AUD-30: pending_tx_recovered alone left a recovered bet out of the audit trail."""
    session_factory = reconcile_session_factory
    await _seed_bet(session_factory, pending_status="building", participant_status="building", age_seconds=300)
    await reconcile_once(session_factory, KnownTxClient("betxid"))
    assert sorted(await _events(session_factory)) == ["bet_placed", "pending_tx_recovered"]
    async with session_factory() as session:
        placed = (await session.scalars(select(AuditLog).where(AuditLog.event_type == "bet_placed"))).one()
        assert json.loads(placed.payload_json) == {"txid": "betxid", "amount_sats": 999_000_000}
        assert placed.user_id is not None and placed.round_id == 1


async def test_a_recovered_withdrawal_and_payout_get_theirs_too(reconcile_session_factory):
    from app.tx.reconcile import _promote

    session_factory = reconcile_session_factory
    async with session_factory() as session:
        withdrawal = Withdrawal(
            user_id=1, external_address=EXTERNAL_ADDRESS, amount_requested_sats=5, amount_sent_sats=4, txid="w", status="building"
        )
        round_ = Round(status="paying_out", winner_user_id=7, winner_amount_sats=700, fee_amount_sats=300)
        session.add_all([withdrawal, round_])
        await session.flush()
        w_row = PendingTransaction(kind="withdrawal", withdrawal_id=withdrawal.id, user_id=1, current_txid="w", fee_rate_sat_vb=1, raw_tx_hex="", status="building")
        p_row = PendingTransaction(kind="payout", round_id=round_.id, current_txid="p", fee_rate_sat_vb=1, raw_tx_hex="", status="building")
        session.add_all([w_row, p_row])
        await session.flush()
        await _promote(session, w_row)
        await _promote(session, p_row)
        await session.commit()
        rows = {e.event_type: e for e in (await session.scalars(select(AuditLog))).all()}
    assert json.loads(rows["withdrawal_sent"].payload_json) == {
        "txid": "w", "amount_sent_sats": 4, "external_address": EXTERNAL_ADDRESS,
    }
    assert json.loads(rows["payout_sent"].payload_json) == {"txid": "p", "winner_sats": 700, "commission_sats": 300}
    assert rows["payout_sent"].user_id == 7


async def test_an_abandoned_payout_leaves_no_amounts_behind(reconcile_session_factory):
    """TX-18 / DAT-22."""
    from app.tx.reconcile import _abandon

    session_factory = reconcile_session_factory
    async with session_factory() as session:
        round_ = Round(
            status="paying_out", payout_txid="p", winner_amount_sats=700, fee_amount_sats=300, bump_reserve_sats=50
        )
        session.add(round_)
        await session.flush()
        row = PendingTransaction(kind="payout", round_id=round_.id, current_txid="p", fee_rate_sat_vb=1, raw_tx_hex="", status="pending")
        session.add(row)
        await session.flush()
        await _abandon(session, row, "no server knows it")
        await session.commit()
        await session.refresh(round_)
        assert (round_.payout_txid, round_.winner_amount_sats, round_.fee_amount_sats, round_.bump_reserve_sats) == (
            None, None, None, None,
        )


async def test_the_reconciler_only_pings_the_dashboards_for_rows_it_changed(reconcile_session_factory, monkeypatch):
    """TX-27: a row found present and already `pending` is examined, not changed."""
    session_factory = reconcile_session_factory
    await _seed_bet(session_factory, pending_status="pending", participant_status="broadcast", age_seconds=30 * 3600)
    pings = []
    monkeypatch.setattr(broadcaster, "publish", lambda: pings.append(1))
    assert await reconcile_once(session_factory, KnownTxClient("betxid")) == 0
    assert pings == []


# --- TX-29 / AUD-46 ------------------------------------------------------------------


async def _stalled_rows(session_factory):
    async with session_factory() as session:
        return list(
            (await session.scalars(select(AuditLog).where(AuditLog.event_type == "closing_stalled").order_by(AuditLog.id))).all()
        )


async def test_closing_stalled_is_written_past_the_threshold_and_repeats_every_30_minutes(scheduler_session_factory):
    session_factory = scheduler_session_factory
    round_id = await _seed_closing_round(session_factory, n=2)
    async with session_factory() as session:
        for participant in (await session.scalars(select(RoundParticipant))).all():
            participant.status = "broadcast"
        await session.commit()
    scheduler = RoundScheduler(session_factory, FakeListener())
    now = datetime.now(timezone.utc)

    await scheduler._note_closing_stall(round_id, now - timedelta(seconds=CLOSING_STALL_THRESHOLD_SECONDS - 60))
    assert await _stalled_rows(session_factory) == []  # not long enough yet

    await scheduler._note_closing_stall(round_id, now - timedelta(seconds=CLOSING_STALL_THRESHOLD_SECONDS + 60))
    (row,) = await _stalled_rows(session_factory)
    payload = json.loads(row.payload_json)
    assert payload["in_flight"] == 2 and len(payload["txids"]) == 2 and payload["seconds_past_deadline"] >= 1860
    assert row.round_id == round_id and row.user_id is None

    await scheduler._note_closing_stall(round_id, now - timedelta(seconds=CLOSING_STALL_THRESHOLD_SECONDS + 120))
    assert len(await _stalled_rows(session_factory)) == 1  # the last row is the clock: not again yet

    async with session_factory() as session:
        last = await session.get(AuditLog, row.id)
        last.created_at = last.created_at - timedelta(seconds=CLOSING_STALL_THRESHOLD_SECONDS + 1)
        await session.commit()
    await scheduler._note_closing_stall(round_id, now - timedelta(seconds=2 * CLOSING_STALL_THRESHOLD_SECONDS))
    assert len(await _stalled_rows(session_factory)) == 2


# --- DEP-15 / DEP-16 / BET-02 / WDR-01 ------------------------------------------------


async def test_refreshes_of_one_user_run_one_at_a_time_and_other_users_in_parallel(scheduler_session_factory):
    from app.electrum.listener import ElectrumListener

    listener = ElectrumListener(lambda endpoint: None, scheduler_session_factory)
    running, peak, order = {}, {}, []

    async def fake_locked(user_id, scripthash):
        running[user_id] = running.get(user_id, 0) + 1
        peak[user_id] = max(peak.get(user_id, 0), running[user_id])
        order.append((user_id, scripthash))
        await asyncio.sleep(0.01)
        running[user_id] -= 1

    listener._refresh_user_locked = fake_locked
    await asyncio.gather(
        listener.refresh_user(1, "a"), listener.refresh_user(1, "b"), listener.refresh_user(2, "c")
    )
    assert peak == {1: 1, 2: 1}
    assert [s for u, s in order if u == 1] == ["a", "b"]  # in order of arrival


async def test_one_users_failed_refresh_does_not_end_the_session(scheduler_session_factory):
    from app.electrum.listener import ElectrumListener

    listener = ElectrumListener(lambda endpoint: None, scheduler_session_factory)
    listener._scripthash_to_user = {"s1": 1, "s2": 2}
    calls = []

    async def fake_refresh(user_id, scripthash):
        calls.append(user_id)
        if user_id == 1:
            raise RuntimeError("duplicate txid/vout")

    listener.refresh_user = fake_refresh
    queue = asyncio.Queue()
    for item in (("s1", None), ("s2", None)):
        queue.put_nowait(item)
    task = asyncio.create_task(listener._consume_scripthash(queue))
    await asyncio.sleep(0.05)
    assert calls == [1, 2] and not task.done()
    task.cancel()


async def test_the_connection_is_read_after_the_user_lock_is_taken():
    """BET-02 / WDR-01: a request queued on the lock across a reconnection must use the
    connection that exists when it gets the lock, not the one it saw on arrival."""
    from contextlib import asynccontextmanager

    from app.api.routes import bets as bets_route

    listener = SimpleNamespace(client="old")

    @asynccontextmanager
    async def acquire(_user_id):
        listener.client = None  # the connection dropped while this request waited
        yield

    request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(electrum_listener=listener, user_locks=SimpleNamespace(acquire=acquire))))
    with pytest.raises(Exception) as excinfo:
        await bets_route.create_bet(request, SimpleNamespace(), SimpleNamespace(id=1), None)
    assert excinfo.value.detail["code"] == "network_unavailable"