Implement full MVP: auth, HD wallet, Electrum client, deposits, bets, round/draw engine, payout, withdrawals, RBF, admin+audit

All 10 build-order stages complete and unit-tested (49 tests). Verified live on
mainnet: registration/address derivation, deposit crediting, a real 10 PLM bet
(broadcast + confirmed + change credited). A full round close->draw->payout
cycle was triggered live and was in progress at commit time. Withdrawal and RBF
bump are unit-tested but not yet exercised against a live broadcast. Known gaps
(scheduler doesn't resume mid-flight rounds after restart, payout has no retry,
no deployment setup, etc.) are documented in CLAUDE.md.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

app/api/routes/admin.py

@@ -0,0 +1,44 @@
+from fastapi import APIRouter, Depends, Header, HTTPException, status
+from pydantic import BaseModel
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.config import settings
+from app.db.session import get_session
+from app.rounds.config import get_round_config
+
+router = APIRouter(prefix="/admin", tags=["admin"])
+
+
+async def require_admin(x_admin_token: str = Header(default="")) -> None:
+    if not settings.admin_token or x_admin_token != settings.admin_token:
+        raise HTTPException(status.HTTP_403_FORBIDDEN, "invalid admin token")
+
+
+class RoundConfigResponse(BaseModel):
+    fee_address: str
+    bet_amount_sats: int
+
+
+class RoundConfigUpdate(BaseModel):
+    fee_address: str | None = None
+    bet_amount_sats: int | None = None
+
+
+@router.get("/config", response_model=RoundConfigResponse, dependencies=[Depends(require_admin)])
+async def read_config(session: AsyncSession = Depends(get_session)) -> RoundConfigResponse:
+    config = await get_round_config(session)
+    await session.commit()
+    return RoundConfigResponse(fee_address=config.fee_address, bet_amount_sats=config.bet_amount_sats)
+
+
+@router.put("/config", response_model=RoundConfigResponse, dependencies=[Depends(require_admin)])
+async def update_config(
+    body: RoundConfigUpdate, session: AsyncSession = Depends(get_session)
+) -> RoundConfigResponse:
+    config = await get_round_config(session)
+    if body.fee_address is not None:
+        config.fee_address = body.fee_address
+    if body.bet_amount_sats is not None:
+        config.bet_amount_sats = body.bet_amount_sats
+    await session.commit()
+    return RoundConfigResponse(fee_address=config.fee_address, bet_amount_sats=config.bet_amount_sats)

app/api/routes/bets.py

@@ -0,0 +1,41 @@
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from pydantic import BaseModel
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.auth.dependencies import get_current_user
+from app.bets.service import BetError, place_bet
+from app.db.models import User
+from app.db.session import get_session
+
+router = APIRouter(prefix="/bets", tags=["bets"])
+
+
+class BetResponse(BaseModel):
+    round_id: int
+    bet_txid: str
+    bet_amount_sats: int
+    status: str
+
+
+@router.post("", response_model=BetResponse, status_code=status.HTTP_201_CREATED)
+async def create_bet(
+    request: Request,
+    user: User = Depends(get_current_user),
+    session: AsyncSession = Depends(get_session),
+) -> BetResponse:
+    listener = request.app.state.electrum_listener
+    if listener.client is None:
+        raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, "not connected to the network, try again shortly")
+
+    async with request.app.state.user_locks.acquire(user.id):
+        try:
+            participant = await place_bet(session, listener.client, user)
+        except BetError as exc:
+            raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
+
+    return BetResponse(
+        round_id=participant.round_id,
+        bet_txid=participant.bet_txid,
+        bet_amount_sats=participant.bet_amount_sats,
+        status=participant.status,
+    )

app/api/routes/users.py

@@ -0,0 +1,18 @@
+from fastapi import APIRouter, Depends
+from pydantic import BaseModel
+
+from app.auth.dependencies import get_current_user
+from app.db.models import User
+
+router = APIRouter(prefix="/users", tags=["users"])
+
+
+class MeResponse(BaseModel):
+    username: str
+    address: str
+    balance_sats: int
+
+
+@router.get("/me", response_model=MeResponse)
+async def me(user: User = Depends(get_current_user)) -> MeResponse:
+    return MeResponse(username=user.username, address=user.address, balance_sats=user.cached_balance_sats)

app/api/routes/withdrawals.py

@@ -0,0 +1,49 @@
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from pydantic import BaseModel
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.auth.dependencies import get_current_user
+from app.db.models import User
+from app.db.session import get_session
+from app.withdrawals.service import WithdrawalError, request_withdrawal
+
+router = APIRouter(prefix="/withdrawals", tags=["withdrawals"])
+
+
+class WithdrawalRequest(BaseModel):
+    external_address: str
+    amount_sats: int
+
+
+class WithdrawalResponse(BaseModel):
+    txid: str
+    amount_requested_sats: int
+    amount_sent_sats: int
+    status: str
+
+
+@router.post("", response_model=WithdrawalResponse, status_code=status.HTTP_201_CREATED)
+async def create_withdrawal(
+    body: WithdrawalRequest,
+    request: Request,
+    user: User = Depends(get_current_user),
+    session: AsyncSession = Depends(get_session),
+) -> WithdrawalResponse:
+    listener = request.app.state.electrum_listener
+    if listener.client is None:
+        raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, "not connected to the network, try again shortly")
+
+    async with request.app.state.user_locks.acquire(user.id):
+        try:
+            withdrawal = await request_withdrawal(
+                session, listener.client, user, body.external_address, body.amount_sats
+            )
+        except WithdrawalError as exc:
+            raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
+
+    return WithdrawalResponse(
+        txid=withdrawal.txid,
+        amount_requested_sats=withdrawal.amount_requested_sats,
+        amount_sent_sats=withdrawal.amount_sent_sats,
+        status=withdrawal.status,
+    )

app/audit/log.py

@@ -0,0 +1,22 @@
+import json
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models import AuditLog
+
+
+async def write_audit_log(
+    session: AsyncSession,
+    event_type: str,
+    payload: dict,
+    user_id: int | None = None,
+    round_id: int | None = None,
+) -> None:
+    session.add(
+        AuditLog(
+            event_type=event_type,
+            payload_json=json.dumps(payload),
+            user_id=user_id,
+            round_id=round_id,
+        )
+    )

app/auth/dependencies.py

@@ -0,0 +1,25 @@
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.auth.security import decode_access_token
+from app.db.models import User
+from app.db.session import get_session
+
+_bearer = HTTPBearer()
+
+
+async def get_current_user(
+    credentials: HTTPAuthorizationCredentials = Depends(_bearer),
+    session: AsyncSession = Depends(get_session),
+) -> User:
+    try:
+        user_id = decode_access_token(credentials.credentials)
+    except Exception as exc:
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid token") from exc
+
+    user = await session.scalar(select(User).where(User.id == user_id))
+    if user is None:
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "user not found")
+    return user

app/auth/routes.py

