import logging from collections.abc import Awaitable, Callable from dataclasses import dataclass, field from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.log import write_audit_log from app.db.base import begin_immediate from app.db.models import PendingTransaction, PendingTransactionTxid, RoundConfig, UtxoEvent from app.rounds.events import broadcaster from app.wallet.balance import recompute_balance from app.wallet.psbt_builder import DUST_LIMIT_SATS logger = logging.getLogger(__name__) # B-82 (DEP-03): confirmations a coinbase output needs before it can be spent, PLM's # COINBASE_MATURITY (palladiumcore/src/consensus/consensus.h). A coinbase paying a user's # address is credited no earlier than this, whatever deposit_confirmations says. COINBASE_MATURITY = 120 # Confirmations an output created by a transaction of the platform itself — the change of # a bet or a withdrawal, a payout — needs: the platform built it, nobody else could have # double-spent its inputs (DEP-03). PLATFORM_TRANSACTION_CONFIRMATIONS = 1 # Default of round_config.deposit_confirmations, used only when the config row does not # exist yet (get_round_config creates it with the same value). DEFAULT_DEPOSIT_CONFIRMATIONS = 6 async def find_new_credit_candidates(session: AsyncSession, user_id: int, entries: list[dict]) -> list[dict]: """The subset of `entries` that would actually credit something: confirmed (height > 0, per the Electrum convention where <= 0 means mempool) and not already recorded, and not dust (B-75: below DUST_LIMIT_SATS nothing is credited). Split out from credit_confirmed_utxos so the caller (electrum/listener.py:refresh_user) can corroborate each *new* outpoint against the other configured servers before any of it is written (B-59) — the mirror image of what B-29 already required before a balance may go *down*. Only new ones: corroborating outpoints already credited would open a connection to every other server on every refresh, for an answer that can no longer change what we hold. """ existing_keys = { (txid, vout) for txid, vout in ( await session.execute(select(UtxoEvent.txid, UtxoEvent.vout).where(UtxoEvent.user_id == user_id)) ).all() } return [ entry for entry in entries if entry["height"] > 0 and entry["value"] >= DUST_LIMIT_SATS # B-75 (DEP-01) and (entry["tx_hash"], entry["tx_pos"]) not in existing_keys ] @dataclass class DepositPolicy: """What decides how many confirmations an incoming output needs (DEP-03), read from the database before any network call: the configured deposit threshold and which of the transactions involved are the platform's own.""" deposit_confirmations: int = DEFAULT_DEPOSIT_CONFIRMATIONS platform_txids: set[str] = field(default_factory=set) async def read_deposit_policy(session: AsyncSession, txids: set[str]) -> DepositPolicy: """The deposit threshold and, among `txids`, the ones created by a platform transaction: any txid of a `pending_transactions` row, replaced RBF versions included (B-77).""" configured = await session.scalar(select(RoundConfig.deposit_confirmations)) platform: set[str] = set() if txids: platform |= set( (await session.scalars(select(PendingTransactionTxid.txid).where(PendingTransactionTxid.txid.in_(txids)))) ) platform |= set( (await session.scalars(select(PendingTransaction.current_txid).where(PendingTransaction.current_txid.in_(txids)))) ) return DepositPolicy( deposit_confirmations=configured if configured else DEFAULT_DEPOSIT_CONFIRMATIONS, platform_txids=platform ) async def classify_new_deposits( new_entries: list[dict], tip_height: int, policy: DepositPolicy, is_coinbase: Callable[[str], Awaitable[bool | None]], ) -> tuple[list[tuple[dict, int]], list[dict]]: """Split the new, confirmed, non-dust entries of `find_new_credit_candidates` into the ones that may now be credited — as `(entry, required_confirmations)`, the number the quorum is then asked to confirm (ELC-17) — and the ones still **waiting** for confirmations (B-82, DEP-03, DEP-14). Confirmations are `tip_height - height + 1`. The threshold is 1 for an output of a platform transaction, 120 (`COINBASE_MATURITY`) for a coinbase output, and `deposit_confirmations` for anything else; a coinbase is only looked for once the plain threshold is met, since below it the output waits anyway. `is_coinbase` answers from the raw transaction (hash-checked by the caller) and may return None when it cannot tell, in which case the output waits and is asked again at the next refresh. No database session is involved: this runs between the read and the write phases, where the network is used. """ candidates: list[tuple[dict, int]] = [] waiting: list[dict] = [] for entry in new_entries: confirmations = tip_height - entry["height"] + 1 if entry["tx_hash"] in policy.platform_txids: required = PLATFORM_TRANSACTION_CONFIRMATIONS else: required = policy.deposit_confirmations if required <= confirmations < COINBASE_MATURITY: # Plain threshold met, maturity not: only now is it worth knowing whether # this is a coinbase output (which would still be immature). coinbase = await is_coinbase(entry["tx_hash"]) if coinbase is None or coinbase: waiting.append(entry) continue if confirmations < required: waiting.append(entry) continue candidates.append((entry, required)) return candidates, waiting async def credit_confirmed_utxos(session: AsyncSession, user_id: int, entries: list[dict]) -> int: """Insert utxo_events for newly-confirmed entries from an Electrum `listunspent` response (idempotent on txid+vout), refresh the user's cached balance. Returns the number of newly-credited UTXOs. entries: [{"tx_hash": ..., "tx_pos": ..., "height": ..., "value": ...}, ...] height <= 0 means unconfirmed (mempool) per the Electrum protocol convention — skipped, since the spec requires 1 confirmation before crediting. B-88: opens with BEGIN IMMEDIATE, so the "already recorded?" read and the inserts it guards are one transaction — the unique (txid, vout) constraint stays the backstop, not the mechanism. """ await begin_immediate(session) existing_keys = { (txid, vout) for txid, vout in ( await session.execute(select(UtxoEvent.txid, UtxoEvent.vout).where(UtxoEvent.user_id == user_id)) ).all() } newly_credited = 0 for entry in entries: if entry["height"] <= 0: continue if entry["value"] < DUST_LIMIT_SATS: # B-75 (DEP-01): anyone can send dust to a public address; crediting it would # make every later refresh of this user's address do more work, for coins # that cost more to spend than they are worth. continue key = (entry["tx_hash"], entry["tx_pos"]) if key in existing_keys: continue session.add( UtxoEvent( user_id=user_id, txid=entry["tx_hash"], vout=entry["tx_pos"], amount_sats=entry["value"], confirmed_height=entry["height"], ) ) await write_audit_log( session, "deposit_credited", {"txid": entry["tx_hash"], "vout": entry["tx_pos"], "amount_sats": entry["value"]}, user_id=user_id, ) newly_credited += 1 if newly_credited: await session.flush() await recompute_balance(session, user_id) await session.commit() broadcaster.publish() # nudges this user's dashboard to refetch its balance instantly return newly_credited _EXTERNAL_SPEND_SENTINEL = "external-spend" async def find_reinstate_candidates(session: AsyncSession, user_id: int, entries: list[dict]) -> list[dict]: """Outpoints flagged as spent outside the platform that this refresh's `listunspent` lists again, confirmed and for the amount we recorded — the *candidates* for reinstatement (B-83, DEP-02, DEP-08). Read-only: like a new credit (B-59) and a candidate external spend (B-29), a reinstatement moves a balance, so the caller (electrum/listener.py:refresh_user) must have the other servers corroborate each candidate before reinstate_reappeared_utxos writes it. An entry whose amount differs from our record is not a candidate: it is not the coin we wrote off.""" flagged = ( await session.scalars( select(UtxoEvent).where(UtxoEvent.user_id == user_id, UtxoEvent.spent_txid == _EXTERNAL_SPEND_SENTINEL) ) ).all() amounts = {(row.txid, row.vout): row.amount_sats for row in flagged} return [ entry for entry in entries if amounts.get((entry["tx_hash"], entry["tx_pos"])) == entry.get("value") and (entry.get("height") or 0) > 0 ] async def reinstate_reappeared_utxos(session: AsyncSession, user_id: int, entries: list[dict]) -> int: """The reverse of a mark applied by find_utxos_missing_from/ mark_utxos_spent_externally (B-29): if an outpoint we'd previously flagged as spent outside the platform reappears as unspent in a later listunspent, undo the mark instead of leaving it permanent no matter what the chain says afterwards. Persistence only, for entries the caller has **already corroborated** against the other servers (B-83, ELC-25) — it used to run on the active server's word alone, so a hostile one could put a spent coin back into a balance. Re-checks, under the write lock, that each row still carries the mark before clearing it (DEP-08). """ current_keys = {(entry["tx_hash"], entry["tx_pos"]) for entry in entries} # B-88: the lock is only worth taking when there is something to write, and this # runs on every refresh — so look first (plain columns), then take it and read the # rows again under it. flagged = ( await session.execute( select(UtxoEvent.txid, UtxoEvent.vout).where( UtxoEvent.user_id == user_id, UtxoEvent.spent_txid == _EXTERNAL_SPEND_SENTINEL ) ) ).all() if not any((txid, vout) in current_keys for txid, vout in flagged): return 0 await begin_immediate(session) marked_rows = ( await session.scalars( select(UtxoEvent).where( UtxoEvent.user_id == user_id, UtxoEvent.spent_txid == _EXTERNAL_SPEND_SENTINEL ) ) ).all() reinstated = 0 for row in marked_rows: if (row.txid, row.vout) not in current_keys: continue row.spent_txid = None await write_audit_log( session, "utxo_external_spend_reinstated", {"txid": row.txid, "vout": row.vout, "amount_sats": row.amount_sats}, user_id=user_id, ) reinstated += 1 if reinstated: await session.flush() await recompute_balance(session, user_id) await session.commit() broadcaster.publish() return reinstated async def find_utxos_missing_from(session: AsyncSession, user_id: int, entries: list[dict]) -> list[UtxoEvent]: """Candidates for an external spend (B-29): unspent UTXOs the DB believes this user still holds that are absent from `entries`, this address's current listunspent. Everything the platform itself spends (bets, withdrawals, payouts) sets spent_txid at broadcast time, before the tx ever reaches the chain — so an outpoint still marked unspent in our own DB that Electrum no longer reports as unspent was never on our own radar. Returning a row here is *not* proof it was actually spent — only that this one server's reply no longer lists it. A single broken, behind, or malicious server could otherwise zero a user's balance on one bad reply, which is why the caller (electrum/listener.py:refresh_user) must independently corroborate each candidate against other configured servers before treating it as genuine, rather than this function marking anything itself. An entirely empty `entries` for an address the DB believes is funded returns no candidates at all: it would otherwise flag every one of this user's UTXOs as missing from a single reply, which is a strong sign of an incomplete or broken response rather than N independent spends landing in the same refresh. """ current_keys = {(entry["tx_hash"], entry["tx_pos"]) for entry in entries} unspent_rows = ( await session.scalars( select(UtxoEvent).where(UtxoEvent.user_id == user_id, UtxoEvent.spent_txid.is_(None)) ) ).all() if not entries and unspent_rows: logger.warning( "listunspent for user_id=%s returned no entries at all while %s UTXO(s) are still recorded " "unspent — treating this as an incomplete response rather than a full external sweep", user_id, len(unspent_rows), ) return [] return [row for row in unspent_rows if (row.txid, row.vout) not in current_keys] async def mark_utxos_spent_externally(session: AsyncSession, user_id: int, utxo_ids: list[int]) -> int: """Applies the external-spend sentinel to UTXOs the caller has already corroborated against other servers (B-29) — this function does no verification of its own, only persistence, so it never runs with a session held open across the network calls that verification needs. Re-checks each row is still unspent before applying the mark: something else may have resolved it (a legitimate platform spend, or a prior refresh) between when the caller read the candidate list and finished corroborating it. """ await begin_immediate(session) # B-88: that re-check and the mark are one transaction marked = 0 for utxo_id in utxo_ids: row = await session.get(UtxoEvent, utxo_id) if row is None or row.spent_txid is not None: continue row.spent_txid = _EXTERNAL_SPEND_SENTINEL await write_audit_log( session, "utxo_spent_externally", {"txid": row.txid, "vout": row.vout, "amount_sats": row.amount_sats}, user_id=user_id, ) marked += 1 if marked: await session.flush() await recompute_balance(session, user_id) await session.commit() broadcaster.publish() return marked