redis

config.py

@@ -11,6 +11,7 @@ SUPABASE_ANON_KEY: str = os.getenv("SUPABASE_ANON_KEY", "")
 GEMINI_API_KEY: str = os.getenv("GEMINI_API_KEY", "")
 FRONTEND_URL: str = os.getenv("FRONTEND_URL", "http://localhost:3000")
 JWT_SECRET: str = os.getenv("JWT_SECRET", "nomoosh-secret-change-in-production")
+REDIS_URL: str = os.getenv("REDIS_URL", "")  # e.g. rediss://default:xxx@xxx.upstash.io:6379
 
 # Storage bucket name in Supabase
 STORAGE_BUCKET: str = "restaurant-media"

redis_client.py

@@ -0,0 +1,101 @@
+"""Redis-backed cart state machine.
+
+Uses Redis Hash Maps for O(1) atomic cart operations.
+HINCRBY ensures perfect concurrency — no race conditions.
+
+Keys:
+  cart:{session_id}       → Hash  { "item_id": quantity, ... }
+  cart:{session_id}:ver   → String  auto-incrementing version
+
+All keys auto-expire after 4 hours (14400s) for cleanup.
+"""
+
+from __future__ import annotations
+import logging
+import redis.asyncio as aioredis
+from config import REDIS_URL
+
+logger = logging.getLogger(__name__)
+
+_pool: aioredis.Redis | None = None
+
+
+async def get_redis() -> aioredis.Redis:
+    global _pool
+    if _pool is None:
+        if not REDIS_URL:
+            raise RuntimeError(
+                "REDIS_URL must be set in .env  "
+                "→ Get free Redis at https://upstash.com"
+            )
+        _pool = aioredis.from_url(
+            REDIS_URL,
+            decode_responses=True,
+            socket_connect_timeout=5,
+            retry_on_timeout=True,
+        )
+    return _pool
+
+
+# ── Atomic Cart Operations ────────────────────────────────
+
+_CART_TTL = 14400  # 4 hours
+
+
+async def cart_incr(session_id: str, item_id: int, delta: int = 1) -> int:
+    """Atomically increment item quantity via HINCRBY.
+    If result ≤ 0, removes the field. Returns new quantity."""
+    r = await get_redis()
+    key = f"cart:{session_id}"
+    new_qty = await r.hincrby(key, str(item_id), delta)
+    if new_qty <= 0:
+        await r.hdel(key, str(item_id))
+        new_qty = 0
+    await r.expire(key, _CART_TTL)
+    return new_qty
+
+
+async def cart_set_qty(session_id: str, item_id: int, quantity: int):
+    """Set exact quantity for an item (for explicit qty changes)."""
+    r = await get_redis()
+    key = f"cart:{session_id}"
+    if quantity <= 0:
+        await r.hdel(key, str(item_id))
+    else:
+        await r.hset(key, str(item_id), quantity)
+        await r.expire(key, _CART_TTL)
+
+
+async def cart_remove_item(session_id: str, item_id: int):
+    """Remove item entirely from cart."""
+    r = await get_redis()
+    await r.hdel(f"cart:{session_id}", str(item_id))
+
+
+async def cart_get_all(session_id: str) -> dict[int, int]:
+    """HGETALL — returns {item_id: quantity} for all items."""
+    r = await get_redis()
+    raw = await r.hgetall(f"cart:{session_id}")
+    return {int(k): int(v) for k, v in raw.items() if int(v) > 0}
+
+
+async def cart_bump_version(session_id: str) -> int:
+    """Atomically increment and return cart version."""
+    r = await get_redis()
+    ver_key = f"cart:{session_id}:ver"
+    ver = await r.incr(ver_key)
+    await r.expire(ver_key, _CART_TTL)
+    return ver
+
+
+async def cart_get_version(session_id: str) -> int:
+    """Get current cart version without bumping."""
+    r = await get_redis()
+    val = await r.get(f"cart:{session_id}:ver")
+    return int(val) if val else 0
+
+
+async def cart_clear(session_id: str):
+    """Delete cart and version (after order confirmed)."""
+    r = await get_redis()
+    await r.delete(f"cart:{session_id}", f"cart:{session_id}:ver")

requirements.txt

