WarpySyncer: Pendle integration

src/utils/config/warpy_syncer.go

@@ -89,20 +89,26 @@ type WarpySyncer struct {
 	// Max number of transactions that wait in the worker queue
 	SyncerDepositWorkerQueueSize int
 
+	// Accepted markets in which token is being deposited
+	SyncerDepositMarkets []string
+
+	// Supported token
+	SyncerDepositToken string
+
 	// Max batch size before last block synced will be inserted into database
 	StoreBatchSize int
 
 	// After this time last block synced will be inserted into database
 	StoreInterval time.Duration
 
-	// Sommelier functions for withdrawal
+	// Functions for withdrawal
 	StoreDepositWithdrawFunctions []string
 
-	// Name of the deposit assets input name
-	StoreDepositDepositAssetsName string
+	// Names of the deposit assets input name
+	StoreDepositDepositAssetsNames []string
 
-	// Name of the withdraw assets input name
-	StoreDepositWithdrawAssetsName string
+	// Names of the withdraw assets input name
+	StoreDepositWithdrawAssetsNames []string
 
 	// Max time between failed retries to save last block synced
 	StoreMaxBackoffInterval time.Duration
@@ -143,8 +149,8 @@ func setWarpySyncerDefaults() {
 	viper.SetDefault("WarpySyncer.BlockDownloaderByHeader", true)
 	viper.SetDefault("WarpySyncer.SyncerContractId", "p5OI99-BaY4QbZts266T7EDwofZqs-wVuYJmMCS0SUU")
 	viper.SetDefault("WarpySyncer.SyncerNameServiceContractId", "p5OI99-BaY4QbZts266T7EDwofZqs-wVuYJmMCS0SUU")
-	viper.SetDefault("WarpySyncer.SyncerChain", eth.Manta)
-	viper.SetDefault("WarpySyncer.SyncerProtocol", eth.LayerBank)
+	viper.SetDefault("WarpySyncer.SyncerChain", eth.Arbitrum)
+	viper.SetDefault("WarpySyncer.SyncerProtocol", eth.Pendle)
 	viper.SetDefault("WarpySyncer.SyncerDreUrl", "https://dre-warpy.warp.cc")
 	viper.SetDefault("WarpySyncer.SyncerWarpyApiUrl", "https://api-warpy.warp.cc")
 	viper.SetDefault("WarpySyncer.SyncerApiKey", "")
@@ -157,15 +163,24 @@ func setWarpySyncerDefaults() {
 	viper.SetDefault("WarpySyncer.SyncerDeltaRedstoneData", "000002ed57011e0000")
 	viper.SetDefault("WarpySyncer.SyncerDeltaNumWorkers", "50")
 	viper.SetDefault("WarpySyncer.SyncerDeltaWorkerQueueSize", "10")
-	viper.SetDefault("WarpySyncer.SyncerDepositContractId", "0xB7A23Fc0b066051dE58B922dC1a08f33DF748bbf")
+	viper.SetDefault("WarpySyncer.SyncerDepositContractId", "0x888888888889758f76e7103c6cbf23abbf58f946")
 	viper.SetDefault("WarpySyncer.SyncerDepositBackoffInterval", "3s")
-	viper.SetDefault("WarpySyncer.SyncerDepositFunctions", []string{"supply", "redeemToken"})
-	viper.SetDefault("WarpySyncer.StoreDepositWithdrawFunctions", []string{"redeemToken"})
+	viper.SetDefault("WarpySyncer.SyncerDepositFunctions", []string{"swapExactTokenForPt", "swapExactPtForToken"})
+	viper.SetDefault("WarpySyncer.SyncerDepositMarkets", []string{
+		// wETH
+		"0x952083cde7aaa11AB8449057F7de23A970AA8472",
+		// rsETH
+		"0x6Ae79089b2CF4be441480801bb741A531d94312b",
+		// ezETH
+		"0x5E03C94Fc5Fb2E21882000A96Df0b63d2c4312e2",
+	})
+	viper.SetDefault("WarpySyncer.SyncerDepositToken", "0x6A0d9584D88D22BcaD7D4F83E7d6AB7949895DDF")
+	viper.SetDefault("WarpySyncer.StoreDepositWithdrawFunctions", []string{"swapExactPtForToken"})
 	viper.SetDefault("WarpySyncer.StoreBatchSize", "500")
 	viper.SetDefault("WarpySyncer.StoreInterval", "2s")
 	viper.SetDefault("WarpySyncer.StoreMaxBackoffInterval", "30s")
-	viper.SetDefault("WarpySyncer.StoreDepositDepositAssetsName", "lAmount")
-	viper.SetDefault("WarpySyncer.StoreDepositWithdrawAssetsName", "uAmount")
+	viper.SetDefault("WarpySyncer.StoreDepositDepositAssetsNames", []string{"input", "netTokenIn"})
+	viper.SetDefault("WarpySyncer.StoreDepositWithdrawAssetsNames", []string{"exactPtIn"})
 	viper.SetDefault("WarpySyncer.PollerDepositChannelBufferLength", 100)
 	viper.SetDefault("WarpySyncer.PollerDepositInterval", "1m")
 	viper.SetDefault("WarpySyncer.PollerDepositTimeout", "90s")

src/utils/eth/client.go

@@ -1,9 +1,12 @@
 package eth
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"math/big"
+	"os"
 	"strings"
 
 	"github.com/ethereum/go-ethereum/accounts/abi"
@@ -28,6 +31,7 @@ const (
 	Delta     Protocol = iota
 	Sommelier Protocol = iota
 	LayerBank Protocol = iota
+	Pendle    Protocol = iota
 )
 
 type Chain int
@@ -84,6 +88,8 @@ func (protocol Protocol) String() string {
 		return "sommelier"
 	case LayerBank:
 		return "layer_bank"
+	case Pendle:
+		return "pendle"
 	}
 	return ""
 }
@@ -189,3 +195,31 @@ func WeiToEther(wei *big.Int) float64 {
 	ether, _ := new(big.Float).Quo(new(big.Float).SetInt(wei), big.NewFloat(params.Ether)).Float64()
 	return ether
 }
+
+func GetContractABIFromFile(fileName string) (*abi.ABI, error) {
+	fileData, err := os.Open(fmt.Sprintf("src/warpy_sync/abi/%s", fileName))
+
+	if err != nil {
+		return nil, err
+	}
+
+	byteValue, err := io.ReadAll(fileData)
+	if err != nil {
+		return nil, err
+	}
+
+	rawABIResponse := &RawABIResponse{}
+
+	err = json.Unmarshal(byteValue, rawABIResponse)
+	if err != nil {
+		return nil, err
+	}
+
+	fileData.Close()
+
+	contractABI, err := abi.JSON(strings.NewReader(*rawABIResponse.Result))
+	if err != nil {
+		return nil, err
+	}
+	return &contractABI, nil
+}

src/warpy_sync/controller.go

@@ -3,6 +3,7 @@ package warpy_sync
 import (
 	"errors"
 
+	"github.com/ethereum/go-ethereum/accounts/abi"
 	"github.com/warp-contracts/syncer/src/utils/config"
 	"github.com/warp-contracts/syncer/src/utils/eth"
 	"github.com/warp-contracts/syncer/src/utils/model"
@@ -90,15 +91,24 @@ func NewController(config *config.Config) (self *Controller, err error) {
 		writerTask = writer.Task
 		syncerTask = syncer.Task
 		syncerOutput = syncer.Output
-	case eth.Sommelier, eth.LayerBank:
-		contractAbi, errAbi := eth.GetContractABI(
-			config.WarpySyncer.SyncerDepositContractId,
-			config.WarpySyncer.SyncerApiKey,
-			config.WarpySyncer.SyncerChain)
-
-		err = errAbi
-
-		// Checks wether block's transactions contain specific Sommelier transactions
+	case eth.Sommelier, eth.LayerBank, eth.Pendle:
+		var contractAbi *abi.ABI
+		switch config.WarpySyncer.SyncerProtocol {
+
+		case eth.Sommelier, eth.LayerBank:
+			contractAbi, err = eth.GetContractABI(
+				config.WarpySyncer.SyncerDepositContractId,
+				config.WarpySyncer.SyncerApiKey,
+				config.WarpySyncer.SyncerChain)
+		case eth.Pendle:
+			contractAbi, err = eth.GetContractABIFromFile("IPActionSwapPTV3.json")
+
+		default:
+			self.Log.WithError(err).Error("ETH Protocol not recognized")
+			return
+		}
+
+		// Checks wether block's transactions contain specific transactions
 		syncer := NewSyncerDeposit(config).
 			WithMonitor(monitor).
 			WithInputChannel(blockDownloader.Output).

src/warpy_sync/poller_deposit.go

@@ -66,10 +66,11 @@ func (self *PollerDeposit) handleNew() (err error) {
 			Raw(`SELECT from_address, 
 				SUM(assets) 
 				FROM warpy_syncer_assets 
-				WHERE timestamp < ? AND chain = ? AND protocol = ? 
+				WHERE timestamp < ? AND chain = ? AND protocol = ?
+				AND from_address = ? 
 				group by from_address;
 		`, time.Now().Unix()-self.Config.WarpySyncer.PollerDepositSecondsForSelect,
-				self.Config.WarpySyncer.SyncerChain, self.Config.WarpySyncer.SyncerProtocol).
+				self.Config.WarpySyncer.SyncerChain, self.Config.WarpySyncer.SyncerProtocol, "0x64937ab314bc1999396De341Aa66897C30008852").
 			Scan(&AssetsSums).Error
 
 		if err != nil {

src/warpy_sync/store_deposit.go

@@ -2,6 +2,7 @@ package warpy_sync
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"math"
 	"math/big"
@@ -77,13 +78,20 @@ func (self *StoreDeposit) run() (err error) {
 					Transaction(func(dbTx *gorm.DB) error {
 						err = self.insertLog(dbTx, payload.Transaction, payload.FromAddress, payload.Block, payload.Method, payload.ParsedInput)
 
-						var ethTxAssetsFieldName string
+						var assets interface{}
 						if slices.Contains(self.Config.WarpySyncer.StoreDepositWithdrawFunctions, payload.Method.Name) {
-							ethTxAssetsFieldName = self.Config.WarpySyncer.StoreDepositDepositAssetsName
+							assets = self.getAssetsFromInput(self.Config.WarpySyncer.StoreDepositWithdrawAssetsNames, payload.Input)
 						} else {
-							ethTxAssetsFieldName = self.Config.WarpySyncer.StoreDepositWithdrawAssetsName
+							assets = self.getAssetsFromInput(self.Config.WarpySyncer.StoreDepositDepositAssetsNames, payload.Input)
 						}
-						err = self.insertAssets(dbTx, payload.Transaction, payload.FromAddress, eth.WeiToEther(payload.Input[ethTxAssetsFieldName].(*big.Int)), payload.Method.Name, payload.Block)
+
+						assetsVal := self.convertAssets(assets)
+
+						if assetsVal == nil {
+							return nil
+						}
+
+						err = self.insertAssets(dbTx, payload.Transaction, payload.FromAddress, eth.WeiToEther(assetsVal), payload.Method.Name, payload.Block)
 						if err != nil {
 							return err
 						}
@@ -237,3 +245,37 @@ func (self *StoreDeposit) insertAssets(dbTx *gorm.DB, tx *types.Transaction, fro
 
 	return
 }
+
+func (self *StoreDeposit) getAssetsFromInput(assetsNames []string, input map[string]interface{}) (assets interface{}) {
+	for i, a := range assetsNames {
+		if i == 0 {
+			assets = input[a]
+		} else {
+			var assetsInterface map[string]interface{}
+			assetsParsed, _ := json.Marshal(assets)
+			err := json.Unmarshal(assetsParsed, &assetsInterface)
+			if err != nil {
+				self.Log.WithError(err).Error("Could not parse assets input")
+				return nil
+			}
+			assets = assetsInterface[a]
+		}
+	}
+
+	return
+}
+
+func (self *StoreDeposit) convertAssets(assets interface{}) (assetsVal *big.Int) {
+	switch assets := assets.(type) {
+	case float64:
+		assetsVal = new(big.Int)
+		new(big.Float).SetFloat64(assets).Int(assetsVal)
+	case *big.Int:
+		assetsVal = assets
+	default:
+		self.Log.WithField("assets", assets).Error("Assets type unsupported")
+		return nil
+	}
+
+	return
+}

src/warpy_sync/syncer_deposit.go

@@ -6,10 +6,12 @@ import (
 	"errors"
 	"fmt"
 	"slices"
+	"strings"
 	"sync"
 
 	"github.com/cenkalti/backoff"
 	"github.com/ethereum/go-ethereum/accounts/abi"
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/warp-contracts/syncer/src/utils/config"
 	"github.com/warp-contracts/syncer/src/utils/eth"
@@ -125,7 +127,7 @@ func (self *SyncerDeposit) checkTx(tx *types.Transaction, block *BlockInfoPayloa
 			return err
 		}).
 		Run(func() error {
-			if tx.To() != nil && tx.To().String() == self.Config.WarpySyncer.SyncerDepositContractId {
+			if tx.To() != nil && strings.EqualFold(tx.To().String(), self.Config.WarpySyncer.SyncerDepositContractId) {
 				self.Log.WithField("tx_id", tx.Hash()).Info("Found new on-chain transaction")
 				method, inputsMap, err := eth.DecodeTransactionInputData(self.contractAbi, tx.Data())
 				if err != nil {
@@ -135,31 +137,26 @@ func (self *SyncerDeposit) checkTx(tx *types.Transaction, block *BlockInfoPayloa
 
 				if slices.Contains(self.Config.WarpySyncer.SyncerDepositFunctions, method.Name) {
 					parsedInputsMap, err := json.Marshal(inputsMap)
+					if err != nil {
+						self.Log.WithError(err).Error("Could not parse transaction input")
+						return err
+					}
 
-					tokenName := inputsMap["lToken"]
-					tokenNameStr := fmt.Sprintf("%v", tokenName)
-					self.Log.WithField("token_name", tokenNameStr).Info("Token set for transfer")
-
-					// TODO: do it properly (i.e. via params, not hardcoded)
-					if self.Config.WarpySyncer.SyncerChain == eth.Mode {
-						if tokenNameStr != "0x6A0d9584D88D22BcaD7D4F83E7d6AB7949895DDF" {
-							self.Log.WithField("token_name", tokenNameStr).Warn("Wrong token set for transfer")
+					if (self.Config.WarpySyncer.SyncerChain == eth.Mode) || (self.Config.WarpySyncer.SyncerChain == eth.Manta) {
+						contains := self.containsSpecificToken(inputsMap)
+						if !contains {
 							return nil
 						}
 					}
 
-					if self.Config.WarpySyncer.SyncerChain == eth.Manta {
-						if tokenNameStr != "0x71384B2c17433Ba1D8F6Fe895E9B2E7953dCED68" {
-							self.Log.WithField("token_id", tokenNameStr).Warn("Wrong token set for transfer")
+					// need to check if deposit is being sent to one of the specific pools
+					if self.Config.WarpySyncer.SyncerProtocol == eth.Pendle {
+						belongs := self.belongsToMarket(inputsMap, tx.Hash())
+						if !belongs {
 							return nil
 						}
 					}
 
-					if err != nil {
-						self.Log.WithError(err).Error("Could not parse transaction input")
-						return err
-					}
-
 					self.Log.WithField("method_name", method.Name).WithField("inputs_map", string(parsedInputsMap)).
 						Info("New transaction decoded")
 
@@ -189,3 +186,39 @@ func (self *SyncerDeposit) checkTx(tx *types.Transaction, block *BlockInfoPayloa
 
 	return
 }
+
+func (self *SyncerDeposit) containsSpecificToken(inputsMap map[string]interface{}) bool {
+	tokenName := inputsMap["lToken"]
+	tokenNameStr := fmt.Sprintf("%v", tokenName)
+	self.Log.WithField("token_name", tokenNameStr).Info("Token set for transfer")
+
+	// TODO: do it properly (i.e. via params, not hardcoded)
+	if self.Config.WarpySyncer.SyncerChain == eth.Mode {
+		if tokenNameStr != self.Config.WarpySyncer.SyncerDepositToken {
+			self.Log.WithField("token_name", tokenNameStr).Warn("Wrong token set for transfer")
+			return false
+		}
+	}
+
+	if self.Config.WarpySyncer.SyncerChain == eth.Manta {
+		if tokenNameStr != self.Config.WarpySyncer.SyncerDepositToken {
+			self.Log.WithField("token_id", tokenNameStr).Warn("Wrong token set for transfer")
+			return false
+		}
+	}
+
+	return true
+}
+
+func (self *SyncerDeposit) belongsToMarket(inputsMap map[string]interface{}, txHash common.Hash) bool {
+	market := inputsMap["market"]
+	marketStr := fmt.Sprintf("%v", market)
+
+	if !(slices.Contains(self.Config.WarpySyncer.SyncerDepositMarkets, marketStr)) {
+		self.Log.WithField("tx_id", txHash).WithField("market", marketStr).Warn("Token deposit does not belong to desired market")
+		return false
+	}
+
+	self.Log.WithField("tx_id", txHash).WithField("market", marketStr).Debug("Token deposit belongs to desired market")
+	return true
+}