plm-lottery/tests/unit/test_broadcast_outcome.py

"""B-76 (TX-04, TX-20, TX-21, TX-05, BET-16, BET-17, BET-23, WDR-11, WDR-12, WDR-16, API-09):
an ambiguous broadcast is never rolled back as if it had been refused.

Only an explicit JSON-RPC refusal that the whole network then confirms it does not know
may release coins. A timeout, a dropped connection, or a refusal nobody could verify
leaves the rows `building` for the reconciler. Each of these tests fails against the
old code, where *any* exception from the broadcast undid everything.
"""

import asyncio
from types import SimpleNamespace

import pytest
from fastapi import Response
from sqlalchemy import select

from app.bets.service import BetError, place_bet
from app.db.models import AuditLog, PendingTransaction, RoundParticipant, User, UtxoEvent, Withdrawal
from app.electrum.client import ElectrumEndpoint, ElectrumError
from app.electrum.listener import ElectrumListener
from app.rounds.events import broadcaster
from app.rounds.scheduler import RoundScheduler
from app.tx.outcome import BroadcastOutcome, broadcast_with_verdict
from app.withdrawals.service import WithdrawalError, request_withdrawal
from tests.unit.fake_network import FakeNetwork, rpc_error
from tests.unit.test_bets import _make_funded_user, session_factory  # noqa: F401  (fixture)
from tests.unit.test_scheduler import (  # noqa: F401  (fixtures and helpers shared with the payout tests)
    FakePayoutClient,
    FakePayoutListener,
    _POOL_AMOUNT_SATS,
    _seed_paying_out_round,
    payout_session_factory,
)

_EXTERNAL = "plm1qqph9qup2mp7w7g5nlsdhdc9m2pp44ampzw0ctx"
_FUNDS = 3_000_000_000


class _FailingClient:
    """A client whose broadcast fails the way `exc` says."""

    def __init__(self, exc: BaseException):
        self._exc = exc
        self.broadcasted: list[str] = []

    async def broadcast(self, raw_tx_hex: str) -> str:
        self.broadcasted.append(raw_tx_hex)
        raise self._exc


_UNCERTAIN_FAILURES = [
    pytest.param(ElectrumError("blockchain.transaction.broadcast timed out after 15s"), id="timeout-after-write"),
    pytest.param(asyncio.TimeoutError(), id="bare-timeout"),
    pytest.param(ElectrumError("connection closed"), id="disconnect"),
    pytest.param(ElectrumError("write failed: broken pipe"), id="write-error"),
    pytest.param(ConnectionResetError("reset by peer"), id="connection-reset"),
]


async def _bet_rows(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.scalars(select(UtxoEvent).where(UtxoEvent.user_id == user_id))).one()
        events = [e.event_type for e in (await session.scalars(select(AuditLog))).all()]
        return participants, pendings, utxo, events


# --- bets ----------------------------------------------------------------------


@pytest.mark.parametrize("failure", _UNCERTAIN_FAILURES)
async def test_a_bet_whose_broadcast_ends_without_an_answer_is_left_building(session_factory, failure):
    """TX-20 / BET-23: nothing is released, the player gets the `building` participant,
    and the network is never asked to verify (there was no refusal to verify)."""
    user_id = await _make_funded_user(session_factory, 4, _FUNDS)
    network = FakeNetwork(known=False)  # would say "release" if asked — it must not be
    queue = broadcaster.subscribe()
    try:
        while not queue.empty():
            queue.get_nowait()
        async with session_factory() as session:
            user = await session.get(User, user_id)
            participant = await place_bet(session, _FailingClient(failure), user, network=network)
        assert participant.status == "building"
        assert not queue.empty()  # SSE
    finally:
        broadcaster.unsubscribe(queue)

    participants, pendings, utxo, events = await _bet_rows(session_factory, user_id)
    assert [p.status for p in participants] == ["building"]
    assert [p.status for p in pendings] == ["building"]
    assert utxo.spent_txid == participants[0].bet_txid  # the coins stay committed to the tx
    assert "bet_broadcast_failed" not in events and "bet_placed" not in events
    assert network.asked == [] and network.fanned_out == []
    async with session_factory() as session:
        assert (await session.get(User, user_id)).cached_balance_sats == 0  # not given back