@@ -9,5 +9,6 @@ Pillow==11.1.0
 PyMuPDF==1.25.3
 pydantic>=2.0
 PyJWT>=2.10.1
+redis>=5.0.0
 passlib[bcrypt]==1.7.4
 bcrypt==4.1.2

routers/orders.py

@@ -1,13 +1,13 @@
-"""Order / Cart / Payment / Chef / Menu-management endpoints.
+"""Order / Payment / Chef / Menu-management endpoints.
+
+Cart lives in Redis (see ws.py). Cart mutations flow through WebSocket.
 
 Endpoints:
-  POST /cart/add                          → Add item to shared cart
-  POST /cart/remove                       → Remove cart item
-  GET  /cart/{session_id}                 → Get enriched cart
   POST /sessions/join/{qr_token}          → Join table session (guest)
   GET  /sessions/{session_id}/status      → Session status (lock, ETA)
   POST /payment/lock                      → Lock cart for payment
-  POST /payment/confirm                   → Confirm payment → create order
+  POST /payment/unlock                    → Unlock cart
+  POST /payment/confirm                   → Confirm payment → create order (reads Redis cart)
   GET  /orders/restaurant/{restaurant_id} → Chef: all orders
   POST /orders/{order_id}/eta             → Chef: set ETA
   POST /menu/create                       → Owner: add dish
@@ -25,6 +25,7 @@ from pydantic import BaseModel
 from supabase_client import get_supabase
 from routers.staff import get_staff_from_token
 from ws_manager import manager as ws_manager
+from redis_client import cart_get_all, cart_clear
 
 router = APIRouter(tags=["orders"])
 
@@ -34,12 +35,20 @@ _payment_timers: dict[str, asyncio.Task] = {}
 _payment_lock_owners: dict[str, str] = {}
 
 
+def invalidate_menu_cache(restaurant_id: int):
+    """Bust the ws.py menu cache when menu CRUD happens."""
+    try:
+        from routers.ws import invalidate_menu_cache as _inv
+        _inv(restaurant_id)
+    except Exception:
+        pass
+
+
 def _safe_session_update(sb, session_id: str, fields: dict):
     """Update session, retrying without new columns if they don't exist yet."""
     try:
         sb.table("sessions").update(fields).eq("id", session_id).execute()
     except Exception:
-        # Fallback: strip columns that may not exist in DB yet
         safe = {k: v for k, v in fields.items() if k in ("status", "payment_lock")}
         if safe:
             sb.table("sessions").update(safe).eq("id", session_id).execute()
@@ -49,25 +58,6 @@ def _safe_session_update(sb, session_id: str, fields: dict):
 # MODELS
 # ═══════════════════════════════════════════════════════════
 
-class AddToCartRequest(BaseModel):
-    session_id: str
-    menu_item_id: int
-    quantity: int = 1
-    participant_id: str | None = None
-    notes: str = ""
-
-
-class RemoveFromCartRequest(BaseModel):
-    session_id: str
-    cart_item_id: str
-
-
-class UpdateCartQtyRequest(BaseModel):
-    session_id: str
-    cart_item_id: str
-    quantity: int
-
-
 class PaymentLockRequest(BaseModel):
     session_id: str
     participant_id: str
@@ -104,227 +94,6 @@ class MenuItemUpdate(BaseModel):
     category_veg: bool | None = None
 
 
