fix(auth): close SA5 timing oracles on register/reset + dedup messages_ws

Loose ends to finish the auth security round:
- register + password-reset-request had an enumeration TIMING oracle: an already-
  registered email skipped the create path (hash + row writes + email scheduling) and
  responded measurably faster than a new one. Move the whole lookup+create (register)
  and lookup+token+email (reset) into a background task with its own DB session, so the
  endpoint returns in the same time for any valid email regardless of whether it exists.
  Verified: existing vs new now ~equal (was 34ms vs 82ms on register); accounts/tokens
  still created off-path.
- messages_ws re-implemented resolved_user_id's per-tab wallet-gated account resolution.
  Generalize resolved_user_id to take any HTTPConnection (Request OR WebSocket) and call
  it from the WS — one shared, wallet-gated resolution. Behavior-identical (unit-checked).

backend/app/routes/messages.py

@@ -20,7 +20,7 @@ from pydantic import BaseModel
 from sqlalchemy import and_, func, or_, select
 from sqlalchemy.orm import Session
 
-from app.auth import require_human
+from app.auth import require_human, resolved_user_id
 from app.db import SessionLocal, get_db
 from app.models import Message, MessageKey, User
 from app.ratelimit import RateLimiter
@@ -363,19 +363,10 @@ 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."""
-    # A browser WebSocket can't send custom headers, so the per-tab account (see
-    # X-Siftlode-Account on HTTP) rides in the ?account= query param instead — honoured only for
-    # an account already in this browser's wallet. Falls back to the session default.
-    sess = ws.session if "session" in ws.scope else {}
-    uid = sess.get("user_id")
-    q = ws.query_params.get("account")
-    if q:
-        try:
-            qid = int(q)
-        except ValueError:
-            qid = None
-        if qid is not None and qid in (sess.get("account_ids") or []):
-            uid = qid
+    # Which signed-in account this socket acts as — the SHARED per-tab resolution (wallet-gated
+    # ?account= override, session default otherwise). A browser WebSocket can't send the
+    # X-Siftlode-Account header, so resolved_user_id falls through to the ?account= query param.
+    uid, _ = resolved_user_id(ws)
     if not uid:
         await ws.close(code=1008)
         return
@@ -385,7 +376,7 @@ async def messages_ws(ws: WebSocket) -> None:
         # SA4: honour server-side session revocation on the live channel too — reject a cookie whose
         # recorded epoch is behind the account's current one (mirrors current_user). Without this a
         # copied cookie could keep receiving pushes after a password reset / "log out everywhere".
-        epochs = sess.get("epochs") or {}
+        epochs = (ws.session if "session" in ws.scope else {}).get("epochs") or {}
         epoch_ok = user is not None and int(epochs.get(str(uid), 0)) == (user.session_epoch or 0)
         ok = epoch_ok and is_messageable_user(user)
     finally: