fix: add refetch blocks and transactions fix command

cmd/fix/blocks/blocks.go

@@ -0,0 +1,118 @@
+package blocks
+
+import (
+	"encoding/hex"
+	"fmt"
+	"strings"
+
+	"github.com/forbole/bdjuno/v2/database"
+	"github.com/forbole/bdjuno/v2/modules"
+	"github.com/forbole/bdjuno/v2/modules/bank"
+	"github.com/forbole/bdjuno/v2/modules/consensus"
+	"github.com/forbole/bdjuno/v2/modules/distribution"
+	"github.com/forbole/bdjuno/v2/modules/gov"
+	"github.com/forbole/bdjuno/v2/modules/history"
+	"github.com/forbole/bdjuno/v2/modules/staking"
+	"github.com/forbole/bdjuno/v2/utils"
+	"github.com/forbole/juno/v2/cmd/parse"
+	junomessages "github.com/forbole/juno/v2/modules/messages"
+	"github.com/forbole/juno/v2/types/config"
+	"github.com/spf13/cobra"
+	tmctypes "github.com/tendermint/tendermint/rpc/core/types"
+)
+
+// blocksCmd returns a Cobra command that allows to fix missing blocks in database
+func blocksCmd(parseConfig *parse.Config) *cobra.Command {
+	return &cobra.Command{
+		Use:   "refetch",
+		Short: "Fix missing blocks and transactions in database from the start height",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			parseCtx, err := parse.GetParsingContext(parseConfig)
+			if err != nil {
+				return err
+			}
+
+			sources, err := modules.BuildSources(config.Cfg.Node, parseCtx.EncodingConfig)
+			if err != nil {
+				return err
+			}
+
+			// Get the database
+			db := database.Cast(parseCtx.Database)
+
+			// Register modules
+			bankModule := bank.NewModule(junomessages.BankMessagesParser, sources.BankSource, parseCtx.EncodingConfig.Marshaler, db)
+			distrModule := distribution.NewModule(config.Cfg, sources.DistrSource, bankModule, parseCtx.EncodingConfig.Marshaler, db)
+			historyModule := history.NewModule(config.Cfg.Chain, junomessages.BankMessagesParser, parseCtx.EncodingConfig.Marshaler, db)
+			stakingModule := staking.NewModule(sources.StakingSource, bankModule, distrModule, historyModule, nil, parseCtx.EncodingConfig.Marshaler, db)
+			govModule := gov.NewModule(sources.GovSource, nil, bankModule, distrModule, nil, nil, stakingModule, parseCtx.EncodingConfig.Marshaler, db)
+
+			// Build the consensus module
+			consensusModule := consensus.NewModule(bankModule, distrModule, govModule, stakingModule, db)
+
+			// Get latest height
+			height, err := parseCtx.Node.LatestHeight()
+			if err != nil {
+				return fmt.Errorf("error while getting chain latest block height: %s", err)
+			}
+
+			k := config.Cfg.Parser.StartHeight
+			fmt.Printf("Refetching missing blocks and transactions from height %d ... \n", k)
+			for ; k <= height; k++ {
+				missingBlock := consensusModule.IsBlockMissing(k)
+				if missingBlock {
+					fmt.Printf("Refetching block %d ... \n", k)
+					err = refreshBlock(parseCtx, k, consensusModule)
+					if err != nil {
+						return err
+					}
+				}
+			}
+
+			return nil
+		},
+	}
+}
+
+func refreshBlock(parseCtx *parse.Context, blockHeight int64, consensusModule *consensus.Module) error {
+	// Get the block details
+	block, err := utils.QueryBlock(parseCtx.Node, blockHeight)
+	if err != nil {
+		return err
+	}
+
+	err = consensusModule.UpdateBlock(block)
+
+	if len(block.Block.Txs) != 0 {
+		for _, tx := range block.Block.Txs {
+			fmt.Printf("Refetching tx %v ... \n", strings.ToUpper(hex.EncodeToString(tx.Hash())))
+			err = refreshTxs(parseCtx, consensusModule, block)
+			if err != nil {
+				return fmt.Errorf("error when updatig tx %s", err)
+			}
+		}
+	}
+	if err != nil {
+		return fmt.Errorf("error while updating block %d: %s", blockHeight, err)
+	}
+
+	return nil
+}
+
+func refreshTxs(parseCtx *parse.Context, consensusModule *consensus.Module, block *tmctypes.ResultBlock) error {
+
+	for _, tx := range block.Block.Txs {
+		// Get the tx details
+		txDetails, err := parseCtx.Node.Tx(hex.EncodeToString(tx.Hash()))
+		if err != nil {
+			return fmt.Errorf("error while encoding tx to string %s", err)
+		}
+
+		err = consensusModule.HandleMessages(txDetails)
+		if err != nil {
+			return fmt.Errorf("error when updating tx message tx %s", err)
+		}
+	}
+
+	return nil
+}

