flowdeck/app/services/realtime_server.py

"""FlowDeck — v5.13.0 Realtime: WebSocket gateway, présences, curseurs live,
merge des opérations de blocs (last-write-wins par bloc) + version de page.

Rooms in-memory (un seul worker uvicorn). Persistance en base (page.content)
avec debounce. Fallback polling côté client si le WS est indisponible.
"""
from __future__ import annotations

import asyncio
import json
import logging

from fastapi import WebSocket

from app.db import get_conn

logger = logging.getLogger(__name__)

COLORS = ["#2383E2", "#46A758", "#E5484D", "#F76B15", "#8E4EC6", "#12A594",
          "#FFC53D", "#D6409F", "#0091FF", "#3E63DD", "#30A46C", "#FF3333"]


def color_for(uid: int) -> str:
    return COLORS[(uid or 0) % len(COLORS)]


def block_id() -> str:
    import time
    return f"b{int(time.time()*1000)}"


def apply_op(blocks: list[dict], op: dict) -> list[dict]:
    """Apply one block op (insert/update/delete/move) — LWW par bloc."""
    t = op.get("type")
    if t == "insert":
        blk = op.get("block") or {}
        if not blk.get("id"):
            blk = dict(blk)
            blk["id"] = block_id()
        idx = op.get("index")
        if not isinstance(idx, int):
            idx = len(blocks)
        idx = max(0, min(idx, len(blocks)))
        return blocks[:idx] + [blk] + blocks[idx:]
    if t == "update":
        nb = op.get("block") or {}
        if not nb.get("id"):
            return blocks
        return [nb if b.get("id") == nb["id"] else b for b in blocks]
    if t == "delete":
        bid = op.get("id")
        return [b for b in blocks if b.get("id") != bid]
    if t == "move":
        bid = op.get("id")
        idx = op.get("index", 0) or 0
        out = [b for b in blocks if b.get("id") != bid]
        idx = max(0, min(idx, len(out)))
        moved = next((b for b in blocks if b.get("id") == bid), None)
        if moved is None:
            return blocks
        out.insert(idx, moved)
        return out
    return blocks


def merge_ops(blocks: list[dict], ops: list[dict]) -> list[dict]:
    """Apply a batch of ops sequentially (arrival order)."""
    out = blocks
    for op in ops or []:
        out = apply_op(out, op)
    return out


class Room:
    __slots__ = ("page_id", "blocks", "title", "version", "conns",
                 "persist_task", "dirty")

    def __init__(self, page_id: int):
        self.page_id = page_id
        self.blocks: list[dict] = []
        self.title = ""
        self.version = 0
        self.conns: set["RTConn"] = set()
        self.persist_task: asyncio.Task | None = None
        self.dirty = False


class RTConn:
    __slots__ = ("ws", "user", "page_id")

    def __init__(self, ws: WebSocket, user: dict, page_id: int):
        self.ws = ws
        self.user = user
        self.page_id = page_id