@@ -0,0 +1,70 @@
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from pydantic import BaseModel
+from sqlalchemy import func, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.auth.security import create_access_token, hash_password, verify_password
+from app.db.models import User
+from app.db.session import get_session
+from app.wallet.hd import derive_user_address
+
+router = APIRouter(prefix="/auth", tags=["auth"])
+
+_MAX_REGISTER_RETRIES = 5
+
+
+class RegisterRequest(BaseModel):
+    username: str
+    password: str
+
+
+class TokenResponse(BaseModel):
+    access_token: str
+    address: str
+
+
+@router.post("/register", response_model=TokenResponse, status_code=status.HTTP_201_CREATED)
+async def register(
+    body: RegisterRequest, request: Request, session: AsyncSession = Depends(get_session)
+) -> TokenResponse:
+    existing = await session.scalar(select(User).where(User.username == body.username))
+    if existing is not None:
+        raise HTTPException(status.HTTP_409_CONFLICT, "username already taken")
+
+    password_hash = hash_password(body.password)
+
+    for _ in range(_MAX_REGISTER_RETRIES):
+        max_index = await session.scalar(select(func.max(User.derivation_index)))
+        next_index = 0 if max_index is None else max_index + 1
+        address = derive_user_address(next_index)
+        user = User(
+            username=body.username,
+            password_hash=password_hash,
+            derivation_index=next_index,
+            address=address,
+        )
+        session.add(user)
+        try:
+            await session.commit()
+        except IntegrityError:
+            await session.rollback()
+            continue
+        await session.refresh(user)
+        request.app.state.electrum_listener.address_for_new_user(user.id, user.address)
+        return TokenResponse(access_token=create_access_token(user.id), address=user.address)
+
+    raise HTTPException(status.HTTP_409_CONFLICT, "could not allocate a derivation index, retry")
+
+
+class LoginRequest(BaseModel):
+    username: str
+    password: str
+
+
+@router.post("/login", response_model=TokenResponse)
+async def login(body: LoginRequest, session: AsyncSession = Depends(get_session)) -> TokenResponse:
+    user = await session.scalar(select(User).where(User.username == body.username))
+    if user is None or not verify_password(body.password, user.password_hash):
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid credentials")
+    return TokenResponse(access_token=create_access_token(user.id), address=user.address)

app/auth/security.py

@@ -0,0 +1,31 @@
+from datetime import datetime, timedelta, timezone
+
+import jwt
+from argon2 import PasswordHasher
+from argon2.exceptions import VerifyMismatchError
+
+from app.config import settings
+
+_hasher = PasswordHasher()
+
+
+def hash_password(password: str) -> str:
+    return _hasher.hash(password)
+
+
+def verify_password(password: str, password_hash: str) -> bool:
+    try:
+        return _hasher.verify(password_hash, password)
+    except VerifyMismatchError:
+        return False
+
+
+def create_access_token(user_id: int) -> str:
+    expires_at = datetime.now(timezone.utc) + timedelta(minutes=settings.jwt_expire_minutes)
+    payload = {"sub": str(user_id), "exp": expires_at}
+    return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
+
+
+def decode_access_token(token: str) -> int:
+    payload = jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm])
+    return int(payload["sub"])

app/bets/confirmation.py

@@ -0,0 +1,19 @@
+from datetime import datetime, timezone
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models import PendingTransaction, RoundParticipant
+from app.tx.confirmation import register_handler
+
+
+async def _on_bet_confirmed(session: AsyncSession, pending: PendingTransaction) -> None:
+    participant = await session.scalar(
+        select(RoundParticipant).where(RoundParticipant.bet_txid == pending.current_txid)
+    )
+    if participant is not None and participant.status == "broadcast":
+        participant.status = "confirmed"
+        participant.confirmed_at = datetime.now(timezone.utc)
+
+
+register_handler("bet", _on_bet_confirmed)

app/bets/service.py

@@ -0,0 +1,104 @@
+from datetime import datetime, timezone
+
+from embit import script
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.audit.log import write_audit_log
+from app.config import settings
+from app.db.models import PendingTransaction, RoundParticipant, User, UtxoEvent
+from app.electrum.client import ElectrumClient
+from app.rounds.config import get_round_config
+from app.rounds.service import open_new_round_if_needed
+from app.wallet.balance import recompute_balance
+from app.wallet.hd import derive_pool_address, derive_user_key
+from app.wallet.psbt_builder import BuiltTransaction, InsufficientFundsError, Utxo, build_signed_transaction
+
+
+class BetError(Exception):
+    pass
+
+
+async def place_bet(session: AsyncSession, client: ElectrumClient, user: User) -> RoundParticipant:
+    round_ = await open_new_round_if_needed(session)
+    if round_.status != "open":
+        raise BetError("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("you already have an active bet in the current round")
+
+    config = await get_round_config(session)
+    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")
+
+    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=settings.fee_rate_sat_vb,
+        )
+    except InsufficientFundsError as exc:
+        raise BetError(str(exc)) from exc
+
+    await client.broadcast(built.raw_hex)
+
+    spent_by_key = {(u.txid, u.vout): u for u in unspent}
+    for spent in built.spent_utxos:
+        row = spent_by_key[(spent.txid, spent.vout)]
+        row.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="broadcast",
+    )
+    session.add(participant)
+    session.add(_pending_transaction(round_.id, user.id, built))
+    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)
+    return participant
+
+
+def _pending_transaction(round_id: int, user_id: int, built: BuiltTransaction) -> PendingTransaction:
+    return PendingTransaction(
+        kind="bet",
+        round_id=round_id,
+        user_id=user_id,
+        current_txid=built.txid,
+        fee_rate_sat_vb=settings.fee_rate_sat_vb,
+        raw_tx_hex=built.raw_hex,
+        status="pending",
+    )

app/config.py

@@ -0,0 +1,29 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
+
+    database_url: str = "sqlite+aiosqlite:///./plm_lottery.db"
+
+    electrum_host: str = "santantonio.sytes.net"
+    electrum_port: int = 50002
+    electrum_use_ssl: bool = True
+
+    xprv_encryption_key: str = ""
+    master_key_path: str = "./master.xprv.enc"
+    jwt_secret: str = ""
+    jwt_algorithm: str = "HS256"
+    jwt_expire_minutes: int = 60 * 24
+    admin_token: str = ""
+
+    round_duration_seconds: int = 600
+
+    bet_amount_sats: int = 10 * 100_000_000
+    min_amount_sats: int = 1 * 100_000_000
+    confirmations_required: int = 1
+    fee_rate_sat_vb: int = 1
+    rbf_timeout_seconds: int = 900
+
+
+settings = Settings()

app/db/base.py

@@ -0,0 +1,11 @@
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from sqlalchemy.orm import DeclarativeBase
+
+from app.config import settings
+
+engine = create_async_engine(settings.database_url)
+AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+
+
+class Base(DeclarativeBase):
+    pass

app/db/models.py

