Merge pull request #5 from dialectlabs/feature/multichannel

Feature/multichannel
dialectlabs · May 25, 2022 · 42f0e14 · 42f0e14
2 parents e6cbb52 + 1ce83a8
commit 42f0e14
Show file tree

Hide file tree

Showing 6 changed files with 503 additions and 103 deletions.
diff --git a/deployment/deployment.yaml b/deployment/deployment.yaml
@@ -62,7 +62,38 @@ spec:
             secretKeyRef:
               key: SABER_MESSAGING_PRIVATE_KEY
               name: env-vars
-        image: dialectlab/saber-monitoring-service:0.0.1
+        - name: TWILIO_ACCOUNT_SID
+          valueFrom:
+            secretKeyRef:
+              key: TWILIO_ACCOUNT_SID
+              name: env-vars
+        - name: TWILIO_AUTH_TOKEN
+          valueFrom:
+            secretKeyRef:
+              key: TWILIO_AUTH_TOKEN
+              name: env-vars
+        - name: TWILIO_SMS_SENDER
+          value: "+15613366687"
+        - name: WEB2_SUBSCRIBER_SERVICE_BASE_URL
+          value: "https://dialectapi.to"
+        - name: WEB2_SUBSCRIBER_SERVICE_BASIC_AUTH
+          valueFrom:
+            secretKeyRef:
+              key: WALLET_ADDRESS_REGISTRY_SERVICE_BASIC_AUTH_USER
+              name: env-vars
+        - name: SENDGRID_KEY
+          valueFrom:
+            secretKeyRef:
+              key: SENDGRID_API_KEY
+              name: env-vars
+        - name: SENDGRID_EMAIL
+          value: "[email protected]"
+        - name: TELEGRAM_TOKEN
+          valueFrom:
+            secretKeyRef:
+              key: TELEGRAM_TOKEN
+              name: env-vars
+        image: dialectlab/saber-monitoring-service:0.1.0
         imagePullPolicy: Always
         name: saber-monitoring-service
         resources: {}

diff --git a/package.json b/package.json
@@ -21,7 +21,7 @@
     "test:e2e": "jest --config ./test/jest-e2e.json"
   },
   "dependencies": {
-    "@dialectlabs/monitor": "^1.2.12",
+    "@dialectlabs/monitor": "^2.0.6",
     "@dialectlabs/web3": "^0.2.0",
     "@gokiprotocol/client": "^0.6.1",
     "@nestjs/common": "^8.0.0",
@@ -64,7 +64,7 @@
     "eslint-config-prettier": "^8.3.0",
     "eslint-plugin-prettier": "^4.0.0",
     "jest": "^27.2.5",
-    "prettier": "^2.3.2",
+    "prettier": "^2.6.2",
     "source-map-support": "^0.5.20",
     "supertest": "^6.1.3",
     "ts-jest": "^27.0.3",

diff --git a/src/app.module.ts b/src/app.module.ts
@@ -1,7 +1,8 @@
 import { Module } from '@nestjs/common';
-import { SaberMonitoringService } from './monitor/saber-monitoring-service';
 import { ScheduleModule } from '@nestjs/schedule';
 import { DialectConnection } from './monitor/dialect-connection';
+import { WhaleMonitoringService } from './monitor/whale-monitoring-service';
+import { FarmMonitoringService } from './monitor/farm-monitoring-service';
 
 @Module({
   imports: [ScheduleModule.forRoot()],
@@ -11,7 +12,8 @@ import { DialectConnection } from './monitor/dialect-connection';
       provide: DialectConnection,
       useValue: DialectConnection.initialize(),
     },
-    SaberMonitoringService,
+    WhaleMonitoringService,
+    FarmMonitoringService,
   ],
 })
 export class AppModule {}
