status-go/services/wallet/reader.go

package wallet

import (
	"context"
	"math"
	"math/big"
	"sync"
	"time"

	"go.uber.org/zap"
	"golang.org/x/exp/maps"

	"github.com/ethereum/go-ethereum/common"
	"github.com/ethereum/go-ethereum/common/hexutil"
	"github.com/ethereum/go-ethereum/event"
	gocommon "github.com/status-im/status-go/common"
	"github.com/status-im/status-go/logutils"
	"github.com/status-im/status-go/rpc/chain"
	"github.com/status-im/status-go/services/wallet/async"
	"github.com/status-im/status-go/services/wallet/market"
	"github.com/status-im/status-go/services/wallet/thirdparty"
	"github.com/status-im/status-go/services/wallet/token"
	"github.com/status-im/status-go/services/wallet/transfer"
	"github.com/status-im/status-go/services/wallet/walletevent"
)

// WalletTickReload emitted every 15mn to reload the wallet balance and history
const EventWalletTickReload walletevent.EventType = "wallet-tick-reload"
const EventWalletTickCheckConnected walletevent.EventType = "wallet-tick-check-connected"

const (
	walletTickReloadPeriod      = 10 * time.Minute
	activityReloadDelay         = 30 // Wait this many seconds after activity is detected before triggering a wallet reload
	activityReloadMarginSeconds = 30 // Trigger a wallet reload if activity is detected this many seconds before the last reload
)

func getFixedCurrencies() []string {
	return []string{"USD"}
}

func belongsToMandatoryTokens(symbol string) bool {
	var mandatoryTokens = []string{"ETH", "DAI", "SNT", "STT"}
	for _, t := range mandatoryTokens {
		if t == symbol {
			return true
		}
	}
	return false
}

func NewReader(tokenManager token.ManagerInterface, marketManager *market.Manager, persistence token.TokenBalancesStorage, walletFeed *event.Feed) *Reader {
	return &Reader{
		tokenManager:        tokenManager,
		marketManager:       marketManager,
		persistence:         persistence,
		walletFeed:          walletFeed,
		refreshBalanceCache: true,
	}
}

type Reader struct {
	tokenManager                   token.ManagerInterface
	marketManager                  *market.Manager
	persistence                    token.TokenBalancesStorage
	walletFeed                     *event.Feed
	cancel                         context.CancelFunc
	walletEventsWatcher            *walletevent.Watcher
	lastWalletTokenUpdateTimestamp sync.Map
	reloadDelayTimer               *time.Timer
	refreshBalanceCache            bool
	rw                             sync.RWMutex
}

func splitVerifiedTokens(tokens []*token.Token) ([]*token.Token, []*token.Token) {
	verified := make([]*token.Token, 0)
	unverified := make([]*token.Token, 0)

	for _, t := range tokens {
		if t.Verified {
			verified = append(verified, t)
		} else {
			unverified = append(unverified, t)
		}
	}

	return verified, unverified
}

func getTokenBySymbols(tokens []*token.Token) map[string][]*token.Token {
	res := make(map[string][]*token.Token)

	for _, t := range tokens {
		if _, ok := res[t.Symbol]; !ok {
			res[t.Symbol] = make([]*token.Token, 0)
		}

		res[t.Symbol] = append(res[t.Symbol], t)
	}

	return res
}

func getTokenAddresses(tokens []*token.Token) []common.Address {
	set := make(map[common.Address]bool)
	for _, token := range tokens {
		set[token.Address] = true
	}
	res := make([]common.Address, 0)
	for address := range set {
		res = append(res, address)
	}
	return res
}

func (r *Reader) Start() error {
	ctx, cancel := context.WithCancel(context.Background())
	r.cancel = cancel

	r.startWalletEventsWatcher()

	go func() {
		defer gocommon.LogOnPanic()
		ticker := time.NewTicker(walletTickReloadPeriod)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				r.triggerWalletReload()
			}
		}
	}()
	return nil
}

func (r *Reader) Stop() {
	if r.cancel != nil {
		r.cancel()
	}

	r.stopWalletEventsWatcher()

	r.cancelDelayedWalletReload()

	r.lastWalletTokenUpdateTimestamp = sync.Map{}
}

func (r *Reader) Restart() error {
	r.Stop()
	return r.Start()
}

func (r *Reader) triggerWalletReload() {
	r.cancelDelayedWalletReload()

	r.walletFeed.Send(walletevent.Event{
		Type: EventWalletTickReload,
	})
}