@@ -0,0 +1,131 @@
+from datetime import datetime, timezone
+
+from sqlalchemy import BigInteger, ForeignKey, String, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+
+def utcnow() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    username: Mapped[str] = mapped_column(String(64), unique=True, index=True)
+    password_hash: Mapped[str] = mapped_column(String(256))
+    derivation_index: Mapped[int] = mapped_column(unique=True)
+    address: Mapped[str] = mapped_column(String(128), unique=True)
+    # Read cache only; must always be written in the same transaction as the
+    # utxo_events rows it summarizes. Source of truth is utxo_events.
+    cached_balance_sats: Mapped[int] = mapped_column(BigInteger, default=0)
+    created_at: Mapped[datetime] = mapped_column(default=utcnow)
+
+
+class UtxoEvent(Base):
+    __tablename__ = "utxo_events"
+    __table_args__ = (UniqueConstraint("txid", "vout"),)
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)
+    txid: Mapped[str] = mapped_column(String(64))
+    vout: Mapped[int]
+    amount_sats: Mapped[int] = mapped_column(BigInteger)
+    confirmed_height: Mapped[int]
+    confirmed_at: Mapped[datetime] = mapped_column(default=utcnow)
+    # Set once this UTXO is consumed by an outgoing bet/withdrawal build.
+    spent_txid: Mapped[str | None] = mapped_column(String(64), default=None)
+
+
+class Round(Base):
+    __tablename__ = "rounds"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    status: Mapped[str] = mapped_column(String(16), default="open")
+    opened_at: Mapped[datetime] = mapped_column(default=utcnow)
+    closed_at: Mapped[datetime | None] = mapped_column(default=None)
+    draw_block_height: Mapped[int | None] = mapped_column(default=None)
+    draw_block_hash: Mapped[str | None] = mapped_column(String(64), default=None)
+    seed_int: Mapped[str | None] = mapped_column(String(128), default=None)
+    winner_user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"), default=None)
+    pool_amount_sats: Mapped[int | None] = mapped_column(BigInteger, default=None)
+    winner_amount_sats: Mapped[int | None] = mapped_column(BigInteger, default=None)
+    fee_amount_sats: Mapped[int | None] = mapped_column(BigInteger, default=None)
+    payout_txid: Mapped[str | None] = mapped_column(String(64), default=None)
+
+
+class RoundParticipant(Base):
+    __tablename__ = "round_participants"
+    __table_args__ = (UniqueConstraint("round_id", "user_id"),)
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    round_id: Mapped[int] = mapped_column(ForeignKey("rounds.id"), index=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)
+    bet_amount_sats: Mapped[int] = mapped_column(BigInteger)
+    bet_txid: Mapped[str] = mapped_column(String(64))
+    # Ordering / tie-break field per spec: broadcast time, not confirmation time.
+    broadcast_at: Mapped[datetime] = mapped_column(default=utcnow)
+    confirmed_at: Mapped[datetime | None] = mapped_column(default=None)
+    status: Mapped[str] = mapped_column(String(16), default="broadcast")
+
+
+class RoundConfig(Base):
+    """Single-row operational config, DB-backed so it's editable without a redeploy.
+
+    round_duration is intentionally NOT here: it stays env-var-driven per spec.
+    Don't move it here without an explicit decision to change that.
+    """
+
+    __tablename__ = "round_config"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    fee_address: Mapped[str] = mapped_column(String(128))
+    bet_amount_sats: Mapped[int] = mapped_column(BigInteger)
+    updated_at: Mapped[datetime] = mapped_column(default=utcnow, onupdate=utcnow)
+
+
+class PendingTransaction(Base):
+    """Single source of truth for the RBF timeout->bump->rebroadcast loop."""
+
+    __tablename__ = "pending_transactions"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    kind: Mapped[str] = mapped_column(String(16))  # bet | payout | withdrawal
+    round_id: Mapped[int | None] = mapped_column(ForeignKey("rounds.id"), default=None)
+    withdrawal_id: Mapped[int | None] = mapped_column(ForeignKey("withdrawals.id"), default=None)
+    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"), default=None)
+    current_txid: Mapped[str] = mapped_column(String(64))
+    fee_rate_sat_vb: Mapped[int]
+    raw_tx_hex: Mapped[str] = mapped_column(String)
+    broadcast_at: Mapped[datetime] = mapped_column(default=utcnow)
+    status: Mapped[str] = mapped_column(String(16), default="pending")
+    replaced_by_txid: Mapped[str | None] = mapped_column(String(64), default=None)
+    attempt_count: Mapped[int] = mapped_column(default=1)
+    updated_at: Mapped[datetime] = mapped_column(default=utcnow, onupdate=utcnow)
+
+
+class Withdrawal(Base):
+    __tablename__ = "withdrawals"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)
+    external_address: Mapped[str] = mapped_column(String(128))
+    amount_requested_sats: Mapped[int] = mapped_column(BigInteger)
+    amount_sent_sats: Mapped[int | None] = mapped_column(BigInteger, default=None)
+    txid: Mapped[str | None] = mapped_column(String(64), default=None)
+    status: Mapped[str] = mapped_column(String(16), default="pending")
+    created_at: Mapped[datetime] = mapped_column(default=utcnow)
+    confirmed_at: Mapped[datetime | None] = mapped_column(default=None)
+
+
+class AuditLog(Base):
+    __tablename__ = "audit_log"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    event_type: Mapped[str] = mapped_column(String(32))
+    payload_json: Mapped[str] = mapped_column(String)
+    user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"), default=None)
+    round_id: Mapped[int | None] = mapped_column(ForeignKey("rounds.id"), default=None)
+    created_at: Mapped[datetime] = mapped_column(default=utcnow)

app/db/session.py

@@ -0,0 +1,10 @@
+from collections.abc import AsyncGenerator
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.base import AsyncSessionLocal
+
+
+async def get_session() -> AsyncGenerator[AsyncSession, None]:
+    async with AsyncSessionLocal() as session:
+        yield session

app/deposits/service.py

@@ -0,0 +1,54 @@
+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.wallet.balance import recompute_balance
+
+
+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()
+
+    return newly_credited

app/electrum/client.py

@@ -0,0 +1,114 @@
+import asyncio
+import itertools
+import json
+import ssl
+
+
+class ElectrumError(Exception):
+    pass
+
+
+class ElectrumClient:
+    """Minimal asyncio Electrum protocol client: line-delimited JSON-RPC over TLS.
+
+    Push notifications (blockchain.headers.subscribe, blockchain.scripthash.subscribe)
+    arrive under the *same* method name as the subscribe call, multiplexed for every
+    scripthash subscribed — callers read `notifications(method)` and, for scripthash
+    pushes, dispatch on `params[0]` (the scripthash) themselves.
+    """
+
+    def __init__(self, host: str, port: int, use_ssl: bool = True):
+        self.host = host
+        self.port = port
+        self.use_ssl = use_ssl
+        self._reader: asyncio.StreamReader | None = None
+        self._writer: asyncio.StreamWriter | None = None
+        self._id_counter = itertools.count(1)
+        self._pending: dict[int, asyncio.Future] = {}
+        self._subscriptions: dict[str, asyncio.Queue] = {}
+        self._read_task: asyncio.Task | None = None
+
+    async def connect(self) -> None:
+        # Electrum servers commonly present self-signed certs; the protocol's trust
+        # model is server consensus, not TLS PKI, so we only use SSL for transport
+        # encryption and don't verify the certificate chain/hostname.
+        ssl_context = None
+        if self.use_ssl:
+            ssl_context = ssl.create_default_context()
+            ssl_context.check_hostname = False
+            ssl_context.verify_mode = ssl.CERT_NONE
+        self._reader, self._writer = await asyncio.open_connection(self.host, self.port, ssl=ssl_context)
+        self._read_task = asyncio.create_task(self._read_loop())
+        await self.request("server.version", ["plm-lottery", "1.4"])
+
+    async def close(self) -> None:
+        if self._read_task is not None:
+            self._read_task.cancel()
+        if self._writer is not None:
+            self._writer.close()
+            try:
+                await asyncio.wait_for(self._writer.wait_closed(), timeout=2)
+            except (ssl.SSLError, TimeoutError, asyncio.TimeoutError):
+                pass  # some Electrum servers don't send a clean TLS close_notify
+
+    async def request(self, method: str, params: list | None = None) -> object:
+        if self._writer is None:
+            raise ElectrumError("not connected")
+        request_id = next(self._id_counter)
+        future: asyncio.Future = asyncio.get_event_loop().create_future()
+        self._pending[request_id] = future
+        payload = json.dumps({"id": request_id, "method": method, "params": params or []}) + "\n"
+        self._writer.write(payload.encode())
+        await self._writer.drain()
+        return await future
+
+    def notifications(self, method: str) -> asyncio.Queue:
+        return self._subscriptions.setdefault(method, asyncio.Queue())
+
+    async def subscribe_headers(self) -> dict:
+        self.notifications("blockchain.headers.subscribe")
+        return await self.request("blockchain.headers.subscribe")
+
+    async def subscribe_scripthash(self, scripthash: str) -> str | None:
+        self.notifications("blockchain.scripthash.subscribe")
+        return await self.request("blockchain.scripthash.subscribe", [scripthash])
+
+    async def listunspent(self, scripthash: str) -> list[dict]:
+        return await self.request("blockchain.scripthash.listunspent", [scripthash])
+
+    async def broadcast(self, raw_tx_hex: str) -> str:
+        return await self.request("blockchain.transaction.broadcast", [raw_tx_hex])
+
+    async def get_transaction(self, txid: str, verbose: bool = False) -> object:
+        return await self.request("blockchain.transaction.get", [txid, verbose])
+
+    async def _read_loop(self) -> None:
+        assert self._reader is not None
+        try:
+            while True:
+                line = await self._reader.readline()
+                if not line:
+                    break
+                message = json.loads(line)
+                self._dispatch(message)
+        finally:
+            error = ElectrumError("connection closed")
+            for future in self._pending.values():
+                if not future.done():
+                    future.set_exception(error)
+            self._pending.clear()
+
+    def _dispatch(self, message: dict) -> None:
+        message_id = message.get("id")
+        if message_id is not None and message_id in self._pending:
+            future = self._pending.pop(message_id)
+            if future.done():
+                return
+            if message.get("error"):
+                future.set_exception(ElectrumError(message["error"]))
+            else:
+                future.set_result(message.get("result"))
+        elif "method" in message:
+            queue = self._subscriptions.get(message["method"])
+            if queue is not None:
+                queue.put_nowait(message.get("params"))

