Add Starknet RPC event indexing package

change/@apibara-indexer-rpc-stream-config.json

@@ -0,0 +1,7 @@
+{
+  "type": "patch",
+  "comment": "Allow indexers to use RPC stream configs",
+  "packageName": "@apibara/indexer",
+  "email": "loothero@provable.games",
+  "dependentChangeType": "patch"
+}

change/@apibara-starknet-rpc-4bb3abe9-7bf6-4477-add1-8177352f12ee.json

@@ -0,0 +1,7 @@
+{
+  "type": "minor",
+  "comment": "Add Starknet RPC event helpers",
+  "packageName": "@apibara/starknet-rpc",
+  "email": "loothero@provable.games",
+  "dependentChangeType": "patch"
+}

change/apibara-rpc-stream-runtime-client.json

@@ -0,0 +1,7 @@
+{
+  "type": "patch",
+  "comment": "Support RPC stream configs in the CLI runtime",
+  "packageName": "apibara",
+  "email": "loothero@provable.games",
+  "dependentChangeType": "patch"
+}

packages/cli/src/runtime/dev.ts

@@ -1,10 +1,10 @@
 import { ReloadIndexerRequest, runWithReconnect } from "@apibara/indexer";
-import { createAuthenticatedClient } from "@apibara/protocol";
 import { getRuntimeDataFromEnv } from "apibara/common";
 import { defineCommand, runMain } from "citty";
 import type { ConsolaInstance } from "consola";
 import { blueBright } from "picocolors";
 import { availableIndexers, createIndexer } from "./internal/app";
