arch: implement real-time trade broadcasting via WebSockets and EventEmitter (#159)

Author: JerryIdoko

package-lock.json

@@ -27,7 +27,8 @@
     "prisma:push": "prisma db push",
     "prisma:migrate": "prisma migrate dev",
     "prisma:studio": "prisma studio",
-    "start:indexer": "node services/eventIndexer.js"
+    "start:indexer": "node services/eventIndexer.js",
+    "start:ws": "node server.js"
   },
   "dependencies": {
     "@nestjs/common": "^10.4.22",
@@ -48,7 +49,8 @@
     "reflect-metadata": "^0.2.2",
     "rxjs": "^7.8.2",
     "swagger-ui-express": "^5.0.0",
-    "typeorm": "^0.3.28"
+    "typeorm": "^0.3.28",
+    "ws": "^8.20.0"
   },
   "devDependencies": {
     "@nestjs/cli": "^11.0.16",

package.json

@@ -0,0 +1,56 @@
+const http = require('http');
+const WebSocket = require('ws');
+const eventEmitter = require('./services/eventEmitter');
+
+/**
+ * TradeFlow WebSocket Server
+ * Broadcasts real-time Soroban ledger updates to connected clients
+ */
+const server = http.createServer();
+const wss = new WebSocket.Server({ server });
+
+// Track connected clients
+let clients = 0;
+
+wss.on('connection', (ws) => {
+    clients++;
+    console.log(`[WS] New client connected. Active: ${clients}`);
+    
+    // Welcome message
+    ws.send(JSON.stringify({ 
+        type: 'SUBSCRIPTION_SUCCESS', 
+        message: 'Connected to TradeFlow Real-time Data Pipeline' 
+    }));
+
+    ws.on('close', () => {
+        clients--;
+        console.log(`[WS] Client disconnected. Active: ${clients}`);
+    });
+});
+
+/**
+ * Handle new trade events from the Soroban Indexer
+ */
+eventEmitter.on('newTrade', (trade) => {
+    console.log(`[WS] Incoming trade from Indexer. Broadcasting to ${wss.clients.size} clients...`);
+    
+    const payload = JSON.stringify({
+        topic: 'TRADE_UPDATE',
+        data: trade,
+        timestamp: new Date().toISOString()
+    });
+
+    wss.clients.forEach((client) => {
+        if (client.readyState === WebSocket.OPEN) {
+            client.send(payload);
+        }
+    });
+});
+
+const WS_PORT = process.env.WS_PORT || 8080;
+
+server.listen(WS_PORT, () => {
+    console.log('==============================================');
+    console.log(`🚀 TradeFlow WebSocket Server active on port ${WS_PORT}`);
+    console.log('==============================================');
+});

server.js

@@ -0,0 +1,9 @@
+const EventEmitter = require('events');
+
+/**
+ * Global Event Emitter for internal service communication
+ * This allows the eventIndexer to communicate with the WebSocket server
+ */
+const tradeFlowEvents = new EventEmitter();
+
+module.exports = tradeFlowEvents;

services/eventEmitter.js

@@ -1,5 +1,6 @@
 const { rpc, xdr, scValToNative } = require('@stellar/stellar-sdk');
 const { PrismaClient } = require('@prisma/client');
+const eventEmitter = require('./eventEmitter');
 
 // Initialize Prisma Client
 const prisma = new PrismaClient();
@@ -164,6 +165,12 @@ async function saveTradeToDb(tradeData) {
     });
 
     console.log(`✅ Trade indexed successfully in DB: ${tradeData.userAddress.substring(0, 8)}...`);
+    
+    // Emit trade event for WebSocket broadcasting
+    eventEmitter.emit('newTrade', {
+        ...tradeData,
+        poolAddress: POOL_ADDRESS
+    });
   } catch (error) {
     console.error('Database insertion error:', error);
   }