plm-lottery/app/wallet/psbt_builder.py

from dataclasses import dataclass

from embit import script
from embit.bip32 import HDKey
from embit.finalizer import finalize_psbt
from embit.psbt import PSBT
from embit.transaction import Transaction, TransactionInput, TransactionOutput

# Standard P2WPKH size estimates (vbytes): 10.5-byte overhead (version+counts+locktime+
# segwit marker/flag), ~68 vbytes per input, ~31 vbytes per output. Used to size the fee
# before signing (fee only needs to be "minimized ~1 sat/vB", not maximally precise).
_TX_OVERHEAD_VBYTES = 11
_P2WPKH_INPUT_VBYTES = 68
_P2WPKH_OUTPUT_VBYTES = 31

# BIP125 opt-in RBF: any sequence < 0xfffffffe signals replaceability. Set on every
# input we create so a stuck tx can later be fee-bumped (tx/broadcast.py, stage 9).
RBF_SEQUENCE = 0xFFFFFFFD


class InsufficientFundsError(Exception):
    """`code` is the machine-readable identifier the API layer forwards to the
    client so it can translate the failure (see app/api/errors.py); the message
    itself stays English."""

    def __init__(self, message: str, code: str = "insufficient_balance") -> None:
        super().__init__(message)
        self.code = code


@dataclass
class Utxo:
    txid: str
    vout: int
    amount_sats: int


@dataclass
class BuiltTransaction:
    raw_hex: str
    txid: str
    fee_sats: int
    recipient_sats: int
    change_sats: int
    spent_utxos: list[Utxo]


def estimate_vsize(n_inputs: int, n_outputs: int) -> int:
    return _TX_OVERHEAD_VBYTES + n_inputs * _P2WPKH_INPUT_VBYTES + n_outputs * _P2WPKH_OUTPUT_VBYTES


def select_utxos(utxos: list[Utxo], target_sats: int) -> tuple[list[Utxo], int]:
    """Greedily select UTXOs (largest first, to minimize input count) covering
    target_sats — the amount deducted from the sender's balance. The fee is paid
    out of target_sats (see build_signed_transaction), not added on top of it."""
    ordered = sorted(utxos, key=lambda u: u.amount_sats, reverse=True)
    selected: list[Utxo] = []
    total = 0
    for utxo in ordered:
        selected.append(utxo)
        total += utxo.amount_sats
        if total >= target_sats:
            return selected, total
    raise InsufficientFundsError("not enough confirmed balance to cover amount")


def build_signed_transaction(
    *,
    signing_key: HDKey,
    from_script: script.Script,
    utxos: list[Utxo],
    to_address: str,
    amount_sats: int,
    change_address: str,
    fee_rate_sat_vb: int,
) -> BuiltTransaction:
    """Build, sign and finalize a single-recipient P2WPKH transaction with change
    back to change_address.

    `amount_sats` is deducted from the sender's balance in full: the recipient
    receives `amount_sats - fee`, change = total_in - amount_sats. This matches the
    spec everywhere a single-recipient tx is used (bet, withdrawal): "fee deducted
    from the amount being moved", not paid on top by the sender.
    """
    selected, total_in = select_utxos(utxos, amount_sats)
    fee = estimate_vsize(len(selected), 2) * fee_rate_sat_vb
    recipient_amount = amount_sats - fee
    if recipient_amount <= 0:
        raise InsufficientFundsError(
            "amount too small to cover the network fee", code="amount_below_network_fee"
        )
    change = total_in - amount_sats

    # TransactionInput.txid is natural/display byte order (as in tx_hash from Electrum);
    # embit reverses it internally when serializing to wire format.
    vin = [TransactionInput(bytes.fromhex(u.txid), u.vout, sequence=RBF_SEQUENCE) for u in selected]
    vout = [TransactionOutput(recipient_amount, script.Script.from_address(to_address))]
    if change > 0:
        vout.append(TransactionOutput(change, script.Script.from_address(change_address)))

    tx = Transaction(vin=vin, vout=vout)
    psbt = PSBT(tx)
    for i, utxo in enumerate(selected):
        psbt.inputs[i].witness_utxo = TransactionOutput(utxo.amount_sats, from_script)

    signed_count = psbt.sign_with(signing_key)
    if signed_count != len(selected):
        raise RuntimeError(f"expected {len(selected)} signatures, got {signed_count}")

    final_tx = finalize_psbt(psbt)
    if final_tx is None:
        raise RuntimeError("failed to finalize PSBT")

    raw = final_tx.serialize()
    return BuiltTransaction(
        raw_hex=raw.hex(),
        txid=final_tx.txid().hex(),
        fee_sats=fee,
        recipient_sats=recipient_amount,
        change_sats=change,
        spent_utxos=selected,
    )


@dataclass
class PayoutTransaction:
    raw_hex: str
    txid: str
    fee_sats: int
    winner_sats: int
    commission_sats: int
    change_sats: int
    spent_utxos: list[Utxo]


def build_payout_transaction(
    *,
    signing_key: HDKey,
    from_script: script.Script,
    utxos: list[Utxo],
    winner_address: str,
    winner_share_sats: int,
    fee_address: str,
    commission_sats: int,
    change_address: str,
    fee_rate_sat_vb: int,
) -> PayoutTransaction:
    """Build, sign and finalize the round payout: pool -> winner + fee address,
    with change back to the pool itself. Per spec, only the winner's share
    absorbs the tx fee — the commission (fee_address) output is untouched."""
    target = winner_share_sats + commission_sats
    selected, total_in = select_utxos(utxos, target)
    fee = estimate_vsize(len(selected), 3) * fee_rate_sat_vb  # winner + commission + pool change
    winner_amount = winner_share_sats - fee
    if winner_amount <= 0:
        raise InsufficientFundsError(
            "winner share too small to cover the network fee", code="winner_share_below_network_fee"
        )
    change = total_in - target

    vin = [TransactionInput(bytes.fromhex(u.txid), u.vout, sequence=RBF_SEQUENCE) for u in selected]
    vout = [
        TransactionOutput(winner_amount, script.Script.from_address(winner_address)),
        TransactionOutput(commission_sats, script.Script.from_address(fee_address)),
    ]
    if change > 0:
        vout.append(TransactionOutput(change, script.Script.from_address(change_address)))

    tx = Transaction(vin=vin, vout=vout)
    psbt = PSBT(tx)
    for i, utxo in enumerate(selected):
        psbt.inputs[i].witness_utxo = TransactionOutput(utxo.amount_sats, from_script)

    signed_count = psbt.sign_with(signing_key)
    if signed_count != len(selected):
        raise RuntimeError(f"expected {len(selected)} signatures, got {signed_count}")

    final_tx = finalize_psbt(psbt)
    if final_tx is None:
        raise RuntimeError("failed to finalize PSBT")

    raw = final_tx.serialize()
    return PayoutTransaction(
        raw_hex=raw.hex(),
        txid=final_tx.txid().hex(),
        fee_sats=fee,
        winner_sats=winner_amount,
        commission_sats=commission_sats,
        change_sats=change,
        spent_utxos=selected,
    )