plm-lottery/app/bets/service.py

import logging
from datetime import datetime, timezone

from embit import script
from sqlalchemy import func, select, update
from sqlalchemy.exc import OperationalError
from sqlalchemy.ext.asyncio import AsyncSession

from app.api.errors import ApiError
from app.audit.log import write_audit_log
from app.db.base import begin_immediate
from app.db.models import PendingTransaction, Round, RoundParticipant, User, UtxoEvent
from app.electrum.client import ElectrumClient
from app.rounds.config import get_round_config
from app.rounds.events import broadcaster
from app.rounds.service import open_new_round_if_needed, round_accepts_bets
from app.tx.versions import delete_versions, record_version
from app.tx.outcome import BroadcastOutcome, TransactionNetwork, broadcast_with_verdict
from app.wallet.balance import recompute_balance
from app.wallet.hd import derive_user_key
from app.wallet.psbt_builder import (
    MAX_PARTICIPANTS_PER_ROUND,
    RECIPIENT_VOUT,
    BuiltTransaction,
    InsufficientFundsError,
    Utxo,
    build_signed_transaction,
)


logger = logging.getLogger(__name__)


class BetError(ApiError):
    pass


class _RoundClosedDuringBuild(Exception):
    """Internal signal (B-53): the round stopped accepting bets while this one was
    being built. Never leaves place_bet — it becomes a `round_closing` BetError."""


