apply pr feedback

consolidate ensdb modules; replace pino logger with native console logger; simplify ensdb worker logic

Author: tk-o

apps/ensapi/package.json

@@ -43,7 +43,7 @@
     "date-fns": "catalog:",
     "drizzle-orm": "catalog:",
     "hono": "catalog:",
-    "p-memoize": "^8.0.0",
+    "p-memoize": "catalog:",
     "p-retry": "catalog:",
     "pg-connection-string": "catalog:",
     "pino": "catalog:",

apps/ensindexer/package.json

@@ -29,15 +29,15 @@
     "@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-memoize": "catalog:",
     "p-retry": "catalog:",
-    "pino": "catalog:",
     "pg-connection-string": "catalog:",
-    "pg": "8.16.3",
     "hono": "catalog:",
     "ponder": "catalog:",
     "viem": "catalog:",

apps/ensindexer/ponder/src/ensdb-writer-worker.ts

@@ -4,8 +4,6 @@
  * - Indexing Status
  * into the ENSDb.
  */
-import config from "@/config";
-
 import { secondsToMilliseconds } from "date-fns";
 import pRetry from "p-retry";
 
@@ -18,14 +16,11 @@ import {
 } from "@ensnode/ensnode-sdk";
 
 import { validateENSIndexerPublicConfigCompatibility } from "@/config/compatibility";
-import { EnsDbConnection, EnsDbMutation, EnsDbQuery } from "@/lib/ensdb";
+import { EnsDbClient } from "@/lib/ensdb";
 import { ensIndexerClient, waitForEnsIndexerToBecomeHealthy } from "@/lib/ensindexer";
-import { makeLogger } from "@/lib/logger";
 
 const INDEXING_STATUS_RECORD_UPDATE_INTERVAL: Duration = 1;
 
-const logger = makeLogger("ensdb-writer-worker");
-
 /**
  * ENSDb Writer Worker
  *
@@ -38,57 +33,45 @@ const logger = makeLogger("ensdb-writer-worker");
  *    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;
 
-  // 1. Create ENSDb Client
-  const ensDbConnection = new EnsDbConnection();
-  const ensDbClient = ensDbConnection.connect({
-    schemaName: config.databaseSchemaName,
-    poolConfig: {
-      connectionString: config.databaseUrl,
-    },
-  });
-
-  logger.info("ENSDb Client connected");
+  console.log("ENSDb Writer Worker: ENSIndexer is healthy, starting tasks.");
 
-  // 2. Create ENSDb Query object for read operations
-  const ensDbQuery = new EnsDbQuery(ensDbClient);
-  // 3. Create ENSDb Mutation object for write operations
-  const ensDbMutation = new EnsDbMutation(ensDbClient);
+  // 1. Create ENSDb Client
+  const ensDbClient = new EnsDbClient();
 
   /**
    * Handle ENSIndexerPublicConfig Record
    */
   const handleEnsIndexerPublicConfigRecord = async () => {
     // Read stored config and in-memory config.
-    // Note: we wrap read operations in pRetry to ensure all of them are
+    // Note: we wrap each operation in pRetry to ensure all of them can be
     // completed successfully.
-    const [storedConfig, inMemoryConfig] = await pRetry(() =>
-      Promise.all([ensDbQuery.getEnsIndexerPublicConfig(), ensIndexerClient.config()]),
-    );
+    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.";
-
-        logger.error(error, errorMessage);
+        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);
+        throw new Error(errorMessage, {
+          cause: error,
+        });
       }
     } else {
       // Upsert ENSIndexerPublicConfig into ENSDb.
-      // Note: we wrap write operation in pRetry to ensure it can complete
-      // successfully, as there will be no other attempt.
-      await pRetry(() => ensDbMutation.upsertEnsIndexerPublicConfig(inMemoryConfig));
-      logger.info("ENSIndexer Public Config successfully stored in ENSDb.");
+      await ensDbClient.upsertEnsIndexerPublicConfig(inMemoryConfig);
     }
   };
 
