feat: replace old transfer detection algorithm with multistandardbalance and transferdetector

Author: dlipicar

services/wallet/activity/service_test.go

@@ -18,7 +18,6 @@ import (
 	"github.com/status-im/status-go/services/wallet/thirdparty"
 	mock_token "github.com/status-im/status-go/services/wallet/token/mock/token"
 	tokenTypes "github.com/status-im/status-go/services/wallet/token/types"
-	"github.com/status-im/status-go/services/wallet/transfer"
 	"github.com/status-im/status-go/services/wallet/walletevent"
 	"github.com/status-im/status-go/t/helpers"
 	"github.com/status-im/status-go/transactions"
@@ -108,13 +107,6 @@ func setupTransactions(t *testing.T, state testState, txCount int, testTxs []tra
 		allAddresses = append(allAddresses, p.From, p.To)
 	}
 
-	txs, fromTrs, toTrs := transfer.GenerateTestTransfers(t, state.service.db, len(pendings), txCount)
-	for i := range txs {
-		transfer.InsertTestTransfer(t, state.service.db, txs[i].To, &txs[i])
-	}
-
-	allAddresses = append(append(allAddresses, fromTrs...), toTrs...)
-
 	state.tokenMock.EXPECT().LookupTokenIdentity(gomock.Any(), gomock.Any(), gomock.Any()).Return(
 		&tokenTypes.Token{
 			ChainID: 5,

services/wallet/activity/session_service.go

@@ -19,7 +19,6 @@ import (
 	"github.com/status-im/status-go/services/wallet/common"
 	"github.com/status-im/status-go/services/wallet/responses"
 	"github.com/status-im/status-go/services/wallet/routeexecution"
-	"github.com/status-im/status-go/services/wallet/transfer"
 	"github.com/status-im/status-go/services/wallet/walletevent"
 	"github.com/status-im/status-go/transactions"
 
@@ -317,12 +316,6 @@ func (s *Service) processEvents(ctx context.Context) {
 					Hash:    payload.Hash,
 				})
 				debounceProcessChangesFn()
-			case transfer.EventNewTransfers:
-				eventCount++
-				// No updates here, these are detected with their final state, just trigger
-				// the detection of new entries
-				newTxs = true
-				debounceProcessChangesFn()
 			case routeexecution.EventRouteExecutionTransactionSent:
 				sentTxs, ok := event.EventParams.(*responses.RouterSentTransactions)
 				if ok && sentTxs != nil {

services/wallet/collectibles/contract_type_db.go

@@ -10,6 +10,28 @@ import (
 	"github.com/status-im/status-go/sqlite"
 )
 
+type ContractTypeDB struct {
+	db *sql.DB
+}
+
+func NewContractTypeDB(sqlDb *sql.DB) *ContractTypeDB {
+	return &ContractTypeDB{
+		db: sqlDb,
+	}
+}
+
+func (o *ContractTypeDB) GetContractTypes(ids []thirdparty.ContractID) (map[thirdparty.ContractID]w_common.ContractType, error) {
+	ret := make(map[thirdparty.ContractID]w_common.ContractType)
+	var err error
+	for _, id := range ids {
+		ret[id], err = readContractType(o.db, id)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return ret, nil
+}
+
 func upsertContractType(creator sqlite.StatementCreator, id thirdparty.ContractID, contractType w_common.ContractType) error {
 	if contractType == w_common.ContractTypeUnknown {
 		return nil

services/wallet/collectibles/ownership/controller.go

@@ -4,12 +4,18 @@ package ownership
 
 import (
 	"context"
+	"slices"
 	"sync"
+	"time"
 
 	"go.uber.org/zap"
 
 	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/event"
+
+	"github.com/status-im/go-wallet-sdk/pkg/balance/multistandardfetcher"
+	"github.com/status-im/go-wallet-sdk/pkg/contracts/erc1155"
+	"github.com/status-im/go-wallet-sdk/pkg/contracts/erc721"
+	"github.com/status-im/go-wallet-sdk/pkg/eventlog"
 
 	gocommon "github.com/status-im/status-go/common"
 	"github.com/status-im/status-go/crypto/types"
@@ -18,8 +24,9 @@ import (
 	"github.com/status-im/status-go/rpc/network"
 	"github.com/status-im/status-go/services/accounts/accountsevent"
 	walletCommon "github.com/status-im/status-go/services/wallet/common"
+	"github.com/status-im/status-go/services/wallet/multistandardbalance"
 	"github.com/status-im/status-go/services/wallet/thirdparty"
-	"github.com/status-im/status-go/services/wallet/transfer"
+	"github.com/status-im/status-go/services/wallet/transferdetector"
 	"github.com/status-im/status-go/services/wallet/walletevent"
 )
 
@@ -30,8 +37,6 @@ const (
 type loaderPerChainID = map[walletCommon.ChainID]*PeriodicalLoader
 type loaderPerAddressAndChainID = map[common.Address]loaderPerChainID
 
-type TransferCb func(common.Address, walletCommon.ChainID, []transfer.Transfer)
-
 type AccountsProvider interface {
 	GetWalletAddresses() ([]types.Address, error)
 }
@@ -41,12 +46,18 @@ type NetworksProvider interface {
 	GetPublisher() *pubsub.Publisher
 }
 
+type BlockChainStateProvider interface {
+	GetEstimatedBlockTime(ctx context.Context, chainID uint64, blockNumber uint64) (time.Time, error)
+}
+
 type Controller struct {
-	fetcher           CollectibleOwnershipFetcher
-	storage           CollectibleOwnershipStorage
-	walletFeed        *event.Feed
-	accountsProvider  AccountsProvider
-	accountsPublisher *pubsub.Publisher
+	fetcher                       CollectibleOwnershipFetcher
+	storage                       CollectibleOwnershipStorage
+	accountsProvider              AccountsProvider
+	accountsPublisher             *pubsub.Publisher
+	multistandardBalancePublisher *pubsub.Publisher
+	transferDetectorPublisher     *pubsub.Publisher
+	blockChainStateProvider       BlockChainStateProvider
 
 	networksProvider NetworksProvider
 
@@ -66,24 +77,28 @@ type Controller struct {
 
 func NewController(
 	storage CollectibleOwnershipStorage,
-	walletFeed *event.Feed,
 	accountsProvider AccountsProvider,
 	accountsPublisher *pubsub.Publisher,
 	networksProvider NetworksProvider,
+	multistandardBalancePublisher *pubsub.Publisher,
+	transferDetectorPublisher *pubsub.Publisher,
+	blockChainStateProvider BlockChainStateProvider,
 	fetcher CollectibleOwnershipFetcher,
 	collectiblesPublisher *pubsub.Publisher,
 	logger *zap.Logger,
 ) *Controller {
 	return &Controller{
-		fetcher:               fetcher,
-		storage:               storage,
-		walletFeed:            walletFeed,
-		accountsProvider:      accountsProvider,
-		accountsPublisher:     accountsPublisher,
-		networksProvider:      networksProvider,
-		periodicalLoaders:     make(loaderPerAddressAndChainID),
-		collectiblesPublisher: collectiblesPublisher,
-		logger:                logger.Named("OwnershipController"),
+		fetcher:                       fetcher,
+		storage:                       storage,
+		accountsProvider:              accountsProvider,
+		accountsPublisher:             accountsPublisher,
+		networksProvider:              networksProvider,
+		multistandardBalancePublisher: multistandardBalancePublisher,
+		transferDetectorPublisher:     transferDetectorPublisher,
+		blockChainStateProvider:       blockChainStateProvider,
+		periodicalLoaders:             make(loaderPerAddressAndChainID),
+		collectiblesPublisher:         collectiblesPublisher,
+		logger:                        logger.Named("OwnershipController"),
 	}
 }
 
@@ -105,11 +120,14 @@ func (c *Controller) Start() {
 	// Setup collectibles fetch when a new account gets added
 	c.startAccountsWatcher()
 
-	// Setup collectibles fetch when relevant activity is detected
-	c.startWalletEventsWatcher()
-
 	// Setup collectibles fetch when active networks change
 	c.startNetworkEventsWatcher()
+
+	// Start balance change watcher
+	c.startBalanceChangeWatcher()
+
+	// Start transfer detection watcher
+	c.startTransferDetectionWatcher()
 }
 
 func (c *Controller) Stop() {
@@ -120,8 +138,6 @@ func (c *Controller) Stop() {
 	close(c.stopCh)
 	c.stopCh = nil
 
-	c.stopWalletEventsWatcher()
-
 	c.stopPeriodicalLoaders()
 }
 
@@ -237,63 +253,118 @@ func (c *Controller) startAccountsWatcher() {
 	}()
 }
 
-func (c *Controller) startWalletEventsWatcher() {
-	if c.walletEventsWatcher != nil {
-		return
-	}
-
-	if c.walletFeed == nil {
+func (c *Controller) startNetworkEventsWatcher() {
+	if c.networksProvider == nil {
 		return
 	}
 
-	walletEventCb := func(event walletevent.Event) {
-		if event.Type != transfer.EventInternalERC721TransferDetected &&
-			event.Type != transfer.EventInternalERC1155TransferDetected {
-			return
-		}
-
-		chainID := walletCommon.ChainID(event.ChainID)
-		for _, account := range event.Accounts {
-			c.refetchOwnershipIfRecentTransfer(account, chainID, event.At)
+	ch, unsub := pubsub.Subscribe[network.EventActiveNetworksChanged](c.networksProvider.GetPublisher(), 10)
+	go func() {
+		defer gocommon.LogOnPanic()
+		defer unsub()
+		for {
+			select {
+			case <-c.stopCh:
+				return
+			case _, ok := <-ch:
+				if !ok {
+					return
+				}
+				c.checkPeriodicalLoaders()
+			}
 		}
-	}
-
-	c.walletEventsWatcher = walletevent.NewWatcher(c.walletFeed, walletEventCb)
-
-	c.walletEventsWatcher.Start()
+	}()
 }
 
-func (c *Controller) stopWalletEventsWatcher() {
-	if c.walletEventsWatcher != nil {
-		c.walletEventsWatcher.Stop()
-		c.walletEventsWatcher = nil
+func (c *Controller) startBalanceChangeWatcher() {
+	if c.multistandardBalancePublisher == nil {
+		return
 	}
+
+	ch, unsub := pubsub.Subscribe[multistandardbalance.EventBalanceFetchFinished](c.multistandardBalancePublisher, 10)
+	go func() {
+		defer gocommon.LogOnPanic()
+		defer unsub()
+		for {
+			select {
+			case <-c.stopCh:
+				return
+			case event, ok := <-ch:
+				if !ok {
+					return
+				}
+				switch event.ResultType {
+				case multistandardfetcher.ResultTypeERC721, multistandardfetcher.ResultTypeERC1155:
+					if event.BalanceChanged {
+						c.refetchOwnershipIfRecentTx(event.Key.Account, walletCommon.ChainID(event.Key.ChainID), event.NewState.FetchedAt)
+					}
+				}
+			}
+		}
+	}()
 }
 
-func (c *Controller) startNetworkEventsWatcher() {
-	if c.networksProvider == nil {
+func (c *Controller) startTransferDetectionWatcher() {
+	if c.transferDetectorPublisher == nil {
 		return
 	}
 
-	ch, unsub := pubsub.Subscribe[network.EventActiveNetworksChanged](c.networksProvider.GetPublisher(), 10)
+	ch, unsub := pubsub.Subscribe[transferdetector.EventTransferDetectionFinished](c.transferDetectorPublisher, 10)
 	go func() {
 		defer gocommon.LogOnPanic()
 		defer unsub()
 		for {
 			select {
 			case <-c.stopCh:
 				return
-			case _, ok := <-ch:
+			case msg, ok := <-ch:
 				if !ok {
 					return
 				}
-				c.checkPeriodicalLoaders()
+				for _, event := range msg.Events {
+					switch event.EventKey {
+					case eventlog.ERC721Transfer:
+						unpackedEvent, ok := event.Unpacked.(erc721.Erc721Transfer)
+						if !ok {
+							c.logger.Error("failed to unpack ERC721Transfer event")
+							continue
+						}
+						c.refetchOwnershipIfRelevantEvent(msg.Accounts, unpackedEvent.From, unpackedEvent.To, msg.ChainID, unpackedEvent.Raw.BlockNumber)
+					case eventlog.ERC1155TransferSingle:
+						unpackedEvent, ok := event.Unpacked.(erc1155.Erc1155TransferSingle)
+						if !ok {
+							c.logger.Error("failed to unpack ERC1155TransferSingle event")
+							continue
+						}
+						c.refetchOwnershipIfRelevantEvent(msg.Accounts, unpackedEvent.From, unpackedEvent.To, msg.ChainID, unpackedEvent.Raw.BlockNumber)
+					case eventlog.ERC1155TransferBatch:
+						unpackedEvent, ok := event.Unpacked.(erc1155.Erc1155TransferBatch)
+						if !ok {
+							c.logger.Error("failed to unpack ERC1155TransferBatch event")
+							continue
+						}
+						c.refetchOwnershipIfRelevantEvent(msg.Accounts, unpackedEvent.From, unpackedEvent.To, msg.ChainID, unpackedEvent.Raw.BlockNumber)
+					}
+				}
 			}
 		}
 	}()
 }
 
-func (c *Controller) refetchOwnershipIfRecentTransfer(account common.Address, chainID walletCommon.ChainID, latestTxTimestamp int64) {
+func (c *Controller) refetchOwnershipIfRelevantEvent(checkedAccounts []common.Address, eventFrom common.Address, eventTo common.Address, chainID uint64, blockNumber uint64) {
+	for _, address := range []common.Address{eventFrom, eventTo} {
+		if slices.Contains(checkedAccounts, address) {
+			blockTime, err := c.blockChainStateProvider.GetEstimatedBlockTime(context.TODO(), chainID, blockNumber)
+			if err != nil {
+				c.logger.Error("failed to get estimated block time", zap.Error(err))
+				continue
+			}
+			c.refetchOwnershipIfRecentTx(address, walletCommon.ChainID(chainID), blockTime.Unix())
+		}
+	}
+}
+
+func (c *Controller) refetchOwnershipIfRecentTx(account common.Address, chainID walletCommon.ChainID, latestTxTimestamp int64) {
 	// Check last ownership update timestamp
 	timestamp, err := c.storage.GetOwnershipUpdateTimestamp(account, chainID)