status-go/pkg/backend/node/get_status_node.go

package node

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"reflect"
	"sync"
	"sync/atomic"
	"time"

	errorspkg "github.com/pkg/errors"

	"go.uber.org/zap"

	"github.com/ethereum/go-ethereum/event"
	gethrpc "github.com/ethereum/go-ethereum/rpc"

	accsmanagement "github.com/status-im/status-go/internal/accounts-management"
	"github.com/status-im/status-go/internal/connection"
	"github.com/status-im/status-go/internal/crypto"
	"github.com/status-im/status-go/internal/db/multiaccounts"
	"github.com/status-im/status-go/internal/db/multiaccounts/accounts"
	"github.com/status-im/status-go/internal/ipfs"
	"github.com/status-im/status-go/internal/panics"
	"github.com/status-im/status-go/internal/pausable"
	"github.com/status-im/status-go/internal/rpc"
	"github.com/status-im/status-go/internal/timesource"
	"github.com/status-im/status-go/internal/transactions"
	"github.com/status-im/status-go/params"
	noderpc "github.com/status-im/status-go/pkg/backend/node/rpc"
	"github.com/status-im/status-go/pkg/pubsub"
	accountssvc "github.com/status-im/status-go/pkg/services/accounts"
	appgeneral "github.com/status-im/status-go/pkg/services/app-general"
	"github.com/status-im/status-go/pkg/services/backup"
	"github.com/status-im/status-go/pkg/services/browsers"
	"github.com/status-im/status-go/pkg/services/chat"
	"github.com/status-im/status-go/pkg/services/communitytokens"
	"github.com/status-im/status-go/pkg/services/connector"
	"github.com/status-im/status-go/pkg/services/ens"
	"github.com/status-im/status-go/pkg/services/eth"
	"github.com/status-im/status-go/pkg/services/gif"
	"github.com/status-im/status-go/pkg/services/linkpreview"
	localnotifications "github.com/status-im/status-go/pkg/services/local-notifications"
	"github.com/status-im/status-go/pkg/services/media"
	"github.com/status-im/status-go/pkg/services/networks"
	"github.com/status-im/status-go/pkg/services/newsfeed"
	"github.com/status-im/status-go/pkg/services/permissions"
	"github.com/status-im/status-go/pkg/services/personal"
	"github.com/status-im/status-go/pkg/services/preferences"
	"github.com/status-im/status-go/pkg/services/rpcstats"
	"github.com/status-im/status-go/pkg/services/sharedurls"
	"github.com/status-im/status-go/pkg/services/stickers"
	"github.com/status-im/status-go/pkg/services/storagestats"
	"github.com/status-im/status-go/pkg/services/updates"
	"github.com/status-im/status-go/pkg/services/utils"
	"github.com/status-im/status-go/pkg/services/wakuv2ext"
	"github.com/status-im/status-go/pkg/services/wallet"
	"github.com/status-im/status-go/pkg/services/wallet/community"
	"github.com/status-im/status-go/pkg/services/wallet/pendingtxtracker"
	"github.com/status-im/status-go/pkg/services/wallet/token"
	"github.com/status-im/status-go/pkg/services/wallet/tokenbalances"
)

// errors
var (
	ErrNodeRunning   = errors.New("node is already running")
	ErrNoRunningNode = errors.New("there is no running node")
)

// StatusNode abstracts contained geth node and provides helper methods to
// interact with it.
type StatusNode struct {
	mu sync.RWMutex

	appDB           *sql.DB
	multiaccountsDB *multiaccounts.Database
	walletDB        *sql.DB

	running atomic.Bool

	config         *params.NodeConfig // Status node configuration
	rpcClient      *rpc.Client        // reference to an RPC client
	networkManager *networks.Manager

	services  []StatusService
	rpcServer *gethrpc.Server

	downloader *ipfs.Downloader

	mediaServerAddress       *string
	mediaServerAdvertizeHost string
	mediaServerAdvertizePort int
	mediaServerEnableTLS     *bool
	mediaServer              *media.Server

	tokenManager *token.Manager

	logger *zap.Logger

	gethAccountsManager *accsmanagement.AccountsManager
	transactor          *transactions.Transactor

	publicMethods map[string]bool
	// we explicitly list every service, we could use interfaces
	// and store them in a nicer way and user reflection, but for now stupid is good
	rpcStatsSrvc           *rpcstats.Service
	accountsSrvc           *accountssvc.Service
	browsersSrvc           *browsers.Service
	permissionsSrvc        *permissions.Service
	walletSrvc             *wallet.Service
	networksSrvc           *networks.Service
	localNotificationsSrvc *localnotifications.Service
	personalSrvc           *personal.Service
	timeSourceSrvc         timesource.Service
	wakuV2ExtSrvc          *wakuv2ext.Service
	ensSrvc                *ens.Service
	communityTokensSrvc    *communitytokens.Service
	gifSrvc                *gif.Service
	stickersSrvc           *stickers.Service
	chatSrvc               *chat.Service
	updatesSrvc            *updates.Service
	pendingTracker         *pendingtxtracker.PendingTxTracker
	connectorSrvc          *connector.Service
	appGeneralSrvc         *appgeneral.Service
	ethSrvc                *eth.Service
	newsfeedSrvc           *newsfeed.Service
	preferencesSrvc        *preferences.Service
	sharedUrlsSrvc         *sharedurls.Service
	storageStatsSrvc       *storagestats.Service
	linkPreviewSrvc        *linkpreview.Service

	walletFeed        event.Feed
	accountsPublisher *pubsub.Publisher

	localBackup *backup.Controller

	tokenManagerStartDone chan struct{}

	serviceRegistry *ServiceRegistry
}

