contact messages: wire the sockets

/_ws routes frames carrying a session token to pagerite.messages (and
now also pushes thread state back down the same socket); the visitor
registry is cleaned on disconnect. /_api/ws/analytics becomes
bidirectional: admins get the contact inbox on connect, visitor messages
are pushed live, and reply/read frames are handled. Analytics payloads
keep their type-less shape; messaging frames always carry a type field.

pagerite/tracking.py

@@ -29,7 +29,7 @@ from fastapi import APIRouter, Request, WebSocket, WebSocketDisconnect
 from fastapi.responses import Response
 from uarite import uaparse
 
-from pagerite import analytics, i18n
+from pagerite import analytics, i18n, messages
 from pagerite.data import resolve
 from pagerite.files import _hash_name, file_store
 from pagerite.state import SITE_URL, _html_response, analytics_store, data
@@ -447,6 +447,10 @@ async def activity_ws(ws: WebSocket) -> None:
     abusive IPs are filtered at display time, not here.  The reverse-DNS and
     DB-IP geoip lookups happen in background tasks so message handling is
     never delayed by slow DNS or the first MMDB decompress.
+
+    Frames carrying a ``token`` field are visitor contact messages
+    (docs/messages.md), routed to pagerite.messages — the same socket
+    serves the contact panel, which also receives thread pushes on it.
     """
     ip = _client_ip(ws)
     ua = ws.headers.get("user-agent", "")
@@ -462,7 +466,14 @@ async def activity_ws(ws: WebSocket) -> None:
         while True:
             text = await ws.receive_text()
             try:
-                msg = msgspec.json.decode(text.encode(), type=analytics.Ping)
+                raw = msgspec.json.decode(text.encode())
+            except msgspec.DecodeError:
+                continue
+            if isinstance(raw, dict) and "token" in raw:
+                await messages.visitor_frame(ws, raw)
+                continue
+            try:
+                msg = msgspec.convert(raw, type=analytics.Ping)
             except msgspec.DecodeError:
                 continue
             new_client = analytics_store.record_msg(
@@ -480,6 +491,8 @@ async def activity_ws(ws: WebSocket) -> None:
             _schedule_favicon_fetch()
     except WebSocketDisconnect:
         pass
+    finally:
+        messages.visitor_disconnect(ws)
 
 
 @router.websocket("/_api/ws/analytics")
@@ -488,15 +501,30 @@ async def analytics_websocket(ws: WebSocket) -> None:
 
     Admin-only via the /_api forward-auth gate, like every management
     endpoint. Powers the analytics viewer rendered at /_a.
+
+    The same socket serves the admin contact inbox (docs/messages.md):
+    on connect the thread list follows the snapshot, new visitor messages
+    are pushed as they arrive, and incoming frames (``reply``/``read``)
+    go to pagerite.messages. Analytics payloads carry no ``type`` field;
+    messaging payloads always do, so the client can tell them apart.
     """
     await ws.accept()
     await ws.send_text(_display_json())
     _analytics_ws_clients.add(ws)
+    await messages.admin_connect(ws)
     try:
-        # Receive until the client goes away; we only push.
+        # Receive until the client goes away; messaging frames are handled,
+        # anything else is ignored (analytics only pushes).
         while True:
-            await ws.receive_text()
+            text = await ws.receive_text()
+            try:
+                raw = msgspec.json.decode(text.encode())
+            except msgspec.DecodeError:
+                continue
+            if isinstance(raw, dict):
+                await messages.admin_frame(ws, raw)
     except WebSocketDisconnect:
         logger.debug("Analytics WS client disconnected")
     finally:
         _analytics_ws_clients.discard(ws)
+        messages.admin_disconnect(ws)