-class CartBatchOp(BaseModel):
-    menu_item_id: int
-    delta: int  # +1 to add, -N to remove N
-
-class CartBatchRequest(BaseModel):
-    session_id: str
-    participant_id: str | None = None
-    operations: list[CartBatchOp]
-
-
-# ═══════════════════════════════════════════════════════════
-# IN-MEMORY MENU CACHE (avoids re-fetching on every cart op)
-# ═══════════════════════════════════════════════════════════
-import time as _time
-
-_menu_cache: dict[int, dict] = {}      # restaurant_id → {data, ts}
-_MENU_CACHE_TTL = 120                    # seconds
-
-def _get_menu_map(sb, restaurant_id: int) -> dict[int, dict]:
-    """Return menu items indexed by id, cached for 120s."""
-    cached = _menu_cache.get(restaurant_id)
-    if cached and (_time.time() - cached["ts"]) < _MENU_CACHE_TTL:
-        return cached["data"]
-    rows = sb.table("menu").select("id, dish_name, price, category, image_link, variant_name").eq("restaurant_id", restaurant_id).execute()
-    menu_map = {m["id"]: m for m in (rows.data or [])}
-    _menu_cache[restaurant_id] = {"data": menu_map, "ts": _time.time()}
-    return menu_map
-
-def invalidate_menu_cache(restaurant_id: int):
-    """Call after menu CRUD to bust cache."""
-    _menu_cache.pop(restaurant_id, None)
-
-
-# ═══════════════════════════════════════════════════════════
-# CART
-# ═══════════════════════════════════════════════════════════
-
-def _enrich_cart(sb, session_id: str, restaurant_id: int | None = None) -> dict:
-    """Return enriched cart data for a session. Uses menu cache if restaurant_id is provided."""
-    cart = sb.table("carts").select("id, version").eq("session_id", session_id).execute()
-    if not cart.data:
-        return {"items": [], "version": 0, "total": 0}
-
-    cart_id = cart.data[0]["id"]
-    items = (
-        sb.table("cart_items")
-        .select("id, menu_item_id, quantity, added_by, notes, created_at")
-        .eq("cart_id", cart_id)
-        .order("created_at")
-        .execute()
-    )
-    if not items.data:
-        return {"items": [], "version": cart.data[0]["version"], "total": 0}
-
-    # Use cache if restaurant_id provided, else fetch by ids
-    if restaurant_id:
-        menu_map = _get_menu_map(sb, restaurant_id)
-    else:
-        menu_ids = list({i["menu_item_id"] for i in items.data})
-        menu_rows = sb.table("menu").select("id, dish_name, price, category, image_link, variant_name").in_("id", menu_ids).execute()
-        menu_map = {m["id"]: m for m in (menu_rows.data or [])}
-
-    enriched = []
-    for item in items.data:
-        m = menu_map.get(item["menu_item_id"])
-        if m:
-            enriched.append({
-                **item,
-                "dish_name": m["dish_name"],
-                "price": m["price"],
-                "category": m.get("category"),
-                "image_link": m.get("image_link"),
-                "variant_name": m.get("variant_name", "Regular"),
-            })
-
-    total = sum(i["price"] * i["quantity"] for i in enriched)
-    return {"items": enriched, "version": cart.data[0]["version"], "total": total}
-
-
-@router.post("/cart/add")
-async def add_to_cart(data: AddToCartRequest):
-    sb = get_supabase()
-
-    # Check payment lock
-    session = sb.table("sessions").select("payment_lock").eq("id", data.session_id).execute()
-    if session.data and session.data[0].get("payment_lock"):
-        raise HTTPException(status_code=423, detail="Cart is locked — payment in progress")
-
-    cart = sb.table("carts").select("id, version").eq("session_id", data.session_id).execute()
-    if not cart.data:
-        raise HTTPException(status_code=404, detail="Cart not found for this session")
-
-    cart_id = cart.data[0]["id"]
-
-    # Upsert: if same item already in cart, bump quantity
-    existing = (
-        sb.table("cart_items")
-        .select("id, quantity")
-        .eq("cart_id", cart_id)
-        .eq("menu_item_id", data.menu_item_id)
-        .execute()
-    )
-    if existing.data:
-        new_qty = existing.data[0]["quantity"] + data.quantity
-        sb.table("cart_items").update({"quantity": new_qty}).eq("id", existing.data[0]["id"]).execute()
-    else:
-        sb.table("cart_items").insert({
-            "cart_id": cart_id,
-            "menu_item_id": data.menu_item_id,
-            "quantity": data.quantity,
-            "added_by": data.participant_id,
-            "notes": data.notes,
-        }).execute()
-
-    # Bump version
-    new_ver = cart.data[0]["version"] + 1
-    sb.table("carts").update({"version": new_ver, "updated_at": datetime.now(timezone.utc).isoformat()}).eq("id", cart_id).execute()
-
-    # Enrich cart once
-    result = _enrich_cart(sb, data.session_id)
-
-    # Broadcast in background (non-blocking)
-    asyncio.create_task(ws_manager.broadcast(data.session_id, {"type": "cart_update", "cart": result}))
-
-    return result
-
-
-@router.post("/cart/batch")
-async def batch_cart(data: CartBatchRequest):
-    """Process multiple cart add/remove operations in a single request.
-    Each operation has menu_item_id and delta (+N to add, -N to remove).
-    This eliminates per-item round trips for rapid tapping."""
-    sb = get_supabase()
-
-    # 1. Check payment lock + get cart (single query each)
-    session = sb.table("sessions").select("payment_lock, restaurant_id").eq("id", data.session_id).execute()
-    if not session.data:
-        raise HTTPException(status_code=404, detail="Session not found")
-    if session.data[0].get("payment_lock"):
-        raise HTTPException(status_code=423, detail="Cart is locked — payment in progress")
-    restaurant_id = session.data[0]["restaurant_id"]
-
-    cart = sb.table("carts").select("id, version").eq("session_id", data.session_id).execute()
-    if not cart.data:
-        raise HTTPException(status_code=404, detail="Cart not found")
-    cart_id = cart.data[0]["id"]
-
-    # 2. Fetch ALL current cart items in one query
-    existing_items = sb.table("cart_items").select("id, menu_item_id, quantity").eq("cart_id", cart_id).execute()
-    existing_map: dict[int, dict] = {}
-    for it in (existing_items.data or []):
-        existing_map[it["menu_item_id"]] = it
-
-    # 3. Process all operations (minimal DB writes)
-    for op in data.operations:
-        cur = existing_map.get(op.menu_item_id)
-        if cur:
-            new_qty = cur["quantity"] + op.delta
-            if new_qty <= 0:
-                sb.table("cart_items").delete().eq("id", cur["id"]).execute()
-                del existing_map[op.menu_item_id]
-            else:
-                sb.table("cart_items").update({"quantity": new_qty}).eq("id", cur["id"]).execute()
-                cur["quantity"] = new_qty
-        elif op.delta > 0:
-            sb.table("cart_items").insert({
-                "cart_id": cart_id,
-                "menu_item_id": op.menu_item_id,
-                "quantity": op.delta,
-                "added_by": data.participant_id,
-            }).execute()
-
-    # 4. Bump version ONCE for entire batch
-    new_ver = cart.data[0]["version"] + 1
-    sb.table("carts").update({"version": new_ver, "updated_at": datetime.now(timezone.utc).isoformat()}).eq("id", cart_id).execute()
-
-    # 5. Enrich with cached menu data
-    result = _enrich_cart(sb, data.session_id, restaurant_id)
-
-    # 6. Broadcast in background
-    asyncio.create_task(ws_manager.broadcast(data.session_id, {"type": "cart_update", "cart": result}))
-
-    return result
-
-
-@router.post("/cart/remove")
-async def remove_from_cart(data: RemoveFromCartRequest):
-    sb = get_supabase()
-    session = sb.table("sessions").select("payment_lock").eq("id", data.session_id).execute()
-    if session.data and session.data[0].get("payment_lock"):
-        raise HTTPException(status_code=423, detail="Cart is locked — payment in progress")
-
-    sb.table("cart_items").delete().eq("id", data.cart_item_id).execute()
-    result = _enrich_cart(sb, data.session_id)
-    asyncio.create_task(ws_manager.broadcast(data.session_id, {"type": "cart_update", "cart": result}))
-    return result
-
-
-@router.post("/cart/update-quantity")
-async def update_cart_quantity(data: UpdateCartQtyRequest):
-    sb = get_supabase()
-    session = sb.table("sessions").select("payment_lock").eq("id", data.session_id).execute()
-    if session.data and session.data[0].get("payment_lock"):
-        raise HTTPException(status_code=423, detail="Cart is locked — payment in progress")
-
-    if data.quantity <= 0:
-        sb.table("cart_items").delete().eq("id", data.cart_item_id).execute()
-    else:
-        sb.table("cart_items").update({"quantity": data.quantity}).eq("id", data.cart_item_id).execute()
-
-    result = _enrich_cart(sb, data.session_id)
-    asyncio.create_task(ws_manager.broadcast(data.session_id, {"type": "cart_update", "cart": result}))
-    return result
-
-
-@router.get("/cart/{session_id}")
-async def get_cart(session_id: str):
-    sb = get_supabase()
-    return _enrich_cart(sb, session_id)
-
-
 # ═══════════════════════════════════════════════════════════
 # SESSION JOIN (customer scans QR)
 # ═══════════════════════════════════════════════════════════