// ServiceRegistry returns the node's service registry for granular pause/resume control.
func (n *StatusNode) ServiceRegistry() *ServiceRegistry {
	return n.serviceRegistry
}

// New makes new instance of StatusNode.
func New(transactor *transactions.Transactor, gethAccountsManager *accsmanagement.AccountsManager, logger *zap.Logger) *StatusNode {
	logger = logger.Named("StatusNode")
	return &StatusNode{
		transactor:          transactor,
		gethAccountsManager: gethAccountsManager,
		logger:              logger,
		publicMethods:       make(map[string]bool),
		accountsPublisher:   pubsub.NewPublisher(),
		rpcServer:           gethrpc.NewServer(),
		serviceRegistry:     newServiceRegistry(),
	}
}

// Config exposes reference to running node's configuration
func (n *StatusNode) Config() *params.NodeConfig {
	n.mu.RLock()
	defer n.mu.RUnlock()

	return n.config
}

func (n *StatusNode) MediaServer() *media.Server {
	n.mu.RLock()
	defer n.mu.RUnlock()

	return n.mediaServer
}

func (n *StatusNode) startMediaServer() error {
	if n.mediaServer != nil {
		if err := n.mediaServer.Stop(); err != nil {
			return err
		}
	}

	var opts []media.MediaServerOption
	if n.mediaServerEnableTLS != nil {
		opts = append(opts, media.WithMediaServerDisableTLS(!*n.mediaServerEnableTLS))
	}
	if n.mediaServerAddress != nil {
		opts = append(opts, media.WithMediaServerAddress(*n.mediaServerAddress))
	}
	opts = append(opts, media.WithMediaServerAdvertizeAddress(n.mediaServerAdvertizeHost, n.mediaServerAdvertizePort))
	mediaServer, err := media.NewServer(nil, nil, n.multiaccountsDB, nil, opts...)
	if err != nil {
		return err
	}

	n.mediaServer = mediaServer

	if err := n.mediaServer.Start(); err != nil {
		return err
	}

	return nil
}

// StartMediaServerWithoutDB starts media server without starting the node
// The server can only handle requests that don't require appdb or IPFS downloader
func (n *StatusNode) StartMediaServerWithoutDB() error {
	if n.IsRunning() {
		n.logger.Debug("node is already running, no need to StartMediaServerWithoutDB")
		return nil
	}

	return n.startMediaServer()
}

// StartWithOptions starts current StatusNode, failing if it's already started.
// It takes some options that allows to further configure starting process.
func (n *StatusNode) Start(config *params.NodeConfig) error {
	n.mu.Lock()
	defer n.mu.Unlock()

	if !n.running.CompareAndSwap(false, true) {
		n.logger.Debug("node is already running")
		return ErrNodeRunning
	}

	n.logger.Debug("starting with options", zap.Stringer("ClusterConfig", &config.ClusterConfig))

	return n.startWithDB(config)
}

