Skip to content

docs: add architecture and implementation documentation for SSE integ… #1354

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
K1NGD4VID merged 1 commit into from
Aug 30, 2026
Merged

docs: add architecture and implementation documentation for SSE integ… #1354

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 27 additions & 18 deletions backend/docs/SSE_ARCHITECTURE.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,27 +3,27 @@
## System Flow

```
┌─────────────────┐
│ Blockchain │
│ Indexer │
│ (Stellar) │
└────────┬────────┘
│ Events
▼
┌─────────────────────────────────────────────────────────┐
│ Stellar Blockchain / Soroban │
│ - On-chain stream contract executions & ledger events │
└────────────────────────────┬────────────────────────────┘
│ On-Chain Events (via Soroban RPC poll)
▼
┌─────────────────────────────────────────────────────────┐
│ Backend Server │
│ │
│ ┌──────────────────────────────────────────────────┐ │
│ │ Stream Controller │ │
│ │ - Creates/updates streams │ │
│ │ - Calls sseService.broadcast() │ │
│ │ Soroban Event Worker (Indexer) │ │
│ │ - Polls Soroban RPC for confirmed events │ │
│ │ - Persists stream state & events to Database │ │
│ │ - Calls sseService.broadcastToStream/Admin() │ │
│ └──────────────┬───────────────────────────────────┘ │
│ │ │
│ │ Dispatch events (indexer-driven) │
│ ▼ │
│ ┌──────────────────────────────────────────────────┐ │
│ │ SSE Service │ │
│ │ - Manages client connections │ │
│ │ - Filters by subscription │ │
│ │ - Filters by subscription (stream/user/admin) │ │
│ │ - Broadcasts to matching clients │ │
│ └──────────────┬───────────────────────────────────┘ │
│ │ │
Expand All @@ -40,6 +40,8 @@
└─────────────────────────────────────┘
```

> **Note on Indexer-Driven Event Origin**: SSE broadcast events originate asynchronously from the background indexer worker (`SorobanEventWorker`) only after transaction confirmation on the Stellar ledger, not synchronously from HTTP API controllers (`stream.controller.ts`, etc.). When a user submits an action (create, pause, withdraw, top-up, cancel), API controllers do not broadcast SSE events directly; events are dispatched once the Soroban event is polled and confirmed on-chain. Additionally, background workers like `StreamRunwayWorker` may dispatch computed alerts (e.g. `STREAM_LOW_BALANCE`).

## Connection Flow

```
Expand Down Expand Up @@ -72,7 +74,7 @@ Client Server
│ [Auto Reconnect - 1s] │
│ │
│ GET /events/subscribe │
├──────────────────────────────>│
│ ├──────────────────────────────>│
│ │
│ 200 OK │
│<──────────────────────────────┤
Expand Down Expand Up @@ -122,30 +124,37 @@ Client Server
└─────────────────┘

Flow:
1. Backend 1 receives stream creation
1. Backend 1 (SorobanEventWorker) indexes confirmed on-chain event
2. Backend 1 publishes to Redis: "stream-events"
3. All backends (1, 2, 3) receive message
3. All backends (1, 2, 3) receive message via Redis subscriber
4. Each backend broadcasts to its connected clients
5. Total: 33 clients receive the event
```

## Event Broadcasting Logic

```typescript
// Broadcast to specific stream
// Broadcast to specific stream (called by SorobanEventWorker / StreamRunwayWorker)
sseService.broadcastToStream("123", "stream.created", data)
↓
Filter clients: subscription includes "123" or "*"
↓
Send to matching clients

// Broadcast to user
sseService.broadcastToUser("GABC...", "stream.created", data)
// Broadcast to user (called by StreamRunwayWorker)
sseService.broadcastToUser("GABC...", "STREAM_LOW_BALANCE", data)
↓
Filter clients: subscription includes "user:GABC..." or "*"
↓
Send to matching clients

// Broadcast to admin (called by SorobanEventWorker for admin/fee events)
sseService.broadcastToAdmin("stream.fee_config_updated", data)
↓
Filter clients: admin subscribers ("admin" or "*")
↓
Send to matching clients

// Broadcast to all
sseService.broadcast("stream.created", data)
↓
Expand Down
2 changes: 1 addition & 1 deletion backend/docs/SSE_IMPLEMENTATION.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -193,7 +193,7 @@ import Redis from 'ioredis';
const redis = new Redis(process.env.REDIS_URL);
const subscriber = new Redis(process.env.REDIS_URL);

// Publisher (in stream controller)
// Publisher (in Soroban event worker / indexer)
redis.publish('stream-events', JSON.stringify({
event: 'stream.created',
data: mockStream,
Expand Down
Loading