feat(cross-chain-indexing): enable multiple concurrent plugins

.env.local.example

@@ -34,5 +34,5 @@ DATABASE_URL=postgresql://dbuser:abcd1234@localhost:5432/my_database
 
 # Plugin configuration
 # Identify which indexer plugins to activate (see `src/plugins` for available plugins)
-# This is a comma separated list of one or more available plugin names.
+# This is a comma separated list of one or more available plugin names (case-sensitive).
 ACTIVE_PLUGINS=eth,base.eth,linea.eth

ponder.config.ts

@@ -1,32 +1,32 @@
 import { deepMergeRecursive } from "./src/lib/helpers";
-import { type IntersectionOf, getActivePlugins } from "./src/lib/plugin-helpers";
+import { type MergedTypes, getActivePlugins } from "./src/lib/plugin-helpers";
 import * as baseEthPlugin from "./src/plugins/base.eth/ponder.config";
 import * as ethPlugin from "./src/plugins/eth/ponder.config";
 import * as lineaEthPlugin from "./src/plugins/linea.eth/ponder.config";
 
 // list of all available plugins
-// any of them can be activated in the runtime
-const plugins = [baseEthPlugin, ethPlugin, lineaEthPlugin] as const;
+// any available plugin can be activated at runtime
+const availablePlugins = [baseEthPlugin, ethPlugin, lineaEthPlugin] as const;
 
-// intersection of all available plugin configs to support correct typechecking
+// merge of all available plugin configs to support correct typechecking
 // of the indexing handlers
-type AllPluginConfigs = IntersectionOf<(typeof plugins)[number]["config"]>;
+type AllPluginConfigs = MergedTypes<(typeof availablePlugins)[number]["config"]>;
 
 // Activates the indexing handlers of activated plugins and
 // returns the intersection of their combined config.
-function getActivePluginConfigs(): AllPluginConfigs {
-  const activePlugins = getActivePlugins(plugins);
+function activatePluginsAndGetConfig(): AllPluginConfigs {
+  const activePlugins = getActivePlugins(availablePlugins);
 
   // load indexing handlers from the active plugins into the runtime
   activePlugins.forEach((plugin) => plugin.activate());
 
-  const config = activePlugins
+  const activePluginsConfig = activePlugins
     .map((plugin) => plugin.config)
     .reduce((acc, val) => deepMergeRecursive(acc, val), {} as AllPluginConfigs);
 
-  return config as AllPluginConfigs;
+  return activePluginsConfig as AllPluginConfigs;
 }
 
 // The type of the default export is the intersection of all active plugin configs
 // configs so that each plugin can be correctly typechecked
-export default getActivePluginConfigs();
+export default activatePluginsAndGetConfig();

src/handlers/Registry.ts

@@ -43,7 +43,6 @@ export async function setupRootNode({ context }: { context: Context }) {
       isMigrated: false,
     })
     // only insert the domain entity into the database if it doesn't already exist
-    // only if it doesn't already exist
     .onConflictDoNothing();
 }
 

src/lib/helpers.ts

@@ -59,7 +59,7 @@ const DEFAULT_RPC_RATE_LIMIT = 50;
  * @param chainId the chain ID to get the rate limit for
  * @returns the rate limit in requests per second (rps)
  */