func (n *StatusNode) StartLocalBackup() error {
	if n.localBackup != nil {
		return errors.New("local backup already started")
	}

	backupPath, err := n.accountsSrvc.GetBackupPath()
	if err != nil {
		return err
	}
	if backupPath == "" {
		// No path set yet, set it to the user's config directory
		dir, err := os.UserConfigDir()
		// We do not return the error as it's not a major issue
		if err != nil {
			n.logger.Error("failed to get user config dir", zap.Error(err))
		} else {
			err = n.accountsSrvc.SetBackupPath(filepath.Join(dir, "Status", "backups"))
			if err != nil {
				n.logger.Error("failed to set backup path", zap.Error(err))
			}
		}
	}

	chatAccount, err := n.gethAccountsManager.SelectedChatAccount()
	if err != nil {
		return err
	}

	privateKey := chatAccount.PrivateKey()

	n.localBackup, err = backup.NewController(backup.Config{
		PrivateKey:       crypto.Keccak256(crypto.FromECDSA(privateKey)),
		FileNameProvider: n,
		BackupEnabled:    true,
		Interval:         time.Minute * 30,
	}, n.logger.Named("LocalBackup"))
	if err != nil {
		return err
	}

	if n.accountsSrvc != nil {
		n.localBackup.Register("settings", n.accountsSrvc)
	}

	if n.walletSrvc != nil {
		n.localBackup.Register("wallet", n.walletSrvc)
	}

	if n.wakuV2ExtSrvc != nil {
		n.localBackup.Register("messenger", n.wakuV2ExtSrvc.Messenger())
	}

	n.serviceRegistry.Register(n.localBackup)
	n.localBackup.Start()

	return nil
}

func (n *StatusNode) GetBackupFilename() (string, error) {
	chatAccount, err := n.gethAccountsManager.SelectedChatAccount()
	if err != nil {
		return "", err
	}

	privateKey := chatAccount.PrivateKey()

	backupPath, err := n.accountsSrvc.GetBackupPath()
	if err != nil {
		return "", err
	}

	compressedPubKey, err := utils.SerializePublicKey(crypto.CompressPubkey(&privateKey.PublicKey))
	if err != nil {
		return "", err
	}

	if backupPath == "" {
		return "", errors.New("backup path is not set")
	}

	fullPath := filepath.Join(backupPath, fmt.Sprintf("%s_user_data.bkp", compressedPubKey[len(compressedPubKey)-6:]))

	return fullPath, nil
}

func (n *StatusNode) PerformLocalBackup() (string, error) {
	return n.localBackup.PerformBackup()
}

func (n *StatusNode) LoadLocalBackup(filePath string) error {
	return n.localBackup.LoadBackup(filePath)
}

func (n *StatusNode) SetMediaServerOptions(address *string, enableTLS *bool, advertizeHost string, advertizePort int) {
	n.mu.Lock()
	defer n.mu.Unlock()

	n.mediaServerAddress = address
	n.mediaServerEnableTLS = enableTLS
	n.mediaServerAdvertizeHost = advertizeHost
	n.mediaServerAdvertizePort = advertizePort

	if n.mediaServer != nil {
		if err := n.startMediaServer(); err != nil {
			n.logger.Error("failed to restart media server with updated options", zap.Error(err))
		}
	}
}

func (n *StatusNode) StopMediaServer() error {
	n.mu.Lock()
	defer n.mu.Unlock()

	if n.mediaServer == nil {
		return nil
	}

	err := n.mediaServer.Stop()
	n.mediaServer = nil
	return err
}

func (n *StatusNode) startWithDB(config *params.NodeConfig) error {
	n.config = config

	if err := n.setupRPCClient(); err != nil {
		return err
	}

	n.downloader = ipfs.NewDownloader(config.RootDataDir)

	if n.mediaServer == nil {
		if err := n.startMediaServer(); err != nil {
			return err
		}
	}
	n.mediaServer.SetDataProviders(n.appDB, n.walletDB, n.downloader)

	if err := n.createTokenManager(); err != nil {
		return err
	}
	if err := n.tokenManager.Start(context.Background()); err != nil {
		return errorspkg.Wrap(err, "failed to start token manager")
	}
	n.tokenManagerStartDone = make(chan struct{})
	close(n.tokenManagerStartDone)

	if err := n.initServices(config, n.mediaServer); err != nil {
		return err
	}

	// Run migrations
	err := n.runServicesMigrations()
	if err != nil {
		return errorspkg.Wrap(err, "failed to run services migrations")
	}

	// Register services

	for _, service := range n.services {
		err := n.registerService(service)
		if err != nil {
			name := reflect.TypeOf(service).Name()
			text := fmt.Sprintf("failed to register service '%s'", name)
			return errorspkg.Wrap(err, text)
		}
	}

	// Start services

	err = n.timeSourceSrvc.Start(context.Background())
	if err != nil {
		return errorspkg.Wrap(err, "failed to start time source")
	}

	for _, service := range n.services {
		err := service.Start()
		if err != nil {
			name := reflect.TypeOf(service).Name()
			text := fmt.Sprintf("failed to start service '%s'", name)
			return errorspkg.Wrap(err, text)
		}
	}

	n.populateServiceRegistry()

	return nil
}

