Add round lifecycle, draw algorithm and scheduler

Periodic scheduler (configurable round duration) that closes a round
only once all broadcast bets confirm, waits for the next block after
closing, draws a winner via block-hash-seeded modulo over participants
ordered by broadcast time, triggers the 70/30 payout, and only opens the
next round once that payout confirms. Draw logic is isolated in
draw.py as a deliberately simple, replaceable component.

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

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_

tests/unit/test_draw.py

@@ -0,0 +1,32 @@
+import pytest
+
+from app.rounds.draw import draw_winner, header_hex_to_block_hash
+
+
+def test_header_hex_to_block_hash_matches_known_mainnet_block():
+    # Real PLM mainnet block 477486: header from blockchain.block.header, hash
+    # cross-checked against the blockhash reported by blockchain.transaction.get
+    # for a tx confirmed in that block.
+    header_hex = (
+        "0020ed30fbc39ee45200d10214f5107c5f36e7bf753032fd1d3279810c170000000000"
+        "009a4ea723f732be4c538f3a2490837dd6560103d73ece606aa7d578a66700447b28875"
+        "e6a47a61b1ad8012582"
+    )
+    known_block_hash = "00000000000008788b55ade13b74d54ceffda9e54315b802411be1ca65064e86"
+    assert header_hex_to_block_hash(header_hex) == known_block_hash
+
+
+def test_draw_winner_is_deterministic_and_within_range():
+    participants = ["addrA", "addrB", "addrC"]
+    block_hash = "00" * 31 + "05"  # seed = 5, index = 5 % 3 = 2
+    assert draw_winner(participants, block_hash) == "addrC"
+
+
+def test_draw_winner_single_participant_always_wins():
+    block_hash = "ff" * 32
+    assert draw_winner(["only"], block_hash) == "only"
+
+
+def test_draw_winner_raises_on_empty_participants():
+    with pytest.raises(ValueError):
+        draw_winner([], "00" * 32)

tests/unit/test_rounds_service.py

@@ -0,0 +1,58 @@
+import pytest
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from app.db.base import Base
+from app.db.models import Round
+from app.rounds.service import get_active_round, open_new_round_if_needed
+
+
+@pytest.fixture
+async def session_factory():
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    yield async_sessionmaker(engine, expire_on_commit=False)
+    await engine.dispose()
+
+
+async def test_opens_a_round_when_none_exists(session_factory):
+    async with session_factory() as session:
+        round_ = await open_new_round_if_needed(session)
+        await session.commit()
+        assert round_.status == "open"
+
+
+async def test_reuses_existing_open_round(session_factory):
+    async with session_factory() as session:
+        first = await open_new_round_if_needed(session)
+        await session.commit()
+        first_id = first.id
+
+    async with session_factory() as session:
+        second = await open_new_round_if_needed(session)
+        assert second.id == first_id
+
+
+@pytest.mark.parametrize("status", ["closing", "drawing", "paying_out"])
+async def test_does_not_open_new_round_while_previous_is_in_progress(session_factory, status):
+    async with session_factory() as session:
+        session.add(Round(status=status))
+        await session.commit()
+
+    async with session_factory() as session:
+        active = await get_active_round(session)
+        assert active is not None
+        assert active.status == status
+        # open_new_round_if_needed must return the in-progress round, not open a new one
+        returned = await open_new_round_if_needed(session)
+        assert returned.status == status
+
+
+async def test_opens_new_round_after_previous_is_closed(session_factory):
+    async with session_factory() as session:
+        session.add(Round(status="closed"))
+        await session.commit()
+
+    async with session_factory() as session:
+        round_ = await open_new_round_if_needed(session)
+        assert round_.status == "open"

tests/unit/test_scheduler.py

@@ -0,0 +1,55 @@
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from app.config import settings
+from app.db.base import Base
+from app.db.models import Round
+from app.rounds.scheduler import RoundScheduler
+
+
+class FakeListener:
+    client = object()  # truthy sentinel; _tick only checks "is not None"
+    tip_height = 100
+    tip_header_hex = "00"
+
+
+@pytest.fixture
+async def session_factory():
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    yield async_sessionmaker(engine, expire_on_commit=False)
+    await engine.dispose()
+
+
+async def test_tick_survives_sqlite_naive_datetime_roundtrip(session_factory, monkeypatch):
+    """Regression test: SQLite drops tzinfo on round-trip, so opened_at comes back
+    naive even though it was written as an aware UTC datetime. A prior bug compared
+    it directly against datetime.now(timezone.utc) and crashed with
+    "can't compare offset-naive and offset-aware datetimes" on every tick once a
+    round existed — this must not happen."""
+    monkeypatch.setattr(settings, "round_duration_seconds", 3600)  # not due yet
+    async with session_factory() as session:
+        session.add(Round(status="open", opened_at=datetime.now(timezone.utc)))
+        await session.commit()
+
+    scheduler = RoundScheduler(session_factory, FakeListener())
+    await scheduler._tick()  # must not raise
+
+
+async def test_tick_closes_round_with_no_participants_once_due(session_factory, monkeypatch):
+    monkeypatch.setattr(settings, "round_duration_seconds", 1)
+    past = datetime.now(timezone.utc) - timedelta(seconds=10)
+    async with session_factory() as session:
+        session.add(Round(status="open", opened_at=past))
+        await session.commit()
+
+    scheduler = RoundScheduler(session_factory, FakeListener())
+    await scheduler._tick()
+
+    async with session_factory() as session:
+        round_ = (await session.scalars(select(Round))).one()
+        assert round_.status == "closed"