Merge pull request #34 from usherlabs/LABS-1100/Integrate_Events_Webstream

Labs 1100/integrate events webstream

Author: rsoury

.github/workflows/ci.yml

@@ -32,7 +32,7 @@ jobs:
 
       - name: Download ROOCH and APTOS
         run: |
-          wget https://github.com/rooch-network/rooch/releases/latest/download/rooch-ubuntu-22.04.zip
+          wget  https://github.com/rooch-network/rooch/releases/latest/download/rooch-ubuntu-22.04.zip
           wget -qO- "https://aptos.dev/scripts/install_cli.py" | python3
           echo "Download completed"
       

.vscode/settings.json

@@ -14,5 +14,8 @@
   ],
   "[typescript]": {
     "editor.defaultFormatter": "biomejs.biome"
+  },
+  "[prisma]": {
+    "editor.defaultFormatter": "Prisma.prisma"
   }
-}
+}

README.md

@@ -171,7 +171,7 @@ pnpm dev
     --args 'string:POST' \
     --args 'string:{}' \
     --args 'string:{
-      "model": "gpt-4",
+      "model": "gpt-4o",
       "messages": [{"role": "user", "content": "Say this is a test!"}],
       "temperature": 0.7
     }' \

docs/APTOS.md

@@ -21,5 +21,5 @@ aptos move publish --named-addresses verity=default,verity_test_foreign_module=d
 Send a new request transaction to have it indexed. Make sure to replace placeholders with actual values relevant to your setup.
 
 ```bash
-aptos move run --function-id  0xa2b7160c0dc70548e8105121b075df9ea3b98c0c82294207ca38cb1165b94f59::example_caller::request_data --sender-account default --args 'string:https://api.x.com/2/users/by/username/elonmusk?user.fields=public_metrics' --args 'string:GET' --args 'string:{}' --args 'string:{}' --args 'string:.data.public_metrics.followers_count' --args 'address:6b516ae2eb4aac47ffadd502cf19ce842020f515f1abea3e154cfc053ab3ab9a'
+aptos move run --function-id 0xa2b7160c0dc70548e8105121b075df9ea3b98c0c82294207ca38cb1165b94f59::example_caller::request_data --sender-account default --args 'string:https://api.x.com/2/users/by/username/elonmusk?user.fields=public_metrics' --args 'string:GET' --args 'string:{}' --args 'string:{}' --args 'string:.data.public_metrics.followers_count' --args 'address:6b516ae2eb4aac47ffadd502cf19ce842020f515f1abea3e154cfc053ab3ab9a'
 ```

orchestrator/prisma/migrations/20250507154028_/migration.sql

@@ -0,0 +1,13 @@
+-- CreateTable
+CREATE TABLE "SupportedUrl" (
+    "domain" TEXT NOT NULL,
+    "supported_path" TEXT[],
+    "authType" TEXT NOT NULL,
+    "authKey" TEXT NOT NULL,
+    "requestRate" BIGINT NOT NULL,
+
+    CONSTRAINT "SupportedUrl_pkey" PRIMARY KEY ("domain")
+);
+
+-- CreateIndex
+CREATE INDEX "SupportedUrl_authKey_idx" ON "SupportedUrl"("authKey");

orchestrator/prisma/schema.prisma

@@ -10,34 +10,43 @@ datasource db {
   url      = env("DATABASE_URL")
 }
 
