Messaging (`devplacepy/services/messaging/`)

This file documents the real-time direct-message chat subsystem. Claude Code auto-loads it when a file under devplacepy/services/messaging/ is read or edited.

Overview

The /messages feature is a live WebSocket chat layered over the SAME messages table and audit/notification cores as before; there is no parallel chat store. The WS only adds live delivery, presence, typing, and read receipts on top of the existing persistence.

messages.py (the router, mounted at /messages) exposes: GET "" (page; the conversation sidebar is one windowed SQL aggregate and an opened thread loads only the newest CONVERSATION_MESSAGE_LIMIT = 500 messages - keep these queries bounded), GET /search, POST /send (no-JS fallback), and WS /messages/ws (NOT lock-owner gated - accepts on every worker; closes 1008 for guests).

Bounded page queries (load-bearing performance rule)

get_conversations is ONE window-function query (ROW_NUMBER() OVER (PARTITION BY other_uid ORDER BY created_at DESC, id DESC) over sender_uid = :me OR receiver_uid = :me, the get_recent_comments_by_target_uids precedent) that returns only each partner's latest row - never load the user's full message history into Python to build the sidebar. get_conversation_messages loads at most CONVERSATION_MESSAGE_LIMIT (500) most-recent rows (ORDER BY created_at DESC, id DESC LIMIT, reversed to ascending in Python); older history stays in the DB and is simply not rendered. Both are covered by idx_messages_conversation/_rev (verified EXPLAIN QUERY PLAN, MULTI-INDEX OR, no table scan). messages is NOT in SOFT_DELETE_TABLES, so neither query filters deleted_at. Any new message listing must stay bounded and indexed the same way.

WS endpoint `/messages/ws`

Lives on the existing messages router (no new router/prefix; already mounted at /messages). It is NOT lock-owner gated (deliberately - unlike /devii/ws): await websocket.accept(), resolve the owner via session cookie / api-key (_resolve_ws_user reuses utils._user_from_session/_user_from_api_key), and accept on EVERY worker. Gating on service_manager.owns_lock() was the original design and was wrong here: when no worker holds the service lock (services disabled, or the lock not yet acquired) every socket got closed 4013 and the client fast-retried every 200ms forever (a connection storm with no stream, and the send form fell back to a full-page POST reload). Chat must not depend on the background-service lock. Messaging requires an authenticated user - guests are closed with 1008. The receive loop is wrapped in try/except WebSocketDisconnect + a broad except (never crash the worker) and ALWAYS unregisters the socket in finally.

Connection registry

The ConnectionManager singleton message_hub (services/messaging/hub.py), one per worker: user_uid -> set[WebSocket], plus an in-process last_seen map and a bounded delivered LRU set (mark_delivered/was_delivered, cap 4000) used to dedupe direct vs relayed delivery. register/unregister return whether the user just came online / went offline (for presence announces). send_to_user/send_to_users fan a frame out to all of a user's sockets (multi-tab) and swallow dead-socket errors. Presence and last-seen are in-process per worker (no DB column).

Cross-worker delivery (the relay)

Because the WS accepts on every worker, a message persisted on worker A must still reach a recipient whose socket lives on worker B. services/messaging/relay.py message_relay (a per-worker singleton asyncio loop, started on the first socket connect, self-stops when message_hub.has_connections() is false) polls SELECT * FROM messages WHERE id > :watermark (uuid7-backed autoincrement id) every ~1s and pushes any row whose sender/receiver has a LOCAL socket and that was not already was_delivered to those local sockets, advancing the watermark to the max row seen. Same-worker sends are delivered INSTANTLY by broadcast_message (which calls message_hub.mark_delivered so the relay skips them); the relay only fills the cross-worker gap, so same-worker latency is zero and cross-worker latency is bounded by the poll interval. The relay primes its watermark to the current MAX(id) on first start so it never replays history. Trade-off: typing/read/presence frames are in-process per worker only - across the two make prod workers those ephemeral signals reach only same-worker peers (message delivery is always correct). On a single dev worker everything is instant and complete.

DRY persist choke point

services/messaging/persist.py persist_message(sender, receiver_uid, content, attachment_uids, *, request=None, origin) is the ONE function that inserts the row, links attachments, fires create_notification + clear_messages_cache + create_mention_notifications, logs, and writes the message.send audit event. Both the WS send handler and the HTTP POST /messages/send handler call it, so audit/notification/mention behavior is byte-identical on both paths (the Devii send_message action is handler="http" -> POST /messages/send, so it also flows through here and broadcasts live). Content is capped at 2000 chars server-side. When request is a Request it audits via audit.record; on the WS path it now passes request=websocket (auditing via audit.record_system with actor_kind="user" + origin="websocket", same message.send key/category - no new event invented; presence/typing/read are ephemeral and NOT audited).

AI correction and AI modifier apply to direct messages, with LIVE delivery of the final content

