Videre PingPong Service

package.json

@@ -28,6 +28,7 @@
     "lint:fix": "npx eslint . --ext .ts --fix && npx prettier --write .",
     "test": "npx mocha -r ts-node/register test/**/*.spec.ts",
     "clean": "rm -rf dist",
+    "coverage": "c8 --all --exclude typechain* --exclude coverage --exclude dist --exclude src/proto* --exclude test npm test && c8 report --all --exclude typechain* --exclude coverage --exclude dist --exclude test --exclude src/proto* -r html",
     "build": "yarn clean && yarn tsc -p tsconfig-build.json",
     "protoc:libs": "cp -pR ./node_modules/@windingtree/stays-models/dist/proto/*.proto ./src/proto/",
     "protoc:local": "protoc --ts_out ./src/proto --proto_path ./src/proto ./src/proto/*.proto",
@@ -42,7 +43,7 @@
   },
   "dependencies": {
     "@protobuf-ts/plugin": "^2.6.0",
-    "@windingtree/stays-models": "^2.0.2",
+    "@windingtree/stays-models": "^2.1.0",
     "@windingtree/videre-sdk": "^0.6.0",
     "axios": "^0.27.2",
     "bcrypt": "^5.0.1",
@@ -77,6 +78,7 @@
     "@typechain/ethers-v5": "^10.0.0",
     "@types/bcrypt": "^5.0.0",
     "@types/chai": "^4.3.1",
+    "@types/chai-as-promised": "^7.1.5",
     "@types/cookie-parser": "^1.4.3",
     "@types/cors": "^2.8.12",
     "@types/luxon": "^2.3.2",
@@ -89,10 +91,10 @@
     "@typescript-eslint/eslint-plugin": "^5.27.0",
     "@typescript-eslint/parser": "^5.27.0",
     "@windingtree/videre-contracts": "^1.0.2",
+    "c8": "^7.11.3",
     "chai": "^4.3.6",
-    "chai-ethers": "^0.0.1",
     "chai-as-promised": "^7.1.1",
-    "@types/chai-as-promised": "^7.1.5",
+    "chai-ethers": "^0.0.1",
     "conventional-changelog-cli": "^2.2.2",
     "eslint": "^8.16.0",
     "eslint-config-prettier": "^8.5.0",

src/config.ts

@@ -1,4 +1,5 @@
 import type { TypedDataDomain } from '@ethersproject/abstract-signer';
+import { VidereConfig } from '@windingtree/videre-sdk';
 import dotenv from 'dotenv';
 
 dotenv.config();
@@ -36,9 +37,13 @@ export const defaultManagerLogin = 'manager';
 export const defaultManagerPassword = 'winwin';
 export const web3StorageKey = process.env.WEB3STORAGE_KEY as string;
 export const wakuConfig = { bootstrap: { default: true } };
+export const videreConfig: VidereConfig = {
+  line: 'stays',
+  version: 1
+};
 export const typedDataDomain: TypedDataDomain = {
-  name: 'stays',
-  version: '1',
+  name: videreConfig.line,
+  version: String(videreConfig.version),
   verifyingContract: String(process.env.APP_VERIFYING_CONTRACT),
   chainId: Number(process.env.APP_CHAIN_ID)
 };

src/index.ts

@@ -3,6 +3,7 @@ import { port, prometheusEnabled } from './config';
 import bootstrapService from './services/BootstrapService';
 import DBService from './services/DBService';
 import { MetricsService } from './services/MetricsService';
+import WakuService from './services/WakuService';
 
 process.on('unhandledRejection', async (error) => {
   console.log(error);
@@ -12,6 +13,7 @@ process.on('unhandledRejection', async (error) => {
 
 const main = async (): Promise<ServerService> => {
   const server = new ServerService(port);
+  const waku = WakuService.getInstance;
 
   await bootstrapService.bootstrap();
 

src/services/PingPongService.ts

@@ -0,0 +1,77 @@
+import { geoToH3 } from 'h3-js';
+import { utils } from 'ethers';
+
+import { eip712, utils as vUtils } from '@windingtree/videre-sdk';
+import { constants } from '@windingtree/videre-sdk/dist/cjs/utils';
+
+import facilityRepository from '../repositories/FacilityRepository';
+
+import log from './LogService';
+import WakuService from './WakuService';
+import WalletService from './WalletService';
+
+import { walletAccountsIndexes } from 'src/types';
+import { Facility } from 'src/proto/facility';
+import { Ping, Pong } from 'src/proto/pingpong';
+import { LatLng } from 'src/proto/latlng';
+
+import { typedDataDomain, videreConfig } from '../config';
+
+export default class PingPongService {
+  protected waku: WakuService;
+  protected unsubscribe: () => void = () => {
+    return;
+  };
+
+  /**
+   * After database initialization, start the ping / pong service
+   * for a specified facility.
+   */
+  public async start(facilityId: string): Promise<void> {
+    const metadata: Facility = (await facilityRepository.getFacilityKey(
+      facilityId,
+      'metadata'
+    )) as Facility;
+    const loc = metadata.location;
+
+    if (!this.waku) this.waku = WakuService.getInstance();
+
+    if (loc) {
+      // get the h3 index to monitor from the location
+      const h3Index = geoToH3(
+        loc.latitude,
+        loc.longitude,
+        constants.DefaultH3Resolution
+      );
+
+      // watch for pings
+      this.unsubscribe = await this.waku.makeWakuObserver(
+        async (message) => {
+          const msg = this.waku.processMessage(Ping, message);
+          if (msg) log.green(`Ping received with timestamp: ${msg.timestamp}`);
+
+          // respond to the ping with a pong
+          this.waku.sendMessage(
+            Pong,
+            await vUtils.createSignedMessage<Pong>(
+              typedDataDomain,
+              eip712.pingpong.Pong,
+              {
+                serviceProvider: utils.arrayify(facilityId),
+                loc: LatLng.toBinary(loc),
+                signature: utils.toUtf8Bytes('') // initially blank signature which gets filled in
+              },
+              await WalletService.getWalletByIndex(walletAccountsIndexes.BIDDER)
+            ),
+            vUtils.generateTopic({ ...videreConfig, topic: 'pong' }, h3Index)
+          );
+        },
+        [vUtils.generateTopic({ ...videreConfig, topic: 'ping' }, h3Index)]
+      );
+    }
+  }
+
+  public async stop(): Promise<void> {
+    this.unsubscribe();
+  }
+}

src/services/WakuService.ts

@@ -1,35 +1,45 @@
 import { Waku, WakuMessage } from 'js-waku';
 import type { MessageType } from '@protobuf-ts/runtime';
 import { wakuConfig } from '../config';
+import log from './LogService';
 
 export type WakuMessageHandler = (message: WakuMessage) => void;
 
-export class WakuService {
-  public waku: Waku;
+export default class WakuService {
+  protected waku: Waku;
+  private static _instance: WakuService = new WakuService();
 
-  public async connect(): Promise<WakuService> {
-    if (this.waku) {
-      return this;
+  constructor() {
+    if (WakuService._instance) {
+      throw new Error(
+        'Error: Instantiation failed: Use WakuService.getInstance() instead of new'
+      );
     }
 
-    console.log('Connecting to Waku...');
-    const waku = await Waku.create(wakuConfig);
-    await waku.waitForRemotePeer();
-    console.log('...Connected');
+    Waku.create(wakuConfig).then(
+      (waku) => {
+        log.green('Connecting to Waku...');
 
-    this.waku = waku;
-    return this;
+        waku.waitForRemotePeer(undefined, 10000).then(() => {
+          log.green('...Connected');
+          this.waku = waku;
+        });
+      },
+      () => {
+        log.red('...Failed');
+      }
+    );
+  }
+
+  public static getInstance(): WakuService {
+    return WakuService._instance;
   }
 
   public async sendMessage<T extends object>(
     protoMessageInstance: MessageType<T>,
     message: T,
     topic: string
   ): Promise<void> {
-    if (!this.waku) {
-      await this.connect();
-    }
-
     const msg = await WakuMessage.fromBytes(
       protoMessageInstance.toBinary(message),
       topic
@@ -49,17 +59,12 @@ export class WakuService {
     messageHandler: WakuMessageHandler,
     topics: string[]
   ) {
-    if (!this.waku) {
-      await this.connect();
-    }
     this.waku.relay.addObserver(messageHandler, topics);
-    console.log('Subscribed to topics:', topics);
+    log.green('Subscribed to topics:' + topics);
 
     return () => {
       this.waku.relay.deleteObserver(messageHandler, topics);
-      console.log('Unsubscribed from topics:', topics);
+      log.yellow('Unsubscribed from topics:' + topics);
     };
   }
 }
-
-export default new WakuService();