import logging from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.log import write_audit_log from app.db.models import UtxoEvent from app.rounds.events import broadcaster from app.wallet.balance import recompute_balance logger = logging.getLogger(__name__) 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. """ 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 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 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. Cheap and purely DB-side — always safe to run on every refresh. """ current_keys = {(entry["tx_hash"], entry["tx_pos"]) for entry in entries} 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. """ 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