feat(ws): WebSocket endpoint for real-time price streaming

package.json

@@ -26,6 +26,7 @@
     "swagger-ui-express": "^5.0.1",
     "winston": "^3.14.0",
     "winston-daily-rotate-file": "^5.0.0",
+    "ws": "^8.21.0",
     "yamljs": "^0.3.0",
     "zod": "^4.4.3"
   },

src/index.js

@@ -18,6 +18,8 @@ const webhooksRouter = require('./routes/webhooks');
 const airdropsRouter = require('./routes/airdrops');
 const apiDocsRouter = require('./routes/apiDocs');
 
+const priceWebSocket = require('./ws/priceWebSocket');
+
 const app = express();
 let server;
 
@@ -52,6 +54,7 @@ function shutdown(signal) {
     logger.info(`${signal} received, shutting down`);
     priceRefreshJob.stop();
     webhookRetryWorker.stop();
+    require('./ws/PriceSubscriptionManager').stopHeartbeat();
     if (server) server.close();
     await cache.disconnect();
     process.exit(0);
@@ -61,6 +64,7 @@ function shutdown(signal) {
 if (require.main === module) {
   server = app.listen(config.port, () => {
     logger.info(`SmartDrop backend running on port ${config.port}`);
+    priceWebSocket.attach(server);
     priceRefreshJob.start();
     webhookRetryWorker.start();
   });

src/jobs/priceRefresh.js

@@ -1,6 +1,7 @@
 const cron = require('node-cron');
 const priceOracle = require('../services/priceOracle');
 const alertsService = require('../services/alerts');
+const subscriptionManager = require('../ws/PriceSubscriptionManager');
 const config = require('../config');
 const logger = require('../logger');
 
@@ -13,8 +14,11 @@ function start() {
   scheduledTask = cron.schedule(cronExpression, async () => {
     try {
       logger.info('Starting scheduled price refresh');
-      await priceOracle.refreshAllCachedPrices();
+      const freshPrices = await priceOracle.refreshAllCachedPrices();
       await alertsService.evaluateAll();
+      if (freshPrices && Object.keys(freshPrices).length > 0) {
+        subscriptionManager.notifyPriceUpdates(freshPrices);
+      }
     } catch (err) {
       logger.error('Scheduled price refresh failed', { error: err.message });
     }

src/services/priceOracle.js

@@ -205,16 +205,22 @@ async function refreshAllCachedPrices() {
     return;
   }
 
+  const freshPrices = {};
+
   const refreshPromises = keys
     .filter((key) => !key.includes(':history:'))
     .map(async (key) => {
       const suffix = key.replace(CACHE_PREFIX, '');
       const parts = suffix.split(':');
       const assetCode = parts[0];
       const issuer = parts.length > 1 ? parts[1] : null;
+      const assetKey = issuer ? `${assetCode}:${issuer}` : assetCode;
 
       try {
-        await fetchFreshPrice(assetCode, issuer);
+        const result = await fetchFreshPrice(assetCode, issuer);
+        if (result && result.price_usd !== null) {
+          freshPrices[assetKey] = { price: result.price_usd, source: result.source };
+        }
         logger.debug('Refreshed price', { assetCode, issuer });
       } catch (err) {
         logger.warn('Price refresh failed', { assetCode, issuer, error: err.message });
@@ -223,6 +229,7 @@ async function refreshAllCachedPrices() {
 
   await Promise.allSettled(refreshPromises);
   logger.info('Price refresh cycle completed', { keysRefreshed: keys.length });
+  return freshPrices;
 }
 
 module.exports = {

src/ws/PriceSubscriptionManager.js

@@ -0,0 +1,174 @@
+'use strict';
+
+const logger = require('../logger');
+
+const MAX_ASSETS_PER_CLIENT = 5;
+const MAX_CONNECTIONS = 100;
+const PING_INTERVAL_MS = 30_000;
+const MAX_MISSED_PINGS = 3;
+const PRICE_CHANGE_THRESHOLD_PCT = 0.1;
+
+// Prometheus gauge — updated whenever a socket connects or disconnects.
+let wsConnectionsGauge = null;
+try {
+  const prom = require('prom-client');
+  wsConnectionsGauge = new prom.Gauge({
+    name: 'ws_connections_current',
+    help: 'Number of currently active WebSocket connections',
+  });
+} catch {
+  // prom-client not installed; gauge is a no-op.
+}
+
+function updateGauge(delta) {
+  if (wsConnectionsGauge) wsConnectionsGauge.inc(delta);
+}
+
+/**
+ * Tracks WebSocket subscriptions and delivers price-change pushes.
+ *
+ * Each socket entry:
+ *   { ws, assets: Set<string>, missedPings: number }
+ */
+class PriceSubscriptionManager {
+  constructor() {
+    this._clients = new Map(); // ws → { assets, missedPings }
+    this._previousPrices = new Map(); // assetKey → number
+    this._pingTimer = null;
+  }
+
+  /** Register a new WebSocket connection. Returns false when at capacity. */
+  add(ws) {
+    if (this._clients.size >= MAX_CONNECTIONS) {
+      ws.close(1013, 'Max connections reached');
+      return false;
+    }
+
+    this._clients.set(ws, { assets: new Set(), missedPings: 0 });
+    updateGauge(1);
+    logger.info('WS client connected', { total: this._clients.size });
+
+    ws.on('message', (raw) => this._handleMessage(ws, raw));
+    ws.on('close', () => this._remove(ws));
+    ws.on('error', (err) => {
+      logger.warn('WS client error', { error: err.message });
+      this._remove(ws);
+    });
+
+    return true;
+  }
+
+  _remove(ws) {
+    if (!this._clients.has(ws)) return;
+    this._clients.delete(ws);
+    updateGauge(-1);
+    logger.info('WS client disconnected', { total: this._clients.size });
+  }
+
+  _handleMessage(ws, raw) {
+    let msg;
+    try {
+      msg = JSON.parse(raw.toString());
+    } catch {
+      this._send(ws, { type: 'error', message: 'Invalid JSON' });
+      return;
+    }
+
+    const client = this._clients.get(ws);
+    if (!client) return;
+
+    if (msg.action === 'subscribe') {
+      const requested = Array.isArray(msg.assets) ? msg.assets : [];
+      const allowed = requested.slice(0, MAX_ASSETS_PER_CLIENT);
+      for (const a of allowed) client.assets.add(String(a));
+      this._send(ws, { type: 'subscribed', assets: [...client.assets] });
+
+    } else if (msg.action === 'unsubscribe') {
+      const toRemove = Array.isArray(msg.assets) ? msg.assets : [];
+      for (const a of toRemove) client.assets.delete(String(a));
+      this._send(ws, { type: 'unsubscribed', assets: [...client.assets] });
+
+    } else if (msg.action === 'pong') {
+      client.missedPings = 0;
+
+    } else {
+      this._send(ws, { type: 'error', message: `Unknown action: ${msg.action}` });
+    }
+  }
+
+  _send(ws, payload) {
+    if (ws.readyState !== ws.constructor.OPEN) return;
+    try {
+      ws.send(JSON.stringify(payload));
+    } catch (err) {
+      logger.warn('WS send failed', { error: err.message });
+    }
+  }
+
+  /**
+   * Called after each price refresh cycle with a map of assetKey → newPrice.
+   * Pushes updates to subscribers whose watched asset changed by > 0.1%.
+   */
+  notifyPriceUpdates(freshPrices) {
+    for (const [assetKey, { price, source }] of Object.entries(freshPrices)) {
+      const prev = this._previousPrices.get(assetKey);
+
+      if (prev !== undefined && prev > 0) {
+        const changePct = ((price - prev) / prev) * 100;
+        if (Math.abs(changePct) > PRICE_CHANGE_THRESHOLD_PCT) {
+          const update = {
+            type: 'price_update',
+            asset: assetKey,
+            price_usd: price,
+            previous_price_usd: prev,
+            change_pct: parseFloat(changePct.toFixed(4)),
+            source,
+            timestamp: new Date().toISOString(),
+          };
+          this._broadcast(assetKey, update);
+        }
+      }
+
+      this._previousPrices.set(assetKey, price);
+    }
+  }
+
+  _broadcast(assetKey, payload) {
+    for (const [ws, client] of this._clients) {
+      if (client.assets.has(assetKey)) {
+        this._send(ws, payload);
+      }
+    }
+  }
+
+  /** Start sending heartbeat pings every 30 s; disconnect idle sockets. */
+  startHeartbeat() {
+    if (this._pingTimer) return;
+    this._pingTimer = setInterval(() => {
+      for (const [ws, client] of this._clients) {
+        if (client.missedPings >= MAX_MISSED_PINGS) {
+          logger.info('WS client timed out, disconnecting');
+          ws.terminate();
+          this._remove(ws);
+          continue;
+        }
+        client.missedPings += 1;
+        this._send(ws, { type: 'ping' });
+      }
+    }, PING_INTERVAL_MS);
+  }
+
+  stopHeartbeat() {
+    if (this._pingTimer) {
+      clearInterval(this._pingTimer);
+      this._pingTimer = null;
+    }
+  }
+
+  get connectionCount() {
+    return this._clients.size;
+  }
+}
+
+module.exports = new PriceSubscriptionManager();
+module.exports.PriceSubscriptionManager = PriceSubscriptionManager;

src/ws/priceWebSocket.js

@@ -0,0 +1,29 @@
+'use strict';
+
+const { WebSocketServer } = require('ws');
+const logger = require('../logger');
+const subscriptionManager = require('./PriceSubscriptionManager');
+
+/**
+ * Attach the WebSocket server to an existing HTTP server.
+ * Clients connect at ws://<host>/ws
+ */
+function attach(httpServer) {
+  const wss = new WebSocketServer({ server: httpServer, path: '/ws' });
+
+  wss.on('connection', (ws, req) => {
+    logger.info('Incoming WS connection', { ip: req.socket.remoteAddress });
+    subscriptionManager.add(ws);
+  });
+
+  wss.on('error', (err) => {
+    logger.error('WebSocket server error', { error: err.message });
+  });
+
+  subscriptionManager.startHeartbeat();
+  logger.info('WebSocket price-stream server attached at /ws');
+
+  return wss;
+}
+
+module.exports = { attach };