Updated stake-update and state-sync to use graph

llamanodes · Aug 22, 2023 · 40857fe · 40857fe
1 parent fe5a3e7
commit 40857fe
Show file tree

Hide file tree

Showing 2 changed files with 118 additions and 11 deletions.
diff --git a/bridge/setu/listener/rootchain_events.go b/bridge/setu/listener/rootchain_events.go
@@ -88,6 +88,10 @@ func (rl *RootChainListener) getCurrentStateID(ctx context.Context) (*big.Int, e
 
 // getStateSync returns the StateSynced event based on the given state ID
 func (rl *RootChainListener) getStateSync(ctx context.Context, stateId int64) (*statesender.StatesenderStateSynced, error) {
+	if rl.subGraph != nil {
+		return rl.getStateSyncGraph(ctx, stateId)
+	}
+
 	rootchainContext, err := rl.getRootChainContext()
 	if err != nil {
 		return nil, err
@@ -155,6 +159,10 @@ func (rl *RootChainListener) getLatestNonce(ctx context.Context, validatorId uin
 
 // getStakeUpdate returns StakeUpdate event based on the given validator ID and nonce
 func (rl *RootChainListener) getStakeUpdate(ctx context.Context, validatorId, nonce uint64) (*stakinginfo.StakinginfoStakeUpdate, error) {
+	if rl.subGraph != nil {
+		return rl.getStakeUpdateGraph(ctx, validatorId, nonce)
+	}
+
 	rootchainContext, err := rl.getRootChainContext()
 	if err != nil {
 		return nil, err

diff --git a/bridge/setu/listener/selfheal_graph.go b/bridge/setu/listener/selfheal_graph.go
@@ -9,13 +9,15 @@ import (
 	"math/big"
 	"net/http"
 	"strconv"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/maticnetwork/heimdall/contracts/stakinginfo"
+	"github.com/maticnetwork/heimdall/contracts/statesender"
+	"github.com/maticnetwork/heimdall/helper"
 )
 
 // StakeUpdate represents the StakeUpdate event
 type stakeUpdate struct {
-	ValidatorID     string `json:"validatorId"`
-	TotalStaked     string `json:"totalStaked"`
-	Block           string `json:"block"`
 	Nonce           string `json:"nonce"`
 	TransactionHash string `json:"transactionHash"`
 	LogIndex        string `json:"logIndex"`
@@ -24,11 +26,8 @@ type stakeUpdate struct {
 // StateSync represents the StateSync event
 type stateSync struct {
 	StateID         string `json:"stateId"`
-	Contract        string `json:"contract"`
-	RawData         string `json:"rawData"`
 	LogIndex        string `json:"logIndex"`
 	TransactionHash string `json:"transactionHash"`
-	BlockNumber     string `json:"blockNumber"`
 }
 
 type stakeUpdateResponse struct {
@@ -43,6 +42,21 @@ type stateSyncResponse struct {
 	} `json:"data"`
 }
 
+func (rl *RootChainListener) querySubGraph(query []byte, ctx context.Context) (data []byte, err error) {
+	request, err := http.NewRequestWithContext(ctx, http.MethodPost, rl.subGraph.graphUrl, bytes.NewBuffer(query))
+	if err != nil {
+		return nil, err
+	}
+
+	response, err := rl.subGraph.client.Do(request)
+	if err != nil {
+		return nil, err
+	}
+	defer response.Body.Close()
+
+	return io.ReadAll(response.Body)
+}
+
 func (rl *RootChainListener) getLatestNonceGraph(ctx context.Context, validatorId uint64) (uint64, error) {
 	query := map[string]string{
 		"query": `
@@ -117,17 +131,102 @@ func (rl *RootChainListener) getLatestStateIDGraph(ctx context.Context) (*big.In
 	return stateID, nil
 }
 
-func (rl *RootChainListener) querySubGraph(query []byte, ctx context.Context) (data []byte, err error) {
-	request, err := http.NewRequestWithContext(ctx, http.MethodPost, rl.subGraph.graphUrl, bytes.NewBuffer(query))
+func (rl *RootChainListener) getStakeUpdateGraph(ctx context.Context, validatorId, nonce uint64) (*stakinginfo.StakinginfoStakeUpdate, error) {
+	query := map[string]string{
+		"query": `
+		{
+			stakeUpdates(where: {validatorId: ` + strconv.Itoa(int(validatorId)) + `, nonce: ` + strconv.Itoa(int(nonce)) + `}){
+				transactionHash
+				logIndex
+		   } 
+		}   
+		`,
+	}
+
+	byteQuery, err := json.Marshal(query)
 	if err != nil {
 		return nil, err
 	}
 
-	response, err := rl.subGraph.client.Do(request)
+	data, err := rl.querySubGraph(byteQuery, ctx)
 	if err != nil {
+		return nil, fmt.Errorf("unable to fetch stake update from graph with err: %s", err)
+	}
+
+	var response stakeUpdateResponse
+	if err = json.Unmarshal(data, &response); err != nil {
 		return nil, err
 	}
-	defer response.Body.Close()
 
-	return io.ReadAll(response.Body)
+	if len(response.Data.StakeUpdates) == 0 {
+		return nil, fmt.Errorf("no stake update found for validator %d and nonce %d", validatorId, nonce)
+	}
+
+	receipt, err := rl.contractConnector.MainChainClient.TransactionReceipt(ctx, common.HexToHash(response.Data.StakeUpdates[0].TransactionHash))
+	if err != nil {
+		return nil, err
+	}
+
+	for _, logs := range receipt.Logs {
+		if strconv.Itoa(int(logs.Index)) == response.Data.StakeUpdates[0].LogIndex {
+			var event stakinginfo.StakinginfoStakeUpdate
+			if err = helper.UnpackLog(rl.stakingInfoAbi, &event, "StakeUpdate", logs); err != nil {
+				return nil, err
+			}
+
+			return &event, nil
+		}
+	}
+
+	return nil, fmt.Errorf("no logs found for given log index %s ,validator %d and nonce %d", response.Data.StakeUpdates[0].LogIndex, validatorId, nonce)
+}
+
+func (rl *RootChainListener) getStateSyncGraph(ctx context.Context, stateId int64) (*statesender.StatesenderStateSynced, error) {
+	query := map[string]string{
+		"query": `
+		{
+			stateSyncs(where: {stateId: ` + strconv.Itoa(int(stateId)) + `}) {
+				logIndex
+				transactionHash
+			}
+		}
+		`,
+	}
+
+	byteQuery, err := json.Marshal(query)
+	if err != nil {
+		return nil, err
+	}
+
+	data, err := rl.querySubGraph(byteQuery, ctx)
+	if err != nil {
+		return nil, fmt.Errorf("unable to fetch latest state id from graph with err: %s", err)
+	}
+
+	var response stateSyncResponse
+	if err = json.Unmarshal(data, &response); err != nil {
+		return nil, fmt.Errorf("unable to unmarshal graph response: %s", err)
+	}
+
+	if len(response.Data.StateSyncs) == 0 {
+		return nil, fmt.Errorf("no state sync found for state id %d", stateId)
+	}
+
+	receipt, err := rl.contractConnector.MainChainClient.TransactionReceipt(ctx, common.HexToHash(response.Data.StateSyncs[0].TransactionHash))
+	if err != nil {
+		return nil, err
+	}
+
+	for _, logs := range receipt.Logs {
+		if strconv.Itoa(int(logs.Index)) == response.Data.StateSyncs[0].LogIndex {
+			var event statesender.StatesenderStateSynced
+			if err = helper.UnpackLog(rl.stateSenderAbi, &event, "StateSynced", logs); err != nil {
+				return nil, err
+			}
+
+			return &event, nil
+		}
+	}
+
+	return nil, fmt.Errorf("no logs found for given log index %s and state id %d", response.Data.StateSyncs[0].LogIndex, stateId)
 }