-model Events{
-  id                        String @id @default(cuid())
-  eventHandleId             String 
-  eventSeq                  BigInt
-  eventType                 String
-  eventData                 String
-  eventIndex                String
-  decoded_event_data        String  //JSON String
-  chain                     String   @default("ROOCH-testnet")
-
-  status                    Int
-  retries                   Int
-  response                  String?
-  indexedAt                 DateTime  @default(now())
-  updateAt                  DateTime  @updatedAt
-
-  @@index([eventHandleId, eventSeq,chain])
+model Events {
+  id                 String @id @default(cuid())
+  eventHandleId      String
+  eventSeq           BigInt
+  eventType          String
+  eventData          String
+  eventIndex         String
+  decoded_event_data String //JSON String
+  chain              String @default("ROOCH-testnet")
+
+  status    Int
+  retries   Int
+  response  String?
+  indexedAt DateTime @default(now())
+  updateAt  DateTime @updatedAt
+
+  @@index([eventHandleId, eventSeq, chain])
 }
 
-model Keeper{
-  id                        String    @id @default(cuid())
-  chain                     String    @default("ROOCH-testnet")
-  module                    String 
-  privateKey                String
+model Keeper {
+  id         String @id @default(cuid())
+  chain      String @default("ROOCH-testnet")
+  module     String
+  privateKey String
 
-  createdAt                 DateTime  @default(now())
-  updateAt                  DateTime  @updatedAt
+  createdAt DateTime @default(now())
+  updateAt  DateTime @updatedAt
 
   @@unique([chain, module])
+}
+
+model SupportedUrl {
+  domain         String    @id
+  supported_path String[]
+  authType       String
+  authKey        String   // BEARER, OAUTH1.0,  OAUTH2.0
+  requestRate    BigInt                   
 
-}
+  @@index([authKey])
+}

orchestrator/src/env.ts

@@ -8,7 +8,7 @@ const baseConfig = {
   // Rooch
   roochChainId: (process.env.ROOCH_CHAIN_ID
     ? process.env.ROOCH_CHAIN_ID.split(",")
-    : ["testnet", "pre-mainnet"]) as RoochNetwork[],
+    : ["testnet", "mainnet"]) as RoochNetwork[],
   roochPrivateKey: process.env.ROOCH_PRIVATE_KEY ?? "",
   roochOracleAddress: process.env.ROOCH_ORACLE_ADDRESS ?? "",
   roochIndexerCron: process.env.ROOCH_INDEXER_CRON,
@@ -25,6 +25,7 @@ const baseConfig = {
   // Integrations
   xBearerToken: process.env.X_BEARER_TOKEN ?? "",
   openAIToken: process.env.OPEN_AI_TOKEN ?? "",
+  azureToken: process.env.AZURE_TOKEN ?? "",
 };
 
 interface IEnvVars {
@@ -43,6 +44,7 @@ interface IEnvVars {
   batchSize: number;
   xBearerToken: string;
   openAIToken: string;
+  azureToken: string;
 }
 
 const envVarsSchema = Joi.object({
@@ -93,6 +95,7 @@ const envVarsSchema = Joi.object({
   // Integrations
   xBearerToken: Joi.string().allow("").required(),
   openAIToken: Joi.string().allow("").required(),
+  azureToken: Joi.string().allow("").required(),
 
   // Common
   sentryDSN: Joi.string().allow("", null),
@@ -117,6 +120,7 @@ export default {
   integrations: {
     xBearerToken: envVars.xBearerToken,
     openAIToken: envVars.openAIToken,
+    azureToken: envVars.azureToken,
   },
   rooch: {
     chainId: envVars.roochChainId,

orchestrator/src/index.ts

@@ -10,8 +10,6 @@ import { log } from "./logger";
   // Start cron job to check for new events from Rooch Oracles
 
   if (env.rooch.privateKey && env.rooch.chainId.length > 0 && env.rooch.oracleAddress && env.chains.includes("ROOCH")) {
-    // https://www.npmjs.com/package/cron#cronjob-class
-
     env.rooch.chainId.map((chain) => {
       const rooch = new RoochIndexer(env.rooch.privateKey, chain, env.rooch.oracleAddress);
       let running = false;

orchestrator/src/indexer/base.ts

@@ -1,10 +1,10 @@
 import { log } from "@/logger";
 import { type ProcessedRequestAdded, RequestStatus } from "@/types";
 
-import { instance as openAIInstance } from "@/integrations/openAI";
-import { instance as xTwitterInstance } from "@/integrations/xtwitter";
+import { azureInstance, openAIInstance } from "@/integrations/openAI";
+import { xTwitterInstance } from "@/integrations/xtwitter";
 
-import type { BasicBearerAPIHandler } from "@/integrations/base";
+import { type BasicBearerAPIHandler, dynamicInstanceManager } from "@/integrations/base";
 import prismaClient from "../../prisma";
 
 // Abstract base class
@@ -50,6 +50,12 @@ export abstract class Indexer {
     if (openAIInstance.isApprovedPath(url)) {
       return openAIInstance;
     }
+    if (azureInstance.isApprovedPath(url)) {
+      return azureInstance;
+    }
+    if (!dynamicInstanceManager.loading) {
+      return dynamicInstanceManager.getInstance(url.host) || null;
+    }
     return null;
   }
 
@@ -94,7 +100,7 @@ export abstract class Indexer {
       if (handler) {
         return handler.submitRequest(data);
       }
-      return { status: 406, message: "URL Not supported" };
+      return { status: 406, message: "URL NOT supported" };
     } catch {
       return { status: 406, message: "Invalid URL" };
     }

orchestrator/src/indexer/rooch.ts

@@ -1,15 +1,23 @@
 import env from "@/env";
-import { instance as xTwitterInstance } from "@/integrations/xtwitter";
+import { xTwitterInstance } from "@/integrations/xtwitter";
 import { log } from "@/logger";
 import type { IEvent, IRequestAdded, JsonRpcResponse, ProcessedRequestAdded, RoochNetwork } from "@/types";
 import { decodeNotifyValueFull } from "@/util";
-import { Args, RoochClient, Secp256k1Keypair, Transaction, getRoochNodeUrl } from "@roochnetwork/rooch-sdk";
+import {
+  Args,
+  RoochClient,
+  RoochWebSocketTransport,
+  Secp256k1Keypair,
+  Transaction,
+  getRoochNodeUrl,
+} from "@roochnetwork/rooch-sdk";
 import axios from "axios";
 import prismaClient from "../../prisma";
 import { Indexer } from "./base";
 
 export default class RoochIndexer extends Indexer {
   private keyPair: Secp256k1Keypair;
+  private client: RoochClient;
 
   constructor(
     private privateKey: string,
@@ -18,6 +26,18 @@ export default class RoochIndexer extends Indexer {
   ) {
     super(oracleAddress, Secp256k1Keypair.fromSecretKey(privateKey).getRoochAddress().toHexAddress());
     this.keyPair = Secp256k1Keypair.fromSecretKey(this.privateKey);
+    const wsTransport = new RoochWebSocketTransport({
+      url: getRoochNodeUrl(this.chainId),
+      reconnectDelay: 1000, // Delay between reconnection attempts (default: 1000ms)
+      maxReconnectAttempts: 5, // Maximum number of reconnection attempts (default: 5)
+      requestTimeout: 30000, // Request timeout (default: 30000ms)
+      connectionReadyTimeout: 5000, // Connection ready timeout (default: 5000ms)
+    });
+
+    // Create client with WebSocket transport
+    this.client = new RoochClient({
+      transport: wsTransport,
+    });
     log.info(`Rooch Indexer initialized`);
     log.info(`Chain ID: ${this.getChainId()} \n\t\tOrchestrator Oracle Node Address: ${this.orchestrator}`);
   }
@@ -35,7 +55,7 @@ export default class RoochIndexer extends Indexer {
    */
   async sendUnfulfilledRequests() {
     // Initialize the Rooch client with the current node URL
-    const client = new RoochClient({ url: this.getRoochNodeUrl() });
+    // const _client = new RoochClient({ url: this.getRoochNodeUrl() });
 
     // Initialize cursor object for pagination
     const cursor = {
@@ -53,7 +73,7 @@ export default class RoochIndexer extends Indexer {
     while (cursor.isNextPage) {
       try {
         // Query for object states with pagination
-        const query = await client.queryObjectStates({
+        const query = await this.client.queryObjectStates({
           filter: {
             object_type: `${this.oracleAddress}::oracles::Request`,
           },
@@ -135,7 +155,7 @@ export default class RoochIndexer extends Indexer {
   }
 
   getRoochNodeUrl() {
-    return this.chainId === "pre-mainnet" ? "https://main-seed.rooch.network" : getRoochNodeUrl(this.chainId);
+    return getRoochNodeUrl(this.chainId);
   }
 
   /**
@@ -216,11 +236,7 @@ export default class RoochIndexer extends Indexer {
   }
 
   async isPreviouslyExecuted(data: ProcessedRequestAdded<any>) {
-    const client = new RoochClient({
-      url: this.getRoochNodeUrl(),
-    });
-
-    const view = await client.executeViewFunction({
+    const view = await this.client.executeViewFunction({
       target: `${this.oracleAddress}::oracles::get_response_status`,
       args: [Args.objectId(data.request_id)],
     });
@@ -240,11 +256,7 @@ export default class RoochIndexer extends Indexer {
    * @returns {Promise<any>} - The receipt of the transaction.
    */
   async sendFulfillment(data: ProcessedRequestAdded<any>, status: number, result: string) {
-    const client = new RoochClient({
-      url: this.getRoochNodeUrl(),
-    });
-
-    const view = await client.executeViewFunction({
+    const view = await this.client.executeViewFunction({
       target: `${this.oracleAddress}::oracles::get_response_status`,
       args: [Args.objectId(data.request_id)],
     });
@@ -282,7 +294,7 @@ export default class RoochIndexer extends Indexer {
 
     tx.setMaxGas(1000000000);
 
-    const receipt = await client.signAndExecuteTransaction({
+    const receipt = await this.client.signAndExecuteTransaction({
       transaction: tx,
       signer: this.keyPair,
     });
@@ -298,7 +310,7 @@ export default class RoochIndexer extends Indexer {
     });
     try {
       if ((data.notify?.length ?? 0) > 66) {
-        const module_abi = await client.getModuleAbi({
+        const module_abi = await this.client.getModuleAbi({
           moduleAddr: notify_module[0] ?? "",
           moduleName: notify_module[1] ?? "",
         });
@@ -312,7 +324,7 @@ export default class RoochIndexer extends Indexer {
             args: function_abi?.params?.length === 0 ? [] : [Args.objectId(data.request_id)],
           });
           tx.setMaxGas(2_0000_0000);
-          const notification_receipt = await client.signAndExecuteTransaction({
+          const notification_receipt = await this.client.signAndExecuteTransaction({
             transaction: tx,
             signer: Secp256k1Keypair.fromSecretKey(keeper_key.privateKey),
           });