[wallet] Fix multiple notifications on a single erc20 transfer

VERSION

@@ -1 +1 @@
-0.64.5
+0.64.6

services/local-notifications/core.go

@@ -166,7 +166,7 @@ func (s *Service) transactionsHandler(payload TransactionEvent) {
 	limit := 20
 	if payload.BlockNumber != nil {
 		for _, address := range payload.Accounts {
-			if payload.BlockNumber.Cmp(payload.MaxKnownBlocks[address]) == 1 {
+			if payload.BlockNumber.Cmp(payload.MaxKnownBlocks[address]) >= 0 {
 				log.Info("Handled transfer for address", "info", address)
 				transfers, err := s.walletDB.GetTransfersByAddressAndBlock(address, payload.BlockNumber, int64(limit))
 				if err != nil {
@@ -235,6 +235,17 @@ func (s *Service) StartWalletWatcher() {
 				return
 			case event := <-events:
 				if event.Type == wallet.EventNewBlock && len(maxKnownBlocks) > 0 {
+					newBlocks := false
+					for _, address := range event.Accounts {
+						if _, ok := maxKnownBlocks[address]; !ok {
+							newBlocks = true
+							maxKnownBlocks[address] = event.BlockNumber
+						} else if event.BlockNumber.Cmp(maxKnownBlocks[address]) == 1 {
+							maxKnownBlocks[address] = event.BlockNumber
+							newBlocks = true
+						}
+					}
+					if newBlocks {
 						s.transmitter.publisher.Send(TransactionEvent{
 							Type:                      string(event.Type),
 							BlockNumber:               event.BlockNumber,
@@ -243,6 +254,7 @@ func (s *Service) StartWalletWatcher() {
 							ERC20:                     event.ERC20,
 							MaxKnownBlocks:            maxKnownBlocks,
 						})
+					}
 				} else if event.Type == wallet.EventMaxKnownBlock {
 					for _, address := range event.Accounts {
 						if _, ok := maxKnownBlocks[address]; !ok {

services/wallet/commands.go

@@ -990,9 +990,27 @@ func (c *findAndCheckBlockRangeCommand) Run(parent context.Context) (err error)
 	for _, address := range c.accounts {
 		ethHeaders := ethHeadersByAddress[address]
 		erc20Headers := erc20HeadersByAddress[address]
-
 		allHeaders := append(ethHeaders, erc20Headers...)
-		foundHeaders[address] = allHeaders
+
+		uniqHeadersByHash := map[common.Hash]*DBHeader{}
+		for _, header := range allHeaders {
+			uniqHeader, ok := uniqHeadersByHash[header.Hash]
+			if ok {
+				if len(header.Erc20Transfers) > 0 {
+					uniqHeader.Erc20Transfers = append(uniqHeader.Erc20Transfers, header.Erc20Transfers...)
+				}
+				uniqHeadersByHash[header.Hash] = uniqHeader
+			} else {
+				uniqHeadersByHash[header.Hash] = header
+			}
+		}
+
+		uniqHeaders := []*DBHeader{}
+		for _, header := range uniqHeadersByHash {
+			uniqHeaders = append(uniqHeaders, header)
+		}
+
+		foundHeaders[address] = uniqHeaders
 
 		for _, header := range allHeaders {
 			if header.Number.Cmp(maxBlockNumber) == 1 {
@@ -1000,8 +1018,8 @@ func (c *findAndCheckBlockRangeCommand) Run(parent context.Context) (err error)
 			}
 		}
 
-		log.Debug("saving headers", "len", len(allHeaders), "address")
-		err = c.db.ProcessBlocks(address, newFromByAddress[address], c.toByAddress[address], allHeaders)
+		log.Debug("saving headers", "len", len(uniqHeaders), "address")
+		err = c.db.ProcessBlocks(address, newFromByAddress[address], c.toByAddress[address], uniqHeaders)
 		if err != nil {
 			return err
 		}

services/wallet/database.go

@@ -18,7 +18,7 @@ type DBHeader struct {
 	Number         *big.Int
 	Hash           common.Hash
 	Timestamp      uint64
-	Erc20Transfer *Transfer
+	Erc20Transfers []*Transfer
 	Network        uint64
 	Address        common.Address
 	// Head is true if the block was a head at the time it was pulled from chain.
@@ -765,8 +765,9 @@ func insertBlocksWithTransactions(creator statementCreator, account common.Addre
 		if err != nil {
 			return err
 		}
-		if header.Erc20Transfer != nil {
-			res, err := updateTx.Exec(&JSONBlob{header.Erc20Transfer.Log}, network, account, header.Erc20Transfer.ID)
+		if len(header.Erc20Transfers) > 0 {
+			for _, transfer := range header.Erc20Transfers {
+				res, err := updateTx.Exec(&JSONBlob{transfer.Log}, network, account, transfer.ID)
 				if err != nil {
 					return err
 				}
@@ -778,13 +779,14 @@ func insertBlocksWithTransactions(creator statementCreator, account common.Addre
 					continue
 				}
 
-			_, err = insertTx.Exec(network, account, account, header.Erc20Transfer.ID, (*SQLBigInt)(header.Number), header.Hash, erc20Transfer, header.Erc20Transfer.Timestamp, &JSONBlob{header.Erc20Transfer.Log})
+				_, err = insertTx.Exec(network, account, account, transfer.ID, (*SQLBigInt)(header.Number), header.Hash, erc20Transfer, transfer.Timestamp, &JSONBlob{transfer.Log})
 				if err != nil {
 					log.Error("error saving erc20transfer", "err", err)
 					return err
 				}
 			}
 		}
+	}
 	return nil
 }
 

services/wallet/downloader.go

@@ -296,7 +296,7 @@ func (d *ERC20TransfersDownloader) blocksFromLogs(parent context.Context, logs [
 		header := &DBHeader{
 			Number: big.NewInt(int64(l.BlockNumber)),
 			Hash:   l.BlockHash,
-			Erc20Transfer: &Transfer{
+			Erc20Transfers: []*Transfer{{
 				Address:     address,
 				BlockNumber: big.NewInt(int64(l.BlockNumber)),
 				BlockHash:   l.BlockHash,
@@ -304,9 +304,7 @@ func (d *ERC20TransfersDownloader) blocksFromLogs(parent context.Context, logs [
 				From:        address,
 				Loaded:      false,
 				Type:        erc20Transfer,
-				Log:         &l,
-			},
-		}
+				Log:         &l}}}
 
 		concurrent.Add(func(ctx context.Context) error {
 			concurrent.PushHeader(header)