WIP refactor(ensindexer): use local ponder client

README.md

@@ -175,6 +175,10 @@ TypeScript library for interacting with the [ENSRainbow API](apps/ensrainbow).
 
 Shared Drizzle schema definitions used by ENSNode
 
+### [`packages/ponder-sdk`](packages/ponder-sdk)
+
+A set of utilities for interacting with a Ponder app.
+
 ### [`packages/ponder-subgraph`](packages/ponder-subgraph)
 
 Subgraph-compatible GraphQL API

apps/ensindexer/ponder/local-client/chain-indexing-status-snapshot.ts

@@ -0,0 +1,161 @@
+import {
+  type ChainIdString,
+  ChainIndexingConfigTypeIds,
+  ChainIndexingStatusIds,
+  type ChainIndexingStatusSnapshot,
+  createIndexingConfig,
+  deserializeBlockRef,
+  deserializeChainId,
+  deserializeChainIndexingStatusSnapshot,
+  deserializeNonNegativeInteger,
+  type SerializedChainIndexingStatusSnapshot,
+  type SerializedChainIndexingStatusSnapshotBackfill,
+  type SerializedChainIndexingStatusSnapshotCompleted,
+  type SerializedChainIndexingStatusSnapshotFollowing,
+  type SerializedChainIndexingStatusSnapshotQueued,
+} from "@ensnode/ensnode-sdk";
+import type {
+  ChainBlockRefs,
+  ChainMetadata,
+  ChainName,
+  PonderMetricsResponse,
+  PonderStatusResponse,
+  UnvalidatedChainMetadata,
+} from "@ensnode/ponder-sdk";
+
+/**
+ * Create {@link ChainIndexingStatusSnapshot} for the indexed chain metadata.
+ */
+export function createChainIndexingSnapshot(
+  chainMetadata: ChainMetadata,
+): ChainIndexingStatusSnapshot {
+  const {
+    config: chainBlocksConfig,
+    backfillEndBlock: chainBackfillEndBlock,
+    isSyncComplete,
+    isSyncRealtime,
+    syncBlock: chainSyncBlock,
+    statusBlock: chainStatusBlock,
+  } = chainMetadata;
+
+  const { startBlock, endBlock } = chainBlocksConfig;
+  const config = createIndexingConfig(startBlock, endBlock);
+
+  // In omnichain ordering, if the startBlock is the same as the
+  // status block, the chain has not started yet.
+  if (chainBlocksConfig.startBlock.number === chainStatusBlock.number) {
+    return deserializeChainIndexingStatusSnapshot({
+      chainStatus: ChainIndexingStatusIds.Queued,
+      config,
+    } satisfies SerializedChainIndexingStatusSnapshotQueued);
+  }
+
+  if (isSyncComplete) {
+    if (config.configType !== ChainIndexingConfigTypeIds.Definite) {
+      throw new Error(
+        `The '${ChainIndexingStatusIds.Completed}' indexing status can be only created with the '${ChainIndexingConfigTypeIds.Definite}' indexing config type.`,
+      );
+    }
+
+    return deserializeChainIndexingStatusSnapshot({
+      chainStatus: ChainIndexingStatusIds.Completed,
+      latestIndexedBlock: chainStatusBlock,
+      config,
+    } satisfies SerializedChainIndexingStatusSnapshotCompleted);
+  }
+
+  if (isSyncRealtime) {
+    if (config.configType !== ChainIndexingConfigTypeIds.Indefinite) {
+      throw new Error(
+        `The '${ChainIndexingStatusIds.Following}' indexing status can be only created with the '${ChainIndexingConfigTypeIds.Indefinite}' indexing config type.`,
+      );
+    }
+
+    return deserializeChainIndexingStatusSnapshot({
+      chainStatus: ChainIndexingStatusIds.Following,
+      latestIndexedBlock: chainStatusBlock,
+      latestKnownBlock: chainSyncBlock,
+      config: {
+        configType: config.configType,
+        startBlock: config.startBlock,
+      },
+    } satisfies SerializedChainIndexingStatusSnapshotFollowing);
+  }
+
+  return deserializeChainIndexingStatusSnapshot({
+    chainStatus: ChainIndexingStatusIds.Backfill,
+    latestIndexedBlock: chainStatusBlock,
+    backfillEndBlock: chainBackfillEndBlock,
+    config,
+  } satisfies SerializedChainIndexingStatusSnapshotBackfill);
+}
+
+/**
+ * Create serialized chain indexing snapshots.
+ *
+ * The output of this function is required for
+ * calling {@link createOmnichainIndexingSnapshot}.
+ */
+export function createSerializedChainSnapshots(
+  chainIds: ChainIdString[],
+  chainsBlockRefs: Map<ChainName, ChainBlockRefs>,
+  metrics: PonderMetricsResponse,
+  status: PonderStatusResponse,
+): Record<ChainIdString, SerializedChainIndexingStatusSnapshot> {
+  const serializedChainIndexingStatusSnapshots = {} as Record<
+    ChainIdString,
+    ChainIndexingStatusSnapshot
+  >;
+
+  // collect unvalidated chain metadata for each indexed chain
+  for (const chainId of chainIds) {
+    const chainBlockRefs = chainsBlockRefs.get(chainId);
+
+    const statusChainId = deserializeChainId(`${status[chainId]?.id}`);
+
+    const backfillEndBlock = deserializeBlockRef(chainBlockRefs?.backfillEndBlock);
+
+    const syncBlock = deserializeBlockRef({
+      number: metrics.getValue("ponder_sync_block", { chain: chainId }),
+      timestamp: metrics.getValue("ponder_sync_block_timestamp", { chain: chainId }),
+    });
+
+    const statusBlock = deserializeBlockRef({
+      number: status[chainId]?.block.number,
+      timestamp: status[chainId]?.block.timestamp,
+    });
+
+    const historicalTotalBlocks = deserializeNonNegativeInteger(
+      metrics.getValue("ponder_historical_total_blocks", {
+        chain: chainId,
+      }),
+    );
+
+    const isSyncComplete = metrics.getValue("ponder_sync_is_complete", { chain: chainId });
+
+    const isSyncRealtime = metrics.getValue("ponder_sync_is_realtime", { chain: chainId });
+
+    const config = {
+      startBlock: deserializeBlockRef(chainBlockRefs?.config.startBlock),
+      endBlock:
+        chainBlockRefs?.config.endBlock === null
+          ? null
+          : deserializeBlockRef(chainBlockRefs?.config.endBlock),
+    };
+
+    const chainMetadata = {
+      chainId: statusChainId,
+      isSyncComplete: String(isSyncComplete) === "1",
+      isSyncRealtime: String(isSyncRealtime) === "1",
+      config,
+      backfillEndBlock,
+      historicalTotalBlocks,
+      syncBlock,
+      statusBlock,
+    } satisfies UnvalidatedChainMetadata;
+
+    serializedChainIndexingStatusSnapshots[chainId] = createChainIndexingSnapshot(chainMetadata);
+  }
+
+  return serializedChainIndexingStatusSnapshots;
+}