func (r *Reader) triggerDelayedWalletReload() {
	r.cancelDelayedWalletReload()

	r.reloadDelayTimer = time.AfterFunc(time.Duration(activityReloadDelay)*time.Second, r.triggerWalletReload)
}

func (r *Reader) cancelDelayedWalletReload() {

	if r.reloadDelayTimer != nil {
		r.reloadDelayTimer.Stop()
		r.reloadDelayTimer = nil
	}
}

func (r *Reader) startWalletEventsWatcher() {
	if r.walletEventsWatcher != nil {
		return
	}

	// Respond to ETH/Token transfers
	walletEventCb := func(event walletevent.Event) {
		if event.Type != transfer.EventInternalETHTransferDetected &&
			event.Type != transfer.EventInternalERC20TransferDetected {
			return
		}

		for _, address := range event.Accounts {
			timestamp, ok := r.lastWalletTokenUpdateTimestamp.Load(address)
			timecheck := int64(0)
			if ok {
				timecheck = timestamp.(int64) - activityReloadMarginSeconds
			}

			if !ok || event.At > timecheck {
				r.triggerDelayedWalletReload()
				r.invalidateBalanceCache()
				break
			}
		}
	}

	r.walletEventsWatcher = walletevent.NewWatcher(r.walletFeed, walletEventCb)

	r.walletEventsWatcher.Start()
}

func (r *Reader) stopWalletEventsWatcher() {
	if r.walletEventsWatcher != nil {
		r.walletEventsWatcher.Stop()
		r.walletEventsWatcher = nil
	}
}

func (r *Reader) tokensCachedForAddresses(addresses []common.Address) bool {
	cachedTokens, err := r.getCachedWalletTokensWithoutMarketData()
	if err != nil {
		return false
	}

	for _, address := range addresses {
		_, ok := cachedTokens[address]
		if !ok {
			return false
		}
	}

	return true
}

func (r *Reader) isCacheTimestampValidForAddress(address common.Address) bool {
	_, ok := r.lastWalletTokenUpdateTimestamp.Load(address)
	return ok
}

func (r *Reader) areCacheTimestampsValid(addresses []common.Address) bool {
	for _, address := range addresses {
		if !r.isCacheTimestampValidForAddress(address) {
			return false
		}
	}

	return true
}

func (r *Reader) isBalanceCacheValid(addresses []common.Address) bool {
	r.rw.RLock()
	defer r.rw.RUnlock()

	return !r.refreshBalanceCache && r.tokensCachedForAddresses(addresses) && r.areCacheTimestampsValid(addresses)
}

func (r *Reader) balanceRefreshed() {
	r.rw.Lock()
	defer r.rw.Unlock()

	r.refreshBalanceCache = false
}

func (r *Reader) invalidateBalanceCache() {
	r.rw.Lock()
	defer r.rw.Unlock()

	r.refreshBalanceCache = true
}

func (r *Reader) FetchOrGetCachedWalletBalances(ctx context.Context, clients map[uint64]chain.ClientInterface, addresses []common.Address, forceRefresh bool) (map[common.Address][]token.StorageToken, error) {
	needFetch := forceRefresh || !r.isBalanceCacheValid(addresses) || r.isBalanceUpdateNeededAnyway(clients, addresses)

	if needFetch {
		_, err := r.FetchBalances(ctx, clients, addresses)
		if err != nil {
			logutils.ZapLogger().Error("FetchOrGetCachedWalletBalances error", zap.Error(err))
		}
	}

	return r.GetCachedBalances(clients, addresses)
}

func (r *Reader) isBalanceUpdateNeededAnyway(clients map[uint64]chain.ClientInterface, addresses []common.Address) bool {
	cachedTokens, err := r.getCachedWalletTokensWithoutMarketData()
	if err != nil {
		return true
	}

	chainIDs := maps.Keys(clients)
	updateAnyway := false
	for _, address := range addresses {
		if res, ok := cachedTokens[address]; !ok || len(res) == 0 {
			updateAnyway = true
			break
		}

		networkFound := map[uint64]bool{}
		for _, token := range cachedTokens[address] {
			for _, chain := range chainIDs {
				if _, ok := token.BalancesPerChain[chain]; ok {
					networkFound[chain] = true
				}
			}
		}

		for _, chain := range chainIDs {
			if !networkFound[chain] {
				updateAnyway = true
				return updateAnyway
			}
		}
	}

	return updateAnyway
}

