feat(messages): end-to-end encrypted real-time direct messaging backend

Private user-to-user messages are end-to-end encrypted: the server only ever
stores ciphertext + iv and acts as a key directory (public keys) plus an opaque
store for each user's private key, wrapped client-side with a passphrase the
server never sees — so not even an admin can read a conversation. A separate
kind=system message (plaintext, no sender) powers a server-authored Siftlode
welcome shown on first open, reusable later for announcements.

- models: rework Message (kind, nullable sender/body, ciphertext+iv) + MessageKey;
  migrations 0026 (table) + 0027 (E2EE rework).
- routes/messages.py: key directory/blob endpoints, ciphertext send, conversations
  + threads (system + user), lazy welcome, all gated by require_human.
- realtime.py: in-process WebSocket connection registry; /ws delivers sent
  messages to a user's open tabs instantly (sync-callable push, single-process).

backend/alembic/versions/0026_messages.py

@@ -0,0 +1,56 @@
+"""user-to-user messages
+
+Revision ID: 0026_messages
+Revises: 0025_install_wizard
+Create Date: 2026-06-25
+
+Adds the `messages` table for direct user-to-user messaging (notification module phase 2).
+A conversation is the set of rows between two users; both FKs cascade so a GDPR account
+erasure removes that user's messages.
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0026_messages"
+down_revision: Union[str, None] = "0025_install_wizard"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "messages",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column(
+            "sender_id",
+            sa.Integer(),
+            sa.ForeignKey("users.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column(
+            "recipient_id",
+            sa.Integer(),
+            sa.ForeignKey("users.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("body", sa.Text(), nullable=False),
+        sa.Column("read_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+    )
+    op.create_index("ix_messages_sender_id", "messages", ["sender_id"])
+    op.create_index("ix_messages_recipient_id", "messages", ["recipient_id"])
+    op.create_index("ix_messages_created_at", "messages", ["created_at"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_messages_created_at", table_name="messages")
+    op.drop_index("ix_messages_recipient_id", table_name="messages")
+    op.drop_index("ix_messages_sender_id", table_name="messages")
+    op.drop_table("messages")

backend/alembic/versions/0027_message_e2ee.py

@@ -0,0 +1,68 @@
+"""end-to-end encrypted messages + system messages
+
+Revision ID: 0027_message_e2ee
+Revises: 0026_messages
+Create Date: 2026-06-25
+
+Reworks direct messaging for end-to-end encryption. The `messages` table now stores ciphertext
+(+ iv) for private user-to-user messages instead of plaintext `body`; `body` is kept (nullable)
+for server-authored `kind="system"` messages (welcome / announcements). Adds `message_keys` to
+hold each user's public key + passphrase-wrapped private key (the server can never decrypt).
+
+Any pre-existing rows were plaintext with no keys, so they're dropped (the feature is unshipped;
+prod has none).
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0027_message_e2ee"
+down_revision: Union[str, None] = "0026_messages"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # Old rows were plaintext with no E2EE keys — can't be migrated; drop them.
+    op.execute("DELETE FROM messages")
+    op.add_column(
+        "messages",
+        sa.Column("kind", sa.String(length=16), nullable=False, server_default="user"),
+    )
+    op.add_column("messages", sa.Column("ciphertext", sa.Text(), nullable=True))
+    op.add_column("messages", sa.Column("iv", sa.String(length=32), nullable=True))
+    # System messages have no human sender and carry plaintext `body`.
+    op.alter_column("messages", "sender_id", existing_type=sa.Integer(), nullable=True)
+    op.alter_column("messages", "body", existing_type=sa.Text(), nullable=True)
+
+    op.create_table(
+        "message_keys",
+        sa.Column(
+            "user_id",
+            sa.Integer(),
+            sa.ForeignKey("users.id", ondelete="CASCADE"),
+            primary_key=True,
+        ),
+        sa.Column("public_key", sa.Text(), nullable=False),
+        sa.Column("wrapped_private_key", sa.Text(), nullable=False),
+        sa.Column("salt", sa.String(length=64), nullable=False),
+        sa.Column("wrap_iv", sa.String(length=32), nullable=False),
+        sa.Column("key_check", sa.Text(), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("message_keys")
+    op.execute("DELETE FROM messages")
+    op.alter_column("messages", "body", existing_type=sa.Text(), nullable=False)
+    op.alter_column("messages", "sender_id", existing_type=sa.Integer(), nullable=False)
+    op.drop_column("messages", "iv")
+    op.drop_column("messages", "ciphertext")
+    op.drop_column("messages", "kind")

backend/app/main.py

@@ -34,6 +34,7 @@ from app.routes import (
     feed,
     health,
     me,
+    messages,
     notifications,
     playlists,
     quota,
@@ -114,6 +115,7 @@ app.include_router(tags.router)
 app.include_router(feed.router)
 app.include_router(me.router)
 app.include_router(notifications.router)
+app.include_router(messages.router)
 app.include_router(channels.router)
 app.include_router(playlists.router)
 app.include_router(admin.router)

backend/app/models.py

@@ -518,3 +518,69 @@ class Notification(Base):
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), server_default=func.now(), index=True
     )
+
+
+class Message(Base):
+    """A message in the notification module (phase 2).
+
+    Two kinds share this table:
+      - `kind="user"`: a private, END-TO-END ENCRYPTED direct message between two users. The
+        server only ever stores `ciphertext` + `iv` (base64); it never holds the plaintext or
+        the keys, so not even an admin can read it. A conversation is just the set of messages
+        between two users — no separate thread entity; queries group by the {sender, recipient}
+        pair.
+      - `kind="system"`: a server-authored plaintext message (e.g. the welcome, or future admin
+        announcements). `sender_id` is NULL and `body` holds the text. These are non-private
+        boilerplate, readable without any E2EE key setup, and surface as a "Siftlode" conversation.
+
+    `read_at` is stamped when the recipient opens the thread (NULL = unread, feeds the nav
+    badge). The recipient FK cascades (and sender, for user messages) so a GDPR account erasure
+    removes that user's messages outright."""
+
+    __tablename__ = "messages"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    kind: Mapped[str] = mapped_column(
+        String(16), default="user", server_default="user"
+    )
+    # NULL for system messages (no human sender).
+    sender_id: Mapped[int | None] = mapped_column(
+        ForeignKey("users.id", ondelete="CASCADE"), index=True
+    )
+    recipient_id: Mapped[int] = mapped_column(
+        ForeignKey("users.id", ondelete="CASCADE"), index=True
+    )
+    # System messages carry plaintext `body`; user messages carry `ciphertext`+`iv` (base64).
+    body: Mapped[str | None] = mapped_column(Text)
+    ciphertext: Mapped[str | None] = mapped_column(Text)
+    iv: Mapped[str | None] = mapped_column(String(32))
+    read_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), index=True
+    )
+
+
+class MessageKey(Base):
+    """A user's end-to-end-encryption key material for direct messaging (phase 2).
+
+    One row per user. `public_key` (base64 SPKI, ECDH P-256) is shared with others so they can
+    derive the pairwise conversation key. `wrapped_private_key` is the user's private key
+    encrypted IN THE BROWSER with a key derived (PBKDF2) from a message passphrase the server
+    never sees — so the server stores only an opaque blob it cannot unwrap. `salt`/`wrap_iv`
+    parameterise that derivation; `key_check` is a small ciphertext the client decrypts to
+    verify a passphrase on unlock. The server is purely a key directory + blob store; it can
+    never read any message."""
+
+    __tablename__ = "message_keys"
+
+    user_id: Mapped[int] = mapped_column(
+        ForeignKey("users.id", ondelete="CASCADE"), primary_key=True
+    )
+    public_key: Mapped[str] = mapped_column(Text)
+    wrapped_private_key: Mapped[str] = mapped_column(Text)
+    salt: Mapped[str] = mapped_column(String(64))
+    wrap_iv: Mapped[str] = mapped_column(String(32))
+    key_check: Mapped[str] = mapped_column(Text)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
+    )

