status-go/rpc/client.go

package rpc

//go:generate mockgen -package=mock_rpcclient -source=client.go -destination=mock/client/client.go

import (
	"context"
	"database/sql"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"net/url"
	"reflect"
	"strings"
	"sync"
	"time"

	"github.com/ethereum/go-ethereum/log"
	gethrpc "github.com/ethereum/go-ethereum/rpc"

	appCommon "github.com/status-im/status-go/common"
	"github.com/status-im/status-go/params"
	"github.com/status-im/status-go/rpc/chain"
	"github.com/status-im/status-go/rpc/chain/ethclient"
	"github.com/status-im/status-go/rpc/chain/rpclimiter"
	"github.com/status-im/status-go/rpc/network"
	"github.com/status-im/status-go/services/rpcstats"
	"github.com/status-im/status-go/services/wallet/common"
)

const (
	// DefaultCallTimeout is a default timeout for an RPC call
	DefaultCallTimeout = time.Minute

	// Names of providers
	providerGrove       = "grove"
	providerInfura      = "infura"
	ProviderStatusProxy = "status-proxy"

	mobile  = "mobile"
	desktop = "desktop"

	// rpcUserAgentFormat 'procurator': *an agent representing others*, aka a "proxy"
	// allows for the rpc client to have a dedicated user agent, which is useful for the proxy server logs.
	rpcUserAgentFormat = "procuratee-%s/%s"

	// rpcUserAgentUpstreamFormat a separate user agent format for upstream, because we should not be using upstream
	// if we see this user agent in the logs that means parts of the application are using a malconfigured http client
	rpcUserAgentUpstreamFormat = "procuratee-%s-upstream/%s"
)

// List of RPC client errors.
var (
	ErrMethodNotFound = fmt.Errorf("the method does not exist/is not available")
)

var (
	// rpcUserAgentName the user agent
	rpcUserAgentName         = fmt.Sprintf(rpcUserAgentFormat, "no-GOOS", params.Version)
	rpcUserAgentUpstreamName = fmt.Sprintf(rpcUserAgentUpstreamFormat, "no-GOOS", params.Version)
)

func init() {
	if appCommon.IsMobilePlatform() {
		rpcUserAgentName = fmt.Sprintf(rpcUserAgentFormat, mobile, params.Version)
		rpcUserAgentUpstreamName = fmt.Sprintf(rpcUserAgentUpstreamFormat, mobile, params.Version)
	} else {
		rpcUserAgentName = fmt.Sprintf(rpcUserAgentFormat, desktop, params.Version)
		rpcUserAgentUpstreamName = fmt.Sprintf(rpcUserAgentUpstreamFormat, desktop, params.Version)
	}
}

// Handler defines handler for RPC methods.
type Handler func(context.Context, uint64, ...interface{}) (interface{}, error)

type ClientInterface interface {
	AbstractEthClient(chainID common.ChainID) (ethclient.BatchCallClient, error)
	EthClient(chainID uint64) (chain.ClientInterface, error)
	EthClients(chainIDs []uint64) (map[uint64]chain.ClientInterface, error)
	CallContext(context context.Context, result interface{}, chainID uint64, method string, args ...interface{}) error
	Call(result interface{}, chainID uint64, method string, args ...interface{}) error
	CallRaw(body string) string
	GetNetworkManager() *network.Manager
}

// Client represents RPC client with custom routing
// scheme. It automatically decides where RPC call
// goes - Upstream or Local node.
type Client struct {
	sync.RWMutex

	upstreamEnabled bool
	upstreamURL     string
	UpstreamChainID uint64

	local              *gethrpc.Client
	upstream           chain.ClientInterface
	rpcClientsMutex    sync.RWMutex
	rpcClients         map[uint64]chain.ClientInterface
	rpsLimiterMutex    sync.RWMutex
	limiterPerProvider map[string]*rpclimiter.RPCRpsLimiter

	router         *router
	NetworkManager *network.Manager

	handlersMx sync.RWMutex       // mx guards handlers
	handlers   map[string]Handler // locally registered handlers
	log        log.Logger

	walletNotifier  func(chainID uint64, message string)
	providerConfigs []params.ProviderConfig
}

// Is initialized in a build-tag-dependent module
var verifProxyInitFn func(c *Client)

