Data pipeline for projected mempool block overview

backend/src/api/mempool-blocks.ts

@@ -112,6 +112,7 @@ class MempoolBlocks {
       medianFee: Common.percentile(transactions.map((tx) => tx.effectiveFeePerVsize), config.MEMPOOL.RECOMMENDED_FEE_PERCENTILE),
       feeRange: Common.getFeesInRange(transactions, rangeLength),
       transactionIds: transactions.map((tx) => tx.txid),
+      transactions: transactions.map((tx) => Common.stripTransaction(tx)),
     };
   }
 }

backend/src/api/websocket-handler.ts

@@ -111,6 +111,22 @@ class WebsocketHandler {
             }
           }
 
+          if (parsedMessage && parsedMessage['track-mempool-block'] != null) {
+            if (Number.isInteger(parsedMessage['track-mempool-block']) && parsedMessage['track-mempool-block'] >= 0) {
+              const index = parsedMessage['track-mempool-block'];
+              client['track-mempool-block'] = index;
+              const mBlocksWithTransactions = mempoolBlocks.getMempoolBlocksWithTransactions();
+              if (mBlocksWithTransactions[index]) {
+                response['projected-mempool-block'] = {
+                  index: index,
+                  block: mBlocksWithTransactions[index],
+                };
+              }
+            } else {
+              client['track-mempool-block'] = null;
+            }
+          }
+
           if (parsedMessage.action === 'init') {
             const _blocks = blocks.getBlocks().slice(-config.MEMPOOL.INITIAL_BLOCKS_AMOUNT);
             if (!_blocks) {
@@ -233,6 +249,7 @@ class WebsocketHandler {
 
     mempoolBlocks.updateMempoolBlocks(newMempool);
     const mBlocks = mempoolBlocks.getMempoolBlocks();
+    const mBlocksWithTransactions = mempoolBlocks.getMempoolBlocksWithTransactions();
     const mempoolInfo = memPool.getMempoolInfo();
     const vBytesPerSecond = memPool.getVBytesPerSecond();
     const rbfTransactions = Common.findRbfTransactions(newTransactions, deletedTransactions);
@@ -370,6 +387,16 @@ class WebsocketHandler {
         }
       }
 
+      if (client['track-mempool-block'] >= 0) {
+        const index = client['track-mempool-block'];
+        if (mBlocksWithTransactions[index]) {
+          response['projected-mempool-block'] = {
+            index: index,
+            block: mBlocksWithTransactions[index],
+          };
+        }
+      }
+
       if (Object.keys(response).length) {
         client.send(JSON.stringify(response));
       }

backend/src/mempool.interfaces.ts

@@ -33,6 +33,7 @@ export interface MempoolBlock {
 
 export interface MempoolBlockWithTransactions extends MempoolBlock {
   transactionIds: string[];
+  transactions: TransactionStripped[];
 }
 
 interface VinStrippedToScriptsig {

frontend/src/app/components/mempool-block-overview/mempool-block-overview.component.html

@@ -0,0 +1,3 @@
+<div class="mempool-block-overview">
+  <p *ngIf="mempoolBlock$ | async as mempoolBlock">{{ mempoolBlock.transactions.length }}</p>
+</div>

frontend/src/app/components/mempool-block-overview/mempool-block-overview.component.ts

@@ -0,0 +1,46 @@
+import { Component, Input, OnInit, OnDestroy,  OnChanges, ChangeDetectionStrategy } from '@angular/core';
+import { StateService } from 'src/app/services/state.service';
+import { MempoolBlockWithTransactions } from 'src/app/interfaces/websocket.interface';
+import { Observable, Subscription } from 'rxjs';
+import { WebsocketService } from 'src/app/services/websocket.service';
+
+@Component({
+  selector: 'app-mempool-block-overview',
+  templateUrl: './mempool-block-overview.component.html',
+  styleUrls: [],
+  changeDetection: ChangeDetectionStrategy.OnPush,
+})
+export class MempoolBlockOverviewComponent implements OnInit, OnDestroy, OnChanges {
+  @Input() index: number;
+
+  sub: Subscription;
+  mempoolBlock$: Observable<MempoolBlockWithTransactions>;
+
+  constructor(
+    public stateService: StateService,
+    private websocketService: WebsocketService,
+  ) { }
+
+  ngOnInit(): void {
+    this.websocketService.startTrackMempoolBlock(this.index);
+    this.mempoolBlock$ = this.stateService.mempoolBlock$
+    this.sub = this.mempoolBlock$.subscribe((block) => {
+      this.updateBlock(block)
+    })
+  }
+
+  ngOnChanges(changes): void {
+    if (changes.index) {
+      this.websocketService.startTrackMempoolBlock(changes.index.currentValue);
+    }
+  }
+
+  ngOnDestroy(): void {
+    this.sub.unsubscribe();
+    this.websocketService.stopTrackMempoolBlock();
+  }
+
+  updateBlock(block: MempoolBlockWithTransactions): void {
+    
+  }
+}

frontend/src/app/components/mempool-block/mempool-block.component.html

@@ -39,9 +39,10 @@
             </tr>
           </tbody>
         </table>
+        <app-fee-distribution-graph [data]="mempoolBlock.feeRange" ></app-fee-distribution-graph>
       </div>
       <div class="col-md chart-container">
-        <app-fee-distribution-graph [data]="mempoolBlock.feeRange" ></app-fee-distribution-graph>
+        <app-mempool-block-overview [index]="mempoolBlockIndex"></app-mempool-block-overview>
       </div>
     </div>
   </div>

frontend/src/app/graphs/graphs.module.ts

@@ -14,6 +14,7 @@ import { LbtcPegsGraphComponent } from '../components/lbtc-pegs-graph/lbtc-pegs-
 import { GraphsComponent } from '../components/graphs/graphs.component';
 import { StatisticsComponent } from '../components/statistics/statistics.component';
 import { MempoolBlockComponent } from '../components/mempool-block/mempool-block.component';
+import { MempoolBlockOverviewComponent } from '../components/mempool-block-overview/mempool-block-overview.component';
 import { PoolRankingComponent } from '../components/pool-ranking/pool-ranking.component';
 import { PoolComponent } from '../components/pool/pool.component';
 import { TelevisionComponent } from '../components/television/television.component';
@@ -40,6 +41,7 @@ import { CommonModule } from '@angular/common';
     BlockFeeRatesGraphComponent,
     BlockSizesWeightsGraphComponent,
     FeeDistributionGraphComponent,
+    MempoolBlockOverviewComponent,
     IncomingTransactionsGraphComponent,
     MempoolGraphComponent,
     LbtcPegsGraphComponent,

frontend/src/app/interfaces/websocket.interface.ts

@@ -25,6 +25,7 @@ export interface WebsocketResponse {
   'track-tx'?: string;
   'track-address'?: string;
   'track-asset'?: string;
+  'track-mempool-block'?: number;
   'watch-mempool'?: boolean;
   'track-bisq-market'?: string;
 }
@@ -44,6 +45,11 @@ export interface MempoolBlock {
   index: number;
 }
 
+export interface MempoolBlockWithTransactions extends MempoolBlock {
+  transactionIds: string[];
+  transactions: TransactionStripped[];
+}
+
 export interface MempoolInfo {
   loaded: boolean;                 //  (boolean) True if the mempool is fully loaded
   size: number;                    //  (numeric) Current tx count

frontend/src/app/services/state.service.ts

@@ -1,7 +1,7 @@
 import { Inject, Injectable, PLATFORM_ID } from '@angular/core';
 import { ReplaySubject, BehaviorSubject, Subject, fromEvent, Observable } from 'rxjs';
 import { Transaction } from '../interfaces/electrs.interface';
-import { IBackendInfo, MempoolBlock, MempoolInfo, Recommendedfees, ReplacedTransaction, TransactionStripped } from '../interfaces/websocket.interface';
+import { IBackendInfo, MempoolBlock, MempoolBlockWithTransactions, MempoolInfo, Recommendedfees, ReplacedTransaction, TransactionStripped } from '../interfaces/websocket.interface';
 import { BlockExtended, DifficultyAdjustment, OptimizedMempoolStats } from '../interfaces/node-api.interface';
 import { Router, NavigationStart } from '@angular/router';
 import { isPlatformBrowser } from '@angular/common';
@@ -80,6 +80,7 @@ export class StateService {
   bsqPrice$ = new ReplaySubject<number>(1);
   mempoolInfo$ = new ReplaySubject<MempoolInfo>(1);
   mempoolBlocks$ = new ReplaySubject<MempoolBlock[]>(1);
+  mempoolBlock$ = new Subject<MempoolBlockWithTransactions>();
   txReplaced$ = new Subject<ReplacedTransaction>();
   utxoSpent$ = new Subject<object>();
   difficultyAdjustment$ = new ReplaySubject<DifficultyAdjustment>(1);

frontend/src/app/services/websocket.service.ts

@@ -27,6 +27,7 @@ export class WebsocketService {
   private lastWant: string | null = null;
   private isTrackingTx = false;
   private trackingTxId: string;
+  private trackingMempoolBlock: number;
   private latestGitCommit = '';
   private onlineCheckTimeout: number;
   private onlineCheckTimeoutTwo: number;
@@ -157,6 +158,16 @@ export class WebsocketService {
     this.websocketSubject.next({ 'track-asset': 'stop' });
   }
 
+  startTrackMempoolBlock(block: number) {
+    this.websocketSubject.next({ 'track-mempool-block': block });
+    this.trackingMempoolBlock = block
+  }
+
+  stopTrackMempoolBlock() {
+    this.websocketSubject.next({ 'track-mempool-block': -1 });
+    this.trackingMempoolBlock = -1
+  }
+
   startTrackBisqMarket(market: string) {
     this.websocketSubject.next({ 'track-bisq-market': market });
   }
@@ -293,6 +304,12 @@ export class WebsocketService {
       });
     }
 
+    if (response['projected-mempool-block']) {
+      if (response['projected-mempool-block'].index == this.trackingMempoolBlock) {
+        this.stateService.mempoolBlock$.next(response['projected-mempool-block'].block);
+      }
+    }
+
     if (response['live-2h-chart']) {
       this.stateService.live2Chart$.next(response['live-2h-chart']);
     }