app/electrum/listener.py

@@ -0,0 +1,111 @@
+import asyncio
+import logging
+from collections.abc import Callable
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import async_sessionmaker
+
+from app.db.models import User
+from app.deposits.service import credit_confirmed_utxos
+from app.electrum.client import ElectrumClient
+from app.electrum.scripthash import address_to_scripthash
+
+logger = logging.getLogger(__name__)
+
+
+class ElectrumListener:
+    """Long-lived background task: keeps one Electrum connection open, subscribes
+    every user's address (plus any address added later via add_address), and
+    credits confirmed deposits as scripthash-change notifications arrive.
+
+    Reconnects with backoff on any failure; a fresh connection re-subscribes to
+    every user pulled straight from the DB, so no in-memory subscription state is
+    ever a stale source of truth.
+    """
+
+    def __init__(self, client_factory: Callable[[], ElectrumClient], session_factory: async_sessionmaker):
+        self._client_factory = client_factory
+        self._session_factory = session_factory
+        self._scripthash_to_user: dict[str, int] = {}
+        self.tip_height: int = 0
+        self.tip_header_hex: str | None = None
+        self.client: ElectrumClient | None = None
+
+    def address_for_new_user(self, user_id: int, address: str) -> None:
+        """Called right after a user registers so their deposit address starts
+        being watched immediately, without waiting for the next reconnect cycle."""
+        scripthash = address_to_scripthash(address)
+        self._scripthash_to_user[scripthash] = user_id
+        if self.client is not None:
+            asyncio.create_task(self._subscribe_and_refresh(scripthash, user_id))
+
+    async def run(self) -> None:
+        backoff = 1
+        while True:
+            try:
+                await self._run_once()
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                logger.exception("Electrum listener error, reconnecting in %ss", backoff)
+                self.client = None
+                await asyncio.sleep(backoff)
+                backoff = min(backoff * 2, 30)
+                continue
+            backoff = 1
+
+    async def _run_once(self) -> None:
+        client = self._client_factory()
+        await client.connect()
+        self.client = client
+
+        header = await client.subscribe_headers()
+        self.tip_height = header["height"]
+        self.tip_header_hex = header.get("hex")
+
+        await self._subscribe_all_users()
+
+        headers_queue = client.notifications("blockchain.headers.subscribe")
+        scripthash_queue = client.notifications("blockchain.scripthash.subscribe")
+        try:
+            await asyncio.gather(
+                self._consume_headers(headers_queue),
+                self._consume_scripthash(scripthash_queue),
+            )
+        finally:
+            await client.close()
+
+    async def _subscribe_all_users(self) -> None:
+        async with self._session_factory() as session:
+            users = (await session.scalars(select(User))).all()
+        for user in users:
+            scripthash = address_to_scripthash(user.address)
+            self._scripthash_to_user[scripthash] = user.id
+            await self._subscribe_and_refresh(scripthash, user.id)
+
+    async def _subscribe_and_refresh(self, scripthash: str, user_id: int) -> None:
+        assert self.client is not None
+        await self.client.subscribe_scripthash(scripthash)
+        await self._refresh_user(user_id, scripthash)
+
+    async def _consume_headers(self, queue: asyncio.Queue) -> None:
+        while True:
+            params = await queue.get()
+            for header in params:
+                self.tip_height = header["height"]
+                self.tip_header_hex = header.get("hex")
+
+    async def _consume_scripthash(self, queue: asyncio.Queue) -> None:
+        while True:
+            scripthash, _status = await queue.get()
+            user_id = self._scripthash_to_user.get(scripthash)
+            if user_id is not None:
+                await self._refresh_user(user_id, scripthash)
+
+    async def _refresh_user(self, user_id: int, scripthash: str) -> None:
+        assert self.client is not None
+        entries = await self.client.listunspent(scripthash)
+        async with self._session_factory() as session:
+            credited = await credit_confirmed_utxos(session, user_id, entries)
+        if credited:
+            logger.info("credited %s new UTXO(s) for user_id=%s", credited, user_id)

app/electrum/scripthash.py

@@ -0,0 +1,15 @@
+import hashlib
+
+from embit.script import Script
+
+
+def address_to_scripthash(address: str) -> str:
+    """Electrum protocol scripthash: sha256(scriptPubKey), byte-reversed, hex.
+
+    Uses `.data` (the raw scriptPubKey bytes), not `.serialize()` — the latter
+    prefixes a compact-size length byte meant for embedding the script as pushdata
+    elsewhere (e.g. a P2SH redeemScript), which is not part of the actual on-chain
+    output script and produces a wrong (unmatchable) scripthash if used here.
+    """
+    script_pubkey = Script.from_address(address).data
+    return hashlib.sha256(script_pubkey).digest()[::-1].hex()

app/main.py

