Architecting High-Throughput FAST AI Platforms, Streaming Data, & DevEx Tooling

async def event_generator(user_id: str, last_event_id: Optional[str] = None):
    queue = asyncio.Queue(maxsize=100) # Bounded backpressure queue
    redis_client = await get_redis_pool()
    
    # Replay missed events from Redis stream if reconnecting
    if last_event_id:
        missed = await redis_client.xrange(f"stream:{user_id}", min=last_event_id)
        for msg_id, payload in missed[1:]:
            yield f"id: {msg_id}\ndata: {payload['data']}\n\n"

    try:
        while True:
            try:
                # Wait for token with 15-second heartbeat timeout
                data = await asyncio.wait_for(queue.get(), timeout=15.0)
                msg_id = await redis_client.xadd(f"stream:{user_id}", {"data": data})
                yield f"id: {msg_id}\ndata: {data}\n\n"
            except asyncio.TimeoutError:
                # Keep-alive comment ping prevents AWS ALB / Cloudflare socket termination
                yield ": ping\n\n"
    except asyncio.CancelledError:
        logger.info(f"SSE stream closed gracefully for user {user_id}")