feat(gateway)!: migrate realtime from socket.io to raw WebSocket - #2810
Conversation
New @mx-space/ws-client workspace package for the socket.io-to-ws migration: reconnecting client (exponential backoff + full jitter), request/ack correlation, ping/pong keepalive, and the shared wire protocol (WsEnvelope/WsAckPayload + guards) under a ./protocol subpath export. Zero runtime dependencies.
Enum keys stay unchanged; values move from SCREAMING_SNAKE to dot-namespaced strings (post.create, ai_agent.message, ...) ahead of the socket.io -> raw WebSocket migration. SERVERLESS_EVENT_PREFIX 'fn#' becomes 'fn.'. Updates the admin socket/types.ts mirror and fixes tests/docs pinning old literal values.
…to 1.0.0 Regenerates src/models.generated.ts from the updated core enum (pnpm -C packages/webhook build) so BusinessEvents/EventScope carry the new dot-namespaced values. Bumps 0.10.1 -> 1.0.0 since every subscriber string a consumer passes to handler.emitter.on(...) is a breaking change. Updates readme examples and the companion-presence contract test to the new values.
…elope) Adds apps/core/src/processors/gateway/ws/ — the raw-ws building blocks the socket.io-to-ws gateway rewrite (next task) will consume: WsBusService (Redis pub/sub fanout across pods), WsPresenceService (Redis-backed connection/room bookkeeping + dead-node GC), WsRoomManager and WsConnectionRegistry (in-process room/connection tracking), and ws-envelope helpers (incoming frame validation, ack building). Existing socket.io gateways are untouched; only gateway.module.ts wiring and cache.constant.ts key additions were needed alongside the new files.
leaveRoom, sweepDeadNode's per-room cleanup, and roomSizes' lazy-prune each did HDEL/HLEN-read then a separate DEL+SREM — a concurrent joinRoom from another node between the read and the write could get its just-written membership destroyed and the room wrongly evicted from the rooms set. Replace all three with one shared atomic EVAL (ROOM_PRUNE_SCRIPT / pruneRoomIfEmpty) that HDELs then re-checks HLEN inside the same script. roomSizes also now checks the per-command error slot of the pipelined HLEN results before treating a room as empty — an errored HLEN no longer falls through to "size 0" and triggers a destructive prune.
Install the WsAdapter with the envelope messageParser in bootstrap, drop RedisIoAdapter and the redis emitter surface, and add the inbound event constants and payload schemas.
Both gateways now run on @nestjs/platform-ws behind /ws/web and /ws/admin. WsGatewayBase carries the shared transport: connection registry, local room manager, redis bus subscription, 30s heartbeat and disconnect release. Broadcast goes out exclusively through the bus, so self-delivery follows the same path as cross-node delivery. Inbound commands are dot-named, validated per event and answered with an ack when the frame carries an id. GatewayService keys metadata by the new connection id and no longer wipes the shared hash at boot, which used to let one replica erase another's entries.
activity resolves a presence sid against either the connection id or the handshake session id, since clients never learn the server-generated id, and reads connectedAt from metadata now that there is no handshake. The pg crud factory composes dot event names to match the BusinessEvents values.
The web gateway resolved the reader session before writing presence, so a socket closing during that lookup let handleDisconnect run its cleanup first and the resumed connect handler then re-created the conns and metadata rows. Nothing reclaimed them: presence GC only sweeps ids owned by dead nodes, so each phantom kept inflating the visitor count for the life of the process. Presence and metadata are now written before the auth lookup, and the handler re-checks that the connection is still tracked before writing anything else, undoing its own writes when it is not. The auth gateway guards its token binding the same way.
…bus/presence Real-socket, real-Redis end-to-end specs for the raw-ws gateway layer: connect/greeting, room-scoped broadcast delivery, lang rooms, unique-session online counting, disconnect broadcast, malformed/unknown frame survival, the three admin auth paths plus auth failure and token-expiry close, ai_task room-scoped delivery, and a dual-instance cross-node spec covering the bus delivery path, the shared presence path, and dead-node reclamation via presence.sweepOnce. Extracts createWsAdapter(app) out of bootstrap.ts (pure extraction, no behavior change) so the harness can wire the identical adapter.
Zero-comments-by-default violation: 5 explanatory comment blocks removed (redis timeout, teardown grace, duplicated testTimeout rationale x3, afterEach presence-settle rationale). Kept the WsTestClient listener-ordering comment, compressed, since it documents an unexpected-behavior workaround a future reader would otherwise revert. Extracted the 50ms teardown grace into a named REDIS_QUIT_GRACE_MS constant instead of prose. No logic changes.
Replaces socket.io-client with the shared @mx-space/ws-client transport against the raw ws gateway at /ws/admin, matching the server-side cutover in T3/T4. SocketBridge now dispatches per-event via ws-client's on() API instead of parsing a single socket.io 'message' frame, and DEV-only lifecycle toasts map from ws-client's $state instead of socket.io's connect/reconnect event set. useTaskSubscription switches ai-task:subscribe/unsubscribe emits to ai_task.subscribe/unsubscribe request()s, marking itself subscribed only after an ok:true ack and resubscribing on $state reconnect. Also removes two dead EventTypes members (PAGE_UPDATED, DANMAKU_CREATE) confirmed unused by grep, and re-exports WsAckPayload/WsRequestError from ws-client's index for consumers.
subscribe()'s ack handler set `subscribed = true` unconditionally, but unmount (or a visibility-hide) could run first and see `subscribed === false`, so cleanup's unsubscribe() silently no-opped. The subscribe ack then resolved into a dead closure with nobody left to unsubscribe it, leaking the server-side ai_task room subscription for the life of the page's WsClient. Track the caller's live intent (wantSubscribed) independently of subscribed/pending so a stale ack that resolves after intent has flipped fires ai_task.unsubscribe itself instead of marking a torn-down subscription active.
…ibers Round 1's stale-ack guard was scoped to a single effect closure, so it only knew whether THIS mount still wanted the subscription. Server-side room membership is per-connection with no refcount, so an unmount followed by a same-payload remount (two detail panels for one task, StrictMode's double-invoke, or a fast route remount) let the unmounted instance's deferred ai_task.unsubscribe silently drop the room out from under the live remounted sibling — subscribed=true locally, no error, no $state change, updates just stop arriving. Track want-count per payload (JSON-keyed) at module scope instead of per-closure. Every instance's intent flip increments/decrements the shared count; the wire-level ai_task.unsubscribe (whether sent immediately or deferred until a stale ack resolves) now only fires when that count reaches zero, i.e. nobody else sharing the payload still wants it live.
- require non-empty sid in UpdatePresenceSchema (empty string matched connections whose socket metadata write failed) - cover WsGatewayBase heartbeat sweep with a unit spec (dead-connection terminate on second sweep, alive connection survives repeated pongs, closed sockets are skipped) - fix ws-client package.json license SPDX id (AGPLv3 -> AGPL-3.0-only)
- heartbeat: implemented semantics is single-cycle isAlive marking (terminate on the sweep after a missed pong), not "two consecutive missed pongs" - unknown event: note the shipped deviation -- WsAdapter silently swallows unmatched events with no UNKNOWN_EVENT ack; client request() falls back to its own timeout - release: spell out the pre-worktree-deletion steps -- web-gateway upgrade routes for /ws/web and /ws/admin, Yohaku/tg-bot file: deps to npm semver, ws-client private:false + AGPL-3.0-only license, and publish order (webhook 1.0.0 + ws-client before core v14) - non-goals: flag apps/mobile (Yohaku repo) as still on socket.io, needing separate follow-up before cutover
SafeDep Report Summary
Package Details
This report is generated by SafeDep Github App |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 7a461168a6
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| this.hooks.onDisconnected.forEach((fn) => fn(conn)) | ||
| leftRooms.forEach((room) => { | ||
| this.hooks.onLeaveRoom.forEach((fn) => fn(conn, room)) |
There was a problem hiding this comment.
When a visitor with an active presence disconnects after joining a room, this invokes both onDisconnected and onLeaveRoom for the same connection. ActivityService registers handlePresencePersistToDb directly for the former and calls it again from the latter, so the same read-duration activity is inserted at least twice—and once more for every additional room such as lang:*—corrupting activity analytics.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in 3e57f39 — added a synchronous check-and-set dedupe keyed on (socket.id, operationTime) in ActivityService, so the onDisconnected + per-room onLeaveRoom invocations for one disconnect persist a single read-duration row. Note this was pre-existing on master (identical hook wiring), not introduced by the migration. Regression test: ActivityService presence persist dedupe fires all three hooks concurrently and asserts one insert.
| await this.redis.del(this.nodeKey()).catch(() => undefined) | ||
| await this.redis.srem(this.nodesKey(), this.nodeId).catch(() => undefined) |
There was a problem hiding this comment.
During a graceful shutdown, removing nodeId from WsNodes makes this node undiscoverable to sweepOnce(), which only iterates IDs still in that set. Because the connection, metadata, and room hashes have no TTL and are not removed here, every graceful deploy leaves permanent phantom connections and room members, inflating online counts and returning nonexistent sockets until Redis is manually cleaned.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in 3e57f39 — onModuleDestroy now runs sweepDeadNode(this.nodeId) to reclaim this node's conn/room/metadata entries before deleting the node key and srem-ing it from the nodes set. Regression test in cross-node.e2e-spec: after a graceful presence shutdown, node B observes zero room members and no lingering connection ids.
| const remaining = await this.presence.roomMemberIds('admin', room) | ||
| const others = remaining.filter((id) => id !== selfId) | ||
| if (others.length === 0) { | ||
| await this.roomSubs.remove(room) |
There was a problem hiding this comment.
When this pod's last subscriber leaves while another pod still has a member, the global roomMemberIds() result keeps this branch from calling roomSubs.remove(). The current pod therefore retains the room in its local heartbeat set and refreshes its stale pod entry indefinitely; after the remote subscriber also leaves, emitToAdminRoom() continues publishing every task update despite there being no recipients. The emptiness check here needs to be local to this gateway's room manager.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in 3e57f39 — dropEmptyRoomSubs now checks the local roomManager.membersOf(room) instead of the global presence registry; roomSubs entries are per-pod, so a remote member keeps only its own pod's entry alive. Regression test in cross-node.e2e-spec: A and B both subscribe, A unsubscribes, and the redis set retains only B's podId.
The duplicated subscriber client inherits enableOfflineQueue: false, so a subscribe issued before its connection was established was rejected outright with no retry — the bus never delivered any broadcast in production shape while the e2e fake redis (offline queue enabled) masked it.
- reclaim this node's conn/room entries during graceful shutdown before deregistering from the nodes set, which sweepOnce alone can no longer reach - decide ai-task roomSubs removal on local room membership: entries are per-pod, so a remote member must not keep this pod's subscription alive - dedupe presence read-duration persistence across the onDisconnected + per-room onLeaveRoom hooks that all fire for one disconnect (pre-existing on master, surfaced by review)
|
Bugbot is not enabled for your account, so this pull request was not reviewed. Enable Bugbot in the Cursor dashboard to get automatic reviews on future PRs. |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 3e57f394b5
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| await subClient.subscribe(this.channel) | ||
| } catch (error) { | ||
| this.warn('Failed to subscribe ws bus channel', error) |
There was a problem hiding this comment.
If the duplicated Redis client takes longer than 15 seconds to become ready, or the initial SUBSCRIBE command fails after the main client is usable, this catch logs the error and leaves the subscriber running without ever retrying. Because both cross-node and local gateway delivery flow exclusively through this channel, that process will accept WebSocket clients but deliver no broadcasts until it is restarted; subscribe again on a later ready event or fail startup instead.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in d7f2962 — a subscribed flag plus a ready-event listener retries the subscription until it succeeds; once subscribed, ioredis restores it across reconnects itself. Regression test: initial waitForReady failure, then a fired ready event ends subscribed.
| async removeConnection(ns: WsNamespace, id: string): Promise<void> { | ||
| try { | ||
| await this.redis.hdel(this.connsKey(ns), id) | ||
| } catch (error) { | ||
| this.warn('Failed to remove ws connection', error) |
There was a problem hiding this comment.
When Redis is briefly unavailable during a disconnect, this failed HDEL is only logged and the connection remains permanently present in mx-ws:conns:*; the same problem affects failed room leaves. sweepOnce() only removes records owned by nodes whose heartbeat has expired, so entries owned by this still-running node are never retried or reclaimed, leaving online counts and room membership polluted until the process itself stops.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in d7f2962 — gateways register a live local-connection index with WsPresenceService, and sweepOnce now also reconciles entries owned by this node against it, reclaiming conn/room/metadata writes lost to transient redis failures. Regression test: a zombie conn + room member owned by a live node is pruned while the live conn survives.
| { | ||
| "name": "@mx-space/ws-client", | ||
| "version": "0.1.0", | ||
| "private": true, |
There was a problem hiding this comment.
The migration's release sequence requires publishing @mx-space/ws-client before deploying core and the external consumers, but npm treats a package with private: true as non-publishable. Workspace builds mask this because they link the source directly, while Yohaku and mx-tg-bot cannot install the required package from the registry, blocking the stated cutover.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in d7f2962 — dropped private: true and added prepublishOnly: tsdown, so npm publish builds dist and ships the publishConfig (dist) shape. The flag was a deliberate guard against premature publishing, but with @mx-space/webhook@1.0.0 already out the guard now costs more than it protects.
…tion - retry the bus subscription on later ready events so a slow or flapping redis at boot cannot leave the node permanently broadcast-dead - reconcile this node's own conn/room entries against the live local registry during sweeps, reclaiming writes lost to transient redis failures on disconnect - drop private: true from @mx-space/ws-client and add prepublishOnly so the package is publishable for the release sequence
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: d7f2962b3e
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| credentials: true, | ||
| }, | ||
| }) | ||
| @WebSocketGateway({ path: '/ws/web' }) |
There was a problem hiding this comment.
For the unauthenticated /ws/web endpoint, specifying only path leaves ws at its 100 MiB maxPayload default, whereas the replaced Socket.IO transport rejected messages above its roughly 1 MiB default. An external client can therefore make the server buffer and JSON-parse very large frames, and several concurrent connections can cause severe memory pressure or process termination; set an explicit, appropriately small maxPayload on this gateway.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in 4d41d60 — both gateways now set maxPayload: 1024 * 1024 (passed through WsAdapter to ws.Server), matching the ~1 MiB limit the socket.io transport enforced. Oversized frames close the socket at the transport layer before any buffering/parse.
| } catch (error) { | ||
| this.warn('Failed to register ws connection', error) | ||
| } | ||
| } |
There was a problem hiding this comment.
When Redis is unavailable during this HSET, the connection remains live in the local registry but is permanently absent from mx-ws:conns:web, so online counts omit it and room/activity lookups cannot find it after Redis recovers; failed joinRoom() writes have the analogous problem. Fresh evidence after the earlier reconciliation change is that reconcileOwnEntries() only deletes stale owned fields and never recreates missing entries for live connections or their rooms, so reconciliation must also restore the local registry state.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in 4d41d60 — the local index is now a full snapshot ({conns, rooms}) and reconciliation works in both directions: owned entries with no live connection are dropped, and live connections/room memberships missing from redis are re-added with this nodeId. Caveat noted: socket metadata contents (sessionId etc.) written by the gateway cannot be reconstructed here, so a restored conn participates in counts/routing but its metadata stays absent until the client reconnects. Regression test covers restore + prune in one sweep.
- cap inbound frames at 1 MiB on both gateways, matching the socket.io transport's limit instead of ws's 100 MiB default - presence reconciliation now works both ways: gateways hand the sweep a full local snapshot (conns + room memberships) so entries lost to transient redis failures at connect/join time are re-added, not only stale ones removed
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 4d41d60489
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
|
|
||
| const candidate = value as Record<string, unknown> | ||
|
|
||
| if (typeof candidate.v !== 'number') return false |
There was a problem hiding this comment.
When a newer or incompatible peer sends a v: 2 frame, this predicate accepts it as a valid v1 envelope and createWsClient dispatches the event—or even resolves a pending request if it is an ack—using v1 semantics. Since the server parser explicitly accepts only version 1, the shared client should likewise require candidate.v === WS_PROTOCOL_VERSION so older clients ignore frames they cannot safely interpret.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in a983b32 — isWsEnvelope now requires candidate.v === WS_PROTOCOL_VERSION, matching the server parser's v:1 literal; v:2/v:0 frames are dropped instead of being dispatched or resolving a pending ack under v1 semantics. Tests added for both.
isWsEnvelope accepted any numeric v, so a hypothetical v2 frame would be dispatched (or resolve a pending ack) under v1 semantics while the server parser pins v to 1 — pin the client predicate to the same literal.
1 participant
Summary
Replaces the socket.io realtime layer with raw WebSocket (
@nestjs/platform-ws). Web clients no longer need any framework — nativeWebSocket+ the new zero-dependency@mx-space/ws-clientpackage.{v:1, event, payload?, id?}; acks for id-carrying uplinks; paths/ws/weband/ws/adminBusinessEventsvalues renamed to Stripe-style dot form (POST_CREATE→post.create,fn#→fn.,companion.presence.changed→companion_presence.changed). Webhook egress carries the new names;@mx-space/webhookbumped to 1.0.0mx-ws:bus) replaces@socket.io/redis-emitter; write-through presence registry (heartbeat + GC, atomic Lua room cleanup) replacesfetchSocketsscatter-gather — soft-timeout/fallback machinery deletedpackages/ws-client: isomorphic client (reconnect backoff, ack requests, ping liveness) shared by admin, Yohaku, tg-botuseTaskSubscriptionis ack-gated with a cross-mount refcountRelease order (see spec §7)
@mx-space/webhook@1.0.0+@mx-space/ws-client(flipprivate, license fixed)/ws/web+/ws/adminupgrade routes to web-gateway (nothing connects without this)file:deps to npm semver first)Notes
UNKNOWN_EVENTack); clientrequest()bounds this via timeoutapps/mobilestill uses socket.io — must follow up before cutoverdocs/superpowers/specs/2026-08-14-socketio-to-ws-migration-design.md