@@ -486,29 +255,24 @@ async def confirm_payment(data: PaymentConfirmRequest):
         raise HTTPException(status_code=404, detail="Session not found")
     s = session.data[0]
 
-    # Get cart
-    cart = sb.table("carts").select("id").eq("session_id", data.session_id).execute()
-    if not cart.data:
-        raise HTTPException(status_code=404, detail="Cart not found")
-
-    cart_items = sb.table("cart_items").select("*").eq("cart_id", cart.data[0]["id"]).execute()
-    if not cart_items.data:
+    # ── Read cart from Redis ──────────────────────────────
+    raw_cart = await cart_get_all(data.session_id)
+    if not raw_cart:
         raise HTTPException(status_code=400, detail="Cart is empty")
 
-    # Batch-fetch all menu data at once
-    menu_ids = list({ci["menu_item_id"] for ci in cart_items.data})
-    menu_rows = sb.table("menu").select("id, price, dish_name, category, variant_name").in_("id", menu_ids).execute()
-    menu_map = {m["id"]: m for m in (menu_rows.data or [])}
+    # Enrich with menu data
+    from routers.ws import _get_menu_map
+    menu_map = _get_menu_map(sb, s["restaurant_id"])
 
     total = 0
     order_items_data = []
-    for ci in cart_items.data:
-        m = menu_map.get(ci["menu_item_id"], {})
+    for item_id, qty in raw_cart.items():
+        m = menu_map.get(item_id, {})
         price = m.get("price", 0)