class RealtimeManager:
    def __init__(self):
        self._rooms: dict[int, Room] = {}

    def room(self, page_id: int) -> Room:
        return self._rooms.setdefault(page_id, Room(page_id))

    @staticmethod
    def _peer(user: dict) -> dict:
        uid = user.get("id") or 0
        return {
            "id": uid,
            "login": user.get("login", ""),
            "full_name": user.get("full_name", "") or user.get("login", ""),
            "color": color_for(uid),
        }

    async def load_room(self, room: Room) -> bool:
        try:
            with get_conn() as conn:
                row = conn.execute(
                    "SELECT title, content, content_format FROM pages WHERE id=? AND deleted_at IS NULL",
                    (room.page_id,),
                ).fetchone()
        except Exception as e:
            logger.warning("realtime load failed: %s", e)
            return False
        if not row:
            return False
        room.title = row["title"] or ""
        if (row["content_format"] or "") == "blocks" and row["content"]:
            try:
                room.blocks = json.loads(row["content"])
            except Exception:
                room.blocks = []
        return True

    async def connect(self, ws: WebSocket, page_id: int, user: dict) -> RTConn | None:
        room = self.room(page_id)
        if not room.conns and not await self.load_room(room):
            await ws.close(code=4404)
            return None
        conn = RTConn(ws, user, page_id)
        room.conns.add(conn)
        me = self._peer(user)
        peers = [self._peer(c.user) for c in room.conns if c is not conn]
        await ws.send_json({"t": "welcome", "self": me,
                            "peers": peers, "color": me["color"]})
        await ws.send_json({"t": "sync", "blocks": room.blocks,
                            "title": room.title, "version": room.version})
        for c in room.conns:
            if c is not conn:
                try:
                    await c.ws.send_json({"t": "peer_join", "peer": me})
                except Exception:
                    pass
        return conn

    async def disconnect(self, conn: RTConn):
        room = self._rooms.get(conn.page_id)
        if not room:
            return
        room.conns.discard(conn)
        for c in room.conns:
            try:
                await c.ws.send_json({"t": "peer_leave",
                                      "id": conn.user.get("id") or 0})
            except Exception:
                pass
        if not room.conns:
            await self.flush(room)
            self._rooms.pop(conn.page_id, None)

    async def flush(self, room: Room):
        """Write current room state to DB (sync, used on idle + disconnect)."""
        if room.persist_task and not room.persist_task.done():
            room.persist_task.cancel()
        await self._persist(room)

    async def _persist(self, room: Room):
        try:
            with get_conn() as conn:
                conn.execute(
                    "UPDATE pages SET content=?, title=?, updated_at=CURRENT_TIMESTAMP WHERE id=?",
                    (json.dumps(room.blocks, ensure_ascii=False), room.title, room.page_id),
                )
                conn.commit()
            room.dirty = False
        except Exception as e:
            logger.warning("realtime persist failed: %s", e)

    def _schedule_persist(self, room: Room):
        if room.persist_task and not room.persist_task.done():
            return
        room.dirty = True

        async def _run():
            try:
                await asyncio.sleep(1.2)
                await self._persist(room)
            except asyncio.CancelledError:
                pass

        room.persist_task = asyncio.create_task(_run())

    async def _broadcast(self, room: Room, msg: dict, exclude: RTConn | None = None):
        for c in room.conns:
            if c is exclude:
                continue
            try:
                await c.ws.send_json(msg)
            except Exception:
                pass

    async def handle(self, conn: RTConn, msg: dict):
        room = self._rooms.get(conn.page_id)
        if not room:
            return
        t = msg.get("t")
        me = (conn.user.get("id") or 0)

        if t == "hello":
            try:
                await conn.ws.send_json({"t": "sync", "blocks": room.blocks,
                                         "title": room.title, "version": room.version})
            except Exception:
                pass
            return

        if t == "sync_req":
            try:
                await conn.ws.send_json({"t": "sync", "blocks": room.blocks,
                                         "title": room.title, "version": room.version})
            except Exception:
                pass
            return

        if t == "ping":
            try:
                await conn.ws.send_json({"t": "pong"})
            except Exception:
                pass
            return

        if t == "op":
            op = msg.get("op") or {}
            client_v = msg.get("v", 0)
            room.blocks = apply_op(room.blocks, op)
            room.version += 1
            self._schedule_persist(room)
            stale = client_v < room.version - 1
            await self._broadcast(room, {"t": "op", "op": op, "from": me,
                                         "v": room.version}, exclude=conn)
            try:
                await conn.ws.send_json({"t": "ack", "v": room.version,
                                         "stale": stale})
            except Exception:
                pass
            if stale:
                try:
                    await conn.ws.send_json({"t": "sync",
                                             "blocks": room.blocks,
                                             "title": room.title,
                                             "version": room.version})
                except Exception:
                    pass
            return

        if t == "title":
            fmt = (msg.get("title") or "").strip()
            if fmt and fmt != room.title:
                room.title = fmt
                room.version += 1
                self._schedule_persist(room)
            await self._broadcast(room, {"t": "title", "title": room.title,
                                         "from": me, "v": room.version}, exclude=conn)
            try:
                await conn.ws.send_json({"t": "ack", "v": room.version,
                                         "stale": False})
            except Exception:
                pass
            return

        if t == "sel":
            await self._broadcast(room, {"t": "sel", "from": me,
                                         "peer": self._peer(conn.user),
                                         "block": msg.get("block"),
                                         "offset": msg.get("offset", 0)}, exclude=conn)
            return

    async def room_state(self, page_id: int) -> dict:
        room = self.room(page_id)
        if not room.conns and not room.blocks:
            await self.load_room(room)
        return {"blocks": room.blocks, "title": room.title, "version": room.version}


manager = RealtimeManager()