use custom ponder transport

ponder-sh · Nov 10, 2023 · 3addeae · 3addeae
1 parent 603fc72
commit 3addeae
Show file tree

Hide file tree

Showing 2 changed files with 29 additions and 12 deletions.
diff --git a/packages/core/src/indexing/service.test.ts b/packages/core/src/indexing/service.test.ts
@@ -2,7 +2,7 @@ import { http } from "viem";
 import { beforeEach, expect, test, vi } from "vitest";
 
 import { usdcContractConfig } from "@/_test/constants";
-import { setupIndexingStore } from "@/_test/setup";
+import { setupIndexingStore, setupSyncStore } from "@/_test/setup";
 import type { IndexingFunctions } from "@/build/functions";
 import { LogEventMetadata } from "@/config/abi";
 import { Source } from "@/config/sources";
@@ -12,6 +12,7 @@ import { SyncGateway } from "@/sync-gateway/service";
 import { IndexingService } from "./service";
 
 beforeEach((context) => setupIndexingStore(context));
+beforeEach((context) => setupSyncStore(context));
 
 const networks = [
   {
@@ -109,10 +110,11 @@ beforeEach(() => {
 });
 
 test("processEvents() calls getEvents with sequential timestamp ranges", async (context) => {
-  const { common, indexingStore } = context;
+  const { common, syncStore, indexingStore } = context;
 
   const service = new IndexingService({
     common,
+    syncStore,
     indexingStore,
     syncGatewayService,
     sources,
@@ -147,10 +149,11 @@ test("processEvents() calls getEvents with sequential timestamp ranges", async (
 });
 
 test("processEvents() calls indexing functions with correct arguments", async (context) => {
-  const { common, indexingStore } = context;
+  const { common, syncStore, indexingStore } = context;
 
   const service = new IndexingService({
     common,
+    syncStore,
     indexingStore,
     syncGatewayService,
     sources,
@@ -183,10 +186,11 @@ test("processEvents() calls indexing functions with correct arguments", async (c
 });
 
 test("processEvents() model methods insert data into the indexing store", async (context) => {
-  const { common, indexingStore } = context;
+  const { common, syncStore, indexingStore } = context;
 
   const service = new IndexingService({
     common,
+    syncStore,
     indexingStore,
     syncGatewayService,
     sources,
@@ -207,10 +211,11 @@ test("processEvents() model methods insert data into the indexing store", async
 });
 
 test("processEvents() updates event count metrics", async (context) => {
-  const { common, indexingStore } = context;
+  const { common, syncStore, indexingStore } = context;
 
   const service = new IndexingService({
     common,
+    syncStore,
     indexingStore,
     syncGatewayService,
     sources,
@@ -248,10 +253,11 @@ test("processEvents() updates event count metrics", async (context) => {
 });
 
 test("reset() reloads the indexing store", async (context) => {
-  const { common, indexingStore } = context;
+  const { common, syncStore, indexingStore } = context;
 
   const service = new IndexingService({
     common,
+    syncStore,
     indexingStore,
     syncGatewayService,
     sources,
@@ -283,10 +289,11 @@ test("reset() reloads the indexing store", async (context) => {
 });
 
 test("handleReorg() updates ponder_handlers_latest_processed_timestamp metric", async (context) => {
-  const { common, indexingStore } = context;
+  const { common, syncStore, indexingStore } = context;
 
   const service = new IndexingService({
     common,
+    syncStore,
     indexingStore,
     syncGatewayService,
     sources,
@@ -314,10 +321,11 @@ test("handleReorg() updates ponder_handlers_latest_processed_timestamp metric",
 });
 
 test("handleReorg() reverts the indexing store", async (context) => {
-  const { common, indexingStore } = context;
+  const { common, syncStore, indexingStore } = context;
 
   const service = new IndexingService({
     common,
+    syncStore,
     indexingStore,
     syncGatewayService,
     sources,
@@ -339,10 +347,11 @@ test("handleReorg() reverts the indexing store", async (context) => {
 });
 
 test("handleReorg() does nothing if there is a user error", async (context) => {
-  const { common, indexingStore } = context;
+  const { common, syncStore, indexingStore } = context;
 
   const service = new IndexingService({
     common,
+    syncStore,
     indexingStore,
     syncGatewayService,
     sources,
@@ -368,10 +377,11 @@ test("handleReorg() does nothing if there is a user error", async (context) => {
 });
 
 test("handleReorg() processes the correct range of events after a reorg", async (context) => {
-  const { common, indexingStore } = context;
+  const { common, syncStore, indexingStore } = context;
 
   const service = new IndexingService({
     common,
+    syncStore,
     indexingStore,
     syncGatewayService,
     sources,
@@ -407,10 +417,11 @@ test("handleReorg() processes the correct range of events after a reorg", async
 });
 
 test("handleReorg() updates ponder_handlers_latest_processed_timestamp metric", async (context) => {
-  const { common, indexingStore } = context;
+  const { common, syncStore, indexingStore } = context;
 
   const service = new IndexingService({
     common,
+    syncStore,
     indexingStore,
     syncGatewayService,
     sources,

diff --git a/packages/core/src/indexing/service.ts b/packages/core/src/indexing/service.ts
@@ -12,6 +12,7 @@ import type { IndexingStore, ModelInstance } from "@/indexing-store/store";
 import type { Common } from "@/Ponder";
 import type { Schema } from "@/schema/types";
 import type { LogEvent, SyncGateway } from "@/sync-gateway/service";
+import { SyncStore } from "@/sync-store/store";
 import type { Model } from "@/types/model";
 import { formatShortDate } from "@/utils/date";
 import { prettyPrint } from "@/utils/print";
@@ -21,6 +22,7 @@ import { wait } from "@/utils/wait";
 import { buildModels } from "./model";
 import { ponderActions } from "./ponderActions";
 import { getStackTrace } from "./trace";
+import { ponderTransport } from "./transport";
 
 type IndexingEvents = {
   eventsProcessed: { toTimestamp: number };
@@ -71,12 +73,14 @@ export class IndexingService extends Emittery<IndexingEvents> {
 
   constructor({
     common,
+    syncStore,
     indexingStore,
     syncGatewayService,
     networks,
     sources,
   }: {
     common: Common;
+    syncStore: SyncStore;
     indexingStore: IndexingStore;
     syncGatewayService: SyncGateway;
     networks: Config["networks"];
@@ -93,6 +97,7 @@ export class IndexingService extends Emittery<IndexingEvents> {
     this.contexts = buildContexts(
       sources,
       networks,
+      syncStore,
       ponderActions(() => this.currentEventBlockNumber)
     );
   }
@@ -537,6 +542,7 @@ export class IndexingService extends Emittery<IndexingEvents> {
 const buildContexts = (
   sources: Source[],
   networks: Config["networks"],
+  syncStore: SyncStore,
   actions: ReturnType<typeof ponderActions>
 ) => {
   const contexts: Record<
@@ -563,7 +569,7 @@ const buildContexts = (
       chains.mainnet;
 
     const client = createClient({
-      transport: network.transport,
+      transport: ponderTransport({ transport: network.transport, syncStore }),
       chain: { ...defaultChain, name: network.name, id: network.chainId },
     });