"messages": ("content",) is in the CORRECTABLE_FIELDS registry and persist_message invokes schedule_correction/schedule_modification (sender = the user), so typing @ai <instruction> in a DM runs the AI modifier (default on + sync) and an enabled correction rewrites the content. Both send paths funnel through routers/messages._finalize_and_broadcast(sender, message, request, client_id=None): it awaits any pending SYNC correction/modification futures stashed on request.scope[PENDING_SCOPE_KEY] (the same PENDING_SCOPE_KEY/loop.run_in_executor mechanism the HTTP await_pending_corrections middleware uses), RE-READS the message row, and broadcasts the FINAL (corrected/modified) content via broadcast_message. The HTTP POST /send and the WS /ws send path both call it; the WS path passes request=websocket to persist_message so the modifier stashes its executor future on the websocket scope and the handler awaits it directly (HTTP middleware does not run for websockets). A message with no @ai directive and correction off has nothing pending, so the broadcast is immediate with zero added latency. Frontend static/js/MessagesLayout.js reconciles the sender's own echo by client_id and replaces the optimistic bubble's .rendered-content with the server's authoritative frame.content, re-rendered - so the sender sees their @ai directive resolved in-place (and corrections become visible to the sender too). See "AI content correction" and "AI modifier".

WS protocol frames

Client -> server: {type:"send", receiver_uid, content, attachment_uids?, client_id}, {type:"typing", receiver_uid} (throttled client-side), {type:"read", with_uid}, {type:"presence", with_uid} (one-shot presence query). Server -> client: {type:"ready", user_uid} (sent first on connect), {type:"message", uid, sender_uid, sender_username, receiver_uid, content, created_at, time_ago, client_id} (broadcast to BOTH sender and receiver sockets so multi-tab and the sender's own optimistic bubble reconcile via the echoed client_id), {type:"typing", from_uid} (to the receiver only), {type:"read", by_uid} (read-receipt to the other user), {type:"presence", user_uid, online, last_seen} (announced to everyone in an open conversation with the user on connect/disconnect, and as a direct reply to a presence query). The WS read path reuses mark_conversation_read (the same UPDATE messages SET read=1 SQL the page uses) and clear_messages_cache.

Renderer integration (XSS control)

Live/echoed message bubbles are NEVER injected as raw HTML. MessagesLayout.buildBubble creates a <div class="rendered-content" data-render>, sets its textContent to the message body, then runs contentRenderer.applyTo(el) + marks img:not(.avatar-img) with data-lightbox + contentRenderer.highlightAll() - exactly the ContentEnhancer.initContentRenderer sequence (emoji shortcode -> marked -> DOMPurify.sanitize -> highlight.js -> image/YouTube/autolink). This gives Discord-style emoji, image and YouTube embeds, autolinking, and sanitization for free.

Frontend

MessagesSocket.js (mirrors DeviiSocket.js: reconnect with backoff, 4013 silent fast-retry, onReady fires on the first server frame not raw open) points at /messages/ws. MessagesLayout.js (instantiated on app in Application.js) opens the socket on the messages page, sends over WS with an optimistic pending bubble keyed by client_id, reconciles on the echoed message frame, appends incoming messages live through the renderer, auto-scrolls, throttles + shows the typing indicator, flips the read-receipt double-check, renders the presence dot / last-seen in the header, and live-updates the conversation-list preview + unread dot + ordering. If the socket is not open the send <form method=POST> submits normally (no-JS fallback). The template carries data-self-uid/data-other-uid/data-other-online/data-other-last-seen on .messages-layout. Styling (presence dot, typing dots, read-receipt, unread dot, pending opacity) is in messages.css using design tokens only, responsive down to 360px.

Attachments stream live

The single message_frame builder (services/messaging/persist.py, used by BOTH broadcast_message and the relay) fetches get_attachments("message", uid) and includes a slimmed list (uid/url/thumbnail_url/is_image/is_video/original_filename/file_size/mime_type) on every frame. MessagesLayout.renderAttachments mirrors _attachment_display.html in the DOM (image -> img.gallery-thumb marked data-lightbox+data-full, video -> <video>, else a download link) - no innerHTML, so it stays XSS-safe. The sender's optimistic bubble shows an "Uploading attachment..." placeholder until its own echo arrives and swaps in the real gallery. Attachment-only messages (empty caption) are allowed: the content input dropped required, sendViaSocket sends when content OR attachments are present, and persist_message accepts empty content when attachment_uids is non-empty (if not content and not attachment_uids: return None). dp-upload (mode attachment) uploads each file to /uploads/upload and writes the uids into its hidden attachment_uids input; collectAttachments() reads that before the send, then dp-upload.clear() resets it. The chat uploader sets hide-chips so only the (N) count badge shows (no filename chips) - a generic dp-upload attribute. The send button shows a spinner and is disabled while busy: MessagesLayout.refreshSendButton() ORs _uploading (driven by the new dp-upload:busy event) with _pendingSends (a set of in-flight send client_ids, cleared on the echo or an 8s safety timeout), toggling .is-sending + disabled on .messages-send-btn.

Messaging (devplacepy/services/messaging/)