Real-time: WebSocket

# realtime/ws.py
from fastapi import APIRouter, WebSocket, WebSocketDisconnect

router = APIRouter()


@router.websocket("/ws/echo")
async def echo(ws: WebSocket):
    await ws.accept()
    try:
        while True:
            text = await ws.receive_text()
            await ws.send_text(f"তুমি বললে: {text}")
    except WebSocketDisconnect:
        pass                                    # client চলে গেছে

brew install websocat
websocat ws://127.0.0.1:8000/ws/echo

# realtime/manager.py
from collections import defaultdict
from fastapi import WebSocket


class ConnectionManager:
    """এই server এ কে কোন order দেখছে তার হিসাব।"""

    def __init__(self) -> None:
        self.rooms: dict[str, set[WebSocket]] = defaultdict(set)

    async def join(self, room: str, ws: WebSocket) -> None:
        await ws.accept()
        self.rooms[room].add(ws)

    def leave(self, room: str, ws: WebSocket) -> None:
        self.rooms[room].discard(ws)
        if not self.rooms[room]:
            del self.rooms[room]

    async def broadcast_local(self, room: str, message: dict) -> None:
        dead = []
        for ws in self.rooms.get(room, ()):
            try:
                await ws.send_json(message)
            except Exception:
                dead.append(ws)                 # বন্ধ connection সরাও
        for ws in dead:
            self.leave(room, ws)


manager = ConnectionManager()

@router.websocket("/ws/orders/{order_id}")
async def order_updates(ws: WebSocket, order_id: int, token: str):
    user = await authenticate_ws(token)                       # login যাচাই
    if user is None or not await can_view_order(user, order_id):
        await ws.close(code=4403)                             # অনুমতি নেই
        return
    room = f"order:{order_id}"
    await manager.join(room, ws)
    try:
        while True:
            await ws.receive_text()                           # client এর ping, বা কিছু না
    except WebSocketDisconnect:
        manager.leave(room, ws)

# realtime/pubsub.py
import asyncio
import json
from app.core.redis import redis


async def publish(room: str, message: dict) -> None:
    await redis.publish(room, json.dumps(message, ensure_ascii=False))


async def listen_forever() -> None:
    """প্রতিটা server চালু হলে এটা background এ চলে।"""
    pubsub = redis.pubsub()
    await pubsub.psubscribe("order:*")
    async for msg in pubsub.listen():
        if msg["type"] == "pmessage":
            room = msg["channel"]
            await manager.broadcast_local(room, json.loads(msg["data"]))

# main.py: app চালু হলে listener শুরু
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app):
    task = asyncio.create_task(listen_forever())
    yield
    task.cancel()

app = FastAPI(lifespan=lifespan)

# order status বদলানোর service এ
async def update_status(self, order_id: int, status: OrderStatus) -> Order:
    order = ...                                             # database এ বদলাও, commit
    await publish(f"order:{order_id}", {"type": "status", "status": status, "at": now_iso()})
    return order

import asyncio
import json
from fastapi.responses import StreamingResponse


@router.get("/orders/{order_id}/events")
async def order_events(order_id: int, user: CurrentUser):
    await ensure_can_view(user, order_id)
    pubsub = redis.pubsub()
    await pubsub.subscribe(f"order:{order_id}")

    async def stream():
        try:
            yield f"data: {json.dumps(await current_status(order_id))}\n\n"     # প্রথমে বর্তমান অবস্থা
            while True:
                msg = await pubsub.get_message(ignore_subscribe_messages=True, timeout=15)
                if msg:
                    yield f"data: {msg['data']}\n\n"
                else:
                    yield ": keep-alive\n\n"                                    # heartbeat
        finally:
            await pubsub.unsubscribe()

    return StreamingResponse(stream(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})

// web_socket_channel package
final channel = WebSocketChannel.connect(
  Uri.parse('wss://api.example.com/ws/orders/42?token=$ticket'),
);
channel.stream.listen(
  (data) => onStatus(jsonDecode(data)),
  onDone: scheduleReconnect,           // বন্ধ হলে backoff সহ আবার
  onError: (_) => scheduleReconnect(),
);