@@ -0,0 +1,63 @@
+import asyncio
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+
+import app.bets.confirmation  # noqa: F401  (registers the "bet" confirmation handler)
+import app.rounds.confirmation  # noqa: F401  (registers the "payout" confirmation handler)
+import app.withdrawals.confirmation  # noqa: F401  (registers the "withdrawal" confirmation handler)
+from app.api.routes.admin import router as admin_router
+from app.api.routes.bets import router as bets_router
+from app.api.routes.users import router as users_router
+from app.api.routes.withdrawals import router as withdrawals_router
+from app.auth.routes import router as auth_router
+from app.config import settings
+from app.db.base import AsyncSessionLocal
+from app.electrum.client import ElectrumClient
+from app.electrum.listener import ElectrumListener
+from app.rounds.scheduler import RoundScheduler
+from app.tx.broadcast import RbfBumper
+from app.tx.confirmation import ConfirmationPoller
+from app.tx.locks import UserLocks
+
+
+def _make_electrum_client() -> ElectrumClient:
+    return ElectrumClient(settings.electrum_host, settings.electrum_port, settings.electrum_use_ssl)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    listener = ElectrumListener(_make_electrum_client, AsyncSessionLocal)
+    app.state.electrum_listener = listener
+    app.state.user_locks = UserLocks()
+
+    scheduler = RoundScheduler(AsyncSessionLocal, listener)
+    poller = ConfirmationPoller(AsyncSessionLocal, lambda: listener.client)
+    bumper = RbfBumper(AsyncSessionLocal, lambda: listener.client)
+
+    tasks = [
+        asyncio.create_task(listener.run()),
+        asyncio.create_task(scheduler.run()),
+        asyncio.create_task(poller.run()),
+        asyncio.create_task(bumper.run()),
+    ]
+    try:
+        yield
+    finally:
+        for task in tasks:
+            task.cancel()
+        if listener.client is not None:
+            await listener.client.close()
+
+
+app = FastAPI(title="PLM Lottery", lifespan=lifespan)
+app.include_router(auth_router)
+app.include_router(users_router)
+app.include_router(bets_router)
+app.include_router(withdrawals_router)
+app.include_router(admin_router)
+
+
+@app.get("/health")
+async def health() -> dict[str, str]:
+    return {"status": "ok"}

app/rounds/config.py

@@ -0,0 +1,17 @@
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.config import settings
+from app.db.models import RoundConfig
+
+
+async def get_round_config(session: AsyncSession) -> RoundConfig:
+    """Single-row operational config, lazily seeded from settings defaults on
+    first use. fee_address starts empty until an operator sets it (admin
+    endpoint, stage 10) — payouts must refuse to run until it's set."""
+    config = await session.scalar(select(RoundConfig))
+    if config is None:
+        config = RoundConfig(fee_address="", bet_amount_sats=settings.bet_amount_sats)
+        session.add(config)
+        await session.flush()
+    return config

app/rounds/confirmation.py

@@ -0,0 +1,16 @@
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models import PendingTransaction, Round
+from app.tx.confirmation import register_handler
+
+
+async def _on_payout_confirmed(session: AsyncSession, pending: PendingTransaction) -> None:
+    round_ = await session.scalar(select(Round).where(Round.payout_txid == pending.current_txid))
+    if round_ is not None and round_.status == "paying_out":
+        round_.status = "closed"
+        # The winner's own address is already watched by the Electrum listener, so
+        # their balance is credited by the normal deposit path once this confirms.
+
+
+register_handler("payout", _on_payout_confirmed)

app/rounds/draw.py

@@ -0,0 +1,21 @@
+import hashlib
+
+
+def header_hex_to_block_hash(header_hex: str) -> str:
+    """Block hash from a raw Electrum header: sha256d, byte-reversed, hex.
+    Verified against a real mainnet block (blockchain.transaction.get's own
+    reported blockhash) during development."""
+    header_bytes = bytes.fromhex(header_hex)
+    digest = hashlib.sha256(hashlib.sha256(header_bytes).digest()).digest()
+    return digest[::-1].hex()
+
+
+def draw_winner(participants: list[str], block_hash_hex: str) -> str:
+    """v1 draw algorithm (flowchart.mmd, node R): seed = block hash as an integer,
+    index = seed mod participant_count, winner = participants[index]. Anyone can
+    recompute and verify it from public data. Deliberately simple/replaceable."""
+    if not participants:
+        raise ValueError("no participants to draw from")
+    seed = int(block_hash_hex, 16)
+    index = seed % len(participants)
+    return participants[index]

app/rounds/scheduler.py

@@ -0,0 +1,206 @@
+import asyncio
+import logging
+from datetime import datetime, timedelta, timezone
+
+from embit import script
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import async_sessionmaker
+
+from app.audit.log import write_audit_log
+from app.config import settings
+from app.db.models import PendingTransaction, Round, RoundParticipant, User
+from app.electrum.listener import ElectrumListener
+from app.electrum.scripthash import address_to_scripthash
+from app.rounds.config import get_round_config
+from app.rounds.draw import draw_winner, header_hex_to_block_hash
+from app.rounds.service import open_new_round_if_needed
+from app.wallet.hd import derive_pool_key
+from app.wallet.plm_network import PLM_MAINNET
+from app.wallet.psbt_builder import InsufficientFundsError, Utxo, build_payout_transaction
+
+logger = logging.getLogger(__name__)
+
+_TICK_INTERVAL_SECONDS = 5
+
+
+class RoundScheduler:
+    """Background task implementing flowchart.mmd's DRAW subgraph: closes the
+    round on its timer (once any in-flight bets have confirmed), draws a winner
+    from the next confirmed block, and broadcasts the payout. The next round only
+    opens once this one is fully closed (rounds/service.get_active_round)."""
+
+    def __init__(self, session_factory: async_sessionmaker, listener: ElectrumListener):
+        self._session_factory = session_factory
+        self._listener = listener
+
+    async def run(self) -> None:
+        while True:
+            try:
+                await self._tick()
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                logger.exception("round scheduler tick failed")
+            await asyncio.sleep(_TICK_INTERVAL_SECONDS)
+
+    async def _tick(self) -> None:
+        if self._listener.client is None:
+            return
+
+        async with self._session_factory() as session:
+            round_ = await open_new_round_if_needed(session)
+            await session.commit()
+            round_id, status, opened_at = round_.id, round_.status, round_.opened_at
+
+        if status != "open":
+            return  # already closing/drawing/paying_out; progress happens elsewhere
+
+        opened_at = opened_at.replace(tzinfo=timezone.utc)
+        if datetime.now(timezone.utc) < opened_at + timedelta(seconds=settings.round_duration_seconds):
+            return
+
+        async with self._session_factory() as session:
+            pending_count = await session.scalar(
+                select(func.count())
+                .select_from(RoundParticipant)
+                .where(RoundParticipant.round_id == round_id, RoundParticipant.status == "broadcast")
+            )
+        if pending_count:
+            return  # wait for in-flight bets to confirm before closing
+
+        await self._close_and_draw(round_id)
+
+    async def _close_and_draw(self, round_id: int) -> None:
+        async with self._session_factory() as session:
+            round_ = await session.get(Round, round_id)
+            round_.status = "closing"
+            round_.closed_at = datetime.now(timezone.utc)
+
+            participants = (
+                await session.scalars(
+                    select(RoundParticipant)
+                    .where(RoundParticipant.round_id == round_id, RoundParticipant.status == "confirmed")
+                    .order_by(RoundParticipant.broadcast_at)
+                )
+            ).all()
+
+            if not participants:
+                round_.status = "closed"
+                await write_audit_log(session, "round_closed", {"participants": 0}, round_id=round_id)
+                await session.commit()
+                logger.info("round %s closed with no participants", round_id)
+                return
+
+            pool_amount = sum(p.bet_amount_sats for p in participants)
+            addresses: list[str] = []
+            user_by_address: dict[str, int] = {}
+            for p in participants:
+                user = await session.get(User, p.user_id)
+                addresses.append(user.address)
+                user_by_address[user.address] = user.id
+
+            round_.status = "drawing"
+            await session.commit()
+
+        tip_at_close = self._listener.tip_height
+        block_height, block_hash = await self._wait_for_next_block(tip_at_close)
+        winner_address = draw_winner(addresses, block_hash)
+
+        async with self._session_factory() as session:
+            round_ = await session.get(Round, round_id)
+            round_.draw_block_height = block_height
+            round_.draw_block_hash = block_hash
+            round_.seed_int = str(int(block_hash, 16))
+            round_.winner_user_id = user_by_address[winner_address]
+            round_.pool_amount_sats = pool_amount
+            round_.status = "paying_out"
+            await write_audit_log(
+                session,
+                "winner_drawn",
+                {
+                    "winner_address": winner_address,
+                    "pool_amount_sats": pool_amount,
+                    "block_height": block_height,
+                    "block_hash": block_hash,
+                    "participants": len(addresses),
+                },
+                user_id=user_by_address[winner_address],
+                round_id=round_id,
+            )
+            await session.commit()
+
+        logger.info("round %s: winner=%s pool=%s", round_id, winner_address, pool_amount)
+        await self._trigger_payout(round_id)
+
+    async def _wait_for_next_block(self, tip_at_close: int) -> tuple[int, str]:
+        while self._listener.tip_height <= tip_at_close or not self._listener.tip_header_hex:
+            await asyncio.sleep(_TICK_INTERVAL_SECONDS)
+        return self._listener.tip_height, header_hex_to_block_hash(self._listener.tip_header_hex)
+
+    async def _trigger_payout(self, round_id: int) -> None:
+        client = self._listener.client
+        if client is None:
+            logger.error("round %s payout deferred: not connected", round_id)
+            return
+
+        async with self._session_factory() as session:
+            round_ = await session.get(Round, round_id)
+            config = await get_round_config(session)
+            if not config.fee_address:
+                logger.error(
+                    "round %s payout blocked: no fee_address configured (set it via the admin endpoint)", round_id
+                )
+                return
+
+            winner = await session.get(User, round_.winner_user_id)
+            winner_share = round_.pool_amount_sats * 70 // 100
+            commission_share = round_.pool_amount_sats - winner_share  # remainder from rounding goes to fees
+
+            pool_key = derive_pool_key()
+            pool_script_obj = script.p2wpkh(pool_key.to_public())
+            pool_address = pool_script_obj.address(network=PLM_MAINNET)
+            pool_scripthash = address_to_scripthash(pool_address)
+            entries = await client.listunspent(pool_scripthash)
+            utxos = [Utxo(e["tx_hash"], e["tx_pos"], e["value"]) for e in entries if e["height"] > 0]
+
+            try:
+                built = build_payout_transaction(
+                    signing_key=pool_key,
+                    from_script=pool_script_obj,
+                    utxos=utxos,
+                    winner_address=winner.address,
+                    winner_share_sats=winner_share,
+                    fee_address=config.fee_address,
+                    commission_sats=commission_share,
+                    change_address=pool_address,
+                    fee_rate_sat_vb=settings.fee_rate_sat_vb,
+                )
+            except InsufficientFundsError:
+                logger.exception("round %s payout failed: insufficient pool UTXOs", round_id)
+                return
+
+            await client.broadcast(built.raw_hex)
+
+            round_.winner_amount_sats = built.winner_sats
+            round_.fee_amount_sats = built.commission_sats
+            round_.payout_txid = built.txid
+            session.add(
+                PendingTransaction(
+                    kind="payout",
+                    round_id=round_id,
+                    current_txid=built.txid,
+                    fee_rate_sat_vb=settings.fee_rate_sat_vb,
+                    raw_tx_hex=built.raw_hex,
+                    status="pending",
+                )
+            )
+            await write_audit_log(
+                session,
+                "payout_sent",
+                {"txid": built.txid, "winner_sats": built.winner_sats, "commission_sats": built.commission_sats},
+                user_id=round_.winner_user_id,
+                round_id=round_id,
+            )
+            await session.commit()
+
+        logger.info("round %s payout broadcast: txid=%s", round_id, built.txid)

