frostfs-node/pkg/innerring/innerring.go

package innerring

import (
	"context"
	"crypto/ecdsa"

	"github.com/nspcc-dev/neo-go/pkg/config/netmode"
	"github.com/nspcc-dev/neo-go/pkg/util"
	crypto "github.com/nspcc-dev/neofs-crypto"
	"github.com/nspcc-dev/neofs-node/pkg/innerring/invoke"
	"github.com/nspcc-dev/neofs-node/pkg/innerring/processors/balance"
	"github.com/nspcc-dev/neofs-node/pkg/innerring/processors/container"
	"github.com/nspcc-dev/neofs-node/pkg/innerring/processors/neofs"
	"github.com/nspcc-dev/neofs-node/pkg/innerring/processors/netmap"
	"github.com/nspcc-dev/neofs-node/pkg/innerring/timers"
	"github.com/nspcc-dev/neofs-node/pkg/morph/client"
	"github.com/nspcc-dev/neofs-node/pkg/morph/event"
	"github.com/nspcc-dev/neofs-node/pkg/morph/subscriber"
	"github.com/pkg/errors"
	"github.com/spf13/viper"
	"go.uber.org/atomic"
	"go.uber.org/zap"
)

type (
	// Server is the inner ring application structure, that contains all event
	// processors, shared variables and event handlers.
	Server struct {
		log *zap.Logger

		// event producers
		morphListener   event.Listener
		mainnetListener event.Listener
		localTimers     *timers.Timers

		// global state
		morphClient   *client.Client
		mainnetClient *client.Client
		epochCounter  atomic.Uint64
		activeState   atomic.Bool

		// todo: export error channel
	}

	contracts struct {
		neofs      util.Uint160 // in mainnet
		netmap     util.Uint160 // in morph
		balance    util.Uint160 // in morph
		container  util.Uint160 // in morph
		audit      util.Uint160 // in morph
		reputation util.Uint160 // in morph
		neofsid    util.Uint160 // in morph
		gas        util.Uint160 // native contract in both chains
	}

	chainParams struct {
		log  *zap.Logger
		cfg  *viper.Viper
		key  *ecdsa.PrivateKey
		name string
		gas  util.Uint160
	}
)

const (
	morphPrefix   = "morph"
	mainnetPrefix = "mainnet"
)

// Start runs all event providers.
func (s *Server) Start(ctx context.Context) error {
	s.localTimers.Start(ctx) // local timers start ticking

	go s.morphListener.Listen(ctx)   // listen for neo:morph events
	go s.mainnetListener.Listen(ctx) // listen for neo:mainnet events

	return nil
}

// Stop closes all subscription channels.
func (s *Server) Stop() {
	go s.morphListener.Stop()
	go s.mainnetListener.Stop()
}

// New creates instance of inner ring sever structure.
func New(ctx context.Context, log *zap.Logger, cfg *viper.Viper) (*Server, error) {
	server := &Server{log: log}

	// prepare inner ring node private key
	key, err := crypto.LoadPrivateKey(cfg.GetString("key"))
	if err != nil {
		return nil, errors.Wrap(err, "ir: can't create private key")
	}

	// get all script hashes of contracts
	contracts, err := parseContracts(cfg)
	if err != nil {
		return nil, err
	}

	// create local timer instance
	server.localTimers = timers.New(&timers.Params{
		Log:           log,
		EpochDuration: cfg.GetDuration("timers.epoch"),
	})

	morphChain := &chainParams{
		log:  log,
		cfg:  cfg,
		key:  key,
		gas:  contracts.gas,
		name: morphPrefix,
	}

	// create morph listener
	server.morphListener, err = createListener(ctx, morphChain)
	if err != nil {
		return nil, err
	}

	// create morph client
	server.morphClient, err = createClient(ctx, morphChain)
	if err != nil {
		return nil, err
	}

	mainnetChain := morphChain
	mainnetChain.name = mainnetPrefix

	// create mainnet listener
	server.mainnetListener, err = createListener(ctx, mainnetChain)
	if err != nil {
		return nil, err
	}

	// create mainnet client
	server.mainnetClient, err = createClient(ctx, mainnetChain)
	if err != nil {
		return nil, err
	}

	// create netmap processor
	netmapProcessor, err := netmap.New(&netmap.Params{
		Log:            log,
		PoolSize:       cfg.GetInt("workers.netmap"),
		NetmapContract: contracts.netmap,
		EpochTimer:     server.localTimers,
		MorphClient:    server.morphClient,
		EpochState:     server,
		ActiveState:    server,
	})
	if err != nil {
		return nil, err
	}

	err = bindMorphProcessor(netmapProcessor, server)
	if err != nil {
		return nil, err
	}

	// container processor
	containerProcessor, err := container.New(&container.Params{
		Log:               log,
		PoolSize:          cfg.GetInt("workers.container"),
		ContainerContract: contracts.container,
		MorphClient:       server.morphClient,
		ActiveState:       server,
	})
	if err != nil {
		return nil, err
	}

	err = bindMorphProcessor(containerProcessor, server)
	if err != nil {
		return nil, err
	}

	// create balance processor
	balanceProcessor, err := balance.New(&balance.Params{
		Log:             log,
		PoolSize:        cfg.GetInt("workers.balance"),
		NeoFSContract:   contracts.neofs,
		BalanceContract: contracts.balance,
		MainnetClient:   server.mainnetClient,
		ActiveState:     server,
	})
	if err != nil {
		return nil, err
	}

	err = bindMorphProcessor(balanceProcessor, server)
	if err != nil {
		return nil, err
	}

	// todo: create reputation processor

	// create mainnnet neofs processor
	neofsProcessor, err := neofs.New(&neofs.Params{
		Log:             log,
		PoolSize:        cfg.GetInt("workers.neofs"),
		NeoFSContract:   contracts.neofs,
		BalanceContract: contracts.balance,
		NetmapContract:  contracts.netmap,
		MorphClient:     server.morphClient,
		EpochState:      server,
		ActiveState:     server,
	})
	if err != nil {
		return nil, err
	}

	err = bindMainnetProcessor(neofsProcessor, server)
	if err != nil {
		return nil, err
	}

	// todo: create vivid id component
	// todo: create audit scheduler

	err = initConfigFromBlockchain(server, contracts, &key.PublicKey)
	if err != nil {
		return nil, errors.Wrap(err, "initializing error")
	}

	return server, nil
}

