app/displays.py

"""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)