backend/app/realtime.py

@@ -0,0 +1,60 @@
+"""In-process WebSocket connection registry for live server→client push.
+
+Generic groundwork (notification module phase 2 is the first user): connections are kept in
+this process's memory, keyed by user_id, so a request handler can push an event to all of a
+user's open tabs/devices. Single-uvicorn-process assumption — if we ever run multiple workers,
+replace the in-memory map with a Redis pub/sub fan-out behind the same `push()` interface (same
+note as app.ratelimit).
+
+`push()` is intentionally SYNC-callable: ordinary (sync) route handlers run in a threadpool, so
+they schedule the actual send onto the WebSocket event loop via run_coroutine_threadsafe. The
+connect/disconnect/_push coroutines all run on that one loop, so the connection map is only ever
+mutated from a single thread.
+"""
+import asyncio
+import logging
+from collections import defaultdict
+
+from starlette.websockets import WebSocket
+
+log = logging.getLogger("siftlode.realtime")
+
+
+class ConnectionManager:
+    def __init__(self) -> None:
+        self._conns: dict[int, set[WebSocket]] = defaultdict(set)
+        self._loop: asyncio.AbstractEventLoop | None = None
+
+    async def connect(self, user_id: int, ws: WebSocket) -> None:
+        self._loop = asyncio.get_running_loop()
+        await ws.accept()
+        self._conns[user_id].add(ws)
+
+    async def disconnect(self, user_id: int, ws: WebSocket) -> None:
+        conns = self._conns.get(user_id)
+        if conns:
+            conns.discard(ws)
+            if not conns:
+                self._conns.pop(user_id, None)
+
+    async def _push(self, user_id: int, payload: dict) -> None:
+        for ws in list(self._conns.get(user_id, ())):
+            try:
+                await ws.send_json(payload)
+            except Exception:
+                # Stale socket; it'll be cleaned up by its own disconnect handler.
+                pass
+
+    def push(self, user_id: int, payload: dict) -> None:
+        """Schedule a push onto the WebSocket event loop. Safe to call from a sync route
+        handler (threadpool). No-op if no WebSocket has ever connected (no loop captured)."""
+        loop = self._loop
+        if loop is None:
+            return
+        try:
+            asyncio.run_coroutine_threadsafe(self._push(user_id, payload), loop)
+        except Exception:
+            log.debug("ws push failed", exc_info=True)
+
+
+manager = ConnectionManager()

