protocol: refactor the new RPC-based stream

Author: fracek

examples/evm-client/package.json

@@ -12,8 +12,8 @@
   },
   "dependencies": {
     "@apibara/evm": "workspace:*",
+    "@apibara/evm-rpc": "workspace:*",
     "@apibara/protocol": "workspace:*",
-    "@apibara/rpc-evm": "workspace:*",
     "citty": "^0.1.6",
     "consola": "^3.4.2",
     "nice-grpc": "^2.1.8",
@@ -23,4 +23,4 @@
     "@types/node": "^20.12.12",
     "jiti": "^1.21.0"
   }
-}
+}

examples/evm-client/src/main-rpc.ts

@@ -1,8 +1,9 @@
 import assert from "node:assert";
-import { type EvmRpcFilter, createEvmRpcClient } from "@apibara/rpc-evm";
+import { EvmRpcStream, type Filter } from "@apibara/evm-rpc";
+import { createRpcClient } from "@apibara/protocol/rpc";
 import { defineCommand, runMain } from "citty";
 import consola from "consola";
-import type { Chain } from "viem";
+import { http, type Chain, createPublicClient } from "viem";
 import { mainnet, sepolia } from "viem/chains";
 
 const command = defineCommand({
@@ -19,7 +20,7 @@ const command = defineCommand({
     },
     network: {
       type: "string",
-      default: "sepolia",
+      default: "mainnet",
       description: "Network name (mainnet, sepolia)",
     },
     contract: {
@@ -29,13 +30,13 @@ const command = defineCommand({
     },
     startBlock: {
       type: "string",
-      default: "23857000",
+      default: "23911400",
       description: "Starting block number",
     },
     finality: {
       type: "string",
       default: "finalized",
-      description: "Block finality (finalized, accepted, pending)",
+      description: "Block finality (finalized, accepted)",
     },
   },
   async run({ args }) {
@@ -44,42 +45,40 @@ const command = defineCommand({
 
     const chain = (args.network === "mainnet" ? mainnet : sepolia) as Chain;
 
-    const client = createEvmRpcClient(args.rpcUrl, chain, {
-      clientConfig: {
-        retryDelay: 10000,
-        onFetchRequest(request) {
-          request
-            .clone()
-            .json()
-            .then((body) => {
-              if (Array.isArray(body)) {
-                consola.debug(`----->>> Batched ${body.length} requests`);
-              } else {
-                consola.debug("----->>> Single request");
-              }
-              // console.dir(body, { depth: null });
-            });
+    const viemClient = createPublicClient({
+      chain,
+      transport: http(args.rpcUrl, {
+        batch: {
+          wait: 10,
         },
-      },
+      }),
     });
 
-    try {
-      const status = await client.status();
-      consola.success("Connected to RPC endpoint");
-      consola.info("Current head:", status.currentHead?.orderKey);
-      consola.info("Finalized:", status.finalized?.orderKey);
-    } catch (error) {
-      consola.error("Failed to connect:", error);
-      process.exit(1);
-    }
+    const client = createRpcClient(
+      new EvmRpcStream(viemClient, {
+        getLogsRangeSize: 10n,
+      }),
+    );
+
+    const status = await client.status();
+    consola.success("Connected to RPC endpoint");
+    consola.info("Current head:", status.currentHead?.orderKey);
+    consola.info("Finalized:", status.finalized?.orderKey);
 
-    const filter: EvmRpcFilter = {
+    const filter: Filter = {
+      header: "always",
       logs: [
         {
-          address: args.contract as `0x${string}`,
-          topics: [
-            "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef" as `0x${string}`,
-          ],
+          id: 1,
+          address: "0xe0e0e08A6A4b9Dc7bD67BCB7aadE5cF48157d444",
+        },
+        {
+          id: 2,
+          address: "0xA37cc341634AFD9E0919D334606E676dbAb63E17",
+        },
+        {
+          id: 3,
+          address: "0xe0e0e08A6A4b9Dc7bD67BCB7aadE5cF48157d444",
         },
       ],
     };
@@ -88,74 +87,70 @@ const command = defineCommand({
     consola.info("Monitoring contract:", args.contract);
     consola.info("Finality:", args.finality);
 
-    try {
-      for await (const message of client.streamData({
-        filter: [filter],
-        finality: args.finality as "finalized" | "accepted" | "pending",
-        startingCursor: {
-          orderKey: BigInt(args.startBlock),
-        },
-      })) {
-        switch (message._tag) {
-          case "data": {
-            const { data, endCursor, finality, production } = message.data;
-
-            for (const block of data) {
-              assert(block !== null);
-              const logs = block.logs;
-
-              if (logs.length > 0) {
-                consola.info(
-                  `📦 Block ${block.header?.blockNumber} [${finality}/${production}]`,
-                );
-                consola.info("   Logs:", logs.length);
-
-                for (const log of logs) {
-                  consola.info("   🔔 Log");
-                  consola.info("      Tx:", log.transactionHash);
-                  consola.info("      Address:", log.address);
-                  consola.info("      Topics:", log.topics.join(", "));
-                  consola.info("      Log Index:", log.logIndex);
-                }
-              }
-            }
+    for await (const message of client.streamData({
+      filter: [filter],
+      finality: args.finality as "finalized" | "accepted" | "pending",
+      startingCursor: {
+        orderKey: BigInt(args.startBlock),
+      },
+    })) {
+      switch (message._tag) {
+        case "data": {
+          const { data, endCursor, finality, production } = message.data;
 
-            break;
-          }
+          for (const block of data) {
+            assert(block !== null);
+            const logs = block.logs;
 
-          case "invalidate": {
-            consola.warn(
-              "⚠️  Reorg detected! Invalidating to block",
-              message.invalidate.cursor?.orderKey,
+            consola.info(
+              `block ${block.header?.blockNumber} [${finality}/${production}]`,
             );
-            break;
+            consola.info("   logs:", logs.length);
+
+            for (const log of logs) {
+              // consola.info(log);
+              // consola.info("   🔔 Log");
+              // consola.info("      Tx:", log.transactionHash);
+              // consola.info("      Address:", log.address);
+              // consola.info("      Topics:", log.topics.join(", "));
+              // consola.info("      Log Index:", log.logIndex);
+            }
           }
 
-          case "finalize": {
-            consola.success(
-              "✅ Finalized up to block",
-              message.finalize.cursor?.orderKey,
-            );
-            break;
-          }
+          break;
+        }
 
-          case "heartbeat": {
-            consola.info("💓 Heartbeat");
-            break;
-          }
+        case "invalidate": {
+          consola.warn(
+            "invalidating to block",
+            message.invalidate.cursor?.orderKey,
+          );
+          break;
+        }
 
-          case "systemMessage": {
-            const output = message.systemMessage.output;
-            if (output._tag === "stderr") {
-              consola.error("System error:", output.stderr);
-            }
-            break;
+        case "finalize": {
+          consola.success(
+            "finalized up to block",
+            message.finalize.cursor?.orderKey,
+          );
+          break;
+        }
+
+        case "heartbeat": {
+          consola.info("heartbeat");
+          break;
+        }
+
+        case "systemMessage": {
+          const output = message.systemMessage.output;
+          if (output._tag === "stderr") {
+            consola.error(output.stderr);
+          } else {
+            consola.info(output.stdout);
           }
+          break;
         }
       }
-    } catch (error) {
-      consola.error("Stream error:", error);
-      process.exit(1);
     }
   },
 });

packages/evm-rpc/README.md

@@ -0,0 +1,3 @@
+# `@apibara/evm-rpc`
+
+This package provides a `EvmRpcStream` that implements the DNA stream using standard JSON-RPC endpoints.

packages/rpc-evm/build.config.ts packages/evm-rpc/build.config.tspackages/rpc-evm/build.config.ts renamed to packages/evm-rpc/build.config.ts

@@ -1,5 +1,5 @@
 {
-  "name": "@apibara/rpc-evm",
+  "name": "@apibara/evm-rpc",
   "version": "2.1.0-beta.40",
   "type": "module",
   "files": [
@@ -33,4 +33,4 @@
     "@apibara/evm": "workspace:*",
     "viem": "^2.13.2"
   }
-}
+}

packages/rpc-evm/package.json packages/evm-rpc/package.jsonpackages/rpc-evm/package.json renamed to packages/evm-rpc/package.json

@@ -3,11 +3,11 @@ import type {
   Log as DnaLog,
 } from "@apibara/evm";
 
-export type EvmRpcBlock = {
+export type BlockHeader = DnaBlockHeader;
+
+export type Log = Omit<DnaLog, "logIndexInTransaction">;
+
+export type Block = {
   header: BlockHeader;
   logs: Log[];
 };
-
-export type BlockHeader = DnaBlockHeader;
-
-export type Log = DnaLog;

packages/rpc-evm/src/block.ts packages/evm-rpc/src/block.tspackages/rpc-evm/src/block.ts renamed to packages/evm-rpc/src/block.ts

@@ -0,0 +1,61 @@
+import type { LogFilter as DnaLogFilter, HeaderFilter } from "@apibara/evm";
+import type { ValidateFilterResult } from "@apibara/protocol/rpc";
+import { isHex } from "viem";
+
+export type Filter = {
+  header?: HeaderFilter;
+  logs: LogFilter[];
+};
+
+export type LogFilter = Pick<
+  DnaLogFilter,
+  "address" | "topics" | "strict" | "id"
+>;
+
+export function validateFilter(filter: Filter): ValidateFilterResult {
+  if (!filter.logs || filter.logs.length === 0) {
+    return { valid: false, error: "Missing logs filter" };
+  }
+
+  let logFilterIndex = 0;
+  for (const logFilter of filter.logs ?? []) {
+    if (
+      logFilter.address === undefined &&
+      (logFilter.topics?.length ?? 0) === 0
+    ) {
+      return {
+        valid: false,
+        error: `Must provide at least one address or topic in log filter at position ${logFilterIndex}`,
+      };
+    }
+
+    if (logFilter.address) {
+      if (!isHex(logFilter.address)) {
+        return {
+          valid: false,
+          error: "Invalid address format. Expected 0x-prefixed hex string",
+        };
+      }
+    }
+
+    if (logFilter.topics) {
+      for (let i = 0; i < logFilter.topics.length; i++) {
+        const topic = logFilter.topics[i];
+        if (topic === null) {
+          continue;
+        }
+
+        if (!isHex(topic)) {
+          return {
+            valid: false,
+            error: `Invalid topic at index ${i}: ${topic}. Must be null or a 0x-prefixed hex string`,
+          };
+        }
+      }
+    }
+
+    logFilterIndex++;
+  }
+
+  return { valid: true };
+}

packages/evm-rpc/src/filter.ts

@@ -0,0 +1,3 @@
+export { EvmRpcStream } from "./stream-config";
+export type { Filter, LogFilter } from "./filter";
+export type { Block, BlockHeader, Log } from "./block";