api: connect and subscribe to streaming endpoints

Author: jamaljsr

app/src/api/lnd.ts

@@ -7,7 +7,6 @@ import GrpcClient from './grpc';
 interface LndEvents {
   transaction: LND.Transaction.AsObject;
   channel: LND.ChannelEventUpdate.AsObject;
-  invoice: LND.Invoice.AsObject;
 }
 
 /**
@@ -56,6 +55,24 @@ class LndApi extends BaseApi<LndEvents> {
     const res = await this._grpc.request(Lightning.ListChannels, req, this._meta);
     return res.toObject();
   }
+
+  /**
+   * Connect to the LND streaming endpoints
+   */
+  connectStreams() {
+    this._grpc.subscribe(
+      Lightning.SubscribeTransactions,
+      new LND.GetTransactionsRequest(),
+      transaction => this.emit('transaction', transaction.toObject()),
+      this._meta,
+    );
+    this._grpc.subscribe(
+      Lightning.SubscribeChannelEvents,
+      new LND.ChannelEventSubscription(),
+      channelEvent => this.emit('channel', channelEvent.toObject()),
+      this._meta,
+    );
+  }
 }
 
 export default LndApi;

app/src/api/loop.ts

@@ -108,6 +108,17 @@ class LoopApi extends BaseApi<LoopEvents> {
     return res.toObject();
   }
 
+  /**
+   * Connect to the Loop streaming endpoint
+   */
+  connectStreams() {
+    this._grpc.subscribe(
+      SwapClient.Monitor,
+      new LOOP.MonitorRequest(),
+      swapStatus => this.emit('monitor', swapStatus.toObject()),
+      this._meta,
+    );
+  }
   /**
    * Calculates the max routing fee params for loop out. this mimics the loop cli
    * behavior of using 2% of the amount

app/src/store/models/channel.ts

@@ -13,6 +13,7 @@ export default class Channel {
 
   @observable chanId = '';
   @observable remotePubkey = '';
+  @observable channelPoint = '';
   @observable capacity = Big(0);
   @observable localBalance = Big(0);
   @observable remoteBalance = Big(0);
@@ -84,6 +85,7 @@ export default class Channel {
   update(lndChannel: LND.Channel.AsObject) {
     this.chanId = lndChannel.chanId;
     this.remotePubkey = lndChannel.remotePubkey;
+    this.channelPoint = lndChannel.channelPoint;
     this.capacity = Big(lndChannel.capacity);
     this.localBalance = Big(lndChannel.localBalance);
     this.remoteBalance = Big(lndChannel.remoteBalance);

app/src/store/store.ts

@@ -48,6 +48,8 @@ export class Store {
   // a flag to indicate when the store has completed all of its
   // API requests requested during initialization
   @observable initialized = false;
+  // a flag to indicate when the websocket streams are connected
+  @observable streamsConnected = false;
 
   constructor(
     lnd: LndApi,
@@ -83,9 +85,14 @@ export class Store {
           this.uiStore.goToLoop();
           // also fetch all the data we need
           this.fetchAllData();
+          // connect and subscribe to the server-side streams
+          this.connectToStreams();
+          this.subscribeToStreams();
         } else {
           // go to auth page if we are not authenticated
           this.uiStore.gotoAuth();
+          // unsubscribe from streams since we are no longer authenticated
+          this.unsubscribeFromStreams();
         }
       },
       { name: 'authenticatedAutorun' },
@@ -102,6 +109,39 @@ export class Store {
     await this.swapStore.fetchSwaps();
     await this.nodeStore.fetchBalances();
   }
+
+  /** connects to the LND and Loop websocket streams if not already connected */
+  @action.bound
+  connectToStreams() {
+    if (this.streamsConnected) return;
+
+    const { lnd, loop } = this.api;
+    lnd.connectStreams();
+    loop.connectStreams();
+    this.streamsConnected = true;
+  }
+
+  /**
+   * subscribes to the LND and Loop streaming endpoints
+   */
+  @action.bound
+  subscribeToStreams() {
+    const { lnd, loop } = this.api;
+    lnd.on('transaction', this.nodeStore.onTransaction);
+    lnd.on('channel', this.channelStore.onChannelEvent);
+    loop.on('monitor', this.swapStore.onSwapUpdate);
+  }
+
+  /**
+   * unsubscribes from the LND and Loop streaming endpoints
+   */
+  @action.bound
+  unsubscribeFromStreams() {
+    const { lnd, loop } = this.api;
+    lnd.off('transaction', this.nodeStore.onTransaction);
+    lnd.off('channel', this.channelStore.onChannelEvent);
+    loop.off('monitor', this.swapStore.onSwapUpdate);
+  }
 }
 
 /**

app/src/store/stores/buildSwapStore.ts

@@ -361,7 +361,6 @@ class BuildSwapStore {
           // hide the swap UI after it is complete
           this.cancel();
           this._store.uiStore.toggleProcessingSwaps();
-          this._store.swapStore.fetchSwaps();
         });
       } catch (error) {
         this._store.uiStore.handleError(error, `Unable to Perform ${direction}`);

app/src/store/stores/channelStore.ts

@@ -7,10 +7,18 @@ import {
   toJS,
   values,
 } from 'mobx';
+import { ChannelEventUpdate, ChannelPoint } from 'types/generated/lnd_pb';
 import Big from 'big.js';
 import { Store } from 'store';
 import { Channel } from '../models';
 
+const {
+  OPEN_CHANNEL,
+  CLOSED_CHANNEL,
+  ACTIVE_CHANNEL,
+  INACTIVE_CHANNEL,
+} = ChannelEventUpdate.UpdateType;
+
 export default class ChannelStore {
   private _store: Store;
 
@@ -90,10 +98,52 @@ export default class ChannelStore {
     }
   }
 
+  /** update the channel list based on events from the API */
+  @action.bound
+  onChannelEvent(event: ChannelEventUpdate.AsObject) {
+    this._store.log.info('handle incoming channel event', event);
+    if (event.type === INACTIVE_CHANNEL && event.inactiveChannel) {
+      // set the channel in state to inactive
+      const point = this._channelPointToString(event.inactiveChannel);
+      values(this.channels)
+        .filter(c => c.channelPoint === point)
+        .forEach(c => {
+          c.active = false;
+          this._store.log.info('updated channel', toJS(c));
+        });
+    } else if (event.type === ACTIVE_CHANNEL && event.activeChannel) {
+      // set the channel in state to active
+      const point = this._channelPointToString(event.activeChannel);
+      values(this.channels)
+        .filter(c => c.channelPoint === point)
+        .forEach(c => {
+          c.active = true;
+          this._store.log.info('updated channel', toJS(c));
+        });
+    } else if (event.type === CLOSED_CHANNEL && event.closedChannel) {
+      // delete the closed channel
+      const channel = this.channels.get(event.closedChannel.chanId);
+      this.channels.delete(event.closedChannel.chanId);
+      this._store.log.info('removed closed channel', toJS(channel));
+    } else if (event.type === OPEN_CHANNEL && event.openChannel) {
+      // add the new opened channel
+      const channel = new Channel(this._store, event.openChannel);
+      this.channels.set(channel.chanId, channel);
+      this._store.log.info('added new open channel', toJS(channel));
+    }
+  }
+
   /** exports the sorted list of channels to CSV file */
   @action.bound
   exportChannels() {
     this._store.log.info('exporting Channels to a CSV file');
     this._store.csv.export('channels', Channel.csvColumns, toJS(this.sortedChannels));
   }
+
+  /** converts a base64 encoded channel point to a hex encoded channel point */
+  private _channelPointToString(channelPoint: ChannelPoint.AsObject) {
+    const txidBytes = channelPoint.fundingTxidBytes as string;
+    const txid = Buffer.from(txidBytes, 'base64').reverse().toString('hex');
+    return `${txid}:${channelPoint.outputIndex}`;
+  }
 }

app/src/store/stores/nodeStore.ts

@@ -1,4 +1,5 @@
 import { action, observable, runInAction, toJS } from 'mobx';
+import { Transaction } from 'types/generated/lnd_pb';
 import Big from 'big.js';
 import { Store } from 'store';
 import { Wallet } from '../models';
@@ -8,6 +9,11 @@ type NodeNetwork = 'mainnet' | 'testnet' | 'regtest';
 
 export default class NodeStore {
   private _store: Store;
+  /**
+   * an internal list of txn ids used to prevent updating the balance
+   * multiple times for the same transaction.
+   */
+  private _knownTxns: string[] = [];
 
   /** the pubkey of the LND node */
   @observable pubkey = '';
@@ -64,4 +70,16 @@ export default class NodeStore {
       this._store.uiStore.handleError(error, 'Unable to fetch balances');
     }
   }
+
+  /**
+   * updates the wallet balance from the transaction provided
+   */
+  @action.bound
+  onTransaction(transaction: Transaction.AsObject) {
+    this._store.log.info('handle incoming transaction', transaction);
+    if (this._knownTxns.includes(transaction.txHash)) return;
+    this._knownTxns.push(transaction.txHash);
+    this.wallet.walletBalance = this.wallet.walletBalance.plus(transaction.amount);
+    this._store.log.info('updated nodeStore.wallet', toJS(this.wallet));
+  }
 }

app/src/store/stores/swapStore.ts

@@ -9,6 +9,7 @@ import {
   toJS,
   values,
 } from 'mobx';
+import { SwapStatus } from 'types/generated/loop_pb';
 import { IS_PROD, IS_TEST } from 'config';
 import { Store } from 'store';
 import { Swap } from '../models';
@@ -89,12 +90,7 @@ export default class SwapStore {
           // update existing swaps or create new ones in state. using this
           // approach instead of overwriting the array will cause fewer state
           // mutations, resulting in better react rendering performance
-          const existing = this.swaps.get(loopSwap.id);
-          if (existing) {
-            existing.update(loopSwap);
-          } else {
-            this.swaps.set(loopSwap.id, new Swap(loopSwap));
-          }
+          this.addOrUpdateSwap(loopSwap);
         });
         // remove any swaps in state that are not in the API response
         const serverIds = swapsList.map(c => c.id);
@@ -111,6 +107,26 @@ export default class SwapStore {
     }
   }
 
+  /** adds a new swap or updates an existing one */
+  @action.bound
+  addOrUpdateSwap(loopSwap: SwapStatus.AsObject) {
+    const existing = this.swaps.get(loopSwap.id);
+    if (existing) {
+      existing.update(loopSwap);
+      this._store.log.info('updated existing swap', toJS(loopSwap));
+    } else {
+      this.swaps.set(loopSwap.id, new Swap(loopSwap));
+      this._store.log.info('added new swap', toJS(loopSwap));
+    }
+  }
+
+  /** updates the swap and refreshes the channel list */
+  @action.bound
+  onSwapUpdate(loopSwap: SwapStatus.AsObject) {
+    this.addOrUpdateSwap(loopSwap);
+    this._store.channelStore.fetchChannels();
+  }
+
   @action.bound
   startPolling() {
     if (this.pollingInterval) this.stopPolling();