feat: query events from breakpoint

dao/mysql/collect.go

@@ -34,13 +34,23 @@ func InsertCollectPaidMwSetEvent(param *model.ParamCollectPaidMwSetEvent) error
 }
 
 func GetCollectPaidMwSetEventParams(chainID uint64, profileID string, essenceID string) (*model.ParamCollectPaidMwSetEvent, error) {
-	collectInfo := new(model.ParamCollectPaidMwSetEvent)
+	collectInfoList := new(model.ParamCollectPaidMwSetEvent)
 	sqlStr := "SELECT chain_name, chain_id, block_number, tx_hash, namespace, profile_id, essence_id, total_supply, amount, recipient, currency, subscribe_required FROM collect_paid_mw_set_events WHERE chain_id = ? AND profile_id = ? AND essence_id = ?"
-	err := db.Get(collectInfo, sqlStr, chainID, profileID, essenceID)
+	err := db.Get(collectInfoList, sqlStr, chainID, profileID, essenceID)
 	if(err != nil) {
 		if err == sql.ErrNoRows {
 			return nil, nil
 		}
 	}
-	return collectInfo, nil
+	return collectInfoList, nil
+}
+
+func GetLatestTrackedCollectPaidMwSetBlockNumber(chainID uint64) (uint64, error) {
+	collectInfoList := make([]*model.ParamCollectPaidMwSetEvent, 0, 2)
+	sqlStr := "SELECT block_number FROM collect_paid_mw_set_events WHERE chain_id = ? ORDER BY id DESC LIMIT 1"
+	err := db.Select(&collectInfoList, sqlStr, chainID)
+	if(err != nil) {
+		return 0 ,err
+	}
+	return collectInfoList[0].BlockNumber, nil
 }

dao/mysql/profile.go

@@ -22,12 +22,22 @@ func InsertCreateProfileEvent(param *model.ParamCreateProfileEvent) (err error)
 
 func GetCreateProfileEventParams(chainID uint64, account string) ([]*model.ParamCreateProfileEvent, error) {
 	profileInfoList := make([]*model.ParamCreateProfileEvent, 0, 2)
-	sqlStr := "SELECT chain_name, chain_id, block_number, tx_hash, `to`, profile_id, handle, avatar, metadata FROM create_profile_events WHERE chain_id = ? AND `to` = ? order by block_number desc"
+	sqlStr := "SELECT chain_name, chain_id, block_number, tx_hash, `to`, profile_id, handle, avatar, metadata FROM create_profile_events WHERE chain_id = ? AND `to` = ? ORDER BY block_number DESC"
 	err := db.Select(&profileInfoList, sqlStr, chainID, account)
 	if(err != nil) {
 		if err == sql.ErrNoRows {
 			return []*model.ParamCreateProfileEvent{}, nil
 		}
 	}
 	return profileInfoList, nil
+}
+
+func GetLatestTrackedCreateProfileBlockNumber(chainID uint64) (uint64, error) {
+	profileInfoList := make([]*model.ParamCreateProfileEvent, 0, 2)
+	sqlStr := "SELECT block_number FROM create_profile_events WHERE chain_id = ? ORDER BY id DESC LIMIT 1"
+	err := db.Select(&profileInfoList, sqlStr, chainID)
+	if(err != nil) {
+		return 0 ,err
+	}
+	return profileInfoList[0].BlockNumber, nil
 }

go.mod

@@ -4,6 +4,7 @@ go 1.21.0
 
 require (
 	github.com/ethereum/go-ethereum v1.13.4
+	github.com/gin-gonic/gin v1.9.1
 	github.com/go-sql-driver/mysql v1.7.1
 )
 
