Add 'mempool delta' websocket subscriptions

backend/src/api/websocket-handler.ts

@@ -3,6 +3,7 @@ import * as WebSocket from 'ws';
 import {
   BlockExtended, TransactionExtended, MempoolTransactionExtended, WebsocketResponse,
   OptimizedStatistic, ILoadingIndicators, GbtCandidates, TxTrackingInfo,
+  MempoolBlockDelta, MempoolDelta, MempoolDeltaTxids
 } from '../mempool.interfaces';
 import blocks from './blocks';
 import memPool from './mempool';
@@ -364,6 +365,18 @@ class WebsocketHandler {
             client['track-donation'] = parsedMessage['track-donation'];
           }
 
+          if (parsedMessage['track-mempool-txids'] === true) {
+            client['track-mempool-txids'] = true;
+          } else if (parsedMessage['track-mempool-txids'] === false) {
+            delete client['track-mempool-txids'];
+          }
+
+          if (parsedMessage['track-mempool'] === true) {
+            client['track-mempool'] = true;
+          } else if (parsedMessage['track-mempool'] === false) {
+            delete client['track-mempool'];
+          }
+
           if (Object.keys(response).length) {
             client.send(this.serializeResponse(response));
           }
@@ -545,6 +558,27 @@ class WebsocketHandler {
 
     const latestTransactions = memPool.getLatestTransactions();
 
+    const replacedTransactions: { replaced: string, by: TransactionExtended }[] = [];
+    for (const tx of newTransactions) {
+      if (rbfTransactions[tx.txid]) {
+        for (const replaced of rbfTransactions[tx.txid]) {
+          replacedTransactions.push({ replaced: replaced.txid, by: tx });
+        }
+      }
+    }
+    const mempoolDeltaTxids: MempoolDeltaTxids = {
+      added: newTransactions.map(tx => tx.txid),
+      removed: deletedTransactions.map(tx => tx.txid),
+      mined: [],
+      replaced: replacedTransactions.map(replacement => ({ replaced: replacement.replaced, by: replacement.by.txid })),
+    };
+    const mempoolDelta: MempoolDelta = {
+      added: newTransactions,
+      removed: deletedTransactions.map(tx => tx.txid),
+      mined: [],
+      replaced: replacedTransactions,
+    };
+
     // update init data
     const socketDataFields = {
       'mempoolInfo': mempoolInfo,
@@ -847,6 +881,14 @@ class WebsocketHandler {
         response['rbfLatestSummary'] = getCachedResponse('rbfLatestSummary', rbfSummary);
       }
 
+      if (client['track-mempool-txids']) {
+        response['mempool-txids'] = getCachedResponse('mempool-txids', mempoolDeltaTxids);
+      }
+
+      if (client['track-mempool']) {
+        response['mempool-transactions'] = getCachedResponse('mempool-transactions', mempoolDelta);
+      }
+
       if (Object.keys(response).length) {
         client.send(this.serializeResponse(response));
       }
@@ -992,6 +1034,25 @@ class WebsocketHandler {
 
     const mBlocksWithTransactions = mempoolBlocks.getMempoolBlocksWithTransactions();
 
+    const replacedTransactions: { replaced: string, by: TransactionExtended }[] = [];
+    for (const txid of Object.keys(rbfTransactions)) {
+      for (const replaced of rbfTransactions[txid].replaced) {
+        replacedTransactions.push({ replaced: replaced.txid, by: rbfTransactions[txid].replacedBy });
+      }
+    }
+    const mempoolDeltaTxids: MempoolDeltaTxids = {
+      added: [],
+      removed: [],
+      mined: transactions.map(tx => tx.txid),
+      replaced: replacedTransactions.map(replacement => ({ replaced: replacement.replaced, by: replacement.by.txid })),
+    };
+    const mempoolDelta: MempoolDelta = {
+      added: [],
+      removed: [],
+      mined: transactions.map(tx => tx.txid),
+      replaced: replacedTransactions,
+    };
+
     const responseCache = { ...this.socketData };
     function getCachedResponse(key, data): string {
       if (!responseCache[key]) {
@@ -1185,6 +1246,14 @@ class WebsocketHandler {
         }
       }
 
+      if (client['track-mempool-txids']) {
+        response['mempool-txids'] = getCachedResponse('mempool-txids', mempoolDeltaTxids);
+      }
+
+      if (client['track-mempool']) {
+        response['mempool-transactions'] = getCachedResponse('mempool-transactions', mempoolDelta);
+      }
+
       if (Object.keys(response).length) {
         client.send(this.serializeResponse(response));
       }

backend/src/mempool.interfaces.ts

@@ -71,6 +71,20 @@ export interface MempoolBlockDelta {
   changed: MempoolDeltaChange[];
 }
 
+export interface MempoolDeltaTxids {
+  added: string[];
+  removed: string[];
+  mined: string[];
+  replaced: { replaced: string, by: string }[];
+}
+
+export interface MempoolDelta {
+  added: MempoolTransactionExtended[];
+  removed: string[];
+  mined: string[];
+  replaced: { replaced: string, by: TransactionExtended }[];
+}
+
 interface VinStrippedToScriptsig {
   scriptsig: string;
 }