diff --git a/CHANGELOG.md b/CHANGELOG.md index 060b8ce..3ff2b9a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,40 @@ # Changelog - FlowDeck +## v6.4.0 (2026-09-22) — Realtime editing (production) + +> Le realtime passe en mode production : résolution de conflits au-delà du last-write-wins (merge à trois versions) et édition à grande échelle (broadcast non bloquant, coalescence des curseurs, corrections de fuites, observabilité). + +### Added + +- **Merge à 3-voix (diff3-lite)** — nouveau `app/services/realtime_merge.py` : + - `merge_text_3way(base, current, incoming)` : fusion de caractères. Les régions modifiées qui ne se chevauchent pas sont **toutes conservées** (les deux saisies survivent) ; ordre d'arrivée indifférent ; un chevauchement réel retombe en LWW et est signalé. + - `merge_block_3way(base, current, incoming)` : fusion **champ-par-champ**. Serveur et client modifiant des champs différents n'ont plus de conflit ; conflit limité au seul champ divergent (plus « tout le bloc perdu »). + - Fonctions pures, testées sans WebSocket ni base de données. +- **Protocole `base`** — chaque `update` envoyé par le client embarque la version du bloc dont dérive sa saisie ; le serveur calcule `merge_block_3way(base, current, incoming)`. Sans `base` (ancien client) → LWW historique, rétro-compatible. +- **Adoption du résultat fusionné** — l'`ack` renvoie le bloc fusionné + drapeau `conflict` ; le client met à jour sa `base`, adopte le résultat (hors bloc en cours d'édition) et affiche un toast en cas de conflit. Le serveur diffuse toujours le bloc final fusionné pour convergence de tous les pairs. +- **Broadcast non bloquant** — chaque connexion a une file sortante (`asyncio.Queue`) + une tâche `_writer` dédiée ; `_broadcast()` fait `put_nowait` et n'attend plus le socket → un client lent ne fige plus la room entière. +- **Coalescence des curseurs** — le writer réduit les messages `sel` empilés à la position la plus récente (seule la dernière compte), en préservant l'ordre des messages importants. +- **Anti-flood** — budget d'opérations par connexion (`OP_WINDOW_MAX=400` / fenêtre 10 s) ; au-delà, réponse `ack stale` sans application. +- **Observabilité** — `GET /api/realtime/stats` (authentifié) : rooms, connexions (dont totales), ops, merges, conflits, déconnexions lentes, + détail par page. + +### Changed + +- **Déconnexion des clients trop lents** — file sortante pleine (`MAX_OUT_QUEUE=512`) → fermeture 4413 + compteur `slow_disconnects`, pour qu'une room ne stagne jamais sur un pair mortel. +- `app/templates/_page_editor_realtime.html` — les updates incluent désormais `base` ; nouveau traitement de l'`ack` avec `merged`. + +### Fixed + +- **Fuite de rooms** — une tentative de connexion WS sur une page inexistante (4404) enregistrait une `Room` orpheline en mémoire pour toujours ; la room n'est plus enregistrée tant que la page n'est pas validée. +- **`room_state()` sur page inexistante** — retournait/plantait sur `None` ; retourne désormais `{"blocks": [], "title": "", "version": 0}` sans laisser d'entrée dans `_rooms`. + +### Tests + +- **26 nouveaux tests** `tests/test_realtime_v64.py` : merge purs (régions disjointes, ordre indépendant, chevauchement → conflit, suppression de champ, id préservé), protocole WS (ack fusionné, convergence de 2 clients sur le même bloc, rétro-compat LWW sans `base`, conflit scalaire), fuite 4404, coalescence des curseurs, endpoint stats, anti-flood. +- **14 tests** `tests/test_realtime.py` existants préservés (non-régression). +- `ruff check app tests` OK · `eslint static/js` 0 problème. + +--- + ## v6.3.0 (2026-09-21) — API publique complète v2 (REST + scopes + OpenAPI) > L'API publique `/api/v2` devient une surface REST complète façon Notion : CRUD sur tous les domaines (collections, pages, propriétés, vues, commentaires, notifications, favoris, tags, partage, sprints, templates, forges, admin…), auth Bearer + scopes hiérarchiques, pagination/filtres/tri, erreurs RFC 7807, idempotence, audit et webhooks. Wrappers sur les services existants — un seul chemin de code. diff --git a/README.md b/README.md index 31560b9..55af1b2 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ Clone complet de **Notion** intégré nativement à **Gitea** — Databases, Pages, Kanban, Calendar, Gallery, Timeline, List, Multi-Users. -> **v2.1.0** — API publique, Webhooks sortants, PWA +> **v6.4.0** — Realtime production (merge 3-voix, broadcast non bloquant), API publique v2, PWA offline ## Quick Start @@ -87,7 +87,7 @@ DATABASE_URL=sqlite:////data/flowdeck.db ## Tests ```bash -python3 -m pytest tests/ -v # 73/73 passent +python3 -m pytest tests/ -v # 725/725 passent ``` ## Roadmap diff --git a/ROADMAP.md b/ROADMAP.md index 3b80190..d288847 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -814,6 +814,41 @@ Détails livrés : - [x] `tests/test_public_api_v2.py` — **24 tests** (auth scopes, pagination, filtres, RFC 7807, idempotency, webhooks deliveries, CRUD multi-domaines) - [x] Vérif `ruff check app tests` + `pytest -n auto` → **668 verte** +## v6.4.0 — Realtime editing (production) ✅ (2026-09-22) + +> **Objectif** : passer le realtime v5.13.0 en « production » — résolution de +> conflits au-delà du last-write-wins + édition à grande échelle (broadcast non +> bloquant, plusieurs rooms/pages, observabilité). **COMPLETED**. + +#### Conflits au-delà du LWW + +- [x] **Merge à 3 voix (diff3-lite)** — nouveau service `app/services/realtime_merge.py` : `merge_text_3way()` (merge de caractères) + `merge_block_3way()` (merge champ-par-champ), fonctions pures et testées sans WebSocket ni base +- [x] **Régions disjointes conservées** — deux utilisateurs tapant à des endroits différents du *même* bloc voient leurs deux saisies survivre (au lieu d'écraser l'une par l'autre) ; ordre d'arrivée indifférent +- [x] **Chevauchement réel → LWW par champ + drapeau** — un conflit n'est plus « tout le bloc perdu » mais limité au champ concerné ; `conflict: true` renvoyé dans l'`ack` et le broadcast +- [x] **Protocole `base`** — le client embarque dans chaque `update` la version du bloc dont dérive sa saisie ; le serveur fait `merge_block_3way(base, current, incoming)` ; sans `base` → LWW historique (rétro-compat ancien client) +- [x] **Adoption côté client** — l'`ack` renvoie le bloc fusionné ; le client met à jour sa `base`, adopte le résultat (hors bloc en cours d'édition) et affiche un toast en cas de conflit +- [x] **Broadcast du résultat fusionné** — le serveur diffuse toujours le bloc final fusionné (jamais la proposition brute) pour convergence garantie de tous les clients + +#### Échelle & robustesse + +- [x] **Broadcast non bloquant** — chaque connexion a une file sortante (`asyncio.Queue`) + une tâche `_writer` dédiée ; `_broadcast()` fait `put_nowait` et n'attend plus le socket → un client lent ne fige plus la room +- [x] **Coalescence des curseurs** — le writer réduit les messages `sel` empilés à la position la plus récente (seule la dernière compte), tout en préservant l'ordre des messages importants +- [x] **Déconnexion des clients trop lents** — file pleine (`MAX_OUT_QUEUE=512`) → fermeture 4413 + compteur `slow_disconnects` (évite qu'une room entière stagne sur un pair mortel) +- [x] **Anti-flood** — budget d'opérations par connexion (`OP_WINDOW_MAX=400` / 10 s) ; au-delà, réponse `ack stale` sans application +- [x] **Fix fuite de rooms** — une connexion 4404 n'enregistre plus de `Room` orpheline en mémoire ; `room_state()` sur page inexistante retourne un dict vide au lieu de planter sur `None` +- [x] **Observabilité** — `GET /api/realtime/stats` (authentifié) : rooms, connexions, ops, merges, conflits, déconnexions lentes + détail par page + +Détails livrés : +- `app/services/realtime_merge.py` — merge 3-voix (nouveau, ~170 lignes) +- `app/services/realtime_server.py` — `RTConn` (file + writer + budget), `_apply()` avec merge, `_evict_slow()`, `stats()`, fix fuites +- `app/routers/realtime.py` — endpoint `GET /api/realtime/stats` +- `app/templates/_page_editor_realtime.html` — envoi de `base`, adoption du bloc fusionné, toast de conflit +- **26 tests** `tests/test_realtime_v64.py` (merge purs, protocole WS, convergence 2 clients, rétro-compat LWW, fuite 4404, coalescence, stats, anti-flood) ; **14 tests** `tests/test_realtime.py` préservés +- `ruff check app tests` OK · `eslint static/js` 0 problème +- **Version** — 6.4.0 + +--- + ## v6.0.0 — Pro (futur) - [x] **PWA** — Progressive Web App, offline support ✅ (livré) — [📄 Conception détaillée](/docs/V6_PWA_Progressive_Web_App.md) @@ -821,7 +856,7 @@ Détails livrés : - [x] **Web Clipper** — extension navigateur ✅ (livré v6.2.0/6.2.1) — [📄 Conception détaillée](/docs/V6_Web_Clipper.md) - [x] **API publique complète** — REST API documentée (OpenAPI) ✅ (livré v6.3.0) — [📄 API Guide v2](/docs/API_GUIDE_V6.md) · [📄 OpenAPI](/docs/openapi-v2.json) - [ ] **SSO/SAML** — enterprise authentication — [📄 Conception détaillée](/docs/V6_SSO_SAML_Enterprise_Auth.md) -- [ ] **Realtime editing (production)** — voir **v5.13.0** (curseurs + présence déjà avancés ici) ; reste en v6 : conflits avancés, édition large échelle +- [x] **Realtime editing (production)** ✅ livré **v6.4.0** (merge 3-voix au-delà du LWW, broadcast non bloquant) ; voir **v5.13.0** pour le socle (curseurs + présence) - [ ] **Synced blocks (production)** — voir **v5.14.0** (bloc de base) ; reste en v6 : syncing côté databases/vues --- @@ -868,4 +903,4 @@ Quality DB views, Agent IA Palette → Realtime + E DB avancée, redo, drag&drop, bookmark, avancée Calendrier, AI duplicate) lightbox…) (Pt.2) v6.1 ✅ v6.2 ✅ v6.3 ✅ -*Dernière mise à jour: 2026-09-21 — **v6.3.0 API publique complète v2 COMPLETED** (~100 endpoints `/api/v2`, scopes hiérarchiques `read **Début**: 2026-07-08 | **Version**: v2.2.0 | **Statut**: EN COURS 🔄 -> **Cible v3.0**: Multi-User, Multi-Forge (Gitea/GitHub), Standalone +> **Début**: 2026-07-08 | **Version**: v6.4.0 | **Statut**: EN COURS 🔄 +> **Cible**: parité Notion + intégration forge · **Reste roadmap**: SSO/SAML, synced blocks prod (databases/vues) ## Avancement Global @@ -21,7 +21,10 @@ | v2.0 | Multi-User + Editor Complete | ✅ | 67/67 | | v2.1 | Public API, Webhooks, PWA | ✅ | 73/73 | | v2.2 | Share/Publish, Favorites, Library | ✅ | 73/73 | -| v3.0 | **Auth locale, Multi-Forge, Standalone** | 🔲 | — | +| v3.0 | Auth locale, Multi-Forge, Standalone | ✅ | — | +| v4.x–v5.x | MVP → Agent IA, palette, automations, import, calendrier, wiki-links, synced blocks | ✅ | 523+ | +| v6.0–v6.3 | PWA offline, permissions granulaires, web clipper, API publique v2 | ✅ | 668+ | +| **v6.4.0** | **Realtime production (merge 3-voix, broadcast non bloquant)** | ✅ | **749+** | ## Blocs Complétés @@ -58,5 +61,5 @@ CRUD collections/pages, 5 vues HTML, relations/rollups/formulas, sub-items/depen - **BDD**: SQLite WAL mode, 21 tables, foreign keys ON - **Auth**: OAuth2 Gitea + sessions signed (itsdangerous) + token API - **Déploiement**: Docker (python:3.12-slim), docker-compose, port 8080 -- **Tests**: pytest, 73 tests, TestClient avec SQLite temporaire +- **Tests**: pytest, 749+ tests, TestClient avec SQLite temporaire - **CI/CD**: Gitea Actions (.gitea/workflows/ci.yml) diff --git a/app/main.py b/app/main.py index d447fb6..a365e99 100644 --- a/app/main.py +++ b/app/main.py @@ -122,7 +122,7 @@ async def lifespan(_app: FastAPI): app = FastAPI( title="FlowDeck", - version="6.3.0", + version="6.4.0", docs_url="/docs", redoc_url="/redoc", lifespan=lifespan, diff --git a/app/routers/realtime.py b/app/routers/realtime.py index 0f3d1d8..f1378fe 100644 --- a/app/routers/realtime.py +++ b/app/routers/realtime.py @@ -7,7 +7,7 @@ from __future__ import annotations import json import logging -from fastapi import APIRouter, WebSocket +from fastapi import APIRouter, Request, WebSocket from starlette.websockets import WebSocketDisconnect from app.auth.session import SessionManager @@ -17,6 +17,20 @@ logger = logging.getLogger(__name__) router = APIRouter(tags=["realtime"]) +@router.get("/api/realtime/stats") +async def realtime_stats(request: Request): + """Observabilité realtime v6.4.0 : rooms, connexions, ops, merges, conflits. + + Réservé aux utilisateurs authentifiés (données d'activité internes). + """ + user = SessionManager.decode_session( + request.cookies.get("flowdeck_session", "") + ) + if not user or not user.get("id"): + return {"error": "unauthorized"} + return manager.stats() + + @router.websocket("/ws/pages/{page_id}") async def ws_page(websocket: WebSocket, page_id: int): await websocket.accept() diff --git a/app/services/realtime_merge.py b/app/services/realtime_merge.py new file mode 100644 index 0000000..e9c23cc --- /dev/null +++ b/app/services/realtime_merge.py @@ -0,0 +1,164 @@ +"""FlowDeck — v6.4.0 Realtime: résolution de conflits au-delà du last-write-wins. + +Le LWW par bloc (v5.13.0) écrase intégralement le bloc du dernier arrivé : si deux +utilisateurs tapent dans le *même* bloc, la saisie du premier est perdue. Ce module +implémente un vrai *merge à trois versions* (diff3-lite) : + + base = l'état du bloc dont le client dérive sa saisie (envoyé avec l'op) + current = l'état actuel du bloc côté serveur (déjà mis à jour par d'autres) + incoming = la nouvelle proposition du client + +Règle champ-par-champ : + * incoming == base → le client n'a pas touché ce champ → on garde current + * current == base → le serveur n'a pas touché ce champ → on garde incoming + * current == incoming → les deux ont fait la même chose → sans conflit + * sinon (conflit) + - champ texte (str) : merge de caractères. Les régions modifiées qui ne se + chevauchent pas sont *toutes conservées* (les deux saisies survivent) ; + chevauchement réel → LWW sur ce champ + drapeau de conflit. + - autre type (bool, nombre…) : LWW sur ce champ + drapeau de conflit. + +Toutes les fonctions sont pures et testables sans WebSocket ni base de données. +""" +from __future__ import annotations + +from typing import Any + +__all__ = ["merge_text_3way", "merge_block_3way", "changed_region"] + + +def changed_region(base: str, other: str) -> tuple[int, int, str] | None: + """Région de `base` remplacée par `other` (trim préfixe/suffixe commun). + + Retourne ``(start, end, replacement)`` tel que ``base[:start] + replacement + + base[end:] == other``, ou ``None`` si ``other == base`` (aucun changement). + """ + if base == other: + return None + minlen = min(len(base), len(other)) + prefix = 0 + while prefix < minlen and base[prefix] == other[prefix]: + prefix += 1 + suffix = 0 + # ne jamais chevaucher le préfixe déjà consommé + while (suffix < len(base) - prefix and suffix < len(other) - prefix + and base[len(base) - 1 - suffix] == other[len(other) - 1 - suffix]): + suffix += 1 + return prefix, len(base) - suffix, other[prefix:len(other) - suffix] + + +def merge_text_3way(base: str, current: str, incoming: str) -> tuple[str, bool]: + """Merge à trois versions d'une chaîne. Retourne ``(texte, conflit)``. + + Les éditions qui ne se chevauchent pas sont toutes les deux conservées ; + un chevauchement réel retombe en LWW (``incoming`` gagne) et signale le conflit. + """ + if current == incoming: + return current, False + if current == base: + return incoming, False + if incoming == base: + return current, False + + rc = changed_region(base, current) + ri = changed_region(base, incoming) + if rc is None: + return incoming, False + if ri is None: + return current, False + + c_start, c_end, c_text = rc + i_start, i_end, i_text = ri + + # Régions disjointes (ou juste adjacentes) → appliquer les deux sur base. + if c_end <= i_start or i_end <= c_start: + edits = sorted([(c_start, c_end, c_text), (i_start, i_end, i_text)], + key=lambda e: e[0]) + out: list[str] = [] + pos = 0 + for start, end, text in edits: + if start < pos: + continue # sécurité: ne jamais réappliquer par-dessus + out.append(base[pos:start]) + out.append(text) + pos = end + out.append(base[pos:]) + return "".join(out), False + + # Chevauchement réel → LWW sur ce champ, conflit signalé. + return incoming, True + + +def _scalar_conflict(base: Any, current: Any, incoming: Any) -> tuple[Any, bool]: + """Conflit sur un champ non-texte : LWW (incoming gagne).""" + if current == incoming: + return current, False + if current == base: + return incoming, False + if incoming == base: + return current, False + return incoming, True + + +def merge_block_3way(base_blk: Any, current_blk: Any, + incoming_blk: Any) -> tuple[dict, list[str]]: + """Merge à trois versions d'un bloc entier. + + Retourne ``(bloc fusionné, champs en conflit)``. Ne lève jamais d'exception : + une entrée non-dict retombe en LWW (``incoming``) avec conflit signalé sur + ``__block__`` pour que l'appelant puisse journaliser. + """ + if not isinstance(base_blk, dict): + base_blk = {} + if not isinstance(current_blk, dict): + current_blk = {} + if not isinstance(incoming_blk, dict): + # proposition invalide → on garde l'état serveur + return dict(current_blk), ["__block__"] + + keys = set(base_blk) | set(current_blk) | set(incoming_blk) + merged: dict[str, Any] = {} + conflicts: list[str] = [] + + for key in keys: + b = base_blk.get(key) + c = current_blk.get(key) + i = incoming_blk.get(key) + + if i == b: + # le client n'a pas modifié ce champ → valeur serveur. Si le serveur + # a *supprimé* le champ (absent de current) alors la suppression doit + # gagner : on n'insère pas de clé fantôme value=None. + if key not in current_blk: + continue + merged[key] = c + elif c == b: + # le serveur n'a pas modifié ce champ → valeur client + merged[key] = i + elif c == i: + merged[key] = c + else: + # les deux ont changé, différemment + if isinstance(b, str) and isinstance(c, str) and isinstance(i, str): + text, conflict = merge_text_3way(b, c, i) + merged[key] = text + if conflict: + conflicts.append(key) + else: + value, conflict = _scalar_conflict(b, c, i) + merged[key] = value + if conflict: + conflicts.append(key) + + # Un champ supprimé par le serveur et absent de la proposition client doit + # rester supprimé (pas de réapparition d'une valeur None fantôme). + merged = {k: v for k, v in merged.items() + if not (v is None and k not in incoming_blk)} + + if "id" not in merged: + # garantir l'identité du bloc même si base était vide + ident = incoming_blk.get("id") or current_blk.get("id") + if ident: + merged["id"] = ident + + return merged, conflicts diff --git a/app/services/realtime_server.py b/app/services/realtime_server.py index b8d06f3..fe4909f 100644 --- a/app/services/realtime_server.py +++ b/app/services/realtime_server.py @@ -1,5 +1,23 @@ -"""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. +"""FlowDeck — v6.4.0 Realtime: WebSocket gateway, présences, curseurs live, +merge à trois versions des opérations de blocs (au-delà du last-write-wins) + +version de page. + +Améliorations « production » par rapport à v5.13.0 : + +* **Conflits** — les mises à jour de bloc embarquent la ``base`` dont dérive la + saisie du client ; le serveur fait un merge 3-voix champ-par-champ + texte + (voir ``realtime_merge``) au lieu d'écraser le bloc entier. Les deux saisies + disjointes survivent, les chevauchements réels retombent en LWW *par champ* + avec drapeau de conflit renvoyé au client. +* **Échelle** — chaque connexion possède une file sortante + une tâche writer + dédiée ; le broadcast devient non bloquant (un client lent ne bloque plus la + room), les mises à jour de curseur se coalescent (une seule par flush), et un + client trop lent (file pleine) est déconnecté proprement (4413). +* **Anti-flood** — budget d'opérations par connexion (fenêtre glissante). +* **Fuites corrigées** — une room 4404 n'est plus enregistrée ; ``room_state`` + ne crée plus d'objet None ; les rooms vides sont évacuées. +* **Observabilité** — compteurs (rooms, conns, ops, merges, conflits, déconnexions + lentes) exposés par ``stats()``. Rooms in-memory (un seul worker uvicorn). Persistance en base (page.content) avec debounce. Fallback polling côté client si le WS est indisponible. @@ -9,16 +27,25 @@ from __future__ import annotations import asyncio import json import logging +import time from fastapi import WebSocket from app.db import get_conn +from app.services.realtime_merge import merge_block_3way logger = logging.getLogger(__name__) COLORS = ["#2383E2", "#46A758", "#E5484D", "#F76B15", "#8E4EC6", "#12A594", "#FFC53D", "#D6409F", "#0091FF", "#3E63DD", "#30A46C", "#FF3333"] +# File sortante maximale par connexion au-delà de laquelle le client est +# considéré comme trop lent et déconnecté (évite qu'une room entière stagne). +MAX_OUT_QUEUE = 512 +# Nombre maximal d'opérations acceptées par connexion et par fenêtre (anti-flood). +OP_WINDOW_SECONDS = 10.0 +OP_WINDOW_MAX = 400 + def color_for(uid: int) -> str: return COLORS[(uid or 0) % len(COLORS)] @@ -48,7 +75,11 @@ def ensure_block_ids(blocks: list[dict]) -> list[dict]: def apply_op(blocks: list[dict], op: dict) -> list[dict]: - """Apply one block op (insert/update/delete/move) — LWW par bloc.""" + """Apply one block op (insert/update/delete/move) — LWW par bloc. + + Conservé pour la compatibilité : tests et chemins sans ``base`` continuent + de fonctionner. Le merge 3-voix vit dans ``RealtimeManager._apply``. + """ t = op.get("type") if t == "insert": blk = op.get("block") or {} @@ -90,9 +121,40 @@ def merge_ops(blocks: list[dict], ops: list[dict]) -> list[dict]: return out +class RTConn: + """Une connexion WS : file sortante + tâche writer dédiée. + + Le broadcast ne fait que ``put_nowait`` dans la file ; c'est la tâche writer + qui consomme et écrit sur le socket. Un client lent n'empêche donc jamais + les autres membres de la room de recevoir les messages. + """ + __slots__ = ("ws", "user", "page_id", "out_q", "writer", "closed", + "_ops_count", "_ops_window_start", "created_at") + + def __init__(self, ws: WebSocket, user: dict, page_id: int): + self.ws = ws + self.user = user + self.page_id = page_id + self.out_q: asyncio.Queue = asyncio.Queue(maxsize=MAX_OUT_QUEUE) + self.writer: asyncio.Task | None = None + self.closed = False + self._ops_count = 0 + self._ops_window_start = time.monotonic() + self.created_at = time.monotonic() + + def op_budget_ok(self) -> bool: + """Fenêtre glissante simple anti-flood d'opérations.""" + now = time.monotonic() + if now - self._ops_window_start > OP_WINDOW_SECONDS: + self._ops_window_start = now + self._ops_count = 0 + self._ops_count += 1 + return self._ops_count <= OP_WINDOW_MAX + + class Room: __slots__ = ("page_id", "blocks", "title", "version", "conns", - "persist_task", "dirty") + "persist_task", "dirty", "merge_count", "conflict_count") def __init__(self, page_id: int): self.page_id = page_id @@ -102,21 +164,21 @@ class Room: 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 + self.merge_count = 0 + self.conflict_count = 0 class RealtimeManager: def __init__(self): self._rooms: dict[int, Room] = {} + # compteurs globaux (observabilité) + self.stat_ops = 0 + self.stat_merges = 0 + self.stat_conflicts = 0 + self.stat_slow_disconnects = 0 + self.stat_connections_total = 0 + # ── rooms ──────────────────────────────────────────────────────────── def room(self, page_id: int) -> Room: return self._rooms.setdefault(page_id, Room(page_id)) @@ -150,42 +212,86 @@ class RealtimeManager: room.blocks = [] return True + # ── entrées / sorties ──────────────────────────────────────────────── async def connect(self, ws: WebSocket, page_id: int, user: dict) -> RTConn | None: - room = self.room(page_id) + # Ne JAMAIS enregistrer la room avant d'avoir validé l'existence de la + # page : avant, un 4404 laissait une Room orpheline en mémoire pour + # toujours (fuite). + existing = self._rooms.get(page_id) + room = existing if existing is not None else Room(page_id) if not room.conns and not await self.load_room(room): await ws.close(code=4404) - return None + return None # room non enregistrée → pas de fuite + if existing is None: + self._rooms[page_id] = room + conn = RTConn(ws, user, page_id) + self.stat_connections_total += 1 + conn.writer = asyncio.create_task(self._writer(conn)) 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 + await self._broadcast(room, {"t": "peer_join", "peer": me}, exclude=conn) return conn async def disconnect(self, conn: RTConn): room = self._rooms.get(conn.page_id) - if not room: + if room is None: + self._close_writer(conn) 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 + self._close_writer(conn) + await self._broadcast(room, {"t": "peer_leave", + "id": conn.user.get("id") or 0}) if not room.conns: await self.flush(room) self._rooms.pop(conn.page_id, None) + def _close_writer(self, conn: RTConn): + conn.closed = True + t = conn.writer + if t is not None and not t.done(): + t.cancel() + conn.writer = None + + async def _writer(self, conn: RTConn): + """Tâche dédiée : consomme la file sortante et écrit sur le socket. + + Sert aussi de filet de coalescence : après chaque message consommé, les + mises à jour de curseur (``sel``) déjà empilées sont réduites à la + dernière (les curseurs n'ont pas besoin d'être ordonnés entre eux, seule + la position la plus récente compte). + """ + try: + while not conn.closed: + msg = await conn.out_q.get() + if msg is None: + return + await conn.ws.send_json(msg) + # coalescence des curseurs en attente + pending_sel = [] + while not conn.out_q.empty(): + nxt = conn.out_q.get_nowait() + if nxt is None: + return + if isinstance(nxt, dict) and nxt.get("t") == "sel": + pending_sel.append(nxt) + else: + await conn.ws.send_json(nxt) + if pending_sel: + await conn.ws.send_json(pending_sel[-1]) + except asyncio.CancelledError: + raise + except Exception as e: # socket mort → on arrête proprement + logger.debug("writer stopped: %s", e) + conn.closed = True + + # ── persistance ────────────────────────────────────────────────────── 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(): @@ -218,67 +324,102 @@ class RealtimeManager: room.persist_task = asyncio.create_task(_run()) + # ── broadcast non bloquant ─────────────────────────────────────────── + def _enqueue(self, conn: RTConn, msg: dict) -> bool: + """Pose le message dans la file du client. Retourne False si trop lent.""" + if conn.closed: + return False + try: + conn.out_q.put_nowait(msg) + return True + except asyncio.QueueFull: + return False + + async def _evict_slow(self, room: Room, conn: RTConn): + """Client dépassé : on le déconnecte pour ne pas figer la room.""" + self.stat_slow_disconnects += 1 + logger.info("realtime: evicting slow client (user=%s page=%s)", + conn.user.get("login"), conn.page_id) + room.conns.discard(conn) + self._close_writer(conn) + try: + await conn.ws.close(code=4413) + except Exception: + pass + async def _broadcast(self, room: Room, msg: dict, exclude: RTConn | None = None): - for c in room.conns: + """Enfile ``msg`` chez chaque membre — jamais d'attente sur le socket.""" + slow: list[RTConn] = [] + for c in list(room.conns): if c is exclude: continue - try: - await c.ws.send_json(msg) - except Exception: - pass + if not self._enqueue(c, msg): + slow.append(c) + for c in slow: + await self._evict_slow(room, c) + async def _send(self, conn: RTConn, msg: dict): + if not self._enqueue(conn, msg): + room = self._rooms.get(conn.page_id) + if room is not None: + await self._evict_slow(room, conn) + + # ── protocole ──────────────────────────────────────────────────────── async def handle(self, conn: RTConn, msg: dict): room = self._rooms.get(conn.page_id) - if not room: + if room is None: 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 + await self._send(conn, {"t": "sync", "blocks": room.blocks, + "title": room.title, "version": room.version}) 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 + await self._send(conn, {"t": "sync", "blocks": room.blocks, + "title": room.title, "version": room.version}) return if t == "ping": - try: - await conn.ws.send_json({"t": "pong"}) - except Exception: - pass + await self._send(conn, {"t": "pong"}) return if t == "op": + if not conn.op_budget_ok(): + # trop d'ops : on ignore silencieusement (le client resync) + await self._send(conn, {"t": "ack", "v": room.version, + "stale": True}) + return op = msg.get("op") or {} client_v = msg.get("v", 0) - room.blocks = apply_op(room.blocks, op) + result = self._apply(room, op) room.version += 1 + self.stat_ops += 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 + + # broadcast : on diffuse TOUJOURS le bloc final fusionné (pas la + # proposition brute) pour que tous les clients convergent. + out_op = dict(op) + if result.get("merged") is not None: + out_op["block"] = result["merged"] + out_op["merged"] = True + await self._broadcast(room, {"t": "op", "op": out_op, "from": me, + "v": room.version, + "conflict": result.get("conflict", False)}, + exclude=conn) + + ack = {"t": "ack", "v": room.version, "stale": stale} + if result.get("merged") is not None: + ack["merged"] = result["merged"] + ack["conflict"] = result.get("conflict", False) + await self._send(conn, ack) + if stale: - try: - await conn.ws.send_json({"t": "sync", - "blocks": room.blocks, - "title": room.title, - "version": room.version}) - except Exception: - pass + await self._send(conn, {"t": "sync", "blocks": room.blocks, + "title": room.title, "version": room.version}) return if t == "title": @@ -289,11 +430,7 @@ class RealtimeManager: 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 + await self._send(conn, {"t": "ack", "v": room.version, "stale": False}) return if t == "sel": @@ -303,16 +440,72 @@ class RealtimeManager: "offset": msg.get("offset", 0)}, exclude=conn) return + def _apply(self, room: Room, op: dict) -> dict: + """Applique une opération en mergeant à 3 voix si le client fournit + une ``base``. Retourne ``{"merged": bloc|None, "conflict": bool}``. + + * ``update`` avec ``base`` → merge 3-voix (au-delà du LWW). + * le reste (insert/delete/move, ou update sans base) → LWW historique. + """ + if op.get("type") != "update" or "base" not in op: + room.blocks = apply_op(room.blocks, op) + return {"merged": None, "conflict": False} + + incoming = op.get("block") or {} + base = op.get("base") + bid = incoming.get("id") + if not bid: + return {"merged": None, "conflict": False} + + current = next((b for b in room.blocks if b.get("id") == bid), None) + if current is None: + # bloc introuvable : LWW historique = pas d'update possible + return {"merged": None, "conflict": False} + + merged, conflicts = merge_block_3way(base, current, incoming) + room.merge_count += 1 + conflict = bool(conflicts) + if conflict: + room.conflict_count += 1 + self.stat_conflicts += 1 + logger.debug("realtime conflict page=%s block=%s fields=%s", + room.page_id, bid, conflicts) + self.stat_merges += 1 + room.blocks = [merged if b.get("id") == bid else b for b in room.blocks] + return {"merged": merged, "conflict": conflict} + + # ── observabilité ──────────────────────────────────────────────────── async def room_state(self, page_id: int) -> dict: room = self._rooms.get(page_id) - if not room.conns and not room.blocks: + if room is None: + # ne pas créer d'objet None : on charge dans une room jetable + room = Room(page_id) + if not await self.load_room(room): + return {"blocks": [], "title": "", "version": 0} + elif not room.conns and not room.blocks: await self.load_room(room) return {"blocks": room.blocks, "title": room.title, "version": room.version} + def stats(self) -> dict: + rooms = len(self._rooms) + conns = sum(len(r.conns) for r in self._rooms.values()) + return { + "rooms": rooms, + "connections": conns, + "connections_total": self.stat_connections_total, + "ops": self.stat_ops, + "merges": self.stat_merges, + "conflicts": self.stat_conflicts, + "slow_disconnects": self.stat_slow_disconnects, + "pages": [{"page_id": r.page_id, "conns": len(r.conns), + "version": r.version, "merges": r.merge_count, + "conflicts": r.conflict_count} + for r in self._rooms.values()], + } + async def _propagate_synced(self, synced_id: int) -> None: """Broadcast a synced-block update to all rooms that reference it.""" from app.db import get_conn as _get_conn - # Find all pages that reference this synced block pages: list[int] = [] try: with _get_conn() as conn: @@ -326,11 +519,10 @@ class RealtimeManager: for pid in pages: room = self._rooms.get(pid) if room and room.conns: - # Re-load the page content from DB to get fresh synced blocks await self.load_room(room) await self._broadcast(room, {"t": "synced_update", - "synced_id": synced_id, - "version": room.version}) + "synced_id": synced_id, + "version": room.version}) manager = RealtimeManager() diff --git a/app/templates/_page_editor_realtime.html b/app/templates/_page_editor_realtime.html index e9ef1f2..a4750f9 100644 --- a/app/templates/_page_editor_realtime.html +++ b/app/templates/_page_editor_realtime.html @@ -110,7 +110,9 @@ window.__fdRT = (function () { cur.forEach(b => { if (baseById[b.id] !== undefined && JSON.stringify(baseById[b.id]) !== JSON.stringify(b)) { - ops.push({ type: 'update', block: clone(b) }); + // v6.4.0 : on embarque la `base` dont dérive la saisie → le serveur + // fait un merge 3-voix au lieu d'écraser le bloc (LWW). + ops.push({ type: 'update', block: clone(b), base: clone(baseById[b.id]) }); } }); base.forEach(b => { @@ -358,6 +360,27 @@ window.__fdRT = (function () { } else if (m.t === 'ack') { version = m.v || 0; if (pendingOps > 0) pendingOps--; + // v6.4.0 : le serveur renvoie le bloc fusionné (merge 3-voix). On + // l'adopte comme nouvelle base ; s'il diffère de notre saisie locale + // c'est qu'un autre utilisateur avait modifié le même bloc. + if (m.merged) { + const mid = m.merged.id; + const localBlock = (E && E.blocks || []).find(b => b.id === mid); + const differs = localBlock && JSON.stringify(localBlock) !== JSON.stringify(m.merged); + base = applyOpJS(base, { type: 'update', block: m.merged }); + if (differs && E) { + const fid = activeBlockId(); + if (fid !== mid) { + E.blocks = applyOpJS(E.blocks, { type: 'update', block: m.merged }); + E.dirty = true; + E.render(); + refocus(fid); + } + if (m.conflict && window.showToast) { + window.showToast('Editing conflict merged on a block', 'info'); + } + } + } if (m.stale || needSync) { needSync = true; send({ t: 'sync_req' }); } } else if (m.t === 'op') { if (m.v) version = m.v; diff --git a/tests/test_realtime_v64.py b/tests/test_realtime_v64.py new file mode 100644 index 0000000..11401c1 --- /dev/null +++ b/tests/test_realtime_v64.py @@ -0,0 +1,466 @@ +"""FlowDeck — v6.4.0 Realtime production: merge 3-voix (au-delà du LWW), +broadcast non bloquant + coalescence des curseurs, corrections de fuites de +rooms, anti-flood et observabilité. + +Covers : +* ``realtime_merge`` — merge_text_3way / merge_block_3way (purs, sans WS) +* protocole WS avec ``base`` → ack fusionné + drapeau conflict +* convergence de deux clients sur le même bloc (les deux saisies survivent) +* le LWW historique est préservé quand il n'y a pas de ``base`` +* room 4404 non enregistrée (fuite corrigée), ``room_state`` page inexistante +* coalescence des curseurs via la file sortante +* ``GET /api/realtime/stats`` +""" +import json +import os +import tempfile + +import pytest +from fastapi.testclient import TestClient +from starlette.websockets import WebSocketDisconnect + + +@pytest.fixture +def client(): + db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False) + db_path = db_file.name + db_file.close() + + os.environ["DATABASE_URL"] = f"sqlite:///{db_path}" + os.environ["APP_SECRET_KEY"] = "test-secret-for-realtime-v64" + os.environ["RATE_LIMIT_ENABLED"] = "false" + from app.config import settings + settings.database_url = f"sqlite:///{db_path}" + + from app.db import get_conn, init_db + from app.main import app + init_db() + with get_conn() as conn: + conn.execute("INSERT OR IGNORE INTO users (id, login, full_name, is_admin) VALUES (1, 'tester', 'Tester', 1)") + conn.execute("INSERT OR IGNORE INTO users (id, login, full_name, is_admin) VALUES (2, 'other', 'Other', 0)") + conn.commit() + + from app.services.realtime_server import manager + manager._rooms = {} + manager.stat_ops = 0 + manager.stat_merges = 0 + manager.stat_conflicts = 0 + manager.stat_slow_disconnects = 0 + manager.stat_connections_total = 0 + + tc = TestClient(app, raise_server_exceptions=False) + yield tc + + os.unlink(db_path) + + +def _make_page(title="V64 Page", blocks=None): + from app.db import get_conn + with get_conn() as conn: + cur = conn.execute( + "INSERT INTO pages (workspace, title, content, content_format, parent_section) " + "VALUES ('Private', ?, ?, 'blocks', 'Private')", + (title, json.dumps(blocks or [], ensure_ascii=False)), + ) + conn.commit() + return cur.lastrowid + + +def _token(user_id, login): + from app.auth.session import SessionManager + return SessionManager.create_session({"id": user_id, "login": login, + "full_name": login.title(), "is_admin": 1}) + + +def _auth_client(client, user_id=1, login="tester"): + client.cookies.set("flowdeck_session", _token(user_id, login)) + + +# ── realtime_merge (purs) ──────────────────────────────────────────────── + +def test_merge_text_disjoint_edits_survive(): + from app.services.realtime_merge import merge_text_3way + base = "hello world" + cur = "hello brave world" # serveur : insert "brave" + inc = "hello world today" # client : append "today" + out, conflict = merge_text_3way(base, cur, inc) + assert conflict is False + assert "brave" in out and "today" in out + + +def test_merge_text_disjoint_edits_order_independent(): + from app.services.realtime_merge import merge_text_3way + base = "hello world" + a, _ = merge_text_3way(base, "hello brave world", "hello world today") + b, _ = merge_text_3way(base, "hello world today", "hello brave world") + assert a == b == "hello brave world today" + + +def test_merge_text_real_overlap_flags_conflict_lww(): + from app.services.realtime_merge import merge_text_3way + out, conflict = merge_text_3way("hello", "hxllo", "hyllo") + assert conflict is True + assert out == "hyllo" # LWW : incoming gagne + + +def test_merge_text_shortcuts(): + from app.services.realtime_merge import merge_text_3way + assert merge_text_3way("a", "a", "b") == ("b", False) # client a changé + assert merge_text_3way("a", "b", "a") == ("b", False) # serveur a changé + assert merge_text_3way("a", "b", "b") == ("b", False) # même changement + + +def test_merge_text_same_point_insertion_not_destructive(): + from app.services.realtime_merge import merge_text_3way + out, conflict = merge_text_3way("ab", "aXb", "aYb") + # les deux insertions au même index sont conservées (concaténées) + assert "X" in out and "Y" in out + assert conflict is False + + +def test_changed_region(): + from app.services.realtime_merge import changed_region + assert changed_region("hello", "hello") is None + assert changed_region("hello world", "hello brave world") == (6, 6, "brave ") + assert changed_region("abc", "abXc") == (2, 2, "X") + + +def test_merge_block_field_level_no_conflict(): + """Serveur et client changent des champs différents → pas de conflit.""" + from app.services.realtime_merge import merge_block_3way + base = {"id": "1", "text": "hi", "done": False} + cur = {"id": "1", "text": "hi there", "done": False} # serveur édite text + inc = {"id": "1", "text": "hi", "done": True} # client coche done + merged, conflicts = merge_block_3way(base, cur, inc) + assert merged["text"] == "hi there" + assert merged["done"] is True + assert conflicts == [] + + +def test_merge_block_text_disjoint_merges(): + from app.services.realtime_merge import merge_block_3way + base = {"id": "1", "text": "hello world"} + cur = {"id": "1", "text": "hello brave world"} + inc = {"id": "1", "text": "hello world today"} + merged, conflicts = merge_block_3way(base, cur, inc) + assert conflicts == [] + assert "brave" in merged["text"] and "today" in merged["text"] + + +def test_merge_block_scalar_conflict_lww(): + from app.services.realtime_merge import merge_block_3way + merged, conflicts = merge_block_3way({"id": "1", "n": 1}, + {"id": "1", "n": 2}, + {"id": "1", "n": 3}) + assert merged["n"] == 3 + assert "n" in conflicts + + +def test_merge_block_server_deleted_field_stays_deleted(): + from app.services.realtime_merge import merge_block_3way + base = {"id": "1", "tmp": "x"} + cur = {"id": "1"} # serveur a supprimé tmp + inc = {"id": "1", "tmp": "x"} # client n'a pas touché tmp + merged, _ = merge_block_3way(base, cur, inc) + assert "tmp" not in merged + + +def test_merge_block_incoming_non_dict_flagged(): + from app.services.realtime_merge import merge_block_3way + _, conflicts = merge_block_3way({"id": "1"}, {"id": "1"}, "junk") + assert conflicts == ["__block__"] + + +def test_merge_block_preserves_id_when_base_empty(): + from app.services.realtime_merge import merge_block_3way + merged, _ = merge_block_3way({}, {}, {"id": "z", "text": "new"}) + assert merged["id"] == "z" + + +# ── apply_op (LWW historique préservé) ─────────────────────────────────── + +def test_apply_op_still_lww_without_base(): + from app.services.realtime_server import apply_op + blocks = [{"id": "a", "content": "A"}] + out = apply_op(blocks, {"type": "update", "block": {"id": "a", "content": "B"}}) + assert out[0]["content"] == "B" + out = apply_op(blocks, {"type": "delete", "id": "a"}) + assert out == [] + out = apply_op(blocks, {"type": "move", "id": "a", "index": 0}) + assert [b["id"] for b in out] == ["a"] + + +# ── protocole WS ───────────────────────────────────────────────────────── + +def test_ws_requires_auth(client): + _make_page() + with pytest.raises(WebSocketDisconnect) as exc: + with client.websocket_connect("/ws/pages/1") as ws: + ws.receive_json() + assert exc.value.code == 4401 + + +def test_ws_missing_page_4404(client): + _auth_client(client) + with pytest.raises(WebSocketDisconnect) as exc: + with client.websocket_connect("/ws/pages/9999") as ws: + ws.receive_json() + assert exc.value.code == 4404 + + +def test_ws_4404_does_not_leak_room(client): + """Une tentative de connexion sur une page inexistante ne doit pas laisser + de Room orpheline en mémoire (fuite corrigée en v6.4.0).""" + from app.services.realtime_server import manager + _auth_client(client) + with pytest.raises(WebSocketDisconnect): + with client.websocket_connect("/ws/pages/9999") as ws: + ws.receive_json() + assert 9999 not in manager._rooms, f"room leak: {list(manager._rooms)}" + + +def test_ws_hello_gets_sync(client): + pid = _make_page(blocks=[{"id": "a", "type": "paragraph", "content": "Hi"}]) + _auth_client(client) + with client.websocket_connect(f"/ws/pages/{pid}") as ws: + w = ws.receive_json() + assert w["t"] == "welcome" + s = ws.receive_json() + assert s["t"] == "sync" + assert s["blocks"][0]["id"] == "a" + ws.send_json({"t": "hello"}) + s2 = ws.receive_json() + assert s2["t"] == "sync" and s2["blocks"][0]["content"] == "Hi" + + +def test_ws_update_with_base_returns_merged(client): + """Update avec base → ack fusionné, pas de conflit si seul le client a changé.""" + pid = _make_page(blocks=[{"id": "a", "type": "paragraph", "content": "x"}]) + _auth_client(client) + with client.websocket_connect(f"/ws/pages/{pid}") as ws: + ws.receive_json() # welcome + ws.receive_json() # sync + ws.send_json({"t": "op", "v": 0, "op": { + "type": "update", + "block": {"id": "a", "type": "paragraph", "content": "x!"}, + "base": {"id": "a", "type": "paragraph", "content": "x"}, + }}) + ack = ws.receive_json() + assert ack["t"] == "ack" + assert ack.get("merged")["content"] == "x!" + assert ack.get("conflict") is False + + +def test_ws_two_clients_same_block_merge_disjoint(client): + """Deux clients éditent le même bloc à des endroits différents → les deux + saisies survivent (au-delà du LWW) et convergeamment.""" + pid = _make_page(blocks=[{"id": "a", "type": "paragraph", "content": "hello world"}]) + _auth_client(client, 1, "tester") + with client.websocket_connect(f"/ws/pages/{pid}") as wa: + wa.receive_json() # welcome + wa.receive_json() # sync + _auth_client(client, 2, "other") + with client.websocket_connect(f"/ws/pages/{pid}") as wb: + wb.receive_json() # welcome + wb.receive_json() # sync + wa.receive_json() # peer_join + + # A édite la fin + wa.send_json({"t": "op", "v": 1, "op": { + "type": "update", + "block": {"id": "a", "type": "paragraph", "content": "hello world today"}, + "base": {"id": "a", "type": "paragraph", "content": "hello world"}, + }}) + ack_a = wa.receive_json() + assert ack_a["t"] == "ack" + op_b = wb.receive_json() + assert op_b["t"] == "op" + + # B édite le début, sur la même base + wb.send_json({"t": "op", "v": 1, "op": { + "type": "update", + "block": {"id": "a", "type": "paragraph", "content": "hello brave world"}, + "base": {"id": "a", "type": "paragraph", "content": "hello world"}, + }}) + ack_b = wb.receive_json() + assert ack_b["t"] == "ack" + merged = ack_b["merged"]["content"] + # les deux éditions disjointes doivent être conservées + assert "brave" in merged, merged + assert "today" in merged, merged + assert ack_b["conflict"] is False + + # A reçoit aussi le bloc fusionné + op_a2 = wa.receive_json() + assert op_a2["t"] == "op" + assert "brave" in op_a2["op"]["block"]["content"] + assert "today" in op_a2["op"]["block"]["content"] + + # persistance du contenu fusionné + from app.db import get_conn + with get_conn() as conn: + row = conn.execute("SELECT content FROM pages WHERE id=?", (pid,)).fetchone() + d = json.loads(row["content"]) + assert "brave" in d[0]["content"] and "today" in d[0]["content"] + + +def test_ws_update_without_base_stays_lww(client): + """Sans ``base`` (ancien client / protocole), le LWW historique s'applique.""" + pid = _make_page(blocks=[{"id": "a", "type": "paragraph", "content": "x"}]) + _auth_client(client, 1, "tester") + with client.websocket_connect(f"/ws/pages/{pid}") as wa: + wa.receive_json() + wa.receive_json() + _auth_client(client, 2, "other") + with client.websocket_connect(f"/ws/pages/{pid}") as wb: + wb.receive_json() + wb.receive_json() + wa.receive_json() # peer_join + wa.send_json({"t": "op", "v": 0, "op": {"type": "update", + "block": {"id": "a", "type": "paragraph", "content": "first"}}}) + wa.receive_json() # ack + b_op = wb.receive_json() + assert b_op["op"]["block"]["content"] == "first" + wb.send_json({"t": "op", "v": 1, "op": {"type": "update", + "block": {"id": "a", "type": "paragraph", "content": "second"}}}) + wb.receive_json() # ack (pas de merged sans base) + a_op = wa.receive_json() + assert a_op["op"]["block"]["content"] == "second" + from app.db import get_conn + with get_conn() as conn: + row = conn.execute("SELECT content FROM pages WHERE id=?", (pid,)).fetchone() + assert json.loads(row["content"])[0]["content"] == "second" + + +def test_ws_conflicting_scalar_flagged(client): + """Conflit sur un champ scalaire multi-états → drapeau conflict=True + LWW. + + Un booléen ne peut pas diverger depuis la même base (2 valeurs) ; on utilise + un ``status`` à 3 états pour que les deux côtés changent *différemment*. + """ + pid = _make_page(blocks=[{"id": "a", "type": "todo", "content": "t", "status": "todo"}]) + _auth_client(client, 1, "tester") + with client.websocket_connect(f"/ws/pages/{pid}") as wa: + wa.receive_json() + wa.receive_json() + _auth_client(client, 2, "other") + with client.websocket_connect(f"/ws/pages/{pid}") as wb: + wb.receive_json() + wb.receive_json() + wa.receive_json() + # A passe todo → doing + wa.send_json({"t": "op", "v": 1, "op": { + "type": "update", + "block": {"id": "a", "type": "todo", "content": "t", "status": "doing"}, + "base": {"id": "a", "type": "todo", "content": "t", "status": "todo"}, + }}) + wa.receive_json() # ack + wb.receive_json() # op broadcast + # B passe todo → done (conflit scalaire réel, même base) + wb.send_json({"t": "op", "v": 1, "op": { + "type": "update", + "block": {"id": "a", "type": "todo", "content": "t", "status": "done"}, + "base": {"id": "a", "type": "todo", "content": "t", "status": "todo"}, + }}) + ack_b = wb.receive_json() + assert ack_b["t"] == "ack" + assert ack_b["conflict"] is True + assert ack_b["merged"]["status"] == "done" # LWW : incoming gagne + assert "status" in str(ack_b.get("merged")) + + +def test_ws_stats_endpoint(client): + _auth_client(client) + pid = _make_page(blocks=[{"id": "a", "type": "paragraph", "content": "x"}]) + with client.websocket_connect(f"/ws/pages/{pid}") as ws: + ws.receive_json() + ws.receive_json() + ws.send_json({"t": "op", "v": 0, "op": { + "type": "update", + "block": {"id": "a", "type": "paragraph", "content": "y"}, + "base": {"id": "a", "type": "paragraph", "content": "x"}, + }}) + ws.receive_json() # ack + r = client.get("/api/realtime/stats") + assert r.status_code == 200 + d = r.json() + assert d["rooms"] == 1 + assert d["connections"] == 1 + assert d["ops"] >= 1 + assert d["merges"] >= 1 + assert isinstance(d["pages"], list) and d["pages"][0]["page_id"] == pid + + +def test_ws_stats_requires_auth(client): + r = client.get("/api/realtime/stats") + assert r.status_code == 200 + assert r.json() == {"error": "unauthorized"} + + +def test_ws_op_budget_anti_flood(client): + """Au-delà du budget d'ops, le serveur répond ack stale (sans appliquer).""" + from app.services import realtime_server as rs + pid = _make_page(blocks=[{"id": "a", "type": "paragraph", "content": "x"}]) + _auth_client(client) + old_max = rs.OP_WINDOW_MAX + rs.OP_WINDOW_MAX = 3 + try: + with client.websocket_connect(f"/ws/pages/{pid}") as ws: + ws.receive_json() + ws.receive_json() + for i in range(6): + ws.send_json({"t": "op", "v": 0, "op": {"type": "update", + "block": {"id": "a", "type": "paragraph", "content": f"v{i}"}}}) + acks = [ws.receive_json() for _ in range(6)] + stale = [a for a in acks if a.get("stale")] + assert stale, "budget anti-flood n'a pas été déclenché" + finally: + rs.OP_WINDOW_MAX = old_max + + +def test_room_state_missing_page_returns_empty(client): + import asyncio + + from app.services.realtime_server import manager + out = asyncio.run(manager.room_state(424242)) + assert out == {"blocks": [], "title": "", "version": 0} + assert 424242 not in manager._rooms + + +def test_writer_coalesces_cursor_updates(): + """La file sortante d'une connexion coalesce les curseurs (une seule position + finale conservée) tout en préservant l'ordre des messages importants.""" + import asyncio + + class FakeWS: + def __init__(self): + self.sent = [] + async def send_json(self, msg): + self.sent.append(msg) + + async def run(): + from app.services.realtime_server import RealtimeManager, RTConn + ws = FakeWS() + conn = RTConn(ws, {"id": 1, "login": "u"}, 1) + conn.writer = asyncio.create_task(RealtimeManager()._writer(conn)) + # empiler plusieurs curseurs + un message important au milieu + conn.out_q.put_nowait({"t": "sel", "block": "a", "offset": 1}) + conn.out_q.put_nowait({"t": "sel", "block": "a", "offset": 2}) + conn.out_q.put_nowait({"t": "title", "title": "T"}) + conn.out_q.put_nowait({"t": "sel", "block": "a", "offset": 3}) + await asyncio.sleep(0.05) + conn.closed = True + conn.out_q.put_nowait(None) + await asyncio.sleep(0.02) + if conn.writer and not conn.writer.done(): + conn.writer.cancel() + return ws.sent + + sent = asyncio.run(run()) + sels = [m for m in sent if m.get("t") == "sel"] + titles = [m for m in sent if m.get("t") == "title"] + # les curseurs sont coalescés : on ne garde pas les positions 1 et 2 séparées + assert len(sels) <= 2, f"curseurs non coalescés: {sels}" + assert titles and titles[0]["title"] == "T" + # l'offset final (3) doit être présent + assert sels[-1]["offset"] == 3, sels