app/rounds/service.py

@@ -0,0 +1,27 @@
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models import Round
+
+_ACTIVE_STATUSES = ("open", "closing", "drawing", "paying_out")
+
+
+async def get_active_round(session: AsyncSession) -> Round | None:
+    """The round currently in progress (in any non-closed state), if any. Rounds
+    never overlap: a new round only opens once the previous one is fully closed
+    (payout confirmed, or no participants to pay out)."""
+    return await session.scalar(select(Round).where(Round.status.in_(_ACTIVE_STATUSES)).order_by(Round.id.desc()))
+
+
+async def open_new_round_if_needed(session: AsyncSession) -> Round:
+    """Returns the active round if one exists (whatever its status), otherwise
+    opens a fresh one. Callers that need to attach a bet must additionally check
+    the returned round's status == "open" — a round in closing/drawing/paying_out
+    isn't accepting new bets, but a new round can't open until it's done."""
+    active = await get_active_round(session)
+    if active is not None:
+        return active
+    round_ = Round(status="open")
+    session.add(round_)
+    await session.flush()
+    return round_

app/tx/broadcast.py

@@ -0,0 +1,163 @@
+import asyncio
+import logging
+from datetime import datetime, timedelta, timezone
+
+from embit import script
+from embit.psbt import PSBT
+from embit.transaction import Transaction, TransactionInput, TransactionOutput
+from embit.finalizer import finalize_psbt
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from app.config import settings
+from app.db.models import PendingTransaction, User
+from app.electrum.client import ElectrumClient
+from app.wallet.hd import derive_pool_key, derive_user_key
+from app.wallet.plm_network import PLM_MAINNET
+from app.wallet.psbt_builder import RBF_SEQUENCE, estimate_vsize
+
+logger = logging.getLogger(__name__)
+
+_POLL_INTERVAL_SECONDS = 30
+_FEE_RATE_INCREMENT = 1  # minimum relay-policy-friendly bump per BIP125
+
+
+class RbfError(Exception):
+    pass
+
+
+def should_bump(pending: PendingTransaction, now: datetime, timeout_seconds: int | None = None) -> bool:
+    """Pure decision: has this pending tx been unconfirmed for longer than the
+    configured timeout? Kept separate from the I/O-heavy bump_fee() so it's
+    trivially unit-testable."""
+    timeout = timeout_seconds if timeout_seconds is not None else settings.rbf_timeout_seconds
+    if pending.status != "pending":
+        return False
+    return now >= pending.broadcast_at.replace(tzinfo=timezone.utc) + timedelta(seconds=timeout)
+
+
+async def _signing_context(session: AsyncSession, pending: PendingTransaction) -> tuple:
+    """Returns (signing_key, own_script, own_address) for the single sender that
+    controls every input of this tx — a user for bet/withdrawal, the pool for
+    payout. All our builders only ever spend one address's UTXOs per tx."""
+    if pending.kind == "payout":
+        key = derive_pool_key()
+    else:
+        user = await session.get(User, pending.user_id)
+        key = derive_user_key(user.derivation_index)
+    own_script = script.p2wpkh(key.to_public())
+    own_address = own_script.address(network=PLM_MAINNET)
+    return key, own_script, own_address
+
+
+async def _prevout_amount(client: ElectrumClient, vin: TransactionInput) -> int:
+    txid_hex = vin.txid.hex()
+    tx = await client.get_transaction(txid_hex, verbose=True)
+    value_coins = tx["vout"][vin.vout]["value"]
+    return round(value_coins * 100_000_000)
+
+
+def _find_change_output(tx: Transaction, change_address: str) -> int | None:
+    for i, out in enumerate(tx.vout):
+        if out.script_pubkey.address(network=PLM_MAINNET) == change_address:
+            return i
+    return None
+
+
+async def bump_fee(session: AsyncSession, client: ElectrumClient, pending: PendingTransaction) -> str:
+    """Rebuild `pending`'s transaction with a higher fee (same inputs, same
+    recipient outputs, the extra fee taken from the change output) and
+    rebroadcast. Returns the new txid.
+
+    Only handles the common case: exactly one change output paying back to the
+    tx's own sender address, large enough to absorb the increase. If there's no
+    such output (e.g. an exact-amount bet with no change), this raises RbfError —
+    bumping such a tx would require selecting additional inputs, which isn't
+    implemented for the MVP; it needs manual operator intervention.
+    """
+    old_tx = Transaction.parse(bytes.fromhex(pending.raw_tx_hex))
+    signing_key, own_script, own_address = await _signing_context(session, pending)
+
+    input_amounts = [await _prevout_amount(client, vin) for vin in old_tx.vin]
+    total_in = sum(input_amounts)
+    old_fee = total_in - sum(o.value for o in old_tx.vout)
+
+    new_fee_rate = pending.fee_rate_sat_vb + _FEE_RATE_INCREMENT
+    new_fee = estimate_vsize(len(old_tx.vin), len(old_tx.vout)) * new_fee_rate
+    fee_delta = new_fee - old_fee
+    if fee_delta <= 0:
+        fee_delta = _FEE_RATE_INCREMENT  # already above the new target vsize*rate; bump by a token amount
+
+    change_index = _find_change_output(old_tx, own_address)
+    if change_index is None or old_tx.vout[change_index].value <= fee_delta:
+        raise RbfError(f"pending_transaction {pending.id}: no change output large enough to absorb a fee bump")
+
+    new_vout = list(old_tx.vout)
+    bumped_change = new_vout[change_index].value - fee_delta
+    new_vout[change_index] = TransactionOutput(bumped_change, new_vout[change_index].script_pubkey)
+
+    new_vin = [TransactionInput(v.txid, v.vout, sequence=RBF_SEQUENCE) for v in old_tx.vin]
+    new_tx = Transaction(vin=new_vin, vout=new_vout)
+    psbt = PSBT(new_tx)
+    for i, amount in enumerate(input_amounts):
+        psbt.inputs[i].witness_utxo = TransactionOutput(amount, own_script)
+
+    signed = psbt.sign_with(signing_key)
+    if signed != len(new_vin):
+        raise RuntimeError(f"expected {len(new_vin)} signatures, got {signed}")
+
+    final_tx = finalize_psbt(psbt)
+    if final_tx is None:
+        raise RuntimeError("failed to finalize bumped PSBT")
+
+    raw_hex = final_tx.serialize().hex()
+    new_txid = final_tx.txid().hex()
+    await client.broadcast(raw_hex)
+
+    pending.current_txid = new_txid
+    pending.raw_tx_hex = raw_hex
+    pending.fee_rate_sat_vb = new_fee_rate
+    pending.attempt_count += 1
+    pending.broadcast_at = datetime.now(timezone.utc)
+    await session.commit()
+
+    logger.info("bumped %s pending_transaction %s: %s -> %s", pending.kind, pending.id, pending.current_txid, new_txid)
+    return new_txid
+
+
+class RbfBumper:
+    def __init__(self, session_factory: async_sessionmaker, get_client):
+        self._session_factory = session_factory
+        self._get_client = get_client
+
+    async def run(self) -> None:
+        while True:
+            client = self._get_client()
+            if client is not None:
+                try:
+                    await self._tick(client)
+                except asyncio.CancelledError:
+                    raise
+                except Exception:
+                    logger.exception("RBF bump tick failed")
+            await asyncio.sleep(_POLL_INTERVAL_SECONDS)
+
+    async def _tick(self, client: ElectrumClient) -> None:
+        now = datetime.now(timezone.utc)
+        async with self._session_factory() as session:
+            candidates = (
+                await session.scalars(select(PendingTransaction).where(PendingTransaction.status == "pending"))
+            ).all()
+            due = [p for p in candidates if should_bump(p, now)]
+
+        for pending in due:
+            async with self._session_factory() as session:
+                row = await session.get(PendingTransaction, pending.id)
+                if row is None or row.status != "pending":
+                    continue
+                try:
+                    await bump_fee(session, client, row)
+                except RbfError:
+                    logger.exception("could not bump pending_transaction %s", row.id)
+                except Exception:
+                    logger.exception("unexpected error bumping pending_transaction %s", row.id)