async def test_a_refused_bet_that_another_server_knows_counts_as_accepted(session_factory):
    """TX-04: the active server refused ("txn-mempool-conflict", whatever it said) but a
    server of the network returns the transaction — it is on the network, so the bet
    stands and is promoted. Releasing here sent the sats to the pool without a participant."""
    user_id = await _make_funded_user(session_factory, 4, _FUNDS)
    network = FakeNetwork(known=True)
    client = _FailingClient(rpc_error("some wording we have never seen"))
    async with session_factory() as session:
        user = await session.get(User, user_id)
        participant = await place_bet(session, client, user, network=network)

    assert participant.status == "broadcast"
    participants, pendings, utxo, events = await _bet_rows(session_factory, user_id)
    assert [p.status for p in pendings] == ["pending"]
    assert "bet_placed" in events and "bet_broadcast_failed" not in events
    assert network.asked == [participants[0].bet_txid]
    assert len(network.fanned_out) == 1  # TX-21: sent to the other servers too


async def test_a_refused_bet_nobody_knows_is_released(session_factory):
    """TX-04 / BET-16: a verified refusal still undoes everything and answers 502."""
    user_id = await _make_funded_user(session_factory, 4, _FUNDS)
    network = FakeNetwork(known=False)
    async with session_factory() as session:
        user = await session.get(User, user_id)
        with pytest.raises(BetError) as excinfo:
            await place_bet(session, _FailingClient(rpc_error("min relay fee not met")), user, network=network)
    assert excinfo.value.code == "broadcast_failed"
    participants, pendings, utxo, events = await _bet_rows(session_factory, user_id)
    assert participants == [] and pendings == [] and utxo.spent_txid is None
    assert "bet_broadcast_failed" in events
    assert network.fanned_out == []


async def test_a_refusal_the_network_cannot_verify_releases_nothing(session_factory):
    """TX-04: too few servers answered the verification — the outcome is uncertain, so
    the bet stays `building` instead of being rolled back on the refusal's word alone."""
    user_id = await _make_funded_user(session_factory, 4, _FUNDS)
    async with session_factory() as session:
        user = await session.get(User, user_id)
        participant = await place_bet(
            session, _FailingClient(rpc_error("min relay fee not met")), user, network=FakeNetwork(known=None)
        )
    assert participant.status == "building"
    participants, pendings, utxo, events = await _bet_rows(session_factory, user_id)
    assert [p.status for p in pendings] == ["building"] and utxo.spent_txid == participants[0].bet_txid
    assert "bet_broadcast_failed" not in events


async def test_a_refusal_with_no_network_to_verify_it_releases_nothing(session_factory):
    user_id = await _make_funded_user(session_factory, 4, _FUNDS)
    async with session_factory() as session:
        user = await session.get(User, user_id)
        participant = await place_bet(session, _FailingClient(rpc_error("min relay fee not met")), user)
    assert participant.status == "building"


@pytest.mark.parametrize(
    "message",
    [
        "txn-already-known",
        "txn-already-in-mempool",
        "Transaction already in block chain",
        "the transaction was rejected by network rules.\n\ntxn-already-in-mempool",
    ],
)
async def test_already_known_is_a_shortcut_that_skips_the_verification(session_factory, message):
    """TX-04: a recognised "already known" message counts as accepted without asking
    anyone; it is a shortcut, never a condition of safety."""
    user_id = await _make_funded_user(session_factory, 4, _FUNDS)
    network = FakeNetwork(known=False)
    async with session_factory() as session:
        user = await session.get(User, user_id)
        participant = await place_bet(session, _FailingClient(rpc_error(message)), user, network=network)
    assert participant.status == "broadcast"
    assert network.asked == []
    assert len(network.fanned_out) == 1


async def test_the_bet_route_answers_202_for_an_uncertain_outcome_and_201_for_success(session_factory):
    """API-09 / BET-23: `building` is HTTP 202; a promoted bet keeps 201."""
    from contextlib import asynccontextmanager

    from app.api.routes.bets import create_bet

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

    def _request(client, network):
        listener = SimpleNamespace(client=client, **{"transaction_known": network.transaction_known,
                                                      "fan_out_transaction": network.fan_out_transaction})
        return SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(electrum_listener=listener, user_locks=_Locks())))

    first = await _make_funded_user(session_factory, 4, _FUNDS)
    second = await _make_funded_user(session_factory, 5, _FUNDS)

    response = Response()
    async with session_factory() as session:
        user = await session.get(User, first)
        body = await create_bet(
            _request(_FailingClient(ElectrumError("connection closed")), FakeNetwork()), response, user=user, session=session
        )
    assert response.status_code == 202 and body.status == "building"

    class _OkClient:
        async def broadcast(self, raw_tx_hex):
            return "txid"

    response = Response()
    async with session_factory() as session:
        user = await session.get(User, second)
        body = await create_bet(_request(_OkClient(), FakeNetwork()), response, user=user, session=session)
    assert response.status_code != 202 and body.status == "broadcast"  # the route's declared 201 stands


