feat(ensindexer): ENSDb Writer Worker

.changeset/moody-tips-run.md

@@ -0,0 +1,5 @@
+---
+"ensindexer": minor
+---
+
+Introduces a worker that writes serialized representations of ENSIndexer Public Config and Indexing Status to ENSDb.

.changeset/rich-buttons-shine.md

@@ -0,0 +1,5 @@
+---
+"@ensnode/ensnode-schema": minor
+---
+
+Includes schema for `ENSNodeMetadata`.

apps/ensapi/package.json

@@ -43,8 +43,8 @@
     "date-fns": "catalog:",
     "drizzle-orm": "catalog:",
     "hono": "catalog:",
-    "p-memoize": "^8.0.0",
-    "p-retry": "^7.1.0",
+    "p-memoize": "catalog:",
+    "p-retry": "catalog:",
     "pg-connection-string": "catalog:",
     "pino": "catalog:",
     "ponder-enrich-gql-docs-middleware": "^0.1.3",

apps/ensindexer/package.json

@@ -29,10 +29,13 @@
     "@ensnode/ensnode-sdk": "workspace:*",
     "@ensnode/ensrainbow-sdk": "workspace:*",
     "@ensnode/ponder-metadata": "workspace:*",
+    "@ponder/client": "catalog:",
     "caip": "catalog:",
     "date-fns": "catalog:",
     "deepmerge-ts": "^7.1.5",
     "dns-packet": "^5.6.1",
+    "drizzle-orm": "catalog:",
+    "p-retry": "catalog:",
     "pg-connection-string": "catalog:",
     "hono": "catalog:",
     "ponder": "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/ensdb-writer-worker.ts

@@ -0,0 +1,134 @@
+/**
+ * This file manages syncing ENSNode metadata:
+ * - ENSIndexer Public Config
+ * - Indexing Status
+ * into the ENSDb.
+ */
+import { secondsToMilliseconds } from "date-fns";
+import pRetry from "p-retry";
+
+import {
+  CrossChainIndexingStatusSnapshot,
+  type Duration,
+  ENSIndexerPublicConfig,
+  IndexingStatusResponseCodes,
+  OmnichainIndexingStatusIds,
+  validateENSIndexerPublicConfigCompatibility,
+} from "@ensnode/ensnode-sdk";
+
+import { EnsDbClient } from "@/lib/ensdb";
+import { ensIndexerClient, waitForEnsIndexerToBecomeHealthy } from "@/lib/ensindexer";
+
+const INDEXING_STATUS_RECORD_UPDATE_INTERVAL: Duration = 1;
+
+/**
+ * ENSDb Writer Worker
+ *
+ * Runs the following tasks:
+ * 1) On application startup, attempt to upsert serialized representation of
+ *    {@link ENSIndexerPublicConfig} into ENSDb.
+ * 2) On application startup, and then on recurring basis,
+ *    following the {@link INDEXING_STATUS_RECORD_UPDATE_INTERVAL}, attempt to
+ *    upsert serialized representation of {@link CrossChainIndexingStatusSnapshot}
+ *    into ENSDb.
+ */
+async function ensDbWriterWorker() {
+  console.log("ENSDb Writer Worker: waiting for ENSIndexer to become healthy.");
+
+  // 0. Wait for ENSIndexer to become healthy before running the worker's logic
+  await waitForEnsIndexerToBecomeHealthy;
+
+  console.log("ENSDb Writer Worker: ENSIndexer is healthy, starting tasks.");
+
+  // 1. Create ENSDb Client
+  const ensDbClient = new EnsDbClient();
+
+  /**
+   * Handle ENSIndexerPublicConfig and ENSDb Version Records
+   */
+  const handleEnsIndexerPublicConfigAndEnsDbVersionRecords = async () => {
+    // Read stored config and in-memory config.
+    // Note: we wrap each operation in pRetry to ensure all of them can be
+    // completed successfully.
+    const [storedConfig, inMemoryConfig] = await Promise.all([
+      pRetry(() => ensDbClient.getEnsIndexerPublicConfig()),
+      pRetry(() => ensIndexerClient.config()),
+    ]);
+
+    // Validate in-memory config object compatibility with the stored one,
+    // if the stored one is available
+    if (storedConfig) {
+      try {
+        validateENSIndexerPublicConfigCompatibility(storedConfig, inMemoryConfig);
+      } catch (error) {
+        const errorMessage = `In-memory ENSIndexerPublicConfig object is not compatible with its counterpart stored in ENSDb.`;
+
+        // Throw the error to terminate the ENSIndexer process due to
+        // found config incompatibility
+        throw new Error(errorMessage, {
+          cause: error,
+        });
+      }
+    } else {
+      // Upsert ENSDb Version into ENSDb.
+      await ensDbClient.upsertEnsDbVersion(inMemoryConfig.versionInfo.ensDb);
+      // Upsert ENSIndexerPublicConfig into ENSDb.
+      await ensDbClient.upsertEnsIndexerPublicConfig(inMemoryConfig);
+    }
+  };
+
+  /**
+   * Handle Indexing Status Record Recursively
+   */
+  const handleIndexingStatusRecordRecursively = async () => {
+    try {
+      // Read in-memory Indexing Status.
+      const inMemoryIndexingStatus = await ensIndexerClient.indexingStatus();
+
+      // Check if Indexing Status is available.
+      if (inMemoryIndexingStatus.responseCode !== IndexingStatusResponseCodes.Ok) {
+        throw new Error("Indexing Status must be available.");
+      }
+
+      const { snapshot } = inMemoryIndexingStatus.realtimeProjection;
+      const { omnichainSnapshot } = snapshot;
+
+      // Check if Indexing Status is in expected status.
+      // The Omnichain Status must indicate that indexing has started already.
+      // Throw an error if Omnichain Status is "Unstarted".
+      if (omnichainSnapshot.omnichainStatus === OmnichainIndexingStatusIds.Unstarted) {
+        throw new Error("Omnichain Status must be different than 'Unstarted'.");
+      }
+
+      // Upsert ENSIndexerPublicConfig into ENSDb.
+      await ensDbClient.upsertIndexingStatus(snapshot);
+    } catch (error) {
+      // Do nothing about this error, but having it logged.
+      const errorMessage = error instanceof Error ? error.message : "Unknown error";
+      console.error(`Could not upsert Indexing Status record due to: ${errorMessage}`);
+    } finally {
+      // Regardless of current iteration result,
+      // schedule the next callback to handle Indexing Status Record.
+      setTimeout(
+        handleIndexingStatusRecordRecursively,
+        secondsToMilliseconds(INDEXING_STATUS_RECORD_UPDATE_INTERVAL),
+      );
+    }
+  };
+
+  // 4. Handle ENSIndexer Public Config and ENSDb Version just once.
+  console.log("Task: store ENSIndexer Public Config and ENSDb Version in ENSDb.");
+  await handleEnsIndexerPublicConfigAndEnsDbVersionRecords();
+  console.log("ENSIndexer Public Config and ENSDb Version successfully stored in ENSDb.");
+
+  // 5. Handle Indexing Status on recurring basis.
+  console.log("Task: store Indexing Status in ENSDb.");
+  await handleIndexingStatusRecordRecursively();
+  console.log("Indexing Status successfully stored in ENSDb.");
+}
+
+// Run ENSDb Writer Worker in the background.
+ensDbWriterWorker().catch((error) => {
+  console.error("ENSDb Writer Worker failed to perform its tasks", error);
+  process.exit(1);
+});

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