-        total += price * ci["quantity"]
+        total += price * qty
         order_items_data.append({
-            "menu_item_id": ci["menu_item_id"],
-            "quantity": ci["quantity"],
+            "menu_item_id": item_id,
+            "quantity": qty,
             "price_at_time": price,
             "dish_name": m.get("dish_name", "Unknown"),
             "category": m.get("category"),
@@ -555,6 +319,9 @@ async def confirm_payment(data: PaymentConfirmRequest):
     })
     _payment_lock_owners.pop(data.session_id, None)
 
+    # Clear Redis cart
+    await cart_clear(data.session_id)
+
     # Table → dirty
     sb.table("restaurant_tables").update({"status": "dirty"}).eq("id", s["table_id"]).execute()
 
@@ -774,4 +541,4 @@ async def delete_menu_item(item_id: int, authorization: str = Header(None)):
     sb = get_supabase()
     sb.table("menu").delete().eq("id", item_id).execute()
     invalidate_menu_cache(payload["restaurant_id"])
-    return {"message": "Deleted"}
+    return {"message": "Deleted"}

routers/ws.py

@@ -1,76 +1,95 @@
-"""WebSocket endpoints — WhatsApp-style push architecture.
+"""WebSocket endpoints — Redis-backed cart state machine.
 
 Three channels:
-  /ws/{session_id}              → Customer session (cart, payment, ETA)
+  /ws/{session_id}              → Customer session (cart via Redis, payment, ETA)
   /ws/staff/{restaurant_id}     → Staff dashboards (tables, orders)
   /ws/table/{qr_token}          → Waiting customers (table activation)
 