cmd/fix/blocks/cmd.go

@@ -0,0 +1,20 @@
+package blocks
+
+import (
+	"github.com/forbole/juno/v2/cmd/parse"
+	"github.com/spf13/cobra"
+)
+
+// NewBlocksCmd returns the Cobra command that allows to fix all the things related to blocks
+func NewBlocksCmd(parseConfig *parse.Config) *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "blocks",
+		Short: "Fix things related to blocks and transactions",
+	}
+
+	cmd.AddCommand(
+		blocksCmd(parseConfig),
+	)
+
+	return cmd
+}

cmd/fix/fix.go

@@ -5,6 +5,7 @@ import (
 	"github.com/spf13/cobra"
 
 	fixauth "github.com/forbole/bdjuno/v2/cmd/fix/auth"
+	fixblocks "github.com/forbole/bdjuno/v2/cmd/fix/blocks"
 	fixgov "github.com/forbole/bdjuno/v2/cmd/fix/gov"
 	fixstaking "github.com/forbole/bdjuno/v2/cmd/fix/staking"
 )
@@ -19,6 +20,7 @@ func NewFixCmd(parseCfg *parse.Config) *cobra.Command {
 
 	cmd.AddCommand(
 		fixauth.NewAuthCmd(parseCfg),
+		fixblocks.NewBlocksCmd(parseCfg),
 		fixgov.NewGovCmd(parseCfg),
 		fixstaking.NewStakingCmd(parseCfg),
 	)

database/bank.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 
 	dbtypes "github.com/forbole/bdjuno/v2/database/types"
+	"github.com/forbole/bdjuno/v2/utils"
 
 	dbutils "github.com/forbole/bdjuno/v2/database/utils"
 
@@ -40,23 +41,34 @@ func (db *Db) SaveAccountBalances(balances []types.AccountBalance) error {
 func (db *Db) saveUpToDateBalances(paramsNumber int, balances []types.AccountBalance) error {
 	stmt := `INSERT INTO account_balance (address, coins, height) VALUES `
 	var params []interface{}
+	var accounts []types.Account
+	// remove duplicate values of accounts balance
+	balances = utils.RemoveDuplicateAccountBalance(balances)
 
 	for i, bal := range balances {
+
 		bi := i * paramsNumber
 		stmt += fmt.Sprintf("($%d, $%d, $%d),", bi+1, bi+2, bi+3)
+		accounts = append(accounts, types.NewAccount(bal.Address))
 
 		coins := pq.Array(dbtypes.NewDbCoins(bal.Balance))
 		params = append(params, bal.Address, coins, bal.Height)
 	}
 
+	// Store the accounts
+	err := db.SaveAccounts(accounts)
+	if err != nil {
+		return fmt.Errorf("error while storing account balance : %s", err)
+	}
+
 	stmt = stmt[:len(stmt)-1]
 	stmt += `
 ON CONFLICT (address) DO UPDATE 
 	SET coins = excluded.coins, 
 	    height = excluded.height 
 WHERE account_balance.height <= excluded.height`
 
-	_, err := db.Sql.Exec(stmt, params...)
+	_, err = db.Sql.Exec(stmt, params...)
 	if err != nil {
 		return fmt.Errorf("error while storing up-to-date balances: %s", err)
 	}

database/consensus.go

@@ -4,9 +4,14 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/cosmos/cosmos-sdk/codec"
 	"github.com/forbole/bdjuno/v2/types"
+	"github.com/lib/pq"
 
 	dbtypes "github.com/forbole/bdjuno/v2/database/types"
+	"github.com/forbole/bdjuno/v2/utils"
+	junotypes "github.com/forbole/juno/v2/types"
+	tmctypes "github.com/tendermint/tendermint/rpc/core/types"
 )
 
 // GetLastBlock returns the last block stored inside the database based on the heights
@@ -184,3 +189,86 @@ func (db *Db) GetGenesis() (*types.Genesis, error) {
 	row := rows[0]
 	return types.NewGenesis(row.ChainID, row.Time, row.InitialHeight), nil
 }
+
+// CheckIfBlockIsMissing checks if block is already stored in database
+func (db *Db) CheckIfBlockIsMissing(height int64) bool {
+	var block []dbtypes.BlockRow
+	stmt := `SELECT * FROM block WHERE height = $1`
+
+	err := db.Sqlx.Select(&block, stmt, height)
+	if err != nil {
+		return true
+	}
+	if len(block) != 0 {
+		return false
+	}
+
+	return true
+}
+
+// UpdateBlocksInDatabase updates given block in database
+func (db *Db) UpdateBlockInDatabase(block *tmctypes.ResultBlock) error {
+	junoBlock := junotypes.NewBlockFromTmBlock(block, uint64(block.Block.Height))
+	err := db.SaveBlock(junoBlock)
+	if err != nil {
+		return fmt.Errorf("error while storing block %d, error:  %s", block.Block.Height, err)
+	}
+	return nil
+}
+
+// UpdateTxInDatabase updates transactions for a given block in database
+func (db *Db) UpdateTxInDatabase(i int, tx *junotypes.Tx) error {
+	err := db.SaveTx(tx)
+	if err != nil {
+		return fmt.Errorf("error while storing tx in database: %s", err)
+	}
+	err = db.UpdateMsgsInDatabase(tx, i)
+	if err != nil {
+		return fmt.Errorf("error while updating tx message in database: %s", err)
+	}
+
+	return nil
+}
+
+// UpdateMsgsInDatabase updates messages for a given block in database
+func (db *Db) UpdateMsgsInDatabase(tx *junotypes.Tx, i int) error {
+	var eventTypes []string
+	var involvedAccounts []string
+	var attributeKeys []string
+	message, err := codec.ProtoMarshalJSON(tx.Body.Messages[i], nil)
+	if err != nil {
+		return fmt.Errorf("error while marshaling tx message: %s", err)
+	}
+
+	for _, event := range tx.Logs {
+		for _, eventType := range event.Events {
+			eventTypess := eventType.Type
+			eventTypes = append(eventTypes, eventTypess)
+			attributes := eventType.Attributes
+			for _, attribute := range attributes {
+				attributeKeys = append(attributeKeys, attribute.Key)
+			}
+		}
+	}
+	attributeKeys = utils.RemoveDuplicateValues(attributeKeys)
+
+	for _, eventType := range eventTypes {
+		event, _ := tx.FindEventByType(i, eventType)
+		for _, key := range attributeKeys {
+			address, _ := tx.FindAttributeByKey(event, key)
+			// process only addresses
+			if len(address) >= 40 {
+				involvedAccounts = append(involvedAccounts, address)
+			}
+		}
+		involvedAccounts = utils.RemoveDuplicateValues(involvedAccounts)
+	}
+
+	junoMsg := junotypes.NewMessage(tx.TxHash, i, tx.Body.Messages[i].TypeUrl, string(message), pq.StringArray(involvedAccounts))
+	err = db.SaveMessage(junoMsg)
+	if err != nil {
+		return fmt.Errorf("error while storing tx message in database, error:  %s", err)
+	}
+
+	return nil
+}

database/staking_delegations.go

@@ -56,7 +56,7 @@ INSERT INTO delegation (validator_address, delegator_address, amount, height) VA
 		// Get the validator consensus address
 		consAddr, err := db.GetValidatorConsensusAddress(delegation.ValidatorOperAddr)
 		if err != nil {
-			return fmt.Errorf("error while gettting validator consensus address: %s", err)
+			return err
 		}
 
 		// Convert the amount

modules/consensus/expected_modules.go

@@ -0,0 +1,22 @@
+package consensus
+
+import (
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/forbole/juno/v2/types"
+)
+
+type BankModule interface {
+	HandleMsg(_ int, msg sdk.Msg, tx *types.Tx) error
+}
+
+type DistrModule interface {
+	HandleMsg(_ int, msg sdk.Msg, tx *types.Tx) error
+}
+
+type GovModule interface {
+	HandleMsg(_ int, msg sdk.Msg, tx *types.Tx) error
+}
+
+type StakingModule interface {
+	HandleMsg(_ int, msg sdk.Msg, tx *types.Tx) error
+}

modules/consensus/module.go

@@ -15,13 +15,25 @@ var (
 
 // Module implements the consensus utils
 type Module struct {
-	db *database.Db
+	db            *database.Db
+	bankModule    BankModule
+	distrModule   DistrModule
+	govModule     GovModule
+	stakingModule StakingModule
 }
 
 // NewModule builds a new Module instance
-func NewModule(db *database.Db) *Module {
+func NewModule(bankModule BankModule,
+	distrModule DistrModule,
+	govModule GovModule,
+	stakingModule StakingModule,
+	db *database.Db) *Module {
 	return &Module{
-		db: db,
+		bankModule:    bankModule,
+		distrModule:   distrModule,
+		govModule:     govModule,
+		stakingModule: stakingModule,
+		db:            db,
 	}
 }