Merge pull request #114 from JerryIdoko/feature/websocket-live-streaming-159

Pull Request: Refactor API for Live Ledger Streaming (#159, #83)

Author: AlAfiz

.env.example

@@ -8,7 +8,8 @@ DB_USERNAME=postgres
 DB_PASSWORD=password
 DB_DATABASE=tradeflow
 
-# Soroban Event Indexer Configuration
+# Soroban Event Indexer & WebSocket Configuration
 SOROBAN_RPC_URL="https://soroban-testnet.stellar.org"
 POOL_ADDRESS="CC..." # Replace with your Pool Contract ID
-INDEXER_POLL_INTERVAL=5000 # Polling interval in ms
+INDEXER_POLL_INTERVAL=5000 # Polling interval in ms
+WS_PORT=3001 # WebSocket server port

package.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:all": "node server.js"
   },
   "dependencies": {
     "@nestjs/common": "^10.4.22",

server.js

@@ -0,0 +1,83 @@
+/**
+ * server.js
+ * 
+ * Main Entry Point for combined Soroban Indexer and WebSocket streaming server.
+ * This file orchestrates the real-time data flow from the blockchain to the frontend.
+ */
+
+const WebSocket = require('ws');
+const wsEvents = require('./services/wsEvents');
+const { startIndexer } = require('./services/eventIndexer');
+
+// Config and Port Setup
+const WS_PORT = process.env.WS_PORT || 3001;
+const wss = new WebSocket.Server({ port: WS_PORT });
+
+console.log('--- 🌐 TradeFlow Real-Time Stream Server ---');
+console.log(`📡 WebSocket server running on ws://localhost:${WS_PORT}`);
+
+// Connection tracking
+let activeConnections = 0;
+
+/**
+ * Listen for incoming WebSocket connections.
+ */
+wss.on('connection', (ws) => {
+  activeConnections++;
+  console.log(`✅ New Web3 client connected. Active: ${activeConnections}`);
+  
+  // Initial Connection ACK
+  ws.send(JSON.stringify({ 
+    event: 'INDEXER_CONNECTED', 
+    status: 'ONLINE',
+    timestamp: new Date().toISOString() 
+  }));
+
+  ws.on('close', () => {
+    activeConnections--;
+    console.log(`❌ Web3 client disconnected. Active: ${activeConnections}`);
+  });
+
+  ws.on('error', (err) => {
+    console.error('⚠️ WS Socket Error:', err.message);
+  });
+});
+
+/**
+ * BROADCASTER: Listens to the internal 'newTrade' event emitter.
+ * Broadcasts every new blockchain event caught by the Indexer daemon
+ * to all connected browser clients.
+ */
+wsEvents.on('newTrade', (tradeData) => {
+  console.log(`📣 BROADCASTING: New trade found in pool ${tradeData.poolId.slice(0, 8)}...`);
+  
+  const payload = JSON.stringify({
+    event: 'NEW_TRADE_EVENT',
+    data: tradeData,
+    receivedAt: new Date().toISOString()
+  });
+
+  // Iterative broadcast to all active subscribers
+  wss.clients.forEach((client) => {
+    if (client.readyState === WebSocket.OPEN) {
+      client.send(payload);
+    }
+  });
+});
+
+/**
+ * 🚀 START DAEMON ORCHESTRATION
+ * We start the Soroban Indexer in the same Node.js process to bridge
+ * the blockchain data with the WebSocket emitter via internal memory.
+ */
+startIndexer().catch((err) => {
+  console.error('❌ CRITICAL ERROR: Event Indexer failed to start:', err.message);
+  process.exit(1);
+});
+
+// Process Management
+process.on('SIGTERM', () => {
+  console.log('🛑 Closing WebSocket connections and shutting down.');
+  wss.close();
+  process.exit(0);
+});

services/eventIndexer.js

@@ -9,6 +9,7 @@
 const { rpc } = require('@stellar/stellar-sdk');
 const { PrismaClient } = require('@prisma/client');
 const { parseScVal } = require('./scValParser');
+const wsEvents = require('./wsEvents');
 
 // In case dotenv is not installed as a top-level dependency,
 // we try to load it safely. Most Node.js environments for this project should have it.
@@ -125,6 +126,9 @@ async function handleContractEvent(event) {
       });
 
       console.log(`💾 Indexed Trade saved. DB ID: ${savedTrade.id}`);
+      
+      // Trigger WebSocket broadcast
+      wsEvents.emit('newTrade', savedTrade);
     }
   } catch (error) {
     console.error('❌ Failed to process event:', error.message);
@@ -138,4 +142,9 @@ process.on('SIGINT', async () => {
   process.exit(0);
 });
 
-startIndexer();
+exports.startIndexer = startIndexer;
+
+// In standalone mode, starting the indexer automatically
+if (require.main === module) {
+  startIndexer();
+}

services/wsEvents.js

@@ -0,0 +1,4 @@
+const EventEmitter = require('events');
+const wsEvents = new EventEmitter();
+
+module.exports = wsEvents;