func tokensToBalancesPerChain(cachedTokens map[common.Address][]token.StorageToken) map[uint64]map[common.Address]map[common.Address]*hexutil.Big {
	cachedBalancesPerChain := map[uint64]map[common.Address]map[common.Address]*hexutil.Big{}
	for address, tokens := range cachedTokens {
		for _, token := range tokens {
			for _, balance := range token.BalancesPerChain {
				if _, ok := cachedBalancesPerChain[balance.ChainID]; !ok {
					cachedBalancesPerChain[balance.ChainID] = map[common.Address]map[common.Address]*hexutil.Big{}
				}
				if _, ok := cachedBalancesPerChain[balance.ChainID][address]; !ok {
					cachedBalancesPerChain[balance.ChainID][address] = map[common.Address]*hexutil.Big{}
				}

				bigBalance, _ := new(big.Int).SetString(balance.RawBalance, 10)
				cachedBalancesPerChain[balance.ChainID][address][balance.Address] = (*hexutil.Big)(bigBalance)
			}
		}
	}

	return cachedBalancesPerChain
}

func (r *Reader) fetchBalances(ctx context.Context, clients map[uint64]chain.ClientInterface, addresses []common.Address, tokenAddresses []common.Address) (map[uint64]map[common.Address]map[common.Address]*hexutil.Big, error) {
	latestBalances, err := r.tokenManager.GetBalancesByChain(ctx, clients, addresses, tokenAddresses)
	if err != nil {
		logutils.ZapLogger().Error("tokenManager.GetBalancesByChain error", zap.Error(err))
		return nil, err
	}

	return latestBalances, nil
}

func toChainBalance(
	balances map[uint64]map[common.Address]map[common.Address]*hexutil.Big,
	tok *token.Token,
	address common.Address,
	decimals uint,
	cachedTokens map[common.Address][]token.StorageToken,
	hasError bool,
	isMandatoryToken bool,
) *token.ChainBalance {
	hexBalance := &big.Int{}
	if balances != nil {
		hexBalance = balances[tok.ChainID][address][tok.Address].ToInt()
	}

	balance := big.NewFloat(0.0)
	if hexBalance != nil {
		balance = new(big.Float).Quo(
			new(big.Float).SetInt(hexBalance),
			big.NewFloat(math.Pow(10, float64(decimals))),
		)
	}

	isVisible := balance.Cmp(big.NewFloat(0.0)) > 0 || isCachedToken(cachedTokens, address, tok.Symbol, tok.ChainID)
	if !isVisible && !isMandatoryToken {
		return nil
	}

	return &token.ChainBalance{
		RawBalance:     hexBalance.String(),
		Balance:        balance,
		Balance1DayAgo: "0",
		Address:        tok.Address,
		ChainID:        tok.ChainID,
		HasError:       hasError,
	}
}

func (r *Reader) getBalance1DayAgo(balance *token.ChainBalance, dayAgoTimestamp int64, symbol string, address common.Address) (*big.Int, error) {
	balance1DayAgo, err := r.tokenManager.GetTokenHistoricalBalance(address, balance.ChainID, symbol, dayAgoTimestamp)
	if err != nil {
		logutils.ZapLogger().Error("tokenManager.GetTokenHistoricalBalance error", zap.Error(err))
		return nil, err
	}

	return balance1DayAgo, nil
}

func (r *Reader) balancesToTokensByAddress(connectedPerChain map[uint64]bool, addresses []common.Address, allTokens []*token.Token, balances map[uint64]map[common.Address]map[common.Address]*hexutil.Big, cachedTokens map[common.Address][]token.StorageToken) map[common.Address][]token.StorageToken {
	verifiedTokens, unverifiedTokens := splitVerifiedTokens(allTokens)

	result := make(map[common.Address][]token.StorageToken)
	dayAgoTimestamp := time.Now().Add(-24 * time.Hour).Unix()

	for _, address := range addresses {
		for _, tokenList := range [][]*token.Token{verifiedTokens, unverifiedTokens} {
			for symbol, tokens := range getTokenBySymbols(tokenList) {
				balancesPerChain := r.createBalancePerChainPerSymbol(address, balances, tokens, cachedTokens, connectedPerChain, dayAgoTimestamp)
				if balancesPerChain == nil {
					continue
				}

				walletToken := token.StorageToken{
					Token: token.Token{
						Name:          tokens[0].Name,
						Symbol:        symbol,
						Decimals:      tokens[0].Decimals,
						PegSymbol:     token.GetTokenPegSymbol(symbol),
						Verified:      tokens[0].Verified,
						CommunityData: tokens[0].CommunityData,
						Image:         tokens[0].Image,
					},
					BalancesPerChain: balancesPerChain,
				}

				result[address] = append(result[address], walletToken)
			}
		}
	}

	return result
}

