Merge pull request #151 from migalabs/feature/add_eth2_deposits_table

Feature/add eth2 deposits table

docs/tables.md

@@ -287,3 +287,16 @@ Table that stores the BLS to execution changes that happened in the network.
 | f_validator_index      | uint64       | validator index that had the change           |
 | f_from_bls_pubkey      | string       | BLS public key corresponding to the validator |
 | f_to_execution_address | string       | execution address after the change            |
+
+# ETH2 Deposits (`t_deposits`)
+
+Table that stores the data of the deposits on the beaconchain.
+
+| Column Name              | Type of Data | Description                                       |     |     |
+| ------------------------ | ------------ | ------------------------------------------------- | --- | --- |
+| f_slot                   | uint64       | slot at which the deposit was included            |
+| f_public_key             | string       | public key of the validator deposited             |
+| f_withdrawal_credentials | string       | withdrawal credentials of the validator deposited |
+| f_amount                 | uint64       | amount of ETH deposited (Gwei)                    |
+| f_signature              | string       | signature of the deposit data                     |
+| f_index                  | uint64       | index of the deposit in the slot                  |

pkg/analyzer/process_block.go

@@ -31,9 +31,33 @@ func (s *ChainAnalyzer) ProcessBlock(slot phase0.Slot) {
 		s.processBlobSidecars(block, block.ExecutionPayload.AgnosticTransactions)
 	}
 	s.processBLSToExecutionChanges(block)
+	s.processDeposits(block)
 	s.processerBook.FreePage(routineKey)
 }
 
