From f04fc009d4db847cc5b110d18e79d507cb7c99fb Mon Sep 17 00:00:00 2001 From: Magic Cat Date: Mon, 22 Jan 2024 21:23:54 +0700 Subject: [PATCH] updated migrator --- cmd/migrate/cmd.go | 4 +- cmd/migrate/msgexec/migrate.go | 44 +++++++++ cmd/migrate/msgexec/utils.go | 31 +++++++ cmd/migrate/{v4 => utils}/types.go | 2 +- cmd/migrate/v4/migrate.go | 8 +- database/legacy/msgexec/migrate.go | 133 ++++++++++++++++++++++++++++ database/legacy/msgexec/migrator.go | 21 +++++ database/migrate/utils/types.go | 11 +++ modules/messages/types.go | 4 +- 9 files changed, 250 insertions(+), 8 deletions(-) create mode 100644 cmd/migrate/msgexec/migrate.go create mode 100644 cmd/migrate/msgexec/utils.go rename cmd/migrate/{v4 => utils}/types.go (98%) create mode 100644 database/legacy/msgexec/migrate.go create mode 100644 database/legacy/msgexec/migrator.go diff --git a/cmd/migrate/cmd.go b/cmd/migrate/cmd.go index bb0a4d49..b9b70f85 100644 --- a/cmd/migrate/cmd.go +++ b/cmd/migrate/cmd.go @@ -8,6 +8,7 @@ import ( "github.com/spf13/cobra" + "github.com/forbole/juno/v5/cmd/migrate/msgexec" v4 "github.com/forbole/juno/v5/cmd/migrate/v4" ) @@ -15,7 +16,8 @@ type Migrator func(parseCfg *parsecmdtypes.Config) error var ( migrations = map[string]Migrator{ - "v4": v4.RunMigration, + "v4": v4.RunMigration, + "msgexec": msgexec.RunMigration, } ) diff --git a/cmd/migrate/msgexec/migrate.go b/cmd/migrate/msgexec/migrate.go new file mode 100644 index 00000000..5fd23b62 --- /dev/null +++ b/cmd/migrate/msgexec/migrate.go @@ -0,0 +1,44 @@ +package msgexec + +import ( + "fmt" + + utils "github.com/forbole/juno/v5/cmd/migrate/utils" + + parse "github.com/forbole/juno/v5/cmd/parse/types" + "github.com/forbole/juno/v5/database" + msgexecdb "github.com/forbole/juno/v5/database/legacy/msgexec" + "github.com/forbole/juno/v5/database/postgresql" +) + +// RunMigration runs the migrations from v4 to v5 +func RunMigration(parseConfig *parse.Config) error { + cfg, err := GetConfig() + if err != nil { + return fmt.Errorf("error while reading config: %s", err) + } + + // Migrate the database + err = migrateDb(cfg, parseConfig) + if err != nil { + return fmt.Errorf("error while migrating database: %s", err) + } + + return nil +} + +func migrateDb(cfg utils.Config, parseConfig *parse.Config) error { + // Build the codec + encodingConfig := parseConfig.GetEncodingConfigBuilder()() + + // Get the db + databaseCtx := database.NewContext(cfg.Database, encodingConfig, parseConfig.GetLogger()) + db, err := postgresql.Builder(databaseCtx) + if err != nil { + return fmt.Errorf("error while building the db: %s", err) + } + + // Build the migrator and perform the migrations + migrator := msgexecdb.NewMigrator(db.(*postgresql.Database)) + return migrator.Migrate() +} diff --git a/cmd/migrate/msgexec/utils.go b/cmd/migrate/msgexec/utils.go new file mode 100644 index 00000000..026f8680 --- /dev/null +++ b/cmd/migrate/msgexec/utils.go @@ -0,0 +1,31 @@ +package msgexec + +import ( + "fmt" + "os" + "path" + + utils "github.com/forbole/juno/v5/cmd/migrate/utils" + "gopkg.in/yaml.v3" + + "github.com/forbole/juno/v5/types/config" +) + +// GetConfig returns the configuration reading it from the config.yaml file present inside the home directory +func GetConfig() (utils.Config, error) { + file := path.Join(config.HomePath, "config.yaml") + + // Make sure the path exists + if _, err := os.Stat(file); os.IsNotExist(err) { + return utils.Config{}, fmt.Errorf("config file does not exist") + } + + bz, err := os.ReadFile(file) + if err != nil { + return utils.Config{}, fmt.Errorf("error while reading config files: %s", err) + } + + var cfg utils.Config + err = yaml.Unmarshal(bz, &cfg) + return cfg, err +} \ No newline at end of file diff --git a/cmd/migrate/v4/types.go b/cmd/migrate/utils/types.go similarity index 98% rename from cmd/migrate/v4/types.go rename to cmd/migrate/utils/types.go index 37e90937..bc8f4fc7 100644 --- a/cmd/migrate/v4/types.go +++ b/cmd/migrate/utils/types.go @@ -1,4 +1,4 @@ -package v4 +package utils import ( databaseconfig "github.com/forbole/juno/v5/database/config" diff --git a/cmd/migrate/v4/migrate.go b/cmd/migrate/v4/migrate.go index 3e03a6f3..c0cd9344 100644 --- a/cmd/migrate/v4/migrate.go +++ b/cmd/migrate/v4/migrate.go @@ -5,7 +5,7 @@ import ( "os" parsecmdtypes "github.com/forbole/juno/v5/cmd/parse/types" - +utils "github.com/forbole/juno/v5/cmd/migrate/utils" "gopkg.in/yaml.v3" v3 "github.com/forbole/juno/v5/cmd/migrate/v3" @@ -40,10 +40,10 @@ func RunMigration(parseConfig *parsecmdtypes.Config) error { return nil } -func migrateConfig() (Config, error) { +func migrateConfig() (utils.Config, error) { cfg, err := v3.GetConfig() if err != nil { - return Config{}, fmt.Errorf("error while reading v3 config: %s", err) + return utils.Config{}, fmt.Errorf("error while reading v3 config: %s", err) } sslMode := cfg.Database.SSLMode @@ -56,7 +56,7 @@ func migrateConfig() (Config, error) { schema = "public" } - return Config{ + return utils.Config{ Node: cfg.Node, Chain: cfg.Chain, Database: databaseconfig.Config{ diff --git a/database/legacy/msgexec/migrate.go b/database/legacy/msgexec/migrate.go new file mode 100644 index 00000000..9218c147 --- /dev/null +++ b/database/legacy/msgexec/migrate.go @@ -0,0 +1,133 @@ +package msgexec + +import ( + "encoding/json" + "fmt" + + sdk "github.com/cosmos/cosmos-sdk/types" + "github.com/lib/pq" + "github.com/rs/zerolog/log" + + dbtypes "github.com/forbole/juno/v5/database/migrate/utils" + "github.com/forbole/juno/v5/modules/messages" + "github.com/forbole/juno/v5/types" +) + +// Migrate implements database.Migrator +func (db *Migrator) Migrate() error { + msgTypes, err := db.getAllMsgExecStoredInDatabase() + if err != nil { + return fmt.Errorf("error while getting message types rows: %s", err) + } + + var skipped = 0 + // Migrate the transactions + log.Info().Msg("** migrating transactions **") + log.Debug().Int("tx count", len(msgTypes)).Msg("processing total transactions") + + for i, msgType := range msgTypes { + log.Debug().Str("tx hash", msgType.TransactionHash).Msg("getting transaction....") + fmt.Printf("\n processing %d/%d transaction \n", i, len(msgTypes)) + + tx, err := db.getMsgExecTransactionsFromDatabase(msgType.TransactionHash) + if err != nil { + return fmt.Errorf("error while getting transaction %s: %s", msgType.TransactionHash, err) + } + + if tx.Success == "true" { + var msgs sdk.ABCIMessageLogs + err = json.Unmarshal([]byte(tx.Logs), &msgs) + if err != nil { + skipped++ + continue + } + + var addresses []string + + for _, msg := range msgs { + for _, event := range msg.Events { + for _, attribute := range event.Attributes { + // Try parsing the address as a validator address + validatorAddress, _ := sdk.ValAddressFromBech32(attribute.Value) + if validatorAddress != nil { + addresses = append(addresses, validatorAddress.String()) + } + + // Try parsing the address as an account address + accountAddress, err := sdk.AccAddressFromBech32(attribute.Value) + if err != nil { + // Skip if the address is not an account address + continue + } + + addresses = append(addresses, accountAddress.String()) + } + } + } + involvedAddresses := messages.RemoveDuplicates(addresses) + + fmt.Printf("\n ADDRESSES BEFORE %s", msgType.InvolvedAccountsAddresses) + fmt.Printf("\n ADDRESSES AFTER %s \n", involvedAddresses) + + err = db.updateMessage(types.NewMessage(msgType.TransactionHash, + int(msgType.Index), + msgType.Type, + msgType.Value, + involvedAddresses, + msgType.Height), msgType.PartitionID) + + if err != nil { + fmt.Printf("error while storing updated message: %s", err) + skipped++ + continue + } + } else { + skipped++ + } + + } + + log.Debug().Int("*** Total Skipped ***", skipped) + + return nil + +} + +// getMsgTypesFromMessageTable retrieves messages types stored in database inside message table +func (db *Migrator) getAllMsgExecStoredInDatabase() ([]dbtypes.MessageRow, error) { + const msgType = "cosmos.authz.v1beta1.MsgExec" + var rows []dbtypes.MessageRow + err := db.SQL.Select(&rows, `SELECT * FROM message WHERE type = $1 ORDER BY height ASC`, msgType) + if err != nil { + return nil, err + } + + return rows, nil +} + +// getMsgTypesFromMessageTable retrieves messages types stored in database inside message table +func (db *Migrator) getMsgExecTransactionsFromDatabase(txHash string) (dbtypes.TransactionRow, error) { + var rows []dbtypes.TransactionRow + err := db.SQL.Select(&rows, `SELECT * FROM transaction WHERE hash = $1`, txHash) + if err != nil { + return dbtypes.TransactionRow{}, err + } + + return rows[0], nil +} + +// updateMessage stores updated message inside the database +func (db *Migrator) updateMessage(msg *types.Message, partitionID int64) error { + stmt := ` +INSERT INTO message(transaction_hash, index, type, value, involved_accounts_addresses, height, partition_id) +VALUES ($1, $2, $3, $4, $5, $6, $7) +ON CONFLICT (transaction_hash, index, partition_id) DO UPDATE + SET height = excluded.height, + type = excluded.type, + value = excluded.value, + involved_accounts_addresses = excluded.involved_accounts_addresses` + + _, err := db.SQL.Exec(stmt, msg.TxHash, msg.Index, msg.Type, msg.Value, pq.Array(msg.Addresses), msg.Height, partitionID) + return err + +} diff --git a/database/legacy/msgexec/migrator.go b/database/legacy/msgexec/migrator.go new file mode 100644 index 00000000..5ba6f727 --- /dev/null +++ b/database/legacy/msgexec/migrator.go @@ -0,0 +1,21 @@ +package msgexec + +import ( + "github.com/jmoiron/sqlx" + + "github.com/forbole/juno/v5/database" + "github.com/forbole/juno/v5/database/postgresql" +) + +var _ database.Migrator = &Migrator{} + +// Migrator represents the database migrator that should be used to migrate from v2 of the database to v3 +type Migrator struct { + SQL *sqlx.DB +} + +func NewMigrator(db *postgresql.Database) *Migrator { + return &Migrator{ + SQL: db.SQL, + } +} \ No newline at end of file diff --git a/database/migrate/utils/types.go b/database/migrate/utils/types.go index 496fd90b..a65dc431 100644 --- a/database/migrate/utils/types.go +++ b/database/migrate/utils/types.go @@ -13,4 +13,15 @@ type TransactionRow struct { GasUsed string `db:"gas_used"` RawLog string `db:"raw_log"` Logs string `db:"logs"` + PartitionID int64 `db:"partition_id"` +} + +type MessageRow struct { + TransactionHash string `db:"transaction_hash"` + Index int64 `db:"index"` + Type string `db:"type"` + Value string `db:"value"` + InvolvedAccountsAddresses string `db:"involved_accounts_addresses"` + Height int64 `db:"height"` + PartitionID int64 `db:"partition_id"` } diff --git a/modules/messages/types.go b/modules/messages/types.go index 6b743c64..c831488a 100644 --- a/modules/messages/types.go +++ b/modules/messages/types.go @@ -21,7 +21,7 @@ func DefaultMessagesParser(tx *types.Tx) ([]string, error) { } // function to remove duplicate values -func removeDuplicates(s []string) []string { +func RemoveDuplicates(s []string) []string { bucket := make(map[string]bool) var result []string for _, str := range s { @@ -55,5 +55,5 @@ func parseAddressesFromEvents(tx *types.Tx) []string { } - return removeDuplicates(addresses) + return RemoveDuplicates(addresses) }