# --- withdrawals -----------------------------------------------------------------


async def _withdrawal_rows(session_factory, user_id):
    async with session_factory() as session:
        withdrawals = (await session.scalars(select(Withdrawal))).all()
        pendings = (await session.scalars(select(PendingTransaction))).all()
        utxo = (await session.scalars(select(UtxoEvent).where(UtxoEvent.user_id == user_id))).one()
        events = [e.event_type for e in (await session.scalars(select(AuditLog))).all()]
        return withdrawals, pendings, utxo, events


@pytest.mark.parametrize("failure", _UNCERTAIN_FAILURES)
async def test_a_withdrawal_whose_broadcast_ends_without_an_answer_is_left_building(session_factory, failure):
    """TX-20 / WDR-16: the transaction may have left; the withdrawal must not be
    marked failed with its coins spendable again (the money would go out twice)."""
    user_id = await _make_funded_user(session_factory, 7, _FUNDS)
    network = FakeNetwork(known=False)
    async with session_factory() as session:
        user = await session.get(User, user_id)
        withdrawal = await request_withdrawal(
            session, _FailingClient(failure), user, _EXTERNAL, 1_000_000_000, network=network
        )
    assert withdrawal.status == "building"
    withdrawals, pendings, utxo, events = await _withdrawal_rows(session_factory, user_id)
    assert [w.status for w in withdrawals] == ["building"]
    assert [p.status for p in pendings] == ["building"]
    assert utxo.spent_txid == withdrawals[0].txid
    assert "withdrawal_broadcast_failed" not in events and "withdrawal_sent" not in events
    assert network.asked == []


async def test_a_refused_withdrawal_that_another_server_knows_counts_as_accepted(session_factory):
    user_id = await _make_funded_user(session_factory, 7, _FUNDS)
    network = FakeNetwork(known=True)
    async with session_factory() as session:
        user = await session.get(User, user_id)
        withdrawal = await request_withdrawal(
            session, _FailingClient(rpc_error("unrecognised")), user, _EXTERNAL, 1_000_000_000, network=network
        )
    assert withdrawal.status == "broadcast"
    withdrawals, pendings, utxo, events = await _withdrawal_rows(session_factory, user_id)
    assert [p.status for p in pendings] == ["pending"] and "withdrawal_sent" in events
    assert len(network.fanned_out) == 1


async def test_a_refused_withdrawal_nobody_knows_is_released_and_marked_failed(session_factory):
    user_id = await _make_funded_user(session_factory, 7, _FUNDS)
    async with session_factory() as session:
        user = await session.get(User, user_id)
        with pytest.raises(WithdrawalError) as excinfo:
            await request_withdrawal(
                session,
                _FailingClient(rpc_error("min relay fee not met")),
                user,
                _EXTERNAL,
                1_000_000_000,
                network=FakeNetwork(known=False),
            )
    assert excinfo.value.code == "broadcast_failed"
    withdrawals, pendings, utxo, events = await _withdrawal_rows(session_factory, user_id)
    assert [w.status for w in withdrawals] == ["failed"] and utxo.spent_txid is None
    assert "withdrawal_broadcast_failed" in events


async def test_an_unverifiable_withdrawal_refusal_releases_nothing(session_factory):
    user_id = await _make_funded_user(session_factory, 7, _FUNDS)
    async with session_factory() as session:
        user = await session.get(User, user_id)
        withdrawal = await request_withdrawal(
            session, _FailingClient(rpc_error("x")), user, _EXTERNAL, 1_000_000_000, network=FakeNetwork(known=None)
        )
    assert withdrawal.status == "building"


async def test_the_withdrawal_route_answers_202_for_an_uncertain_outcome(session_factory):
    from contextlib import asynccontextmanager

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

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

    user_id = await _make_funded_user(session_factory, 7, _FUNDS)
    network = FakeNetwork()
    listener = SimpleNamespace(
        client=_FailingClient(ElectrumError("connection closed")),
        transaction_known=network.transaction_known,
        fan_out_transaction=network.fan_out_transaction,
    )
    request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(electrum_listener=listener, user_locks=_Locks())))
    response = Response()
    async with session_factory() as session:
        user = await session.get(User, user_id)
        body = await create_withdrawal(
            WithdrawalRequest(external_address=_EXTERNAL, amount_sats=1_000_000_000), request, response, user=user, session=session
        )
    assert response.status_code == 202 and body.status == "building"