-On connect, server pushes FULL current state immediately.
-All mutations broadcast incremental updates — zero polling needed.
+Cart mutations flow through WebSocket → Redis HINCRBY (atomic) → broadcast.
+On connect, server pushes FULL current state from Redis immediately.
 """
 
 from __future__ import annotations
-import json, logging
+import json, logging, time as _time
 from fastapi import APIRouter, WebSocket, WebSocketDisconnect
 from ws_manager import manager
 from supabase_client import get_supabase
+from redis_client import (
+    cart_incr, cart_set_qty, cart_remove_item,
+    cart_get_all, cart_bump_version, cart_get_version, cart_clear,
+)
 
 router = APIRouter()
 logger = logging.getLogger(__name__)
 
-# Import in-memory lock owners from orders module (lazy to avoid circular)
-def _get_lock_owner(session_id: str) -> str | None:
-    try:
-        from routers.orders import _payment_lock_owners
-        return _payment_lock_owners.get(session_id)
-    except Exception:
-        return None
 
+# ── Menu cache (shared enrichment) ─────────────────────────
 
-# ── Shared helpers: build state payloads ───────────────────
+_menu_cache: dict[int, dict] = {}
+_MENU_CACHE_TTL = 120
 
-def _get_cart_state(session_id: str) -> dict:
-    """Full enriched cart for a session."""
-    sb = get_supabase()
-    cart = sb.table("carts").select("id, version").eq("session_id", session_id).execute()
-    if not cart.data:
-        return {"items": [], "version": 0, "total": 0}
-
-    cart_id = cart.data[0]["id"]
-    items = (
-        sb.table("cart_items")
-        .select("id, menu_item_id, quantity, added_by, notes, created_at")
-        .eq("cart_id", cart_id)
-        .order("created_at")
-        .execute()
-    )
-    if not items.data:
-        return {"items": [], "version": cart.data[0]["version"], "total": 0}
 
-    menu_ids = list({i["menu_item_id"] for i in items.data})
-    menu_rows = (
+def _get_menu_map(sb, restaurant_id: int) -> dict[int, dict]:
+    """Menu items indexed by id, cached in-memory for 120s."""
+    cached = _menu_cache.get(restaurant_id)
+    if cached and (_time.time() - cached["ts"]) < _MENU_CACHE_TTL:
+        return cached["data"]
+    rows = (
         sb.table("menu")
         .select("id, dish_name, price, category, image_link, variant_name")
-        .in_("id", menu_ids)
+        .eq("restaurant_id", restaurant_id)
         .execute()
     )
-    menu_map = {m["id"]: m for m in (menu_rows.data or [])}
-
-    enriched = []
-    for item in items.data:
-        m = menu_map.get(item["menu_item_id"])
-        if m:
-            enriched.append({
-                **item,
-                "dish_name": m["dish_name"],
-                "price": m["price"],
-                "category": m.get("category"),
-                "image_link": m.get("image_link"),
-                "variant_name": m.get("variant_name", "Regular"),
-            })
+    menu_map = {m["id"]: m for m in (rows.data or [])}
+    _menu_cache[restaurant_id] = {"data": menu_map, "ts": _time.time()}
+    return menu_map
+
+
+def invalidate_menu_cache(restaurant_id: int):
+    _menu_cache.pop(restaurant_id, None)
+
+
+# ── Cart enrichment from Redis ─────────────────────────────
+
+async def get_enriched_cart(session_id: str, restaurant_id: int) -> dict:
+    """Read cart from Redis, enrich with cached menu data."""
+    raw = await cart_get_all(session_id)
+    ver = await cart_get_version(session_id)
+
+    if not raw:
+        return {"items": [], "total": 0, "version": ver}
+
+    sb = get_supabase()
+    menu_map = _get_menu_map(sb, restaurant_id)
 
-    total = sum(i["price"] * i["quantity"] for i in enriched)
-    return {"items": enriched, "version": cart.data[0]["version"], "total": total}
+    items = []
+    total = 0
+    for item_id, qty in raw.items():
+        m = menu_map.get(item_id, {})
+        price = m.get("price", 0)
+        items.append({
+            "menu_item_id": item_id,
+            "quantity": qty,
+            "dish_name": m.get("dish_name", "Unknown"),
+            "price": price,
+            "category": m.get("category"),
+            "image_link": m.get("image_link"),
+            "variant_name": m.get("variant_name", "Regular"),
+        })
+        total += price * qty
+
+    return {"items": items, "total": total, "version": ver}
+
+
+# ── Helpers: build state payloads ──────────────────────────
+
+# Import in-memory lock owners from orders module (lazy to avoid circular)
+def _get_lock_owner(session_id: str) -> str | None:
+    try:
+        from routers.orders import _payment_lock_owners
+        return _payment_lock_owners.get(session_id)
+    except Exception:
+        return None
 
 
 def _get_session_state(session_id: str) -> dict:
@@ -179,15 +198,28 @@ def _get_staff_state(restaurant_id: int) -> dict:
 
 @router.websocket("/ws/{session_id}")
 async def session_ws(websocket: WebSocket, session_id: str):
-    """Customer WS — pushes cart/payment/ETA updates instantly."""
+    """Customer WS — cart mutations go through here → Redis → broadcast."""
     await manager.connect(session_id, websocket)
     logger.info(f"[WS] Customer connected: session={session_id[:8]}")
 
-    # Push full state immediately on connect (like WhatsApp message sync)
+    # Resolve restaurant_id for menu enrichment
+    restaurant_id: int | None = None
+    try:
+        sb = get_supabase()
+        sess = sb.table("sessions").select("restaurant_id").eq("id", session_id).execute()
+        if sess.data:
+            restaurant_id = sess.data[0]["restaurant_id"]
+    except Exception:
+        pass
+
+    # Push full state immediately on connect
     try:
-        cart = _get_cart_state(session_id)
-        session = _get_session_state(session_id)
-        await websocket.send_json({"type": "init", "cart": cart, "session": session})
+        if restaurant_id:
+            cart_state = await get_enriched_cart(session_id, restaurant_id)
+        else:
+            cart_state = {"items": [], "total": 0, "version": 0}
+        session_state = _get_session_state(session_id)
+        await websocket.send_json({"type": "init", "cart": cart_state, "session": session_state})
     except Exception as e:
         logger.error(f"[WS] Init push failed {session_id[:8]}: {e}")
 
@@ -196,15 +228,53 @@ async def session_ws(websocket: WebSocket, session_id: str):
             data = await websocket.receive_text()
             try:
                 msg = json.loads(data)
-                if msg.get("type") == "ping":
+                msg_type = msg.get("type", "")
+
+                if msg_type == "ping":
                     await websocket.send_json({"type": "pong"})
-                elif msg.get("type") == "sync":
-                    # Client requests full re-sync (e.g. after wake from sleep)
-                    cart = _get_cart_state(session_id)
-                    session = _get_session_state(session_id)
-                    await websocket.send_json({"type": "init", "cart": cart, "session": session})
+
+                elif msg_type == "sync":
+                    if restaurant_id:
+                        cart_state = await get_enriched_cart(session_id, restaurant_id)
+                    else:
+                        cart_state = {"items": [], "total": 0, "version": 0}
+                    session_state = _get_session_state(session_id)
+                    await websocket.send_json({"type": "init", "cart": cart_state, "session": session_state})
+
+                # ── Cart mutations via Redis (atomic) ─────
+                elif msg_type == "cart_add":
+                    item_id = int(msg["item_id"])
+                    delta = int(msg.get("delta", 1))
+                    await cart_incr(session_id, item_id, delta)
+                    await cart_bump_version(session_id)
+                    if restaurant_id:
+                        enriched = await get_enriched_cart(session_id, restaurant_id)
+                        await manager.broadcast(session_id, {"type": "cart_update", "cart": enriched})
+
+                elif msg_type == "cart_remove":
+                    item_id = int(msg["item_id"])
+                    await cart_remove_item(session_id, item_id)
+                    await cart_bump_version(session_id)
+                    if restaurant_id:
+                        enriched = await get_enriched_cart(session_id, restaurant_id)
+                        await manager.broadcast(session_id, {"type": "cart_update", "cart": enriched})
+
+                elif msg_type == "cart_set_qty":
+                    item_id = int(msg["item_id"])
+                    qty = int(msg["quantity"])
+                    if qty <= 0:
+                        await cart_remove_item(session_id, item_id)
+                    else:
+                        await cart_set_qty(session_id, item_id, qty)
+                    await cart_bump_version(session_id)
+                    if restaurant_id:
+                        enriched = await get_enriched_cart(session_id, restaurant_id)
+                        await manager.broadcast(session_id, {"type": "cart_update", "cart": enriched})
+
             except json.JSONDecodeError:
                 logger.warning(f"[WS] Bad JSON from {session_id[:8]}")
+            except Exception as e:
+                logger.error(f"[WS] Cart action error {session_id[:8]}: {e}")
     except WebSocketDisconnect:
         logger.info(f"[WS] Customer disconnected: session={session_id[:8]}")
     except Exception as e: