feat: forward request sub (#594)

* feat: forward request sub

* chore: log change

src/server/modules/api/amboss/amboss.gql.ts

@@ -129,3 +129,12 @@ export const getEdgeInfoBatchQuery = gql`
     }
   }
 `;
+
+export const getPhantomPayment = gql`
+  query GetPhantomPayment($input: PhantomPaymentInput!) {
+    getPhantomPayment(input: $input) {
+      preimage
+      payment_amount
+    }
+  }
+`;

src/server/modules/api/amboss/amboss.service.ts

@@ -10,6 +10,7 @@ import { FetchService } from '../../fetch/fetch.service';
 import {
   getEdgeInfoBatchQuery,
   getNodeAliasBatchQuery,
+  getPhantomPayment,
   pingHealthCheckMutation,
   pushNodeBalancesMutation,
   saveBackupMutation,
@@ -78,6 +79,21 @@ export class AmbossService {
     return data.getNodeAliasBatch;
   }
 
+  async getPhantomPayment(paymentHash: string, signature: string) {
+    const { data, error } = await this.fetchService.graphqlFetchWithProxy(
+      this.configService.get('urls.amboss'),
+      getPhantomPayment,
+      { input: { payment_hash: paymentHash, signature } }
+    );
+
+    if (!data?.getPhantomPayment || error) {
+      this.logger.error('Error getting phantom payment info', { data, error });
+      return null;
+    }
+
+    return data.getPhantomPayment;
+  }
+
   async getNodeAliasBatch(pubkeys: string[]): Promise<(NodeAlias | null)[]> {
     this.logger.info('Fetching information for nodes', {
       amount: pubkeys.length,

src/server/modules/sub/sub.service.ts

@@ -6,6 +6,8 @@ import {
   subscribeToInvoices,
   subscribeToBackups,
   subscribeToPastPayments,
+  subscribeToForwardRequests,
+  SubscribeToForwardRequestsForwardRequestEvent,
 } from 'lightning';
 import { auto, each, map, forever } from 'async';
 import { Logger } from 'winston';
@@ -18,6 +20,8 @@ import { UserConfigService } from '../api/userConfig/userConfig.service';
 import { getNetwork } from 'src/server/utils/network';
 import { AmbossService } from '../api/amboss/amboss.service';
 
+const SHORT_CHANNEL_ID = '1052673x257x257';
+
 const restartSubscriptionTimeMs = 1000 * 30;
 
 type NodeType = {
@@ -277,6 +281,115 @@ export class SubService implements OnApplicationBootstrap {
               },
             ],
 
+            // Subscribe to node forward requests
+            forwardRequests: [
+              'checkAvailable',
+              async ({ checkAvailable }, callback) => {
+                const names = checkAvailable.map(a => a.name);
+
+                this.logger.info('Forward request subscription', {
+                  connections: names.join(', '),
+                });
+
+                return each(
+                  checkAvailable,
+                  (node, cbk) => {
+                    const sub = subscribeToForwardRequests({ lnd: node.lnd });
+
+                    this.subscriptions.push(sub);
+
+                    sub.on(
+                      'forward_request',
+                      async (
+                        data: SubscribeToForwardRequestsForwardRequestEvent
+                      ) => {
+                        this.logger.silly('New forward request event', {
+                          node: node.name,
+                          amount_msats: data.mtokens,
+                          fee_msats: data.fee_mtokens,
+                          in_channel: data.in_channel,
+                          out_channel: data.out_channel,
+                        });
+
+                        if (data.out_channel !== SHORT_CHANNEL_ID) {
+                          this.logger.debug(
+                            'Accepting non phantom forward request'
+                          );
+                          data.accept();
+
+                          return;
+                        }
+
+                        this.logger.info('Accepting phantom payment');
+
+                        const { signature } =
+                          await this.nodeService.signMessage(
+                            node.id,
+                            data.hash
+                          );
+
+                        const info = await this.ambossService.getPhantomPayment(
+                          data.hash,
+                          signature
+                        );
+
+                        if (!info) {
+                          this.logger.error('Unable to accept phantom payment');
+                          data.reject();
+
+                          return;
+                        }
+
+                        if (data.mtokens < info.payment_amount) {
+                          this.logger.error(
+                            'Unable to accept phantom payment because size is below expected',
+                            {
+                              expected: info.payment_amount,
+                              received: data.mtokens,
+                            }
+                          );
+                          data.reject();
+
+                          return;
+                        }
+
+                        if (!!info.preimage) {
+                          data.settle({ secret: info.preimage });
+
+                          this.logger.info('Accepted phantom payment request', {
+                            info,
+                          });
+
+                          return;
+                        }
+
+                        this.logger.error(
+                          'Error accepting phantom payment request',
+                          { info }
+                        );
+                        data.reject();
+                      }
+                    );
+
+                    sub.on('error', async err => {
+                      sub.removeAllListeners();
+
+                      this.logger.error(
+                        `ErrorInForwardRequestsSubscribe: ${node.name}`,
+                        { err }
+                      );
+
+                      cbk([
+                        'ErrorInForwardRequestsSubscribe',
+                        { node: node.name, err },
+                      ]);
+                    });
+                  },
+                  callback
+                );
+              },
+            ],
+
             // Subscribe to node channels
             channels: [
               'checkAvailable',