Add server-push (SSE) notifications for round/bet/balance state changes

Frontend dashboards previously found out about state changes only on their
next poll tick (up to 15s, or 3s during a draw) — this adds a push channel
so updates land as soon as they happen instead.

- app/rounds/events.py: a small in-process pub/sub (RoundEventBroadcaster).
  The message carries no payload — it's just a "something changed, go
  refetch" signal, so it needs no auth and no knowledge of who's allowed to
  see what; personalization stays entirely in the existing REST endpoints.
- GET /rounds/stream: an SSE endpoint exposing that channel, with keep-alive
  comments so it survives idle periods behind a reverse proxy, and a
  defensive MAX_SUBSCRIBERS cap (well above the ~100 concurrent users
  expected) — past it, the endpoint returns 503 instead of opening a stream,
  and callers just keep working off polling.
- publish() calls added at every point that actually changes what a
  dashboard would want to know: new round opened, round status transitions
  (closing/drawing/paying_out/closed), a bet or withdrawal broadcast, any
  pending tx confirming (bet/withdrawal/payout), a deposit credited, and a
  new block tip arriving (the exact moment the "drawing" phase is waiting on).
- Caddyfile: excludes /rounds/stream from gzip encoding, since compression
  would buffer output and defeat the point of a live stream.

Single-process only by design for now (no cross-worker fan-out) and the
notification is a generic broadcast rather than a per-user channel — both
are deliberate scope decisions for the current ~100-user, single-container
deployment, not oversights. Polling is left fully in place as a fallback;
this is purely additive.

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

Caddyfile

@@ -6,6 +6,12 @@
 # instead, via its internal CA. Browsers will still warn on first visit
 # unless that CA is explicitly trusted — expected for local/dev use.
 {$SITE_ADDRESS:localhost} {
-	encode gzip
+	# gzip buffers output, which would delay delivery on the SSE stream
+	# (/rounds/stream, app/api/routes/rounds.py) — it needs each event flushed
+	# to the client immediately, not batched. Everything else still compresses.
+	@not_sse {
+		not path /rounds/stream
+	}
+	encode @not_sse gzip
 	reverse_proxy app:8123
 }

app/api/routes/rounds.py

@@ -1,6 +1,9 @@
+import asyncio
+import json
 from datetime import datetime, timedelta, timezone
 
 from fastapi import APIRouter, Depends, Request
+from fastapi.responses import JSONResponse, Response, StreamingResponse
 from pydantic import BaseModel
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -9,10 +12,69 @@ from app.auth.dependencies import get_optional_user
 from app.db.models import RoundParticipant, User
 from app.db.session import get_session
 from app.rounds.config import get_round_config
+from app.rounds.events import RoundEventCapacityError, broadcaster
 from app.rounds.service import get_active_round
 
 router = APIRouter(prefix="/rounds", tags=["rounds"])
 
+# How often request.is_disconnected() gets (re-)checked while idle — bounds
+# how long a subscriber slot lingers after a client goes away without a clean
+# TCP close (e.g. the network just vanishes). Kept short since the check
+# itself is cheap; it does NOT control how often anything is sent on the wire.
+_SSE_DISCONNECT_CHECK_SECONDS = 5
+
+# How often to send an SSE keep-alive comment on an otherwise-idle connection —
+# well under any reasonable reverse-proxy/load-balancer idle-connection timeout
+# (Caddy's default is 5 minutes) so the stream isn't silently dropped. Expressed
+# as a multiple of the disconnect-check interval above.
+_SSE_KEEPALIVE_TICKS = 4  # 4 * 5s = 20s between keep-alive comments
+
+
+@router.get("/stream")
+async def round_stream(request: Request) -> Response:
+    """Server-Sent Events channel: pushes a content-free "update" notification
+    the instant round/bet/balance state changes anywhere (see app/rounds/events.py
+    for the publish() call sites), instead of clients only finding out on their
+    next poll. No payload and no auth: it's a public "go refetch" signal, and
+    the actual data still comes from the normal per-user REST endpoints, which
+    is where authorization and personalization (e.g. user_played) already live.
+
+    Frontend polling (app/static/index.html) is left in place as a fallback —
+    this is purely additive, so a dropped/blocked SSE connection degrades to
+    the pre-existing polling behavior rather than losing updates outright.
+    That's also what happens past MAX_SUBSCRIBERS (app/rounds/events.py): this
+    returns 503 rather than opening a stream, and the browser's EventSource
+    just retries later while the frontend keeps working off polling meanwhile.
+    """
+    try:
+        queue = broadcaster.subscribe()
+    except RoundEventCapacityError:
+        return JSONResponse(status_code=503, content={"detail": "too many concurrent update streams"})
+
+    async def event_generator():
+        ticks_since_keepalive = 0
+        try:
+            while True:
+                if await request.is_disconnected():
+                    break
+                try:
+                    await asyncio.wait_for(queue.get(), timeout=_SSE_DISCONNECT_CHECK_SECONDS)
+                    yield "event: update\ndata: {}\n\n".format(json.dumps({}))
+                    ticks_since_keepalive = 0
+                except asyncio.TimeoutError:
+                    ticks_since_keepalive += 1
+                    if ticks_since_keepalive >= _SSE_KEEPALIVE_TICKS:
+                        yield ": keep-alive\n\n"
+                        ticks_since_keepalive = 0
+        finally:
+            broadcaster.unsubscribe(queue)
+
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-store", "X-Accel-Buffering": "no"},
+    )
+
 
 class CurrentRoundResponse(BaseModel):
     server_time: str