// NewClient initializes Client and tries to connect to both,
// upstream and local node.
//
// Client is safe for concurrent use and will automatically
// reconnect to the server if connection is lost.
func NewClient(client *gethrpc.Client, upstreamChainID uint64, upstream params.UpstreamRPCConfig, networks []params.Network, db *sql.DB, providerConfigs []params.ProviderConfig) (*Client, error) {
	var err error

	log := log.New("package", "status-go/rpc.Client")
	networkManager := network.NewManager(db)
	if networkManager == nil {
		return nil, errors.New("failed to create network manager")
	}

	err = networkManager.Init(networks)
	if err != nil {
		log.Error("Network manager failed to initialize", "error", err)
	}

	c := Client{
		local:              client,
		NetworkManager:     networkManager,
		handlers:           make(map[string]Handler),
		rpcClients:         make(map[uint64]chain.ClientInterface),
		limiterPerProvider: make(map[string]*rpclimiter.RPCRpsLimiter),
		log:                log,
		providerConfigs:    providerConfigs,
	}

	var opts []gethrpc.ClientOption
	opts = append(opts,
		gethrpc.WithHeaders(http.Header{
			"User-Agent": {rpcUserAgentUpstreamName},
		}),
	)

	if upstream.Enabled {
		c.UpstreamChainID = upstreamChainID
		c.upstreamEnabled = upstream.Enabled
		c.upstreamURL = upstream.URL
		upstreamClient, err := gethrpc.DialOptions(context.Background(), c.upstreamURL, opts...)
		if err != nil {
			return nil, fmt.Errorf("dial upstream server: %s", err)
		}
		limiter, err := c.getRPCRpsLimiter(c.upstreamURL)
		if err != nil {
			return nil, fmt.Errorf("get RPC limiter: %s", err)
		}
		hostPortUpstream, err := extractHostFromURL(c.upstreamURL)
		if err != nil {
			hostPortUpstream = "upstream"
		}

		// Include the chain-id in the rpc client
		rpcName := fmt.Sprintf("%s-chain-id-%d", hostPortUpstream, upstreamChainID)

		ethClients := []ethclient.RPSLimitedEthClientInterface{
			ethclient.NewRPSLimitedEthClient(upstreamClient, limiter, rpcName),
		}
		c.upstream = chain.NewClient(ethClients, upstreamChainID)
	}

	c.router = newRouter(c.upstreamEnabled)

	if verifProxyInitFn != nil {
		verifProxyInitFn(&c)
	}

	return &c, nil
}

func (c *Client) GetNetworkManager() *network.Manager {
	return c.NetworkManager
}

func (c *Client) SetWalletNotifier(notifier func(chainID uint64, message string)) {
	c.walletNotifier = notifier
}

func extractHostFromURL(inputURL string) (string, error) {
	parsedURL, err := url.Parse(inputURL)
	if err != nil {
		return "", err
	}

	return parsedURL.Host, nil
}

func (c *Client) getRPCRpsLimiter(key string) (*rpclimiter.RPCRpsLimiter, error) {
	c.rpsLimiterMutex.Lock()
	defer c.rpsLimiterMutex.Unlock()
	if limiter, ok := c.limiterPerProvider[key]; ok {
		return limiter, nil
	}
	limiter := rpclimiter.NewRPCRpsLimiter()
	c.limiterPerProvider[key] = limiter
	return limiter, nil
}

func getProviderConfig(providerConfigs []params.ProviderConfig, providerName string) (params.ProviderConfig, error) {
	for _, providerConfig := range providerConfigs {
		if providerConfig.Name == providerName {
			return providerConfig, nil
		}
	}
	return params.ProviderConfig{}, fmt.Errorf("provider config not found for provider: %s", providerName)
}

func (c *Client) getClientUsingCache(chainID uint64) (chain.ClientInterface, error) {
	c.rpcClientsMutex.Lock()
	defer c.rpcClientsMutex.Unlock()
	if rpcClient, ok := c.rpcClients[chainID]; ok {
		if rpcClient.GetWalletNotifier() == nil {
			rpcClient.SetWalletNotifier(c.walletNotifier)
		}
		return rpcClient, nil
	}

	network := c.NetworkManager.Find(chainID)
	if network == nil {
		if c.UpstreamChainID == chainID {
			return c.upstream, nil
		}
		return nil, fmt.Errorf("could not find network: %d", chainID)
	}

	ethClients := c.getEthClients(network)
	if len(ethClients) == 0 {
		return nil, fmt.Errorf("could not find any RPC URL for chain: %d", chainID)
	}

	client := chain.NewClient(ethClients, chainID)
	client.SetWalletNotifier(c.walletNotifier)
	c.rpcClients[chainID] = client
	return client, nil
}

