ENSDb Writer Worker for ENSIndexer

.changeset/afraid-ducks-strive.md

@@ -0,0 +1,5 @@
+---
+"ensindexer": minor
+---
+
+Introduced `EnsDbClient` and `EnsDbWriterWorker` to enable storing metadata in ENSDb.

apps/ensapi/package.json

@@ -53,7 +53,7 @@
     "graphql-yoga": "^5.16.0",
     "hono": "catalog:",
     "p-memoize": "^8.0.0",
-    "p-retry": "^7.1.0",
+    "p-retry": "catalog:",
     "pg-connection-string": "catalog:",
     "pino": "catalog:",
     "ponder-enrich-gql-docs-middleware": "^0.1.3",

apps/ensindexer/package.json

@@ -29,11 +29,14 @@
     "@ensnode/ensnode-sdk": "workspace:*",
     "@ensnode/ensrainbow-sdk": "workspace:*",
     "@ensnode/ponder-sdk": "workspace:*",
+    "@ponder/client": "catalog:",
     "caip": "catalog:",
     "date-fns": "catalog:",
     "deepmerge-ts": "^7.1.5",
     "dns-packet": "^5.6.1",
+    "drizzle-orm": "catalog:",
     "pg-connection-string": "catalog:",
+    "p-retry": "catalog:",
     "hono": "catalog:",
     "ponder": "catalog:",
     "viem": "catalog:",
@@ -43,6 +46,7 @@
     "@ensnode/shared-configs": "workspace:*",
     "@types/dns-packet": "^5.6.5",
     "@types/node": "catalog:",
+    "@types/pg": "8.16.0",
     "typescript": "catalog:",
     "vitest": "catalog:"
   }

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

@@ -14,11 +14,9 @@ import {
 } from "@ensnode/ensnode-sdk";
 
 import { buildENSIndexerPublicConfig } from "@/config/public";
-import { IndexingStatusBuilder } from "@/lib/indexing-status-builder";
-import { localPonderClient } from "@/lib/local-ponder-client";
+import { indexingStatusBuilder } from "@/lib/indexing-status-builder/singleton";
 
 const app = new Hono();
