Merge branch 'feature/cherry-picked-to-master' of MQL-Development/market-data-service into master

src/controllers/livePriceController.js

@@ -58,162 +58,140 @@ class LivePriceController {
       });
 
       if (!livePrice) {
-        const error = new Error('Live price not found for this symbol');
-        error.statusCode = 404;
-        return next(error);
+        return res.status(404).json({ success: false, message: 'Live price not found' });
       }
 
-      res.json({
-        success: true,
-        data: livePrice
-      });
+      res.json({ success: true, data: livePrice });
     } catch (error) {
       next(error);
     }
   }
 
-  // Update or create live price
+  // ✅ Upsert single live price (used by EA if sending one at a time)
   async upsertLivePrice(req, res, next) {
     try {
       const { symbolId, price, bid, ask, bidSize, askSize } = req.body;
 
-      // Verify symbol exists and is active
       const symbol = await Symbol.findByPk(symbolId);
-      if (!symbol) {
-        const error = new Error('Symbol not found');
-        error.statusCode = 404;
-        return next(error);
+      if (!symbol || !symbol.isActive) {
+        return res.status(400).json({ success: false, message: 'Invalid or inactive symbol' });
       }
 
-      if (!symbol.isActive) {
-        const error = new Error('Cannot update live price for inactive symbol');
-        error.statusCode = 400;
-        return next(error);
-      }
+      const now = new Date();
 
-      const [livePrice, created] = await LivePrice.upsert({
-        symbolId: parseInt(symbolId),
-        price,
-        bid,
-        ask,
+      // 🔥 Emit immediately — no DB blocking
+      const io = req.app.get('io');
+      const eventData = {
+        symbol: symbol.symbol,
+        symbolId,
+        price: Number(price).toFixed(5),
+        bid: Number(bid).toFixed(5),
+        ask: Number(ask).toFixed(5),
         bidSize,
         askSize,
-        lastUpdated: new Date()
-      });
-
-      // Emit WebSocket event for real-time updates
-      const io = req.app.get('io');
-      if (io) {
-        const eventData = {
-          symbol: symbol.symbol,
-          symbolId: symbol.id,
-          price,
-          bid,
-          ask,
-          bidSize,
-          askSize,
-          lastUpdated: livePrice.lastUpdated,
-          exchange: symbol.exchange,
-          instrumentType: symbol.instrumentType
-        };
-
-        // Emit to all clients subscribed to this symbol
-        io.to(`symbol:${symbol.symbol}`).emit('livePriceUpdate', eventData);
-
-        // Also emit to general live price updates
-        io.emit('livePriceUpdate', eventData);
-      }
-
-      res.status(created ? 201 : 200).json({
-        success: true,
-        data: livePrice,
-        message: created ? 'Live price created successfully' : 'Live price updated successfully'
+        lastUpdated: now,
+        exchange: symbol.exchange,
+        instrumentType: symbol.instrumentType
+      };
+
+      io?.to(`symbol:${symbol.symbol}`).emit('livePriceUpdate', eventData);
+      io?.emit('livePriceUpdate', eventData);
+
+      // Respond immediately
+      res.json({ success: true, message: 'Emitted instantly', data: eventData });
+
+      // Background DB upsert (async)
+      setImmediate(async () => {
+        try {
+          await LivePrice.upsert({
+            symbolId,
+            price,
+            bid,
+            ask,
+            bidSize,
+            askSize,
+            lastUpdated: now
+          });
+        } catch (err) {
+          console.error('[DB ERROR] upsertLivePrice:', err);
+        }
       });
     } catch (error) {
       next(error);
     }
   }
 
-  // Bulk update live prices
+  // ✅ BULK UPDATE (Optimized for speed & instant WebSocket)
   async bulkUpdateLivePrices(req, res, next) {
     try {
       const { prices } = req.body;
-
       if (!Array.isArray(prices) || prices.length === 0) {
-        const error = new Error('Prices array is required');
-        error.statusCode = 400;
-        return next(error);
+        return res.status(400).json({ success: false, message: 'Prices array is required' });
       }
 
-      // Verify all symbols exist and are active
+      const io = req.app.get('io');
+      const now = new Date();
+
+      // Step 1: Fetch all active symbols once
       const symbolIds = prices.map(p => p.symbolId);
-      const existingSymbols = await Symbol.findAll({
-        where: {
-          id: symbolIds,
-          isActive: true
-        },
-        attributes: ['id']
+      const symbols = await Symbol.findAll({
+        where: { id: symbolIds, isActive: true },
+        attributes: ['id', 'symbol', 'exchange', 'instrumentType']
       });
 
-      const existingSymbolIds = existingSymbols.map(s => s.id);
-      const invalidSymbolIds = symbolIds.filter(id => !existingSymbolIds.includes(id));
+      const symbolMap = new Map(symbols.map(s => [s.id, s]));
+
+      // Step 2: Emit immediately (non-blocking)
+      for (const p of prices) {
+        const s = symbolMap.get(p.symbolId);
+        if (!s) continue;
+
+        const payload = {
+          symbol: s.symbol,
+          symbolId: s.id,
+          price: Number(p.price).toFixed(5),
+          bid: Number(p.bid).toFixed(5),
+          ask: Number(p.ask).toFixed(5),
+          bidSize: p.bidSize,
+          askSize: p.askSize,
+          lastUpdated: now,
+          exchange: s.exchange,
+          instrumentType: s.instrumentType
+        };
 
-      if (invalidSymbolIds.length > 0) {
-        const error = new Error(`Invalid or inactive symbol IDs: ${invalidSymbolIds.join(', ')}`);
-        error.statusCode = 400;
-        return next(error);
-      }
-
-      // Prepare data for bulk upsert
-      const upsertData = prices.map(price => ({
-        symbolId: parseInt(price.symbolId),
-        price: price.price,
-        bid: price.bid,
-        ask: price.ask,
-        bidSize: price.bidSize,
-        askSize: price.askSize,
-        lastUpdated: new Date()
-      }));
-
-      const updatedPrices = [];
-      const io = req.app.get('io');
-
-      for (const data of upsertData) {
-        const [livePrice] = await LivePrice.upsert(data);
-        updatedPrices.push(livePrice);
-
-        // Emit WebSocket event for each updated price
-        if (io) {
-          const symbol = await Symbol.findByPk(data.symbolId);
-          if (symbol) {
-            const eventData = {
-              symbol: symbol.symbol,
-              symbolId: symbol.id,
-              price: data.price,
-              bid: data.bid,
-              ask: data.ask,
-              bidSize: data.bidSize,
-              askSize: data.askSize,
-              lastUpdated: data.lastUpdated,
-              exchange: symbol.exchange,
-              instrumentType: symbol.instrumentType
-            };
-
-            // Emit to all clients subscribed to this symbol
-            io.to(`symbol:${symbol.symbol}`).emit('livePriceUpdate', eventData);
-
-            // Also emit to general live price updates
-            io.emit('livePriceUpdate', eventData);
-          }
-        }
+        io?.to(`symbol:${s.symbol}`).emit('livePriceUpdate', payload);
+        io?.emit('livePriceUpdate', payload);
       }
 
+      // Step 3: Respond immediately
       res.json({
         success: true,
-        data: updatedPrices,
-        message: `${updatedPrices.length} live prices updated successfully`
+        message: `${prices.length} live prices emitted instantly`,
+      });
+
+      // Step 4: Async background write (DB non-blocking)
+      setImmediate(async () => {
+        try {
+          await LivePrice.bulkCreate(
+            prices.map(p => ({
+              symbolId: p.symbolId,
+              price: p.price,
+              bid: p.bid,
+              ask: p.ask,
+              bidSize: p.bidSize,
+              askSize: p.askSize,
+              lastUpdated: now
+            })),
+            { updateOnDuplicate: ['price', 'bid', 'ask', 'bidSize', 'askSize', 'lastUpdated'] }
+          );
+
+          console.log(`[DB] ${prices.length} prices written at ${now.toISOString()}`);
+        } catch (err) {
+          console.error('[DB ERROR] bulkUpdateLivePrices:', err);
+        }
       });
     } catch (error) {
+      console.error('[ERROR] bulkUpdateLivePrices:', error);
       next(error);
     }
   }
@@ -293,16 +271,14 @@ class LivePriceController {
         where: { symbolId: parseInt(symbolId) }
       });
 
-      if (deletedRowsCount === 0) {
-        const error = new Error('Live price not found for this symbol');
-        error.statusCode = 404;
-        return next(error);
+      if (!deletedRowsCount) {
+        return res.status(404).json({ success: false, message: 'Live price not found' });
       }
 
-      res.json({
-        success: true,
+      res.json({ 
+        success: true, 
         message: 'Live price deleted successfully'
-      });
+       });
     } catch (error) {
       next(error);
     }

src/server.js

@@ -15,13 +15,20 @@ const startServer = async () => {
     // Create HTTP server
     const server = createServer(app);
 
-    // Initialize Socket.io
+    // Initialize Socket.io - Force WebSocket transport to prevent polling fallback
     const io = new Server(server, {
       cors: {
         origin: process.env.CORS_ORIGIN || "*",
         methods: ["GET", "POST"],
         credentials: true
-      }
+      },
+      // Force WebSocket transport only - no HTTP polling fallback
+      transports: ['websocket'],
+      // Connection management
+      pingTimeout: 60000,  // 60 seconds
+      pingInterval: 25000, // 25 seconds
+      // Allow Engine.IO v3 for compatibility
+      allowEIO3: true
     });
 
     // Socket.io connection handling