feat: multiple thegraph api keys (#669)

KyberNetwork · Jan 8, 2025 · 0cca88c · 0cca88c
1 parent 62083d3
commit 0cca88c
Show file tree

Hide file tree

Showing 70 changed files with 635 additions and 479 deletions.
diff --git a/pkg/liquidity-source/algebra/integral/constant.go b/pkg/liquidity-source/algebra/integral/constant.go
@@ -1,20 +1,17 @@
 package integral
 
 import (
-	"time"
-
 	"github.com/holiman/uint256"
 )
 
 const (
-	DexType               = "algebra-integral"
-	graphSkipLimit        = 5000
-	graphFirstLimit       = 1000
-	defaultTokenDecimals  = 18
-	defaultTokenWeight    = 50
-	zeroString            = "0"
-	emptyString           = ""
-	graphQLRequestTimeout = 20 * time.Second
+	DexType              = "algebra-integral"
+	graphSkipLimit       = 5000
+	graphFirstLimit      = 1000
+	defaultTokenDecimals = 18
+	defaultTokenWeight   = 50
+	zeroString           = "0"
+	emptyString          = ""
 
 	timepointPageSize = uint16(300)
 	maxSwapLoop       = 1000000

diff --git a/pkg/liquidity-source/algebra/integral/pool_list_updater.go b/pkg/liquidity-source/algebra/integral/pool_list_updater.go
@@ -10,26 +10,20 @@ import (
 	"github.com/KyberNetwork/kutils"
 	"github.com/KyberNetwork/logger"
 	"github.com/goccy/go-json"
-	"github.com/machinebox/graphql"
 
 	"github.com/KyberNetwork/kyberswap-dex-lib/pkg/entity"
 	graphqlpkg "github.com/KyberNetwork/kyberswap-dex-lib/pkg/util/graphql"
 )
 
 type PoolsListUpdater struct {
 	config        *Config
-	graphqlClient *graphql.Client
+	graphqlClient *graphqlpkg.Client
 }
 
 func NewPoolsListUpdater(
 	cfg *Config,
+	graphqlClient *graphqlpkg.Client,
 ) *PoolsListUpdater {
-	graphqlClient := graphqlpkg.New(graphqlpkg.Config{
-		Url:     cfg.SubgraphAPI,
-		Header:  cfg.SubgraphHeaders,
-		Timeout: graphQLRequestTimeout,
-	})
-
 	return &PoolsListUpdater{
 		config:        cfg,
 		graphqlClient: graphqlClient,
@@ -39,7 +33,7 @@ func NewPoolsListUpdater(
 func (d *PoolsListUpdater) getPoolsList(ctx context.Context, lastCreatedAtTimestamp *big.Int, lastPoolIds []string, first, skip int) ([]SubgraphPool, error) {
 	allowSubgraphError := d.config.AllowSubgraphError
 
-	req := graphql.NewRequest(getPoolsListQuery(allowSubgraphError, lastCreatedAtTimestamp, lastPoolIds, first, skip))
+	req := graphqlpkg.NewRequest(getPoolsListQuery(allowSubgraphError, lastCreatedAtTimestamp, lastPoolIds, first, skip))
 
 	var response struct {
 		Pools []SubgraphPool `json:"pools"`

diff --git a/pkg/liquidity-source/algebra/integral/pool_tracker.go b/pkg/liquidity-source/algebra/integral/pool_tracker.go
@@ -11,7 +11,6 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/goccy/go-json"
 	"github.com/holiman/uint256"
-	"github.com/machinebox/graphql"
 	"github.com/sourcegraph/conc/pool"
 
 	"github.com/KyberNetwork/kyberswap-dex-lib/pkg/entity"
@@ -24,19 +23,14 @@ import (
 type PoolTracker struct {
 	config        *Config
 	ethrpcClient  *ethrpc.Client
-	graphqlClient *graphql.Client
+	graphqlClient *graphqlpkg.Client
 }
 
 func NewPoolTracker(
 	cfg *Config,
 	ethrpcClient *ethrpc.Client,
+	graphqlClient *graphqlpkg.Client,
 ) (*PoolTracker, error) {
-	graphqlClient := graphqlpkg.New(graphqlpkg.Config{
-		Url:     cfg.SubgraphAPI,
-		Header:  cfg.SubgraphHeaders,
-		Timeout: graphQLRequestTimeout,
-	})
-
 	return &PoolTracker{
 		config:        cfg,
 		ethrpcClient:  ethrpcClient,
@@ -537,7 +531,7 @@ func (d *PoolTracker) getPoolTimepoints(ctx context.Context, blockNumber *big.In
 		end = begin
 		begin = end - timepointPageSize
 		if begin <= currentIndex && currentIndex < end {
-			//we've wrapped around full circle, so break here
+			// we've wrapped around full circle, so break here
 			break
 		}
 	}
@@ -571,7 +565,7 @@ func (d *PoolTracker) getPoolTicks(ctx context.Context, poolAddress string) ([]T
 	var ticks []TickResp
 
 	for {
-		req := graphql.NewRequest(getPoolTicksQuery(allowSubgraphError, poolAddress, skip))
+		req := graphqlpkg.NewRequest(getPoolTicksQuery(allowSubgraphError, poolAddress, skip))
 
 		var resp struct {
 			Pool *SubgraphPoolTicks        `json:"pool"`

diff --git a/pkg/liquidity-source/algebra/v1/constant.go b/pkg/liquidity-source/algebra/v1/constant.go
@@ -2,21 +2,19 @@ package algebrav1
 
 import (
 	"math/big"
-	"time"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/holiman/uint256"
 )
 
 const (
-	DexTypeAlgebraV1      = "algebra-v1"
-	graphSkipLimit        = 5000
-	graphFirstLimit       = 1000
-	defaultTokenDecimals  = 18
-	defaultTokenWeight    = 50
-	zeroString            = "0"
-	emptyString           = ""
-	graphQLRequestTimeout = 20 * time.Second
+	DexTypeAlgebraV1     = "algebra-v1"
+	graphSkipLimit       = 5000
+	graphFirstLimit      = 1000
+	defaultTokenDecimals = 18
+	defaultTokenWeight   = 50
+	zeroString           = "0"
+	emptyString          = ""
 
 	methodGetLiquidity           = "liquidity"
 	methodGetGlobalState         = "globalState"

diff --git a/pkg/liquidity-source/algebra/v1/pool_list_updater.go b/pkg/liquidity-source/algebra/v1/pool_list_updater.go
@@ -3,32 +3,26 @@ package algebrav1
 import (
 	"context"
 	"fmt"
+	graphqlpkg "github.com/KyberNetwork/kyberswap-dex-lib/pkg/util/graphql"
+
 	"math/big"
 	"strconv"
 
 	"github.com/KyberNetwork/kutils"
+	"github.com/KyberNetwork/kyberswap-dex-lib/pkg/entity"
 	"github.com/KyberNetwork/logger"
 	"github.com/goccy/go-json"
-	"github.com/machinebox/graphql"
-
-	"github.com/KyberNetwork/kyberswap-dex-lib/pkg/entity"
-	graphqlpkg "github.com/KyberNetwork/kyberswap-dex-lib/pkg/util/graphql"
 )
 
 type PoolsListUpdater struct {
 	config        *Config
-	graphqlClient *graphql.Client
+	graphqlClient *graphqlpkg.Client
 }
 
 func NewPoolsListUpdater(
 	cfg *Config,
+	graphqlClient *graphqlpkg.Client,
 ) *PoolsListUpdater {
-	graphqlClient := graphqlpkg.New(graphqlpkg.Config{
-		Url:     cfg.SubgraphAPI,
-		Header:  cfg.SubgraphHeaders,
-		Timeout: graphQLRequestTimeout,
-	})
-
 	return &PoolsListUpdater{
 		config:        cfg,
 		graphqlClient: graphqlClient,
@@ -38,7 +32,7 @@ func NewPoolsListUpdater(
 func (d *PoolsListUpdater) getPoolsList(ctx context.Context, lastCreatedAtTimestamp *big.Int, lastPoolIds []string, first, skip int) ([]SubgraphPool, error) {
 	allowSubgraphError := d.config.AllowSubgraphError
 
-	req := graphql.NewRequest(getPoolsListQuery(allowSubgraphError, lastCreatedAtTimestamp, lastPoolIds, first, skip))
+	req := graphqlpkg.NewRequest(getPoolsListQuery(allowSubgraphError, lastCreatedAtTimestamp, lastPoolIds, first, skip))
 
 	var response struct {
 		Pools []SubgraphPool `json:"pools"`

diff --git a/pkg/liquidity-source/algebra/v1/pool_tracker.go b/pkg/liquidity-source/algebra/v1/pool_tracker.go
@@ -11,7 +11,6 @@ import (
 	"github.com/ethereum/go-ethereum/accounts/abi"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/goccy/go-json"
-	"github.com/machinebox/graphql"
 	"github.com/sourcegraph/conc/pool"
 
 	"github.com/KyberNetwork/kyberswap-dex-lib/pkg/entity"
@@ -24,19 +23,14 @@ import (
 type PoolTracker struct {
 	config        *Config
 	ethrpcClient  *ethrpc.Client
-	graphqlClient *graphql.Client
+	graphqlClient *graphqlpkg.Client
 }
 
 func NewPoolTracker(
 	cfg *Config,
 	ethrpcClient *ethrpc.Client,
+	graphqlClient *graphqlpkg.Client,
 ) (*PoolTracker, error) {
-	graphqlClient := graphqlpkg.New(graphqlpkg.Config{
-		Url:     cfg.SubgraphAPI,
-		Header:  cfg.SubgraphHeaders,
-		Timeout: graphQLRequestTimeout,
-	})
-
 	return &PoolTracker{
 		config:        cfg,
 		ethrpcClient:  ethrpcClient,
@@ -622,7 +616,7 @@ func (d *PoolTracker) getPoolTicks(ctx context.Context, poolAddress string) ([]T
 	var ticks []TickResp
 
 	for {
-		req := graphql.NewRequest(getPoolTicksQuery(allowSubgraphError, poolAddress, skip))
+		req := graphqlpkg.NewRequest(getPoolTicksQuery(allowSubgraphError, poolAddress, skip))
 
 		var resp struct {
 			Pool *SubgraphPoolTicks        `json:"pool"`

diff --git a/pkg/liquidity-source/ambient/config.go b/pkg/liquidity-source/ambient/config.go
@@ -10,11 +10,11 @@ import (
 )
 
 type Config struct {
-	DexID                  string                `json:"dexID"`
-	SubgraphURL            string                `json:"subgraphUrl"`
-	SubgraphHeaders        http.Header           `json:"subgraphHeaders"`
-	SubgraphRequestTimeout durationjson.Duration `json:"subgraphRequestTimeout"`
-	SubgraphLimit          uint64                `json:"subgraphLimit"`
+	DexID           string                `json:"dexID"`
+	SubgraphAPI     string                `json:"subgraphAPI"`
+	SubgraphHeaders http.Header           `json:"subgraphHeaders"`
+	SubgraphTimeout durationjson.Duration `json:"subgraphTimeout"`
+	SubgraphLimit   uint64                `json:"subgraphLimit"`
 
 	// Ambient doesn't use ERC20 wrapped native token when swapping with native token, it uses 0x0 address instead.
 	// kyberswap-dex-lib uses ERC20 wrapped native token to store pool's tokens that are native.

diff --git a/pkg/liquidity-source/ambient/pool_list_updater.go b/pkg/liquidity-source/ambient/pool_list_updater.go
@@ -9,7 +9,6 @@ import (
 	"github.com/KyberNetwork/logger"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/goccy/go-json"
-	"github.com/machinebox/graphql"
 
 	"github.com/KyberNetwork/kyberswap-dex-lib/pkg/entity"
 	"github.com/KyberNetwork/kyberswap-dex-lib/pkg/util"
@@ -20,24 +19,21 @@ import (
 type PoolListUpdater struct {
 	cfg           Config
 	poolDatastore IPoolDatastore
-	subgraph      *graphql.Client
+	graphqlClient *graphqlpkg.Client
 }
 
 func NewPoolsListUpdater(
 	cfg Config,
 	poolDatastore IPoolDatastore,
+	graphqlClient *graphqlpkg.Client,
 ) (*PoolListUpdater, error) {
 	if err := cfg.Validate(); err != nil {
 		return nil, fmt.Errorf("invalid config: %w", err)
 	}
 	return &PoolListUpdater{
 		cfg:           cfg,
 		poolDatastore: poolDatastore,
-		subgraph: graphqlpkg.New(graphqlpkg.Config{
-			Url:     cfg.SubgraphURL,
-			Header:  cfg.SubgraphHeaders,
-			Timeout: cfg.SubgraphRequestTimeout.Duration,
-		}),
+		graphqlClient: graphqlClient,
 	}, nil
 }
 
@@ -118,7 +114,7 @@ func (u *PoolListUpdater) fetchSubgraph(ctx context.Context, lastCreateTime uint
 		limit = u.cfg.SubgraphLimit
 	}
 	var (
-		req = graphql.NewRequest(fmt.Sprintf(`{
+		req = graphqlpkg.NewRequest(fmt.Sprintf(`{
 	pools(
 		where: {
 			timeCreate_gt: %d,
@@ -138,7 +134,7 @@ func (u *PoolListUpdater) fetchSubgraph(ctx context.Context, lastCreateTime uint
 		resp SubgraphPoolsResponse
 	)
 
-	if err := u.subgraph.Run(ctx, req, &resp); err != nil {
+	if err := u.graphqlClient.Run(ctx, req, &resp); err != nil {
 		return nil, err
 	}
 

diff --git a/pkg/liquidity-source/ambient/pool_list_updater_test.go b/pkg/liquidity-source/ambient/pool_list_updater_test.go
@@ -9,12 +9,14 @@ import (
 	"time"
 
 	"github.com/KyberNetwork/blockchain-toolkit/time/durationjson"
+	"github.com/KyberNetwork/kutils"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/goccy/go-json"
 	"github.com/stretchr/testify/require"
 
 	"github.com/KyberNetwork/kyberswap-dex-lib/pkg/entity"
 	"github.com/KyberNetwork/kyberswap-dex-lib/pkg/liquidity-source/ambient"
+	graphqlpkg "github.com/KyberNetwork/kyberswap-dex-lib/pkg/util/graphql"
 )
 
 type mockPoolDataStore struct {
@@ -41,21 +43,26 @@ func TestPoolListUpdater(t *testing.T) {
 
 		config = ambient.Config{
 			DexID:                    "ambient",
-			SubgraphURL:              "https://api.studio.thegraph.com/query/47610/croc-mainnet/version/latest",
-			SubgraphRequestTimeout:   durationjson.Duration{Duration: time.Second * 10},
+			SubgraphAPI:              "https://api.studio.thegraph.com/query/47610/croc-mainnet/version/latest",
+			SubgraphTimeout:          durationjson.Duration{Duration: time.Second * 10},
 			SubgraphLimit:            10,
 			PoolIdx:                  big.NewInt(420),
 			NativeTokenAddress:       "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2",
 			QueryContractAddress:     "0xCA00926b6190c2C59336E73F02569c356d7B6b56",
 			SwapDexContractAddress:   "0xAaAaAAAaA24eEeb8d57D431224f73832bC34f688",
 			MulticallContractAddress: multicallAddress,
 		}
+
+		httpCfg = &kutils.HttpCfg{
+			Timeout: config.SubgraphTimeout.Duration,
+		}
+		graphqlClient = graphqlpkg.NewClient(config.SubgraphAPI, graphqlpkg.WithRestyClient(httpCfg.NewRestyClient()))
 	)
 
 	{
 		t.Logf("first run with limit = 10")
 
-		pu, err := ambient.NewPoolsListUpdater(config, mockPoolDataStore{})
+		pu, err := ambient.NewPoolsListUpdater(config, mockPoolDataStore{}, graphqlClient)
 		require.NoError(t, err)
 		pools, metadataBytes, err = pu.GetNewPools(context.Background(), metadataBytes)
 		require.NoError(t, err)
@@ -71,7 +78,7 @@ func TestPoolListUpdater(t *testing.T) {
 		t.Logf("second run with metadata from first run and limit = 1000")
 
 		config.SubgraphLimit = 1000
-		pu, err := ambient.NewPoolsListUpdater(config, mockPoolDataStore{pool: &firstRunPool})
+		pu, err := ambient.NewPoolsListUpdater(config, mockPoolDataStore{pool: &firstRunPool}, graphqlClient)
 		require.NoError(t, err)
 		pools, metadataBytes, err = pu.GetNewPools(context.Background(), metadataBytes)
 		require.NoError(t, err)

diff --git a/pkg/liquidity-source/balancer-v1/config.go b/pkg/liquidity-source/balancer-v1/config.go
@@ -7,9 +7,9 @@ import (
 )
 
 type Config struct {
-	DexID                  string                `json:"dexID"`
-	NewPoolLimit           int                   `json:"newPoolLimit"`
-	SubgraphURL            string                `json:"subgraphUrl"`
-	SubgraphHeaders        http.Header           `json:"subgraphHeaders"`
-	SubgraphRequestTimeout durationjson.Duration `json:"subgraphRequestTimeout"`
+	DexID           string                `json:"dexID"`
+	NewPoolLimit    int                   `json:"newPoolLimit"`
+	SubgraphAPI     string                `json:"subgraphAPI"`
+	SubgraphHeaders http.Header           `json:"subgraphHeaders"`
+	SubgraphTimeout durationjson.Duration `json:"subgraphTimeout"`
 }