Merge pull request #3 from usherlabs/bugfix/Deployment-issue

feat: index request, improved Readme and deployment

.gitignore

@@ -121,4 +121,8 @@ cache
 
 # Move
 build
-temp*
+temp*
+
+
+# else
+.script

README.md

@@ -9,3 +9,86 @@ The first Oracle supported is an Oracle for X (Twitter) data.
 - [x] [Rooch Network](https://rooch.network/)
 - [ ] [Aptos](https://aptosfoundation.org/)
 - [ ] ~Sui~
+
+
+## Running Rooch orchestrator (Locally)
+
+### Prerequisites
+
+Before running the script, ensure you have the following prerequisites installed:
+
+- **Roach**: A blockchain development toolset.
+- **Node.js**: Alongside npm, yarn, or pnpm package managers.
+
+### Step-by-Step Instructions
+
+#### Step 1: Create a Roach Account
+
+First, you need to create a Roach account. This account will be used throughout the setup process.
+
+```bash
+rooch account create
+```
+
+#### Step 2: Clear and Start Local Network
+
+Clear any existing state and start the local network.
+
+```bash
+rooch server clean
+rooch server start
+```
+
+#### Step 3: Deploy Contracts
+
+Navigate to the `rooch` directory, build the contracts for development, publish them with named addresses and update `.env` `ROOCH_ORACLE_ADDRESS` with deployed Address
+
+```bash
+cd rooch
+rooch move build --dev
+rooch move publish --named-addresses verity_test_foreign_module=default,verity=default
+cd ..
+```
+
+#### Step 4: Install Node Dependencies
+
+Install the necessary Node.js dependencies using npm, yarn, or pnpm. Ensure you are in the root project directory.
+
+```bash
+npm install
+# or
+yarn install
+# or
+pnpm install
+```
+
+#### Step 5: Run Prisma Migration
+
+Run the Prisma migration to update your database schema according to your models.
+
+```bash
+npx prisma migrate dev
+```
+
+#### Step 6: Run Orchestrator
+
+Start the development server for your application. This step might vary depending on your project setup; the command below assumes a typical setup.
+
+```bash
+npm run dev
+# or
+yarn dev
+# or
+pnpm dev
+```
+
+#### Step 7: Send New Request Transaction
+
+Finally, send a new request transaction to have it indexed. Make sure to replace placeholders with actual values relevant to your setup.
+
+```bash
+cd rooch
+rooch move run --function <deploymentAddress>::example_caller::request_data --sender-account default --args 'string:v2v3v' --args 'string:v2v3v' --args 'string:v2v3v' --args 'string:v2v3v' --args 'string:v2v3v' --args 'address:0x9a759932a6640790b3e2a5fefdf23917c8830dcd8998fe8af3f3b49b0ab5ca35'
+```
+
+

orchestrator/prisma/index.ts

@@ -0,0 +1,4 @@
+import { PrismaClient } from "@prisma/client";
+
+const prismaClient = new PrismaClient();
+export default prismaClient;

orchestrator/prisma/migrations/20240822103300_init/migration.sql → orchestrator/prisma/migrations/20240822213740_init/migration.sql

@@ -9,8 +9,8 @@ CREATE TABLE "Events" (
     "decoded_event_data" TEXT NOT NULL,
     "status" INTEGER NOT NULL,
     "retries" INTEGER NOT NULL,
-    "response" TEXT NOT NULL,
-    "executedAt" DATETIME NOT NULL,
+    "response" TEXT,
+    "executedAt" DATETIME,
     "indexedAt" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
     "updateAt" DATETIME NOT NULL
 );

orchestrator/prisma/schema.prisma

@@ -21,10 +21,10 @@ model Events{
 
   status                    Int
   retries                   Int
-  response                  String  //JSON String
-  executedAt                DateTime
+  response                  String?  //JSON String
+  executedAt                DateTime?
   indexedAt                 DateTime  @default(now())
-  updateAt   DateTime @updatedAt
+  updateAt                  DateTime  @updatedAt
 
   @@index([eventHandleId, eventSeq])
 }

orchestrator/src/env.ts

@@ -7,8 +7,10 @@ const baseConfig = {
   roochChainId: process.env.ROOCH_CHAIN_ID,
   roochPrivateKey: process.env.ROOCH_PRIVATE_KEY ?? "",
   roochOracleAddress: process.env.ROOCH_ORACLE_ADDRESS ?? "",
+  roochIndexerCron: process.env.ROOCH_INDEXER_CRON,
   sentryDSN: process.env.SENTRY_DSN ?? "",
   ecdsaPrivateKey: process.env.SENTRY_DSN ?? "",
+  batchSize: process.env.BATCH_SIZE ?? 1000,
 };
 interface IEnvVars {
   preferredChain: SupportedChain;
@@ -18,6 +20,7 @@ interface IEnvVars {
   roochIndexerCron: string;
   sentryDSN?: string;
   ecdsaPrivateKey?: string;
+  batchSize: number;
 }
 
 const envVarsSchema = Joi.object({
@@ -42,6 +45,7 @@ const envVarsSchema = Joi.object({
   roochIndexerCron: Joi.string().default("*/5 * * * * *"),
   sentryDSN: Joi.string().allow("", null),
   ecdsaPrivateKey: Joi.string().allow("", null),
+  batchSize: Joi.number().default(1000),
 });
 
 const { value, error } = envVarsSchema.validate({
@@ -54,6 +58,7 @@ if (error) {
 const envVars = value as IEnvVars;
 
 export default {
+  batchSize: envVars.batchSize,
   chain: envVars.preferredChain,
   ecdsaPrivateKey: envVars.ecdsaPrivateKey,
   sentryDSN: envVars.sentryDSN,

orchestrator/src/indexer/rooch.ts

@@ -1,7 +1,9 @@
+import env from "@/env";
 import { log } from "@/logger";
-import type { RoochNetwork } from "@/types";
+import { type IRequestAdded, type JsonRpcResponse, RequestStatus, type RoochNetwork } from "@/types";
 import { getRoochNodeUrl } from "@roochnetwork/rooch-sdk";
 import axios from "axios";
+import prismaClient from "../../prisma";
 
 export default class RoochIndexer {
   constructor(
@@ -14,15 +16,24 @@ export default class RoochIndexer {
     log.info(`Oracle Address: ${this.oracleAddress}`);
   }
 
-  async fetchEvents(eventName: string) {
+  async fetchEvents<T>(
+    eventName: "RequestAdded" | "FulfilmentAdded",
+    last_processed: null | number = null,
+  ): Promise<JsonRpcResponse<T> | null> {
     try {
       const response = await axios.post(
         getRoochNodeUrl(this.chainId),
         {
           id: 101,
           jsonrpc: "2.0",
           method: "rooch_getEventsByEventHandle",
-          params: [`${this.oracleAddress}::oracles::${eventName}`, null, "1000", false, { decode: true }],
+          params: [
+            `${this.oracleAddress}::oracles::${eventName}`,
+            last_processed,
+            `${env.batchSize}`,
+            false,
+            { decode: true },
+          ],
         },
         {
           headers: {
@@ -36,16 +47,41 @@ export default class RoochIndexer {
       return response.data;
     } catch (error) {
       log.error("Error fetching events", error);
+      return null;
     }
-
-    return [];
   }
 
   async run() {
     log.info("Rooch indexer running...", Date.now());
 
+    const latestCommit = await prismaClient.events.findFirst({
+      orderBy: {
+        eventSeq: "desc",
+        // indexedAt: "desc", // Order by date in descending order
+      },
+    });
+
     // Fetch the latest events from the Rooch Oracles Contract
-    const newRequestsEvents = await this.fetchEvents("RequestAdded");
+    const newRequestsEvents = await this.fetchEvents<IRequestAdded>("RequestAdded", latestCommit?.eventSeq ?? null);
+
+    if (!newRequestsEvents || "data" in newRequestsEvents) {
+      //TODO: HANDLE ERROR
+      return;
+    }
+
+    await prismaClient.events.createMany({
+      data: newRequestsEvents?.result.data.map((request) => ({
+        eventHandleId: request.event_id.event_handle_id,
+        eventSeq: +request.event_id.event_seq,
+        eventData: request.event_data,
+        eventType: request.event_type,
+        eventIndex: request.event_index,
+        decoded_event_data: JSON.stringify(request.decoded_event_data),
+        retries: 0,
+        status: RequestStatus.INDEXED,
+      })),
+    });
+
     // const newFulfilmentEvents = await this.fetchEvents("FulfilmentAdded");
 
     // Filter the events to if they're only relevant to this Oracle (Orchestrator)

orchestrator/src/types.ts

@@ -20,3 +20,64 @@ export const SupportedChain = ChainList.reduce(
   },
   {} as Record<(typeof ChainList)[number], string>,
 );
+
+interface ParamsValue {
+  abilities: number;
+  type: string;
+  value: {
+    body: string;
+    headers: string;
+    method: string;
+    url: string;
+  };
+}
+
+interface NotifyValue {
+  abilities: number;
+  type: string;
+  value: VecValue;
+}
+interface VecValue {
+  vec: string[];
+}
+
+interface Value {
+  notify: NotifyValue;
+  oracle: string;
+  params: ParamsValue;
+  pick: string;
+}
+
+export interface IRequestAdded {
+  abilities: number;
+  type: string;
+  value: Value;
+}
+
+export interface IEvent<T> {
+  event_id: {
+    event_handle_id: string;
+    event_seq: string;
+  };
+  event_type: string;
+  event_data: string;
+  event_index: string;
+  decoded_event_data: T;
+}
+
+interface Result<T> {
+  data: IEvent<T>[];
+}
+
+export interface JsonRpcResponse<T> {
+  jsonrpc: string;
+  result: Result<T>;
+}
+
+export const RequestStatus = {
+  INDEXED: 1,
+  SUCCESS: 2,
+  INVALID_URL: 3,
+  INVALID_PAYLOAD: 4,
+  UNREACHABLE: 5,
+};

rooch/sources/example_caller.move

@@ -28,8 +28,10 @@ module verity_test_foreign_module::example_caller {
     // Initiate the module with an empty vector of pending requests
     // Requests are managed in the caller to prevent other modules from impersonating the calling module, and spoofing new data.
     fun init(){
-      let params = account::borrow_mut_resource<GlobalParams>(@verity_test_foreign_module);
-      params.pending_requests = vector::empty<ObjectID>();
+      // let params = account::borrow_mut_resource<GlobalParams>(@verity_test_foreign_module); // account::borrow_mut_resource  in init throws an error on deployment
+      // params.pending_requests = vector::empty<ObjectID>();
+      let signer = moveos_std::signer::module_signer<GlobalParams>();
+      account::move_resource_to(&signer, GlobalParams { pending_requests: vector::empty<ObjectID>() });
     }
 
     public entry fun request_data(