app/bets/service.py

@@ -8,6 +8,7 @@ from app.audit.log import write_audit_log
 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.events import broadcaster
 from app.rounds.service import open_new_round_if_needed, round_accepts_bets
 from app.wallet.balance import recompute_balance
 from app.wallet.hd import derive_pool_address, derive_user_key
@@ -91,6 +92,7 @@ async def place_bet(session: AsyncSession, client: ElectrumClient, user: User) -
 
     await session.commit()
     await session.refresh(participant)
+    broadcaster.publish()  # participant_count/jackpot changed — nudge every dashboard to refetch
     return participant
 
 

app/deposits/service.py

@@ -3,6 +3,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.audit.log import write_audit_log
 from app.db.models import UtxoEvent
+from app.rounds.events import broadcaster
 from app.wallet.balance import recompute_balance
 
 
@@ -50,5 +51,6 @@ async def credit_confirmed_utxos(session: AsyncSession, user_id: int, entries: l
         await session.flush()
         await recompute_balance(session, user_id)
         await session.commit()
+        broadcaster.publish()  # nudges this user's dashboard to refetch its balance instantly
 
     return newly_credited

app/electrum/listener.py

@@ -9,6 +9,7 @@ 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
+from app.rounds.events import broadcaster
 
 logger = logging.getLogger(__name__)
 
@@ -94,6 +95,10 @@ class ElectrumListener:
             for header in params:
                 self.tip_height = header["height"]
                 self.tip_header_hex = header.get("hex")
+            # A new block is exactly what the "drawing" phase is waiting on
+            # (rounds/scheduler.py:_wait_for_next_block) — nudge dashboards to
+            # refetch instead of waiting for their next poll.
+            broadcaster.publish()
 
     async def _consume_scripthash(self, queue: asyncio.Queue) -> None:
         while True:

app/rounds/events.py

@@ -0,0 +1,50 @@
+import asyncio
+
+# Defensive cap on concurrent SSE subscribers. Expected load is on the order of
+# ~100 concurrent users; this is set well above that so it never engages under
+# normal use — it exists purely so a runaway/DoS-y number of open connections
+# degrades (new connections fall back to polling, see round_stream()) instead
+# of growing the in-memory subscriber set without bound. Revisit this number if
+# expected concurrency grows well past it.
+MAX_SUBSCRIBERS = 500
+
+
+class RoundEventCapacityError(Exception):
+    """Raised by subscribe() when MAX_SUBSCRIBERS is already reached."""
+
+
+class RoundEventBroadcaster:
+    """In-process pub/sub so SSE clients (GET /rounds/stream) get pushed a
+    notification the instant round/bet/balance state changes, instead of only
+    finding out on their next poll. The message carries no payload — it's just
+    a "something changed, go refetch" signal; the client re-hits the existing
+    per-user REST endpoints (/rounds/current, /users/me, ...) for the actual
+    data, so this never needs to know what changed or who's allowed to see it.
+
+    Single-process only (no cross-worker fan-out) — fine for this deployment
+    (one uvicorn process, see docker-compose.yml). A multi-worker deployment
+    would need a shared channel (e.g. Redis pub/sub) instead.
+    """
+
+    def __init__(self, max_subscribers: int = MAX_SUBSCRIBERS):
+        self._subscribers: set[asyncio.Queue] = set()
+        self.max_subscribers = max_subscribers
+
+    def subscribe(self) -> asyncio.Queue:
+        if len(self._subscribers) >= self.max_subscribers:
+            raise RoundEventCapacityError(f"already at the {self.max_subscribers}-subscriber cap")
+        queue: asyncio.Queue = asyncio.Queue(maxsize=1)
+        self._subscribers.add(queue)
+        return queue
+
+    def unsubscribe(self, queue: asyncio.Queue) -> None:
+        self._subscribers.discard(queue)
+
+    def publish(self) -> None:
+        for queue in self._subscribers:
+            if queue.full():
+                continue  # a not-yet-delivered notification already covers this one
+            queue.put_nowait(None)
+
+
+broadcaster = RoundEventBroadcaster()

app/rounds/scheduler.py

@@ -12,6 +12,7 @@ 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.events import broadcaster
 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
@@ -71,6 +72,7 @@ class RoundScheduler:
                 round_.status = "closing"
                 round_.closed_at = datetime.now(timezone.utc)
                 await session.commit()
+                broadcaster.publish()
 
         async with self._session_factory() as session:
             pending_count = await session.scalar(
@@ -99,6 +101,7 @@ class RoundScheduler:
                 round_.status = "closed"
                 await write_audit_log(session, "round_closed", {"participants": 0}, round_id=round_id)
                 await session.commit()
+                broadcaster.publish()
                 logger.info("round %s closed with no participants", round_id)
                 return
 
@@ -112,6 +115,7 @@ class RoundScheduler:
 
             round_.status = "drawing"
             await session.commit()
+        broadcaster.publish()
 
         tip_at_close = self._listener.tip_height
         block_height, block_hash = await self._wait_for_next_block(tip_at_close)
@@ -139,6 +143,7 @@ class RoundScheduler:
                 round_id=round_id,
             )
             await session.commit()
+        broadcaster.publish()
 
         logger.info("round %s: winner=%s pool=%s", round_id, winner_address, pool_amount)
         await self._trigger_payout(round_id)

app/rounds/service.py

@@ -5,6 +5,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.db.models import Round
 from app.rounds.config import get_round_config
+from app.rounds.events import broadcaster
 
 _ACTIVE_STATUSES = ("open", "closing", "drawing", "paying_out")
 
@@ -57,4 +58,9 @@ async def open_new_round_if_needed(session: AsyncSession) -> Round | None:
     round_ = Round(status="open")
     session.add(round_)
     await session.flush()
+    # Published pre-commit (the caller commits right after) — acceptable: this
+    # only tells subscribers "go refetch", and by the time an SSE client's
+    # refetch request actually lands, this in-process commit (microseconds
+    # away) has essentially always already happened.
+    broadcaster.publish()
     return round_

app/tx/confirmation.py

@@ -7,6 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from app.db.models import PendingTransaction
 from app.electrum.client import ElectrumClient
+from app.rounds.events import broadcaster
 
 logger = logging.getLogger(__name__)
 
@@ -44,6 +45,7 @@ async def poll_once(session_factory: async_sessionmaker, client: ElectrumClient)
             if handler is not None:
                 await handler(session, row)
             await session.commit()
+        broadcaster.publish()  # a bet/withdrawal/payout just confirmed — balance and/or round state changed
         confirmed += 1
 
     return confirmed

app/withdrawals/service.py

@@ -6,6 +6,7 @@ from app.audit.log import write_audit_log
 from app.db.models import PendingTransaction, User, UtxoEvent, Withdrawal
 from app.electrum.client import ElectrumClient
 from app.rounds.config import get_round_config
+from app.rounds.events import broadcaster
 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
@@ -84,4 +85,5 @@ async def request_withdrawal(
 
     await session.commit()
     await session.refresh(withdrawal)
+    broadcaster.publish()  # balance just went "pending" — nudge the dashboard to refetch
     return withdrawal

tests/unit/test_round_events.py

@@ -0,0 +1,62 @@
+import asyncio
+
+import pytest
+
+from app.rounds.events import RoundEventBroadcaster, RoundEventCapacityError
+
+
+async def test_publish_wakes_up_subscriber():
+    broadcaster = RoundEventBroadcaster()
+    queue = broadcaster.subscribe()
+
+    broadcaster.publish()
+
+    await asyncio.wait_for(queue.get(), timeout=1)
+
+
+async def test_publish_with_no_subscribers_does_not_raise():
+    broadcaster = RoundEventBroadcaster()
+    broadcaster.publish()  # no subscribers yet — must be a no-op, not an error
+
+
+async def test_publish_coalesces_when_subscriber_has_not_drained():
+    """The queue is maxsize=1: a second publish() before the subscriber reads
+    the first notification must not block or raise — it's fine to drop it,
+    since the subscriber will refetch full state anyway on the first one."""
+    broadcaster = RoundEventBroadcaster()
+    queue = broadcaster.subscribe()
+
+    broadcaster.publish()
+    broadcaster.publish()  # would raise QueueFull if not guarded
+
+    assert queue.qsize() == 1
+
+
+async def test_unsubscribe_stops_delivery():
+    broadcaster = RoundEventBroadcaster()
+    queue = broadcaster.subscribe()
+    broadcaster.unsubscribe(queue)
+
+    broadcaster.publish()
+
+    assert queue.empty()
+
+
+async def test_subscribe_rejects_past_the_cap():
+    broadcaster = RoundEventBroadcaster(max_subscribers=3)
+    for _ in range(3):
+        broadcaster.subscribe()
+
+    with pytest.raises(RoundEventCapacityError):
+        broadcaster.subscribe()
+
+
+async def test_unsubscribe_frees_a_capacity_slot():
+    broadcaster = RoundEventBroadcaster(max_subscribers=1)
+    queue = broadcaster.subscribe()
+
+    with pytest.raises(RoundEventCapacityError):
+        broadcaster.subscribe()
+
+    broadcaster.unsubscribe(queue)
+    broadcaster.subscribe()  # no longer at capacity