plm-lottery/app/bets/service.py

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.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.wallet.balance import recompute_balance
from app.wallet.hd import derive_pool_address, derive_user_key
from app.wallet.psbt_builder import (
    MAX_PARTICIPANTS_PER_ROUND,
    BuiltTransaction,
    InsufficientFundsError,
    Utxo,
    build_signed_transaction,
)


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) -> RoundParticipant:
    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,
        )

    bet_amount = config.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=derive_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,
        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)

    # 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, OperationalError) as exc:
        # OperationalError here is SQLite's write-snapshot conflict: the round row
        # changed under us, which is the same situation as the guard matching nothing.
        # 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

    # --- Phase 2: broadcast, then promote both rows to their live state ---------
    try:
        await client.broadcast(built.raw_hex)
    except Exception as exc:
        # The node refused it (fee too low, dust, mempool conflict, or simply an
        # unreachable server) — nothing is on-chain, so undo phase 1 completely and
        # give the user a translatable failure instead of a bare 500 (B-07).
        await _release_failed_bet(session, participant, pending, built, user.id, str(exc))
        raise BetError("broadcast_failed", f"the network refused the transaction: {exc}") from exc

    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."""
    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 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 _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",
    )