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

357 lines
9.5 KiB
Go

package localnotifications
import (
"database/sql"
"math/big"
"sync"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/rpc"
"github.com/status-im/status-go/eth-node/types"
"github.com/status-im/status-go/multiaccounts/accounts"
"github.com/status-im/status-go/services/wallet"
"github.com/status-im/status-go/signal"
)
type PushCategory string
type transactionState string
const walletDeeplinkPrefix = "status-im://wallet/"
const (
failed transactionState = "failed"
inbound transactionState = "inbound"
outbound transactionState = "outbound"
)
type notificationBody 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"`
}
type Notification struct {
ID common.Hash `json:"id"`
Platform float32 `json:"platform,omitempty"`
Body notificationBody `json:"body"`
Category PushCategory `json:"category,omitempty"`
Deeplink string `json:"deepLink,omitempty"`
Image string `json:"imageUrl,omitempty"`
IsScheduled bool `json:"isScheduled,omitempty"`
ScheduledTime string `json:"scheduleTime,omitempty"`
}
// 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"`
NewTransactionsPerAccount map[common.Address]int `json:"new-transactions"`
ERC20 bool `json:"erc20"`
MaxKnownBlocks map[common.Address]*big.Int `json:"max-known-blocks"`
}
// MessageEvent - structure used to pass messages from chat to bus
type MessageEvent struct{}
// CustomEvent - structure used to pass custom user set messages to bus
type CustomEvent struct{}
type transmitter struct {
publisher *event.Feed
wg sync.WaitGroup
quit chan struct{}
}
// Service keeps the state of message bus
type Service struct {
started bool
transmitter *transmitter
walletTransmitter *transmitter
db *Database
walletDB *wallet.Database
accountsDB *accounts.Database
}
func NewService(appDB *sql.DB, network uint64) *Service {
db := NewDB(appDB, network)
walletDB := wallet.NewDB(appDB, network)
accountsDB := accounts.NewDB(appDB)
trans := &transmitter{}
walletTrans := &transmitter{}
return &Service{
db: db,
walletDB: walletDB,
accountsDB: accountsDB,
transmitter: trans,
walletTransmitter: walletTrans,
}
}
func pushMessage(notification *Notification) {
log.Info("Pushing a new push notification", "info", notification)
signal.SendLocalNotifications(notification)
}
func (s *Service) buildTransactionNotification(rawTransfer wallet.Transfer) *Notification {
log.Info("Handled a new transfer in buildTransactionNotification", "info", rawTransfer)
var deeplink string
var state transactionState
transfer := wallet.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 {
log.Debug("Could not select From account by address", "error", err)
}
to, err := s.accountsDB.GetAccountByAddress(types.Address(transfer.To))
if err != nil {
log.Debug("Could not select To account by address", "error", err)
}
if from != nil {
deeplink = walletDeeplinkPrefix + from.Address.String()
} else if to != nil {
deeplink = walletDeeplinkPrefix + to.Address.String()
}
body := notificationBody{
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{
ID: transfer.ID,
Body: body,
Deeplink: deeplink,
Category: "transaction",
}
}
func (s *Service) transactionsHandler(payload TransactionEvent) {
log.Info("Handled a new transaction", "info", payload)
limit := 20
if payload.BlockNumber != nil {
for _, address := range payload.Accounts {
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 {
log.Error("Could not fetch transfers", "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 {
log.Error("Failed to get wallet preference", "error", err)
return nil
}
if preference.Enabled {
s.StartWalletWatcher()
}
return nil
}
// 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 {
log.Error("wallet publisher was not initialized")
return
}
s.walletTransmitter.quit = make(chan struct{})
events := make(chan wallet.Event, 10)
sub := s.walletTransmitter.publisher.Subscribe(events)
s.walletTransmitter.wg.Add(1)
maxKnownBlocks := map[common.Address]*big.Int{}
go func() {
defer s.walletTransmitter.wg.Done()
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 {
log.Error("wallet signals transmitter failed with", "error", err)
}
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,
Accounts: event.Accounts,
NewTransactionsPerAccount: event.NewTransactionsPerAccount,
ERC20: event.ERC20,
MaxKnownBlocks: maxKnownBlocks,
})
}
} else if event.Type == wallet.EventMaxKnownBlock {
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
}
// Start Worker which processes all incoming messages
func (s *Service) Start(_ *p2p.Server) error {
s.started = true
s.transmitter.quit = make(chan struct{})
s.transmitter.publisher = &event.Feed{}
events := make(chan TransactionEvent, 10)
sub := s.transmitter.publisher.Subscribe(events)
s.transmitter.wg.Add(1)
go func() {
defer s.transmitter.wg.Done()
for {
select {
case <-s.transmitter.quit:
sub.Unsubscribe()
return
case err := <-sub.Err():
if err != nil {
log.Error("Local notifications transmitter failed with", "error", err)
}
return
case event := <-events:
s.transactionsHandler(event)
}
}
}()
log.Info("Successful start")
return nil
}
// Stop worker
func (s *Service) Stop() error {
s.started = false
if s.transmitter.quit != nil {
close(s.transmitter.quit)
s.transmitter.wg.Wait()
s.transmitter.quit = nil
}
if s.walletTransmitter.quit != nil {
close(s.walletTransmitter.quit)
s.walletTransmitter.wg.Wait()
s.walletTransmitter.quit = nil
}
return nil
}
// APIs returns list of available RPC APIs.
func (s *Service) APIs() []rpc.API {
return []rpc.API{
{
Namespace: "localnotifications",
Version: "0.1.0",
Service: NewAPI(s),
Public: true,
},
}
}
// Protocols returns list of p2p protocols.
func (s *Service) Protocols() []p2p.Protocol {
return nil
}
func (s *Service) IsStarted() bool {
return s.started
}