status-go/services/local-notifications/transaction.go

package localnotifications

import (
	"encoding/json"
	"math/big"

	"go.uber.org/zap"

	"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/eth-node/types"
	"github.com/status-im/status-go/logutils"
	"github.com/status-im/status-go/multiaccounts/accounts"
	"github.com/status-im/status-go/services/wallet/transfer"
	"github.com/status-im/status-go/services/wallet/walletevent"
)

type transactionState string

const (
	walletDeeplinkPrefix = "status-app://wallet/"

	failed   transactionState = "failed"
	inbound  transactionState = "inbound"
	outbound transactionState = "outbound"
)

// TransactionEvent - structure used to pass messages from wallet to bus
type TransactionEvent struct {
	Type           string                      `json:"type"`
	BlockNumber    *big.Int                    `json:"block-number"`
	Accounts       []common.Address            `json:"accounts"`
	MaxKnownBlocks map[common.Address]*big.Int `json:"max-known-blocks"`
}

type transactionBody struct {
	State       transactionState  `json:"state"`
	From        common.Address    `json:"from"`
	To          common.Address    `json:"to"`
	FromAccount *accounts.Account `json:"fromAccount,omitempty"`
	ToAccount   *accounts.Account `json:"toAccount,omitempty"`
	Value       *hexutil.Big      `json:"value"`
	ERC20       bool              `json:"erc20"`
	Contract    common.Address    `json:"contract"`
	Network     uint64            `json:"network"`
}

func (t transactionBody) MarshalJSON() ([]byte, error) {
	type Alias transactionBody
	item := struct{ *Alias }{Alias: (*Alias)(&t)}
	return json.Marshal(item)
}

func (s *Service) buildTransactionNotification(rawTransfer transfer.Transfer) *Notification {
	logutils.ZapLogger().Debug("Handled a new transfer in buildTransactionNotification", zap.Any("info", rawTransfer))

	var deeplink string
	var state transactionState
	transfer := transfer.CastToTransferView(rawTransfer)

	switch {
	case transfer.TxStatus == hexutil.Uint64(0):
		state = failed
	case transfer.Address == transfer.To:
		state = inbound
	default:
		state = outbound
	}

	from, err := s.accountsDB.GetAccountByAddress(types.Address(transfer.From))

	if err != nil {
		logutils.ZapLogger().Debug("Could not select From account by address", zap.Error(err))
	}

	to, err := s.accountsDB.GetAccountByAddress(types.Address(transfer.To))

	if err != nil {
		logutils.ZapLogger().Debug("Could not select To account by address", zap.Error(err))
	}

	if from != nil {
		deeplink = walletDeeplinkPrefix + from.Address.String()
	} else if to != nil {
		deeplink = walletDeeplinkPrefix + to.Address.String()
	}

	body := transactionBody{
		State:       state,
		From:        transfer.From,
		To:          transfer.Address,
		FromAccount: from,
		ToAccount:   to,
		Value:       transfer.Value,
		ERC20:       string(transfer.Type) == "erc20",
		Contract:    transfer.Contract,
		Network:     transfer.NetworkID,
	}

	return &Notification{
		BodyType: TypeTransaction,
		ID:       transfer.ID,
		Body:     body,
		Deeplink: deeplink,
		Category: CategoryTransaction,
	}
}

func (s *Service) transactionsHandler(payload TransactionEvent) {
	logutils.ZapLogger().Info("Handled a new transaction", zap.Any("info", payload))

	limit := 20
	if payload.BlockNumber != nil {
		for _, address := range payload.Accounts {
			if payload.BlockNumber.Cmp(payload.MaxKnownBlocks[address]) >= 0 {
				logutils.ZapLogger().Info("Handled transfer for address", zap.Stringer("info", address))
				transfers, err := s.walletDB.GetTransfersByAddressAndBlock(s.chainID, address, payload.BlockNumber, int64(limit))
				if err != nil {
					logutils.ZapLogger().Error("Could not fetch transfers", zap.Error(err))
				}

				for _, transaction := range transfers {
					n := s.buildTransactionNotification(transaction)
					pushMessage(n)
				}
			}
		}
	}
}

// SubscribeWallet - Subscribes to wallet signals
func (s *Service) SubscribeWallet(publisher *event.Feed) error {
	s.walletTransmitter.publisher = publisher

	preference, err := s.db.GetWalletPreference()

	if err != nil {
		logutils.ZapLogger().Error("Failed to get wallet preference", zap.Error(err))
		s.WatchingEnabled = false
	} else {
		s.WatchingEnabled = preference.Enabled
	}

	s.StartWalletWatcher()

	return err
}

// StartWalletWatcher - Forward wallet events to notifications
func (s *Service) StartWalletWatcher() {
	if s.walletTransmitter.quit != nil {
		// already running, nothing to do
		return
	}

	if s.walletTransmitter.publisher == nil {
		logutils.ZapLogger().Error("wallet publisher was not initialized")
		return
	}

	s.walletTransmitter.quit = make(chan struct{})
	events := make(chan walletevent.Event, 10)
	sub := s.walletTransmitter.publisher.Subscribe(events)

	s.walletTransmitter.wg.Add(1)

	maxKnownBlocks := map[common.Address]*big.Int{}
	go func() {
		defer gocommon.LogOnPanic()
		defer s.walletTransmitter.wg.Done()
		historyReady := false
		for {
			select {
			case <-s.walletTransmitter.quit:
				sub.Unsubscribe()
				return
			case err := <-sub.Err():
				// technically event.Feed cannot send an error to subscription.Err channel.
				// the only time we will get an event is when that channel is closed.
				if err != nil {
					logutils.ZapLogger().Error("wallet signals transmitter failed with", zap.Error(err))
				}
				return
			case event := <-events:
				if event.Type == transfer.EventNewTransfers && historyReady && event.BlockNumber != nil {
					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.WatchingEnabled {
						s.transmitter.publisher.Send(TransactionEvent{
							Type:           string(event.Type),
							BlockNumber:    event.BlockNumber,
							Accounts:       event.Accounts,
							MaxKnownBlocks: maxKnownBlocks,
						})
					}
				} else if event.Type == transfer.EventRecentHistoryReady {
					historyReady = true
					if event.BlockNumber != nil {
						for _, address := range event.Accounts {
							if _, ok := maxKnownBlocks[address]; !ok {
								maxKnownBlocks[address] = event.BlockNumber
							}
						}
					}
				}
			}
		}
	}()
}

// StopWalletWatcher - stops watching for new wallet events
func (s *Service) StopWalletWatcher() {
	if s.walletTransmitter.quit != nil {
		close(s.walletTransmitter.quit)
		s.walletTransmitter.wg.Wait()
		s.walletTransmitter.quit = nil
	}
}

// IsWatchingWallet - check if local-notifications are subscribed to wallet updates
func (s *Service) IsWatchingWallet() bool {
	return s.walletTransmitter.quit != nil
}