"""TV tại bàn (bàn giao 15, 05/08/2026) — WS ``/display/{id}`` + registry. Thiết kế: FullVision §5.4 (TV client) + §8 (``WS /display/{display_id}``, pair mã 6 số). Demo "đánh xong liếc TV" với trigger TAY (bấm Gợi ý trên điện thoại) — KHÔNG camera, KHÔNG edge agent, phần đó của bàn giao sau. Registry IN-MEMORY (dict), KHÔNG bảng DB, KHÔNG migration — noi gương drills.json (bàn giao 11): Space không Postgres vẫn demo được; bảng ``displays`` trong design (FullVision §7) chờ có auth/dashboard. Hệ quả đã chốt trước trong BRIEF: server restart là mất pair — TV thấy WS rớt, tự reconnect và hiện lại mã để ghép lại. Khoá ghép là ``session_id`` DẠNG CHUỖI MỜ, cố ý không đụng DB ở đây: - có phiên QR (bàn giao 12) → điện thoại gửi đúng session_id đó ("bàn của phiên" như BRIEF Bước 2.2); - Space degraded không DB → điện thoại không thể có phiên QR, FE sinh id cục bộ chỉ để ghép (bối cảnh #4: TV demo không được phụ thuộc DB/Redis). Server chỉ so khớp chuỗi giữa pair và ``/api/recommend`` — không validate, không FK, cùng tinh thần ``scan_id``. Mọi thao tác hub chạy trên MỘT event loop (1 uvicorn worker, không thread) — không cần lock quanh dict. Cap + TTL (bàn giao 20, chuẩn bị /tv lên Space public): registry không còn lớn vô hạn. Purge LAZY trong ``register()`` — giữ đúng kiến trúc một event loop, KHÔNG task nền, KHÔNG lock; entry ``ws=None`` rớt quá ``DISPLAY_TTL_S`` bị xoá (TV rớt quá 1h mất mã, reconnect được cấp mã mới — hành vi chấp nhận). Đầy ``MAX_DISPLAYS`` → xoá entry ``ws=None`` cũ nhất; vẫn đầy (toàn kết nối sống) → endpoint accept rồi đóng WS ngay, không cấp entry — TV client sẵn auto-reconnect. """ from __future__ import annotations import asyncio import secrets import time from dataclasses import dataclass, field from datetime import datetime, timezone from fastapi import APIRouter, HTTPException, WebSocket, WebSocketDisconnect from .schemas import DisplayPairIn, DisplayPairOut router = APIRouter() DISPLAY_TTL_S = 60 * 60.0 # entry ws=None rớt quá 60 phút → purge (BRIEF 3.2) MAX_DISPLAYS = 50 # trần registry (BRIEF 3.3) def _now() -> float: """Đồng hồ đơn điệu cho ``last_seen_at`` — tách hàm để test monkeypatch (không sleep), và monotonic để NTP chỉnh giờ không purge oan.""" return time.monotonic() @dataclass class Display: """Một TV đã từng mở WS. ``ws`` None = đang rớt (entry giữ nguyên: mã pair + pairing sống qua reconnect — nhưng từ bàn giao 20 chỉ tới ``DISPLAY_TTL_S`` kể từ lần thấy cuối; quá hạn là purge lazy).""" code: str # mã pair 6 số, cấp MỘT lần cho mỗi display ws: WebSocket | None = None session_id: str | None = None # phiên đã ghép | None = chưa ghép last_seen_at: float = field(default_factory=_now) # mốc _now() thấy cuối class DisplayHub: """Registry display_id → Display, thuần in-memory.""" def __init__(self) -> None: self._displays: dict[str, Display] = {} def reset(self) -> None: """Về trạng thái rỗng — cho test, như jobqueue.teardown().""" self._displays = {} def _new_code(self) -> str: """Mã 6 số ngẫu nhiên (secrets — không đoán tuần tự được), tránh trùng với mã đang cấp; 10^6 mã cho vài display, vòng while an toàn.""" while True: code = f"{secrets.randbelow(1_000_000):06d}" if all(d.code != code for d in self._displays.values()): return code def _purge_expired(self) -> None: """Xoá entry ``ws=None`` rớt quá ``DISPLAY_TTL_S`` — TV mất mã, phải ghép lại với mã mới khi quay lại (hành vi chấp nhận, docstring module). Entry đang KẾT NỐI không bao giờ bị đụng, ghép hay chưa.""" now = _now() for did, d in list(self._displays.items()): if d.ws is None and now - d.last_seen_at > DISPLAY_TTL_S: del self._displays[did] def _evict_oldest_detached(self) -> bool: """Xoá entry ``ws=None`` có ``last_seen_at`` cũ nhất (nhường chỗ khi đầy cap). False = không còn gì để xoá (toàn kết nối sống).""" cands = [(d.last_seen_at, did) for did, d in self._displays.items() if d.ws is None] if not cands: return False del self._displays[min(cands)[1]] return True def register(self, display_id: str, ws: WebSocket) -> Display | None: """TV mở/reconnect WS. display_id đã biết → GIỮ NGUYÊN mã + pairing (TV reload không đổi mã — client giữ display_id trong sessionStorage); chưa biết → cấp entry mới với mã mới. Purge lazy chạy ở ĐÂY (bàn giao 20) — điểm vào duy nhất làm registry lớn lên, nên cũng là chỗ duy nhất cần dọn; không task nền, không lock. Trả None khi đầy ``MAX_DISPLAYS`` toàn kết nối sống và display_id là TV mới — endpoint accept rồi đóng WS, không cấp entry.""" self._purge_expired() d = self._displays.get(display_id) if d is None: while len(self._displays) >= MAX_DISPLAYS: if not self._evict_oldest_detached(): return None d = Display(code=self._new_code()) self._displays[display_id] = d d.ws = ws d.last_seen_at = _now() return d def detach(self, display_id: str, ws: WebSocket) -> None: """TV rớt. Chỉ gỡ nếu ``ws`` đúng là connection đang giữ — một display_id mở connection mới đè connection cũ (register), lúc đó finally của endpoint cũ không được phép gỡ nhầm cái mới. ``last_seen_at`` cập nhật tại đây để TTL đếm TỪ LÚC RỚT: "TV rớt quá 1h mất mã" — TV nối liền 2h rồi mới rớt không được tính là đã cũ 2h.""" d = self._displays.get(display_id) if d is not None and d.ws is ws: d.ws = None d.last_seen_at = _now() def pair(self, code: str, session_id: str) -> Display | None: """Ghép display có mã ``code`` với phiên. None = không display nào mang mã đó (mã gõ sai / TV chưa từng mở / server đã restart). Ghép ĐƯỢC cả khi TV đang rớt (ws None): entry còn là mã còn hiệu lực, TV reconnect sẽ thấy trạng thái paired trong hello. Pair lần nữa với phiên khác → phiên mới đè phiên cũ (một TV một bàn).""" for d in self._displays.values(): if d.code == code: d.session_id = session_id d.last_seen_at = _now() # có người đứng trước TV gõ mã return d return None # ---------------------------------------------- push sau recommend (B3) PUSH_TIMEOUT_S = 1.0 # trần MỖI TV — send bình thường là micro giây async def push_recommend(self, session_id: str, balls: dict, resp: dict) -> None: """Đẩy kết quả recommend xuống MỌI TV đã ghép với ``session_id``. BEST-EFFORT đúng nghĩa đen (BRIEF Bước 3.1): không TV / TV rớt / TV kẹt → bỏ qua lặng lẽ, KHÔNG BAO GIỜ ném — response ``/api/recommend`` không được đổi một byte hay chậm đi vì chuyện chiếu hình. ``wait_for`` chặn trần thời gian một socket nghẽn TCP có thể ghim response; quá trần coi như TV chết, buông socket chờ reconnect. ``balls`` là thế bàn ĐÃ GỬI cho recommend (JSON-thuần) — TV vẽ đúng thế của cú đang chiếu, không đoán từ ``balls_final``. """ msg = {"type": "recommend", "ts": datetime.now(timezone.utc).isoformat(timespec="seconds"), "balls": balls, "result": resp} for d in list(self._displays.values()): if d.session_id != session_id or d.ws is None: continue try: await asyncio.wait_for(d.ws.send_json(msg), self.PUSH_TIMEOUT_S) d.last_seen_at = _now() # push thành công = TV còn sống except Exception: # noqa: BLE001 — TV chết kiểu gì cũng bỏ qua d.ws = None hub = DisplayHub() @router.post("/api/displays/pair", response_model=DisplayPairOut) async def api_displays_pair(req: DisplayPairIn): """Điện thoại nhập mã 6 số trên TV → ghép display với phiên của bàn. Lệch design CÓ CHỦ ĐÍCH (BRIEF Bước 2.3): design cho owner pair từ dashboard (route /t/{tenant}/... + auth), nhưng chưa có auth → pair từ điện thoại đang trong phiên. Không kiểm quyền — mã 6 số chỉ hiện trên chính màn hình TV, biết mã nghĩa là đang đứng nhìn nó. """ d = hub.pair(req.code.strip(), req.session_id) if d is None: raise HTTPException(404, "Mã ghép không đúng hoặc TV chưa mở trang " "/tv — kiểm tra mã 6 số trên màn hình TV " "rồi nhập lại.") if d.ws is not None: # báo TV đổi màn hình ngay; TV đang rớt thì thôi — hello lúc # reconnect đã mang paired=True, không cần gửi bù try: await d.ws.send_json({"type": "paired"}) except Exception: # noqa: BLE001 — TV chết đúng lúc ghép: vẫn ghép xong d.ws = None return DisplayPairOut(paired=True) @router.websocket("/display/{display_id}") async def ws_display(ws: WebSocket, display_id: str): """Kênh push xuống MỘT TV. Server nói, TV nghe — chiều lên chỉ để phát hiện disconnect (TV read-only theo design §5.4, mọi thao tác ở điện thoại). Message đầu luôn là ``hello`` mang mã pair + trạng thái ghép — TV vẽ được ngay màn hình đúng (mã to khi chưa ghép / chờ cú khi đã ghép) mà không cần hỏi thêm gì. """ await ws.accept() d = hub.register(display_id, ws) if d is None: # Đầy MAX_DISPLAYS toàn kết nối sống (bàn giao 20): accept rồi đóng # ngay với lý do ngắn — 1013 Try Again Later, TV client sẵn # auto-reconnect nên không sửa FE. Không cấp entry, không hello. await ws.close(code=1013, reason=f"Đủ {MAX_DISPLAYS} TV đang kết " f"nối — thử lại sau.") return await ws.send_json({"type": "hello", "code": d.code, "paired": d.session_id is not None}) try: while True: # TV không có gì để nói — nhận rồi bỏ, chờ WebSocketDisconnect. await ws.receive_text() except WebSocketDisconnect: pass finally: hub.detach(display_id, ws)