-export const rpcRequestRateLimit = (chainId: number): number => {
+export const rpcMaxRequestsPerSecond = (chainId: number): number => {
   /**
    * Reads the RPC request rate limit for a given chain ID from the environment
    * variable: RPC_REQUEST_RATE_LIMIT_{chainId}.
@@ -79,18 +79,16 @@ export const rpcRequestRateLimit = (chainId: number): number => {
   // otherwise
   try {
     // parse the rate limit value from the environment variable
-    const rpcRequestRateLimit = parseInt(envVarValue, 10);
-    
-    if (Number.isNaN(rpcRequestRateLimit)) {
-      throw new Error(`Could not parse rate limit value '${rpcRequestRateLimit}'`);
+    const parsedEnvVarValue = parseInt(envVarValue, 10);
+
+    if (Number.isNaN(parsedEnvVarValue) || parsedEnvVarValue <= 0) {
+      throw new Error(`Rate limit value must be an integer greater than 0.`);
     }
 
-    return rpcRequestRateLimit;
-  } catch (e) {
-    console.log(e);
-
+    return parsedEnvVarValue;
+  } catch (e: any) {
     throw new Error(
-      `Invalid '${envVarName}' environment variable value: '${envVarValue}'. Please provide a valid RPC RATE LIMIT integer.`,
+      `Invalid '${envVarName}' environment variable value: '${envVarValue}'. ${e.message}`,
     );
   }
 };

src/lib/plugin-helpers.ts

@@ -79,28 +79,29 @@ type PluginNamespacePath<T extends PluginNamespacePath = "/"> =
   | `/${string}`
   | `/${string}${T}`;
 
-/** @var comma separated list of the requested active plugin names (see `src/plugins` for available plugins) */
-export const ACTIVE_PLUGINS = process.env.ACTIVE_PLUGINS;
-
 /**
- * Returns the list of 1 or more active plugins based on the `ACTIVE_PLUGINS` environment variable.
+ * Returns a list of 1 or more distinct active plugins based on the `ACTIVE_PLUGINS` environment variable.
  *
  * The `ACTIVE_PLUGINS` environment variable is a comma-separated list of plugin
  * names. The function returns the plugins that are included in the list.
  *
- * @param plugins is a list of available plugins
+ * @param availablePlugins is a list of available plugins
  * @returns the active plugins
  */
-export function getActivePlugins<T extends { ownedName: string }>(plugins: readonly T[]): T[] {
-  const pluginsToActivateByOwnedName = ACTIVE_PLUGINS ? ACTIVE_PLUGINS.split(",") : [];
+export function getActivePlugins<T extends { ownedName: string }>(
+  availablePlugins: readonly T[],
+): T[] {
+  /** @var comma separated list of the requested plugin names (see `src/plugins` for available plugins) */
+  const requestedPluginsEnvVar = process.env.ACTIVE_PLUGINS;
+  const requestedPlugins = requestedPluginsEnvVar ? requestedPluginsEnvVar.split(",") : [];
 
-  if (!pluginsToActivateByOwnedName.length) {
-    throw new Error("No active plugins found. Please set the ACTIVE_PLUGINS environment variable.");
+  if (!requestedPlugins.length) {
+    throw new Error("Set the ACTIVE_PLUGINS environment variable to activate one or more plugins.");
   }
 
   // Check if the requested plugins are valid and can become active
-  const invalidPlugins = pluginsToActivateByOwnedName.filter(
-    (plugin) => !plugins.some((p) => p.ownedName === plugin),
+  const invalidPlugins = requestedPlugins.filter(
+    (plugin) => !availablePlugins.some((p) => p.ownedName === plugin),
   );
 
   if (invalidPlugins.length) {
@@ -112,12 +113,19 @@ export function getActivePlugins<T extends { ownedName: string }>(plugins: reado
     );
   }
 
-  // Return the active plugins
-  return plugins.filter((plugin) => pluginsToActivateByOwnedName.includes(plugin.ownedName));
+  const uniquePluginsToActivate = availablePlugins.reduce((acc, plugin) => {
+    // Only add the plugin if it's not already in the map
+    if (acc.has(plugin.ownedName) === false) {
+      acc.set(plugin.ownedName, plugin);
+    }
+    return acc;
+  }, new Map<string, T>());
+
+  return Array.from(uniquePluginsToActivate.values());
 }
 
-// Helper type to get the intersection of all config types
-export type IntersectionOf<T> = (T extends any ? (x: T) => void : never) extends (
+// Helper type to merge multiple types into one
+export type MergedTypes<T> = (T extends any ? (x: T) => void : never) extends (
   x: infer R,
 ) => void
   ? R

src/plugins/base.eth/handlers/Registrar.ts

@@ -23,10 +23,10 @@ export default function () {
 
   ponder.on(pluginNamespace("BaseRegistrar:NameRegistered"), async ({ context, event }) => {
     await upsertAccount(context, event.args.owner);
-    // base has 'preminted' names via Registrar#registerOnly, which explicitly does not update Registry.
-    // this breaks a subgraph assumption, as it expects a domain to exist (via Registry:NewOwner) before
-    // any Registrar:NameRegistered events. in the future we will likely happily upsert domains, but
-    // in order to avoid prematurely drifting from subgraph equivalancy, we insert the domain entity here,
+    // Base has 'preminted' names via Registrar#registerOnly, which explicitly
+    // does not update the Registry. This breaks a subgraph assumption, as it
+    // expects a domain to exist (via Registry:NewOwner) before any
+    // Registrar:NameRegistered events. We insert the domain entity here,
     // allowing the base indexer to progress.
     await context.db.insert(schema.domain).values({
       id: makeSubnodeNamehash(ownedSubnameNode, tokenIdToLabel(event.args.id)),
@@ -45,12 +45,12 @@ export default function () {
 
     if (event.args.from === zeroAddress) {
       // The ens-subgraph `handleNameTransferred` handler implementation
-      // assumes the domain record exists. However, when an NFT token is
-      // minted, there's no domain entity in the database yet. The very first
-      // transfer event has to ensure the domain entity for the requested
-      // token ID has been inserted into the database. This is a workaround to
-      // meet expectations of the `handleNameTransferred` subgraph
-      // implementation.
+      // assumes an indexed record for the domain already exists. However,
+      // when an NFT token is minted (transferred from `0x0` address),
+      // there's no domain entity in the database yet. That very first transfer
+      // event has to ensure the domain entity for the requested token ID
+      // has been inserted into the database. This is a workaround to meet
+      // expectations of the `handleNameTransferred` subgraph implementation.
       await context.db.insert(schema.domain).values({
         id: makeSubnodeNamehash(ownedSubnameNode, tokenIdToLabel(tokenId)),
         ownerId: to,

src/plugins/base.eth/ponder.config.ts

@@ -2,7 +2,7 @@ import { type ContractConfig, createConfig, factory } from "ponder";
 import { http, getAbiItem } from "viem";
 import { base } from "viem/chains";
 
-import { blockConfig, rpcEndpointUrl, rpcRequestRateLimit } from "../../lib/helpers";
+import { blockConfig, rpcEndpointUrl, rpcMaxRequestsPerSecond } from "../../lib/helpers";
 import { createPluginNamespace } from "../../lib/plugin-helpers";
 import { BaseRegistrar } from "./abis/BaseRegistrar";
 import { EarlyAccessRegistrarController } from "./abis/EARegistrarController";
@@ -24,7 +24,7 @@ export const config = createConfig({
     base: {
       chainId: base.id,
       transport: http(rpcEndpointUrl(base.id)),
-      maxRequestsPerSecond: rpcRequestRateLimit(base.id),
+      maxRequestsPerSecond: rpcMaxRequestsPerSecond(base.id),
     },
   },
   contracts: {

src/plugins/eth/ponder.config.ts

@@ -2,7 +2,7 @@ import { ContractConfig, createConfig, factory, mergeAbis } from "ponder";
 import { http, getAbiItem } from "viem";
 
 import { mainnet } from "viem/chains";
-import { blockConfig, rpcEndpointUrl, rpcRequestRateLimit } from "../../lib/helpers";
+import { blockConfig, rpcEndpointUrl, rpcMaxRequestsPerSecond } from "../../lib/helpers";
 import { createPluginNamespace } from "../../lib/plugin-helpers";
 import { BaseRegistrar } from "./abis/BaseRegistrar";
 import { EthRegistrarController } from "./abis/EthRegistrarController";
@@ -28,7 +28,7 @@ export const config = createConfig({
     mainnet: {
       chainId: mainnet.id,
       transport: http(rpcEndpointUrl(mainnet.id)),
-      maxRequestsPerSecond: rpcRequestRateLimit(mainnet.id),
+      maxRequestsPerSecond: rpcMaxRequestsPerSecond(mainnet.id),
     },
   },
   contracts: {

src/plugins/linea.eth/handlers/EthRegistrar.ts

@@ -23,12 +23,12 @@ export default function () {
 
     if (event.args.from === zeroAddress) {
       // The ens-subgraph `handleNameTransferred` handler implementation
-      // assumes the domain record exists. However, when an NFT token is
-      // minted, there's no domain entity in the database yet. The very first
-      // transfer event has to ensure the domain entity for the requested
-      // token ID has been inserted into the database. This is a workaround to
-      // meet expectations of the `handleNameTransferred` subgraph
-      // implementation.
+      // assumes an indexed record for the domain already exists. However,
+      // when an NFT token is minted (transferred from `0x0` address),
+      // there's no domain entity in the database yet. That very first transfer
+      // event has to ensure the domain entity for the requested token ID
+      // has been inserted into the database. This is a workaround to meet
+      // expectations of the `handleNameTransferred` subgraph implementation.
       await context.db.insert(schema.domain).values({
         id: makeSubnodeNamehash(ownedSubnameNode, tokenIdToLabel(tokenId)),
         ownerId: to,

src/plugins/linea.eth/ponder.config.ts

@@ -2,7 +2,7 @@ import { ContractConfig, createConfig, factory, mergeAbis } from "ponder";
 import { http, getAbiItem } from "viem";
 
 import { linea } from "viem/chains";
-import { blockConfig, rpcEndpointUrl, rpcRequestRateLimit } from "../../lib/helpers";
+import { blockConfig, rpcEndpointUrl, rpcMaxRequestsPerSecond } from "../../lib/helpers";
 import { createPluginNamespace } from "../../lib/plugin-helpers";
 import { BaseRegistrar } from "./abis/BaseRegistrar";
 import { EthRegistrarController } from "./abis/EthRegistrarController";
@@ -24,7 +24,7 @@ export const config = createConfig({
     linea: {
       chainId: linea.id,
       transport: http(rpcEndpointUrl(linea.id)),
-      maxRequestsPerSecond: rpcRequestRateLimit(linea.id),
+      maxRequestsPerSecond: rpcMaxRequestsPerSecond(linea.id),
     },
   },
   contracts: {