-const indexingStatusBuilder = new IndexingStatusBuilder(localPonderClient);
 
 // include ENSIndexer Public Config endpoint
 app.get("/config", async (c) => {

apps/ensindexer/ponder/src/api/index.ts

@@ -5,8 +5,16 @@ import { cors } from "hono/cors";
 
 import type { ErrorResponse } from "@ensnode/ensnode-sdk";
 
+import { startEnsDbWriterWorker } from "@/lib/ensdb-writer-worker/singleton";
+
 import ensNodeApi from "./handlers/ensnode-api";
 
+// The entry point for the ENSDb Writer Worker. It must be placed inside
+// the `api` directory of the Ponder app to avoid the following build issue:
+// Error: Invalid dependency graph. Config, schema, and indexing function files
+// cannot import objects from the API function file "src/api/index.ts".
+startEnsDbWriterWorker();
+
 const app = new Hono();
 
 // set the X-ENSIndexer-Version header to the current version

apps/ensindexer/src/lib/ensdb-client/drizzle.ts

@@ -0,0 +1,26 @@
+// This file is based on `packages/ponder-subgraph/src/drizzle.ts` file.
+// We currently duplicate the makeDrizzle function, as we don't have
+// a shared package for backend code yet. When we do, we can move
+// this function to the shared package and import it in both places.
+import { setDatabaseSchema } from "@ponder/client";
+import { drizzle } from "drizzle-orm/node-postgres";
+
+type Schema = { [name: string]: unknown };
+
+/**
+ * Makes a Drizzle DB object.
+ */
+export const makeDrizzle = <SCHEMA extends Schema>({
+  schema,
+  databaseUrl,
+  databaseSchema,
+}: {
+  schema: SCHEMA;
+  databaseUrl: string;
+  databaseSchema: string;
+}) => {
+  // monkeypatch schema onto tables
+  setDatabaseSchema(schema, databaseSchema);
+
+  return drizzle(databaseUrl, { schema, casing: "snake_case" });
+};

apps/ensindexer/src/lib/ensdb-client/ensdb-client.mock.ts

@@ -0,0 +1,66 @@
+import {
+  type BlockRef,
+  ChainIndexingStatusIds,
+  CrossChainIndexingStrategyIds,
+  type EnsIndexerPublicConfig,
+  OmnichainIndexingStatusIds,
+  PluginName,
+  RangeTypeIds,
+  type SerializedCrossChainIndexingStatusSnapshot,
+} from "@ensnode/ensnode-sdk";
+
+export const earlierBlockRef = {
+  timestamp: 1672531199,
+  number: 1024,
+} as const satisfies BlockRef;
+
+export const laterBlockRef = {
+  timestamp: 1672531200,
+  number: 1025,
+} as const satisfies BlockRef;
+
+export const databaseUrl = "postgres://user:pass@localhost:5432/ensdb";
+
+export const databaseSchemaName = "public";
+
+export const publicConfig = {
+  databaseSchemaName,
+  labelSet: {
+    labelSetId: "subgraph",
+    labelSetVersion: 0,
+  },
+  indexedChainIds: new Set([1]),
+  isSubgraphCompatible: true,
+  namespace: "mainnet",
+  plugins: [PluginName.Subgraph],
+  versionInfo: {
+    nodejs: "v22.10.12",
+    ponder: "0.11.25",
+    ensDb: "0.32.0",
+    ensIndexer: "0.32.0",
+    ensNormalize: "1.11.1",
+    ensRainbow: "0.32.0",
+    ensRainbowSchema: 2,
+  },
+} satisfies EnsIndexerPublicConfig;
+
+export const serializedSnapshot = {
+  strategy: CrossChainIndexingStrategyIds.Omnichain,
+  slowestChainIndexingCursor: earlierBlockRef.timestamp,
+  snapshotTime: earlierBlockRef.timestamp + 20,
+  omnichainSnapshot: {
+    omnichainStatus: OmnichainIndexingStatusIds.Following,
+    chains: {
+      "1": {
+        chainStatus: ChainIndexingStatusIds.Following,
+        config: {
+          rangeType: RangeTypeIds.LeftBounded,
+          startBlock: earlierBlockRef,
+        },
+        latestIndexedBlock: earlierBlockRef,
+        latestKnownBlock: laterBlockRef,
+      },
+    },
+    omnichainIndexingCursor: earlierBlockRef.timestamp,
+  },
+} satisfies SerializedCrossChainIndexingStatusSnapshot;

apps/ensindexer/src/lib/ensdb-client/ensdb-client.test.ts

@@ -0,0 +1,200 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+import { ensNodeMetadata } from "@ensnode/ensnode-schema";
+import {
+  deserializeCrossChainIndexingStatusSnapshot,
+  EnsNodeMetadataKeys,
+  serializeCrossChainIndexingStatusSnapshot,
+  serializeEnsIndexerPublicConfig,
+} from "@ensnode/ensnode-sdk";
+
+import { makeDrizzle } from "./drizzle";
+import { EnsDbClient } from "./ensdb-client";
+import * as ensDbClientMock from "./ensdb-client.mock";
+
+// Mock the config module to prevent it from trying to load actual environment variables during tests
+vi.mock("@/config", () => ({ default: {} }));
+
+// Mock the makeDrizzle function to return a mock database instance
+vi.mock("./drizzle", () => ({ makeDrizzle: vi.fn() }));
+
+describe("EnsDbClient", () => {
+  // Mock database query results and methods
+  const selectResult = { current: [] as Array<{ value: unknown }> };
+  const whereMock = vi.fn(async () => selectResult.current);
+  const fromMock = vi.fn(() => ({ where: whereMock }));
+  const selectMock = vi.fn(() => ({ from: fromMock }));
+  const onConflictDoUpdateMock = vi.fn(async () => undefined);
+  const valuesMock = vi.fn(() => ({ onConflictDoUpdate: onConflictDoUpdateMock }));
+  const insertMock = vi.fn(() => ({ values: valuesMock }));
+  const dbMock = { select: selectMock, insert: insertMock };
+
+  beforeEach(() => {
+    selectResult.current = [];
+    whereMock.mockClear();
+    fromMock.mockClear();
+    selectMock.mockClear();
+    onConflictDoUpdateMock.mockClear();
+    valuesMock.mockClear();
+    insertMock.mockClear();
+    vi.mocked(makeDrizzle).mockReturnValue(dbMock as unknown as ReturnType<typeof makeDrizzle>);
+  });
+
+  describe("getEnsDbVersion", () => {
+    it("returns undefined when no record exists", async () => {
+      // arrange
+      const client = new EnsDbClient(
+        ensDbClientMock.databaseUrl,
+        ensDbClientMock.databaseSchemaName,
+      );
+
+      // act & assert
+      await expect(client.getEnsDbVersion()).resolves.toBeUndefined();
+
+      expect(selectMock).toHaveBeenCalledTimes(1);
+      expect(fromMock).toHaveBeenCalledWith(ensNodeMetadata);
+    });
+
+    it("returns value when one record exists", async () => {
+      // arrange
+      selectResult.current = [{ value: "0.1.0" }];
+
+      const client = new EnsDbClient(
+        ensDbClientMock.databaseUrl,
+        ensDbClientMock.databaseSchemaName,
+      );
+
+      // act & assert
+      await expect(client.getEnsDbVersion()).resolves.toBe("0.1.0");
+    });
+
+    // This scenario should be impossible due to the primary key constraint on
+    // the 'key' column of 'ensnode_metadata' table.
+    it("throws when multiple records exist", async () => {
+      // arrange
+      selectResult.current = [{ value: "0.1.0" }, { value: "0.1.1" }];
+
+      const client = new EnsDbClient(
+        ensDbClientMock.databaseUrl,
+        ensDbClientMock.databaseSchemaName,
+      );
+
+      // act & assert
+      await expect(client.getEnsDbVersion()).rejects.toThrowError(/ensdb_version/i);
+    });
+  });
+
+  describe("getEnsIndexerPublicConfig", () => {
+    it("returns undefined when no record exists", async () => {
+      // arrange
+      const client = new EnsDbClient(
+        ensDbClientMock.databaseUrl,
+        ensDbClientMock.databaseSchemaName,
+      );
+
+      // act & assert
+      await expect(client.getEnsIndexerPublicConfig()).resolves.toBeUndefined();
+    });
+
+    it("deserializes the stored config", async () => {
+      // arrange
+      const serializedConfig = serializeEnsIndexerPublicConfig(ensDbClientMock.publicConfig);
+      selectResult.current = [{ value: serializedConfig }];
+
+      const client = new EnsDbClient(
+        ensDbClientMock.databaseUrl,
+        ensDbClientMock.databaseSchemaName,
+      );
+
+      // act & assert
+      await expect(client.getEnsIndexerPublicConfig()).resolves.toStrictEqual(
+        ensDbClientMock.publicConfig,
+      );
+    });
+  });
+
+  describe("getIndexingStatusSnapshot", () => {
+    it("deserializes the stored indexing status snapshot", async () => {
+      // arrange
+      selectResult.current = [{ value: ensDbClientMock.serializedSnapshot }];
+
+      const client = new EnsDbClient(
+        ensDbClientMock.databaseUrl,
+        ensDbClientMock.databaseSchemaName,
+      );
+      const expected = deserializeCrossChainIndexingStatusSnapshot(
+        ensDbClientMock.serializedSnapshot,
+      );
+
+      // act & assert
+      await expect(client.getIndexingStatusSnapshot()).resolves.toStrictEqual(expected);
+    });
+  });
+
+  describe("upsertEnsDbVersion", () => {
+    it("writes the database version metadata", async () => {
+      // arrange
+      const client = new EnsDbClient(
+        ensDbClientMock.databaseUrl,
+        ensDbClientMock.databaseSchemaName,
+      );
+
+      // act
+      await client.upsertEnsDbVersion("0.2.0");
+
+      // assert
+      expect(insertMock).toHaveBeenCalledWith(ensNodeMetadata);
+      expect(valuesMock).toHaveBeenCalledWith({
+        key: EnsNodeMetadataKeys.EnsDbVersion,
+        value: "0.2.0",
+      });
+      expect(onConflictDoUpdateMock).toHaveBeenCalledWith({
+        target: ensNodeMetadata.key,
+        set: { value: "0.2.0" },
+      });
+    });
+  });
+
+  describe("upsertEnsIndexerPublicConfig", () => {
+    it("serializes and writes the public config", async () => {
+      // arrange
+      const client = new EnsDbClient(
+        ensDbClientMock.databaseUrl,
+        ensDbClientMock.databaseSchemaName,
+      );
+      const expectedValue = serializeEnsIndexerPublicConfig(ensDbClientMock.publicConfig);
+
+      // act
+      await client.upsertEnsIndexerPublicConfig(ensDbClientMock.publicConfig);
+
+      // assert
+      expect(valuesMock).toHaveBeenCalledWith({
+        key: EnsNodeMetadataKeys.EnsIndexerPublicConfig,
+        value: expectedValue,
+      });
+    });
+  });
+
+  describe("upsertIndexingStatusSnapshot", () => {
+    it("serializes and writes the indexing status snapshot", async () => {
+      // arrange
+      const client = new EnsDbClient(
+        ensDbClientMock.databaseUrl,
+        ensDbClientMock.databaseSchemaName,
+      );
+      const snapshot = deserializeCrossChainIndexingStatusSnapshot(
+        ensDbClientMock.serializedSnapshot,
+      );
+      const expectedValue = serializeCrossChainIndexingStatusSnapshot(snapshot);
+
+      // act
+      await client.upsertIndexingStatusSnapshot(snapshot);
+
+      // assert
+      expect(valuesMock).toHaveBeenCalledWith({
+        key: EnsNodeMetadataKeys.EnsIndexerIndexingStatus,
+        value: expectedValue,
+      });
+    });
+  });
+});