colaborative tích hợp redis để đồng bộ toàn bộ annotation

backend/app/redis_client.py

@@ -0,0 +1,75 @@
+"""
+Redis helper — returns a connected client or None when Redis is unavailable.
+
+Key namespace design (prevents cross-file / cross-user data leakage):
+
+    ann:{pdf_id}:{page_number}:{user_id}
+
+Every segment is mandatory, so:
+  - User A opening file 1 never touches User A's data on file 2
+  - User A opening file 1 never touches User B's data on file 1
+  - Data for page 3 never affects page 7
+
+TTL: 24 h — temp entry expires automatically if the user never returns.
+On permanent save the entry is deleted immediately.
+"""
+
+import logging
+import os
+from typing import Optional
+
+import redis as redis_lib
+
+logger = logging.getLogger(__name__)
+
+_client: Optional[redis_lib.Redis] = None
+_warned = False  # log the "unavailable" warning only once
+
+
+def get_redis() -> Optional[redis_lib.Redis]:
+    """Return a live Redis client, or None if Redis is unreachable."""
+    global _client, _warned
+
+    if _client is not None:
+        try:
+            _client.ping()
+            return _client
+        except Exception:
+            _client = None  # connection dropped — try to reconnect below
+
+    url = os.getenv("REDIS_URL", "redis://localhost:6379/0")
+    try:
+        r: redis_lib.Redis = redis_lib.from_url(
+            url,
+            decode_responses=True,
+            socket_connect_timeout=2,
+            socket_timeout=2,
+        )
+        r.ping()
+        _client = r
+        _warned = False
+        logger.info("Redis connected: %s", url)
+        return _client
+    except Exception as exc:
+        if not _warned:
+            logger.warning(
+                "Redis unavailable (%s) — temp annotation cache disabled. "
+                "Set REDIS_URL or start a local Redis instance to enable it.",
+                exc,
+            )
+            _warned = True
+        return None
+
+
+# Key helpers — centralised so there's one place to change the format.
+ANN_TTL = 86_400  # 24 hours
+
+
+def ann_temp_key(pdf_id: int, page_number: int, user_id: int) -> str:
+    """Fully-namespaced Redis key for one user's unsaved canvas on one PDF page."""
+    return f"ann:{pdf_id}:{page_number}:{user_id}"
+
+
+def ann_temp_page_pattern(pdf_id: int, page_number: int) -> str:
+    """Glob pattern to list all users' temp entries for a given page."""
+    return f"ann:{pdf_id}:{page_number}:*"

backend/app/routers/annotations.py

@@ -1,11 +1,13 @@
 from datetime import datetime, timezone
+import json
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, status as http_status
 from sqlalchemy.orm import Session
 
 from ..database import get_db
 from ..dependencies import get_current_user
 from ..models import Annotation, PDF, User
+from ..redis_client import ANN_TTL, ann_temp_key, ann_temp_page_pattern, get_redis
 from ..schemas import AnnotationIn, AnnotationOut
 
 router = APIRouter(prefix="/annotations", tags=["annotations"])
@@ -82,7 +84,35 @@ def get_all_annotations(
         .all()
     )
 
-    if not rows:
+    # Build user_id → canvas_data from DB (permanent saves)
+    # Key = pdf_id + page_number + user_id → completely isolated per file/page/user
+    user_data: dict[int, dict] = {}
+    for row in rows:
+        user_data[row.user_id] = dict(row.canvas_data) if row.canvas_data else {}
+
+    # Overlay with Redis temp data — unsaved strokes written by path:created / clear.
+    # If a user has BOTH a DB record and a temp entry, Redis wins (more recent).
+    r = get_redis()
+    if r is not None:
+        try:
+            pattern = ann_temp_page_pattern(pdf_id, page_number)
+            temp_keys: list[str] = r.keys(pattern)
+            for key in temp_keys:
+                # key format: ann:{pdf_id}:{page_number}:{user_id}
+                try:
+                    uid = int(key.split(":")[-1])
+                except ValueError:
+                    continue
+                raw = r.get(key)
+                if raw:
+                    try:
+                        user_data[uid] = json.loads(raw)
+                    except Exception:
+                        pass
+        except Exception:
+            pass  # Redis hiccup — fall through with DB data only
+
+    if not user_data:
         return AnnotationOut(
             id=0,
             pdf_id=pdf_id,
@@ -91,28 +121,50 @@ def get_all_annotations(
             updated_at=datetime.now(timezone.utc),
         )
 
-    # Merge all objects arrays; use canvas metadata (background etc.) from latest record
-    latest = max(rows, key=lambda r: r.updated_at)
-    base: dict = dict(latest.canvas_data) if latest.canvas_data else {}
+    # Canvas-level metadata (version, background…) comes from the latest DB record
+    latest = max(rows, key=lambda row: row.updated_at) if rows else None
+    base: dict = dict(latest.canvas_data) if latest and latest.canvas_data else {}
+
     merged_objects: list = []
-    for row in rows:
-        for obj in (row.canvas_data or {}).get("objects", []):
-            # Tag each object with its owner so the frontend can avoid re-saving
-            # other users' objects under the current user's record (prevents duplicates).
+    for uid, canvas in user_data.items():
+        for obj in (canvas or {}).get("objects", []):
             obj_copy = dict(obj)
-            obj_copy["_owner_id"] = row.user_id
+            obj_copy["_owner_id"] = uid
             merged_objects.append(obj_copy)
     base["objects"] = merged_objects
 
     return AnnotationOut(
-        id=latest.id,
+        id=latest.id if latest else 0,
         pdf_id=pdf_id,
         page_number=page_number,
         canvas_data=base,
-        updated_at=latest.updated_at,
+        updated_at=latest.updated_at if latest else datetime.now(timezone.utc),
     )
 
 
+# ── PUT /api/annotations/{pdf_id}/{page_number}/temp ──────────────────────────
+# Writes unsaved canvas state to Redis so every subsequent /all call includes it.
+# Called automatically (debounced) by the frontend after each stroke.
+
+@router.put("/{pdf_id}/{page_number}/temp", status_code=http_status.HTTP_204_NO_CONTENT)
+def upsert_temp_annotation(
+    pdf_id: int,
+    page_number: int,
+    body: AnnotationIn,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+):
+    _any_pdf_or_404(db, pdf_id)
+    r = get_redis()
+    if r is None:
+        return  # Redis unavailable — silently skip
+    key = ann_temp_key(pdf_id, page_number, current_user.id)
+    try:
+        r.setex(key, ANN_TTL, json.dumps(body.canvas_data))
+    except Exception:
+        pass  # non-critical
+
+
 # ── PUT /api/annotations/{pdf_id}/{page_number} ───────────────────────────────
 
 @router.put("/{pdf_id}/{page_number}", response_model=AnnotationOut)
@@ -149,4 +201,13 @@ def upsert_annotation(
 
     db.commit()
     db.refresh(ann)
+
+    # Data is now in DB — remove the Redis temp entry so /all doesn't double-count
+    r = get_redis()
+    if r is not None:
+        try:
+            r.delete(ann_temp_key(pdf_id, page_number, current_user.id))
+        except Exception:
+            pass
+
     return ann

backend/app/routers/ws.py

@@ -161,7 +161,7 @@ async def pdf_collaboration(websocket: WebSocket, pdf_id: int):
                 continue
 
             # Inject sender identity and forward to all other clients
-            if msg_type in {"object_add", "object_remove", "clear", "cursor"}:
+            if msg_type in {"object_add", "object_remove", "clear", "cursor", "color_sync"}:
                 msg.update(user_info)
                 await room.broadcast(msg, exclude=websocket)