@@ -110,16 +93,14 @@ async function ensDbWriterWorker() {
 
       // Check if Indexing Status is in expected status.
       if (omnichainSnapshot.omnichainStatus === OmnichainIndexingStatusIds.Unstarted) {
-        throw new Error("Omnichain Status  must be different that 'Unstarted'.");
+        throw new Error("Omnichain Status must be different than 'Unstarted'.");
       }
 
       // Upsert ENSIndexerPublicConfig into ENSDb.
-      await ensDbMutation.upsertIndexingStatus(snapshot);
-
-      logger.info("Indexing Status successfully stored in ENSDb.");
+      await ensDbClient.upsertIndexingStatus(snapshot);
     } catch (error) {
       // Do nothing about this error, but having it logged.
-      logger.error(error, "Could not upsert Indexing Status record");
+      console.error(error, "Could not upsert Indexing Status record");
     } finally {
       // Regardless of current iteration result,
       // schedule the next callback to handle Indexing Status Record.
@@ -131,12 +112,20 @@ async function ensDbWriterWorker() {
   };
 
   // 4. Handle ENSIndexer Public Config just once.
-  await handleEnsIndexerPublicConfigRecord();
+  console.log("Task: store ENSIndexer Public Config in ENSDb.");
+  await handleEnsIndexerPublicConfigRecord().then(() =>
+    console.log("ENSIndexer Public Config successfully stored in ENSDb."),
+  );
 
   // 5. Handle Indexing Status on recurring basis.
-  await handleIndexingStatusRecordRecursively();
+  console.log("Task: store Indexing Status in ENSDb.");
+  await handleIndexingStatusRecordRecursively().then(() =>
+    console.log("Indexing Status successfully stored in ENSDb."),
+  );
 }
 
 // Run ENSDb Writer Worker in a non-blocking way to
 // allow database migrations to proceed in the background.
-setTimeout(ensDbWriterWorker, 0);
+ensDbWriterWorker().catch((error) =>
+  console.error("ENSDb Writer Worker failed to perform its tasks", error),
+);

apps/ensindexer/src/config/compatibility.ts

@@ -9,10 +9,7 @@ export type ENSIndexerPublicConfigCompatibilityCheck = Pick<
  * Validate if `configB` is compatible with `configA`, such that `configA` is
  * a subset of `configB`.
  *
- * @throws error if 'indexedChainIds' were incompatible.
- * @throws error if 'isSubgraphCompatible' flag was incompatible.
- * @throws error if 'namespace' was incompatible.
- * @throws error if 'plugins' were incompatible.
+ * @throws error if configs are incompatible.
  */
 export function validateENSIndexerPublicConfigCompatibility(
   configA: ENSIndexerPublicConfigCompatibilityCheck,

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

@@ -1,41 +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 { isTable, Table } from "drizzle-orm";
-import { drizzle, type NodePgDatabase } from "drizzle-orm/node-postgres";
-import { isPgEnum } from "drizzle-orm/pg-core";
-import type { Pool } from "pg";
+import { setDatabaseSchema } from "@ponder/client";
+import { drizzle } from "drizzle-orm/node-postgres";
 
 type Schema = { [name: string]: unknown };
 
-// https://github.com/ponder-sh/ponder/blob/f7f6444ab8d1a870fe6492023941091df7b7cddf/packages/client/src/index.ts#L226C1-L239C3
-const setDatabaseSchema = <T extends Schema>(schema: T, schemaName: string) => {
-  for (const table of Object.values(schema)) {
-    if (isTable(table)) {
-      // @ts-expect-error
-      table[Table.Symbol.Schema] = schemaName;
-    } else if (isPgEnum(table)) {
-      // @ts-expect-error
-      table.schema = schemaName;
-    }
-  }
-};
-
 /**
  * Makes a Drizzle DB object.
  */
 export const makeDrizzle = <SCHEMA extends Schema>({
   schema,
-  connectionPool,
+  databaseUrl,
   databaseSchema,
 }: {
   schema: SCHEMA;
-  connectionPool: Pool;
+  databaseUrl: string;
   databaseSchema: string;
-}): NodePgDatabase<SCHEMA> & {
-  $client: Pool;
-} => {
+}) => {
   // monkeypatch schema onto tables
   setDatabaseSchema(schema, databaseSchema);
 
-  return drizzle(connectionPool, { schema, casing: "snake_case" });
+  return drizzle(databaseUrl, { schema, casing: "snake_case" });
 };

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

