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 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 round_id: int | None = None status: str | None = None opened_at: str | None = None closes_at: str | None = None participant_count: int = 0 bet_amount_sats: int jackpot_sats: int = 0 draw_animation_seconds: int winner_user_id: int | None = None winner_amount_sats: int | None = None draw_block_height: int | None = None draw_block_hash: str | None = None chain_tip_height: int | None = None lottery_paused: bool = False user_played: bool = False @router.get("/current", response_model=CurrentRoundResponse) async def current_round( request: Request, session: AsyncSession = Depends(get_session), user: User | None = Depends(get_optional_user), ) -> CurrentRoundResponse: config = await get_round_config(session) round_ = await get_active_round(session) listener = request.app.state.electrum_listener chain_tip_height = listener.tip_height or None if round_ is None: await session.commit() return CurrentRoundResponse( server_time=datetime.now(timezone.utc).isoformat(), bet_amount_sats=config.bet_amount_sats, draw_animation_seconds=config.draw_animation_seconds, chain_tip_height=chain_tip_height, lottery_paused=config.paused, ) participant_count = await session.scalar( select(func.count()).select_from(RoundParticipant).where(RoundParticipant.round_id == round_.id) ) or 0 opened_at = round_.opened_at.replace(tzinfo=timezone.utc) closes_at = opened_at + timedelta(seconds=config.round_duration_seconds) # Lets the frontend show the personalized win/lose reveal only to players in # this round — everyone else (not logged in, or logged in but didn't bet) # just sees the generic phase progress instead of a "non hai vinto" that # wouldn't mean anything to them. user_played = False if user is not None: user_played = ( await session.scalar( select(RoundParticipant).where( RoundParticipant.round_id == round_.id, RoundParticipant.user_id == user.id ) ) ) is not None await session.commit() # Shown to players as "jackpot": the winner's 70% share of the pool (same # split rounds/scheduler.py applies at payout time), not the full pool — # what's displayed should match what the winner actually receives. pool_amount_sats = participant_count * config.bet_amount_sats jackpot_sats = pool_amount_sats * 70 // 100 return CurrentRoundResponse( server_time=datetime.now(timezone.utc).isoformat(), round_id=round_.id, status=round_.status, opened_at=opened_at.isoformat(), closes_at=closes_at.isoformat(), participant_count=participant_count, bet_amount_sats=config.bet_amount_sats, jackpot_sats=jackpot_sats, draw_animation_seconds=config.draw_animation_seconds, winner_user_id=round_.winner_user_id, winner_amount_sats=round_.winner_amount_sats, draw_block_height=round_.draw_block_height, draw_block_hash=round_.draw_block_hash, chain_tip_height=chain_tip_height, lottery_paused=config.paused, user_played=user_played, )