async def place_bet(
    session: AsyncSession, client: ElectrumClient, user: User, network: TransactionNetwork | None = None
) -> RoundParticipant:
    """`network` (the ElectrumListener) is what verifies a refused broadcast and fans an
    accepted one out (B-76); without it a refusal can never be proven, so it never
    releases anything."""
    # B-88 (DAT-74, BET-11): take SQLite's write lock before the first read, so the
    # balance, the free UTXOs, the participants and the round's status read below are
    # the state the phase-1 write lands on — a plain BEGIN only starts at the first
    # write, leaving every read before it outside any transaction. Failing to get the
    # lock within busy_timeout is not a conflict about this bet: nothing has been
    # built or broadcast yet, so the player is simply asked to retry.
    try:
        await begin_immediate(session)
    except OperationalError as exc:
        raise _service_busy() from exc

    round_ = await open_new_round_if_needed(session)
    if round_ is None:
        raise BetError("no_round_open", "no round open right now, please try again shortly")

    config = await get_round_config(session)
    if not round_accepts_bets(round_):
        raise BetError("round_closing", "the current round is closing, please try again shortly")

    already_playing = await session.scalar(
        select(RoundParticipant).where(
            RoundParticipant.round_id == round_.id, RoundParticipant.user_id == user.id
        )
    )
    if already_playing is not None:
        raise BetError("already_betting", "you already have an active bet in the current round")

    # B-52: "this round can always be paid out" is an invariant, and this is where it
    # gets enforced — before any of this user's money moves. The payout has to spend
    # one pool UTXO per bet, so a round that grew past what a single payout
    # transaction may spend was unpayable: it stayed "paying_out" retrying forever,
    # and because no new round may open while one is active, the whole lottery
    # stopped. Refusing the bet costs the player one round of waiting; accepting it
    # cost everyone the platform. Counted over every participant row, not just the
    # confirmed ones: a bet that later fails frees a slot, so counting them all is
    # the conservative direction.
    participant_count = await session.scalar(
        select(func.count()).select_from(RoundParticipant).where(RoundParticipant.round_id == round_.id)
    )
    if participant_count >= MAX_PARTICIPANTS_PER_ROUND:
        raise BetError(
            "round_full",
            f"this round already has its maximum of {MAX_PARTICIPANTS_PER_ROUND} players, "
            "wait for the next one",
            max_participants=MAX_PARTICIPANTS_PER_ROUND,
        )

    # B-95 (BET-20, DAT-27): the round's own cap on the pool, when it has one. The sum counts
    # every participant row (a bet still in flight may yet confirm — the conservative side), and
    # a bet is refused once the sum has *reached* the cap, since its own net amount is not known
    # before it is built: the pool can therefore end up above the cap by at most one bet.
    if round_.max_round_pool_sats:
        pool_so_far = await session.scalar(
            select(func.coalesce(func.sum(RoundParticipant.bet_amount_sats), 0)).where(
                RoundParticipant.round_id == round_.id
            )
        )
        if pool_so_far >= round_.max_round_pool_sats:
            raise BetError(
                "round_full",
                "this round has reached its limit, wait for the next one",
                max_participants=MAX_PARTICIPANTS_PER_ROUND,
                max_round_pool_sats=round_.max_round_pool_sats,
            )

    if not round_.pool_address:
        # B-80: open_new_round_if_needed always writes it; a row without one cannot
        # take money that would then have nowhere verifiable to go.
        raise RuntimeError(f"round {round_.id} has no pool address")

    # B-95 (BET-21): the stake of the round — what every player of it pays — not the value in
    # round_config now, which an operator may have edited since the round opened.
    bet_amount = round_.bet_amount_sats

    unspent = (
        await session.scalars(
            select(UtxoEvent).where(UtxoEvent.user_id == user.id, UtxoEvent.spent_txid.is_(None))
        )
    ).all()
    if sum(u.amount_sats for u in unspent) < bet_amount:
        raise BetError("insufficient_balance", "insufficient balance", required_sats=bet_amount)

    user_key = derive_user_key(user.derivation_index)
    from_script = script.p2wpkh(user_key.to_public())
    utxos = [Utxo(u.txid, u.vout, u.amount_sats) for u in unspent]

    try:
        built = build_signed_transaction(
            signing_key=user_key,
            from_script=from_script,
            utxos=utxos,
            to_address=round_.pool_address,  # B-80: this round's own pool address
            amount_sats=bet_amount,
            change_address=user.address,
            fee_rate_sat_vb=config.fee_rate_sat_vb,
        )
    except InsufficientFundsError as exc:
        raise BetError(exc.code, str(exc), **exc.params) from exc

    # --- Phase 1: record the intent, *then* broadcast (B-08) --------------------
    # Broadcasting first meant a failure (or a crash) between the broadcast and the
    # commit left the coins irreversibly spent on-chain with no trace in the DB: no
    # participant, so no entry in the draw; no pending row, so no RBF and no
    # confirmation tracking; and the UTXOs not even marked spent, so the next bet
    # would try to double-spend them. Writing "building" rows first means the worst
    # case is a row the reconciler (app/tx/reconcile.py) can resolve either way by
    # asking the chain whether the tx exists.
    spent_by_key = {(u.txid, u.vout): u for u in unspent}
    for spent in built.spent_utxos:
        spent_by_key[(spent.txid, spent.vout)].spent_txid = built.txid
    await recompute_balance(session, user.id)

    broadcast_at = datetime.now(timezone.utc)
    participant = RoundParticipant(
        round_id=round_.id,
        user_id=user.id,
        bet_amount_sats=built.recipient_sats,
        bet_txid=built.txid,
        bet_vout=RECIPIENT_VOUT,  # B-73 (DAT-63)
        broadcast_at=broadcast_at,
        status="building",
    )
    session.add(participant)
    pending = _pending_transaction(round_.id, user.id, built, config.fee_rate_sat_vb)
    session.add(pending)
    await record_version(session, pending, built.txid, built.raw_hex)  # B-77: the first version, with the row

    # B-53: the deadline check at the top of this function happened before the UTXO
    # scan and the signing above, so re-check it here against the clock as it is now —
    # a slow build must not sneak a bet past the round's deadline.
    #
    # And then the part the clock can't cover: a compare-and-set on the round's own
    # row, in the *same* transaction as the participant insert. The scheduler flips
    # "open" -> "closing" in a transaction of its own and only counts in-flight
    # participants afterwards, so without this a bet could commit its "building" row
    # in between and be paid into the pool while the round drew and paid out without
    # it — money credited to no round, no participant and no refund path. The UPDATE
    # takes SQLite's write lock, so the two transactions can no longer interleave:
    # either this commits first and the scheduler's subsequent in-flight count sees
    # the row, or the flip commits first and this matches zero rows and refuses the
    # bet before anything is broadcast.
    try:
        if not round_accepts_bets(round_):
            raise _RoundClosedDuringBuild
        guard = await session.execute(
            update(Round)
            .where(Round.id == round_.id, Round.status == "open")
            .values(status="open")
            .execution_options(synchronize_session=False)
        )
        if guard.rowcount != 1:
            raise _RoundClosedDuringBuild
        await session.commit()
    except _RoundClosedDuringBuild as exc:
        # Nothing has been broadcast yet, so the rollback undoes phase 1 entirely —
        # the UTXOs stay unspent and no participant row survives.
        await session.rollback()
        raise BetError(
            "round_closing", "the current round is closing, please try again shortly"
        ) from exc
    except OperationalError as exc:
        # B-88: SQLite has no write-snapshot conflicts — an OperationalError here is
        # the lock wait (busy_timeout) expiring, which says nothing about the round.
        # Same rollback, but the honest answer is "busy", not "closing".
        await session.rollback()
        raise _service_busy() from exc

    # --- Phase 2: broadcast, then promote both rows to their live state ---------
    # B-76 (TX-04, TX-20): only a *verified* refusal undoes phase 1. A timeout, a
    # dropped connection or a refusal nobody could verify leaves the tx possibly on the
    # network, and releasing the coins then sent the bet's sats into the pool with no
    # participant behind them — so those rows stay "building" for the reconciler.
    verdict = await broadcast_with_verdict(client, network, built.raw_hex, built.txid)
    if verdict.outcome is BroadcastOutcome.REFUSED:
        # The network refused it (fee too low, dust, mempool conflict) and no server
        # knows it — nothing is on-chain, so undo phase 1 completely and give the user
        # a translatable failure instead of a bare 500 (B-07).
        try:
            await _release_failed_bet(session, participant, pending, built, user.id, verdict.reason)
        except OperationalError:
            # B-88: the database is busy, so the release could not be written. The
            # rows stay "building" — the state phase 1 committed — and the
            # reconciler resolves them against the chain.
            await session.rollback()
            logger.error("could not release the failed bet %s: database busy, left to the reconciler", built.txid)
        raise BetError("broadcast_failed", f"the network refused the transaction: {verdict.reason}")
    if verdict.outcome is BroadcastOutcome.UNCERTAIN:
        logger.warning("bet %s: broadcast outcome uncertain (%s), left 'building' for the reconciler", built.txid, verdict.reason)
        await session.refresh(participant)  # reports the status the DB really holds: building
        broadcaster.publish()  # an in-flight bet counts towards the round's figures
        return participant

    # B-88: after the broadcast nothing may be undone. If the lock cannot be had the
    # rows stay "building" (the tx is on the network) and the reconciler promotes
    # them; the caller still gets the participant, with the status the DB holds.
    try:
        await begin_immediate(session)
    except OperationalError:
        await session.rollback()
        logger.error("bet %s broadcast but the database is busy: left 'building' for the reconciler", built.txid)
        await session.refresh(participant)  # reports the status the DB really holds
        return participant
    participant.status = "broadcast"
    pending.status = "pending"
    await write_audit_log(
        session,
        "bet_placed",
        {"txid": built.txid, "amount_sats": built.recipient_sats},
        user_id=user.id,
        round_id=round_.id,
    )
    await session.commit()
    await session.refresh(participant)
    broadcaster.publish()  # participant_count/jackpot changed — nudge every dashboard to refetch
    return participant