func (r *Reader) createBalancePerChainPerSymbol(
	address common.Address,
	balances map[uint64]map[common.Address]map[common.Address]*hexutil.Big,
	tokens []*token.Token,
	cachedTokens map[common.Address][]token.StorageToken,
	clientConnectionPerChain map[uint64]bool,
	dayAgoTimestamp int64,
) map[uint64]token.ChainBalance {
	var balancesPerChain map[uint64]token.ChainBalance
	decimals := tokens[0].Decimals
	isMandatoryToken := belongsToMandatoryTokens(tokens[0].Symbol) // we expect all tokens in the list to have the same symbol
	for _, tok := range tokens {
		hasError := false
		if connected, ok := clientConnectionPerChain[tok.ChainID]; ok {
			hasError = !connected
		}

		if _, ok := balances[tok.ChainID][address][tok.Address]; !ok {
			hasError = true
		}

		// TODO: Avoid passing the entire balances map to toChainBalance. Iterate over the balances map once and pass the balance per address per token to toChainBalance
		balance := toChainBalance(balances, tok, address, decimals, cachedTokens, hasError, isMandatoryToken)
		if balance != nil {
			balance1DayAgo, _ := r.getBalance1DayAgo(balance, dayAgoTimestamp, tok.Symbol, address) // Ignore error
			if balance1DayAgo != nil {
				balance.Balance1DayAgo = balance1DayAgo.String()
			}

			if balancesPerChain == nil {
				balancesPerChain = make(map[uint64]token.ChainBalance)
			}
			balancesPerChain[tok.ChainID] = *balance
		}
	}

	return balancesPerChain
}

func (r *Reader) GetWalletToken(ctx context.Context, clients map[uint64]chain.ClientInterface, addresses []common.Address, currency string) (map[common.Address][]token.StorageToken, error) {
	currencies := make([]string, 0)
	currencies = append(currencies, currency)
	currencies = append(currencies, getFixedCurrencies()...)

	result, err := r.FetchOrGetCachedWalletBalances(ctx, clients, addresses, true)
	if err != nil {
		return nil, err
	}

	tokenSymbols := make([]string, 0)
	for _, storageTokens := range result {
		for _, t := range storageTokens {
			tokenSymbols = append(tokenSymbols, t.Token.Symbol)
		}
	}

	var (
		group             = async.NewAtomicGroup(ctx)
		prices            = map[string]map[string]market.DataPoint{}
		tokenDetails      = map[string]thirdparty.TokenDetails{}
		tokenMarketValues = map[string]thirdparty.TokenMarketValues{}
	)

	group.Add(func(parent context.Context) error {
		prices, err = r.marketManager.GetOrFetchPrices(tokenSymbols, currencies, market.MaxAgeInSecondsForBalances)
		if err != nil {
			logutils.ZapLogger().Info("marketManager.GetOrFetchPrices", zap.Error(err))
		}
		return nil
	})

	group.Add(func(parent context.Context) error {
		tokenDetails, err = r.marketManager.FetchTokenDetails(tokenSymbols)
		if err != nil {
			logutils.ZapLogger().Info("marketManager.FetchTokenDetails", zap.Error(err))
		}
		return nil
	})

	group.Add(func(parent context.Context) error {
		tokenMarketValues, err = r.marketManager.GetOrFetchTokenMarketValues(tokenSymbols, currency, market.MaxAgeInSecondsForBalances)
		if err != nil {
			logutils.ZapLogger().Info("marketManager.GetOrFetchTokenMarketValues", zap.Error(err))
		}
		return nil
	})

	select {
	case <-group.WaitAsync():
	case <-ctx.Done():
		return nil, ctx.Err()
	}
	err = group.Error()
	if err != nil {
		return nil, err
	}

	for address, tokens := range result {
		for index, tok := range tokens {
			marketValuesPerCurrency := make(map[string]token.TokenMarketValues)
			for _, currency := range currencies {
				if _, ok := tokenMarketValues[tok.Symbol]; !ok {
					continue
				}
				marketValuesPerCurrency[currency] = token.TokenMarketValues{
					MarketCap:       tokenMarketValues[tok.Symbol].MKTCAP,
					HighDay:         tokenMarketValues[tok.Symbol].HIGHDAY,
					LowDay:          tokenMarketValues[tok.Symbol].LOWDAY,
					ChangePctHour:   tokenMarketValues[tok.Symbol].CHANGEPCTHOUR,
					ChangePctDay:    tokenMarketValues[tok.Symbol].CHANGEPCTDAY,
					ChangePct24hour: tokenMarketValues[tok.Symbol].CHANGEPCT24HOUR,
					Change24hour:    tokenMarketValues[tok.Symbol].CHANGE24HOUR,
					Price:           prices[tok.Symbol][currency].Price,
					HasError:        !r.marketManager.IsConnected,
				}
			}

			if _, ok := tokenDetails[tok.Symbol]; !ok {
				continue
			}

			result[address][index].Description = tokenDetails[tok.Symbol].Description
			result[address][index].AssetWebsiteURL = tokenDetails[tok.Symbol].AssetWebsiteURL
			result[address][index].BuiltOn = tokenDetails[tok.Symbol].BuiltOn
			result[address][index].MarketValuesPerCurrency = marketValuesPerCurrency
		}
	}

	r.updateTokenUpdateTimestamp(addresses)

	return result, r.persistence.SaveTokens(result)
}