app/tx/confirmation.py

@@ -0,0 +1,67 @@
+import asyncio
+import logging
+from collections.abc import Awaitable, Callable
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from app.db.models import PendingTransaction
+from app.electrum.client import ElectrumClient
+
+logger = logging.getLogger(__name__)
+
+_POLL_INTERVAL_SECONDS = 10
+
+ConfirmationHandler = Callable[[AsyncSession, PendingTransaction], Awaitable[None]]
+_handlers: dict[str, ConfirmationHandler] = {}
+
+
+def register_handler(kind: str, handler: ConfirmationHandler) -> None:
+    """Domain modules (bets, rounds, withdrawals) register here so this generic
+    poller can notify them when one of their outgoing txs gets its 1st
+    confirmation, without this module importing them directly."""
+    _handlers[kind] = handler
+
+
+async def poll_once(session_factory: async_sessionmaker, client: ElectrumClient) -> int:
+    async with session_factory() as session:
+        pending = (
+            await session.scalars(select(PendingTransaction).where(PendingTransaction.status == "pending"))
+        ).all()
+        pending_ids = [p.id for p in pending]
+
+    confirmed = 0
+    for pending_id, txid, kind in [(p.id, p.current_txid, p.kind) for p in pending]:
+        tx = await client.get_transaction(txid, verbose=True)
+        if not tx or tx.get("confirmations", 0) < 1:
+            continue
+        async with session_factory() as session:
+            row = await session.get(PendingTransaction, pending_id)
+            if row is None or row.status != "pending":
+                continue
+            row.status = "confirmed"
+            handler = _handlers.get(kind)
+            if handler is not None:
+                await handler(session, row)
+            await session.commit()
+        confirmed += 1
+
+    return confirmed
+
+
+class ConfirmationPoller:
+    def __init__(self, session_factory: async_sessionmaker, get_client: Callable[[], ElectrumClient | None]):
+        self._session_factory = session_factory
+        self._get_client = get_client
+
+    async def run(self) -> None:
+        while True:
+            client = self._get_client()
+            if client is not None:
+                try:
+                    await poll_once(self._session_factory, client)
+                except asyncio.CancelledError:
+                    raise
+                except Exception:
+                    logger.exception("confirmation poll failed")
+            await asyncio.sleep(_POLL_INTERVAL_SECONDS)

app/tx/locks.py

@@ -0,0 +1,22 @@
+import asyncio
+from contextlib import asynccontextmanager
+
+
+class UserLocks:
+    """Per-user asyncio.Lock registry, shared by PLAY and WITHDRAW so a user can
+    never have a bet-build and a withdrawal-build in flight at once (both would
+    otherwise spend from the same UTXO set on the user's dedicated address).
+
+    Single-process-only by design (an in-memory dict of asyncio.Lock) — this is an
+    accepted MVP constraint; a multi-process deployment would need a DB or Redis
+    lock instead (e.g. a Postgres advisory lock).
+    """
+
+    def __init__(self) -> None:
+        self._locks: dict[int, asyncio.Lock] = {}
+
+    @asynccontextmanager
+    async def acquire(self, user_id: int):
+        lock = self._locks.setdefault(user_id, asyncio.Lock())
+        async with lock:
+            yield

app/wallet/balance.py