@@ -0,0 +1,25 @@
+// This file was copied 1-to-1 from ENSApi.
+// TODO: deduplicate with apps/ensapi/src/lib/handlers/drizzle.ts when ensnode nodejs internal package is created
+
+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/ensdb-client.ts

@@ -0,0 +1,192 @@
+import config from "@/config";
+
+import { eq } from "drizzle-orm/sql";
+
+import * as schema from "@ensnode/ensnode-schema";
+import {
+  type CrossChainIndexingStatusSnapshot,
+  deserializeCrossChainIndexingStatusSnapshot,
+  deserializeENSIndexerPublicConfig,
+  type ENSIndexerPublicConfig,
+  serializeCrossChainIndexingStatusSnapshotOmnichain,
+  serializeENSIndexerPublicConfig,
+} from "@ensnode/ensnode-sdk";
+
+import { makeDrizzle } from "./drizzle";
+import {
+  EnsNodeMetadataKeys,
+  type SerializedEnsNodeMetadata,
+  type SerializedEnsNodeMetadataEnsDbVersion,
+  type SerializedEnsNodeMetadataEnsIndexerPublicConfig,
+  type SerializedEnsNodeMetadataIndexingStatus,
+} from "./ensnode-metadata";
+
+/**
+ * ENSDb Client Query
+ * 
+  Includes methods for reading from ENSDb.
+ */
+export interface EnsDbClientQuery {
+  /**
+   * Get ENSDb Version
+   *
+   * @returns the existing record, or `undefined`.
+   * @throws if not exactly one record was found.
+   */
+  getEnsDbVersion(): Promise<string | undefined>;
+
+  /**
+   * Get ENSIndexer Public Config
+   *
+   * @returns the existing record, or `undefined`.
+   * @throws if not exactly one record was found.
+   */
+  getEnsIndexerPublicConfig(): Promise<ENSIndexerPublicConfig | undefined>;
+
+  /**
+   * Get Indexing Status
+   *
+   * @returns the existing record, or `undefined`.
+   * @throws if not exactly one record was found.
+   */
+  getIndexingStatus(): Promise<CrossChainIndexingStatusSnapshot | undefined>;
+}
+
+/**
+ * ENSDb Client Mutation
+ *
+ * Includes methods for writing into ENSDb.
+ */
+export interface EnsDbClientMutation {
+  /**
+   * Upsert ENSDb Version
+   *
+   * @throws when upsert operation failed.
+   */
+  upsertEnsDbVersion(ensDbVersion: string): Promise<void>;
+
+  /**
+   * Upsert ENSIndexer Public Config
+   *
+   * @throws when upsert operation failed.
+   */
+  upsertEnsIndexerPublicConfig(ensIndexerPublicConfig: ENSIndexerPublicConfig): Promise<void>;
+
+  /**
+   * Upsert Indexing Status
+   *
+   * @throws when upsert operation failed.
+   */
+  upsertIndexingStatus(indexingStatus: CrossChainIndexingStatusSnapshot): Promise<void>;
+}
+
+/**
+ * ENSDb Client
+ */
+export class EnsDbClient implements EnsDbClientQuery, EnsDbClientMutation {
+  #db = makeDrizzle({
+    databaseSchema: config.databaseSchemaName,
+    databaseUrl: config.databaseUrl,
+    schema,
+  });
+
+  async getEnsDbVersion(): Promise<string | undefined> {
+    const record = await this.getEnsNodeMetadata<SerializedEnsNodeMetadataEnsDbVersion>({
+      key: EnsNodeMetadataKeys.EnsDbVersion,
+    });
+
+    return record;
+  }
+
+  async getEnsIndexerPublicConfig(): Promise<ENSIndexerPublicConfig | undefined> {
+    const record = await this.getEnsNodeMetadata<SerializedEnsNodeMetadataEnsIndexerPublicConfig>({
+      key: EnsNodeMetadataKeys.EnsIndexerPublicConfig,
+    });
+
+    if (!record) {
+      return undefined;
+    }
+
+    return deserializeENSIndexerPublicConfig(record);
+  }
+
+  async getIndexingStatus(): Promise<CrossChainIndexingStatusSnapshot | undefined> {
+    const record = await this.getEnsNodeMetadata<SerializedEnsNodeMetadataIndexingStatus>({
+      key: EnsNodeMetadataKeys.IndexingStatus,
+    });
+
+    if (!record) {
+      return undefined;
+    }
+
+    return deserializeCrossChainIndexingStatusSnapshot(record);
+  }
+
+  async upsertEnsDbVersion(ensDbVersion: string): Promise<void> {
+    await this.upsertEnsNodeMetadata({
+      key: EnsNodeMetadataKeys.EnsDbVersion,
+      value: ensDbVersion,
+    });
+  }
+
+  async upsertEnsIndexerPublicConfig(
+    ensIndexerPublicConfig: ENSIndexerPublicConfig,
+  ): Promise<void> {
+    await this.upsertEnsNodeMetadata({
+      key: EnsNodeMetadataKeys.EnsIndexerPublicConfig,
+      value: serializeENSIndexerPublicConfig(ensIndexerPublicConfig),
+    });
+  }
+
+  async upsertIndexingStatus(indexingStatus: CrossChainIndexingStatusSnapshot): Promise<void> {
+    await this.upsertEnsNodeMetadata({
+      key: EnsNodeMetadataKeys.IndexingStatus,
+      value: serializeCrossChainIndexingStatusSnapshotOmnichain(indexingStatus),
+    });
+  }
+
+  /**
+   * Get ENSNode metadata record
+   *
+   * @returns selected record in ENSDb.
+   * @throws when exactly one matching metadata record was not found
+   */
+  private async getEnsNodeMetadata<
+    EnsNodeMetadataType extends SerializedEnsNodeMetadata = SerializedEnsNodeMetadata,
+  >(metadata: Pick<EnsNodeMetadataType, "key">): Promise<EnsNodeMetadataType["value"] | undefined> {
+    const result = await this.#db
+      .select()
+      .from(schema.ensNodeMetadata)
+      .where(eq(schema.ensNodeMetadata.key, metadata.key));
+
+    if (result.length === 0) {
+      return undefined;
+    }
+
+    if (result.length === 1 && result[0]) {
+      return result[0].value as EnsNodeMetadataType["value"];
+    }
+
+    throw new Error(`There must be exactly one ENSNodeMetadata record for '${metadata.key}' key`);
+  }
+
+  /**
+   * Upsert ENSNode metadata
+   *
+   * @throws when upsert operation failed.
+   */
+  private async upsertEnsNodeMetadata<
+    EnsNodeMetadataType extends SerializedEnsNodeMetadata = SerializedEnsNodeMetadata,
+  >(metadata: EnsNodeMetadataType): Promise<void> {
+    await this.#db
+      .insert(schema.ensNodeMetadata)
+      .values({
+        key: metadata.key,
+        value: metadata.value,
+      })
+      .onConflictDoUpdate({
+        target: schema.ensNodeMetadata.key,
+        set: { value: metadata.value },
+      });
+  }
+}