backend/app/routes/messages.py

@@ -0,0 +1,373 @@
+"""Direct messaging (notification module, phase 2): end-to-end encrypted user-to-user chat
+plus server-authored system messages, with live WebSocket delivery.
+
+E2EE model: every private message is encrypted IN THE BROWSER under a key both parties derive
+from ECDH(their private key, the other's public key). The server only ever stores ciphertext +
+iv and acts as a key directory (public keys) and an opaque blob store (each user's private key,
+wrapped client-side with a passphrase the server never sees). So not even an admin can read a
+user-to-user conversation. `kind="system"` messages (the welcome; future announcements) are
+plain server boilerplate — non-private, readable without any key setup, shown as a "Siftlode"
+conversation.
+
+Every endpoint is gated by `require_human` (the shared demo account can't message). Metadata
+(who messaged whom, when, read state) is NOT encrypted — standard for E2EE and needed for
+routing and unread counts.
+"""
+from datetime import datetime, timezone
+
+from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect
+from pydantic import BaseModel
+from sqlalchemy import and_, func, or_, select
+from sqlalchemy.orm import Session
+
+from app.auth import require_human
+from app.db import SessionLocal, get_db
+from app.models import Message, MessageKey, User
+from app.ratelimit import RateLimiter
+from app.realtime import manager
+
+router = APIRouter(prefix="/api/messages", tags=["messages"])
+
+MAX_CIPHERTEXT = 8000  # base64 of a generous plaintext + AES-GCM overhead
+SYSTEM_PARTNER_ID = 0  # synthetic conversation partner for system messages
+SYSTEM_NAME = "Siftlode"
+_send_limiter = RateLimiter(max_events=30, window_seconds=60)
+
+# The welcome is rendered server-side at creation time, so it carries its own translations
+# (one per supported UI language). Kept short and non-private (every new user gets the same text).
+WELCOME = {
+    "en": (
+        "Welcome to Siftlode! 👋\n\n"
+        "This is your private message inbox. Conversations with other members are end-to-end "
+        "encrypted — only you and the person you're chatting with can read them, not even an "
+        "admin. To start messaging someone, set up secure messaging with a passphrase first.\n\n"
+        "Enjoy your feed!"
+    ),
+    "hu": (
+        "Üdv a Siftlode-on! 👋\n\n"
+        "Ez a privát üzenet-postafiókod. A többi taggal folytatott beszélgetések végpontig "
+        "titkosítottak — csak te és a beszélgetőpartnered olvashatja őket, még az admin sem. "
+        "Ha üzenni szeretnél valakinek, előbb állítsd be a biztonságos üzenetküldést egy "
+        "jelszóval.\n\n"
+        "Jó böngészést!"
+    ),
+    "de": (
+        "Willkommen bei Siftlode! 👋\n\n"
+        "Das ist dein privater Nachrichten-Posteingang. Unterhaltungen mit anderen Mitgliedern "
+        "sind Ende-zu-Ende-verschlüsselt — nur du und dein Gegenüber könnt sie lesen, nicht "
+        "einmal ein Admin. Um jemandem zu schreiben, richte zuerst die sichere "
+        "Nachrichtenübermittlung mit einem Passwort ein.\n\n"
+        "Viel Spaß mit deinem Feed!"
+    ),
+}
+
+
+class KeySetupIn(BaseModel):
+    public_key: str
+    wrapped_private_key: str
+    salt: str
+    wrap_iv: str
+    key_check: str
+
+
+class SendIn(BaseModel):
+    recipient_id: int
+    ciphertext: str
+    iv: str
+
+
+def _user_label(u: User) -> str:
+    return u.display_name or u.email.split("@")[0]
+
+
+def _serialize_user(u: User) -> dict:
+    return {"id": u.id, "name": _user_label(u), "avatar_url": u.avatar_url}
+
+
+def _system_partner() -> dict:
+    return {"id": SYSTEM_PARTNER_ID, "name": SYSTEM_NAME, "avatar_url": None}
+
+
+def _serialize_msg(m: Message) -> dict:
+    return {
+        "id": m.id,
+        "kind": m.kind,
+        "sender_id": m.sender_id,
+        "recipient_id": m.recipient_id,
+        "body": m.body,  # only set for system messages
+        "ciphertext": m.ciphertext,
+        "iv": m.iv,
+        "read_at": m.read_at.isoformat() if m.read_at else None,
+        "created_at": m.created_at.isoformat() if m.created_at else None,
+    }
+
+
+def _messageable() -> list:
+    return [
+        User.is_demo.is_(False),
+        User.is_active.is_(True),
+        User.is_suspended.is_(False),
+    ]
+
+
+def ensure_welcome(db: Session, user: User) -> None:
+    """Create the one-time system welcome message for a (human) user the first time it's needed.
+    Idempotent: guarded on whether any system message already exists for them."""
+    if user.is_demo:
+        return
+    exists = db.scalar(
+        select(Message.id)
+        .where(Message.recipient_id == user.id, Message.kind == "system")
+        .limit(1)
+    )
+    if exists:
+        return
+    lang = (user.preferences or {}).get("language")
+    body = WELCOME.get(lang, WELCOME["en"])
+    m = Message(kind="system", sender_id=None, recipient_id=user.id, body=body)
+    db.add(m)
+    db.commit()
+    db.refresh(m)
+    manager.push(user.id, {"type": "message", "message": _serialize_msg(m)})
+
+
+# --- E2EE key directory / blob store -----------------------------------------
+
+@router.get("/keys/me")
+def my_key(user: User = Depends(require_human), db: Session = Depends(get_db)) -> dict:
+    """My own key bundle. `configured` says whether secure messaging is set up; if so the
+    wrapped blob + params are returned so this device can unlock with the passphrase."""
+    k = db.get(MessageKey, user.id)
+    if k is None:
+        return {"configured": False}
+    return {
+        "configured": True,
+        "public_key": k.public_key,
+        "wrapped_private_key": k.wrapped_private_key,
+        "salt": k.salt,
+        "wrap_iv": k.wrap_iv,
+        "key_check": k.key_check,
+    }
+
+
+@router.post("/keys")
+def setup_keys(
+    payload: KeySetupIn,
+    user: User = Depends(require_human),
+    db: Session = Depends(get_db),
+) -> dict:
+    """Store this user's public key + passphrase-wrapped private key. Set-once: changing the
+    key would orphan every existing conversation key, so re-setup is rejected here."""
+    if db.get(MessageKey, user.id) is not None:
+        raise HTTPException(status_code=409, detail="Secure messaging is already set up.")
+    db.add(
+        MessageKey(
+            user_id=user.id,
+            public_key=payload.public_key,
+            wrapped_private_key=payload.wrapped_private_key,
+            salt=payload.salt,
+            wrap_iv=payload.wrap_iv,
+            key_check=payload.key_check,
+        )
+    )
+    db.commit()
+    return {"configured": True}
+
+
+@router.get("/keys/{user_id}")
+def public_key(
+    user_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> dict:
+    """A member's public key, for deriving the pairwise conversation key."""
+    k = db.get(MessageKey, user_id)
+    if k is None:
+        raise HTTPException(status_code=404, detail="That member hasn't set up messaging yet.")
+    return {"user_id": user_id, "public_key": k.public_key}
+
+
+# --- directory + conversations -----------------------------------------------
+
+@router.get("/users")
+def list_recipients(
+    user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> list[dict]:
+    """Member directory for the recipient picker: messageable members (except me) who have set
+    up secure messaging (so a conversation key can be derived)."""
+    rows = (
+        db.execute(
+            select(User)
+            .join(MessageKey, MessageKey.user_id == User.id)
+            .where(User.id != user.id, *_messageable())
+            .order_by(func.lower(func.coalesce(User.display_name, User.email)))
+        )
+        .scalars()
+        .all()
+    )
+    return [_serialize_user(u) for u in rows]
+
+
+@router.get("/conversations")
+def list_conversations(
+    user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> dict:
+    """One entry per conversation: each user partner (E2EE) plus the system "Siftlode" thread,
+    with the latest message + my unread count, newest first. Lazily creates the welcome on the
+    first open. User-message previews are ciphertext — the client decrypts them."""
+    ensure_welcome(db, user)
+    msgs = (
+        db.execute(
+            select(Message)
+            .where(or_(Message.sender_id == user.id, Message.recipient_id == user.id))
+            .order_by(Message.created_at.desc())
+        )
+        .scalars()
+        .all()
+    )
+    convs: dict[int, dict] = {}
+    for m in msgs:
+        if m.kind == "system":
+            pid = SYSTEM_PARTNER_ID
+        else:
+            pid = m.recipient_id if m.sender_id == user.id else m.sender_id
+        c = convs.setdefault(pid, {"last": m, "unread": 0})
+        if m.recipient_id == user.id and m.read_at is None:
+            c["unread"] += 1
+    user_ids = [pid for pid in convs if pid != SYSTEM_PARTNER_ID]
+    partners = {}
+    if user_ids:
+        partners = {
+            u.id: u
+            for u in db.execute(select(User).where(User.id.in_(user_ids))).scalars().all()
+        }
+    items = []
+    for pid, c in convs.items():
+        if pid == SYSTEM_PARTNER_ID:
+            partner = _system_partner()
+        elif pid in partners:
+            partner = _serialize_user(partners[pid])
+        else:
+            continue
+        items.append(
+            {"partner": partner, "last_message": _serialize_msg(c["last"]), "unread": c["unread"]}
+        )
+    items.sort(key=lambda x: x["last_message"]["created_at"] or "", reverse=True)
+    return {"items": items}
+
+
+@router.get("/conversations/{partner_id}")
+def get_thread(
+    partner_id: int,
+    mark_read: bool = True,
+    user: User = Depends(require_human),
+    db: Session = Depends(get_db),
+) -> dict:
+    """Full thread with a partner (or the system thread when partner_id == 0), oldest first.
+    Stamps unread incoming messages read unless mark_read=false."""
+    if partner_id == SYSTEM_PARTNER_ID:
+        ensure_welcome(db, user)
+        partner = _system_partner()
+        cond = and_(Message.recipient_id == user.id, Message.kind == "system")
+    else:
+        p = db.get(User, partner_id)
+        if p is None:
+            raise HTTPException(status_code=404, detail="User not found")
+        partner = _serialize_user(p)
+        cond = and_(
+            Message.kind == "user",
+            or_(
+                and_(Message.sender_id == user.id, Message.recipient_id == partner_id),
+                and_(Message.sender_id == partner_id, Message.recipient_id == user.id),
+            ),
+        )
+    msgs = (
+        db.execute(select(Message).where(cond).order_by(Message.created_at.asc()))
+        .scalars()
+        .all()
+    )
+    if mark_read:
+        now = datetime.now(timezone.utc)
+        changed = False
+        for m in msgs:
+            if m.recipient_id == user.id and m.read_at is None:
+                m.read_at = now
+                changed = True
+        if changed:
+            db.commit()
+    return {"partner": partner, "items": [_serialize_msg(m) for m in msgs]}
+
+
+@router.post("")
+def send_message(
+    payload: SendIn,
+    user: User = Depends(require_human),
+    db: Session = Depends(get_db),
+) -> dict:
+    """Send an end-to-end-encrypted message. The server stores only the ciphertext the client
+    produced, then pushes it live to the recipient (and the sender's other devices)."""
+    if not _send_limiter.allow(f"msg:{user.id}"):
+        raise HTTPException(status_code=429, detail="Too many messages — slow down a moment.")
+    ct = (payload.ciphertext or "").strip()
+    iv = (payload.iv or "").strip()
+    if not ct or not iv:
+        raise HTTPException(status_code=400, detail="Message is empty.")
+    if len(ct) > MAX_CIPHERTEXT:
+        raise HTTPException(status_code=400, detail="Message is too long.")
+    if payload.recipient_id == user.id:
+        raise HTTPException(status_code=400, detail="You can't message yourself.")
+    recipient = db.get(User, payload.recipient_id)
+    if recipient is None or recipient.is_demo or not recipient.is_active or recipient.is_suspended:
+        raise HTTPException(status_code=404, detail="Recipient not available.")
+    if db.get(MessageKey, recipient.id) is None:
+        raise HTTPException(status_code=404, detail="That member hasn't set up messaging yet.")
+    m = Message(
+        kind="user", sender_id=user.id, recipient_id=recipient.id, ciphertext=ct, iv=iv
+    )
+    db.add(m)
+    db.commit()
+    db.refresh(m)
+    event = {"type": "message", "message": _serialize_msg(m)}
+    manager.push(recipient.id, event)
+    manager.push(user.id, event)  # the sender's other open devices
+    return _serialize_msg(m)
+
+
+@router.get("/unread_count")
+def unread_count(
+    user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> dict:
+    n = db.scalar(
+        select(func.count())
+        .select_from(Message)
+        .where(Message.recipient_id == user.id, Message.read_at.is_(None))
+    )
+    return {"count": n or 0}
+
+
+# --- live delivery -----------------------------------------------------------
+
+@router.websocket("/ws")
+async def messages_ws(ws: WebSocket) -> None:
+    """Live push channel: authenticated by the session cookie, registers the connection so
+    sent messages reach this user's open tabs instantly. We don't consume client→server frames
+    (sending goes through POST /api/messages); the receive loop only detects disconnect."""
+    uid = ws.session.get("user_id") if "session" in ws.scope else None
+    if not uid:
+        await ws.close(code=1008)
+        return
+    db = SessionLocal()
+    try:
+        u = db.get(User, uid)
+        ok = bool(u and not u.is_demo and not u.is_suspended and u.is_active)
+    finally:
+        db.close()
+    if not ok:
+        await ws.close(code=1008)
+        return
+    await manager.connect(uid, ws)
+    try:
+        while True:
+            await ws.receive_text()
+    except WebSocketDisconnect:
+        pass
+    finally:
+        await manager.disconnect(uid, ws)