🐛 Fix nft_income migration errors (#96)

* 🐛 Fix infinity loop error in migration

* 🐛 Prevent variables to be reset too early

* 🔊 Add log when `count == 0`

* 🐛 Query events from txs table and filter by correct action types

* 💡 Add comments

db/schema/parallel/nft_income.go

@@ -8,6 +8,7 @@ import (
 	"github.com/jackc/pgx/v4/pgxpool"
 	"github.com/likecoin/likecoin-chain-tx-indexer/db"
 	"github.com/likecoin/likecoin-chain-tx-indexer/logger"
+	"github.com/likecoin/likecoin-chain-tx-indexer/rest"
 	"github.com/likecoin/likecoin-chain-tx-indexer/utils"
 )
 
@@ -21,10 +22,21 @@ func MigrateNftIncome(conn *pgxpool.Conn, batchSize uint64) error {
 		return err
 	}
 	return conn.BeginFunc(context.Background(), func(dbTx pgx.Tx) error {
+		// use `WHERE rn = 1` to ensure only the latest event with same tx_hash is processed,
+		// that way we can skip most unused `mint_nft` actions mixed in the `/cosmos.nft.v1beta1.MsgSend` actions
 		_, err := dbTx.Exec(context.Background(), `
 			DECLARE nft_income_migration_cursor CURSOR FOR
 				SELECT id, class_id, nft_id, tx_hash, events
-				FROM nft_event
+				FROM (
+					SELECT e.id, e.class_id, e.nft_id, e.tx_hash, t.events, ROW_NUMBER() OVER (PARTITION BY e.tx_hash ORDER BY e.id DESC) AS rn
+					FROM nft_event AS e
+					JOIN (
+						SELECT DISTINCT ON (tx ->> 'txhash') tx ->> 'txhash' AS tx_hash, events
+						FROM txs
+					) AS t ON e.tx_hash = t.tx_hash
+					WHERE e.action IN ('/cosmos.nft.v1beta1.MsgSend', 'buy_nft', 'sell_nft')
+				) subquery
+				WHERE rn = 1
 				ORDER BY id
 			;
 		`)
@@ -57,15 +69,20 @@ func MigrateNftIncome(conn *pgxpool.Conn, batchSize uint64) error {
 					return err
 				}
 
-				events, err := utils.ParseEvents(eventRaw)
+				// `spreadEvents` includes events of 'ALL' messages in the tx, not just a message.
+				// this is different from the case in `extractNftIncomes()`
+				spreadEvents, err := utils.ParseEvents(eventRaw)
 				if err != nil {
 					logger.L.Errorw("Error when parsing events", "error", err)
 					return err
 				}
 
-				msgAction := utils.GetEventsValue(events, "message", "action")
-				if msgAction == "/cosmos.bank.v1beta1.MsgSend" || msgAction == string(db.ACTION_BUY) || msgAction == string(db.ACTION_SELL) {
-					incomeMap := utils.GetIncomeMap(events)
+				firstMsgAction := utils.GetEventsValue(spreadEvents, "message", "action")
+				if firstMsgAction == "/cosmos.authz.v1beta1.MsgExec" || firstMsgAction == string(db.ACTION_BUY) || firstMsgAction == string(db.ACTION_SELL) {
+					incomeMap := utils.GetIncomeMap(spreadEvents)
+					for _, address := range rest.DefaultApiAddresses {
+						delete(incomeMap, address)
+					}
 					for address, amount := range incomeMap {
 						incomes = append(incomes, db.NftIncome{
 							ClassId: classId,
@@ -77,8 +94,16 @@ func MigrateNftIncome(conn *pgxpool.Conn, batchSize uint64) error {
 					}
 				}
 			}
+			if pkeyId == 0 {
+				break
+			}
 			count := len(incomes)
 			if count == 0 {
+				logger.L.Infow(
+					"NFT income table migration progress",
+					"pkey_id", pkeyId,
+					"count", count,
+				)
 				continue
 			}
 			for i := 0; i < count; i++ {
@@ -103,6 +128,7 @@ func MigrateNftIncome(conn *pgxpool.Conn, batchSize uint64) error {
 				"amount", lastIncome.Amount,
 			)
 		}
+		logger.L.Infow("NFT income table migration completed")
 		return nil
 	})
 }

utils/events.go

@@ -78,10 +78,10 @@ func ParseCoinFromEventString(coinStr string) (uint64, error) {
 
 func GetIncomeMap(events types.StringEvents) map[string]uint64 {
 	incomeMap := make(map[string]uint64)
+	address := ""
+	amount := uint64(0)
 	for _, event := range events {
 		if event.Type == "coin_received" {
-			address := ""
-			amount := uint64(0)
 			for _, attr := range event.Attributes {
 				if attr.Key == "receiver" {
 					address = attr.Value