// populateServiceRegistry registers all services that implement Pausable and wraps the
// media server as a pausable. Called once after all services are started.
func (n *StatusNode) populateServiceRegistry() {
	// Register services implementing Pausable
	for _, service := range n.services {
		if p, ok := service.(pausable.Pausable); ok {
			n.serviceRegistry.Register(p)
		}
	}
	// Wrap and register the media server
	if n.mediaServer != nil {
		n.serviceRegistry.Register(newPausableMediaServer(n.mediaServer))
	}
	// Wrap and register the messenger so that PauseServices/ResumeServices
	// gate filter health-check pings and mailserver syncs in background.
	if n.wakuV2ExtSrvc != nil {
		if m := n.wakuV2ExtSrvc.Messenger(); m != nil {
			n.serviceRegistry.Register(newPausableMessenger(m))
		}
	}
	// Register infrastructure components that are not go-ethereum services
	if n.downloader != nil {
		n.serviceRegistry.Register(n.downloader)
	}
	if p, ok := n.timeSourceSrvc.(pausable.Pausable); ok {
		n.serviceRegistry.Register(p)
	}
}

func (n *StatusNode) createTokenManager() error {
	const (
		defaultAutoRefreshInterval      = 30 * time.Minute // interval after which we should fetch the token lists from the remote source (or use the default one if remote source is not set)
		defaultAutoRefreshCheckInterval = 3 * time.Minute  // interval after which we should check if we should trigger the auto-refresh
	)

	autoRefreshInterval := defaultAutoRefreshInterval
	autoRefreshCheckInterval := defaultAutoRefreshCheckInterval
	if n.config.WalletConfig.TokensListsAutoRefreshInterval > 0 &&
		n.config.WalletConfig.TokensListsAutoRefreshCheckInterval > 0 &&
		n.config.WalletConfig.TokensListsAutoRefreshInterval > n.config.WalletConfig.TokensListsAutoRefreshCheckInterval {
		autoRefreshInterval = time.Duration(n.config.WalletConfig.TokensListsAutoRefreshInterval) * time.Second
		autoRefreshCheckInterval = time.Duration(n.config.WalletConfig.TokensListsAutoRefreshCheckInterval) * time.Second
	}

	accDB, err := accounts.NewDB(n.appDB)
	if err != nil {
		return err
	}

	n.tokenManager, err = token.NewTokenManager(n.walletDB, n.rpcClient, community.NewManager(n.appDB, n.mediaServer, nil),
		n.rpcClient.GetNetworkManager(), n.appDB, n.mediaServer, &n.walletFeed, n.accountsPublisher, accDB,
		autoRefreshInterval, autoRefreshCheckInterval)
	if err != nil {
		return err
	}

	// check for possible custom tokens in the config
	if len(n.config.WalletConfig.CustomTokens) > 0 {
		for _, token := range n.config.WalletConfig.CustomTokens {
			err := n.tokenManager.UpsertCustom(*token)
			if err != nil {
				return err
			}
		}
	}

	return nil
}

func (n *StatusNode) StartTokenManager() {
	n.mu.Lock()
	if !n.running.Load() || n.tokenManager == nil || n.tokenManagerStartDone != nil {
		n.mu.Unlock()
		return
	}

	tokenManager := n.tokenManager
	done := make(chan struct{})
	n.tokenManagerStartDone = done
	n.mu.Unlock()

	go func() {
		defer panics.LogOnPanic()
		defer close(done)
		if err := tokenManager.Start(context.Background()); err != nil {
			n.logger.Error("failed to start token manager", zap.Error(err))
		}
	}()
}

// NetworkManager returns the network manager owned by the node.
func (n *StatusNode) NetworkManager() *networks.Manager {
	return n.networkManager
}

func (n *StatusNode) setupRPCClient() (err error) {
	networkManager := networks.NewManager(n.appDB, n.accountsPublisher)
	if networkManager == nil {
		return errorspkg.New("failed to create network manager")
	}
	if err = networkManager.InitEmbeddedNetworks(n.config.Networks); err != nil {
		return errorspkg.Wrap(err, "network manager failed to initialize")
	}
	n.networkManager = networkManager

	config := rpc.ClientConfig{
		NetworkManager:    networkManager,
		AccountsPublisher: n.accountsPublisher,
	}
	n.rpcClient, err = rpc.NewClient(config)
	if err != nil {
		return
	}
	n.rpcClient.Start(context.Background())
	return
}