apps/ensindexer/ponder/local-client/index.ts

@@ -0,0 +1 @@
+export * from "./local-ponder-client";

apps/ensindexer/ponder/local-client/local-ponder-client.ts

@@ -0,0 +1,97 @@
+import { publicClients } from "ponder:api";
+import type { PublicClient } from "viem";
+
+import {
+  deserializeOmnichainIndexingStatusSnapshot,
+  type OmnichainIndexingStatusSnapshot,
+} from "@ensnode/ensnode-sdk";
+import {
+  buildHistoricalTotalBlocksForChains,
+  type ChainBlockRefs,
+  type ChainName,
+  getChainsBlockRefs,
+  getChainsBlockrange,
+  PonderClient,
+  type PonderMetricsResponse,
+} from "@ensnode/ponder-sdk";
+
+import { createSerializedChainSnapshots } from "./chain-indexing-status-snapshot";
+import ponderConfig from "./config";
+import { createSerializedOmnichainIndexingStatusSnapshot } from "./omnichain-indexing-status-snapshot";
+
+export class LocalPonderClient {
+  /**
+   * Cached Chain Block Refs
+   *
+   * {@link ChainBlockRefs} for each indexed chain.
+   */
+  private chainsBlockRefs = new Map<ChainName, ChainBlockRefs>();
+
+  private readonly ponderClient: PonderClient;
+
+  constructor(ponderApplicationUrl: URL) {
+    this.ponderClient = new PonderClient(ponderApplicationUrl);
+  }
+
+  public async buildCrossChainIndexingStatusSnapshot(): Promise<OmnichainIndexingStatusSnapshot> {
+    const [metrics, status] = await Promise.all([
+      this.ponderClient.metrics(),
+      this.ponderClient.status(),
+    ]);
+
+    const chainsBlockRefs = await this.getChainsBlockRefsCached(metrics);
+
+    // create serialized chain indexing snapshot for each indexed chain
+    const serializedChainSnapshots = createSerializedChainSnapshots(
+      this.indexedChainNames,
+      chainsBlockRefs,
+      metrics,
+      status,
+    );
+
+    const serializedOmnichainSnapshot =
+      createSerializedOmnichainIndexingStatusSnapshot(serializedChainSnapshots);
+
+    return deserializeOmnichainIndexingStatusSnapshot(serializedOmnichainSnapshot);
+  }
+
+  /**
+   * Get cached {@link IndexedChainBlockRefs} for indexed chains.
+   *
+   * Guaranteed to include {@link ChainBlockRefs} for each indexed chain.
+   *
+   * Note: performs a network request only once and caches response to
+   * re-use it for further `getChainsBlockRefs` calls.
+   *
+   * @throws when RPC calls fail or data model invariants are not met.
+   */
+  private async getChainsBlockRefsCached(
+    metrics: PonderMetricsResponse,
+  ): Promise<Map<ChainName, ChainBlockRefs>> {
+    // early-return the cached chain block refs
+    if (this.chainsBlockRefs.size > 0) {
+      return this.chainsBlockRefs;
+    }
+
+    this.chainsBlockRefs = await getChainsBlockRefs(
+      this.indexedChainNames,
+      getChainsBlockrange(this.ponderConfig),
+      buildHistoricalTotalBlocksForChains(this.indexedChainNames, metrics),
+      this.publicClients,
+    );
+
+    return this.chainsBlockRefs;
+  }
+
+  private get ponderConfig() {
+    return ponderConfig;
+  }
+
+  private get publicClients(): Record<ChainName, PublicClient> {
+    return publicClients;
+  }
+
+  private get indexedChainNames(): ChainName[] {
+    return Object.keys(this.ponderConfig.chains) as ChainName[];
+  }
+}