func (c *Client) getEthClients(network *params.Network) []ethclient.RPSLimitedEthClientInterface {
	urls := make(map[string]string)
	keys := make([]string, 0)
	authMap := make(map[string]string)

	// find proxy provider
	proxyProvider, err := getProviderConfig(c.providerConfigs, ProviderStatusProxy)
	if err != nil {
		c.log.Warn("could not find provider config for status-proxy", "error", err)
	}

	if proxyProvider.Enabled {
		key := ProviderStatusProxy
		keyFallback := ProviderStatusProxy + "-fallback"
		keyFallback2 := ProviderStatusProxy + "-fallback2"
		urls[key] = network.DefaultRPCURL
		urls[keyFallback] = network.DefaultFallbackURL
		urls[keyFallback2] = network.DefaultFallbackURL2
		keys = []string{key, keyFallback, keyFallback2}
		authMap[key] = proxyProvider.User + ":" + proxyProvider.Password
		authMap[keyFallback] = authMap[key]
		authMap[keyFallback2] = authMap[key]
	}
	keys = append(keys, []string{"main", "fallback"}...)
	urls["main"] = network.RPCURL
	urls["fallback"] = network.FallbackURL

	ethClients := make([]ethclient.RPSLimitedEthClientInterface, 0)
	for index, key := range keys {
		var rpcClient *gethrpc.Client
		var rpcLimiter *rpclimiter.RPCRpsLimiter
		var err error
		var hostPort string
		url := urls[key]

		if len(url) > 0 {
			// For now, we only support auth for status-proxy.
			authStr, ok := authMap[key]
			var opts []gethrpc.ClientOption
			if ok {
				authEncoded := base64.StdEncoding.EncodeToString([]byte(authStr))
				opts = append(opts,
					gethrpc.WithHeaders(http.Header{
						"Authorization": {"Basic " + authEncoded},
						"User-Agent":    {rpcUserAgentName},
					}),
				)
			}

			rpcClient, err = gethrpc.DialOptions(context.Background(), url, opts...)
			if err != nil {
				c.log.Error("dial server "+key, "error", err)
			}

			// If using the status-proxy, consider each endpoint as a separate provider
			circuitKey := fmt.Sprintf("%s-%d", key, index)
			// Otherwise host is good enough
			if !strings.Contains(url, "status.im") {
				hostPort, err = extractHostFromURL(url)
				if err == nil {
					circuitKey = hostPort
				}
			}

			rpcLimiter, err = c.getRPCRpsLimiter(circuitKey)
			if err != nil {
				c.log.Error("get RPC limiter "+key, "error", err)
			}

			ethClients = append(ethClients, ethclient.NewRPSLimitedEthClient(rpcClient, rpcLimiter, circuitKey))
		}
	}

	return ethClients
}

// EthClient returns ethclient.Client per chain
func (c *Client) EthClient(chainID uint64) (chain.ClientInterface, error) {
	client, err := c.getClientUsingCache(chainID)
	if err != nil {
		return nil, err
	}

	return client, nil
}

// AbstractEthClient returns a partial abstraction used by new components for testing purposes
func (c *Client) AbstractEthClient(chainID common.ChainID) (ethclient.BatchCallClient, error) {
	client, err := c.getClientUsingCache(uint64(chainID))
	if err != nil {
		return nil, err
	}

	return client, nil
}

func (c *Client) EthClients(chainIDs []uint64) (map[uint64]chain.ClientInterface, error) {
	clients := make(map[uint64]chain.ClientInterface, 0)
	for _, chainID := range chainIDs {
		client, err := c.getClientUsingCache(chainID)
		if err != nil {
			return nil, err
		}
		clients[chainID] = client
	}

	return clients, nil
}

// SetClient strictly for testing purposes
func (c *Client) SetClient(chainID uint64, client chain.ClientInterface) {
	c.rpcClientsMutex.Lock()
	defer c.rpcClientsMutex.Unlock()
	c.rpcClients[chainID] = client
}

// UpdateUpstreamURL changes the upstream RPC client URL, if the upstream is enabled.
func (c *Client) UpdateUpstreamURL(url string) error {
	if c.upstream == nil {
		return nil
	}

	rpcClient, err := gethrpc.Dial(url)
	if err != nil {
		return err
	}
	rpsLimiter, err := c.getRPCRpsLimiter(url)
	if err != nil {
		return err
	}
	c.Lock()
	hostPortUpstream, err := extractHostFromURL(url)
	if err != nil {
		hostPortUpstream = "upstream"
	}

	ethClients := []ethclient.RPSLimitedEthClientInterface{
		ethclient.NewRPSLimitedEthClient(rpcClient, rpsLimiter, hostPortUpstream),
	}
	c.upstream = chain.NewClient(ethClients, c.UpstreamChainID)
	c.upstreamURL = url
	c.Unlock()

	return nil
}

