Feature: update validators

internal/storage/block.go

@@ -54,6 +54,7 @@ type Block struct {
 	Addresses       map[string]*Address       `bun:"-"` // internal field for saving address
 	Rollups         map[string]*Rollup        `bun:"-"` // internal field for saving rollups
 	RollupAddress   map[string]*RollupAddress `bun:"-"` // internal field for saving rollup address
+	Validators      map[string]*Validator     `bun:"-"` // internal field for update validators
 	BlockSignatures []BlockSignature          `bun:"-"` // internal field for saving block signatures
 
 	Txs      []*Tx       `bun:"rel:has-many"`

internal/storage/generic.go

@@ -69,6 +69,7 @@ type Transaction interface {
 	RollbackValidators(ctx context.Context, height types.Level) (err error)
 	UpdateAddresses(ctx context.Context, address ...*Address) error
 	UpdateRollups(ctx context.Context, rollups ...*Rollup) error
+	UpdateValidators(ctx context.Context, validators ...*Validator) error
 
 	LastBlock(ctx context.Context) (block Block, err error)
 	State(ctx context.Context, name string) (state State, err error)

internal/storage/postgres/transaction.go

@@ -360,3 +360,21 @@ func (tx Transaction) RetentionBlockSignatures(ctx context.Context, height types
 		Exec(ctx)
 	return err
 }
+
+func (tx Transaction) UpdateValidators(ctx context.Context, validators ...*models.Validator) error {
+	if len(validators) == 0 {
+		return nil
+	}
+
+	for _, val := range validators {
+		_, err := tx.Tx().NewUpdate().
+			Model(val).
+			Where("pubkey = ?", val.PubKey).
+			Set("power = ?", val.Power).
+			Exec(ctx)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}

internal/storage/postgres/transaction_test.go

@@ -732,3 +732,27 @@ func (s *TransactionTestSuite) TestRetentionBlockSignatures() {
 	s.Require().NoError(err)
 	s.Require().Len(signs, 3)
 }
+
+func (s *TransactionTestSuite) TestUpdateValidators() {
+	ctx, ctxCancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer ctxCancel()
+
+	tx, err := BeginTransaction(ctx, s.storage.Transactable)
+	s.Require().NoError(err)
+
+	pk, err := hex.DecodeString("32415f09dbee4297cc9a841c2c2312bf903fc53c48860d788ae66097355a585f")
+	s.Require().NoError(err)
+
+	err = tx.UpdateValidators(ctx, &storage.Validator{
+		PubKey: pk,
+		Power:  decimal.NewFromInt(10000),
+	})
+	s.Require().NoError(err)
+
+	s.Require().NoError(tx.Flush(ctx))
+	s.Require().NoError(tx.Close(ctx))
+
+	validator, err := s.storage.Validator.GetByID(ctx, 1)
+	s.Require().NoError(err)
+	s.Require().EqualValues("10000", validator.Power.String())
+}

pkg/indexer/decode/actions.go

@@ -296,8 +296,10 @@ func parseValidatorUpdateAction(body *astria.Action_ValidatorUpdateAction, heigh
 	action.Type = storageTypes.ActionTypeValidatorUpdate
 	action.Data = make(map[string]any)
 	if body.ValidatorUpdateAction != nil {
-		action.Data["power"] = body.ValidatorUpdateAction.GetPower()
-		action.Data["pubkey"] = body.ValidatorUpdateAction.GetPubKey().GetEd25519()
+		power := body.ValidatorUpdateAction.GetPower()
+		action.Data["power"] = power
+		pubKey := body.ValidatorUpdateAction.GetPubKey().GetEd25519()
+		action.Data["pubkey"] = pubKey
 
 		address := AddressFromPubKey(body.ValidatorUpdateAction.GetPubKey().GetEd25519())
 		addr := ctx.Addresses.Set(address, height, decimal.Zero, 1, 0)
@@ -308,6 +310,7 @@ func parseValidatorUpdateAction(body *astria.Action_ValidatorUpdateAction, heigh
 			Height:     action.Height,
 			ActionType: action.Type,
 		})
+		ctx.Validators.Set(pubKey, power)
 	}
 	return nil
 }

pkg/indexer/decode/actions_test.go

@@ -459,6 +459,12 @@ func TestDecodeActions(t *testing.T) {
 		err := parseValidatorUpdateAction(message, 1000, &decodeContext, &action)
 		require.NoError(t, err)
 		require.Equal(t, wantAction, action)
+
+		require.Len(t, decodeContext.Validators, 1)
+		pk := hex.EncodeToString(message.ValidatorUpdateAction.GetPubKey().GetEd25519())
+		v, ok := decodeContext.Validators[pk]
+		require.True(t, ok)
+		require.EqualValues(t, "10", v.Power.String())
 	})
 
 	t.Run("fee asset change: addition", func(t *testing.T) {

pkg/indexer/decode/tx.go

@@ -16,6 +16,7 @@ import (
 type Context struct {
 	Addresses      Addresses
 	Rollups        Rollups
+	Validators     Validators
 	RollupAddress  map[string]*storage.RollupAddress
 	AddressActions map[string]*storage.AddressAction
 	SupplyChange   decimal.Decimal
@@ -32,6 +33,7 @@ func NewContext() Context {
 		Rollups:       NewRollups(),
 		RollupAddress: make(map[string]*storage.RollupAddress),
 		SupplyChange:  decimal.Zero,
+		Validators:    NewValidators(),
 	}
 }
 

pkg/indexer/decode/validators.go

@@ -0,0 +1,34 @@
+// SPDX-FileCopyrightText: 2024 PK Lab AG <[email protected]>
+// SPDX-License-Identifier: MIT
+
+package decode
+
+import (
+	"encoding/hex"
+
+	"github.com/celenium-io/astria-indexer/internal/storage"
+	"github.com/shopspring/decimal"
+)
+
+type Validators map[string]*storage.Validator
+
+func NewValidators() Validators {
+	return make(map[string]*storage.Validator)
+}
+
+func (v Validators) Set(pubKey []byte, power int64) *storage.Validator {
+	sPubKey := hex.EncodeToString(pubKey)
+
+	pow := decimal.NewFromInt(power)
+	if validator, ok := v[sPubKey]; ok {
+		validator.Power = pow
+		return validator
+	}
+
+	validator := &storage.Validator{
+		PubKey: pubKey,
+		Power:  pow,
+	}
+	v[sPubKey] = validator
+	return validator
+}

pkg/indexer/parser/parse.go

@@ -50,6 +50,7 @@ func (p *Module) parse(b types.BlockData) error {
 		Addresses:     decodeCtx.Addresses,
 		Rollups:       decodeCtx.Rollups,
 		RollupAddress: decodeCtx.RollupAddress,
+		Validators:    decodeCtx.Validators,
 		ActionTypes:   decodeCtx.ActionTypes,
 
 		Txs: txs,

pkg/indexer/parser/parser_test.go

@@ -61,6 +61,7 @@ func getExpectedBlock() storage.Block {
 		Addresses:       make(map[string]*storage.Address),
 		Rollups:         make(map[string]*storage.Rollup),
 		RollupAddress:   make(map[string]*storage.RollupAddress),
+		Validators:      make(map[string]*storage.Validator),
 		BlockSignatures: []storage.BlockSignature{},
 	}
 }

pkg/indexer/storage/storage.go

@@ -194,6 +194,10 @@ func (module *Module) processBlockInTransaction(ctx context.Context, tx storage.
 		return state, err
 	}
 
+	if err := saveValidators(ctx, tx, block.Validators); err != nil {
+		return state, err
+	}
+
 	updateState(block, totalAccounts, totalRollups, &state)
 	if err := tx.Update(ctx, &state); err != nil {
 		return state, err

pkg/indexer/storage/validators.go

@@ -0,0 +1,27 @@
+// SPDX-FileCopyrightText: 2024 PK Lab AG <[email protected]>
+// SPDX-License-Identifier: MIT
+
+package storage
+
+import (
+	"context"
+
+	"github.com/celenium-io/astria-indexer/internal/storage"
+)
+
+func saveValidators(
+	ctx context.Context,
+	tx storage.Transaction,
+	validators map[string]*storage.Validator,
+) error {
+	if len(validators) == 0 {
+		return nil
+	}
+
+	vals := make([]*storage.Validator, 0)
+	for _, val := range validators {
+		vals = append(vals, val)
+	}
+
+	return tx.UpdateValidators(ctx, vals...)
+}