feat(cannon): Add BLS_To_Execution

pkg/cannon/cannon.go

@@ -69,6 +69,8 @@ func New(ctx context.Context, log logrus.FieldLogger, config *Config) (*Cannon,
 		v2.NewProposerSlashingDeriver(log),
 		v2.NewVoluntaryExitDeriver(log),
 		v2.NewDepositDeriver(log),
+		v2.NewBLSToExecutionChangeDeriver(log),
+		v2.NewExecutionTransactionDeriver(log),
 	}
 
 	return &Cannon{
@@ -279,7 +281,7 @@ func (c *Cannon) startBeaconBlockProcessor(ctx context.Context) error {
 		c.log.Info("Internal beacon node is ready, firing up beacon block processor")
 
 		// TODO: Fetch our starting point from xatu-server
-		start := uint64(0)
+		start := uint64(5193791)
 
 		for {
 			select {

pkg/cannon/event/beacon/eth/v2/beacon_block.go

@@ -123,3 +123,7 @@ func (e *BeaconBlockMetadata) Process(ctx context.Context) ([]*xatu.DecoratedEve
 
 	return events, nil
 }
+
+func (e *BeaconBlockMetadata) Block() *spec.VersionedSignedBeaconBlock {
+	return e.block
+}

pkg/cannon/event/beacon/eth/v2/bls_to_execution_change.go

@@ -0,0 +1,116 @@
+package v2
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/attestantio/go-eth2-client/spec"
+	xatuethv2 "github.com/ethpandaops/xatu/pkg/proto/eth/v2"
+	"github.com/ethpandaops/xatu/pkg/proto/xatu"
+	"github.com/google/uuid"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	"google.golang.org/protobuf/types/known/timestamppb"
+	"google.golang.org/protobuf/types/known/wrapperspb"
+)
+
+const (
+	BLSToExecutionChangeDeriverName = "BEACON_API_ETH_V2_BEACON_BLOCK_BLS_TO_EXECUTION_CHANGE"
+)
+
+type BLSToExecutionChangeDeriver struct {
+	log logrus.FieldLogger
+}
+
+func NewBLSToExecutionChangeDeriver(log logrus.FieldLogger) *BLSToExecutionChangeDeriver {
+	return &BLSToExecutionChangeDeriver{
+		log: log.WithField("module", "cannon/event/beacon/eth/v2/bls_to_execution_change"),
+	}
+}
+
+func (b *BLSToExecutionChangeDeriver) Name() string {
+	return BLSToExecutionChangeDeriverName
+}
+
+func (b *BLSToExecutionChangeDeriver) Filter(ctx context.Context) bool {
+	return false
+}
+
+func (b *BLSToExecutionChangeDeriver) Process(ctx context.Context, metadata *BeaconBlockMetadata, block *spec.VersionedSignedBeaconBlock) ([]*xatu.DecoratedEvent, error) {
+	events := []*xatu.DecoratedEvent{}
+
+	changes, err := b.getBLSToExecutionChanges(ctx, block)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, change := range changes {
+		event, err := b.createEvent(ctx, metadata, change)
+		if err != nil {
+			b.log.WithError(err).Error("Failed to create event")
+
+			return nil, errors.Wrapf(err, "failed to create event for BLS to execution change %s", change.String())
+		}
+
+		events = append(events, event)
+	}
+
+	return events, nil
+}
+
+func (b *BLSToExecutionChangeDeriver) getBLSToExecutionChanges(ctx context.Context, block *spec.VersionedSignedBeaconBlock) ([]*xatuethv2.SignedBLSToExecutionChangeV2, error) {
+	changes := []*xatuethv2.SignedBLSToExecutionChangeV2{}
+
+	switch block.Version {
+	case spec.DataVersionPhase0:
+		return changes, nil
+	case spec.DataVersionAltair:
+		return changes, nil
+	case spec.DataVersionBellatrix:
+		return changes, nil
+	case spec.DataVersionCapella:
+		for _, change := range block.Capella.Message.Body.BLSToExecutionChanges {
+			changes = append(changes, &xatuethv2.SignedBLSToExecutionChangeV2{
+				Message: &xatuethv2.BLSToExecutionChangeV2{
+					ValidatorIndex:     wrapperspb.UInt64(uint64(change.Message.ValidatorIndex)),
+					FromBlsPubkey:      change.Message.FromBLSPubkey.String(),
+					ToExecutionAddress: change.Message.ToExecutionAddress.String(),
+				},
+				Signature: change.Signature.String(),
+			})
+		}
+	default:
+		return nil, fmt.Errorf("unsupported block version: %s", block.Version.String())
+	}
+
+	return changes, nil
+}
+
+func (b *BLSToExecutionChangeDeriver) createEvent(ctx context.Context, metadata *BeaconBlockMetadata, change *xatuethv2.SignedBLSToExecutionChangeV2) (*xatu.DecoratedEvent, error) {
+	decoratedEvent := &xatu.DecoratedEvent{
+		Event: &xatu.Event{
+			Name:     xatu.Event_BEACON_API_ETH_V2_BEACON_BLOCK_BLS_TO_EXECUTION_CHANGE,
+			DateTime: timestamppb.New(metadata.Now),
+			Id:       uuid.New().String(),
+		},
+		Meta: &xatu.Meta{
+			Client: metadata.ClientMeta,
+		},
+		Data: &xatu.DecoratedEvent_EthV2BeaconBlockBlsToExecutionChange{
+			EthV2BeaconBlockBlsToExecutionChange: change,
+		},
+	}
+
+	blockIdentifier, err := metadata.BlockIdentifier()
+	if err != nil {
+		return nil, err
+	}
+
+	decoratedEvent.Meta.Client.AdditionalData = &xatu.ClientMeta_EthV2BeaconBlockBlsToExecutionChange{
+		EthV2BeaconBlockBlsToExecutionChange: &xatu.ClientMeta_AdditionalEthV2BeaconBlockBLSToExecutionChangeData{
+			Block: blockIdentifier,
+		},
+	}
+
+	return decoratedEvent, nil
+}

pkg/cannon/event/beacon/eth/v2/execution_transaction.go

@@ -0,0 +1,163 @@
+package v2
+
+import (
+	"context"
+	"encoding/hex"
+	"fmt"
+
+	"github.com/attestantio/go-eth2-client/spec"
+	"github.com/ethereum/go-ethereum/core/types"
+	xatuethv1 "github.com/ethpandaops/xatu/pkg/proto/eth/v1"
+	"github.com/ethpandaops/xatu/pkg/proto/xatu"
+	"github.com/google/uuid"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	"google.golang.org/protobuf/types/known/timestamppb"
+	"google.golang.org/protobuf/types/known/wrapperspb"
+)
+
+type ExecutionTransactionDeriver struct {
+	log logrus.FieldLogger
+}
+
+const (
+	ExecutionTransactionDeriverName = "BEACON_API_ETH_V2_BEACON_BLOCK_EXECUTION_TRANSACTION"
+)
+
+func NewExecutionTransactionDeriver(log logrus.FieldLogger) *ExecutionTransactionDeriver {
+	return &ExecutionTransactionDeriver{
+		log: log.WithField("module", "cannon/event/beacon/eth/v2/execution_transaction"),
+	}
+}
+
+func (b *ExecutionTransactionDeriver) Name() string {
+	return ExecutionTransactionDeriverName
+}
+
+func (b *ExecutionTransactionDeriver) Filter(ctx context.Context) bool {
+	return false
+}
+
+func (b *ExecutionTransactionDeriver) Process(ctx context.Context, metadata *BeaconBlockMetadata, block *spec.VersionedSignedBeaconBlock) ([]*xatu.DecoratedEvent, error) {
+	events := []*xatu.DecoratedEvent{}
+
+	transactions, err := b.getExecutionTransactions(ctx, block, metadata)
+	if err != nil {
+		return nil, err
+	}
+
+	for index, transaction := range transactions {
+		from, err := types.Sender(types.LatestSignerForChainID(transaction.ChainId()), transaction)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get transaction sender: %v", err)
+		}
+
+		gasPrice := transaction.GasPrice()
+		if gasPrice == nil {
+			return nil, fmt.Errorf("failed to get transaction gas price")
+		}
+
+		value := transaction.Value()
+		if value == nil {
+			return nil, fmt.Errorf("failed to get transaction value")
+		}
+
+		to := ""
+
+		if transaction.To() != nil {
+			to = transaction.To().Hex()
+		}
+
+		chainID := transaction.ChainId()
+		if chainID == nil {
+			return nil, fmt.Errorf("failed to get transaction chain ID")
+		}
+
+		tx := &xatuethv1.Transaction{
+			Nonce:    wrapperspb.UInt64(transaction.Nonce()),
+			GasPrice: gasPrice.String(),
+			Gas:      wrapperspb.UInt64(transaction.Gas()),
+			To:       to,
+			From:     from.Hex(),
+			Value:    value.String(),
+			Input:    hex.EncodeToString(transaction.Data()),
+			Hash:     transaction.Hash().Hex(),
+			ChainId:  chainID.String(),
+			Type:     wrapperspb.UInt32(uint32(transaction.Type())),
+		}
+
+		event, err := b.createEvent(ctx, metadata, tx, uint64(index))
+		if err != nil {
+			b.log.WithError(err).Error("Failed to create event")
+
+			return nil, errors.Wrapf(err, "failed to create event for execution transaction %s", transaction.Hash())
+		}
+
+		events = append(events, event)
+	}
+
+	return events, nil
+}
+
+func (b *ExecutionTransactionDeriver) getExecutionTransactions(ctx context.Context, block *spec.VersionedSignedBeaconBlock, metadata *BeaconBlockMetadata) ([]*types.Transaction, error) {
+	transactions := []*types.Transaction{}
+
+	switch block.Version {
+	case spec.DataVersionPhase0:
+		return transactions, nil
+	case spec.DataVersionAltair:
+		return transactions, nil
+	case spec.DataVersionBellatrix:
+		for _, transaction := range block.Bellatrix.Message.Body.ExecutionPayload.Transactions {
+			ethTransaction := new(types.Transaction)
+			if err := ethTransaction.UnmarshalBinary(transaction); err != nil {
+				return nil, fmt.Errorf("failed to unmarshal transaction: %v", err)
+			}
+
+			transactions = append(transactions, ethTransaction)
+		}
+	case spec.DataVersionCapella:
+		for _, transaction := range block.Capella.Message.Body.ExecutionPayload.Transactions {
+			ethTransaction := new(types.Transaction)
+			if err := ethTransaction.UnmarshalBinary(transaction); err != nil {
+				return nil, fmt.Errorf("failed to unmarshal transaction: %v", err)
+			}
+
+			transactions = append(transactions, ethTransaction)
+		}
+	default:
+		return nil, fmt.Errorf("unsupported block version: %s", block.Version.String())
+	}
+
+	return transactions, nil
+}
+
+func (b *ExecutionTransactionDeriver) createEvent(ctx context.Context, metadata *BeaconBlockMetadata, transaction *xatuethv1.Transaction, positionInBlock uint64) (*xatu.DecoratedEvent, error) {
+	decoratedEvent := &xatu.DecoratedEvent{
+		Event: &xatu.Event{
+			Name:     xatu.Event_BEACON_API_ETH_V2_BEACON_BLOCK_EXECUTION_TRANSACTION,
+			DateTime: timestamppb.New(metadata.Now),
+			Id:       uuid.New().String(),
+		},
+		Meta: &xatu.Meta{
+			Client: metadata.ClientMeta,
+		},
+		Data: &xatu.DecoratedEvent_EthV2BeaconBlockExecutionTransaction{
+			EthV2BeaconBlockExecutionTransaction: transaction,
+		},
+	}
+
+	blockIdentifier, err := metadata.BlockIdentifier()
+	if err != nil {
+		return nil, err
+	}
+
+	decoratedEvent.Meta.Client.AdditionalData = &xatu.ClientMeta_EthV2BeaconBlockExecutionTransaction{
+		EthV2BeaconBlockExecutionTransaction: &xatu.ClientMeta_AdditionalEthV2BeaconBlockExecutionTransactionData{
+			Block:           blockIdentifier,
+			PositionInBlock: wrapperspb.UInt64(positionInBlock),
+		},
+	}
+
+	return decoratedEvent, nil
+}