func createListener(ctx context.Context, p *chainParams) (event.Listener, error) {
	sub, err := subscriber.New(ctx, &subscriber.Params{
		Log:         p.log,
		Endpoint:    p.cfg.GetString(p.name + ".endpoint.notification"),
		DialTimeout: p.cfg.GetDuration(p.name + ".dial_timeouts"),
	})
	if err != nil {
		return nil, err
	}

	listener, err := event.NewListener(event.ListenerParams{
		Logger:     p.log,
		Subscriber: sub,
	})
	if err != nil {
		return nil, err
	}

	return listener, err
}

func createClient(ctx context.Context, p *chainParams) (*client.Client, error) {
	return client.New(
		p.key,
		p.cfg.GetString(p.name+".endpoint.client"),
		client.WithContext(ctx),
		client.WithLogger(p.log),
		client.WithDialTimeout(p.cfg.GetDuration(p.name+".dial_timeouts")),
		client.WithMagic(netmode.Magic(p.cfg.GetUint32(p.name+".magic_number"))),
		client.WithGasContract(p.gas),
	)
}

func parseContracts(cfg *viper.Viper) (*contracts, error) {
	var (
		result = new(contracts)
		err    error
	)

	netmapContractStr := cfg.GetString("contracts.netmap")
	neofsContractStr := cfg.GetString("contracts.neofs")
	balanceContractStr := cfg.GetString("contracts.balance")
	nativeGasContractStr := cfg.GetString("contracts.gas")
	containerContractStr := cfg.GetString("contracts.container")

	result.netmap, err = util.Uint160DecodeStringLE(netmapContractStr)
	if err != nil {
		return nil, errors.Wrap(err, "ir: can't read netmap script-hash")
	}

	result.neofs, err = util.Uint160DecodeStringLE(neofsContractStr)
	if err != nil {
		return nil, errors.Wrap(err, "ir: can't read neofs script-hash")
	}

	result.balance, err = util.Uint160DecodeStringLE(balanceContractStr)
	if err != nil {
		return nil, errors.Wrap(err, "ir: can't read balance script-hash")
	}

	result.gas, err = util.Uint160DecodeStringLE(nativeGasContractStr)
	if err != nil {
		return nil, errors.Wrap(err, "ir: can't read native gas script-hash")
	}

	result.container, err = util.Uint160DecodeStringLE(containerContractStr)
	if err != nil {
		return nil, errors.Wrap(err, "ir: can't read container script-hash")
	}

	return result, nil
}

func initConfigFromBlockchain(s *Server, c *contracts, key *ecdsa.PublicKey) error {
	// get current epoch
	epoch, err := invoke.Epoch(s.morphClient, c.netmap)
	if err != nil {
		return errors.Wrap(err, "can't read epoch")
	}

	// check if node inside inner ring list
	state, err := invoke.IsInnerRing(s.mainnetClient, c.neofs, key)
	if err != nil {
		return errors.Wrap(err, "can't read inner ring list")
	}

	s.epochCounter.Store(uint64(epoch))
	s.activeState.Store(state)

	s.log.Debug("read config from blockchain",
		zap.Bool("active", state),
		zap.Int64("epoch", epoch),
	)

	return nil
}