Implement validator data process structure

packages/brain/src/calls/getStakerConfig.ts

@@ -1,8 +1,8 @@
 import { StakerConfig } from "@stakingbrain/common";
 import {
   network,
-  executionClientSelected,
-  consensusClientSelected,
+  executionClient,
+  consensusClient,
   isMevBoostSet,
   executionClientUrl,
   validatorUrl,
@@ -13,8 +13,8 @@ import {
 export async function getStakerConfig(): Promise<StakerConfig> {
   return {
     network,
-    executionClientSelected,
-    consensusClientSelected,
+    executionClient,
+    consensusClient,
     isMevBoostSet,
     executionClientUrl,
     validatorUrl,

packages/brain/src/index.ts

@@ -33,8 +33,8 @@ export const __dirname = process.cwd();
 // Load staker config
 export const {
   network,
-  executionClientSelected,
-  consensusClientSelected,
+  executionClient,
+  consensusClient,
   isMevBoostSet,
   executionClientUrl,
   validatorUrl,
@@ -53,7 +53,7 @@ export const {
   tlsCert
 } = brainConfig();
 logger.debug(
-  `Loaded staker config:\n  - Network: ${network}\n  - Execution client: ${executionClientSelected}\n  - Consensus client: ${consensusClientSelected}\n  - Execution client url: ${executionClientUrl}\n  - Validator url: ${validatorUrl}\n  - Beaconcha url: ${blockExplorerUrl}\n  - Beaconchain url: ${beaconchainUrl}\n  - Signer url: ${signerUrl}\n  - Token: ${token}\n  - Host: ${host}}\n  - Postgres url: ${postgresUrl}\n}`
+  `Loaded staker config:\n  - Network: ${network}\n  - Execution client: ${executionClient}\n  - Consensus client: ${consensusClient}\n  - Execution client url: ${executionClientUrl}\n  - Validator url: ${validatorUrl}\n  - Beaconcha url: ${blockExplorerUrl}\n  - Beaconchain url: ${beaconchainUrl}\n  - Signer url: ${signerUrl}\n  - Token: ${token}\n  - Host: ${host}}\n  - Postgres url: ${postgresUrl}\n}`
 );
 
 // Create API instances. Must preceed db initialization
@@ -104,7 +104,15 @@ const proofOfValidationCron = new CronJob(shareCronInterval, () =>
 proofOfValidationCron.start();
 const trackValidatorsPerformanceCron = new CronJob(slotsPerEpoch * secondsPerSlot * 1000, () =>
   // once every epoch
-  trackValidatorsPerformance({ brainDb, postgresClient, beaconchainApi, minGenesisTime, secondsPerSlot })
+  trackValidatorsPerformance({
+    brainDb,
+    postgresClient,
+    beaconchainApi,
+    minGenesisTime,
+    secondsPerSlot,
+    executionClient,
+    consensusClient
+  })
 );
 const secondsToNextEpoch = getSecondsToNextEpoch({ minGenesisTime, secondsPerSlot });
 // start the cron within the first minute of an epoch

packages/brain/src/modules/apiClients/postgres/index.ts

@@ -2,6 +2,20 @@ import postgres from "postgres";
 import logger from "../../logger/index.js";
 import { BlockProposalStatus, ValidatorPerformance } from "./types.js";
 import { PostgresApiError } from "./error.js";
+import { ConsensusClient, ExecutionClient } from "@stakingbrain/common";
+
+enum Columns {
+  validatorIndex = "validator_index",
+  epoch = "epoch",
+  executionClient = "execution_client",
+  consensusClient = "consensus_client",
+  slot = "slot",
+  liveness = "liveness",
+  blockProposalStatus = "block_proposal_status",
+  syncCommitteeRewards = "sync_comittee_rewards",
+  attestationsRewards = "attestations_rewards",
+  error = "error"
+}
 
 export class PostgresClient {
   private readonly tableName = "validators_performance";
@@ -51,25 +65,40 @@ SELECT pg_total_relation_size('${this.tableName}');
    */
   public async initialize() {
     const query = `
-    DO $$ 
-    BEGIN
-      IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'block_proposal_status') THEN
-        CREATE TYPE block_proposal_status AS ENUM('${BlockProposalStatus.Missed}', '${BlockProposalStatus.Proposed}', '${BlockProposalStatus.Unchosen}');
-      END IF;
-    END $$;
-        
-    CREATE TABLE IF NOT EXISTS ${this.tableName} (
-      validator_index BIGINT NOT NULL,
-      epoch BIGINT NOT NULL,
-      slot BIGINT,
-      liveness BOOLEAN,
-      block_proposal_status block_proposal_status,
-      sync_comittee_rewards BIGINT,
-      attestations_rewards JSONB,
-      error TEXT,
-      PRIMARY KEY (validator_index, epoch)
-    );
-  `;
+DO $$ 
+BEGIN
+  -- Check and create BLOCK_PROPOSAL_STATUS ENUM type if not exists
+  IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'BLOCK_PROPOSAL_STATUS') THEN
+    CREATE TYPE BLOCK_PROPOSAL_STATUS AS ENUM('${BlockProposalStatus.Missed}', '${BlockProposalStatus.Proposed}', '${BlockProposalStatus.Unchosen}');
+  END IF;
+
+  -- Check and create EXECUTION_CLIENT ENUM type if not exists
+  IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'EXECUTION_CLIENT') THEN
+    CREATE TYPE EXECUTION_CLIENT AS ENUM('${ExecutionClient.Besu}', '${ExecutionClient.Nethermind}', '${ExecutionClient.Geth}', '${ExecutionClient.Erigon}', '${ExecutionClient.Unknown}');
+  END IF;
+
+  -- Check and create CONSENSUS_CLIENT ENUM type if not exists
+  IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'CONSENSUS_CLIENT') THEN
+    CREATE TYPE CONSENSUS_CLIENT AS ENUM('${ConsensusClient.Teku}', '${ConsensusClient.Prysm}', '${ConsensusClient.Lighthouse}', '${ConsensusClient.Nimbus}', '${ConsensusClient.Unknown}');
+  END IF;
+END $$;
+
+-- Create the table if not exists
+CREATE TABLE IF NOT EXISTS ${this.tableName} (
+  ${Columns.validatorIndex} BIGINT NOT NULL,
+  ${Columns.epoch} BIGINT NOT NULL,
+  ${Columns.executionClient} EXECUTION_CLIENT NOT NULL,
+  ${Columns.consensusClient} CONSENSUS_CLIENT NOT NULL,
+  ${Columns.slot} BIGINT,
+  ${Columns.liveness} BOOLEAN,
+  ${Columns.blockProposalStatus} BLOCK_PROPOSAL_STATUS,
+  ${Columns.syncCommitteeRewards} BIGINT,
+  ${Columns.attestationsRewards} JSONB,
+  ${Columns.error} TEXT,
+  PRIMARY KEY (${Columns.validatorIndex}, ${Columns.epoch})
+);
+`;
+
     try {
       await this.sql.unsafe(query);
       logger.info("Table created or already exists.");
@@ -94,6 +123,23 @@ SELECT pg_total_relation_size('${this.tableName}');
     }
   }
 
+  /**
+   * Delete enum types.
+   */
+  public async deleteEnumTypes(): Promise<void> {
+    const query = `
+    DROP TYPE IF EXISTS BLOCK_PROPOSAL_STATUS;
+    DROP TYPE IF EXISTS EXECUTION_CLIENT;
+    DROP TYPE IF EXISTS CONSENSUS_CLIENT;
+  `;
+    try {
+      await this.sql.unsafe(query);
+      logger.info("Enum types deleted.");
+    } catch (err) {
+      logger.error("Error deleting enum types:", err);
+    }
+  }
+
   /**
    * Inserts the given performance data into the database.
    *
@@ -102,13 +148,15 @@ SELECT pg_total_relation_size('${this.tableName}');
    */
   public async insertPerformanceData(data: ValidatorPerformance): Promise<void> {
     const query = `
-INSERT INTO ${this.tableName} (validator_index, epoch, slot, liveness, block_proposal_status, sync_comittee_rewards, attestations_rewards, error)
+INSERT INTO ${this.tableName} (${Columns.validatorIndex}, ${Columns.epoch}, ${Columns.slot}, ${Columns.liveness}, ${Columns.blockProposalStatus}, ${Columns.syncCommitteeRewards}, ${Columns.attestationsRewards}, ${Columns.error})
 VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
     `;
     try {
       await this.sql.unsafe(query, [
         data.validatorIndex,
         data.epoch,
+        data.executionClient,
+        data.consensusClient,
         data.slot ?? null,
         data.liveness ?? null,
         data.blockProposalStatus ?? null,
@@ -122,6 +170,39 @@ VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
     }
   }
 
+  /**
+   * Get the validators data for the given validator indexes from all epochs.
+   *
+   * @param validatorIndexes - The indexes of the validators to get the data for.
+   * @returns The performance data for the given validators.
+   */
+  public async getValidatorsDataFromAllEpochs(validatorIndexes: string[]): Promise<ValidatorPerformance[]> {
+    const query = `
+SELECT * FROM ${this.tableName}
+WHERE ${Columns.validatorIndex} = ANY($1)
+    `;
+    try {
+      const result = await this.sql.unsafe(query, [validatorIndexes]);
+      // TODO: add type for result
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      return result.map((row: any) => ({
+        validatorIndex: row.validator_index,
+        epoch: row.epoch,
+        executionClient: row.execution_client,
+        consensusClient: row.consensus_client,
+        slot: row.slot,
+        liveness: row.liveness,
+        blockProposalStatus: row.block_proposal_status,
+        syncCommitteeRewards: row.sync_comittee_rewards,
+        attestationsRewards: row.attestations_rewards,
+        error: row.error
+      }));
+    } catch (err) {
+      logger.error("Error getting data:", err);
+      return [];
+    }
+  }
+
   /**
    * Method to close the database connection.
    */

packages/brain/src/modules/apiClients/postgres/types.ts

@@ -1,3 +1,5 @@
+import { ConsensusClient, ExecutionClient } from "@stakingbrain/common";
+
 export enum BlockProposalStatus {
   Missed = "Missed",
   Proposed = "Proposed",
@@ -8,6 +10,8 @@ export enum BlockProposalStatus {
 export interface ValidatorPerformance {
   validatorIndex: number;
   epoch: number;
+  executionClient: ExecutionClient;
+  consensusClient: ConsensusClient;
   slot?: number;
   liveness?: boolean;
   blockProposalStatus?: BlockProposalStatus;

packages/brain/src/modules/config/index.ts

@@ -10,19 +10,18 @@ import {
 } from "./networks/index.js";
 
 export const brainConfig = (): BrainConfig => {
-  const { network, executionClientSelected, consensusClientSelected, isMevBoostSet, shareDataWithDappnode } =
-    loadEnvs();
+  const { network, executionClient, consensusClient, isMevBoostSet, shareDataWithDappnode } = loadEnvs();
   switch (network) {
     case Network.Holesky:
-      return holeskyBrainConfig(executionClientSelected, consensusClientSelected, isMevBoostSet, shareDataWithDappnode);
+      return holeskyBrainConfig(executionClient, consensusClient, isMevBoostSet, shareDataWithDappnode);
     case Network.Mainnet:
-      return mainnetBrainConfig(executionClientSelected, consensusClientSelected, isMevBoostSet, shareDataWithDappnode);
+      return mainnetBrainConfig(executionClient, consensusClient, isMevBoostSet, shareDataWithDappnode);
     case Network.Gnosis:
-      return gnosisBrainConfig(executionClientSelected, consensusClientSelected, isMevBoostSet, shareDataWithDappnode);
+      return gnosisBrainConfig(executionClient, consensusClient, isMevBoostSet, shareDataWithDappnode);
     case Network.Lukso:
-      return luksoBrainConfig(executionClientSelected, consensusClientSelected, isMevBoostSet, shareDataWithDappnode);
+      return luksoBrainConfig(executionClient, consensusClient, isMevBoostSet, shareDataWithDappnode);
     case Network.Prater:
-      return praterBrainConfig(executionClientSelected, consensusClientSelected, isMevBoostSet, shareDataWithDappnode);
+      return praterBrainConfig(executionClient, consensusClient, isMevBoostSet, shareDataWithDappnode);
     default:
       throw Error(`Network ${network} is not supported`);
   }

packages/brain/src/modules/config/loadEnvs.ts

@@ -1,31 +1,61 @@
-import { Network } from "@stakingbrain/common";
+import { ConsensusClient, ExecutionClient, Network } from "@stakingbrain/common";
 
 export function loadEnvs(): {
   network: Network;
-  executionClientSelected: string;
-  consensusClientSelected: string;
+  executionClient: ExecutionClient;
+  consensusClient: ConsensusClient;
   isMevBoostSet: boolean;
   shareDataWithDappnode: boolean;
 } {
-  const network = process.env.NETWORK;
-  if (!network) throw Error("NETWORK env is required");
-  if (!Object.values(Network).includes(network as Network))
-    throw Error(`NETWORK env must be one of ${Object.values(Network).join(", ")}`);
-
-  const executionClientSelected = process.env[`_DAPPNODE_GLOBAL_EXECUTION_CLIENT_${network.toUpperCase()}`];
-  if (!executionClientSelected)
-    throw Error(`_DAPPNODE_GLOBAL_EXECUTION_CLIENT_${network.toUpperCase()} env is required`);
-  const consensusClientSelected = process.env[`_DAPPNODE_GLOBAL_CONSENSUS_CLIENT_${network.toUpperCase()}`];
-  if (!consensusClientSelected)
-    throw Error(`_DAPPNODE_GLOBAL_CONSENSUS_CLIENT_${network.toUpperCase()} env is required`);
+  const network = getNetwork();
+
+  const executionClient = getExecutionClient(network);
+  const consensusClient = getConsensusClient(network);
+
   const isMevBoostSet = process.env[`_DAPPNODE_GLOBAL_MEVBOOST_${network.toUpperCase()}`] === "true";
   const shareDataWithDappnode = process.env.SHARE_DATA_WITH_DAPPNODE === "true";
 
   return {
     network: network as Network,
-    executionClientSelected,
-    consensusClientSelected,
+    executionClient,
+    consensusClient,
     isMevBoostSet,
     shareDataWithDappnode
   };
 }
+
+function getNetwork(): Network {
+  const network = process.env.NETWORK;
+  if (!network) throw Error("NETWORK env is required");
+
+  if (network === Network.Mainnet) return Network.Mainnet;
+  if (network === Network.Prater) return Network.Prater;
+  if (network === Network.Gnosis) return Network.Gnosis;
+  if (network === Network.Lukso) return Network.Lukso;
+  if (network === Network.Holesky) return Network.Holesky;
+
+  throw Error(`NETWORK env must be one of ${Object.values(Network).join(", ")}`);
+}
+
+function getExecutionClient(network: Network): ExecutionClient {
+  const executionClientStr = process.env[`_DAPPNODE_GLOBAL_EXECUTION_CLIENT_${network.toUpperCase()}`];
+  if (!executionClientStr) throw Error(`_DAPPNODE_GLOBAL_EXECUTION_CLIENT_${network.toUpperCase()} env is required`);
+
+  if (executionClientStr.includes(ExecutionClient.Geth)) return ExecutionClient.Geth;
+  if (executionClientStr.includes(ExecutionClient.Besu)) return ExecutionClient.Besu;
+  if (executionClientStr.includes(ExecutionClient.Nethermind)) return ExecutionClient.Nethermind;
+  if (executionClientStr.includes(ExecutionClient.Erigon)) return ExecutionClient.Erigon;
+  return ExecutionClient.Unknown;
+}
+
+function getConsensusClient(network: Network): ConsensusClient {
+  const consensusClientStr = process.env[`_DAPPNODE_GLOBAL_CONSENSUS_CLIENT_${network.toUpperCase()}`];
+  if (!consensusClientStr) throw Error(`_DAPPNODE_GLOBAL_CONSENSUS_CLIENT_${network.toUpperCase()} env is required`);
+
+  if (consensusClientStr.includes(ConsensusClient.Teku)) return ConsensusClient.Teku;
+  if (consensusClientStr.includes(ConsensusClient.Prysm)) return ConsensusClient.Prysm;
+  if (consensusClientStr.includes(ConsensusClient.Lighthouse)) return ConsensusClient.Lighthouse;
+  if (consensusClientStr.includes(ConsensusClient.Nimbus)) return ConsensusClient.Nimbus;
+  if (consensusClientStr.includes(ConsensusClient.Lodestar)) return ConsensusClient.Lodestar;
+  return ConsensusClient.Unknown;
+}

packages/brain/src/modules/config/networks/gnosis.ts

@@ -1,25 +1,25 @@
-import { Network } from "@stakingbrain/common";
+import { ConsensusClient, ExecutionClient, Network } from "@stakingbrain/common";
 import { BrainConfig } from "../types.js";
 import { tlsCert } from "./tlsCert.js";
 import { validatorToken } from "./validatorToken.js";
 
 export const gnosisBrainConfig = (
-  executionClientSelected: string,
-  consensusClientSelected: string,
+  executionClient: ExecutionClient,
+  consensusClient: ConsensusClient,
   isMevBoostSet: boolean,
   shareDataWithDappnode: boolean
 ): BrainConfig => {
   return {
     network: Network.Gnosis,
-    executionClientSelected,
-    consensusClientSelected,
+    executionClient,
+    consensusClient,
     isMevBoostSet,
     executionClientUrl: "http://execution.gnosis.dncore.dappnode:8545",
     validatorUrl: "http://validator.gnosis.dncore.dappnode:3500",
     beaconchainUrl: "http:/beacon-chain.gnosis.dncore.dappnode:3500",
     blockExplorerUrl: "https://gnosischa.in",
     signerUrl: "http://web3signer.web3signer-gnosis.dappnode:9000",
-    token: validatorToken(consensusClientSelected),
+    token: validatorToken(consensusClient),
     host: "brain.web3signer-gnosis.dappnode",
     shareDataWithDappnode,
     validatorsMonitorUrl: "https://validators-proofs.dappnode.io",
@@ -28,6 +28,6 @@ export const gnosisBrainConfig = (
     postgresUrl: "postgres://postgres:[email protected]:5432/web3signer-gnosis",
     secondsPerSlot: 5,
     slotsPerEpoch: 16,
-    tlsCert: tlsCert(consensusClientSelected)
+    tlsCert: tlsCert(consensusClient)
   };
 };