# --- payouts ---------------------------------------------------------------------


def _payout_entries():
    return [{"tx_hash": "44" * 32, "tx_pos": 0, "height": 10, "value": _POOL_AMOUNT_SATS + 100_000}]


class _RaisingPayoutClient(FakePayoutClient):
    def __init__(self, entries, exc):
        super().__init__(entries)
        self._exc = exc

    async def broadcast(self, raw_tx_hex):
        raise self._exc


@pytest.mark.parametrize(
    "exc, known",
    [
        pytest.param(ElectrumError("connection closed"), False, id="uncertain"),
        pytest.param(rpc_error("min relay fee not met"), False, id="verified-refusal"),
        pytest.param(rpc_error("min relay fee not met"), None, id="unverifiable-refusal"),
    ],
)
async def test_a_payout_that_is_not_accepted_stays_building_with_a_payout_failed_entry(payout_session_factory, exc, known):
    """TX-05: refusal or uncertainty alike leave the row `building` — the reconciler,
    asking the chain, decides whether it left (TX-16, TX-17)."""
    await _seed_paying_out_round(payout_session_factory)
    scheduler = RoundScheduler(payout_session_factory, FakePayoutListener(_RaisingPayoutClient(_payout_entries(), exc), known=known))

    await scheduler._trigger_payout(1)

    async with payout_session_factory() as session:
        pending = (await session.scalars(select(PendingTransaction))).one()
        assert pending.status == "building"
        events = [e.event_type for e in (await session.scalars(select(AuditLog))).all()]
        assert "payout_failed" in events and "payout_sent" not in events


async def test_a_payout_refused_as_already_known_is_promoted(payout_session_factory):
    """TX-05: "already known" is an accepted broadcast — `pending` + `payout_sent`."""
    await _seed_paying_out_round(payout_session_factory)
    listener = FakePayoutListener(_RaisingPayoutClient(_payout_entries(), rpc_error("txn-already-in-mempool")))
    await RoundScheduler(payout_session_factory, listener)._trigger_payout(1)

    async with payout_session_factory() as session:
        pending = (await session.scalars(select(PendingTransaction))).one()
        assert pending.status == "pending"
        events = [e.event_type for e in (await session.scalars(select(AuditLog))).all()]
        assert "payout_sent" in events
    assert len(listener.fanned_out) == 1


async def test_an_accepted_payout_is_fanned_out_to_the_other_servers(payout_session_factory):
    """TX-21."""
    await _seed_paying_out_round(payout_session_factory)
    client = FakePayoutClient(_payout_entries())
    listener = FakePayoutListener(client)
    await RoundScheduler(payout_session_factory, listener)._trigger_payout(1)
    assert listener.fanned_out == client.broadcasted


# --- the verdict itself -------------------------------------------------------------


async def test_the_verdict_distinguishes_the_three_outcomes():
    class _Ok:
        async def broadcast(self, raw):
            return "txid"

    ok = await broadcast_with_verdict(_Ok(), FakeNetwork(), "00", "aa")
    refused = await broadcast_with_verdict(_FailingClient(rpc_error("bad")), FakeNetwork(known=False), "00", "aa")
    uncertain = await broadcast_with_verdict(_FailingClient(ElectrumError("timed out")), FakeNetwork(known=False), "00", "aa")
    assert (ok.outcome, refused.outcome, uncertain.outcome) == (
        BroadcastOutcome.ACCEPTED,
        BroadcastOutcome.REFUSED,
        BroadcastOutcome.UNCERTAIN,
    )


# --- the listener: transaction_known / fan_out_transaction ----------------------------


_ENDPOINTS = [
    ElectrumEndpoint("active.example", 50002, True),
    ElectrumEndpoint("first.example", 50002, True),
    ElectrumEndpoint("second.example", 50002, True),
    ElectrumEndpoint("third.example", 50002, True),
]


def _real_tx():
    """A real serialised transaction and its txid — transaction_known checks the
    returned bytes hash to the txid asked about."""
    from embit.transaction import Transaction, TransactionInput, TransactionOutput
    from embit import script

    tx = Transaction(
        vin=[TransactionInput(bytes.fromhex("11" * 32), 0)],
        vout=[TransactionOutput(1000, script.p2wpkh(__import__("embit").ec.PrivateKey(b"\x01" * 32).get_public_key()))],
    )
    return tx.serialize().hex(), tx.txid().hex()