@@ -0,0 +1,151 @@
+import config from "@/config";
+
+import { eq } from "drizzle-orm/sql";
+
+import * as schema from "@ensnode/ensnode-schema";
+import type {
+  SerializedCrossChainIndexingStatusSnapshot,
+  SerializedENSIndexerPublicConfig,
+} from "@ensnode/ensnode-sdk";
+
+import { makeDrizzle } from "./drizzle";
+import {
+  type EnsNodeMetadata,
+  type EnsNodeMetadataEnsIndexerPublicConfig,
+  type EnsNodeMetadataIndexingStatus,
+  EnsNodeMetadataKeys,
+} from "./ensnode-metadata";
+
+export interface EnsDbClientQuery {
+  getEnsIndexerPublicConfig(): Promise<SerializedENSIndexerPublicConfig | undefined>;
+
+  getIndexingStatus(): Promise<SerializedCrossChainIndexingStatusSnapshot | undefined>;
+}
+
+export interface EnsDbClientMutation {
+  upsertEnsIndexerPublicConfig(
+    ensIndexerPublicConfig: SerializedENSIndexerPublicConfig,
+  ): Promise<SerializedENSIndexerPublicConfig>;
+
+  upsertIndexingStatus(
+    indexingStatus: SerializedCrossChainIndexingStatusSnapshot,
+  ): Promise<SerializedCrossChainIndexingStatusSnapshot>;
+}
+
+/**
+ * ENSDb Client
+ */
+export class EnsDbClient implements EnsDbClientQuery, EnsDbClientMutation {
+  #db = makeDrizzle({
+    databaseSchema: config.databaseSchemaName,
+    databaseUrl: config.databaseUrl,
+    schema,
+  });
+
+  /**
+   * Upsert ENSIndexer Public Config
+   *
+   * @returns updated record in ENSDb.
+   * @throws when upsert operation failed.
+   */
+  async getEnsIndexerPublicConfig(): Promise<SerializedENSIndexerPublicConfig | undefined> {
+    return this.getEnsNodeMetadata<EnsNodeMetadataEnsIndexerPublicConfig>({
+      key: EnsNodeMetadataKeys.EnsIndexerPublicConfig,
+    });
+  }
+
+  /**
+   * Upsert Indexing Status
+   *
+   * @returns updated record in ENSDb.
+   * @throws when upsert operation failed.
+   */
+  async getIndexingStatus(): Promise<SerializedCrossChainIndexingStatusSnapshot | undefined> {
+    return this.getEnsNodeMetadata<EnsNodeMetadataIndexingStatus>({
+      key: EnsNodeMetadataKeys.IndexingStatus,
+    });
+  }
+
+  /**
+   * Upsert ENSIndexer Public Config
+   *
+   * @returns updated record in ENSDb.
+   * @throws when upsert operation failed.
+   */
+  async upsertEnsIndexerPublicConfig(
+    ensIndexerPublicConfig: SerializedENSIndexerPublicConfig,
+  ): Promise<SerializedENSIndexerPublicConfig> {
+    return this.upsertEnsNodeMetadata({
+      key: EnsNodeMetadataKeys.EnsIndexerPublicConfig,
+      value: ensIndexerPublicConfig,
+    });
+  }
+
+  /**
+   * Upsert Indexing Status
+   *
+   * @returns updated record in ENSDb.
+   * @throws when upsert operation failed.
+   */
+  async upsertIndexingStatus(
+    indexingStatus: SerializedCrossChainIndexingStatusSnapshot,
+  ): Promise<SerializedCrossChainIndexingStatusSnapshot> {
+    return this.upsertEnsNodeMetadata({
+      key: EnsNodeMetadataKeys.IndexingStatus,
+      value: 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 EnsNodeMetadata = EnsNodeMetadata>(
+    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
+   *
+   * @returns updated record in ENSDb.
+   * @throws when upsert operation failed.
+   */
+  private async upsertEnsNodeMetadata<
+    EnsNodeMetadataType extends EnsNodeMetadata = EnsNodeMetadata,
+  >(metadata: EnsNodeMetadataType): Promise<EnsNodeMetadataType["value"]> {
+    const [result] = await this.#db
+      .insert(schema.ensNodeMetadata)
+      .values({
+        key: metadata.key,
+        value: metadata.value,
+      })
+      .onConflictDoUpdate({
+        target: schema.ensNodeMetadata.key,
+        set: { value: metadata.value },
+      })
+      .returning({ value: schema.ensNodeMetadata.value });
+
+    if (!result) {
+      throw new Error(`Failed to upsert metadata for key: ${metadata.key}`);
+    }
+
+    return result.value as EnsNodeMetadataType["value"];
+  }
+}