Authenticated Real-Time WebSocket Streaming with Resumable Event Replay

[robertocarlous]

Problem Statement

Every user-facing update is pull-based: the client polls /api/portfolio, /api/transactions, and /api/analytics. That means either stale dashboards or chatty, rate-limit-burning polling. The existing webhook system (src/services/webhookDispatcher.ts) solves this for operator-configured endpoints, but end users need a first-party, authenticated, real-time channel with no message loss across reconnects. The platform already produces a rich, ordered event stream (transactions, agent actions, portfolio changes, alert triggers, strategy notifications) — this issue exposes it to the user over an authenticated WebSocket with resumable, sequence-numbered replay and clean scaling.

Current State

• src/stellar/events.ts persists processed on-chain events with a cursor; src/services/webhookDispatcher.ts fan-outs typed events (WEBHOOK_EVENTS in src/validators/webhook-validators.ts) to external URLs with retry/DLQ. src/agent/loop.ts and src/jobs/alertRules.ts emit domain events through the same dispatcher.
• src/middleware/authenticate.ts provides JWT auth (req.auth); src/config/redis.ts configures ioredis. There is no WebSocket server today.
• docs/STRATEGY_MARKETPLACE.md and src/jobs/alertRules.ts already establish the notion of per-user events (alert_rule.triggered payload carries userId), and sub-accounts (actingAsUserId) establish the notion of acting-on-behalf-of.

Proposed Solution

1. Authenticated WebSocket transport

• New WebSocket endpoint mounted on the existing Express app (e.g. /api/v1/ws), using the same JWT verification as the REST middleware. Reject unauthenticated handshakes with a close frame before any data is sent; never fall back to an anonymous channel.
• Connections are per-user, and may be ?actor=-scoped so a parent can subscribe to a child sub-account's stream (actingAsUserId semantics — the same enforcement the REST routes use, enforced server-side, not client-side).
• Heartbeat (ping/pong) with disconnect timeout; documented idle policy. Rate-limit handshakes and message rates (reuse the rate-limit philosophy from src/middleware/rateLimiter.ts; a connection flood is a DoS vector).

2. Sequence-numbered events with resumable replay

• Every event pushed to a user is stamped with a monotonic per-user sequence number (seq) and a stream topic (portfolio, transactions, agent, alerts, strategies).
• Client sends { type: 'subscribe', topics: [...] } and, on reconnect, { type: 'resume', topics: [...], afterSeq: N }. The server replays missed events from a durable per-user stream, then live-switches — no gap, no duplicate (a small duplication window with dedup-by-seq on the client is acceptable; document it).
• The per-user stream must be bounded and bounded-duration: a new UserEventStream/UserEvent store (or a Redis list, documented tradeoff) with configurable retention and eviction, so a stale afterSeq is answered with a gap frame containing the newest available seq, not an unbounded replay.
• Ordering contract: within a topic, events are ordered; across topics, no cross-topic ordering is guaranteed (documented). On-chain event ordering inherits ProcessedEvent's ledger ordering.

3. Event emission integration

• Introduce a single publishUserEvent(userId, topic, type, payload) that both the WebSocket fan-out and the existing webhook dispatcher call from a common place — so adding a domain event emits to both channels without bespoke wiring. Map existing domains onto topics:
  • transactions: deposit/withdraw/settlement confirmations (from src/stellar/events.ts).
  • portfolio: value/position changes (from src/agent/loop.ts, snapshots, rebalance decisions).
  • alerts: alert_rule.triggered (from src/jobs/alertRules.ts).
  • strategies: publish/unpublish/material-config-change/follow events (from src/strategy/service.ts).
• Sub-account fan-out: an event for a child user must be deliverable to the parent's connection when the parent holds the right scope — reuse the existing SubAccountPermission model and enforce the same rules as the REST routes.

4. Backpressure, scaling, and lifecycle