@@ -0,0 +1,16 @@
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models import User, UtxoEvent
+
+
+async def recompute_balance(session: AsyncSession, user_id: int) -> int:
+    """Source of truth: sum of this user's confirmed, unspent UTXOs. Updates and
+    returns the read-cache column (User.cached_balance_sats). Must be called
+    within the same transaction as whatever inserted/updated utxo_events rows."""
+    balance = await session.scalar(
+        select(func.sum(UtxoEvent.amount_sats)).where(UtxoEvent.user_id == user_id, UtxoEvent.spent_txid.is_(None))
+    )
+    user = await session.get(User, user_id)
+    user.cached_balance_sats = balance or 0
+    return user.cached_balance_sats

app/wallet/hd.py

@@ -0,0 +1,52 @@
+import os
+
+from embit import script
+from embit.bip32 import HDKey
+
+from app.config import settings
+from app.wallet.keystore import decrypt_xprv, encrypt_xprv
+from app.wallet.plm_network import ACCOUNT_PATH, PLM_MAINNET
+
+_account_key: HDKey | None = None
+
+
+def generate_master_key(overwrite: bool = False) -> None:
+    """One-time ops bootstrap: create a random master seed, encrypt it, write it to
+    disk. Not exposed via any API endpoint — run manually before first launch."""
+    if os.path.exists(settings.master_key_path) and not overwrite:
+        raise FileExistsError(f"{settings.master_key_path} already exists")
+    root = HDKey.from_seed(os.urandom(32), version=PLM_MAINNET["xprv"])
+    with open(settings.master_key_path, "wb") as f:
+        f.write(encrypt_xprv(root.to_base58(version=PLM_MAINNET["xprv"])))
+
+
+def _load_account_key() -> HDKey:
+    global _account_key
+    if _account_key is None:
+        with open(settings.master_key_path, "rb") as f:
+            token = f.read()
+        root = HDKey.from_base58(decrypt_xprv(token))
+        _account_key = root.derive(ACCOUNT_PATH)
+    return _account_key
+
+
+def derive_user_key(derivation_index: int) -> HDKey:
+    return _load_account_key().derive(f"0/{derivation_index}")
+
+
+def derive_user_address(derivation_index: int) -> str:
+    pub = derive_user_key(derivation_index).to_public()
+    return script.p2wpkh(pub).address(network=PLM_MAINNET)
+
+
+def derive_pool_key() -> HDKey:
+    """The "indirizzo padre" from the flowchart: all bets are sent here, and
+    payouts are signed with this key. Reserved on branch 1 of the account (branch 0
+    is user addresses), index 0 — not a spec requirement, an implementation choice
+    to keep it in the same encrypted master key rather than a separate secret."""
+    return _load_account_key().derive("1/0")
+
+
+def derive_pool_address() -> str:
+    pub = derive_pool_key().to_public()
+    return script.p2wpkh(pub).address(network=PLM_MAINNET)

app/wallet/keystore.py

@@ -0,0 +1,11 @@
+from cryptography.fernet import Fernet
+
+from app.config import settings
+
+
+def encrypt_xprv(xprv_base58: str) -> bytes:
+    return Fernet(settings.xprv_encryption_key).encrypt(xprv_base58.encode())
+
+
+def decrypt_xprv(token: bytes) -> str:
+    return Fernet(settings.xprv_encryption_key).decrypt(token).decode()

app/wallet/plm_network.py

@@ -0,0 +1,28 @@
+"""PLM mainnet params for embit, verified against PalladiumWallet/src/Core/Chain/ChainProfiles.cs.
+
+Threaded explicitly through every embit call via `network=PLM_MAINNET` rather than
+registered globally, since this is a long-lived async server (embit has no
+concept of "current network" beyond what you pass in).
+"""
+
+PLM_MAINNET = {
+    "name": "PLM Mainnet",
+    "wif": bytes([0x80]),
+    "p2pkh": bytes([55]),
+    "p2sh": bytes([5]),
+    "bech32": "plm",
+    "xprv": bytes.fromhex("0488ade4"),
+    "xpub": bytes.fromhex("0488b21e"),
+    "yprv": bytes.fromhex("049d7878"),
+    "ypub": bytes.fromhex("049d7cb2"),
+    "zprv": bytes.fromhex("04b2430c"),
+    "zpub": bytes.fromhex("04b24746"),
+    "Yprv": bytes.fromhex("0295b005"),
+    "Ypub": bytes.fromhex("0295b43f"),
+    "Zprv": bytes.fromhex("02aa7a99"),
+    "Zpub": bytes.fromhex("02aa7ed3"),
+    "bip32": 0,
+}
+
+BIP44_COIN_TYPE = 746
+ACCOUNT_PATH = f"m/84h/{BIP44_COIN_TYPE}h/0h"

app/wallet/psbt_builder.py

@@ -0,0 +1,181 @@
+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):
+    pass
+
+
+@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")
+    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")
+    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,
+    )

app/withdrawals/confirmation.py

@@ -0,0 +1,18 @@
+from datetime import datetime, timezone
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models import PendingTransaction, Withdrawal
+from app.tx.confirmation import register_handler
+
+
+async def _on_withdrawal_confirmed(session: AsyncSession, pending: PendingTransaction) -> None:
+    if pending.withdrawal_id is None:
+        return
+    withdrawal = await session.get(Withdrawal, pending.withdrawal_id)
+    if withdrawal is not None and withdrawal.status == "broadcast":
+        withdrawal.status = "confirmed"
+        withdrawal.confirmed_at = datetime.now(timezone.utc)
+
+
+register_handler("withdrawal", _on_withdrawal_confirmed)

app/withdrawals/service.py

@@ -0,0 +1,86 @@
+from embit import script
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.audit.log import write_audit_log
+from app.config import settings
+from app.db.models import PendingTransaction, User, UtxoEvent, Withdrawal
+from app.electrum.client import ElectrumClient
+from app.wallet.balance import recompute_balance
+from app.wallet.hd import derive_user_key
+from app.wallet.psbt_builder import InsufficientFundsError, Utxo, build_signed_transaction
+
+
+class WithdrawalError(Exception):
+    pass
+
+
+async def request_withdrawal(
+    session: AsyncSession, client: ElectrumClient, user: User, external_address: str, amount_sats: int
+) -> Withdrawal:
+    if amount_sats < settings.min_amount_sats:
+        raise WithdrawalError(f"amount below the minimum of {settings.min_amount_sats} 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) < amount_sats:
+        raise WithdrawalError("insufficient balance")
+
+    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=external_address,
+            amount_sats=amount_sats,
+            change_address=user.address,
+            fee_rate_sat_vb=settings.fee_rate_sat_vb,
+        )
+    except InsufficientFundsError as exc:
+        raise WithdrawalError(str(exc)) from exc
+
+    await client.broadcast(built.raw_hex)
+
+    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)
+
+    withdrawal = Withdrawal(
+        user_id=user.id,
+        external_address=external_address,
+        amount_requested_sats=amount_sats,
+        amount_sent_sats=built.recipient_sats,
+        txid=built.txid,
+        status="broadcast",
+    )
+    session.add(withdrawal)
+    await session.flush()
+    session.add(
+        PendingTransaction(
+            kind="withdrawal",
+            withdrawal_id=withdrawal.id,
+            user_id=user.id,
+            current_txid=built.txid,
+            fee_rate_sat_vb=settings.fee_rate_sat_vb,
+            raw_tx_hex=built.raw_hex,
+            status="pending",
+        )
+    )
+    await write_audit_log(
+        session,
+        "withdrawal_sent",
+        {"txid": built.txid, "amount_sent_sats": built.recipient_sats, "external_address": external_address},
+        user_id=user.id,
+    )
+
+    await session.commit()
+    await session.refresh(withdrawal)
+    return withdrawal