// GetLastTokenUpdateTimestamps returns last timestamps of successful token updates
func (r *Reader) GetLastTokenUpdateTimestamps() map[common.Address]int64 {
	result := make(map[common.Address]int64)

	r.lastWalletTokenUpdateTimestamp.Range(func(key, value interface{}) bool {
		addr, ok1 := key.(common.Address)
		timestamp, ok2 := value.(int64)
		if ok1 && ok2 {
			result[addr] = timestamp
		}
		return true
	})

	return result
}

func isCachedToken(cachedTokens map[common.Address][]token.StorageToken, address common.Address, symbol string, chainID uint64) bool {
	if tokens, ok := cachedTokens[address]; ok {
		for _, t := range tokens {
			if t.Symbol != symbol {
				continue
			}
			_, ok := t.BalancesPerChain[chainID]
			if ok {
				return true
			}
		}
	}
	return false
}

// getCachedWalletTokensWithoutMarketData returns the latest fetched balances, minus
// price information
func (r *Reader) getCachedWalletTokensWithoutMarketData() (map[common.Address][]token.StorageToken, error) {
	return r.persistence.GetTokens()
}

func (r *Reader) updateTokenUpdateTimestamp(addresses []common.Address) {
	for _, address := range addresses {
		r.lastWalletTokenUpdateTimestamp.Store(address, time.Now().Unix())
	}
}

func (r *Reader) FetchBalances(ctx context.Context, clients map[uint64]chain.ClientInterface, addresses []common.Address) (map[common.Address][]token.StorageToken, error) {
	cachedTokens, err := r.getCachedWalletTokensWithoutMarketData()
	if err != nil {
		return nil, err
	}

	chainIDs := maps.Keys(clients)
	allTokens, err := r.tokenManager.GetTokensByChainIDs(chainIDs)
	if err != nil {
		return nil, err
	}

	tokenAddresses := getTokenAddresses(allTokens)
	balances, err := r.fetchBalances(ctx, clients, addresses, tokenAddresses)
	if err != nil {
		logutils.ZapLogger().Error("failed to update balances", zap.Error(err))
		return nil, err
	}

	connectedPerChain := map[uint64]bool{}
	for chainID, client := range clients {
		connectedPerChain[chainID] = client.IsConnected()
	}

	tokens := r.balancesToTokensByAddress(connectedPerChain, addresses, allTokens, balances, cachedTokens)

	err = r.persistence.SaveTokens(tokens)
	if err != nil {
		logutils.ZapLogger().Error("failed to save tokens", zap.Error(err)) // Do not return error, as it is not critical
	}

	r.updateTokenUpdateTimestamp(addresses)
	r.balanceRefreshed()

	return tokens, err
}

func (r *Reader) GetCachedBalances(clients map[uint64]chain.ClientInterface, addresses []common.Address) (map[common.Address][]token.StorageToken, error) {
	cachedTokens, err := r.getCachedWalletTokensWithoutMarketData()
	if err != nil {
		return nil, err
	}

	chainIDs := maps.Keys(clients)
	allTokens, err := r.tokenManager.GetTokensByChainIDs(chainIDs)
	if err != nil {
		return nil, err
	}

	connectedPerChain := map[uint64]bool{}
	for chainID, client := range clients {
		connectedPerChain[chainID] = client.IsConnected()
	}

	balances := tokensToBalancesPerChain(cachedTokens)
	return r.balancesToTokensByAddress(connectedPerChain, addresses, allTokens, balances, cachedTokens), nil
}