apps/ensindexer/ponder/local-client/omnichain-indexing-status-snapshot.ts

@@ -0,0 +1,70 @@
+import {
+  type ChainIdString,
+  type ChainIndexingStatusSnapshotForOmnichainIndexingStatusSnapshotBackfill,
+  getOmnichainIndexingCursor,
+  getOmnichainIndexingStatus,
+  OmnichainIndexingStatusIds,
+  type SerializedChainIndexingStatusSnapshot,
+  type SerializedChainIndexingStatusSnapshotCompleted,
+  type SerializedChainIndexingStatusSnapshotQueued,
+  type SerializedOmnichainIndexingStatusSnapshot,
+  type SerializedOmnichainIndexingStatusSnapshotBackfill,
+  type SerializedOmnichainIndexingStatusSnapshotCompleted,
+  type SerializedOmnichainIndexingStatusSnapshotFollowing,
+  type SerializedOmnichainIndexingStatusSnapshotUnstarted,
+} from "@ensnode/ensnode-sdk";
+
+/**
+ * Create Serialized Omnichain Indexing Snapshot
+ *
+ * Creates {@link SerializedOmnichainIndexingStatusSnapshot} from serialized chain snapshots.
+ */
+export function createSerializedOmnichainIndexingStatusSnapshot(
+  serializedChainSnapshots: Record<ChainIdString, SerializedChainIndexingStatusSnapshot>,
+): SerializedOmnichainIndexingStatusSnapshot {
+  const chains = Object.values(serializedChainSnapshots);
+  const omnichainStatus = getOmnichainIndexingStatus(chains);
+  const omnichainIndexingCursor = getOmnichainIndexingCursor(chains);
+
+  switch (omnichainStatus) {
+    case OmnichainIndexingStatusIds.Unstarted: {
+      return {
+        omnichainStatus: OmnichainIndexingStatusIds.Unstarted,
+        chains: serializedChainSnapshots as Record<
+          ChainIdString,
+          SerializedChainIndexingStatusSnapshotQueued
+        >, // forcing the type here, will be validated in the following 'check' step
+        omnichainIndexingCursor,
+      } satisfies SerializedOmnichainIndexingStatusSnapshotUnstarted;
+    }
+
+    case OmnichainIndexingStatusIds.Backfill: {
+      return {
+        omnichainStatus: OmnichainIndexingStatusIds.Backfill,
+        chains: serializedChainSnapshots as Record<
+          ChainIdString,
+          ChainIndexingStatusSnapshotForOmnichainIndexingStatusSnapshotBackfill
+        >, // forcing the type here, will be validated in the following 'check' step
+        omnichainIndexingCursor,
+      } satisfies SerializedOmnichainIndexingStatusSnapshotBackfill;
+    }
+
+    case OmnichainIndexingStatusIds.Completed: {
+      return {
+        omnichainStatus: OmnichainIndexingStatusIds.Completed,
+        chains: serializedChainSnapshots as Record<
+          ChainIdString,
+          SerializedChainIndexingStatusSnapshotCompleted
+        >, // forcing the type here, will be validated in the following 'check' step
+        omnichainIndexingCursor,
+      } satisfies SerializedOmnichainIndexingStatusSnapshotCompleted;
+    }
+
+    case OmnichainIndexingStatusIds.Following:
+      return {
+        omnichainStatus: OmnichainIndexingStatusIds.Following,
+        chains: serializedChainSnapshots,
+        omnichainIndexingCursor,
+      } satisfies SerializedOmnichainIndexingStatusSnapshotFollowing;
+  }
+}

apps/ensindexer/ponder/ponder.config.ts

@@ -3,7 +3,8 @@ import config from "@/config";
 import { prettyPrintJson } from "@ensnode/ensnode-sdk/internal";
 
 import { redactENSIndexerConfig } from "@/config/redact";
-import ponderConfig from "@/ponder/config";
+
+import ponderConfig from "./local-client/config";
 
 ////////
 // Log redacted ENSIndexerConfig for debugging.

apps/ensindexer/ponder/src/api/handlers/ensnode-api.ts

@@ -1,6 +1,5 @@
 import config from "@/config";
 
-import { publicClients } from "ponder:api";
 import { getUnixTime } from "date-fns";
 import { Hono } from "hono";
 
@@ -38,7 +37,7 @@ app.get("/indexing-status", async (c) => {
   let omnichainSnapshot: OmnichainIndexingStatusSnapshot | undefined;
 
   try {
-    omnichainSnapshot = await buildOmnichainIndexingStatusSnapshot(publicClients);
+    omnichainSnapshot = await buildOmnichainIndexingStatusSnapshot();
   } catch (error) {
     const errorMessage = error instanceof Error ? error.message : "Unknown error";
     console.error(`Omnichain snapshot is currently not available: ${errorMessage}`);