// Call performs a JSON-RPC call with the given arguments and unmarshals into
// result if no error occurred.
//
// The result must be a pointer so that package json can unmarshal into it. You
// can also pass nil, in which case the result is ignored.
//
// It uses custom routing scheme for calls.
func (c *Client) Call(result interface{}, chainID uint64, method string, args ...interface{}) error {
	ctx := context.Background()
	return c.CallContext(ctx, result, chainID, method, args...)
}

// CallContext performs a JSON-RPC call with the given arguments. If the context is
// canceled before the call has successfully returned, CallContext returns immediately.
//
// The result must be a pointer so that package json can unmarshal into it. You
// can also pass nil, in which case the result is ignored.
//
// It uses custom routing scheme for calls.
// If there are any local handlers registered for this call, they will handle it.
func (c *Client) CallContext(ctx context.Context, result interface{}, chainID uint64, method string, args ...interface{}) error {
	rpcstats.CountCall(method)
	if c.router.routeBlocked(method) {
		return ErrMethodNotFound
	}

	// check locally registered handlers first
	if handler, ok := c.handler(method); ok {
		return c.callMethod(ctx, result, chainID, handler, args...)
	}

	return c.CallContextIgnoringLocalHandlers(ctx, result, chainID, method, args...)
}

// CallContextIgnoringLocalHandlers performs a JSON-RPC call with the given
// arguments.
//
// If there are local handlers registered for this call, they would
// be ignored. It is useful if the call is happening from within a local
// handler itself.
// Upstream calls routing will be used anyway.
func (c *Client) CallContextIgnoringLocalHandlers(ctx context.Context, result interface{}, chainID uint64, method string, args ...interface{}) error {
	if c.router.routeBlocked(method) {
		return ErrMethodNotFound
	}

	if c.router.routeRemote(method) {
		client, err := c.getClientUsingCache(chainID)
		if err != nil {
			return err
		}
		return client.CallContext(ctx, result, method, args...)
	}

	if c.local == nil {
		c.log.Warn("Local JSON-RPC endpoint missing", "method", method)
		return errors.New("missing local JSON-RPC endpoint")
	}
	return c.local.CallContext(ctx, result, method, args...)
}

// RegisterHandler registers local handler for specific RPC method.
//
// If method is registered, it will be executed with given handler and
// never routed to the upstream or local servers.
func (c *Client) RegisterHandler(method string, handler Handler) {
	c.handlersMx.Lock()
	defer c.handlersMx.Unlock()

	c.handlers[method] = handler
}

// UnregisterHandler removes a previously registered handler.
func (c *Client) UnregisterHandler(method string) {
	c.handlersMx.Lock()
	defer c.handlersMx.Unlock()

	delete(c.handlers, method)
}

// callMethod calls registered RPC handler with given args and pointer to result.
// It handles proper params and result converting
//
// TODO(divan): use cancellation via context here?
func (c *Client) callMethod(ctx context.Context, result interface{}, chainID uint64, handler Handler, args ...interface{}) error {
	response, err := handler(ctx, chainID, args...)
	if err != nil {
		return err
	}

	// if result is nil, just ignore result -
	// the same way as gethrpc.CallContext() caller would expect
	if result == nil {
		return nil
	}

	return setResultFromRPCResponse(result, response)
}

// handler is a concurrently safe method to get registered handler by name.
func (c *Client) handler(method string) (Handler, bool) {
	c.handlersMx.RLock()
	defer c.handlersMx.RUnlock()
	handler, ok := c.handlers[method]
	return handler, ok
}

// setResultFromRPCResponse tries to set result value from response using reflection
// as concrete types are unknown.
func setResultFromRPCResponse(result, response interface{}) (err error) {
	defer func() {
		if r := recover(); r != nil {
			err = fmt.Errorf("invalid result type: %s", r)
		}
	}()

	responseValue := reflect.ValueOf(response)

	// If it is called via CallRaw, result has type json.RawMessage and
	// we should marshal the response before setting it.
	// Otherwise, it is called with CallContext and result is of concrete type,
	// thus we should try to set it as it is.
	// If response type and result type are incorrect, an error should be returned.
	// TODO(divan): add additional checks for result underlying value, if needed:
	// some example: https://golang.org/src/encoding/json/decode.go#L596
	switch reflect.ValueOf(result).Elem().Type() {
	case reflect.TypeOf(json.RawMessage{}), reflect.TypeOf([]byte{}):
		data, err := json.Marshal(response)
		if err != nil {
			return err
		}

		responseValue = reflect.ValueOf(data)
	}

	value := reflect.ValueOf(result).Elem()
	if !value.CanSet() {
		return errors.New("can't assign value to result")
	}
	value.Set(responseValue)

	return nil
}