Implement a real-time per-wallet event subscription system over SSE with guaranteed at-least-once delivery, cursor-based replay, and per-connection backpressure

[Chucks1093]

Summary

Clients currently poll the transaction history endpoint to see updates. This issue replaces polling with a persistent Server-Sent Events (SSE) stream per wallet: the server pushes every new event matching the wallet's subscriptions in real time, guarantees at-least-once delivery by tracking the last acknowledged cursor per connection, supports reconnection with cursor-based replay so no events are missed during a disconnect, and enforces backpressure so a slow client cannot exhaust server memory.

Scope

1. Subscription registry

• POST /subscriptions accepts a JWT-authenticated request with a topics array (e.g. ['key_buy', 'key_sell', 'follower_added']) and returns a subscriptionId
• Subscriptions are stored in Redis as a sorted set per wallet: subscriptions:{walletAddress} → set of (subscriptionId, topics[])
• A wallet may hold at most 5 concurrent subscriptions; a 6th returns 409 subscription_limit_reached
• Subscriptions expire after 24 hours of inactivity (no connected SSE client); a background worker prunes expired entries every 5 minutes

2. SSE connection and event fan-out

• GET /subscriptions/:subscriptionId/stream upgrades to an SSE connection; requires the same JWT that created the subscription
• On connect, the server registers the connection in an in-process connection map keyed by subscriptionId
• When a new event is ingested by the event pipeline, the fan-out worker looks up all subscriptions whose topics include the event type and pushes the serialised event to every connected SSE client matching the event's wallet
• Fan-out must not block the ingestion pipeline — publish to a per-subscription in-process queue and drain asynchronously
• Each SSE message includes: id: {eventCursor}, event: {topic}, data: {JSON payload}

3. Cursor-based replay and at-least-once delivery

• The eventCursor is an opaque string encoding (walletAddress, topic, createdAt, eventId) — monotonically increasing per wallet
• On reconnect, the client sends Last-Event-ID: {cursor} in the request header; the server replays all events after that cursor from the database before switching to live fan-out
• Store the last acknowledged cursor per subscription in Redis; update it each time an event is successfully written to the SSE stream
• If a client disconnects without acknowledging, events since the last stored cursor are replayed on the next reconnect (at-least-once guarantee)
• Replay is capped at 1000 events; if the gap exceeds this, return an X-Replay-Truncated: true header and start from the oldest available event in the window

4. Backpressure and slow-client protection

• Each SSE connection has an in-process event queue capped at 500 messages
• If the queue reaches 500, stop accepting new events for that connection and emit a warn log sse_queue_full with subscriptionId and queueDepth
• If the queue remains full for more than 30 seconds, forcibly close the connection with SSE comment # connection closed: slow client and mark the subscription as THROTTLED in Redis
• A THROTTLED subscription cannot reconnect for 60 seconds (return 429 with Retry-After)

5. Heartbeat and connection hygiene

• Send an SSE comment (:) heartbeat every 15 seconds to keep the connection alive through proxies
• Track connection count per wallet in Redis; enforce a maximum of 2 concurrent SSE connections per wallet (3rd connection returns 429 too_many_connections)
• On server shutdown, send event: server_closing to all open connections before closing, giving clients time to reconnect to another instance

6. Integration tests

• Subscribe to key_buy events, ingest a buy event, assert the SSE stream receives the event within 200ms
• Disconnect mid-stream and reconnect with the last cursor, assert only missed events are replayed (no duplicates)
• Ingest 1100 events while disconnected, reconnect, assert replay is capped at 1000 and X-Replay-Truncated: true is set
• Saturate the per-connection queue to 500 and hold it for 31 seconds, assert the connection is forcibly closed and the subscription is THROTTLED
• Open 3 concurrent SSE connections for the same wallet, assert the 3rd returns 429
• Attempt to create a 6th subscription for the same wallet, assert 409

Acceptance Criteria

• POST /subscriptions creates a subscription with topic filtering and enforces the 5-subscription limit
• GET /subscriptions/:id/stream opens an SSE stream and delivers matching events in real time
• Fan-out does not block the event ingestion pipeline
• Cursor-based replay delivers all missed events on reconnect with no duplicates
• Replay truncated at 1000 events with X-Replay-Truncated: true header
• Per-connection queue capped at 500; slow clients forcibly closed after 30 seconds
• THROTTLED subscriptions blocked for 60 seconds with 429 and Retry-After
• Maximum 2 concurrent SSE connections per wallet enforced
• Heartbeat comment sent every 15 seconds
• server_closing event sent to all connections on graceful shutdown

ETA: 24 hours

---

Coordinate on Telegram

[salienne]

Can I work on this chief?

[sammycee769]

I’d like to work on this issue by implementing a robust, production-grade SSE streaming system with subscription management, cursor-based replay, and backpressure handling. I will build the Redis-backed subscription registry, real-time fan-out pipeline, replay logic with at-least-once guarantees, and safeguards for slow clients and connection limits. I’ll also add full integration tests to validate delivery timing, replay correctness, throttling, and edge cases. I can complete the implementation within 12–14 hours.