wallet tracking backend support

backend/src/api/bitcoin/bitcoin-api-abstract-factory.ts

@@ -29,6 +29,7 @@ export interface AbstractBitcoinApi {
   $getBatchedOutspendsInternal(txId: string[]): Promise<IEsploraApi.Outspend[][]>;
   $getOutSpendsByOutpoint(outpoints: { txid: string, vout: number }[]): Promise<IEsploraApi.Outspend[]>;
   $getCoinbaseTx(blockhash: string): Promise<IEsploraApi.Transaction>;
+  $getAddressTransactionSummary(address: string): Promise<IEsploraApi.AddressTxSummary[]>;
 
   startHealthChecks(): void;
   getHealthStatus(): HealthCheckHost[];

backend/src/api/bitcoin/bitcoin-api.ts

@@ -251,6 +251,10 @@ class BitcoinApi implements AbstractBitcoinApi {
     return this.$getRawTransaction(txids[0]);
   }
 
+  async $getAddressTransactionSummary(address: string): Promise<IEsploraApi.AddressTxSummary[]> {
+    throw new Error('Method getAddressTransactionSummary not supported by the Bitcoin RPC API.');
+  }
+
   $getEstimatedHashrate(blockHeight: number): Promise<number> {
     // 120 is the default block span in Core
     return this.bitcoindClient.getNetworkHashPs(120, blockHeight);

backend/src/api/bitcoin/esplora-api.interface.ts

@@ -179,4 +179,11 @@ export namespace IEsploraApi {
     burn_count: number;
   }
 
+  export interface AddressTxSummary {
+    txid: string;
+    value: number;
+    height: number;
+    time: number;
+    tx_position?: number;
+  }
 }

backend/src/api/bitcoin/esplora-api.ts

@@ -357,6 +357,10 @@ class ElectrsApi implements AbstractBitcoinApi {
     return this.failoverRouter.$get<IEsploraApi.Transaction>('/tx/' + txid);
   }
 
+  async $getAddressTransactionSummary(address: string): Promise<IEsploraApi.AddressTxSummary[]> {
+    return this.failoverRouter.$get<IEsploraApi.AddressTxSummary[]>('/address/' + address + '/txs/summary');
+  }
+
   public startHealthChecks(): void {
     this.failoverRouter.startHealthChecks();
   }

backend/src/api/services/services-routes.ts

@@ -0,0 +1,26 @@
+import { Application, Request, Response } from 'express';
+import config from '../../config';
+import WalletApi from './wallets';
+
+class ServicesRoutes {
+  public initRoutes(app: Application): void {
+    app
+      .get(config.MEMPOOL.API_URL_PREFIX + 'wallet/:walletId', this.$getWallet)
+    ;
+  }
+
+  private async $getWallet(req: Request, res: Response): Promise<void> {
+    try {
+      res.header('Pragma', 'public');
+      res.header('Cache-control', 'public');
+      res.setHeader('Expires', new Date(Date.now() + 1000 * 5).toUTCString());
+      const walletId = req.params.walletId;
+      const wallet = await WalletApi.getWallet(walletId);
+      res.status(200).send(wallet);
+    } catch (e) {
+      res.status(500).send(e instanceof Error ? e.message : e);
+    }
+  }
+}
+
+export default new ServicesRoutes();

backend/src/api/services/wallets.ts

@@ -0,0 +1,131 @@
+import config from '../../config';
+import logger from '../../logger';
+import { IEsploraApi } from '../bitcoin/esplora-api.interface';
+import bitcoinApi from '../bitcoin/bitcoin-api-factory';
+import axios from 'axios';
+import { TransactionExtended } from '../../mempool.interfaces';
+
+interface WalletAddress {
+  address: string;
+  active: boolean;
+  transactions?: IEsploraApi.AddressTxSummary[];
+}
+
+interface WalletConfig {
+  url: string;
+  name: string;
+  apiKey: string;
+}
+
+interface Wallet extends WalletConfig {
+  addresses: Record<string, WalletAddress>;
+  lastPoll: number;
+}
+
+const POLL_FREQUENCY = 60 * 60 * 1000; // 1 hour
+
+class WalletApi {
+  private wallets: Record<string, Wallet> = {};
+  private syncing = false;
+
+  constructor() {
+    this.wallets = (config.WALLETS.WALLETS as WalletConfig[]).reduce((acc, wallet) => {
+      acc[wallet.name] = { ...wallet, addresses: {}, lastPoll: 0 };
+      return acc;
+    }, {} as Record<string, Wallet>);
+  }
+
+  public getWallet(wallet: string): Record<string, WalletAddress> {
+    return this.wallets?.[wallet]?.addresses || {};
+  }
+
+  // resync wallet addresses from the provided API
+  async $syncWallets(): Promise<void> {
+    this.syncing = true;
+    for (const walletKey of Object.keys(this.wallets)) {
+      const wallet = this.wallets[walletKey];
+      if (wallet.lastPoll < (Date.now() - POLL_FREQUENCY)) {
+        try {
+          const response = await axios.get(`${wallet.url}/${wallet.name}`, { headers: { 'Authorization': `${wallet.apiKey}` } });
+          const data: { walletBalances: WalletAddress[] } = response.data;
+          const addresses = data.walletBalances;
+          const newAddresses: Record<string, boolean> = {};
+          // sync all current addresses
+          for (const address of addresses) {
+            await this.$syncWalletAddress(wallet, address);
+            newAddresses[address.address] = true;
+          }
+          // remove old addresses
+          for (const address of Object.keys(wallet.addresses)) {
+            if (!newAddresses[address]) {
+              delete wallet.addresses[address];
+            }
+          }
+          wallet.lastPoll = Date.now();
+          logger.debug(`Synced ${Object.keys(wallet.addresses).length} addresses for wallet ${wallet.name}`);
+        } catch (e) {
+          logger.err(`Error syncing wallet ${wallet.name}: ${(e instanceof Error ? e.message : e)}`);
+        }
+      }
+    }
+    this.syncing = false;
+  }
+
+  // resync address transactions from esplora
+  async $syncWalletAddress(wallet: Wallet, address: WalletAddress): Promise<void> {
+    // fetch full transaction data if the address is new or still active
+    const refreshTransactions = !wallet.addresses[address.address] || address.active;
+    if (refreshTransactions) {
+      try {
+        const walletAddress: WalletAddress = {
+          address: address.address,
+          active: address.active,
+          transactions: await bitcoinApi.$getAddressTransactionSummary(address.address),
+        };
+        logger.debug(`Synced ${walletAddress.transactions?.length || 0} transactions for wallet ${wallet.name} address ${address.address}`);
+        wallet.addresses[address.address] = walletAddress;
+      } catch (e) {
+        logger.err(`Error syncing wallet address ${address.address}: ${(e instanceof Error ? e.message : e)}`);
+      }
+    }
+  }
+
+  // check a new block for transactions that affect wallet address balances, and add relevant transactions to wallets
+  processBlock(block: IEsploraApi.Block, blockTxs: TransactionExtended[]): Record<string, Record<string, IEsploraApi.AddressTxSummary[]>> {
+    const walletTransactions: Record<string, Record<string, IEsploraApi.AddressTxSummary[]>> = {};
+    for (const walletKey of Object.keys(this.wallets)) {
+      const wallet = this.wallets[walletKey];
+      walletTransactions[walletKey] = {};
+      for (const tx of blockTxs) {
+        const funded: Record<string, number> = {};
+        const spent: Record<string, number> = {};
+        for (const vin of tx.vin) {
+          const address = vin.prevout?.scriptpubkey_address;
+          if (address && wallet.addresses[address]) {
+            spent[address] = (spent[address] ?? 0) + (vin.prevout?.value ?? 0);
+          }
+        }
+        for (const vout of tx.vout) {
+          const address = vout.scriptpubkey_address;
+          if (address && wallet.addresses[address]) {
+            funded[address] = (funded[address] ?? 0) + (vout.value ?? 0);
+          }
+        }
+        for (const address of Object.keys({ ...funded, ...spent })) {
+          if (!walletTransactions[walletKey][address]) {
+            walletTransactions[walletKey][address] = [];
+          }
+          walletTransactions[walletKey][address].push({
+            txid: tx.txid,
+            value: (funded[address] ?? 0) - (spent[address] ?? 0),
+            height: block.height,
+            time: block.timestamp,
+          });
+        }
+      }
+    }
+    return walletTransactions;
+  }
+}
+
+export default new WalletApi();

backend/src/api/websocket-handler.ts

@@ -26,6 +26,7 @@ import mempool from './mempool';
 import statistics from './statistics/statistics';
 import accelerationRepository from '../repositories/AccelerationRepository';
 import bitcoinApi from './bitcoin/bitcoin-api-factory';
+import walletApi from './services/wallets';
 
 interface AddressTransactions {
   mempool: MempoolTransactionExtended[],
@@ -305,6 +306,14 @@ class WebsocketHandler {
             }
           }
 
+          if (parsedMessage && parsedMessage['track-wallet']) {
+            if (parsedMessage['track-wallet'] === 'stop') {
+              client['track-wallet'] = null;
+            } else {
+              client['track-wallet'] = parsedMessage['track-wallet'];
+            }
+          }
+
           if (parsedMessage && parsedMessage['track-asset']) {
             if (/^[a-fA-F0-9]{64}$/.test(parsedMessage['track-asset'])) {
               client['track-asset'] = parsedMessage['track-asset'];
@@ -1085,6 +1094,9 @@ class WebsocketHandler {
       replaced: replacedTransactions,
     };
 
+    // check for wallet transactions
+    const walletTransactions = config.WALLETS.ENABLED ? walletApi.processBlock(block, transactions) : [];
+
     const responseCache = { ...this.socketData };
     function getCachedResponse(key, data): string {
       if (!responseCache[key]) {
@@ -1287,6 +1299,11 @@ class WebsocketHandler {
         response['mempool-transactions'] = getCachedResponse('mempool-transactions', mempoolDelta);
       }
 
+      if (client['track-wallet']) {
+        const trackedWallet = client['track-wallet'];
+        response['wallet-transactions'] = getCachedResponse(`wallet-transactions-${trackedWallet}`, walletTransactions[trackedWallet] ?? {});
+      }
+
       if (Object.keys(response).length) {
         client.send(this.serializeResponse(response));
       }