+func (s *ChainAnalyzer) processDeposits(block *spec.AgnosticBlock) {
+	if len(block.Deposits) == 0 {
+		return
+	}
+	var deposits []spec.Deposit
+	for i, item := range block.Deposits {
+		deposits = append(deposits, spec.Deposit{
+			Slot:                  block.Slot,
+			PublicKey:             item.Data.PublicKey,
+			WithdrawalCredentials: item.Data.WithdrawalCredentials,
+			Amount:                item.Data.Amount,
+			Signature:             item.Data.Signature,
+			Index:                 uint8(i),
+		})
+	}
+
+	err := s.dbClient.PersistDeposits(deposits)
+	if err != nil {
+		log.Errorf("error persisting deposits: %s", err.Error())
+	}
+
+}
+
 func (s *ChainAnalyzer) processBLSToExecutionChanges(block *spec.AgnosticBlock) {
 	if len(block.BLSToExecutionChanges) == 0 {
 		return

pkg/db/bls_to_execution_change.go → pkg/db/bls_to_execution_changes.go

@@ -32,13 +32,13 @@ func blsToExecutionChangeInput(blsToExecutionChanges []spec.BLSToExecutionChange
 		f_to_execution_address proto.ColStr
 	)
 
-	for _, withdrawal := range blsToExecutionChanges {
+	for _, blsToExecutionChange := range blsToExecutionChanges {
 
-		f_slot.Append(uint64(withdrawal.Slot))
-		f_epoch.Append(uint64(withdrawal.Epoch))
-		f_validator_index.Append(uint64(withdrawal.ValidatorIndex))
-		f_from_bls_pubkey.Append(withdrawal.FromBLSPublicKey.String())
-		f_to_execution_address.Append(withdrawal.ToExecutionAddress.String())
+		f_slot.Append(uint64(blsToExecutionChange.Slot))
+		f_epoch.Append(uint64(blsToExecutionChange.Epoch))
+		f_validator_index.Append(uint64(blsToExecutionChange.ValidatorIndex))
+		f_from_bls_pubkey.Append(blsToExecutionChange.FromBLSPublicKey.String())
+		f_to_execution_address.Append(blsToExecutionChange.ToExecutionAddress.String())
 	}
 
 	return proto.Input{

pkg/db/deposits.go

@@ -0,0 +1,75 @@
+package db
+
+import (
+	"fmt"
+
+	"github.com/ClickHouse/ch-go/proto"
+	"github.com/migalabs/goteth/pkg/spec"
+)
+
+var (
+	depositsTable      = "t_deposits"
+	insertDepositQuery = `
+	INSERT INTO %s (
+		f_slot,
+		f_public_key,
+		f_withdrawal_credentials,
+		f_amount,
+		f_signature,
+		f_index
+		)
+		VALUES`
+
+	deleteDepositsQuery = `
+		DELETE FROM %s
+		WHERE f_slot = $1;`
+)
+
+func depositsInput(depositss []spec.Deposit) proto.Input {
+	// one object per column
+	var (
+		f_slot                   proto.ColUInt64
+		f_public_key             proto.ColStr
+		f_withdrawal_credentials proto.ColStr
+		f_amount                 proto.ColUInt64
+		f_signature              proto.ColStr
+		f_index                  proto.ColUInt8
+	)
+
+	for _, deposit := range depositss {
+
+		f_slot.Append(uint64(deposit.Slot))
+		f_public_key.Append(deposit.PublicKey.String())
+		f_withdrawal_credentials.Append(fmt.Sprintf("%#x", deposit.WithdrawalCredentials))
+		f_amount.Append(uint64(deposit.Amount))
+		f_signature.Append(deposit.Signature.String())
+		f_index.Append(uint8(deposit.Index))
+	}
+
+	return proto.Input{
+		{Name: "f_slot", Data: f_slot},
+		{Name: "f_public_key", Data: f_public_key},
+		{Name: "f_withdrawal_credentials", Data: f_withdrawal_credentials},
+		{Name: "f_amount", Data: f_amount},
+		{Name: "f_signature", Data: f_signature},
+		{Name: "f_index", Data: f_index},
+	}
+}
+
+func (p *DBService) PersistDeposits(data []spec.Deposit) error {
+	persistObj := PersistableObject[spec.Deposit]{
+		input: depositsInput,
+		table: depositsTable,
+		query: insertDepositQuery,
+	}
+
+	for _, item := range data {
+		persistObj.Append(item)
+	}
+
+	err := p.Persist(persistObj.ExportPersist())
+	if err != nil {
+		log.Errorf("error persisting deposits: %s", err.Error())
+	}
+	return err
+}

pkg/db/migrations/000017_create_deposits_table.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS t_deposits;

pkg/db/migrations/000017_create_deposits_table.up.sql

@@ -0,0 +1,10 @@
+CREATE TABLE t_deposits(
+	f_slot UInt64,
+	f_public_key TEXT,
+	f_withdrawal_credentials TEXT,
+	f_amount UInt64,
+	f_signature TEXT,
+	f_index UInt8,
+	)
+	ENGINE = ReplacingMergeTree()
+	ORDER BY (f_slot, f_index);

pkg/db/prometheus_metrics.go

@@ -94,6 +94,7 @@ func (r *DBService) initMonitorMetrics() {
 		withdrawalsTable,
 		slashingsTable,
 		blsToExecutionChangeTable,
+		depositsTable,
 	}
 
 	for _, tableName := range tablesArr {

pkg/db/service.go

@@ -148,7 +148,8 @@ type PersistableObject[
 		spec.BlobSideCarEventWraper |
 		BlockReward |
 		spec.AgnosticSlashing |
-		spec.BLSToExecutionChange] struct {
+		spec.BLSToExecutionChange |
+		spec.Deposit] struct {
 	table string
 	query string
 	data  []T

pkg/spec/constants.go

@@ -71,6 +71,7 @@ const (
 	ValidatorRewardsAggregationModel
 	SlashingModel
 	BLSToExecutionChangeModel
+	DepositModel
 )
 
 type ValidatorStatus int8

pkg/spec/deposit.go

@@ -0,0 +1,30 @@
+package spec
+
+import (
+	"github.com/attestantio/go-eth2-client/spec/phase0"
+)
+
+type Deposit struct {
+	Slot                  phase0.Slot
+	PublicKey             phase0.BLSPubKey
+	WithdrawalCredentials []byte
+	Amount                phase0.Gwei
+	Signature             phase0.BLSSignature
+	Index                 uint8
+}
+
+func (f Deposit) Type() ModelType {
+	return DepositModel
+}
+
+func (f Deposit) ToArray() []interface{} {
+	rows := []interface{}{
+		f.Slot,
+		f.PublicKey,
+		f.WithdrawalCredentials,
+		f.Amount,
+		f.Signature,
+		f.Index,
+	}
+	return rows
+}