class _ServerClient:
    """A fake connection to one server: `answer` is a raw tx hex (it knows the tx), the
    string "unknown" (an RPC error: no such transaction), or None (unreachable)."""

    def __init__(self, answer, log):
        self._answer = answer
        self._log = log

    async def connect(self):
        if self._answer is None:
            raise ConnectionRefusedError("down")

    async def get_transaction(self, txid, verbose=False):
        if self._answer == "unknown":
            raise rpc_error("No such mempool or blockchain transaction", code=-5)
        return self._answer

    async def broadcast(self, raw):
        self._log.append(raw)
        return "txid"

    async def close(self):
        pass


def _listener(session_factory, answers: dict, active_answer, log=None, allow_no_quorum=False, endpoints=None):
    log = log if log is not None else []
    listener = ElectrumListener(
        lambda endpoint: _ServerClient(answers[endpoint.host], log),
        session_factory,
        endpoints or _ENDPOINTS,
        allow_no_quorum=allow_no_quorum,
    )
    listener._endpoint_index = 0
    listener.client = None if active_answer is Ellipsis else _ServerClient(active_answer, log)
    return listener


async def test_transaction_known_is_true_when_any_server_returns_it(session_factory):
    raw, txid = _real_tx()
    listener = _listener(
        session_factory,
        {"first.example": "unknown", "second.example": raw, "third.example": None},
        active_answer="unknown",
    )
    assert await listener.transaction_known(txid) is True


async def test_transaction_known_is_true_when_only_the_active_server_has_it(session_factory):
    raw, txid = _real_tx()
    listener = _listener(
        session_factory,
        {"first.example": "unknown", "second.example": "unknown", "third.example": "unknown"},
        active_answer=raw,
    )
    assert await listener.transaction_known(txid) is True


async def test_transaction_known_is_false_only_when_the_active_server_and_a_quorum_deny_it(session_factory):
    _, txid = _real_tx()
    listener = _listener(
        session_factory,
        {"first.example": "unknown", "second.example": "unknown", "third.example": None},
        active_answer="unknown",
    )
    assert await listener.transaction_known(txid) is False


async def test_transaction_known_is_uncertain_with_too_few_answers(session_factory):
    """Only one of the other servers answered: below MIN_QUORUM_RESPONSES, so the
    denial proves nothing."""
    _, txid = _real_tx()
    listener = _listener(
        session_factory,
        {"first.example": "unknown", "second.example": None, "third.example": None},
        active_answer="unknown",
    )
    assert await listener.transaction_known(txid) is None


async def test_transaction_known_is_uncertain_when_the_active_server_does_not_answer(session_factory):
    _, txid = _real_tx()
    listener = _listener(
        session_factory,
        {"first.example": "unknown", "second.example": "unknown", "third.example": "unknown"},
        active_answer=Ellipsis,  # no active connection
    )
    assert await listener.transaction_known(txid) is None


async def test_transaction_known_ignores_a_reply_that_is_not_the_transaction(session_factory):
    """A server answering with some other transaction does not make this one known —
    and does not count as a denial either."""
    listener = _listener(
        session_factory,
        {"first.example": "unknown", "second.example": "00", "third.example": "unknown"},
        active_answer="unknown",
    )
    assert await listener.transaction_known("ab" * 32) is False  # 2 real denials, 1 ignored reply


async def test_transaction_known_follows_the_reduced_rule_under_allow_no_quorum(session_factory):
    _, txid = _real_tx()
    single = [ElectrumEndpoint("active.example", 50002, True)]
    listener = _listener(session_factory, {}, active_answer="unknown", allow_no_quorum=True, endpoints=single)
    assert await listener.transaction_known(txid) is False
    strict = _listener(session_factory, {}, active_answer="unknown", endpoints=single)
    assert await strict.transaction_known(txid) is None


async def test_fan_out_sends_to_every_other_server_and_never_to_the_active_one(session_factory):
    log: list[str] = []
    listener = _listener(
        session_factory,
        {"first.example": "unknown", "second.example": "unknown", "third.example": None},
        active_answer="unknown",
        log=log,
    )
    listener.fan_out_transaction("deadbeef")
    await asyncio.gather(*listener._background_tasks)
    assert log == ["deadbeef", "deadbeef"]  # first + second; third unreachable; active untouched