async def _release_failed_bet(
    session: AsyncSession,
    participant: RoundParticipant,
    pending: PendingTransaction,
    built: BuiltTransaction,
    user_id: int,
    reason: str,
) -> None:
    """Undo phase 1 after a failed broadcast: free the UTXOs the build reserved, drop
    the two rows, and restore the balance. Same shape as what the reconciler does for
    a tx that turns out never to have made it onto the chain."""
    await begin_immediate(session)  # B-88
    for spent in built.spent_utxos:
        row = await session.scalar(
            select(UtxoEvent).where(UtxoEvent.txid == spent.txid, UtxoEvent.vout == spent.vout)
        )
        if row is not None:
            row.spent_txid = None
    await session.delete(participant)
    await delete_versions(session, pending.id)  # B-77 (DAT-38): the versions go with the row
    await session.delete(pending)
    await recompute_balance(session, user_id)
    await write_audit_log(
        session,
        "bet_broadcast_failed",
        {"txid": built.txid, "reason": reason[:200]},
        user_id=user_id,
    )
    await session.commit()
    # The rollback moved as much state as the successful path did — the balance is
    # back, the participant is gone, so participant_count and jackpot shrank again.
    # Without this the dashboards kept showing the phantom bet until their next poll
    # (B-49); the reconciler's own abandon path has always published here.
    broadcaster.publish()


def _service_busy() -> BetError:
    return BetError(
        "service_busy",
        "the service is busy right now, nothing was sent — please try again",
    )


def _pending_transaction(
    round_id: int, user_id: int, built: BuiltTransaction, fee_rate_sat_vb: int
) -> PendingTransaction:
    return PendingTransaction(
        kind="bet",
        round_id=round_id,
        user_id=user_id,
        current_txid=built.txid,
        fee_rate_sat_vb=fee_rate_sat_vb,
        raw_tx_hex=built.raw_hex,
        # "building" until the broadcast succeeds — see place_bet's two phases. It
        # matters which one this starts as: the reconciler gives a "building" row a
        # short grace period (we may have died mid-broadcast) and a "pending" one a
        # long one (a node accepted it once, so it deserves the RBF attempts first).
        status="building",
    )