+import { createRuntimeClient } from "./internal/client";
 
 async function startIndexer(indexer: string, signal: AbortSignal) {
   let _logger: ConsolaInstance | undefined;
@@ -25,11 +25,11 @@ async function startIndexer(indexer: string, signal: AbortSignal) {
         return;
       }
 
-      const client = createAuthenticatedClient(
-        indexerInstance.streamConfig,
-        indexerInstance.options.streamUrl,
-        indexerInstance.options.clientOptions,
-      );
+      const client = createRuntimeClient({
+        streamConfig: indexerInstance.streamConfig,
+        streamUrl: indexerInstance.options.streamUrl,
+        clientOptions: indexerInstance.options.clientOptions,
+      });
 
       if (logger) {
         logger.info(`Indexer ${blueBright(indexer)} started`);

packages/cli/src/runtime/internal/client.ts

@@ -0,0 +1,27 @@
+import type { IndexerStreamConfig } from "@apibara/indexer";
+import {
+  type Client,
+  type CreateClientOptions,
+  createAuthenticatedClient,
+} from "@apibara/protocol";
+import { createRpcClient } from "@apibara/protocol/rpc";
+
+export function createRuntimeClient<TFilter, TBlock>({
+  streamConfig,
+  streamUrl,
+  clientOptions,
+}: {
+  streamConfig: IndexerStreamConfig<TFilter, TBlock>;
+  streamUrl?: string;
+  clientOptions?: CreateClientOptions;
+}): Client<TFilter, TBlock> {
+  if ("Request" in streamConfig) {
+    if (!streamUrl) {
+      throw new Error("streamUrl is required when using a DNA StreamConfig");
+    }
+
+    return createAuthenticatedClient(streamConfig, streamUrl, clientOptions);
+  }
+
+  return createRpcClient(streamConfig);
+}

packages/cli/src/runtime/project-info.ts

@@ -59,7 +59,7 @@ const startCommand = defineCommand({
         projectInfo.indexers[indexer] = {
           ...(projectInfo.indexers[indexer] ?? {}),
           [preset]: {
-            type: indexerInstance.streamConfig.name,
+            type: streamConfigType(indexerInstance.streamConfig),
             isFactory: indexerInstance.options.factory !== undefined,
           },
         };
@@ -75,6 +75,29 @@ const startCommand = defineCommand({
   },
 });
 
+function streamConfigType(streamConfig: unknown): string {
+  if (
+    typeof streamConfig === "object" &&
+    streamConfig !== null &&
+    "name" in streamConfig &&
+    typeof streamConfig.name === "string"
+  ) {
+    return streamConfig.name;
+  }
+
+  if (
+    typeof streamConfig === "object" &&
+    streamConfig !== null &&
+    "constructor" in streamConfig &&
+    typeof streamConfig.constructor === "function" &&
+    streamConfig.constructor.name
+  ) {
+    return streamConfig.constructor.name;
+  }
+
+  return "RpcStreamConfig";
+}
+
 export const mainCli = defineCommand({
   meta: {
     name: "write-project-info-runner",

packages/cli/src/runtime/start.ts

@@ -1,5 +1,4 @@
 import { ReloadIndexerRequest, runWithReconnect } from "@apibara/indexer";
-import { createAuthenticatedClient } from "@apibara/protocol";
 import {
   checkForUnknownArgs,
   getProcessedRuntimeConfig,
@@ -17,6 +16,7 @@ import {
   userEnvRuntimeConfig,
 } from "#apibara-internal-virtual/static-config";
 import { createIndexer } from "./internal/app";
+import { createRuntimeClient } from "./internal/client";
 
 const startCommand = defineCommand({
   meta: {
@@ -83,11 +83,11 @@ const startCommand = defineCommand({
           process.exit(1);
         }
 
-        const client = createAuthenticatedClient(
-          indexerInstance.streamConfig,
-          indexerInstance.options.streamUrl,
-          indexerInstance.options.clientOptions,
-        );
+        const client = createRuntimeClient({
+          streamConfig: indexerInstance.streamConfig,
+          streamUrl: indexerInstance.options.streamUrl,
+          clientOptions: indexerInstance.options.clientOptions,
+        });
 
         if (register) {
           consola.start("Registering from instrumentation");

packages/indexer/src/indexer.ts

@@ -15,6 +15,7 @@ import {
   type StreamDataResponse,
   type SystemMessage,
 } from "@apibara/protocol";
+import type { RpcStreamConfig } from "@apibara/protocol/rpc";
 import consola from "consola";
 import {
   type Hookable,
@@ -116,8 +117,7 @@ export type HandlerArgs<TBlock> = {
   abortSignal?: AbortSignal;
 };
 
-export type IndexerConfig<TFilter, TBlock> = {
-  streamUrl: string;
+export type BaseIndexerConfig<TFilter, TBlock> = {
   filter: TFilter;
   finality?: DataFinality;
   clientOptions?: CreateClientOptions;
@@ -130,27 +130,68 @@ export type IndexerConfig<TFilter, TBlock> = {
   debug?: boolean;
 } & IndexerStartingCursor;
 
-export type IndexerWithStreamConfig<TFilter, TBlock> = IndexerConfig<
+export type IndexerConfig<TFilter, TBlock> = BaseIndexerConfig<
+  TFilter,
+  TBlock
+> & {
+  streamUrl: string;
+};
+
+export type RpcIndexerConfig<TFilter, TBlock> = BaseIndexerConfig<
+  TFilter,
+  TBlock
+> & {
+  streamUrl?: string;
+};
+
+export type IndexerStreamConfig<TFilter, TBlock> =
+  | StreamConfig<TFilter, TBlock>
+  | RpcStreamConfig<TFilter, TBlock>;
+
+export type IndexerWithDnaStreamConfig<TFilter, TBlock> = IndexerConfig<
   TFilter,
   TBlock
 > & {
   streamConfig: StreamConfig<TFilter, TBlock>;
 };
 
+export type IndexerWithRpcStreamConfig<TFilter, TBlock> = RpcIndexerConfig<
+  TFilter,
+  TBlock
+> & {
+  streamConfig: RpcStreamConfig<TFilter, TBlock>;
+};
+
+export type IndexerWithStreamConfig<TFilter, TBlock> =
+  | IndexerWithDnaStreamConfig<TFilter, TBlock>
+  | IndexerWithRpcStreamConfig<TFilter, TBlock>;
+
 export function defineIndexer<TFilter, TBlock>(
   streamConfig: StreamConfig<TFilter, TBlock>,
-) {
+): (
+  config: IndexerConfig<TFilter, TBlock>,
+) => IndexerWithDnaStreamConfig<TFilter, TBlock>;
+
+export function defineIndexer<TFilter, TBlock>(
+  streamConfig: RpcStreamConfig<TFilter, TBlock>,
+): (
+  config: RpcIndexerConfig<TFilter, TBlock>,
+) => IndexerWithRpcStreamConfig<TFilter, TBlock>;
+
+export function defineIndexer<TFilter, TBlock>(
+  streamConfig: IndexerStreamConfig<TFilter, TBlock>,
+): unknown {
   return (
-    config: IndexerConfig<TFilter, TBlock>,
-  ): IndexerWithStreamConfig<TFilter, TBlock> => ({
+    config: IndexerConfig<TFilter, TBlock> | RpcIndexerConfig<TFilter, TBlock>,
+  ) => ({
     streamConfig,
     ...config,
   });
 }
 
 export interface Indexer<TFilter, TBlock> {
-  streamConfig: StreamConfig<TFilter, TBlock>;
-  options: IndexerConfig<TFilter, TBlock>;
+  streamConfig: IndexerStreamConfig<TFilter, TBlock>;
+  options: IndexerConfig<TFilter, TBlock> | RpcIndexerConfig<TFilter, TBlock>;
   hooks: Hookable<IndexerHooks<TFilter, TBlock>>;
 }
 
@@ -177,6 +218,20 @@ export function createIndexer<TFilter, TBlock>({
   return indexer;
 }
 
+function mergeFilterWithStreamConfig<TFilter, TBlock>(
+  streamConfig: IndexerStreamConfig<TFilter, TBlock>,
+  a: TFilter,
+  b: TFilter,
+): TFilter {
+  if ("mergeFilter" in streamConfig) {
+    return streamConfig.mergeFilter(a, b);
+  }
+
+  throw new Error(
+    "Factory mode requires a stream config with mergeFilter support.",
+  );
+}
+
 export interface ReconnectOptions {
   maxRetries?: number;
   retryDelay?: number;
@@ -434,7 +489,8 @@ export async function run<TFilter, TBlock>(
                   if (filter) {
                     // when filter is defined
                     // merge old and new filters
-                    mainFilter = indexer.streamConfig.mergeFilter(
+                    mainFilter = mergeFilterWithStreamConfig(
+                      indexer.streamConfig,
                       mainFilter,
                       filter,
                     );

packages/indexer/src/testing/index.ts

@@ -67,6 +67,14 @@ export function createVcr() {
           throw new Error("Cannot record cassette in CI");
         }
 
+        if (!("Request" in indexer.streamConfig)) {
+          throw new Error("VCR recording requires a DNA StreamConfig");
+        }
+
+        if (!indexer.options.streamUrl) {
+          throw new Error("VCR recording requires streamUrl");
+        }
+
         const client = createAuthenticatedClient(
           indexer.streamConfig,
           indexer.options.streamUrl,