@@ -13,7 +14,6 @@ require (
 	github.com/chenzhuoyu/iasm v0.9.0 // indirect
 	github.com/gabriel-vasile/mimetype v1.4.3 // indirect
 	github.com/gin-contrib/sse v0.1.0 // indirect
-	github.com/gin-gonic/gin v1.9.1 // indirect
 	github.com/go-playground/locales v0.14.1 // indirect
 	github.com/go-playground/universal-translator v0.18.1 // indirect
 	github.com/go-playground/validator/v10 v10.15.5 // indirect

listener/collect.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"cyber-events-tracker/logic"
 	"cyber-events-tracker/utils"
+	"database/sql"
 	"log"
 	"math/big"
 	"time"
@@ -75,10 +76,22 @@ func QueryCollectPaidMwSetEvents(chainID uint64, contractAddress common.Address,
 
 		currentBlockNumber := big.NewInt(int64(_currentBlockNumber))
 
-		log.Printf("[%d]: Start query CollectPaidMwSet events...", chainID)
-
-		var _startAt *big.Int = startAt
+		var _startAt *big.Int
 		var _endAt *big.Int = big.NewInt(0)
+		previousAt, err := logic.GetPreviousTrackedCollectPaidMwSetBlockNumber(chainID)
+		if err != nil {
+			if err == sql.ErrNoRows {
+				_startAt = startAt
+				log.Printf("[%d]: Start query CollectPaidMwSet events at [%d]...", chainID, _startAt.Uint64())
+			} else {
+				log.Fatalf("[%d]: GetPreviousTrackedCollectPaidMwSetBlockNumber failed, %v", chainID, err)
+				return
+			}
+		} else {
+			_startAt = big.NewInt(int64(previousAt))
+			log.Printf("[%d]: Continue query CollectPaidMwSet events at [%d]...", chainID, _startAt.Uint64())
+		}
+
 		for {
 			_endAt.Add(_startAt, big.NewInt(49999))
 			if _endAt.Cmp(currentBlockNumber) > 0 {

listener/profile.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"cyber-events-tracker/logic"
 	"cyber-events-tracker/utils"
+	"database/sql"
 	"log"
 	"math/big"
 	"time"
@@ -75,10 +76,22 @@ func QueryCreateProfileEvents(chainID uint64, contractAddress common.Address, st
 
 		currentBlockNumber := big.NewInt(int64(_currentBlockNumber))
 
-		log.Printf("[%d]: Start query CreateProfile events...", chainID)
-
-		var _startAt *big.Int = big.NewInt(25455191)
+		var _startAt *big.Int
 		var _endAt *big.Int = big.NewInt(0)
+		previousAt, err := logic.GetPreviousTrackedCreateProfileBlockNumber(chainID)
+		if err != nil {
+			if err == sql.ErrNoRows {
+				_startAt = startAt
+				log.Printf("[%d]: Start query CreateProfile events at [%d]...", chainID, _startAt.Uint64())
+			} else {
+				log.Fatalf("[%d]: GetPreviousTrackedCreateProfileBlockNumber failed, %v", chainID, err)
+				return
+			}
+		} else {
+			_startAt = big.NewInt(int64(previousAt))
+			log.Printf("[%d]: Continue query CreateProfile events at [%d]...", chainID, _startAt.Uint64())
+		}
+
 		for {
 			_endAt.Add(_startAt, big.NewInt(49999))
 			if _endAt.Cmp(currentBlockNumber) > 0 {
@@ -97,10 +110,10 @@ func QueryCreateProfileEvents(chainID uint64, contractAddress common.Address, st
 				log.Fatalf("[%d]: FilterLogs CreateProfile failed, %v", chainID, err)
 				return
 			}
-		
+
 			for _, historyLog := range historyLogs {
 				err = logic.SetProfilesInfo(chainID, historyLog)
-				if(err != nil) {
+				if err != nil {
 					log.Fatalf("[%d]: SetProfileInfo failed, %v", chainID, err)
 				}
 			}

logic/collect.go

@@ -50,3 +50,7 @@ func SetCollectInfo(chainID uint64, vLog types.Log) (err error) {
 func GetCollectInfo(chainID uint64, profileID string, essenceID string) (*model.ParamCollectPaidMwSetEvent, error) {
 	return mysql.GetCollectPaidMwSetEventParams(chainID, profileID, essenceID)
 }
+
+func GetPreviousTrackedCollectPaidMwSetBlockNumber(chainID uint64) (uint64, error) {
+	return mysql.GetLatestTrackedCollectPaidMwSetBlockNumber(chainID)
+}

logic/profile.go

@@ -45,3 +45,7 @@ func SetProfilesInfo(chainID uint64, vLog types.Log) (err error) {
 func GetProfilesInfo(chainID uint64, account string) ([]*model.ParamCreateProfileEvent, error) {
 	return mysql.GetCreateProfileEventParams(chainID, account)
 }
+
+func GetPreviousTrackedCreateProfileBlockNumber(chainID uint64) (uint64, error) {
+	return mysql.GetLatestTrackedCreateProfileBlockNumber(chainID)
+}

utils/chain.go

@@ -27,9 +27,9 @@ func GetChainName(chainID uint64) (string, error) {
 func GetChainRPC(chainID uint64) (string) {
 	if(chainID == 97) {
 		return os.Getenv("BSCT_RPC_URL")
-	}
-	if(chainID == 56) {
+	} else if(chainID == 56) {
 		return os.Getenv("BSC_RPC_URL")
+	} else {
+		return ""
 	}
-	return ""
 }

utils/ethclient.go

@@ -1,7 +1,8 @@
 package utils
 
-import "github.com/ethereum/go-ethereum/ethclient"
-
+import (
+	"github.com/ethereum/go-ethereum/ethclient"
+)
 
 func GetEthClient(rpcUrl string) (*ethclient.Client, error) {
 	if ethClient, err := ethclient.Dial(rpcUrl); err != nil {