• Backpressure: if a subscriber's socket is slow, buffer within a per-connection bound then drop-with-seq-marker (client resumes via resume afterSeq) — never unbounded memory growth, never crash the process.
• Multi-instance: when scaled horizontally, connections live on arbitrary pods. Use Redis pub/sub (ioredis is already configured) to bridge emits to the pod holding the socket; document the at-most-once/loss window across pod death and how resume closes it.
• Graceful shutdown: close sockets with a draining frame before SIGTERM (integrate with src/index.ts graceful-shutdown ordering; connection cleanup must be awaited).
• Observability: track connected users, message rates, replay volume, and gaps via Prometheus metrics (consistent with src/utils/metrics.ts).

5. API/docs

• docs/openapi.yaml documents the WebSocket subprotocol (message schema, topics, seq/resume semantics, error frames) — the handshake itself is not a REST route but must be spec'd for clients.
• Document the client contract: auth flow, subscribe/resume messages, heartbeat, gap handling, and reconnect policy.

Edge Cases & Failure Modes

• Reconnect with stale afterSeq (older than retention): return gap with current seq and a flag so the client can decide to fetch a REST snapshot and resubscribe.
• User deleted / account deactivated mid-connection: server must close the socket with an auth-revoked frame and stop fan-out.
• Duplicate delivery: at-least-once with client dedup-by-seq; document the window.
• Event storm (a rebalance affecting many positions): batching/coalescing per topic within a tick — documented policy, no unbounded per-user queue.
• Subscription to a topic the user has no permission for: reject at subscribe time with an error frame (mirrors REST 403 semantics).
• Redis down: emits must not drop silently — fall back to a bounded in-process per-pod broadcast for the current connections, with an alert (matching alertingService dedupe patterns).

Security & Privacy Considerations

• JWT verification at handshake; tokens must be checked for expiry and the session's current validity (a revoked session must kill live sockets — coordinate with Session/refresh-token revocation).
• Event payloads to the client must go through the same allowlist/redaction discipline as REST responses — never leak userId, keys, or internal fields into a payload just because it's a socket. Reuse the response-mapper allowlists from src/utils/api-formatters.ts.
• Topic access for parent/child must be enforced at publish time (the publisher checks scope), not trusted from the subscriber.
• No secrets travel over the socket; the connection carries the JWT only.

Out of Scope

• Bidirectional trading messages (e.g. place order over WS) — receive side is server→client only in v1.
• Compression/tracing beyond what the stack already provides (gzip upgrade is optional, keep it documented if added).

Suggested Implementation Plan

1. Durable per-user event store (schema + retention) or Redis-stream design decision, with the choice justified in code comments.
2. WebSocket server with JWT handshake, subscribe/resume/gap protocol, heartbeat, per-connection bounds.
3. publishUserEvent unifying socket + webhook emission; migrate domain sites onto it.
4. Redis pub/sub bridge for multi-instance; backpressure with drop-with-marker.
5. Metrics, graceful shutdown, docs/openapi.yaml subprotocol spec, integration tests (a real socket client driving subscribe → event → resume-replay → gap).

Acceptance Criteria

• Authenticated handshake; unauthenticated and revoked-session connections rejected; permission-scoped topic subscription enforced server-side
• Monotonic per-user seq; resume afterSeq replays without gaps or unbounded replay; stale afterSeq returns gap + current seq
• Event emission unified through one publishUserEvent used by both the socket and webhook paths
• Sub-account topics delivered to permitted parents with actingAsUserId semantics
• Backpressure bounds a slow consumer without unbounded memory; drop-with-marker is resumable
• Multi-instance delivery via Redis pub/sub works with a documented loss window closed by resume
• Metrics for connections/messages/replay/gaps; sockets drain on graceful shutdown
• Subprotocol documented in docs/openapi.yaml; integration tests green

[grantfox-oss]

🦊 GrantFox — @coderDom-x has been assigned to this issue as part of the Third Campaign campaign!

Next steps:

1. Open a Pull Request referencing this issue (e.g., Closes #316)
2. Your PR will be reviewed by the Neurowealth maintainers

Good luck! Track your progress on GrantFox.