// Stop will stop current StatusNode. A stopped node cannot be resumed.
func (n *StatusNode) Stop() error {
	n.mu.Lock()
	defer n.mu.Unlock()

	n.logger.Debug("stopping")

	if !n.running.CompareAndSwap(true, false) {
		return ErrNoRunningNode
	}

	var errs []error
	if n.tokenManagerStartDone != nil {
		<-n.tokenManagerStartDone
		n.tokenManagerStartDone = nil
		n.tokenManager.Stop()
	}
	n.timeSourceSrvc.Stop()

	for _, service := range n.services {
		err := service.Stop()
		errs = append(errs, err)
	}

	if n.localBackup != nil {
		n.localBackup.Stop()
		n.localBackup = nil
	}

	n.accountsPublisher.Close()

	n.rpcClient.Stop()
	n.rpcClient = nil
	n.config = nil

	if n.mediaServer != nil {
		n.mediaServer.SetDataProviders(nil, nil, nil)
	}

	n.downloader.Stop()
	n.downloader = nil

	n.rpcStatsSrvc = nil
	n.accountsSrvc = nil
	n.browsersSrvc = nil
	n.permissionsSrvc = nil
	n.walletSrvc = nil
	n.networksSrvc = nil
	n.networkManager = nil
	n.localNotificationsSrvc = nil
	n.personalSrvc = nil
	n.timeSourceSrvc = nil
	n.wakuV2ExtSrvc = nil
	n.ensSrvc = nil
	n.communityTokensSrvc = nil
	n.stickersSrvc = nil
	n.connectorSrvc = nil
	n.publicMethods = make(map[string]bool)
	n.pendingTracker = nil
	n.appGeneralSrvc = nil
	n.newsfeedSrvc = nil
	n.preferencesSrvc = nil
	n.storageStatsSrvc = nil

	n.logger.Debug("status node stopped")
	return errors.Join(errs...)
}

// IsRunning confirm that node is running.
func (n *StatusNode) IsRunning() bool {
	return n.running.Load()
}

// Pause reduces non-essential background work while preserving core messaging/Waku runtime.
// Delegates to the ServiceRegistry so callers can also pause individual services via PauseService.
func (n *StatusNode) Pause() error {
	if n.serviceRegistry == nil {
		return nil
	}
	return n.serviceRegistry.PauseAll()
}

// Resume restores regular service work after paused background mode.
// Delegates to the ServiceRegistry so callers can also resume individual services via ResumeService.
func (n *StatusNode) Resume() error {
	if n.serviceRegistry == nil {
		return nil
	}
	return n.serviceRegistry.ResumeAll()
}

func (n *StatusNode) CallInProcessRPC(inputJSON string) string {
	codec := noderpc.NewSingleRequestCodec(inputJSON)
	n.rpcServer.ServeCodec(codec.GethCodec(), 0)
	return codec.Output()
}

// RPCClient exposes reference to RPC client connected to the running node.
func (n *StatusNode) RPCClient() *rpc.Client {
	n.mu.RLock()
	defer n.mu.RUnlock()
	return n.rpcClient
}

func (n *StatusNode) SetAppDB(db *sql.DB) {
	n.appDB = db
}

func (n *StatusNode) GetAppDB() *sql.DB {
	return n.appDB
}

func (n *StatusNode) SetMultiaccountsDB(db *multiaccounts.Database) {
	n.multiaccountsDB = db
}

func (n *StatusNode) SetWalletDB(db *sql.DB) {
	n.walletDB = db
}

func (n *StatusNode) GetWalletDB() *sql.DB {
	return n.walletDB
}

func (n *StatusNode) TokenManager() *token.Manager {
	return n.tokenManager
}

func (n *StatusNode) TokenBalancesFetcher() tokenbalances.FetcherIface {
	if n.walletSrvc != nil {
		return n.walletSrvc.GetTokenBalancesFetcher()
	}
	return nil
}

func (n *StatusNode) TokenBalancesStorage() tokenbalances.Storage {
	if n.walletSrvc != nil {
		return n.walletSrvc.GetTokenBalancesStorage()
	}
	return nil
}

func (n *StatusNode) ConnectionChanged(state connection.State) {
	if n.wakuV2ExtSrvc != nil {
		n.wakuV2ExtSrvc.ConnectionChanged(state)
	}
}