diff --git a/src/monitor/farm-monitoring-service.ts b/src/monitor/farm-monitoring-service.ts
@@ -0,0 +1,173 @@
+import {
+  Injectable,
+  Logger,
+  OnModuleDestroy,
+  OnModuleInit,
+} from '@nestjs/common';
+import {
+  DialectNotification,
+  Monitors,
+  SourceData,
+  Web2SubscriberRepository,
+} from '@dialectlabs/monitor';
+import {
+  InMemoryWeb2SubscriberRepository,
+  RestWeb2SubscriberRepository,
+} from '@dialectlabs/monitor/lib/cjs/internal/rest-web2-subscriber.repository';
+import { findAllDistinct } from '@dialectlabs/monitor/lib/cjs/internal/subsbscriber-repository-utilts';
+import { toDecimals } from '../saber-wars-api/saber-wars-api';
+import { DialectConnection } from './dialect-connection';
+import { Subject } from 'rxjs';
+import { QuarryEventSubscription } from '../saber-wars-api/quarry-event-api';
+import { getOwner, quarrySDK } from '../saber-wars-api/quarry-sdk-factory';
+import { getTokenInfo } from '../saber-wars-api/token-info-api';
+import { OnChainSubscriberRepository } from '@dialectlabs/monitor/lib/cjs/internal/on-chain-subscriber.repository';
+import { InMemorySubscriberRepository } from '@dialectlabs/monitor/lib/cjs/internal/in-memory-subscriber.repository';
+import { PublicKey } from '@solana/web3.js';
+
+@Injectable()
+export class FarmMonitoringService implements OnModuleInit, OnModuleDestroy {
+  private readonly logger = new Logger(FarmMonitoringService.name);
+  private readonly numberFormat = new Intl.NumberFormat('en-US');
+  private readonly inMemoryWeb2SubscriberRepository: Web2SubscriberRepository;
+  private readonly subscriberRepository: InMemorySubscriberRepository;
+
+  constructor(private readonly dialectConnection: DialectConnection) {
+    this.subscriberRepository = InMemorySubscriberRepository.decorate(
+      new OnChainSubscriberRepository(
+        dialectConnection.getProgram(),
+        dialectConnection.getKeypair(),
+      ),
+    );
+    this.inMemoryWeb2SubscriberRepository =
+      new InMemoryWeb2SubscriberRepository(
+        dialectConnection.getKeypair().publicKey,
+        new RestWeb2SubscriberRepository(
+          process.env.WEB2_SUBSCRIBER_SERVICE_BASE_URL!,
+          dialectConnection.getKeypair().publicKey,
+        ),
+      );
+  }
+
+  onModuleInit() {
+    this.initFarmMonitor();
+  }
+
+  async onModuleDestroy() {
+    await Monitors.shutdown();
+  }
+
+  private async initFarmMonitor() {
+    const subscribers = await findAllDistinct(
+      this.subscriberRepository,
+      this.inMemoryWeb2SubscriberRepository,
+    );
+
+    const quarryEvents = new Subject<SourceData<DialectNotification>>();
+    new QuarryEventSubscription(quarrySDK.programs.Mine, async (evt) => {
+      if (subscribers.length === 0) {
+        this.logger.warn('No subscribers, skipping event');
+        return;
+      }
+      const resourceId = process.env.TEST_MODE
+        ? subscribers[0]
+        : await getOwner(evt.data.authority);
+      if (evt.name === 'StakeEvent') {
+        const tokenInfo = await getTokenInfo(evt.data.token);
+        quarryEvents.next({
+          data: {
+            message: `Success! You staked ${this.numberFormat.format(
+              toDecimals(evt.data.amount, tokenInfo.decimals),
+            )} ${tokenInfo.symbol} to ${tokenInfo.name}`,
+          },
+          groupingKey: resourceId.toBase58(),
+        });
+      }
+      if (evt.name === 'ClaimEvent') {
+        const tokenInfo = await getTokenInfo(evt.data.stakedToken);
+        quarryEvents.next({
+          data: {
+            message: `Success! You claimed ${this.numberFormat.format(
+              toDecimals(evt.data.amount, tokenInfo.decimals),
+            )} ${tokenInfo.symbol} from ${tokenInfo.name}`,
+          },
+          groupingKey: resourceId.toBase58(),
+        });
+      }
+    }).start();
+
+    const farmMonitor = Monitors.builder({
+      monitorKeypair: this.dialectConnection.getKeypair(),
+      dialectProgram: this.dialectConnection.getProgram(),
+      sinks: {
+        sms: {
+          twilioUsername: process.env.TWILIO_ACCOUNT_SID!,
+          twilioPassword: process.env.TWILIO_AUTH_TOKEN!,
+          senderSmsNumber: process.env.TWILIO_SMS_SENDER!,
+        },
+        email: {
+          apiToken: process.env.SENDGRID_KEY!,
+          senderEmail: process.env.SENDGRID_EMAIL!,
+        },
+        telegram: {
+          telegramBotToken: process.env.TELEGRAM_TOKEN!,
+        },
+      },
+      web2SubscriberRepositoryUrl: process.env.WEB2_SUBSCRIBER_SERVICE_BASE_URL,
+    })
+      .defineDataSource<DialectNotification>()
+      .push(quarryEvents)
+      .notify()
+      .dialectThread(
+        ({ value }) => {
+          this.logger.log(`Sending message ${value.message}`);
+          return {
+            message: value.message,
+          };
+        },
+        {
+          dispatch: 'unicast',
+          to: ({ groupingKey }) => new PublicKey(groupingKey),
+        },
+      )
+      .telegram(
+        ({ value }) => {
+          return {
+            body: `⚔️ SABER: ` + value.message,
+          };
+        },
+        {
+          dispatch: 'unicast',
+          to: ({ groupingKey }) => new PublicKey(groupingKey),
+        },
+      )
+      .sms(
+        ({ value }) => {
+          return {
+            body: `⚔️ SABER: ` + value.message,
+          };
+        },
+        {
+          dispatch: 'unicast',
+          to: ({ groupingKey }) => new PublicKey(groupingKey),
+        },
+      )
+      .email(
+        ({ value }) => {
+          return {
+            subject: `⚔️ SABER: Successful ${
+              value.message.includes('claimed') ? 'Claim' : 'Stake'
+            }`,
+            text: value.message,
+          };
+        },
+        {
+          dispatch: 'unicast',
+          to: ({ groupingKey }) => new PublicKey(groupingKey),
+        },
+      )
+      .and()
+      .build();
+    farmMonitor.start();
+  }
+}