<file_sep>package profiling

import (
	"net/http"
	_ "net/http/pprof"
	"runtime"

	"github.com/pkg/errors"

	"go.uber.org/dig"

	"github.com/gohornet/hornet/pkg/node"
	"github.com/iotaledger/hive.go/configuration"
)

func init() {
	Plugin = &node.Plugin{
		Status: node.StatusEnabled,
		Pluggable: node.Pluggable{
			Name:     "Profiling",
			DepsFunc: func(cDeps dependencies) { deps = cDeps },
			Params:   params,
			Run:      run,
		},
	}
}

var (
	Plugin *node.Plugin
	deps   dependencies
)

type dependencies struct {
	dig.In
	NodeConfig *configuration.Configuration `name:"nodeConfig"`
}

func run() {
	runtime.SetMutexProfileFraction(5)
	runtime.SetBlockProfileRate(5)

	bindAddr := deps.NodeConfig.String(CfgProfilingBindAddress)

	go func() {
		Plugin.LogInfof("You can now access the profiling server using: http://%s/debug/pprof/", bindAddr)

		// pprof Server for Debugging
		if err := http.ListenAndServe(bindAddr, nil); err != nil && !errors.Is(err, http.ErrServerClosed) {
			Plugin.LogWarnf("Stopped profiling server due to an error (%s)", err)
		}
	}()
}
<file_sep>package gracefulshutdown

import (
	"go.uber.org/dig"

	"github.com/gohornet/hornet/pkg/node"
	"github.com/gohornet/hornet/pkg/shutdown"
)

func init() {
	CorePlugin = &node.CorePlugin{
		Pluggable: node.Pluggable{
			Name:      "Graceful Shutdown",
			Provide:   provide,
			DepsFunc:  func(cDeps dependencies) { deps = cDeps },
			Configure: configure,
		},
	}
}

var (
	CorePlugin *node.CorePlugin
	deps       dependencies
)

type dependencies struct {
	dig.In
	ShutdownHandler *shutdown.ShutdownHandler
}

func provide(c *dig.Container) {

	if err := c.Provide(func() *shutdown.ShutdownHandler {
		return shutdown.NewShutdownHandler(CorePlugin.Logger(), CorePlugin.Daemon())
	}); err != nil {
		CorePlugin.Panic(err)
	}
}

func configure() {
	deps.ShutdownHandler.Run()
}
<file_sep>package pow

import (
	"time"

	"go.uber.org/dig"

	"github.com/gohornet/hornet/pkg/node"
	"github.com/gohornet/hornet/pkg/pow"
	"github.com/gohornet/hornet/pkg/shutdown"
	"github.com/gohornet/hornet/pkg/utils"
	"github.com/iotaledger/hive.go/configuration"
)

func init() {
	CorePlugin = &node.CorePlugin{
		Pluggable: node.Pluggable{
			Name:     "PoW",
			DepsFunc: func(cDeps dependencies) { deps = cDeps },
			Params:   params,
			Provide:  provide,
			Run:      run,
		},
	}
}

var (
	CorePlugin *node.CorePlugin
	deps       dependencies
)

const (
	powsrvInitCooldown = 30 * time.Second
)

type dependencies struct {
	dig.In
	Handler *pow.Handler
}

func provide(c *dig.Container) {

	type handlerDeps struct {
		dig.In
		NodeConfig  *configuration.Configuration `name:"nodeConfig"`
		MinPoWScore float64                      `name:"minPoWScore"`
	}

	if err := c.Provide(func(deps handlerDeps) *pow.Handler {
		// init the pow handler with all possible settings
		powsrvAPIKey, err := utils.LoadStringFromEnvironment("POWSRV_API_KEY")
		if err == nil && len(powsrvAPIKey) > 12 {
			powsrvAPIKey = powsrvAPIKey[:12]
		}
		return pow.New(CorePlugin.Logger(), deps.MinPoWScore, deps.NodeConfig.Duration(CfgPoWRefreshTipsInterval), powsrvAPIKey, powsrvInitCooldown)
	}); err != nil {
		CorePlugin.Panic(err)
	}
}

func run() {

	// close the PoW handler on shutdown
	if err := CorePlugin.Daemon().BackgroundWorker("PoW Handler", func(shutdownSignal <-chan struct{}) {
		CorePlugin.LogInfo("Starting PoW Handler ... done")
		<-shutdownSignal
		CorePlugin.LogInfo("Stopping PoW Handler ...")
		deps.Handler.Close()
		CorePlugin.LogInfo("Stopping PoW Handler ... done")
	}, shutdown.PriorityPoWHandler); err != nil {
		CorePlugin.Panicf("failed to start worker: %s", err)
	}
}
<file_sep>// TODO: obviously move all this into its separate pkg
package p2p

import (
	"context"
	"path/filepath"
	"time"

	"github.com/libp2p/go-libp2p"
	connmgr "github.com/libp2p/go-libp2p-connmgr"
	"github.com/libp2p/go-libp2p-core/crypto"
	"github.com/libp2p/go-libp2p-core/host"
	"github.com/libp2p/go-libp2p-core/peer"
	"github.com/multiformats/go-multiaddr"
	"go.uber.org/dig"

	"github.com/gohornet/hornet/pkg/database"
	"github.com/gohornet/hornet/pkg/node"
	"github.com/gohornet/hornet/pkg/p2p"
	"github.com/gohornet/hornet/pkg/shutdown"
	"github.com/iotaledger/hive.go/configuration"
	"github.com/iotaledger/hive.go/logger"
)

func init() {
	CorePlugin = &node.CorePlugin{
		Pluggable: node.Pluggable{
			Name:           "P2P",
			DepsFunc:       func(cDeps dependencies) { deps = cDeps },
			Params:         params,
			InitConfigPars: initConfigPars,
			Provide:        provide,
			Configure:      configure,
			Run:            run,
		},
	}
}

var (
	CorePlugin *node.CorePlugin
	deps       dependencies
)

type dependencies struct {
	dig.In
	Manager              *p2p.Manager
	Host                 host.Host
	NodeConfig           *configuration.Configuration `name:"nodeConfig"`
	PeerStoreContainer   *p2p.PeerStoreContainer
	PeeringConfig        *configuration.Configuration `name:"peeringConfig"`
	PeeringConfigManager *p2p.ConfigManager
}

func initConfigPars(c *dig.Container) {

	type cfgDeps struct {
		dig.In
		NodeConfig *configuration.Configuration `name:"nodeConfig"`
	}

	type cfgResult struct {
		dig.Out
		P2PDatabasePath       string   `name:"p2pDatabasePath"`
		P2PBindMultiAddresses []string `name:"p2pBindMultiAddresses"`
	}

	if err := c.Provide(func(deps cfgDeps) cfgResult {
		return cfgResult{
			P2PDatabasePath:       deps.NodeConfig.String(CfgP2PDatabasePath),
			P2PBindMultiAddresses: deps.NodeConfig.Strings(CfgP2PBindMultiAddresses),
		}
	}); err != nil {
		CorePlugin.Panic(err)
	}
}

func provide(c *dig.Container) {

	type hostDeps struct {
		dig.In
		NodeConfig            *configuration.Configuration `name:"nodeConfig"`
		DatabaseEngine        database.Engine              `name:"databaseEngine"`
		P2PDatabasePath       string                       `name:"p2pDatabasePath"`
		P2PBindMultiAddresses []string                     `name:"p2pBindMultiAddresses"`
	}

	type p2presult struct {
		dig.Out
		PeerStoreContainer *p2p.PeerStoreContainer
		NodePrivateKey     crypto.PrivKey `name:"nodePrivateKey"`
		Host               host.Host
	}

	if err := c.Provide(func(deps hostDeps) p2presult {

		res := p2presult{}

		privKeyFilePath := filepath.Join(deps.P2PDatabasePath, p2p.PrivKeyFileName)

		peerStoreContainer, err := p2p.NewPeerStoreContainer(filepath.Join(deps.P2PDatabasePath, "peers"), deps.DatabaseEngine, true)
		if err != nil {
			CorePlugin.Panic(err)
		}
		res.PeerStoreContainer = peerStoreContainer

		// TODO: temporary migration logic
		// this should be removed after some time / hornet versions (20.08.21: muXxer)
		identityPrivKey := deps.NodeConfig.String(CfgP2PIdentityPrivKey)
		migrated, err := p2p.MigrateDeprecatedPeerStore(deps.P2PDatabasePath, identityPrivKey, peerStoreContainer)
		if err != nil {
			CorePlugin.Panicf("migration of deprecated peer store failed: %s", err)
		}
		if migrated {
			CorePlugin.LogInfof(`The peer store was migrated successfully!

Your node identity private key can now be found at "%s".
`, privKeyFilePath)
		}

		// make sure nobody copies around the peer store since it contains the private key of the node
		CorePlugin.LogInfof(`WARNING: never share your "%s" folder as it contains your node's private key!`, deps.P2PDatabasePath)

		// load up the previously generated identity or create a new one
		privKey, newlyCreated, err := p2p.LoadOrCreateIdentityPrivateKey(deps.P2PDatabasePath, identityPrivKey)
		if err != nil {
			CorePlugin.Panic(err)
		}
		res.NodePrivateKey = privKey

		if newlyCreated {
			CorePlugin.LogInfof(`stored new private key for peer identity under "%s"`, privKeyFilePath)
		} else {
			CorePlugin.LogInfof(`loaded existing private key for peer identity from "%s"`, privKeyFilePath)
		}

		createdHost, err := libp2p.New(context.Background(),
			libp2p.Identity(privKey),
			libp2p.ListenAddrStrings(deps.P2PBindMultiAddresses...),
			libp2p.Peerstore(peerStoreContainer.Peerstore()),
			libp2p.DefaultTransports,
			libp2p.ConnectionManager(connmgr.NewConnManager(
				deps.NodeConfig.Int(CfgP2PConnMngLowWatermark),
				deps.NodeConfig.Int(CfgP2PConnMngHighWatermark),
				time.Minute,
			)),
			libp2p.NATPortMap(),
		)
		if err != nil {
			CorePlugin.Panicf("unable to initialize peer: %s", err)
		}
		res.Host = createdHost

		return res
	}); err != nil {
		CorePlugin.Panic(err)
	}

	type mngDeps struct {
		dig.In
		Host                      host.Host
		Config                    *configuration.Configuration `name:"nodeConfig"`
		AutopeeringRunAsEntryNode bool                         `name:"autopeeringRunAsEntryNode"`
	}

	if err := c.Provide(func(deps mngDeps) *p2p.Manager {
		if !deps.AutopeeringRunAsEntryNode {
			return p2p.NewManager(deps.Host,
				p2p.WithManagerLogger(logger.NewLogger("P2P-Manager")),
				p2p.WithManagerReconnectInterval(deps.Config.Duration(CfgP2PReconnectInterval), 1*time.Second),
			)
		}
		return nil
	}); err != nil {
		CorePlugin.Panic(err)
	}

	type configManagerDeps struct {
		dig.In
		PeeringConfig         *configuration.Configuration `name:"peeringConfig"`
		PeeringConfigFilePath string                       `name:"peeringConfigFilePath"`
	}

	if err := c.Provide(func(deps configManagerDeps) *p2p.ConfigManager {

		p2pConfigManager := p2p.NewConfigManager(func(peers []*p2p.PeerConfig) error {
			if err := deps.PeeringConfig.Set(CfgPeers, peers); err != nil {
				return err
			}

			return deps.PeeringConfig.StoreFile(deps.PeeringConfigFilePath, []string{"p2p"})
		})

		// peers from peering config
		var peers []*p2p.PeerConfig
		if err := deps.PeeringConfig.Unmarshal(CfgPeers, &peers); err != nil {
			CorePlugin.Panicf("invalid peer config: %s", err)
		}

		for i, p := range peers {
			multiAddr, err := multiaddr.NewMultiaddr(p.MultiAddress)
			if err != nil {
				CorePlugin.Panicf("invalid config peer address at pos %d: %s", i, err)
			}

			if err = p2pConfigManager.AddPeer(multiAddr, p.Alias); err != nil {
				CorePlugin.LogWarnf("unable to add peer to config manager %s: %s", p.MultiAddress, err)
			}
		}

		// peers from CLI arguments
		peerIDsStr := deps.PeeringConfig.Strings(CfgP2PPeers)
		peerAliases := deps.PeeringConfig.Strings(CfgP2PPeerAliases)

		applyAliases := true
		if len(peerIDsStr) != len(peerAliases) {
			CorePlugin.LogWarnf("won't apply peer aliases: you must define aliases for all defined static peers (got %d aliases, %d peers).", len(peerAliases), len(peerIDsStr))
			applyAliases = false
		}

		for i, peerIDStr := range peerIDsStr {
			multiAddr, err := multiaddr.NewMultiaddr(peerIDStr)
			if err != nil {
				CorePlugin.Panicf("invalid CLI peer address at pos %d: %s", i, err)
			}

			var alias string
			if applyAliases {
				alias = peerAliases[i]
			}

			if err = p2pConfigManager.AddPeer(multiAddr, alias); err != nil {
				CorePlugin.LogWarnf("unable to add peer to config manager %s: %s", peerIDStr, err)
			}
		}

		p2pConfigManager.StoreOnChange(true)

		return p2pConfigManager
	}); err != nil {
		CorePlugin.Panic(err)
	}
}

func configure() {

	CorePlugin.LogInfof("peer configured, ID: %s", deps.Host.ID())

	if err := CorePlugin.Daemon().BackgroundWorker("Close p2p peer database", func(shutdownSignal <-chan struct{}) {
		<-shutdownSignal

		closeDatabases := func() error {
			if err := deps.PeerStoreContainer.Flush(); err != nil {
				return err
			}

			return deps.PeerStoreContainer.Close()
		}

		CorePlugin.LogInfo("Syncing p2p peer database to disk...")
		if err := closeDatabases(); err != nil {
			CorePlugin.Panicf("Syncing p2p peer database to disk... failed: %s", err)
		}
		CorePlugin.LogInfo("Syncing p2p peer database to disk... done")
	}, shutdown.PriorityCloseDatabase); err != nil {
		CorePlugin.Panicf("failed to start worker: %s", err)
	}
}

func run() {
	if deps.Manager == nil {
		// Manager is optional, due to autopeering entry node
		return
	}

	// register a daemon to disconnect all peers up on shutdown
	if err := CorePlugin.Daemon().BackgroundWorker("Manager", func(shutdownSignal <-chan struct{}) {
		CorePlugin.LogInfof("listening on: %s", deps.Host.Addrs())
		go deps.Manager.Start(shutdownSignal)
		connectConfigKnownPeers()
		<-shutdownSignal
		if err := deps.Host.Peerstore().Close(); err != nil {
			CorePlugin.LogError("unable to cleanly closing peer store: %s", err)
		}
	}, shutdown.PriorityP2PManager); err != nil {
		CorePlugin.Panicf("failed to start worker: %s", err)
	}
}

// connects to the peers defined in the config.
func connectConfigKnownPeers() {
	for _, p := range deps.PeeringConfigManager.Peers() {
		multiAddr, err := multiaddr.NewMultiaddr(p.MultiAddress)
		if err != nil {
			CorePlugin.Panicf("invalid peer address: %s", err)
		}

		addrInfo, err := peer.AddrInfoFromP2pAddr(multiAddr)
		if err != nil {
			CorePlugin.Panicf("invalid peer address info: %s", err)
		}

		if err = deps.Manager.ConnectPeer(addrInfo, p2p.PeerRelationKnown, p.Alias); err != nil {
			CorePlugin.LogInfof("can't connect to peer (%s): %s", multiAddr.String(), err)
		}
	}
}
<file_sep>package database

import (
	"errors"
	"fmt"
	"os"
	"path/filepath"

	"github.com/gohornet/hornet/pkg/utils"

	"github.com/iotaledger/hive.go/kvstore"
	"github.com/iotaledger/hive.go/kvstore/pebble"
	"github.com/iotaledger/hive.go/kvstore/rocksdb"
)

type databaseInfo struct {
	Engine string `toml:"databaseEngine"`
}

// DatabaseEngine parses a string and returns an engine.
// Returns an error if the engine is unknown.
func DatabaseEngine(engine string) (Engine, error) {

	switch engine {
	case EngineRocksDB:
	case EnginePebble:
	default:
		return "", fmt.Errorf("unknown database engine: %s, supported engines: pebble/rocksdb", engine)
	}

	return Engine(engine), nil
}

// CheckDatabaseEngine checks if the correct database engine is used.
// This function stores a so called "database info file" in the database folder or
// checks if an existing "database info file" contains the correct engine.
// Otherwise the files in the database folder are not compatible.
func CheckDatabaseEngine(dbPath string, createDatabaseIfNotExists bool, dbEngine ...Engine) (Engine, error) {

	if createDatabaseIfNotExists && len(dbEngine) == 0 {
		return EngineRocksDB, errors.New("the database engine must be specified if the database should be newly created")
	}

	// check if the database exists and if it should be created
	_, err := os.Stat(dbPath)
	if err != nil {
		if !os.IsNotExist(err) {
			return EngineUnknown, fmt.Errorf("unable to check database path (%s): %w", dbPath, err)
		}

		if !createDatabaseIfNotExists {
			return EngineUnknown, fmt.Errorf("database not found (%s)", dbPath)
		}

		// database will be newly created
	}

	var targetEngine Engine

	// check if the database info file exists and if it should be created
	dbInfoFilePath := filepath.Join(dbPath, "dbinfo")
	_, err = os.Stat(dbInfoFilePath)
	if err != nil {
		if !os.IsNotExist(err) {
			return EngineUnknown, fmt.Errorf("unable to check database info file (%s): %w", dbInfoFilePath, err)
		}

		if len(dbEngine) == 0 {
			return EngineUnknown, fmt.Errorf("database info file not found (%s)", dbInfoFilePath)
		}

		// if the dbInfo file does not exist and the dbEngine is given, create the dbInfo file.
		if err := storeDatabaseInfoToFile(dbInfoFilePath, dbEngine[0]); err != nil {
			return EngineUnknown, err
		}

		targetEngine = dbEngine[0]
	} else {
		dbEngineFromInfoFile, err := LoadDatabaseEngineFromFile(dbInfoFilePath)
		if err != nil {
			return EngineUnknown, err
		}

		// if the dbInfo file exists and the dbEngine is given, compare the engines.
		if len(dbEngine) > 0 {

			if dbEngineFromInfoFile != dbEngine[0] {
				return EngineUnknown, fmt.Errorf(`database engine does not match the configuration: '%v' != '%v'

If you want to use another database engine, you can use the tool './hornet tool db-migration' to convert the current database.`, dbEngineFromInfoFile, dbEngine[0])
			}
		}

		targetEngine = dbEngineFromInfoFile
	}

	return targetEngine, nil
}

// LoadDatabaseEngineFromFile returns the engine from the "database info file".
func LoadDatabaseEngineFromFile(path string) (Engine, error) {

	var info databaseInfo

	if err := utils.ReadTOMLFromFile(path, &info); err != nil {
		return "", fmt.Errorf("unable to read database info file: %w", err)
	}

	return DatabaseEngine(info.Engine)
}

// storeDatabaseInfoToFile stores the used engine in a "database info file".
func storeDatabaseInfoToFile(filePath string, engine Engine) error {
	dirPath := filepath.Dir(filePath)

	if err := os.MkdirAll(dirPath, 0700); err != nil {
		return fmt.Errorf("could not create database dir '%s': %w", dirPath, err)
	}

	info := &databaseInfo{
		Engine: string(engine),
	}

	return utils.WriteTOMLToFile(filePath, info, 0660, "# auto-generated\n# !!! do not modify this file !!!")
}

// StoreWithDefaultSettings returns a kvstore with default settings.
// It also checks if the database engine is correct.
func StoreWithDefaultSettings(path string, createDatabaseIfNotExists bool, dbEngine ...Engine) (kvstore.KVStore, error) {

	targetEngine, err := CheckDatabaseEngine(path, createDatabaseIfNotExists, dbEngine...)
	if err != nil {
		return nil, err
	}

	switch targetEngine {
	case EnginePebble:
		db, err := NewPebbleDB(path, nil, false)
		if err != nil {
			return nil, err
		}
		return pebble.New(db), nil

	case EngineRocksDB:
		db, err := NewRocksDB(path)
		if err != nil {
			return nil, err
		}
		return rocksdb.New(db), nil

	default:
		return nil, fmt.Errorf("unknown database engine: %s, supported engines: pebble/rocksdb", dbEngine)
	}
}
<file_sep>package faucet

import (
	"fmt"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/labstack/echo/v4"
	"github.com/labstack/echo/v4/middleware"
	"github.com/pkg/errors"
	"go.uber.org/dig"
	"golang.org/x/time/rate"

	"github.com/gohornet/hornet/pkg/model/faucet"
	"github.com/gohornet/hornet/pkg/model/storage"
	"github.com/gohornet/hornet/pkg/model/utxo"
	"github.com/gohornet/hornet/pkg/node"
	"github.com/gohornet/hornet/pkg/pow"
	"github.com/gohornet/hornet/pkg/protocol/gossip"
	"github.com/gohornet/hornet/pkg/restapi"
	"github.com/gohornet/hornet/pkg/shutdown"
	"github.com/gohornet/hornet/pkg/tipselect"
	"github.com/gohornet/hornet/pkg/utils"
	"github.com/iotaledger/hive.go/configuration"
	iotago "github.com/iotaledger/iota.go/v2"
	"github.com/iotaledger/iota.go/v2/ed25519"
)

const (

	// RouteFaucetInfo is the route to give info about the faucet address.
	// GET returns address and balance of the faucet.
	RouteFaucetInfo = "/info"

	// RouteFaucetEnqueue is the route to tell the faucet to pay out some funds to the given address.
	// POST enqueues a new request.
	RouteFaucetEnqueue = "/enqueue"
)

func init() {
	Plugin = &node.Plugin{
		Status: node.StatusDisabled,
		Pluggable: node.Pluggable{
			Name:      "Faucet",
			DepsFunc:  func(cDeps dependencies) { deps = cDeps },
			Params:    params,
			Provide:   provide,
			Configure: configure,
			Run:       run,
		},
	}
}

var (
	Plugin *node.Plugin
	deps   dependencies
)

type dependencies struct {
	dig.In
	NodeConfig            *configuration.Configuration `name:"nodeConfig"`
	RestAPIBindAddress    string                       `name:"restAPIBindAddress"`
	FaucetAllowedAPIRoute restapi.AllowedRoute         `name:"faucetAllowedAPIRoute"`
	Faucet                *faucet.Faucet
	Echo                  *echo.Echo
	ShutdownHandler       *shutdown.ShutdownHandler
}

func provide(c *dig.Container) {

	privateKeys, err := utils.LoadEd25519PrivateKeysFromEnvironment("FAUCET_PRV_KEY")
	if err != nil {
		Plugin.Panicf("loading faucet private key failed, err: %s", err)
	}

	if len(privateKeys) == 0 {
		Plugin.Panic("loading faucet private key failed, err: no private keys given")
	}

	if len(privateKeys) > 1 {
		Plugin.Panic("loading faucet private key failed, err: too many private keys given")
	}

	privateKey := privateKeys[0]
	if len(privateKey) != ed25519.PrivateKeySize {
		Plugin.Panic("loading faucet private key failed, err: wrong private key length")
	}

	faucetAddress := iotago.AddressFromEd25519PubKey(privateKey.Public().(ed25519.PublicKey))
	faucetSigner := iotago.NewInMemoryAddressSigner(iotago.NewAddressKeysForEd25519Address(&faucetAddress, privateKey))

	type faucetDeps struct {
		dig.In
		Storage          *storage.Storage
		PowHandler       *pow.Handler
		UTXO             *utxo.Manager
		NodeConfig       *configuration.Configuration `name:"nodeConfig"`
		NetworkID        uint64                       `name:"networkId"`
		BelowMaxDepth    int                          `name:"belowMaxDepth"`
		Bech32HRP        iotago.NetworkPrefix         `name:"bech32HRP"`
		TipSelector      *tipselect.TipSelector
		MessageProcessor *gossip.MessageProcessor
	}

	if err := c.Provide(func(deps faucetDeps) *faucet.Faucet {
		return faucet.New(
			deps.Storage,
			deps.NetworkID,
			deps.BelowMaxDepth,
			deps.UTXO,
			&faucetAddress,
			faucetSigner,
			deps.TipSelector.SelectNonLazyTips,
			deps.PowHandler,
			deps.MessageProcessor.Emit,
			faucet.WithLogger(Plugin.Logger()),
			faucet.WithHRPNetworkPrefix(deps.Bech32HRP),
			faucet.WithAmount(uint64(deps.NodeConfig.Int64(CfgFaucetAmount))),
			faucet.WithSmallAmount(uint64(deps.NodeConfig.Int64(CfgFaucetSmallAmount))),
			faucet.WithMaxAddressBalance(uint64(deps.NodeConfig.Int64(CfgFaucetMaxAddressBalance))),
			faucet.WithMaxOutputCount(deps.NodeConfig.Int(CfgFaucetMaxOutputCount)),
			faucet.WithIndexationMessage(deps.NodeConfig.String(CfgFaucetIndexationMessage)),
			faucet.WithBatchTimeout(deps.NodeConfig.Duration(CfgFaucetBatchTimeout)),
			faucet.WithPowWorkerCount(deps.NodeConfig.Int(CfgFaucetPoWWorkerCount)),
		)
	}); err != nil {
		Plugin.Panic(err)
	}
}

func configure() {

	routeGroup := deps.Echo.Group("/api/plugins/faucet")

	allowedRoutes := map[string][]string{
		http.MethodGet: {
			"/api/plugins/faucet/info",
		},
	}

	rateLimiterSkipper := func(context echo.Context) bool {
		// Check for which route we will skip the rate limiter
		routesForMethod, exists := allowedRoutes[context.Request().Method]
		if !exists {
			return false
		}

		path := context.Request().URL.EscapedPath()
		for _, prefix := range routesForMethod {
			if strings.HasPrefix(path, prefix) {
				return true
			}
		}

		return false
	}

	rateLimiterConfig := middleware.RateLimiterConfig{
		Skipper: rateLimiterSkipper,
		Store: middleware.NewRateLimiterMemoryStoreWithConfig(
			middleware.RateLimiterMemoryStoreConfig{
				Rate:      rate.Limit(1 / 300.0), // 1 request every 5 minutes
				Burst:     10,                    // additional burst of 10 requests
				ExpiresIn: 5 * time.Minute,
			},
		),
		IdentifierExtractor: func(ctx echo.Context) (string, error) {
			id := ctx.RealIP()
			return id, nil
		},
		ErrorHandler: func(context echo.Context, err error) error {
			return context.JSON(http.StatusForbidden, nil)
		},
		DenyHandler: func(context echo.Context, identifier string, err error) error {
			return context.JSON(http.StatusTooManyRequests, nil)
		},
	}
	routeGroup.Use(middleware.RateLimiterWithConfig(rateLimiterConfig))

	routeGroup.GET(RouteFaucetInfo, func(c echo.Context) error {
		resp, err := getFaucetInfo(c)
		if err != nil {
			return err
		}

		return restapi.JSONResponse(c, http.StatusOK, resp)
	})

	routeGroup.POST(RouteFaucetEnqueue, func(c echo.Context) error {
		resp, err := addFaucetOutputToQueue(c)
		if err != nil {
			// own error handler to have nicer user facing error messages.
			var statusCode int
			var message string

			var e *echo.HTTPError
			if errors.As(err, &e) {
				statusCode = e.Code
				if errors.Is(err, restapi.ErrInvalidParameter) {
					message = strings.Replace(err.Error(), ": "+errors.Unwrap(err).Error(), "", 1)
				} else {
					message = err.Error()
				}
			} else {
				statusCode = http.StatusInternalServerError
				message = fmt.Sprintf("internal server error. error: %s", err)
			}

			return c.JSON(statusCode, restapi.HTTPErrorResponseEnvelope{Error: restapi.HTTPErrorResponse{Code: strconv.Itoa(statusCode), Message: message}})
		}

		return restapi.JSONResponse(c, http.StatusAccepted, resp)
	})
}

func run() {
	// create a background worker that handles the enqueued faucet requests
	if err := Plugin.Daemon().BackgroundWorker("Faucet", func(shutdownSignal <-chan struct{}) {
		if err := deps.Faucet.RunFaucetLoop(shutdownSignal); err != nil {
			deps.ShutdownHandler.SelfShutdown(fmt.Sprintf("faucet plugin hit a critical error: %s", err.Error()))
		}
	}, shutdown.PriorityFaucet); err != nil {
		Plugin.Panicf("failed to start worker: %s", err)
	}

	websiteEnabled := deps.NodeConfig.Bool(CfgFaucetWebsiteEnabled)

	if websiteEnabled {
		bindAddr := deps.NodeConfig.String(CfgFaucetWebsiteBindAddress)

		e := echo.New()
		e.HideBanner = true
		e.Use(middleware.Recover())

		setupRoutes(e)

		go func() {
			Plugin.LogInfof("You can now access the faucet website using: http://%s", bindAddr)

			if err := e.Start(bindAddr); err != nil && !errors.Is(err, http.ErrServerClosed) {
				Plugin.LogWarnf("Stopped faucet website server due to an error (%s)", err)
			}
		}()
	}
}
<file_sep>module github.com/gohornet/hornet

go 1.16

replace github.com/labstack/gommon => github.com/muXxer/gommon v0.3.1-0.20210805133353-359faea1baf6

replace github.com/linxGnu/grocksdb => github.com/gohornet/grocksdb v1.6.34-0.20210518222204-d6ea5eedcfb9

require (
	github.com/DataDog/zstd v1.4.8 // indirect
	github.com/Microsoft/go-winio v0.5.0 // indirect
	github.com/Shopify/sarama v1.29.1 // indirect
	github.com/StackExchange/wmi v1.2.1 // indirect
	github.com/bits-and-blooms/bitset v1.2.0
	github.com/blang/vfs v1.0.0
	github.com/cockroachdb/errors v1.8.6 // indirect
	github.com/cockroachdb/pebble v0.0.0-20210831135706-8731fd6ed157
	github.com/cockroachdb/redact v1.1.3 // indirect
	github.com/containerd/containerd v1.5.5 // indirect
	github.com/dgraph-io/ristretto v0.1.0 // indirect
	github.com/docker/docker v20.10.8+incompatible
	github.com/docker/go-connections v0.4.0
	github.com/dustin/go-humanize v1.0.0
	github.com/eclipse/paho.mqtt.golang v1.3.5
	github.com/fhmq/hmq v0.0.0-20210810024638-1d6979189a22
	github.com/gin-gonic/gin v1.7.4 // indirect
	github.com/go-echarts/go-echarts v1.0.0
	github.com/go-playground/validator/v10 v10.9.0 // indirect
	github.com/gobuffalo/packr/v2 v2.8.1
	github.com/golang-jwt/jwt v3.2.2+incompatible
	github.com/google/go-querystring v1.1.0 // indirect
	github.com/gorilla/websocket v1.4.2
	github.com/hashicorp/go-version v1.3.0 // indirect
	github.com/iotaledger/go-ds-kvstore v0.0.0-20210819121432-6e2ce2d41200
	github.com/iotaledger/hive.go v0.0.0-20210830134454-9d969a8cc0a2
	github.com/iotaledger/iota.go v1.0.0
	github.com/iotaledger/iota.go/v2 v2.0.1-0.20210830162758-173bada804f9
	github.com/ipfs/go-cid v0.1.0 // indirect
	github.com/ipfs/go-datastore v0.4.6
	github.com/ipfs/go-ds-badger v0.2.7
	github.com/karrick/godirwalk v1.16.1 // indirect
	github.com/labstack/echo/v4 v4.5.0
	github.com/labstack/gommon v0.3.0
	github.com/libp2p/go-libp2p v0.15.0-rc.1
	github.com/libp2p/go-libp2p-connmgr v0.2.4
	github.com/libp2p/go-libp2p-core v0.9.0
	github.com/libp2p/go-libp2p-peerstore v0.2.9-0.20210814101051-ca567f210575
	github.com/moby/term v0.0.0-20210619224110-3f7ff695adc6 // indirect
	github.com/morikuni/aec v1.0.0 // indirect
	github.com/mr-tron/base58 v1.2.0
	github.com/multiformats/go-multiaddr v0.4.0
	github.com/pelletier/go-toml v1.9.3 // indirect
	github.com/pelletier/go-toml/v2 v2.0.0-beta.3
	github.com/pkg/errors v0.9.1
	github.com/prometheus/client_golang v1.11.0
	github.com/segmentio/fasthash v1.0.3 // indirect
	github.com/shirou/gopsutil v3.21.8+incompatible
	github.com/spf13/pflag v1.0.5
	github.com/stretchr/testify v1.7.1-0.20210427113832-6241f9ab9942
	github.com/tcnksm/go-latest v0.0.0-20170313132115-e3007ae9052e
	github.com/tidwall/gjson v1.8.1 // indirect
	github.com/tklauser/go-sysconf v0.3.9 // indirect
	github.com/ugorji/go v1.2.6 // indirect
	github.com/wollac/iota-crypto-demo v0.0.0-20210820085437-1a7b8ead2881
	gitlab.com/powsrv.io/go/client v0.0.0-20210203192329-84583796cd46
	go.uber.org/atomic v1.9.0
	go.uber.org/dig v1.12.0
	golang.org/x/crypto v0.0.0-20210817164053-32db794688a5
	golang.org/x/exp v0.0.0-20210831221722-b4e88ed8e8aa // indirect
	golang.org/x/net v0.0.0-20210825183410-e898025ed96a
	golang.org/x/sys v0.0.0-20210831042530-f4d43177bf5e // indirect
	golang.org/x/term v0.0.0-20210615171337-6886f2dfbf5b
	golang.org/x/time v0.0.0-20210723032227-1f47c861a9ac
	google.golang.org/genproto v0.0.0-20210831024726-fe130286e0e2 // indirect
)
<file_sep>package gossip

import (
	"fmt"
	"sync"
	"time"

	"github.com/gohornet/hornet/pkg/dag"
	"github.com/gohornet/hornet/pkg/model/hornet"
	"github.com/gohornet/hornet/pkg/model/milestone"
	"github.com/gohornet/hornet/pkg/model/storage"
	"github.com/iotaledger/hive.go/events"
	"github.com/iotaledger/hive.go/syncutils"
)

// NewWarpSync creates a new WarpSync instance with the given advancement range and criteria func.
// If no advancement func is provided, the WarpSync uses AdvanceAtPercentageReached with DefaultAdvancementThreshold.
func NewWarpSync(advRange int, advanceCheckpointCriteriaFunc ...AdvanceCheckpointCriteria) *WarpSync {
	ws := &WarpSync{
		AdvancementRange: advRange,
		Events: Events{
			CheckpointUpdated: events.NewEvent(CheckpointCaller),
			TargetUpdated:     events.NewEvent(TargetCaller),
			Start:             events.NewEvent(SyncStartCaller),
			Done:              events.NewEvent(SyncDoneCaller),
		},
	}
	if len(advanceCheckpointCriteriaFunc) > 0 {
		ws.advCheckpointCriteria = advanceCheckpointCriteriaFunc[0]
	} else {
		ws.advCheckpointCriteria = AdvanceAtPercentageReached(DefaultAdvancementThreshold)
	}
	return ws
}

func SyncStartCaller(handler interface{}, params ...interface{}) {
	handler.(func(target milestone.Index, newCheckpoint milestone.Index, msRange int32))(params[0].(milestone.Index), params[1].(milestone.Index), params[2].(int32))
}

func SyncDoneCaller(handler interface{}, params ...interface{}) {
	handler.(func(delta int, referencedMessagesTotal int, dur time.Duration))(params[0].(int), params[1].(int), params[2].(time.Duration))
}

func CheckpointCaller(handler interface{}, params ...interface{}) {
	handler.(func(newCheckpoint milestone.Index, oldCheckpoint milestone.Index, msRange int32, target milestone.Index))(params[0].(milestone.Index), params[1].(milestone.Index), params[2].(int32), params[3].(milestone.Index))
}

func TargetCaller(handler interface{}, params ...interface{}) {
	handler.(func(checkpoint milestone.Index, target milestone.Index))(params[0].(milestone.Index), params[1].(milestone.Index))
}

// Events holds WarpSync related events.
type Events struct {
	// Fired when a new set of milestones should be requested.
	CheckpointUpdated *events.Event
	// Fired when the target milestone is updated.
	TargetUpdated *events.Event
	// Fired when warp synchronization starts.
	Start *events.Event
	// Fired when the warp synchronization is done.
	Done *events.Event
}

// AdvanceCheckpointCriteria is a function which determines whether the checkpoint should be advanced.
type AdvanceCheckpointCriteria func(currentConfirmed, previousCheckpoint, currentCheckpoint milestone.Index) bool

// DefaultAdvancementThreshold is the default threshold at which a checkpoint advancement is done.
// Per default an advancement is always done as soon the confirmed milestone enters the range between
// the previous and current checkpoint.
const DefaultAdvancementThreshold = 0.0

// AdvanceAtPercentageReached is an AdvanceCheckpointCriteria which advances the checkpoint
// when the current one was reached by >=X% by the current confirmed milestone in relation to the previous checkpoint.
func AdvanceAtPercentageReached(threshold float64) AdvanceCheckpointCriteria {
	return func(currentConfirmed, previousCheckpoint, currentCheckpoint milestone.Index) bool {
		// the previous checkpoint can be over the current confirmed milestone,
		// as advancements move the checkpoint window above the confirmed milestone
		if currentConfirmed < previousCheckpoint {
			return false
		}
		checkpointDelta := currentCheckpoint - previousCheckpoint
		progress := currentConfirmed - previousCheckpoint
		return float64(progress)/float64(checkpointDelta) >= threshold
	}
}

// WarpSync is metadata about doing a synchronization via STING messages.
type WarpSync struct {
	sync.Mutex

	// The used advancement range per checkpoint.
	AdvancementRange int
	// The Events of the warpsync.
	Events Events
	// The criteria whether to advance to the next checkpoint.
	advCheckpointCriteria AdvanceCheckpointCriteria

	// The current confirmed milestone of the node.
	CurrentConfirmedMilestone milestone.Index
	// The starting time of the synchronization.
	StartTime time.Time
	// The starting point of the synchronization.
	InitMilestone milestone.Index
	// The target milestone to which to synchronize to.
	TargetMilestone milestone.Index
	// The previous checkpoint of the synchronization.
	PreviousCheckpoint milestone.Index
	// The current checkpoint of the synchronization.
	CurrentCheckpoint milestone.Index
	// The amount of referenced messages during this warpsync run.
	referencedMessagesTotal int
}

// UpdateCurrentConfirmedMilestone updates the current confirmed milestone index state.
func (ws *WarpSync) UpdateCurrentConfirmedMilestone(current milestone.Index) {
	ws.Lock()
	defer ws.Unlock()

	if current <= ws.CurrentConfirmedMilestone {
		return
	}
	ws.CurrentConfirmedMilestone = current

	// synchronization not started
	if ws.CurrentCheckpoint == 0 {
		return
	}

	// finished
	if ws.TargetMilestone != 0 && ws.CurrentConfirmedMilestone >= ws.TargetMilestone {
		ws.Events.Done.Trigger(int(ws.TargetMilestone-ws.InitMilestone), ws.referencedMessagesTotal, time.Since(ws.StartTime))
		ws.reset()
		return
	}

	// check whether advancement criteria is fulfilled
	if !ws.advCheckpointCriteria(ws.CurrentConfirmedMilestone, ws.PreviousCheckpoint, ws.CurrentCheckpoint) {
		return
	}

	oldCheckpoint := ws.CurrentCheckpoint
	if msRange := ws.advanceCheckpoint(); msRange != 0 {
		ws.Events.CheckpointUpdated.Trigger(ws.CurrentCheckpoint, oldCheckpoint, msRange, ws.TargetMilestone)
	}
}

// UpdateTargetMilestone updates the synchronization target if it is higher than the current one and
// triggers a synchronization start if the target was set for the first time.
func (ws *WarpSync) UpdateTargetMilestone(target milestone.Index) {
	ws.Lock()
	defer ws.Unlock()

	if target <= ws.TargetMilestone {
		return
	}

	ws.TargetMilestone = target

	// as a special case, while we are warp syncing and within the last checkpoint range,
	// new target milestones need to shift the checkpoint to the new target, in order
	// to fire an 'updated checkpoint event'/respectively updating the request queue filter.
	// since we will request missing parents for the new target, it will still solidify
	// even though we discarded requests for a short period of time parents when the
	// request filter wasn't yet updated.
	if ws.CurrentCheckpoint != 0 && ws.CurrentCheckpoint+milestone.Index(ws.AdvancementRange) > ws.TargetMilestone {
		oldCheckpoint := ws.CurrentCheckpoint
		reqRange := ws.TargetMilestone - ws.CurrentCheckpoint
		ws.CurrentCheckpoint = ws.TargetMilestone
		ws.Events.CheckpointUpdated.Trigger(ws.CurrentCheckpoint, oldCheckpoint, int32(reqRange), ws.TargetMilestone)
	}

	if ws.CurrentCheckpoint != 0 {
		// if synchronization was already started, only update the target
		ws.Events.TargetUpdated.Trigger(ws.CurrentCheckpoint, ws.TargetMilestone)
		return
	}

	// do not start the synchronization if current confirmed is newer than the target or the delta is smaller than 2
	if ws.CurrentConfirmedMilestone >= ws.TargetMilestone || target-ws.CurrentConfirmedMilestone < 2 {
		return
	}

	// start the synchronization
	ws.StartTime = time.Now()
	ws.InitMilestone = ws.CurrentConfirmedMilestone
	ws.PreviousCheckpoint = ws.CurrentConfirmedMilestone
	advancementRange := ws.advanceCheckpoint()
	ws.Events.Start.Trigger(ws.TargetMilestone, ws.CurrentCheckpoint, advancementRange)
}

// AddReferencedMessagesCount adds the amount of referenced messages to collect stats.
func (ws *WarpSync) AddReferencedMessagesCount(referencedMessagesCount int) {
	ws.Lock()
	defer ws.Unlock()

	ws.referencedMessagesTotal += referencedMessagesCount
}

// advances the next checkpoint by either incrementing from the current
// via the checkpoint range or max to the target of the synchronization.
// returns the chosen range.
func (ws *WarpSync) advanceCheckpoint() int32 {
	if ws.CurrentCheckpoint != 0 {
		ws.PreviousCheckpoint = ws.CurrentCheckpoint
	}

	advRange := milestone.Index(ws.AdvancementRange)

	// make sure we advance max to the target milestone
	if ws.TargetMilestone-ws.CurrentConfirmedMilestone <= advRange || ws.TargetMilestone-ws.CurrentCheckpoint <= advRange {
		deltaRange := ws.TargetMilestone - ws.CurrentCheckpoint
		if deltaRange > ws.TargetMilestone-ws.CurrentConfirmedMilestone {
			deltaRange = ws.TargetMilestone - ws.CurrentConfirmedMilestone
		}
		ws.CurrentCheckpoint = ws.TargetMilestone
		return int32(deltaRange)
	}

	// at start simply advance from the current confirmed
	if ws.CurrentCheckpoint == 0 {
		ws.CurrentCheckpoint = ws.CurrentConfirmedMilestone + advRange
		return int32(advRange)
	}

	ws.CurrentCheckpoint = ws.CurrentCheckpoint + advRange
	return int32(advRange)
}

// resets the warp sync.
func (ws *WarpSync) reset() {
	ws.StartTime = time.Time{}
	ws.InitMilestone = 0
	ws.TargetMilestone = 0
	ws.PreviousCheckpoint = 0
	ws.CurrentCheckpoint = 0
	ws.referencedMessagesTotal = 0
}

// NewWarpSyncMilestoneRequester creates a new WarpSyncMilestoneRequester instance.
func NewWarpSyncMilestoneRequester(storage *storage.Storage, requester *Requester, preventDiscard bool) *WarpSyncMilestoneRequester {
	return &WarpSyncMilestoneRequester{
		storage:        storage,
		requester:      requester,
		preventDiscard: preventDiscard,
		traversed:      make(map[string]struct{}),
	}
}

// WarpSyncMilestoneRequester walks the cones of existing but non-solid milestones and memoizes already walked messages and milestones.
type WarpSyncMilestoneRequester struct {
	syncutils.Mutex

	storage        *storage.Storage
	requester      *Requester
	preventDiscard bool
	traversed      map[string]struct{}
}

// RequestMissingMilestoneParents traverses the parents of a given milestone and requests each missing parent.
// Already requested milestones or traversed messages will be ignored, to circumvent requesting
// the same parents multiple times.
func (w *WarpSyncMilestoneRequester) RequestMissingMilestoneParents(msIndex milestone.Index) {
	w.Lock()
	defer w.Unlock()

	if msIndex <= w.storage.ConfirmedMilestoneIndex() {
		return
	}

	cachedMs := w.storage.CachedMilestoneOrNil(msIndex) // milestone +1
	if cachedMs == nil {
		panic(fmt.Sprintf("milestone %d wasn't found", msIndex))
	}

	milestoneMessageID := cachedMs.Milestone().MessageID
	cachedMs.Release(true) // message -1

	// error is ignored because the next milestone will repeat the process anyway
	_ = dag.TraverseParentsOfMessage(w.storage, milestoneMessageID,
		// traversal stops if no more messages pass the given condition
		// Caution: condition func is not in DFS order
		func(cachedMsgMeta *storage.CachedMetadata) (bool, error) { // meta +1
			defer cachedMsgMeta.Release(true) // meta -1

			mapKey := cachedMsgMeta.Metadata().MessageID().ToMapKey()
			if _, previouslyTraversed := w.traversed[mapKey]; previouslyTraversed {
				return false, nil
			}
			w.traversed[mapKey] = struct{}{}

			if cachedMsgMeta.Metadata().IsSolid() {
				return false, nil
			}

			return true, nil
		},
		// consumer
		nil,
		// called on missing parents
		func(parentMessageID hornet.MessageID) error {
			w.requester.Request(parentMessageID, msIndex, w.preventDiscard)
			return nil
		},
		// called on solid entry points
		// Ignore solid entry points (snapshot milestone included)
		nil,
		false, nil)
}

// Cleanup cleans up traversed messages to free memory.
func (w *WarpSyncMilestoneRequester) Cleanup() {
	w.Lock()
	defer w.Unlock()

	w.traversed = make(map[string]struct{})
}
<file_sep>package toolset

import (
	"fmt"
	"os"
	"path/filepath"

	"github.com/iotaledger/hive.go/configuration"

	"github.com/libp2p/go-libp2p-core/peer"

	p2pCore "github.com/gohornet/hornet/core/p2p"
	"github.com/gohornet/hornet/pkg/jwt"
	"github.com/gohornet/hornet/pkg/p2p"
	"github.com/gohornet/hornet/plugins/restapi"
)

func generateJWTApiToken(nodeConfig *configuration.Configuration, args []string) error {

	printUsage := func() {
		println("Usage:")
		println(fmt.Sprintf("	%s [P2P_DATABASE_PATH]", ToolJWTApi))
		println()
		println("	[P2P_DATABASE_PATH] - the path to the p2p database folder (optional)")
		println()
		println(fmt.Sprintf("example: %s %s", ToolJWTApi, "p2pstore"))
	}

	if len(args) > 1 {
		printUsage()
		return fmt.Errorf("too many arguments for '%s'", ToolJWTApi)
	}

	salt := nodeConfig.String(restapi.CfgRestAPIJWTAuthSalt)
	if len(salt) == 0 {
		return fmt.Errorf("'%s' should not be empty", restapi.CfgRestAPIJWTAuthSalt)
	}

	p2pDatabasePath := nodeConfig.String(p2pCore.CfgP2PDatabasePath)
	if len(args) > 0 {
		p2pDatabasePath = args[0]
	}
	privKeyFilePath := filepath.Join(p2pDatabasePath, p2p.PrivKeyFileName)

	_, err := os.Stat(privKeyFilePath)
	switch {
	case os.IsNotExist(err):
		// private key does not exist
		return fmt.Errorf("private key file (%s) does not exist", privKeyFilePath)

	case err == nil || os.IsExist(err):
		// private key file exists

	default:
		return fmt.Errorf("unable to check private key file (%s): %w", privKeyFilePath, err)
	}

	privKey, err := p2p.ReadEd25519PrivateKeyFromPEMFile(privKeyFilePath)
	if err != nil {
		return fmt.Errorf("reading private key file for peer identity failed: %w", err)
	}

	peerID, err := peer.IDFromPublicKey(privKey.GetPublic())
	if err != nil {
		return fmt.Errorf("unable to get peer identity from public key: %w", err)
	}

	// API tokens do not expire.
	jwtAuth, err := jwt.NewJWTAuth(salt,
		0,
		peerID.String(),
		privKey,
	)
	if err != nil {
		return fmt.Errorf("JWT auth initialization failed: %w", err)
	}

	jwtToken, err := jwtAuth.IssueJWT(true, false)
	if err != nil {
		return fmt.Errorf("issuing JWT token failed: %w", err)
	}

	fmt.Println("Your API JWT token: ", jwtToken)

	return nil
}
<file_sep>package restapi

import (
	"net/http"

	"github.com/labstack/echo/v4"
)

func setupHealthRoute() {
	deps.Echo.GET(nodeAPIHealthRoute, func(c echo.Context) error {

		// node mode
		if deps.Tangle != nil && !deps.Tangle.IsNodeHealthy() {
			return c.NoContent(http.StatusServiceUnavailable)
		}

		return c.NoContent(http.StatusOK)
	})
}
<file_sep>package faucet

import (
	"time"

	flag "github.com/spf13/pflag"

	"github.com/gohornet/hornet/pkg/node"
	iotago "github.com/iotaledger/iota.go/v2"
)

const (
	// the amount of funds the requester receives.
	CfgFaucetAmount = "faucet.amount"
	// the amount of funds the requester receives if the target address has more funds than the faucet amount and less than maximum.
	CfgFaucetSmallAmount = "faucet.smallAmount"
	// the maximum allowed amount of funds on the target address.
	CfgFaucetMaxAddressBalance = "faucet.maxAddressBalance"
	// the maximum output count per faucet message.
	CfgFaucetMaxOutputCount = "faucet.maxOutputCount"
	// the faucet transaction indexation payload.
	CfgFaucetIndexationMessage = "faucet.indexationMessage"
	// the maximum duration for collecting faucet batches.
	CfgFaucetBatchTimeout = "faucet.batchTimeout"
	// the amount of workers used for calculating PoW when issuing faucet messages.
	CfgFaucetPoWWorkerCount = "faucet.powWorkerCount"
	// the bind address on which the faucet website can be accessed from
	CfgFaucetWebsiteBindAddress = "faucet.website.bindAddress"
	// whether to host the faucet website
	CfgFaucetWebsiteEnabled = "faucet.website.enabled"
)

var params = &node.PluginParams{
	Params: map[string]*flag.FlagSet{
		"nodeConfig": func() *flag.FlagSet {
			fs := flag.NewFlagSet("", flag.ContinueOnError)
			fs.Int64(CfgFaucetAmount, 10000000, "the amount of funds the requester receives")
			fs.Int64(CfgFaucetSmallAmount, 1000000, "the amount of funds the requester receives if the target address has more funds than the faucet amount and less than maximum")
			fs.Int64(CfgFaucetMaxAddressBalance, 20000000, "the maximum allowed amount of funds on the target address")
			fs.Int(CfgFaucetMaxOutputCount, iotago.MaxOutputsCount, "the maximum output count per faucet message")
			fs.String(CfgFaucetIndexationMessage, "HORNET FAUCET", "the faucet transaction indexation payload")
			fs.Duration(CfgFaucetBatchTimeout, 2*time.Second, "the maximum duration for collecting faucet batches")
			fs.Int(CfgFaucetPoWWorkerCount, 0, "the amount of workers used for calculating PoW when issuing faucet messages")
			fs.String(CfgFaucetWebsiteBindAddress, "localhost:8091", "the bind address on which the faucet website can be accessed from")
			fs.Bool(CfgFaucetWebsiteEnabled, false, "whether to host the faucet website")
			return fs
		}(),
	},
	Masked: nil,
}
<file_sep>package database

import (
	"encoding/json"
	"errors"
	"time"

	"github.com/iotaledger/hive.go/events"
	"github.com/iotaledger/hive.go/kvstore"
	"github.com/iotaledger/hive.go/logger"
	"github.com/iotaledger/hive.go/syncutils"
)

type Engine string

const (
	EngineUnknown = "unknown"
	EngineRocksDB = "rocksdb"
	EnginePebble  = "pebble"
)

var (
	// ErrNothingToCleanUp is returned when nothing is there to clean up in the database.
	ErrNothingToCleanUp = errors.New("Nothing to clean up in the databases")
)

type DatabaseCleanup struct {
	Start time.Time
	End   time.Time
}

func (c *DatabaseCleanup) MarshalJSON() ([]byte, error) {

	cleanup := struct {
		Start int64 `json:"start"`
		End   int64 `json:"end"`
	}{
		Start: 0,
		End:   0,
	}

	if !c.Start.IsZero() {
		cleanup.Start = c.Start.Unix()
	}

	if !c.End.IsZero() {
		cleanup.End = c.End.Unix()
	}

	return json.Marshal(cleanup)
}

func DatabaseCleanupCaller(handler interface{}, params ...interface{}) {
	handler.(func(*DatabaseCleanup))(params[0].(*DatabaseCleanup))
}

type Events struct {
	DatabaseCleanup    *events.Event
	DatabaseCompaction *events.Event
}

// New creates a new Database instance.
func New(log *logger.Logger, kvStore kvstore.KVStore, events *Events, compactionSupported bool, compactionRunningFunc func() bool) *Database {
	return &Database{
		log:                   log,
		store:                 kvStore,
		events:                events,
		compactionSupported:   compactionSupported,
		compactionRunningFunc: compactionRunningFunc,
	}
}

// Database holds the underlying KVStore and database specific functions.
type Database struct {
	log                   *logger.Logger
	store                 kvstore.KVStore
	events                *Events
	compactionSupported   bool
	compactionRunningFunc func() bool
	garbageCollectionLock syncutils.Mutex
}

// KVStore returns the underlying KVStore.
func (db *Database) KVStore() kvstore.KVStore {
	return db.store
}

// Events returns the events of the database.
func (db *Database) Events() *Events {
	return db.events
}

// CompactionSupported returns whether the database engine supports compaction.
func (db *Database) CompactionSupported() bool {
	return db.compactionSupported
}

// CompactionRunning returns whether a compaction is running.
func (db *Database) CompactionRunning() bool {
	return db.compactionRunningFunc()
}

func (db *Database) DatabaseSupportsCleanup() bool {
	// ToDo: add this to the db initialization of the different database engines in the core module
	return false
}

func (db *Database) CleanupDatabases() error {
	// ToDo: add this to the db initialization of the different database engines in the core module
	return ErrNothingToCleanUp
}

func (db *Database) RunGarbageCollection() {
	if !db.DatabaseSupportsCleanup() {
		return
	}

	db.garbageCollectionLock.Lock()
	defer db.garbageCollectionLock.Unlock()

	db.log.Info("running full database garbage collection. This can take a while...")

	start := time.Now()

	db.events.DatabaseCleanup.Trigger(&DatabaseCleanup{
		Start: start,
	})

	err := db.CleanupDatabases()

	end := time.Now()

	db.events.DatabaseCleanup.Trigger(&DatabaseCleanup{
		Start: start,
		End:   end,
	})

	if err != nil {
		if !errors.Is(err, ErrNothingToCleanUp) {
			db.log.Warnf("full database garbage collection failed with error: %s. took: %v", err, end.Sub(start).Truncate(time.Millisecond))
			return
		}
	}

	db.log.Infof("full database garbage collection finished. took %v", end.Sub(start).Truncate(time.Millisecond))
}
<file_sep>package node

import (
	"strings"
	"sync"

	flag "github.com/spf13/pflag"

	"github.com/iotaledger/hive.go/configuration"
	"github.com/iotaledger/hive.go/daemon"
	"github.com/iotaledger/hive.go/logger"
)

type PluginStatus int

const (
	StatusDisabled PluginStatus = iota
	StatusEnabled
)

// PluginParams defines the parameters configuration of a plugin.
type PluginParams struct {
	// The parameters of the plugin under for the defined configuration.
	Params map[string]*flag.FlagSet
	// The configuration values to mask.
	Masked []string
}

// Pluggable is something which extends the Node's capabilities.
type Pluggable struct {
	// A reference to the Node instance.
	Node *Node
	// The name of the plugin.
	Name string
	// The config parameters for this plugin.
	Params *PluginParams
	// The function to call to initialize the plugin dependencies.
	DepsFunc interface{}
	// InitConfigPars gets called in the init stage of node initialization.
	// This can be used to provide config parameters even if the pluggable is disabled.
	InitConfigPars InitConfigParsFunc
	// PreProvide gets called before the provide stage of node initialization.
	// This can be used to force disable other pluggables before they get initialized.
	PreProvide PreProvideFunc
	// Provide gets called in the provide stage of node initialization (enabled pluggables only).
	Provide ProvideFunc
	// Configure gets called in the configure stage of node initialization (enabled pluggables only).
	Configure Callback
	// Run gets called in the run stage of node initialization (enabled pluggables only).
	Run Callback

	// The logger instance used in this plugin.
	log     *logger.Logger
	logOnce sync.Once
}

// Logger instantiates and returns a logger with the name of the plugin.
func (p *Pluggable) Logger() *logger.Logger {
	p.logOnce.Do(func() {
		p.log = logger.NewLogger(p.Name)
	})

	return p.log
}

func (p *Pluggable) Daemon() daemon.Daemon {
	return p.Node.Daemon()
}

func (p *Pluggable) Identifier() string {
	return strings.ToLower(strings.Replace(p.Name, " ", "", -1))
}

// LogDebug uses fmt.Sprint to construct and log a message.
func (p *Pluggable) LogDebug(args ...interface{}) {
	p.Logger().Debug(args...)
}

// LogDebugf uses fmt.Sprintf to log a templated message.
func (p *Pluggable) LogDebugf(template string, args ...interface{}) {
	p.Logger().Debugf(template, args...)
}

// LogError uses fmt.Sprint to construct and log a message.
func (p *Pluggable) LogError(args ...interface{}) {
	p.Logger().Error(args...)
}

// LogErrorf uses fmt.Sprintf to log a templated message.
func (p *Pluggable) LogErrorf(template string, args ...interface{}) {
	p.Logger().Errorf(template, args...)
}

// LogFatal uses fmt.Sprint to construct and log a message, then calls os.Exit.
func (p *Pluggable) LogFatal(args ...interface{}) {
	p.Logger().Fatal(args...)
}

// LogFatalf uses fmt.Sprintf to log a templated message, then calls os.Exit.
func (p *Pluggable) LogFatalf(template string, args ...interface{}) {
	p.Logger().Fatalf(template, args...)
}

// LogInfo uses fmt.Sprint to construct and log a message.
func (p *Pluggable) LogInfo(args ...interface{}) {
	p.Logger().Info(args...)
}

// LogInfof uses fmt.Sprintf to log a templated message.
func (p *Pluggable) LogInfof(template string, args ...interface{}) {
	p.Logger().Infof(template, args...)
}

// LogWarn uses fmt.Sprint to construct and log a message.
func (p *Pluggable) LogWarn(args ...interface{}) {
	p.Logger().Warn(args...)
}

// LogWarnf uses fmt.Sprintf to log a templated message.
func (p *Pluggable) LogWarnf(template string, args ...interface{}) {
	p.Logger().Warnf(template, args...)
}

// Panic uses fmt.Sprint to construct and log a message, then panics.
func (p *Pluggable) Panic(args ...interface{}) {
	p.Logger().Panic(args...)
}

// Panicf uses fmt.Sprintf to log a templated message, then panics.
func (p *Pluggable) Panicf(template string, args ...interface{}) {
	p.Logger().Panicf(template, args...)
}

// InitPlugin is the module initializing configuration of the node.
// A Node can only have one of such modules.
type InitPlugin struct {
	Pluggable
	// Init gets called in the initialization stage of the node.
	Init InitFunc
	// The configs this InitPlugin brings to the node.
	Configs map[string]*configuration.Configuration
}

// CorePlugin is a plugin essential for node operation.
// It can not be disabled.
type CorePlugin struct {
	Pluggable
}

type Plugin struct {
	Pluggable
	// The status of the plugin.
	Status PluginStatus
}
<file_sep>package toolset

import (
	"encoding/hex"
	"fmt"
	"os"
	"path/filepath"

	"github.com/libp2p/go-libp2p-core/peer"
	"github.com/mr-tron/base58"

	"github.com/iotaledger/hive.go/configuration"

	p2pCore "github.com/gohornet/hornet/core/p2p"
	"github.com/gohornet/hornet/pkg/p2p"
)

func extractP2PIdentity(nodeConfig *configuration.Configuration, args []string) error {
	printUsage := func() {
		println("Usage:")
		println(fmt.Sprintf("	%s [P2P_DATABASE_PATH]", ToolP2PExtractIdentity))
		println()
		println("	[P2P_DATABASE_PATH] - the path to the p2p database folder (optional)")
		println()
		println(fmt.Sprintf("example: %s %s", ToolP2PExtractIdentity, "p2pstore"))
	}

	if len(args) > 1 {
		printUsage()
		return fmt.Errorf("too many arguments for '%s'", ToolP2PExtractIdentity)
	}

	p2pDatabasePath := nodeConfig.String(p2pCore.CfgP2PDatabasePath)
	if len(args) > 0 {
		p2pDatabasePath = args[0]
	}
	privKeyFilePath := filepath.Join(p2pDatabasePath, p2p.PrivKeyFileName)

	_, err := os.Stat(privKeyFilePath)
	switch {
	case os.IsNotExist(err):
		// private key does not exist
		return fmt.Errorf("private key file (%s) does not exist", privKeyFilePath)

	case err == nil || os.IsExist(err):
		// private key file exists

	default:
		return fmt.Errorf("unable to check private key file (%s): %w", privKeyFilePath, err)
	}

	privKey, err := p2p.ReadEd25519PrivateKeyFromPEMFile(privKeyFilePath)
	if err != nil {
		return fmt.Errorf("reading private key file for peer identity failed: %w", err)
	}

	peerID, err := peer.IDFromPublicKey(privKey.GetPublic())
	if err != nil {
		return fmt.Errorf("unable to get peer identity from public key: %w", err)
	}

	privKeyBytes, err := privKey.Raw()
	if err != nil {
		return fmt.Errorf("unable to get raw private key bytes: %w", err)
	}

	pubKeyBytes, err := privKey.GetPublic().Raw()
	if err != nil {
		return fmt.Errorf("unable to get raw public key bytes: %w", err)
	}

	fmt.Println("Your p2p private key (hex):   ", hex.EncodeToString(privKeyBytes))
	fmt.Println("Your p2p public key (hex):    ", hex.EncodeToString(pubKeyBytes))
	fmt.Println("Your p2p public key (base58): ", base58.Encode(pubKeyBytes))
	fmt.Println("Your p2p PeerID:              ", peerID.String())

	return nil
}
<file_sep>package faucet

import (
	"fmt"
	"net"
	"net/http"
	"net/url"
	"strings"

	"github.com/gobuffalo/packr/v2"
	"github.com/labstack/echo/v4"
	"github.com/labstack/echo/v4/middleware"
)

var (
	// holds faucet website assets
	appBox = packr.New("Faucet_App", "./frontend/public")
)

func appBoxMiddleware() echo.MiddlewareFunc {
	return func(next echo.HandlerFunc) echo.HandlerFunc {
		return func(c echo.Context) (err error) {
			contentType := calculateMimeType(c)

			path := strings.TrimPrefix(c.Request().URL.Path, "/")
			if len(path) == 0 {
				path = "index.html"
				contentType = echo.MIMETextHTMLCharsetUTF8
			}
			staticBlob, err := appBox.Find(path)
			if err != nil {
				// If the asset cannot be found, fall back to the index.html for routing
				path = "index.html"
				contentType = echo.MIMETextHTMLCharsetUTF8
				staticBlob, err = appBox.Find(path)
				if err != nil {
					return next(c)
				}
			}
			return c.Blob(http.StatusOK, contentType, staticBlob)
		}
	}
}

func apiMiddlewares() []echo.MiddlewareFunc {

	proxySkipper := func(context echo.Context) bool {
		// Only proxy allowed routes, skip all others
		return !deps.FaucetAllowedAPIRoute(context)
	}

	apiBindAddr := deps.RestAPIBindAddress
	_, apiBindPort, err := net.SplitHostPort(apiBindAddr)
	if err != nil {
		Plugin.LogFatalf("wrong REST API bind address: %s", err)
	}

	apiURL, err := url.Parse(fmt.Sprintf("http://localhost:%s", apiBindPort))
	if err != nil {
		Plugin.LogFatalf("wrong faucet website API url: %s", err)
	}

	balancer := middleware.NewRoundRobinBalancer([]*middleware.ProxyTarget{
		{
			URL: apiURL,
		},
	})

	config := middleware.ProxyConfig{
		Skipper:  proxySkipper,
		Balancer: balancer,
	}

	return []echo.MiddlewareFunc{
		middleware.ProxyWithConfig(config),
	}
}

func calculateMimeType(e echo.Context) string {
	url := e.Request().URL.String()

	switch {
	case strings.HasSuffix(url, ".html"):
		return echo.MIMETextHTMLCharsetUTF8
	case strings.HasSuffix(url, ".css"):
		return "text/css"
	case strings.HasSuffix(url, ".js"):
		return echo.MIMEApplicationJavaScript
	case strings.HasSuffix(url, ".json"):
		return echo.MIMEApplicationJSONCharsetUTF8
	case strings.HasSuffix(url, ".png"):
		return "image/png"
	case strings.HasSuffix(url, ".svg"):
		return "image/svg+xml"
	default:
		return echo.MIMETextHTMLCharsetUTF8
	}
}

func enforceMaxOneDotPerURL(next echo.HandlerFunc) echo.HandlerFunc {
	return func(c echo.Context) error {
		if strings.Count(c.Request().URL.Path, "..") != 0 {
			return c.String(http.StatusForbidden, "path not allowed")
		}
		return next(c)
	}
}

func setupRoutes(e *echo.Echo) {

	e.Pre(enforceMaxOneDotPerURL)
	//e.Use(middleware.CSRF())

	e.Group("/*").Use(appBoxMiddleware())

	// Pass all the requests through to the local rest API
	e.Group("/api", apiMiddlewares()...)
}
<file_sep>package autopeering

import (
	"strings"
	"time"

	"github.com/libp2p/go-libp2p-core/crypto"
	libp2p "github.com/libp2p/go-libp2p-core/peer"
	"github.com/pkg/errors"
	"go.uber.org/dig"

	"github.com/iotaledger/hive.go/crypto/ed25519"

	databaseCore "github.com/gohornet/hornet/core/database"
	"github.com/gohornet/hornet/core/gossip"
	"github.com/gohornet/hornet/core/pow"
	"github.com/gohornet/hornet/core/snapshot"
	"github.com/gohornet/hornet/core/tangle"
	"github.com/gohornet/hornet/pkg/database"
	"github.com/gohornet/hornet/pkg/node"
	"github.com/gohornet/hornet/pkg/p2p"
	"github.com/gohornet/hornet/pkg/p2p/autopeering"
	"github.com/gohornet/hornet/pkg/shutdown"
	"github.com/gohornet/hornet/plugins/coordinator"
	"github.com/gohornet/hornet/plugins/dashboard"
	"github.com/gohornet/hornet/plugins/debug"
	"github.com/gohornet/hornet/plugins/faucet"
	"github.com/gohornet/hornet/plugins/migrator"
	"github.com/gohornet/hornet/plugins/mqtt"
	"github.com/gohornet/hornet/plugins/prometheus"
	"github.com/gohornet/hornet/plugins/receipt"
	restapiv1 "github.com/gohornet/hornet/plugins/restapi/v1"
	"github.com/gohornet/hornet/plugins/spammer"
	"github.com/gohornet/hornet/plugins/urts"
	"github.com/gohornet/hornet/plugins/warpsync"

	"github.com/iotaledger/hive.go/autopeering/discover"
	"github.com/iotaledger/hive.go/autopeering/peer/service"
	"github.com/iotaledger/hive.go/autopeering/selection"
	"github.com/iotaledger/hive.go/configuration"
	"github.com/iotaledger/hive.go/events"
)

func init() {
	Plugin = &node.Plugin{
		Status: node.StatusDisabled,
		Pluggable: node.Pluggable{
			Name:       "Autopeering",
			DepsFunc:   func(cDeps dependencies) { deps = cDeps },
			Params:     params,
			PreProvide: preProvide,
			Configure:  configure,
			Run:        run,
		},
	}
}

var (
	Plugin *node.Plugin
	deps   dependencies

	localPeerContainer *autopeering.LocalPeerContainer

	onDiscoveryPeerDiscovered  *events.Closure
	onDiscoveryPeerDeleted     *events.Closure
	onSelectionSaltUpdated     *events.Closure
	onSelectionOutgoingPeering *events.Closure
	onSelectionIncomingPeering *events.Closure
	onSelectionDropped         *events.Closure
	onPeerDisconnected         *events.Closure
	onAutopeerBecameKnown      *events.Closure
)

type dependencies struct {
	dig.In
	NodeConfig                *configuration.Configuration `name:"nodeConfig"`
	NodePrivateKey            crypto.PrivKey               `name:"nodePrivateKey"`
	P2PDatabasePath           string                       `name:"p2pDatabasePath"`
	P2PBindMultiAddresses     []string                     `name:"p2pBindMultiAddresses"`
	DatabaseEngine            database.Engine              `name:"databaseEngine"`
	NetworkIDName             string                       `name:"networkIdName"`
	AutopeeringRunAsEntryNode bool                         `name:"autopeeringRunAsEntryNode"`
	Manager                   *p2p.Manager                 `optional:"true"`
}

func preProvide(c *dig.Container, configs map[string]*configuration.Configuration, initConfig *node.InitConfig) {

	pluginEnabled := true

	containsPlugin := func(pluginsList []string, pluginIdentifier string) bool {
		contains := false
		for _, plugin := range pluginsList {
			if strings.ToLower(plugin) == pluginIdentifier {
				contains = true
				break
			}
		}
		return contains
	}

	if disabled := containsPlugin(initConfig.DisabledPlugins, Plugin.Identifier()); disabled {
		// Autopeering is disabled
		pluginEnabled = false
	}

	if enabled := containsPlugin(initConfig.EnabledPlugins, Plugin.Identifier()); !enabled {
		// Autopeering was not enabled
		pluginEnabled = false
	}

	runAsEntryNode := pluginEnabled && configs["nodeConfig"].Bool(CfgNetAutopeeringRunAsEntryNode)
	if runAsEntryNode {
		// the following pluggables stay enabled
		// - profile
		// - protocfg
		// - gracefulshutdown
		// - p2p
		// - profiling
		// - versioncheck
		// - autopeering

		// disable the other plugins if the node runs as an entry node for autopeering
		initConfig.ForceDisablePluggable(databaseCore.CorePlugin.Identifier())
		initConfig.ForceDisablePluggable(pow.CorePlugin.Identifier())
		initConfig.ForceDisablePluggable(gossip.CorePlugin.Identifier())
		initConfig.ForceDisablePluggable(tangle.CorePlugin.Identifier())
		initConfig.ForceDisablePluggable(snapshot.CorePlugin.Identifier())
		initConfig.ForceDisablePluggable(restapiv1.Plugin.Identifier())
		initConfig.ForceDisablePluggable(warpsync.Plugin.Identifier())
		initConfig.ForceDisablePluggable(urts.Plugin.Identifier())
		initConfig.ForceDisablePluggable(dashboard.Plugin.Identifier())
		initConfig.ForceDisablePluggable(spammer.Plugin.Identifier())
		initConfig.ForceDisablePluggable(mqtt.Plugin.Identifier())
		initConfig.ForceDisablePluggable(coordinator.Plugin.Identifier())
		initConfig.ForceDisablePluggable(migrator.Plugin.Identifier())
		initConfig.ForceDisablePluggable(receipt.Plugin.Identifier())
		initConfig.ForceDisablePluggable(prometheus.Plugin.Identifier())
		initConfig.ForceDisablePluggable(debug.Plugin.Identifier())
		initConfig.ForceDisablePluggable(faucet.Plugin.Identifier())
	}

	// the parameter has to be provided in the preProvide stage.
	// this is a special case, since it only should be true if the plugin is enabled
	type cfgResult struct {
		dig.Out
		AutopeeringRunAsEntryNode bool `name:"autopeeringRunAsEntryNode"`
	}

	if err := c.Provide(func() cfgResult {
		return cfgResult{
			AutopeeringRunAsEntryNode: runAsEntryNode,
		}
	}); err != nil {
		Plugin.Panic(err)
	}
}

func configure() {
	selection.SetParameters(selection.Parameters{
		InboundNeighborSize:  deps.NodeConfig.Int(CfgNetAutopeeringInboundPeers),
		OutboundNeighborSize: deps.NodeConfig.Int(CfgNetAutopeeringOutboundPeers),
		SaltLifetime:         deps.NodeConfig.Duration(CfgNetAutopeeringSaltLifetime),
	})

	if err := autopeering.RegisterAutopeeringProtocolInMultiAddresses(); err != nil {
		Plugin.Panicf("unable to register autopeering protocol for multi addresses: %s", err)
	}

	rawPrvKey, err := deps.NodePrivateKey.Raw()
	if err != nil {
		Plugin.Panicf("unable to obtain raw private key: %s", err)
	}

	localPeerContainer, err = autopeering.NewLocalPeerContainer(p2pServiceKey(),
		rawPrvKey[:ed25519.SeedSize],
		deps.P2PDatabasePath,
		deps.DatabaseEngine,
		deps.P2PBindMultiAddresses,
		deps.NodeConfig.String(CfgNetAutopeeringBindAddr),
		deps.AutopeeringRunAsEntryNode)
	if err != nil {
		Plugin.Panicf("unable to initialize local peer container: %s", err)
	}

	Plugin.LogInfof("initialized local autopeering: %s@%s", localPeerContainer.Local().PublicKey(), localPeerContainer.Local().Address())

	if deps.AutopeeringRunAsEntryNode {
		entryNodeMultiAddress, err := autopeering.GetEntryNodeMultiAddress(localPeerContainer.Local())
		if err != nil {
			Plugin.Panicf("unable to parse entry node multiaddress: %s", err)
		}

		Plugin.LogInfof("\n\nentry node multiaddress: %s\n", entryNodeMultiAddress.String())
	}

	configureAutopeering(localPeerContainer)
	configureEvents()
}

func run() {
	if err := Plugin.Node.Daemon().BackgroundWorker(Plugin.Name, func(shutdownSignal <-chan struct{}) {
		attachEvents()
		start(localPeerContainer, shutdownSignal)
		detachEvents()
	}, shutdown.PriorityAutopeering); err != nil {
		Plugin.Panicf("failed to start worker: %s", err)
	}
}

// gets the peering service key from the config.
func p2pServiceKey() service.Key {
	return service.Key(deps.NetworkIDName)
}

func configureEvents() {

	onDiscoveryPeerDiscovered = events.NewClosure(func(ev *discover.DiscoveredEvent) {
		if peerID := autopeering.ConvertHivePubKeyToPeerIDOrLog(ev.Peer.PublicKey(), Plugin.LogWarnf); peerID != nil {
			Plugin.LogInfof("discovered: %s / %s", ev.Peer.Address(), *peerID)
		}
	})

	onDiscoveryPeerDeleted = events.NewClosure(func(ev *discover.DeletedEvent) {
		if peerID := autopeering.ConvertHivePubKeyToPeerIDOrLog(ev.Peer.PublicKey(), Plugin.LogWarnf); peerID != nil {
			Plugin.LogInfof("removed offline: %s / %s", ev.Peer.Address(), *peerID)
		}
	})

	onPeerDisconnected = events.NewClosure(func(peerOptErr *p2p.PeerOptError) {
		if peerOptErr.Peer.Relation != p2p.PeerRelationAutopeered {
			return
		}

		if id := autopeering.ConvertPeerIDToHiveIdentityOrLog(peerOptErr.Peer, Plugin.LogWarnf); id != nil {
			Plugin.LogInfof("removing: %s", peerOptErr.Peer.ID)
			selectionProtocol.RemoveNeighbor(id.ID())
		}
	})

	onAutopeerBecameKnown = events.NewClosure(func(p *p2p.Peer, oldRel p2p.PeerRelation) {
		if oldRel != p2p.PeerRelationAutopeered {
			return
		}
		if id := autopeering.ConvertPeerIDToHiveIdentityOrLog(p, Plugin.LogWarnf); id != nil {
			Plugin.LogInfof("removing %s from autopeering selection protocol", p.ID)
			selectionProtocol.RemoveNeighbor(id.ID())
		}
	})

	onSelectionSaltUpdated = events.NewClosure(func(ev *selection.SaltUpdatedEvent) {
		Plugin.LogInfof("salt updated; expires=%s", ev.Public.GetExpiration().Format(time.RFC822))
	})

	onSelectionOutgoingPeering = events.NewClosure(func(ev *selection.PeeringEvent) {
		if !ev.Status {
			return
		}
		Plugin.LogInfof("[outgoing peering] adding autopeering peer %s", ev.Peer.ID())

		addrInfo, err := autopeering.HivePeerToAddrInfo(ev.Peer, p2pServiceKey())
		if err != nil {
			Plugin.LogWarnf("unable to convert outgoing selection autopeering peer to addr info: %s", err)
			return
		}

		handleSelection(ev, addrInfo, func() {
			Plugin.LogInfof("connecting to %s", addrInfo)
			if err := deps.Manager.ConnectPeer(addrInfo, p2p.PeerRelationAutopeered); err != nil {
				Plugin.LogWarnf("couldn't add autopeering peer %s", err)
			}
		})
	})

	onSelectionIncomingPeering = events.NewClosure(func(ev *selection.PeeringEvent) {
		if !ev.Status {
			return
		}
		Plugin.LogInfof("[incoming peering] whitelisting %s", ev.Peer.ID())

		addrInfo, err := autopeering.HivePeerToAddrInfo(ev.Peer, p2pServiceKey())
		if err != nil {
			Plugin.LogWarnf("unable to convert incoming selection autopeering peer to addr info: %s", err)
			return
		}

		handleSelection(ev, addrInfo, func() {
			// TODO: maybe do whitelisting instead?
			//Plugin.LogInfof("connecting to %s", addrInfo)
			//if err := deps.Manager.ConnectPeer(addrInfo, p2p.PeerRelationAutopeered); err != nil {
			//	Plugin.LogWarnf("couldn't add autopeering peer %s", err)
			//}
		})
	})

	onSelectionDropped = events.NewClosure(func(ev *selection.DroppedEvent) {
		peerID := autopeering.ConvertHivePubKeyToPeerIDOrLog(ev.Peer.PublicKey(), Plugin.LogWarnf)
		if peerID == nil {
			return
		}

		Plugin.LogInfof("[dropped event] disconnecting %s", peerID)
		var peerRelation p2p.PeerRelation
		deps.Manager.Call(*peerID, func(p *p2p.Peer) {
			peerRelation = p.Relation
		})

		if len(peerRelation) == 0 {
			Plugin.LogWarnf("didn't find autopeered peer %s for disconnecting", peerID)
			return
		}

		if peerRelation != p2p.PeerRelationAutopeered {
			Plugin.LogWarnf("won't disconnect %s as it its relation is not '%s' but '%s'", peerID, p2p.PeerRelationAutopeered, peerRelation)
			return
		}

		if err := deps.Manager.DisconnectPeer(*peerID, errors.New("removed via autopeering selection")); err != nil {
			Plugin.LogWarnf("couldn't disconnect selection dropped autopeer: %s", err)
		}
	})
}

// handles a peer gotten from the autopeering selection according to its existing relation.
// if the peer is not yet part of the peering manager, the given noRelationFunc is called.
func handleSelection(ev *selection.PeeringEvent, addrInfo *libp2p.AddrInfo, noRelationFunc func()) {
	// extract peer relation
	var peerRelation p2p.PeerRelation
	deps.Manager.Call(addrInfo.ID, func(p *p2p.Peer) {
		peerRelation = p.Relation
	})

	switch peerRelation {
	case p2p.PeerRelationKnown:
		clearFromAutopeeringSelector(ev)

	case p2p.PeerRelationUnknown:
		updatePeerRelationToDiscovered(addrInfo)

	case p2p.PeerRelationAutopeered:
		handleAlreadyAutopeered(addrInfo)

	default:
		noRelationFunc()
	}
}

// logs a warning about a from the selector seen peer which is already autopeered.
func handleAlreadyAutopeered(addrInfo *libp2p.AddrInfo) {
	Plugin.LogWarnf("peer is already autopeered %s", addrInfo.ID)
}

// updates the given peers relation to discovered.
func updatePeerRelationToDiscovered(addrInfo *libp2p.AddrInfo) {
	if err := deps.Manager.ConnectPeer(addrInfo, p2p.PeerRelationAutopeered); err != nil {
		Plugin.LogWarnf("couldn't update unknown peer to 'discovered' %s", err)
	}
}

// clears an already statically peered from the autopeering selector.
func clearFromAutopeeringSelector(ev *selection.PeeringEvent) {
	Plugin.LogInfof("peer is statically peered already %s, removing from autopeering selection protocol", ev.Peer.ID())
	selectionProtocol.RemoveNeighbor(ev.Peer.ID())
}

func attachEvents() {
	discoveryProtocol.Events().PeerDiscovered.Attach(onDiscoveryPeerDiscovered)
	discoveryProtocol.Events().PeerDeleted.Attach(onDiscoveryPeerDeleted)

	if deps.Manager == nil {
		return
	}

	// notify the selection when a connection is closed or failed.
	deps.Manager.Events.Disconnected.Attach(onPeerDisconnected)
	deps.Manager.Events.RelationUpdated.Attach(onAutopeerBecameKnown)
	selectionProtocol.Events().SaltUpdated.Attach(onSelectionSaltUpdated)
	selectionProtocol.Events().OutgoingPeering.Attach(onSelectionOutgoingPeering)
	selectionProtocol.Events().IncomingPeering.Attach(onSelectionIncomingPeering)
	selectionProtocol.Events().Dropped.Attach(onSelectionDropped)
}

func detachEvents() {
	discoveryProtocol.Events().PeerDiscovered.Detach(onDiscoveryPeerDiscovered)
	discoveryProtocol.Events().PeerDeleted.Detach(onDiscoveryPeerDeleted)

	if deps.Manager == nil {
		return
	}

	deps.Manager.Events.Disconnected.Detach(onPeerDisconnected)
	deps.Manager.Events.RelationUpdated.Detach(onAutopeerBecameKnown)
	selectionProtocol.Events().SaltUpdated.Detach(onSelectionSaltUpdated)
	selectionProtocol.Events().OutgoingPeering.Detach(onSelectionOutgoingPeering)
	selectionProtocol.Events().IncomingPeering.Detach(onSelectionIncomingPeering)
	selectionProtocol.Events().Dropped.Detach(onSelectionDropped)
}
<file_sep>package protocfg

import (
	"encoding/json"

	flag "github.com/spf13/pflag"
	"go.uber.org/dig"

	"github.com/gohornet/hornet/pkg/model/coordinator"
	"github.com/gohornet/hornet/pkg/node"
	"github.com/iotaledger/hive.go/configuration"
	iotago "github.com/iotaledger/iota.go/v2"
)

func init() {
	_ = flag.CommandLine.MarkHidden(CfgProtocolPublicKeyRangesJSON)

	CorePlugin = &node.CorePlugin{
		Pluggable: node.Pluggable{
			Name:           "ProtoCfg",
			Params:         params,
			InitConfigPars: initConfigPars,
		},
	}
}

var (
	CorePlugin *node.CorePlugin

	cooPubKeyRangesFlag = flag.String(CfgProtocolPublicKeyRangesJSON, "", "overwrite public key ranges (JSON)")
)

func initConfigPars(c *dig.Container) {

	type cfgDeps struct {
		dig.In
		NodeConfig *configuration.Configuration `name:"nodeConfig"`
	}

	type cfgResult struct {
		dig.Out
		PublicKeyRanges         coordinator.PublicKeyRanges
		NetworkID               uint64               `name:"networkId"`
		NetworkIDName           string               `name:"networkIdName"`
		Bech32HRP               iotago.NetworkPrefix `name:"bech32HRP"`
		MinPoWScore             float64              `name:"minPoWScore"`
		MilestonePublicKeyCount int                  `name:"milestonePublicKeyCount"`
	}

	if err := c.Provide(func(deps cfgDeps) cfgResult {

		res := cfgResult{
			NetworkID:               iotago.NetworkIDFromString(deps.NodeConfig.String(CfgProtocolNetworkIDName)),
			NetworkIDName:           deps.NodeConfig.String(CfgProtocolNetworkIDName),
			Bech32HRP:               iotago.NetworkPrefix(deps.NodeConfig.String(CfgProtocolBech32HRP)),
			MinPoWScore:             deps.NodeConfig.Float64(CfgProtocolMinPoWScore),
			MilestonePublicKeyCount: deps.NodeConfig.Int(CfgProtocolMilestonePublicKeyCount),
		}

		if *cooPubKeyRangesFlag != "" {
			// load from special CLI flag
			if err := json.Unmarshal([]byte(*cooPubKeyRangesFlag), &res.PublicKeyRanges); err != nil {
				CorePlugin.Panic(err)
			}
			return res
		}

		if err := deps.NodeConfig.SetDefault(CfgProtocolPublicKeyRanges, coordinator.PublicKeyRanges{
			{
				Key:        "<KEY>",
				StartIndex: 0,
				EndIndex:   777600,
			}, {
				Key:        "365fb85e7568b9b32f7359d6cbafa9814472ad0ecbad32d77beaf5dd9e84c6ba",
				StartIndex: 0,
				EndIndex:   1555200,
			}, {
				Key:        "<KEY>",
				StartIndex: 552960,
				EndIndex:   2108160,
			}, {
				Key:        "760d88e112c0fd210cf16a3dce3443ecf7e18c456c2fb9646cabb2e13e367569",
				StartIndex: 1333460,
				EndIndex:   2888660,
			}, {
				Key:        "7bac2209b576ea2235539358c7df8ca4d2f2fc35a663c760449e65eba9f8a6e7",
				StartIndex: 2111060,
				EndIndex:   3666260,
			}, {
				Key:        "<KEY>",
				StartIndex: 2888660,
				EndIndex:   4443860,
			},
		}); err != nil {
			CorePlugin.Panic(err)
		}

		// load from config
		if err := deps.NodeConfig.Unmarshal(CfgProtocolPublicKeyRanges, &res.PublicKeyRanges); err != nil {
			CorePlugin.Panic(err)
		}

		return res
	}); err != nil {
		CorePlugin.Panic(err)
	}
}
<file_sep>package keymanager_test

import (
	"testing"

	"github.com/stretchr/testify/assert"

	"github.com/gohornet/hornet/pkg/keymanager"
	iotago "github.com/iotaledger/iota.go/v2"
	"github.com/iotaledger/iota.go/v2/ed25519"
)

func TestMilestoneKeyManager(t *testing.T) {

	pubKey1, privKey1, err := ed25519.GenerateKey(nil)
	assert.NoError(t, err)

	pubKey2, privKey2, err := ed25519.GenerateKey(nil)
	assert.NoError(t, err)

	pubKey3, privKey3, err := ed25519.GenerateKey(nil)
	assert.NoError(t, err)

	km := keymanager.New()
	km.AddKeyRange(pubKey1, 0, 0)
	km.AddKeyRange(pubKey2, 3, 10)
	km.AddKeyRange(pubKey3, 8, 15)

	keysIndex0 := km.PublicKeysForMilestoneIndex(0)
	assert.Len(t, keysIndex0, 1)

	keysIndex3 := km.PublicKeysForMilestoneIndex(3)
	assert.Len(t, keysIndex3, 2)

	keysIndex7 := km.PublicKeysForMilestoneIndex(7)
	assert.Len(t, keysIndex7, 2)

	keysIndex8 := km.PublicKeysForMilestoneIndex(8)
	assert.Len(t, keysIndex8, 3)

	keysIndex10 := km.PublicKeysForMilestoneIndex(10)
	assert.Len(t, keysIndex10, 3)

	keysIndex11 := km.PublicKeysForMilestoneIndex(11)
	assert.Len(t, keysIndex11, 2)

	keysIndex15 := km.PublicKeysForMilestoneIndex(15)
	assert.Len(t, keysIndex15, 2)

	keysIndex16 := km.PublicKeysForMilestoneIndex(16)
	assert.Len(t, keysIndex16, 1)

	keysIndex1000 := km.PublicKeysForMilestoneIndex(1000)
	assert.Len(t, keysIndex1000, 1)

	keysSet8 := km.PublicKeysSetForMilestoneIndex(8)
	assert.Len(t, keysSet8, 3)

	var msPubKey1 iotago.MilestonePublicKey
	copy(msPubKey1[:], pubKey1)

	var msPubKey2 iotago.MilestonePublicKey
	copy(msPubKey2[:], pubKey2)

	var msPubKey3 iotago.MilestonePublicKey
	copy(msPubKey3[:], pubKey3)

	assert.Contains(t, keysSet8, msPubKey1)
	assert.Contains(t, keysSet8, msPubKey2)
	assert.Contains(t, keysSet8, msPubKey3)

	keyMapping8 := km.MilestonePublicKeyMappingForMilestoneIndex(8, []ed25519.PrivateKey{privKey1, privKey2, privKey3}, 2)
	assert.Len(t, keyMapping8, 2)

	assert.Equal(t, keyMapping8[msPubKey1], privKey1)
	assert.Equal(t, keyMapping8[msPubKey2], privKey2)
}
<file_sep>package database

import (
	"runtime"

	"github.com/iotaledger/hive.go/kvstore/rocksdb"
)

// NewRocksDB creates a new RocksDB instance.
func NewRocksDB(path string) (*rocksdb.RocksDB, error) {

	opts := []rocksdb.Option{
		rocksdb.IncreaseParallelism(runtime.NumCPU() - 1),
	}

	return rocksdb.CreateDB(path, opts...)
}
<file_sep>package faucet

// faucetEnqueueRequest defines the request for a POST RouteFaucetEnqueue REST API call.
type faucetEnqueueRequest struct {
	// The bech32 address.
	Address string `json:"address"`
}
<file_sep>package faucet

import (
	"bytes"
	"fmt"
	"runtime"
	"time"

	"github.com/labstack/echo/v4"
	"github.com/pkg/errors"

	"github.com/gohornet/hornet/pkg/common"
	"github.com/gohornet/hornet/pkg/dag"
	"github.com/gohornet/hornet/pkg/model/hornet"
	"github.com/gohornet/hornet/pkg/model/milestone"
	"github.com/gohornet/hornet/pkg/model/storage"
	"github.com/gohornet/hornet/pkg/model/utxo"
	"github.com/gohornet/hornet/pkg/pow"
	"github.com/gohornet/hornet/pkg/restapi"
	"github.com/iotaledger/hive.go/events"
	"github.com/iotaledger/hive.go/logger"
	"github.com/iotaledger/hive.go/syncutils"
	iotago "github.com/iotaledger/iota.go/v2"
)

// SendMessageFunc is a function which sends a message to the network.
type SendMessageFunc = func(msg *storage.Message) error

// TipselFunc selects tips for the faucet.
type TipselFunc = func() (tips hornet.MessageIDs, err error)

var (
	// ErrNoTipsGiven is returned when no tips were given to issue a message.
	ErrNoTipsGiven = errors.New("no tips given")
)

// Events are the events issued by the faucet.
type Events struct {
	// Fired when a faucet message is issued.
	IssuedMessage *events.Event
	// SoftError is triggered when a soft error is encountered.
	SoftError *events.Event
}

// queueItem is an item for the faucet requests queue.
type queueItem struct {
	Bech32         string
	Amount         uint64
	Ed25519Address *iotago.Ed25519Address
}

// FaucetInfoResponse defines the response of a GET RouteFaucetInfo REST API call.
type FaucetInfoResponse struct {
	// The bech32 address of the faucet.
	Address string `json:"address"`
	// The remaining balance of faucet.
	Balance uint64 `json:"balance"`
}

// FaucetEnqueueResponse defines the response of a POST RouteFaucetEnqueue REST API call.
type FaucetEnqueueResponse struct {
	// The bech32 address.
	Address string `json:"address"`
	// The number of waiting requests in the queue.
	WaitingRequests int `json:"waitingRequests"`
}

// Faucet is used to issue transaction to users that requested funds via a REST endpoint.
type Faucet struct {
	syncutils.Mutex

	// used to access the node storage.
	storage *storage.Storage
	// id of the network the faucet is running in.
	networkID uint64
	// belowMaxDepth is the maximum allowed delta
	// value between OCRI of a given message in relation to the current CMI before it gets lazy.
	belowMaxDepth milestone.Index
	// used to get the outputs.
	utxoManager *utxo.Manager
	// the address of the faucet.
	address *iotago.Ed25519Address
	// used to sign the faucet transactions.
	addressSigner iotago.AddressSigner
	// used to get valid tips for new faucet messages.
	tipselFunc TipselFunc
	// used to do the PoW for the faucet messages.
	powHandler *pow.Handler
	// the function used to send a message.
	sendMessageFunc SendMessageFunc
	// holds the faucet options.
	opts *Options

	// events of the faucet.
	Events *Events

	// the message ID of the last sent faucet message.
	lastMessageID hornet.MessageID
	// map with all queued requests per address.
	queueMap map[string]*queueItem
	// queue of new requests.
	queue chan *queueItem
}

// the default options applied to the faucet.
var defaultOptions = []Option{
	WithHRPNetworkPrefix(iotago.PrefixTestnet),
	WithAmount(10000000),            // 10 Mi
	WithSmallAmount(1000000),        // 1 Mi
	WithMaxAddressBalance(20000000), // 20 Mi
	WithMaxOutputCount(iotago.MaxOutputsCount),
	WithIndexationMessage("HORNET FAUCET"),
	WithBatchTimeout(2 * time.Second),
	WithPowWorkerCount(0),
}

// Options define options for the faucet.
type Options struct {
	logger *logger.Logger

	hrpNetworkPrefix  iotago.NetworkPrefix
	amount            uint64
	smallAmount       uint64
	maxAddressBalance uint64
	maxOutputCount    int
	indexationMessage []byte
	batchTimeout      time.Duration
	powWorkerCount    int
}

// applies the given Option.
func (so *Options) apply(opts ...Option) {
	for _, opt := range opts {
		opt(so)
	}
}

// WithLogger enables logging within the faucet.
func WithLogger(logger *logger.Logger) Option {
	return func(opts *Options) {
		opts.logger = logger
	}
}

// WithHRPNetworkPrefix sets the bech32 HRP network prefix.
func WithHRPNetworkPrefix(networkPrefix iotago.NetworkPrefix) Option {
	return func(opts *Options) {
		opts.hrpNetworkPrefix = networkPrefix
	}
}

// WithAmount defines the amount of funds the requester receives.
func WithAmount(amount uint64) Option {
	return func(opts *Options) {
		opts.amount = amount
	}
}

// WithSmallAmount defines the amount of funds the requester receives
// if the target address has more funds than the faucet amount and less than maximum.
func WithSmallAmount(smallAmount uint64) Option {
	return func(opts *Options) {
		opts.smallAmount = smallAmount
	}
}

// WithMaxAddressBalance defines the maximum allowed amount of funds on the target address.
// If there are more funds already, the faucet request is rejected.
func WithMaxAddressBalance(maxAddressBalance uint64) Option {
	return func(opts *Options) {
		opts.maxAddressBalance = maxAddressBalance
	}
}

// WithMaxOutputCount defines the maximum output count per faucet message.
func WithMaxOutputCount(maxOutputCount int) Option {
	return func(opts *Options) {
		if maxOutputCount > iotago.MaxOutputsCount {
			maxOutputCount = iotago.MaxOutputsCount
		}
		opts.maxOutputCount = maxOutputCount
	}
}

// WithIndexationMessage defines the faucet transaction indexation payload.
func WithIndexationMessage(indexationMessage string) Option {
	return func(opts *Options) {
		opts.indexationMessage = []byte(indexationMessage)
	}
}

// WithBatchTimeout sets the maximum duration for collecting faucet batches.
func WithBatchTimeout(timeout time.Duration) Option {
	return func(opts *Options) {
		opts.batchTimeout = timeout
	}
}

// WithPowWorkerCount defines the amount of workers used for calculating PoW when issuing faucet messages.
func WithPowWorkerCount(powWorkerCount int) Option {

	if powWorkerCount == 0 {
		powWorkerCount = runtime.NumCPU() - 1
	}

	if powWorkerCount < 1 {
		powWorkerCount = 1
	}

	return func(opts *Options) {
		opts.powWorkerCount = powWorkerCount
	}
}

// Option is a function setting a faucet option.
type Option func(opts *Options)

// New creates a new faucet instance.
func New(
	storage *storage.Storage,
	networkID uint64,
	belowMaxDepth int,
	utxoManager *utxo.Manager,
	address *iotago.Ed25519Address,
	addressSigner iotago.AddressSigner,
	tipselFunc TipselFunc,
	powHandler *pow.Handler,
	sendMessageFunc SendMessageFunc,
	opts ...Option) *Faucet {

	options := &Options{}
	options.apply(defaultOptions...)
	options.apply(opts...)

	faucet := &Faucet{
		storage:         storage,
		networkID:       networkID,
		belowMaxDepth:   milestone.Index(belowMaxDepth),
		utxoManager:     utxoManager,
		address:         address,
		addressSigner:   addressSigner,
		tipselFunc:      tipselFunc,
		powHandler:      powHandler,
		sendMessageFunc: sendMessageFunc,
		opts:            options,

		Events: &Events{
			IssuedMessage: events.NewEvent(events.VoidCaller),
			SoftError:     events.NewEvent(events.ErrorCaller),
		},
	}
	faucet.init()

	return faucet
}

func (f *Faucet) init() {
	f.queue = make(chan *queueItem, 5000)
	f.queueMap = make(map[string]*queueItem)
	f.lastMessageID = hornet.NullMessageID()
}

// NetworkPrefix returns the used network prefix.
func (f *Faucet) NetworkPrefix() iotago.NetworkPrefix {
	return f.opts.hrpNetworkPrefix
}

// Info returns the used faucet address and remaining balance.
func (f *Faucet) Info() (*FaucetInfoResponse, error) {
	balance, _, err := f.utxoManager.AddressBalanceWithoutLocking(f.address)
	if err != nil {
		return nil, err
	}

	return &FaucetInfoResponse{
		Address: f.address.Bech32(f.opts.hrpNetworkPrefix),
		Balance: balance,
	}, nil
}

// Enqueue adds a new faucet request to the queue.
func (f *Faucet) Enqueue(bech32 string, ed25519Addr *iotago.Ed25519Address) (*FaucetEnqueueResponse, error) {
	f.Lock()
	defer f.Unlock()

	if _, exists := f.queueMap[bech32]; exists {
		return nil, errors.WithMessage(restapi.ErrInvalidParameter, "Address is already in the queue.")
	}

	amount := f.opts.amount
	balance, _, err := f.utxoManager.AddressBalanceWithoutLocking(ed25519Addr)
	if err == nil && balance >= f.opts.amount {
		amount = f.opts.smallAmount

		if balance >= f.opts.maxAddressBalance {
			return nil, errors.WithMessage(restapi.ErrInvalidParameter, "You already have enough coins on your address.")
		}
	}

	request := &queueItem{
		Bech32:         bech32,
		Amount:         amount,
		Ed25519Address: ed25519Addr,
	}

	select {
	case f.queue <- request:
		f.queueMap[bech32] = request
		return &FaucetEnqueueResponse{
			Address:         bech32,
			WaitingRequests: len(f.queueMap),
		}, nil

	default:
		// queue is full
		return nil, errors.WithMessage(echo.ErrInternalServerError, "Faucet queue is full. Please try again later!")
	}
}

// clearRequests clears the old requests from the map.
// this is necessary to be able to send new requests to the same addresses.
func (f *Faucet) clearRequests(batchedRequests []*queueItem) {
	f.Lock()
	defer f.Unlock()

	for _, request := range batchedRequests {
		delete(f.queueMap, request.Bech32)
	}
}

// createMessage creates a new message and references the last faucet message (also reattaches if below max depth).
func (f *Faucet) createMessage(txPayload iotago.Serializable, shutdownSignal <-chan struct{}) (*storage.Message, error) {

	tips, err := f.tipselFunc()
	if err != nil {
		return nil, err
	}

	reattachMessage := func(messageID hornet.MessageID) (*storage.Message, error) {
		cachedMsg := f.storage.CachedMessageOrNil(f.lastMessageID)
		if cachedMsg == nil {
			// message unknown
			return nil, fmt.Errorf("message not found: %s", messageID.ToHex())
		}
		defer cachedMsg.Release(true)

		tips, err := f.tipselFunc()
		if err != nil {
			return nil, err
		}

		iotaMsg := &iotago.Message{
			NetworkID: f.networkID,
			Parents:   tips.ToSliceOfArrays(),
			Payload:   cachedMsg.Message().Message().Payload,
		}

		if err := f.powHandler.DoPoW(iotaMsg, shutdownSignal, 1); err != nil {
			return nil, err
		}

		msg, err := storage.NewMessage(iotaMsg, iotago.DeSeriModePerformValidation)
		if err != nil {
			return nil, err
		}

		if err := f.sendMessageFunc(msg); err != nil {
			return nil, err
		}

		return msg, nil
	}

	// check if the last faucet message was already confirmed.
	// if not, check if it is already below max depth and reattach in case.
	// we need to check for the last faucet message, because we reference the last message as a tip
	// to be sure the tangle consumes our UTXOs in the correct order.
	if err = func() error {
		if bytes.Equal(f.lastMessageID, hornet.NullMessageID()) {
			// do not reference NullMessage
			return nil
		}

		cachedMsgMeta := f.storage.CachedMessageMetadataOrNil(f.lastMessageID)
		if cachedMsgMeta == nil {
			// message unknown
			return nil
		}
		defer cachedMsgMeta.Release(true)

		if cachedMsgMeta.Metadata().IsReferenced() {
			// message is already confirmed, no need to reference
			return nil
		}

		_, ocri := dag.ConeRootIndexes(f.storage, cachedMsgMeta.Retain(), f.storage.ConfirmedMilestoneIndex()) // meta +
		if (f.storage.LatestMilestoneIndex() - ocri) > f.belowMaxDepth {
			// the last faucet message is not confirmed yet, but it is already below max depth
			// we need to reattach it
			msg, err := reattachMessage(f.lastMessageID)
			if err != nil {
				return common.CriticalError(fmt.Errorf("faucet message was below max depth and couldn't be reattached: %w", err))
			}

			// update the lastMessasgeID because we reattached the message
			f.lastMessageID = msg.MessageID()
		}

		tips[0] = f.lastMessageID
		tips = tips.RemoveDupsAndSortByLexicalOrder()

		return nil
	}(); err != nil {
		return nil, err
	}

	// create the message
	iotaMsg := &iotago.Message{
		NetworkID: f.networkID,
		Parents:   tips.ToSliceOfArrays(),
		Payload:   txPayload,
	}

	if err := f.powHandler.DoPoW(iotaMsg, shutdownSignal, 1); err != nil {
		return nil, err
	}

	msg, err := storage.NewMessage(iotaMsg, iotago.DeSeriModePerformValidation)
	if err != nil {
		return nil, err
	}

	return msg, nil
}

// buildTransactionPayload creates a signed transaction payload with all UTXO and batched requests.
func (f *Faucet) buildTransactionPayload(unspentOutputs []*utxo.Output, batchedRequests []*queueItem) (*iotago.Transaction, *iotago.UTXOInput, uint64, error) {

	txBuilder := iotago.NewTransactionBuilder()
	txBuilder.AddIndexationPayload(&iotago.Indexation{Index: f.opts.indexationMessage, Data: nil})

	outputCount := 0
	var remainderAmount int64 = 0

	// collect all unspent output of the faucet address
	for _, unspentOutput := range unspentOutputs {
		outputCount++
		remainderAmount += int64(unspentOutput.Amount())
		txBuilder.AddInput(&iotago.ToBeSignedUTXOInput{Address: f.address, Input: unspentOutput.UTXOInput()})
	}

	// add all requests as outputs
	for _, req := range batchedRequests {
		outputCount++

		if outputCount >= f.opts.maxOutputCount {
			// do not collect further requests
			// the last slot is for the remainder
			break
		}

		if remainderAmount == 0 {
			// do not collect further requests
			break
		}

		amount := req.Amount
		if remainderAmount < int64(amount) {
			// not enough funds left
			amount = uint64(remainderAmount)
		}
		remainderAmount -= int64(amount)

		txBuilder.AddOutput(&iotago.SigLockedSingleOutput{Address: req.Ed25519Address, Amount: uint64(amount)})
	}

	if remainderAmount > 0 {
		txBuilder.AddOutput(&iotago.SigLockedSingleOutput{Address: f.address, Amount: uint64(remainderAmount)})
	}

	txPayload, err := txBuilder.Build(f.addressSigner)
	if err != nil {
		return nil, nil, 0, err
	}

	if remainderAmount == 0 {
		// no remainder available
		return txPayload, nil, 0, nil
	}

	transactionID, err := txPayload.ID()
	if err != nil {
		return nil, nil, 0, fmt.Errorf("can't compute the transaction ID, error: %w", err)
	}

	remainderOutput := &iotago.UTXOInput{}
	copy(remainderOutput.TransactionID[:], transactionID[:iotago.TransactionIDLength])

	// search remainder address in the outputs
	found := false
	var outputIndex uint16 = 0
	for _, output := range txPayload.Essence.(*iotago.TransactionEssence).Outputs {
		sigLock := output.(*iotago.SigLockedSingleOutput)
		ed25519Addr := sigLock.Address.(*iotago.Ed25519Address)

		if bytes.Equal(ed25519Addr[:], f.address[:]) {
			// found the remainder address in the outputs
			found = true
			remainderOutput.TransactionOutputIndex = outputIndex
			break
		}
		outputIndex++
	}

	if !found {
		return nil, nil, 0, errors.New("can't find the faucet remainder output")
	}

	return txPayload, remainderOutput, uint64(remainderAmount), nil
}

// sendFaucetMessage creates a faucet transaction payload and remembers the last sent messageID.
func (f *Faucet) sendFaucetMessage(unspentOutputs []*utxo.Output, batchedRequests []*queueItem, shutdownSignal <-chan struct{}) (*utxo.Output, error) {

	txPayload, remainderIotaGoOutput, remainderAmount, err := f.buildTransactionPayload(unspentOutputs, batchedRequests)
	if err != nil {
		return nil, fmt.Errorf("build transaction payload failed, error: %w", err)
	}

	msg, err := f.createMessage(txPayload, shutdownSignal)
	if err != nil {
		return nil, fmt.Errorf("build faucet message failed, error: %w", err)
	}

	if err := f.sendMessageFunc(msg); err != nil {
		return nil, fmt.Errorf("send faucet message failed, error: %w", err)
	}

	f.lastMessageID = msg.MessageID()
	remainderIotaGoOutputID := remainderIotaGoOutput.ID()
	remainderOutput := utxo.CreateOutput(&remainderIotaGoOutputID, msg.MessageID(), iotago.OutputSigLockedSingleOutput, f.address, uint64(remainderAmount))

	return remainderOutput, nil
}

// logSoftError logs a soft error and triggers the event.
func (f *Faucet) logSoftError(err error) {
	if f.opts.logger != nil {
		f.opts.logger.Warn(err)
	}
	f.Events.SoftError.Trigger(err)
}

// RunFaucetLoop collects unspent outputs on the faucet address and batches the requests from the queue.
func (f *Faucet) RunFaucetLoop(shutdownSignal <-chan struct{}) error {

	var lastRemainderOutput *utxo.Output

	for {
		select {
		case <-shutdownSignal:
			// faucet was stopped
			return nil

		default:

			// only collect unspent outputs if the lastRemainderOutput is not pending
			shouldCollectUnspentOutputs := func() bool {
				if lastRemainderOutput == nil {
					return true
				}

				if _, err := f.utxoManager.ReadOutputByOutputIDWithoutLocking(lastRemainderOutput.OutputID()); err != nil {
					return false
				}

				return true
			}

			var err error
			unspentOutputs := []*utxo.Output{}
			if shouldCollectUnspentOutputs() {
				unspentOutputs, err = f.utxoManager.UnspentOutputs(utxo.FilterAddress(f.address), utxo.ReadLockLedger(false), utxo.MaxResultCount(f.opts.maxOutputCount-2), utxo.FilterOutputType(iotago.OutputSigLockedSingleOutput))
				if err != nil {
					return fmt.Errorf("reading unspent outputs failed: %s, error: %w", f.address.Bech32(f.opts.hrpNetworkPrefix), err)
				}
			} else {
				unspentOutputs = append(unspentOutputs, lastRemainderOutput)
			}

			var amount uint64 = 0
			found := false
			for _, unspentOutput := range unspentOutputs {
				amount += unspentOutput.Amount()
				if lastRemainderOutput != nil && bytes.Equal(unspentOutput.OutputID()[:], lastRemainderOutput.OutputID()[:]) {
					found = true
				}
			}

			if lastRemainderOutput != nil && !found {
				unspentOutputs = append(unspentOutputs, lastRemainderOutput)
				amount += lastRemainderOutput.Amount()
			}

			collectedRequestsCounter := len(unspentOutputs)
			batchWriterTimeoutChan := time.After(f.opts.batchTimeout)
			batchedRequests := []*queueItem{}

		CollectValues:
			for collectedRequestsCounter < f.opts.maxOutputCount-1 && amount > f.opts.amount {
				select {
				case <-shutdownSignal:
					// faucet was stopped
					return nil

				case <-batchWriterTimeoutChan:
					// timeout was reached => stop collecting requests
					break CollectValues

				case request := <-f.queue:
					batchedRequests = append(batchedRequests, request)
					collectedRequestsCounter++
					amount -= request.Amount
				}
			}

			f.clearRequests(batchedRequests)

			if len(unspentOutputs) < 2 && len(batchedRequests) == 0 {
				// no need to sweep or send funds
				continue
			}

			remainderOutput, err := f.sendFaucetMessage(unspentOutputs, batchedRequests, shutdownSignal)
			if err != nil {
				if common.IsCriticalError(err) != nil {
					// error is a critical error
					// => stop the faucet
					return err
				}

				f.logSoftError(err)
				continue
			}

			lastRemainderOutput = remainderOutput
		}
	}
}
<file_sep>package storage

import (
	"time"

	"github.com/gohornet/hornet/pkg/common"
	"github.com/gohornet/hornet/pkg/model/hornet"
	"github.com/gohornet/hornet/pkg/profile"
	"github.com/iotaledger/hive.go/kvstore"
	"github.com/iotaledger/hive.go/objectstorage"
	iotago "github.com/iotaledger/iota.go/v2"
)

// CachedChild represents a cached Child.
type CachedChild struct {
	objectstorage.CachedObject
}

type CachedChildren []*CachedChild

// Retain registers a new consumer for the cached children.
func (cachedChildren CachedChildren) Retain() CachedChildren {
	cachedResult := make(CachedChildren, len(cachedChildren))
	for i, cachedChild := range cachedChildren {
		cachedResult[i] = cachedChild.Retain()
	}
	return cachedResult
}

// Release releases the cached children, to be picked up by the persistence layer (as soon as all consumers are done).
func (cachedChildren CachedChildren) Release(force ...bool) {
	for _, cachedChild := range cachedChildren {
		cachedChild.Release(force...)
	}
}

// Retain registers a new consumer for the cached child.
func (c *CachedChild) Retain() *CachedChild {
	return &CachedChild{c.CachedObject.Retain()}
}

// Child retrieves the child, that is cached in this container.
func (c *CachedChild) Child() *Child {
	return c.Get().(*Child)
}

func childrenFactory(key []byte, _ []byte) (objectstorage.StorableObject, error) {
	child := NewChild(hornet.MessageIDFromSlice(key[:iotago.MessageIDLength]), hornet.MessageIDFromSlice(key[iotago.MessageIDLength:iotago.MessageIDLength+iotago.MessageIDLength]))
	return child, nil
}

func (s *Storage) ChildrenStorageSize() int {
	return s.childrenStorage.GetSize()
}

func (s *Storage) configureChildrenStorage(store kvstore.KVStore, opts *profile.CacheOpts) error {

	cacheTime, err := time.ParseDuration(opts.CacheTime)
	if err != nil {
		return err
	}

	leakDetectionMaxConsumerHoldTime, err := time.ParseDuration(opts.LeakDetectionOptions.MaxConsumerHoldTime)
	if err != nil {
		return err
	}

	s.childrenStorage = objectstorage.New(
		store.WithRealm([]byte{common.StorePrefixChildren}),
		childrenFactory,
		objectstorage.CacheTime(cacheTime),
		objectstorage.PersistenceEnabled(true),
		objectstorage.PartitionKey(iotago.MessageIDLength, iotago.MessageIDLength),
		objectstorage.KeysOnly(true),
		objectstorage.StoreOnCreation(true),
		objectstorage.ReleaseExecutorWorkerCount(opts.ReleaseExecutorWorkerCount),
		objectstorage.LeakDetectionEnabled(opts.LeakDetectionOptions.Enabled,
			objectstorage.LeakDetectionOptions{
				MaxConsumersPerObject: opts.LeakDetectionOptions.MaxConsumersPerObject,
				MaxConsumerHoldTime:   leakDetectionMaxConsumerHoldTime,
			}),
	)

	return nil
}

// ChildrenMessageIDs returns the message IDs of the children of the given message.
// children +-0
func (s *Storage) ChildrenMessageIDs(messageID hornet.MessageID, iteratorOptions ...IteratorOption) hornet.MessageIDs {
	var childrenMessageIDs hornet.MessageIDs

	s.childrenStorage.ForEachKeyOnly(func(key []byte) bool {
		childrenMessageIDs = append(childrenMessageIDs, hornet.MessageIDFromSlice(key[iotago.MessageIDLength:iotago.MessageIDLength+iotago.MessageIDLength]))
		return true
	}, append(iteratorOptions, objectstorage.WithIteratorPrefix(messageID))...)

	return childrenMessageIDs
}

// ContainsChild returns if the given child exists in the cache/persistence layer.
func (s *Storage) ContainsChild(messageID hornet.MessageID, childMessageID hornet.MessageID, readOptions ...ReadOption) bool {
	return s.childrenStorage.Contains(append(messageID, childMessageID...), readOptions...)
}

// CachedChildrenOfMessageID returns the cached children of a message.
// children +1
func (s *Storage) CachedChildrenOfMessageID(messageID hornet.MessageID, iteratorOptions ...IteratorOption) CachedChildren {
	cachedChildren := make(CachedChildren, 0)
	s.childrenStorage.ForEach(func(_ []byte, cachedObject objectstorage.CachedObject) bool {
		cachedChildren = append(cachedChildren, &CachedChild{CachedObject: cachedObject})
		return true
	}, append(iteratorOptions, objectstorage.WithIteratorPrefix(messageID))...)
	return cachedChildren
}

// ChildConsumer consumes the given child during looping through all children.
type ChildConsumer func(messageID hornet.MessageID, childMessageID hornet.MessageID) bool

// ForEachChild loops over all children.
func (s *Storage) ForEachChild(consumer ChildConsumer, iteratorOptions ...IteratorOption) {
	s.childrenStorage.ForEachKeyOnly(func(key []byte) bool {
		return consumer(hornet.MessageIDFromSlice(key[:iotago.MessageIDLength]), hornet.MessageIDFromSlice(key[iotago.MessageIDLength:iotago.MessageIDLength+iotago.MessageIDLength]))
	}, iteratorOptions...)
}

// StoreChild stores the child in the persistence layer and returns a cached object.
// child +1
func (s *Storage) StoreChild(parentMessageID hornet.MessageID, childMessageID hornet.MessageID) *CachedChild {
	child := NewChild(parentMessageID, childMessageID)
	return &CachedChild{CachedObject: s.childrenStorage.Store(child)}
}

// DeleteChild deletes the child in the cache/persistence layer.
// child +-0
func (s *Storage) DeleteChild(messageID hornet.MessageID, childMessageID hornet.MessageID) {
	child := NewChild(messageID, childMessageID)
	s.childrenStorage.Delete(child.ObjectStorageKey())
}

// DeleteChildren deletes the children of the given message in the cache/persistence layer.
// child +-0
func (s *Storage) DeleteChildren(messageID hornet.MessageID, iteratorOptions ...IteratorOption) {

	var keysToDelete [][]byte

	s.childrenStorage.ForEachKeyOnly(func(key []byte) bool {
		keysToDelete = append(keysToDelete, key)
		return true
	}, append(iteratorOptions, objectstorage.WithIteratorPrefix(messageID))...)

	for _, key := range keysToDelete {
		s.childrenStorage.Delete(key)
	}
}

// ShutdownChildrenStorage shuts down the children storage.
func (s *Storage) ShutdownChildrenStorage() {
	s.childrenStorage.Shutdown()
}

// FlushChildrenStorage flushes the children storage.
func (s *Storage) FlushChildrenStorage() {
	s.childrenStorage.Flush()
}
<file_sep>package p2p

import (
	"bytes"
	"context"
	stded25519 "crypto/ed25519"
	"crypto/x509"
	"encoding/hex"
	"encoding/pem"
	"fmt"
	"io/ioutil"
	"os"
	"path/filepath"
	"runtime"

	"github.com/ipfs/go-datastore/query"
	badger "github.com/ipfs/go-ds-badger"
	"github.com/libp2p/go-libp2p-core/crypto"
	"github.com/libp2p/go-libp2p-core/peer"
	"github.com/libp2p/go-libp2p-core/peerstore"
	"github.com/libp2p/go-libp2p-peerstore/pstoreds"
	"github.com/pkg/errors"

	"github.com/gohornet/hornet/pkg/database"
	"github.com/gohornet/hornet/pkg/utils"
	kvstoreds "github.com/iotaledger/go-ds-kvstore"
	"github.com/iotaledger/hive.go/kvstore"
)

const (
	DeprecatedPubKeyFileName = "key.pub"
	PrivKeyFileName          = "identity.key"
)

var (
	ErrPrivKeyInvalid = errors.New("invalid private key")
	ErrNoPrivKeyFound = errors.New("no private key found")
)

// PeerStoreContainer is a container for a libp2p peer store.
type PeerStoreContainer struct {
	store     kvstore.KVStore
	peerStore peerstore.Peerstore
}

// Peerstore returns the libp2p peer store from the container.
func (psc *PeerStoreContainer) Peerstore() peerstore.Peerstore {
	return psc.peerStore
}

// Flush persists all outstanding write operations to disc.
func (psc *PeerStoreContainer) Flush() error {
	return psc.store.Flush()
}

// Close flushes all outstanding write operations and closes the store.
func (psc *PeerStoreContainer) Close() error {
	psc.peerStore.Close()

	if err := psc.store.Flush(); err != nil {
		return err
	}

	return psc.store.Close()
}

// NewPeerStoreContainer creates a peerstore using kvstore.
func NewPeerStoreContainer(peerStorePath string, dbEngine database.Engine, createDatabaseIfNotExists bool) (*PeerStoreContainer, error) {

	dirPath := filepath.Dir(peerStorePath)

	if createDatabaseIfNotExists {
		if err := os.MkdirAll(dirPath, 0700); err != nil {
			return nil, fmt.Errorf("could not create peer store database dir '%s': %w", dirPath, err)
		}
	}

	store, err := database.StoreWithDefaultSettings(peerStorePath, createDatabaseIfNotExists, dbEngine)
	if err != nil {
		return nil, fmt.Errorf("unable to initialize peer store database: %w", err)
	}

	peerStore, err := pstoreds.NewPeerstore(context.Background(), kvstoreds.NewDatastore(store), pstoreds.DefaultOpts())
	if err != nil {
		return nil, fmt.Errorf("unable to initialize peer store: %w", err)
	}

	return &PeerStoreContainer{
		store:     store,
		peerStore: peerStore,
	}, nil
}

// parseEd25519PrivateKeyFromString parses an Ed25519 private key from a hex encoded string.
func parseEd25519PrivateKeyFromString(identityPrivKey string) (crypto.PrivKey, error) {
	if identityPrivKey == "" {
		return nil, ErrNoPrivKeyFound
	}

	hivePrivKey, err := utils.ParseEd25519PrivateKeyFromString(identityPrivKey)
	if err != nil {
		return nil, fmt.Errorf("unable to parse private key: %w", ErrPrivKeyInvalid)
	}

	stdPrvKey := stded25519.PrivateKey(hivePrivKey)
	p2pPrvKey, _, err := crypto.KeyPairFromStdKey(&stdPrvKey)
	if err != nil {
		return nil, fmt.Errorf("unable to convert private key: %w", err)
	}

	return p2pPrvKey, nil
}

// ReadEd25519PrivateKeyFromPEMFile reads an Ed25519 private key from a file with PEM format.
func ReadEd25519PrivateKeyFromPEMFile(filepath string) (crypto.PrivKey, error) {

	pemPrivateBlockBytes, err := os.ReadFile(filepath)
	if err != nil {
		return nil, fmt.Errorf("unable to read private key: %w", err)
	}

	pemPrivateBlock, _ := pem.Decode(pemPrivateBlockBytes)
	if pemPrivateBlock == nil {
		return nil, fmt.Errorf("unable to decode private key: %w", err)
	}

	stdCryptoPrvKey, err := x509.ParsePKCS8PrivateKey(pemPrivateBlock.Bytes)
	if err != nil {
		return nil, fmt.Errorf("unable to parse private key: %w", err)
	}

	stdPrvKey, ok := stdCryptoPrvKey.(stded25519.PrivateKey)
	if !ok {
		return nil, fmt.Errorf("unable to type assert private key: %w", err)
	}

	privKey, err := crypto.UnmarshalEd25519PrivateKey((stdPrvKey)[:])
	if err != nil {
		return nil, fmt.Errorf("unable to unmarshal private key: %w", err)
	}

	return privKey, nil
}

// WriteEd25519PrivateKeyToPEMFile stores an Ed25519 private key to a file with PEM format.
func WriteEd25519PrivateKeyToPEMFile(filepath string, privateKey crypto.PrivKey) error {

	stdCryptoPrvKey, err := crypto.PrivKeyToStdKey(privateKey)
	if err != nil {
		return fmt.Errorf("unable to convert private key: %w", err)
	}

	stdPrvKey, ok := stdCryptoPrvKey.(*stded25519.PrivateKey)
	if !ok {
		return fmt.Errorf("unable to type assert private key: %w", err)
	}

	pkcs8Bytes, err := x509.MarshalPKCS8PrivateKey(*stdPrvKey)
	if err != nil {
		return fmt.Errorf("unable to mashal private key: %w", err)
	}

	pemPrivateBlock := &pem.Block{
		Type:  "PRIVATE KEY",
		Bytes: pkcs8Bytes,
	}

	var pemBuffer bytes.Buffer
	if err := pem.Encode(&pemBuffer, pemPrivateBlock); err != nil {
		return fmt.Errorf("unable to encode private key: %w", err)
	}

	if err := utils.WriteToFile(filepath, pemBuffer.Bytes(), 0660); err != nil {
		return fmt.Errorf("unable to write private key: %w", err)
	}

	return nil
}

// LoadOrCreateIdentityPrivateKey loads an existing Ed25519 based identity private key
// or creates a new one and stores it as a PEM file in the p2p store folder.
func LoadOrCreateIdentityPrivateKey(p2pStorePath string, identityPrivKey string) (crypto.PrivKey, bool, error) {

	privKeyFromConfig, err := parseEd25519PrivateKeyFromString(identityPrivKey)
	if err != nil {
		if errors.Is(err, ErrPrivKeyInvalid) {
			return nil, false, errors.New("configuration contains an invalid private key")
		}

		if !errors.Is(err, ErrNoPrivKeyFound) {
			return nil, false, fmt.Errorf("unable to parse private key from config: %w", err)
		}
	}

	privKeyFilePath := filepath.Join(p2pStorePath, PrivKeyFileName)

	_, err = os.Stat(privKeyFilePath)
	switch {
	case err == nil || os.IsExist(err):
		// private key already exists, load and return it
		privKey, err := ReadEd25519PrivateKeyFromPEMFile(privKeyFilePath)
		if err != nil {
			return nil, false, fmt.Errorf("unable to load Ed25519 private key for peer identity: %w", err)
		}

		if privKeyFromConfig != nil && !privKeyFromConfig.Equals(privKey) {
			storedPrivKeyBytes, err := crypto.MarshalPrivateKey(privKey)
			if err != nil {
				return nil, false, fmt.Errorf("unable to marshal stored Ed25519 private key for peer identity: %w", err)
			}
			configPrivKeyBytes, err := crypto.MarshalPrivateKey(privKeyFromConfig)
			if err != nil {
				return nil, false, fmt.Errorf("unable to marshal configured Ed25519 private key for peer identity: %w", err)
			}

			return nil, false, fmt.Errorf("stored Ed25519 private key (%s) for peer identity doesn't match private key in config (%s)", hex.EncodeToString(storedPrivKeyBytes[:]), hex.EncodeToString(configPrivKeyBytes[:]))
		}

		return privKey, false, nil

	case os.IsNotExist(err):
		var privKey crypto.PrivKey

		if privKeyFromConfig != nil {
			privKey = privKeyFromConfig
		} else {
			// private key does not exist, create a new one
			privKey, _, err = crypto.GenerateKeyPair(crypto.Ed25519, -1)
			if err != nil {
				return nil, false, fmt.Errorf("unable to generate Ed25519 private key for peer identity: %w", err)
			}
		}
		if err := WriteEd25519PrivateKeyToPEMFile(privKeyFilePath, privKey); err != nil {
			return nil, false, fmt.Errorf("unable to store private key file for peer identity: %w", err)
		}
		return privKey, true, nil

	default:
		return nil, false, fmt.Errorf("unable to check private key file for peer identity (%s): %w", privKeyFilePath, err)
	}
}

// MigrateDeprecatedPeerStore extracts the old peer identity private key from the configuration or peer store,
// migrates the old database and stores the private key in a new file with PEM format.
func MigrateDeprecatedPeerStore(p2pStorePath string, identityPrivKey string, newPeerStoreContainer *PeerStoreContainer) (bool, error) {

	privKeyFilePath := filepath.Join(p2pStorePath, PrivKeyFileName)

	_, err := os.Stat(privKeyFilePath)
	switch {
	case err == nil || os.IsExist(err):
		// migration not necessary since the private key file already exists
		return false, nil
	case os.IsNotExist(err):
		// migration maybe necessary
	default:
		return false, fmt.Errorf("unable to check private key file for peer identity (%s): %w", privKeyFilePath, err)
	}

	deprecatedPubKeyFilePath := filepath.Join(p2pStorePath, DeprecatedPubKeyFileName)
	if _, err := os.Stat(deprecatedPubKeyFilePath); err != nil {
		if os.IsNotExist(err) {
			// migration not necessary since no old public key file exists
			return false, nil
		}

		return false, fmt.Errorf("unable to check deprecated public key file for peer identity (%s): %w", deprecatedPubKeyFilePath, err)
	}

	// migrates the deprecated badger DB peerstore to the new kvstore based peerstore.
	migrateDeprecatedPeerStore := func(deprecatedPeerStorePath string, newStore kvstore.KVStore) error {
		defaultOpts := badger.DefaultOptions

		// needed under Windows otherwise peer store is 'corrupted' after a restart
		defaultOpts.Truncate = runtime.GOOS == "windows"

		badgerStore, err := badger.NewDatastore(deprecatedPeerStorePath, &defaultOpts)
		if err != nil {
			return fmt.Errorf("unable to initialize data store for deprecated peer store: %w", err)
		}
		defer func() { _ = badgerStore.Close() }()

		results, err := badgerStore.Query(query.Query{})
		if err != nil {
			return fmt.Errorf("unable to query deprecated peer store: %w", err)
		}

		for res := range results.Next() {
			if err := newStore.Set([]byte(res.Key), res.Value); err != nil {
				return fmt.Errorf("unable to migrate data to new peer store: %w", err)
			}
		}
		if err := newStore.Flush(); err != nil {
			return fmt.Errorf("unable to flush new peer store: %w", err)
		}

		return nil
	}

	if err := migrateDeprecatedPeerStore(p2pStorePath, newPeerStoreContainer.store); err != nil {
		return false, err
	}

	privKey, err := parseEd25519PrivateKeyFromString(identityPrivKey)
	if err != nil {
		if errors.Is(err, ErrPrivKeyInvalid) {
			return false, errors.New("configuration contains an invalid private key")
		}

		if !errors.Is(err, ErrNoPrivKeyFound) {
			return false, fmt.Errorf("unable to parse private key from config: %w", err)
		}

		// there was no private key specified, retrieve it from the peer store with the public key from the deprecated file
		existingPubKeyBytes, err := ioutil.ReadFile(deprecatedPubKeyFilePath)
		if err != nil {
			return false, fmt.Errorf("unable to read deprecated public key file for peer identity: %w", err)
		}

		pubKey, err := crypto.UnmarshalPublicKey(existingPubKeyBytes)
		if err != nil {
			return false, fmt.Errorf("unable to unmarshal deprecated public key for peer identity: %w", err)
		}

		peerID, err := peer.IDFromPublicKey(pubKey)
		if err != nil {
			return false, fmt.Errorf("unable to get peer identity from deprecated public key: %w", err)
		}

		// retrieve this node's private key from the new peer store
		privKey = newPeerStoreContainer.peerStore.PrivKey(peerID)
		if privKey == nil {
			return false, errors.New("error while fetching private key for peer identity from peer store")
		}
	}

	if err := WriteEd25519PrivateKeyToPEMFile(privKeyFilePath, privKey); err != nil {
		return false, err
	}

	// delete the deprecated public key file
	if err := os.Remove(deprecatedPubKeyFilePath); err != nil {
		return false, fmt.Errorf("unable to remove deprecated public key file for peer identity: %w", err)
	}

	return true, nil
}
<file_sep># HORNET Changelog

All notable changes to this project will be documented in this file.

## [1.0.5] - 02.09.2021

### Added
    - Add faucet plugin
    - Add faucet website
    - Add database migration tool
    - Add io utils to read/write a TOML file
    - Add private tangle scripts to deb/rpm files
    - New public keys applicable for milestone ranges between 1333460-4443860

### Changed
    - Move p2p identity private key to PEM file
    - Run autopeering entry node in standalone mode
    - Merge peerstore and autopeering db path in config
    - Expose autopeering port in dockerfile
    - Add detailed reason for snapshot download failure
    - Print bech32 address in toolset
    - Use kvstore for libp2p peerstore and autopeering database
    - Use internal bech32 HRP instead of additional config parameter
    - Update go version in workflows and dockerfile
    - Update github workflow actions
    - Update go modules
    - Update libp2p
    - Autopeering entry nodes for mainnet
    - Snapshot download sources for mainnet/comnet

### Fixed
    - Fix data size units according to SI units and IEC 60027
    - Fix possible jwt token vulnerabilities
    - Fix nil pointer exception in optimalSnapshotType
    - Set missing config default values for mainnet
    - Do not disable plugins for the entry nodes in the integration tests
    - Fix dependency injection of global config pars if pluggables are disabled by adding a InitConfigPars stage
    - Do not panic in MilestoneRetrieverFromStorage
    - Fix logger init in test cases

### Removed
    - Remove support for boltdb

### Cleanups
    - Refactor node/pluggable logger interface
    - Check returned error at the initialization of new background workers
    - Fixed return values order of StreamSnapshotDataTo
    - Do not use unbuffered channels for signal.Notify
    - Change some variable names according to linter suggestions
    - Remove duplicate imports
    - Remove dead code
    - Cleanup toolset listTools func
    - Replace deprecated prometheus functions
    - Fix comments
    - Remove 'Get' from getter function names
    - Add error return value to SetConfirmedMilestoneIndex
    - Add error return value to SetSnapshotMilestone
    - Add error return value to database health functions
    - Add error return value to loadSnaphotInfo
    - Replace http.Get with client.Do and a context
    - Add missing checks for returned errors
    - Add linter ignore rules
    - Cleanup unused parameters and return values
    - Fix some printouts and move hornet logo to startup
    - Return error if database init fails
    - Reduce dependencies between plugins/core modules
    - Replace deprecated libp2p crypto Bytes function
    - Add error return value to loadSolidEntryPoints
    - Rename util files to utils
    - Move database events to package
    - Move graceful shutdown logic to package
    - Move autopeering local to pkg
    - Add error return value to authorization init functions

### Config file changes

`config.json`
```diff
  "snapshots": {
    "downloadURLs": [
      {
-        "full": "https://mainnet.tanglebay.com/ls/full_snapshot.bin",
+        "full": "https://cdn.tanglebay.com/snapshots/mainnet/full_snapshot.bin",
-        "delta": "https://mainnet.tanglebay.com/ls/delta_snapshot.bin"
+        "delta": "https://cdn.tanglebay.com/snapshots/mainnet/delta_snapshot.bin"
      }
    ]
  },
  "protocol": {
    "publicKeyRanges": [
      {
        "key": "<KEY>",
        "start": 552960,
        "end": 2108160
-      }
+      },
+      {
+        "key": "760d88e112c0fd210cf16a3dce3443ecf7e18c456c2fb9646cabb2e13e367569",
+        "start": 1333460,
+        "end": 2888660
+      },
+      {
+        "key": "<KEY>76ea2235539358c7df8ca4d2f2fc35a663c760449e65eba9f8a6e7",
+        "start": 2111060,
+        "end": 3666260
+      },
+      {
+        "key": "<KEY>",
+        "start": 2888660,
+        "end": 4443860
+      }
    ]
  },
  "p2p": {
-    "identityPrivateKey": "",
-    "peerStore": {
-      "path": "./p2pstore"
-    }
+    "db": {
+      "path": "p2pstore"
+    },
    "autopeering": {
-      "db": {
-        "path": "./p2pstore"
-      },
      "entryNodes": [
-        "/dns/lucamoser.ch/udp/14926/autopeering/4H6WV54tB29u8xCcEaMGQMn37LFvM1ynNpp27TTXaqNM",
+        "/dns/lucamoser.ch/udp/14826/autopeering/4H6WV54tB29u8xCcEaMGQMn37LFvM1ynNpp27TTXaqNM",
+        "/dns/entry-hornet-0.h.chrysalis-mainnet.iotaledger.net/udp/14626/autopeering/iotaPHdAn7eueBnXtikZMwhfPXaeGJGXDt4RBuLuGgb",
+        "/dns/entry-hornet-1.h.chrysalis-mainnet.iotaledger.net/udp/14626/autopeering/iotaJJqMd5CQvv1A61coSQCYW9PNT1QKPs7xh2Qg5K2",
         "/dns/entry-mainnet.tanglebay.com/udp/14626/autopeering/iot4By1FD4pFLrGJ6AAe7YEeSu9RbW9xnPUmxMdQenC"
      ],
    }
  },
+  "faucet": {
+    "amount": 10000000,
+    "smallAmount": 1000000,
+    "maxAddressBalance": 20000000,
+    "maxOutputCount": 127,
+    "indexationMessage": "HORNET FAUCET",
+    "batchTimeout": "2s",
+    "powWorkerCount": 0,
+    "website": {
+      "bindAddress": "localhost:8091",
+      "enabled": true
+    },
  }
```

## [1.0.4] - 04.08.2021

### Added
    - Autopeering with two default entry nodes (disabled by default)
    - Default config for community network

### Changed
    - Reduces the default WarpSync advancement range back to 150 as the previous value was only a workaround.
    - Cleaned up parameter names and values in all config files
    - Rate limit for dashboard login changed from 1 request every 5 minutes to 5 requests every minute

### Config file changes

`config.json`
```diff
  "receipts": {
    "backup": {
-      "folder": "receipts"
+      "path": "receipts"
    },
  }
  "p2p": {
    "bindMultiAddresses": [
      "/ip4/0.0.0.0/tcp/15600",
+      "/ip6/::/tcp/15600"
    ],
-    "gossipUnknownPeersLimit": 4,
+    "gossip": {
+      "unknownPeersLimit": 4,
+      "streamReadTimeout": "1m0s",
+      "streamWriteTimeout": "10s"
+    },
  "autopeering": {
-    "dirPath": "./p2pstore"
+    "db": {
+      "path": "./p2pstore"
+    },
  }
  "warpsync": {
-    "advancementRange": 10000
+    "advancementRange": 150
  },
  "spammer": {
-    "mpsRateLimit": 5.0,
+    "mpsRateLimit": 0.0,
  },
```

## [1.0.3] - 02.06.2021

### Added
    - A new public key applicable for milestone ranges between 552960-2108160

### Changed
    - Increases the WarpSync advancement range to 10k which allows nodes to resync even if
      they already stored the milestone for which they lacked the applicable public key beforehand.

## [1.0.2] - 28.05.2021

### Added
    - p2pidentity-extract tool (#1090)
    - tool to generate JWT token for REST API (#1085)
    - Add database pruning based on database size (#1115)

### Changed
    - Improved documentation (#1060 + #1083 + #1087 + #1090)
    - Build Dockerfile with rocksdb (#1077)
    - Default DB engine changed to rocksdb (#1078)
    - Renamed alphanet scripts to private_tangle (#1078)
    - Updated rocksdb (#1080)
    - Updated go modules, containers and dashboard (#1103)
    - Check if files exist in the p2pstore directory (needed for docker) (#1084)
    - Disabled MQTT http port (#1094)
    - Download the latest snapshots from the given targets (#1097)
    - Adds "ledgerIndex" field to some REST HTTP and MQTT API responses (#1106)
    - Add delta snapshots to control endpoint (#1039)
    - Changed node control endpoints to use POST (#1039)
    - Expose MQTT port. Remove no longer needed ports (#1105)
    - Re-add private Tangle doc (#1113)

### Fixed
    - Added workdir to docker/Dockerfile. (#1068)
    - JWT subject verification (#1076)
    - Send on closed channel in coo quorum (#1082)
    - Database revalidation (#1096)
    - Mask sensitive config parameters in log (#1100)
    - Fix ulimits and dependencies at node startup (#1107)
    - Do not print API JWT auth tokens to the log (unsafe) (#1039)
    - Check if node is busy before accepting snapshot commands via API (#1039)

### Config file changes

`config.json`
```diff
  "pruning": {
-    "enabled": true,
-    "delay": 60480,
+    "milestones": {
+      "enabled": false,
+      "maxMilestonesToKeep": 60480
+    },
+    "size": {
+      "enabled": true,
+      "targetSize": "30GB",
+      "thresholdPercentage": 10.0,
+      "cooldownTime": "5m"
+    },
    "pruneReceipts": false
  },
```

## [1.0.1] - 28.04.2021

### Fixed

    - Receipt validation deadlock
    - Docker Image Build Tags for RocksDB


## [1.0.0] - 28.04.2021

### Changed

    - IOTA: A new dawn.<file_sep>package utils

import (
	"os"
	"path/filepath"
)

// FolderSize returns the size of a folder.
func FolderSize(target string) (int64, error) {

	var size int64

	err := filepath.Walk(target, func(_ string, info os.FileInfo, err error) error {
		if err != nil {
			return err
		}

		if !info.IsDir() {
			size += info.Size()
		}

		return err
	})

	return size, err
}
<file_sep>package database

import (
	flag "github.com/spf13/pflag"

	"github.com/gohornet/hornet/pkg/database"
	"github.com/gohornet/hornet/pkg/node"
)

const (
	// the used database engine (pebble/rocksdb).
	CfgDatabaseEngine = "db.engine"
	// the path to the database folder.
	CfgDatabasePath = "db.path"
	// whether to automatically start revalidation on startup if the database is corrupted.
	CfgDatabaseAutoRevalidation = "db.autoRevalidation"
	// ignore the check for corrupted databases (should only be used for debug reasons).
	CfgDatabaseDebug = "db.debug"
)

var params = &node.PluginParams{
	Params: map[string]*flag.FlagSet{
		"nodeConfig": func() *flag.FlagSet {
			fs := flag.NewFlagSet("", flag.ContinueOnError)
			fs.String(CfgDatabaseEngine, database.EngineRocksDB, "the used database engine (pebble/rocksdb)")
			fs.String(CfgDatabasePath, "mainnetdb", "the path to the database folder")
			fs.Bool(CfgDatabaseAutoRevalidation, false, "whether to automatically start revalidation on startup if the database is corrupted")
			fs.Bool(CfgDatabaseDebug, false, "ignore the check for corrupted databases (should only be used for debug reasons)")
			return fs
		}(),
	},
	Masked: nil,
}
<file_sep>package storage

import (
	"sync"

	"github.com/gohornet/hornet/pkg/keymanager"
	"github.com/gohornet/hornet/pkg/model/milestone"
	"github.com/gohornet/hornet/pkg/model/utxo"
	"github.com/gohornet/hornet/pkg/profile"
	"github.com/gohornet/hornet/pkg/utils"
	"github.com/iotaledger/hive.go/events"
	"github.com/iotaledger/hive.go/kvstore"
	"github.com/iotaledger/hive.go/logger"
	"github.com/iotaledger/hive.go/objectstorage"
	"github.com/iotaledger/hive.go/syncutils"
)

type packageEvents struct {
	ReceivedValidMilestone *events.Event
	PruningStateChanged    *events.Event
}

type ReadOption = objectstorage.ReadOption
type IteratorOption = objectstorage.IteratorOption

type Storage struct {
	log *logger.Logger

	// database
	databaseDir   string
	store         kvstore.KVStore
	belowMaxDepth milestone.Index

	// kv storages
	healthStore   kvstore.KVStore
	snapshotStore kvstore.KVStore

	// object storages
	childrenStorage             *objectstorage.ObjectStorage
	indexationStorage           *objectstorage.ObjectStorage
	messagesStorage             *objectstorage.ObjectStorage
	metadataStorage             *objectstorage.ObjectStorage
	milestoneStorage            *objectstorage.ObjectStorage
	unreferencedMessagesStorage *objectstorage.ObjectStorage

	// solid entry points
	solidEntryPoints     *SolidEntryPoints
	solidEntryPointsLock sync.RWMutex

	// snapshot info
	snapshot      *SnapshotInfo
	snapshotMutex syncutils.RWMutex

	// milestones
	confirmedMilestoneIndex milestone.Index
	confirmedMilestoneLock  syncutils.RWMutex
	latestMilestoneIndex    milestone.Index
	latestMilestoneLock     syncutils.RWMutex

	// node synced
	isNodeSynced                    bool
	isNodeAlmostSynced              bool
	isNodeSyncedWithinBelowMaxDepth bool
	waitForNodeSyncedChannelsLock   syncutils.Mutex
	waitForNodeSyncedChannels       []chan struct{}

	// milestones
	keyManager              *keymanager.KeyManager
	milestonePublicKeyCount int

	// utxo
	utxoManager *utxo.Manager

	// events
	Events *packageEvents
}

func New(log *logger.Logger, databaseDirectory string, store kvstore.KVStore, cachesProfile *profile.Caches, belowMaxDepth int, keyManager *keymanager.KeyManager, milestonePublicKeyCount int) (*Storage, error) {

	utxoManager := utxo.New(store)

	s := &Storage{
		log:                     log,
		databaseDir:             databaseDirectory,
		store:                   store,
		keyManager:              keyManager,
		milestonePublicKeyCount: milestonePublicKeyCount,
		utxoManager:             utxoManager,
		belowMaxDepth:           milestone.Index(belowMaxDepth),
		Events: &packageEvents{
			ReceivedValidMilestone: events.NewEvent(MilestoneWithRequestedCaller),
			PruningStateChanged:    events.NewEvent(events.BoolCaller),
		},
	}

	if err := s.configureStorages(s.store, cachesProfile); err != nil {
		return nil, err
	}

	if err := s.loadConfirmedMilestoneFromDatabase(); err != nil {
		return nil, err
	}

	if err := s.loadSnapshotInfo(); err != nil {
		return nil, err
	}

	if err := s.loadSolidEntryPoints(); err != nil {
		return nil, err
	}

	return s, nil
}

func (s *Storage) KVStore() kvstore.KVStore {
	return s.store
}

func (s *Storage) UTXO() *utxo.Manager {
	return s.utxoManager
}

func (s *Storage) configureStorages(store kvstore.KVStore, caches *profile.Caches) error {

	if err := s.configureHealthStore(store); err != nil {
		return err
	}

	if err := s.configureMessageStorage(store, caches.Messages); err != nil {
		return err
	}

	if err := s.configureChildrenStorage(store, caches.Children); err != nil {
		return err
	}

	if err := s.configureMilestoneStorage(store, caches.Milestones); err != nil {
		return err
	}

	if err := s.configureUnreferencedMessageStorage(store, caches.UnreferencedMessages); err != nil {
		return err
	}

	if err := s.configureIndexationStorage(store, caches.Indexations); err != nil {
		return err
	}

	s.configureSnapshotStore(store)

	return nil
}

// FlushStorages flushes all storages.
func (s *Storage) FlushStorages() {
	s.FlushMilestoneStorage()
	s.FlushMessagesStorage()
	s.FlushChildrenStorage()
	s.FlushIndexationStorage()
	s.FlushUnreferencedMessagesStorage()
}

// ShutdownStorages shuts down all storages.
func (s *Storage) ShutdownStorages() {

	s.ShutdownMilestoneStorage()
	s.ShutdownMessagesStorage()
	s.ShutdownChildrenStorage()
	s.ShutdownIndexationStorage()
	s.ShutdownUnreferencedMessagesStorage()
}

func (s *Storage) loadConfirmedMilestoneFromDatabase() error {

	ledgerMilestoneIndex, err := s.UTXO().ReadLedgerIndex()
	if err != nil {
		return err
	}

	// set the confirmed milestone index based on the ledger milestone
	return s.SetConfirmedMilestoneIndex(ledgerMilestoneIndex, false)
}

// DatabaseSize returns the size of the database.
func (s *Storage) DatabaseSize() (int64, error) {
	return utils.FolderSize(s.databaseDir)
}
<file_sep>package toolset

import (
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"time"

	"github.com/dustin/go-humanize"

	"github.com/gohornet/hornet/pkg/database"
	"github.com/gohornet/hornet/pkg/utils"
	"github.com/iotaledger/hive.go/configuration"
	"github.com/iotaledger/hive.go/kvstore"
)

func databaseMigration(_ *configuration.Configuration, args []string) error {
	printUsage := func() {
		println("Usage:")
		println(fmt.Sprintf("	%s [SOURCE_DATABASE_PATH] [TARGET_DATABASE_PATH] [TARGET_DATABASE_ENGINE]", ToolDatabaseMigration))
		println()
		println("   [SOURCE_DATABASE_PATH]   - the path to the source database")
		println("   [TARGET_DATABASE_PATH]   - the path to the target database")
		println("   [TARGET_DATABASE_ENGINE] - the engine of the target database (values: pebble, rocksdb)")
		println()
		println(fmt.Sprintf("example: %s %s %s %s", ToolDatabaseMigration, "mainnetdb", "mainnetdb_new", "rocksdb"))
	}

	// check arguments
	if len(args) != 3 {
		printUsage()
		return fmt.Errorf("wrong argument count for '%s'", ToolDatabaseMigration)
	}

	sourcePath := args[0]
	if _, err := os.Stat(sourcePath); err != nil || os.IsNotExist(err) {
		return fmt.Errorf("SOURCE_DATABASE_PATH (%s) does not exist", sourcePath)
	}

	targetPath := args[1]
	if _, err := os.Stat(targetPath); err == nil || !os.IsNotExist(err) {
		return fmt.Errorf("TARGET_DATABASE_PATH (%s) already exist", targetPath)
	}

	dbEngineTarget := strings.ToLower(args[2])
	engineTarget, err := database.DatabaseEngine(dbEngineTarget)
	if err != nil {
		return err
	}

	storeSource, err := database.StoreWithDefaultSettings(sourcePath, false)
	if err != nil {
		return fmt.Errorf("source database initialization failed: %w", err)
	}
	defer func() { _ = storeSource.Close() }()

	storeTarget, err := database.StoreWithDefaultSettings(targetPath, true, engineTarget)
	if err != nil {
		return fmt.Errorf("target database initialization failed: %w", err)
	}
	defer func() { _ = storeTarget.Close() }()

	copyBytes := func(source []byte) []byte {
		cpy := make([]byte, len(source))
		copy(cpy, source)
		return cpy
	}

	ts := time.Now()
	lastStatusTime := time.Now()

	sourcePathAbs, err := filepath.Abs(sourcePath)
	if err != nil {
		sourcePathAbs = sourcePath
	}
	targetPathAbs, err := filepath.Abs(targetPath)
	if err != nil {
		targetPathAbs = targetPath
	}

	fmt.Printf("Migrating database... (source: \"%s\", target: \"%s\")\n", sourcePathAbs, targetPathAbs)

	var errDB error
	if err := storeSource.Iterate(kvstore.EmptyPrefix, func(key []byte, value kvstore.Value) bool {
		dstKey := copyBytes(key)
		dstValue := copyBytes(value)

		if errDB = storeTarget.Set(dstKey, dstValue); errDB != nil {
			return false
		}

		if time.Since(lastStatusTime) >= printStatusInterval {
			lastStatusTime = time.Now()

			sourceSizeBytes, _ := utils.FolderSize(sourcePath)
			targetSizeBytes, _ := utils.FolderSize(targetPath)

			percentage, remaining := utils.EstimateRemainingTime(ts, targetSizeBytes, sourceSizeBytes)
			fmt.Printf("Source database size: %s, target database size: %s, estimated percentage: %0.2f%%. %v elapsed, %v left...)\n", humanize.Bytes(uint64(sourceSizeBytes)), humanize.Bytes(uint64(targetSizeBytes)), percentage, time.Since(ts).Truncate(time.Second), remaining.Truncate(time.Second))
		}

		return true
	}); err != nil {
		return fmt.Errorf("source database iteration failed: %w", err)
	}

	if errDB != nil {
		return fmt.Errorf("target database set failed: %w", err)
	}

	if err := storeTarget.Flush(); err != nil {
		return fmt.Errorf("target database flush failed: %w", err)
	}

	fmt.Printf("Migration successful! took: %v\n", time.Since(ts).Truncate(time.Second))

	return nil
}
<file_sep>package node

import (
	"fmt"
	"strings"

	flag "github.com/spf13/pflag"
	"go.uber.org/dig"

	"github.com/iotaledger/hive.go/configuration"
	"github.com/iotaledger/hive.go/daemon"
	"github.com/iotaledger/hive.go/logger"
)

type Node struct {
	enabledPlugins          map[string]struct{}
	disabledPlugins         map[string]struct{}
	forceDisabledPluggables map[string]struct{}
	corePluginsMap          map[string]*CorePlugin
	corePlugins             []*CorePlugin
	pluginsMap              map[string]*Plugin
	plugins                 []*Plugin
	container               *dig.Container
	log                     *logger.Logger
	options                 *NodeOptions
}

func New(optionalOptions ...NodeOption) *Node {
	nodeOpts := &NodeOptions{}
	nodeOpts.apply(defaultNodeOptions...)
	nodeOpts.apply(optionalOptions...)

	node := &Node{
		enabledPlugins:          make(map[string]struct{}),
		disabledPlugins:         make(map[string]struct{}),
		forceDisabledPluggables: make(map[string]struct{}),
		corePluginsMap:          make(map[string]*CorePlugin),
		corePlugins:             make([]*CorePlugin, 0),
		pluginsMap:              make(map[string]*Plugin),
		plugins:                 make([]*Plugin, 0),
		container:               dig.New(dig.DeferAcyclicVerification()),
		options:                 nodeOpts,
	}

	// initialize the core plugins and plugins
	node.init()

	// initialize logger after init phase because plugins could modify it
	node.log = logger.NewLogger("Node")

	// configure the core plugins and enabled plugins
	node.configure()

	return node
}

// LogDebug uses fmt.Sprint to construct and log a message.
func (n *Node) LogDebug(args ...interface{}) {
	n.log.Debug(args...)
}

// LogDebugf uses fmt.Sprintf to log a templated message.
func (n *Node) LogDebugf(template string, args ...interface{}) {
	n.log.Debugf(template, args...)
}

// LogError uses fmt.Sprint to construct and log a message.
func (n *Node) LogError(args ...interface{}) {
	n.log.Error(args...)
}

// LogErrorf uses fmt.Sprintf to log a templated message.
func (n *Node) LogErrorf(template string, args ...interface{}) {
	n.log.Errorf(template, args...)
}

// LogFatal uses fmt.Sprint to construct and log a message, then calls os.Exit.
func (n *Node) LogFatal(args ...interface{}) {
	n.log.Fatal(args...)
}

// LogFatalf uses fmt.Sprintf to log a templated message, then calls os.Exit.
func (n *Node) LogFatalf(template string, args ...interface{}) {
	n.log.Fatalf(template, args...)
}

// LogInfo uses fmt.Sprint to construct and log a message.
func (n *Node) LogInfo(args ...interface{}) {
	n.log.Info(args...)
}

// LogInfof uses fmt.Sprintf to log a templated message.
func (n *Node) LogInfof(template string, args ...interface{}) {
	n.log.Infof(template, args...)
}

// LogWarn uses fmt.Sprint to construct and log a message.
func (n *Node) LogWarn(args ...interface{}) {
	n.log.Warn(args...)
}

// LogWarnf uses fmt.Sprintf to log a templated message.
func (n *Node) LogWarnf(template string, args ...interface{}) {
	n.log.Warnf(template, args...)
}

// Panic uses fmt.Sprint to construct and log a message, then panics.
func (n *Node) Panic(args ...interface{}) {
	n.log.Panic(args...)
}

// Panicf uses fmt.Sprintf to log a templated message, then panics.
func (n *Node) Panicf(template string, args ...interface{}) {
	n.log.Panicf(template, args...)
}

func Start(optionalOptions ...NodeOption) *Node {
	node := New(optionalOptions...)
	node.Start()

	return node
}

func Run(optionalOptions ...NodeOption) *Node {
	node := New(optionalOptions...)
	node.Run()

	return node
}

func (n *Node) isEnabled(identifier string) bool {
	_, exists := n.enabledPlugins[identifier]
	return exists
}

func (n *Node) isDisabled(identifier string) bool {
	_, exists := n.disabledPlugins[identifier]
	return exists
}

func (n *Node) isForceDisabled(identifier string) bool {
	_, exists := n.forceDisabledPluggables[identifier]
	return exists
}

// IsSkipped returns whether the plugin is loaded or skipped.
func (n *Node) IsSkipped(plugin *Plugin) bool {
	// list of disabled plugins has the highest priority
	if n.isDisabled(plugin.Identifier()) || n.isForceDisabled(plugin.Identifier()) {
		return true
	}

	// if the plugin was not in the list of disabled plugins, it is only skipped if
	// the plugin was not enabled and not in the list of enabled plugins.
	return plugin.Status != StatusEnabled && !n.isEnabled(plugin.Identifier())
}

func (n *Node) init() {

	if n.options.initPlugin == nil {
		panic("you must configure the node with an InitPlugin")
	}

	params := map[string][]*flag.FlagSet{}
	masked := []string{}

	//
	// Collect parameters
	//
	n.options.initPlugin.Node = n
	if n.options.initPlugin.Params != nil {
		for k, v := range n.options.initPlugin.Params.Params {
			params[k] = append(params[k], v)
		}
	}
	if n.options.initPlugin.Params.Masked != nil {
		masked = append(masked, n.options.initPlugin.Params.Masked...)
	}

	forEachCorePlugin(n.options.corePlugins, func(corePlugin *CorePlugin) bool {
		corePlugin.Node = n

		if corePlugin.Params == nil {
			return true
		}
		for k, v := range corePlugin.Params.Params {
			params[k] = append(params[k], v)
		}
		if corePlugin.Params.Masked != nil {
			masked = append(masked, corePlugin.Params.Masked...)
		}
		return true
	})

	forEachPlugin(n.options.plugins, func(plugin *Plugin) bool {
		plugin.Node = n

		if plugin.Params == nil {
			return true
		}
		for k, v := range plugin.Params.Params {
			params[k] = append(params[k], v)
		}
		if plugin.Params.Masked != nil {
			masked = append(masked, plugin.Params.Masked...)
		}
		return true
	})

	//
	// Init Stage
	//

	// the init plugin parses the config files and initializes the global logger
	initCfg, err := n.options.initPlugin.Init(params, masked)
	if err != nil {
		panic(fmt.Errorf("unable to initialize node: %w", err))
	}

	//
	// InitConfig Stage
	//
	if n.options.initPlugin.InitConfigPars == nil {
		panic(fmt.Errorf("the init plugin must have an InitConfigPars func"))
	}
	n.options.initPlugin.InitConfigPars(n.container)

	forEachCorePlugin(n.options.corePlugins, func(corePlugin *CorePlugin) bool {
		if corePlugin.InitConfigPars != nil {
			corePlugin.InitConfigPars(n.container)
		}
		return true
	})

	forEachPlugin(n.options.plugins, func(plugin *Plugin) bool {
		if plugin.InitConfigPars != nil {
			plugin.InitConfigPars(n.container)
		}
		return true
	})

	//
	// Pre-Provide Stage
	//
	if n.options.initPlugin.PreProvide != nil {
		n.options.initPlugin.PreProvide(n.container, n.options.initPlugin.Configs, initCfg)
	}

	forEachCorePlugin(n.options.corePlugins, func(corePlugin *CorePlugin) bool {
		if corePlugin.PreProvide != nil {
			corePlugin.PreProvide(n.container, n.options.initPlugin.Configs, initCfg)
		}
		return true
	})

	forEachPlugin(n.options.plugins, func(plugin *Plugin) bool {
		if plugin.PreProvide != nil {
			plugin.PreProvide(n.container, n.options.initPlugin.Configs, initCfg)
		}
		return true
	})

	//
	// Enable / (Force-) disable Pluggables
	//
	for _, name := range initCfg.EnabledPlugins {
		n.enabledPlugins[strings.ToLower(name)] = struct{}{}
	}

	for _, name := range initCfg.DisabledPlugins {
		n.disabledPlugins[strings.ToLower(name)] = struct{}{}
	}

	for _, name := range initCfg.forceDisabledPluggables {
		n.forceDisabledPluggables[strings.ToLower(name)] = struct{}{}
	}

	forEachCorePlugin(n.options.corePlugins, func(corePlugin *CorePlugin) bool {
		if n.isForceDisabled(corePlugin.Identifier()) {
			return true
		}

		n.addCorePlugin(corePlugin)
		return true
	})

	forEachPlugin(n.options.plugins, func(plugin *Plugin) bool {
		if n.IsSkipped(plugin) {
			return true
		}

		n.addPlugin(plugin)
		return true
	})

	//
	// Provide Stage
	//
	if n.options.initPlugin.Provide != nil {
		n.options.initPlugin.Provide(n.container)
	}

	n.ForEachCorePlugin(func(corePlugin *CorePlugin) bool {
		if corePlugin.Provide != nil {
			corePlugin.Provide(n.container)
		}
		return true
	})

	n.ForEachPlugin(func(plugin *Plugin) bool {
		if plugin.Provide != nil {
			plugin.Provide(n.container)
		}
		return true
	})

	//
	// Invoke Stage
	//
	if n.options.initPlugin.DepsFunc != nil {
		if err := n.container.Invoke(n.options.initPlugin.DepsFunc); err != nil {
			panic(err)
		}
	}

	n.ForEachCorePlugin(func(corePlugin *CorePlugin) bool {
		if corePlugin.DepsFunc != nil {
			if err := n.container.Invoke(corePlugin.DepsFunc); err != nil {
				panic(err)
			}
		}
		return true
	})

	n.ForEachPlugin(func(plugin *Plugin) bool {
		if plugin.DepsFunc != nil {
			if err := n.container.Invoke(plugin.DepsFunc); err != nil {
				panic(err)
			}
		}
		return true
	})
}

func (n *Node) configure() {

	if n.options.initPlugin.Configure != nil {
		n.options.initPlugin.Configure()
	}

	n.ForEachCorePlugin(func(corePlugin *CorePlugin) bool {
		if corePlugin.Configure != nil {
			corePlugin.Configure()
		}
		n.LogInfof("Loading core plugin: %s ... done", corePlugin.Name)
		return true
	})

	n.ForEachPlugin(func(plugin *Plugin) bool {
		if plugin.Configure != nil {
			plugin.Configure()
		}
		n.LogInfof("Loading plugin: %s ... done", plugin.Name)
		return true
	})
}

func (n *Node) execute() {
	n.LogInfo("Executing core plugin ...")

	if n.options.initPlugin.Run != nil {
		n.options.initPlugin.Run()
	}

	n.ForEachCorePlugin(func(corePlugin *CorePlugin) bool {
		if corePlugin.Run != nil {
			corePlugin.Run()
		}
		n.LogInfof("Starting core plugin: %s ... done", corePlugin.Name)
		return true
	})

	n.LogInfo("Executing plugins ...")

	n.ForEachPlugin(func(plugin *Plugin) bool {
		if plugin.Run != nil {
			plugin.Run()
		}
		n.LogInfof("Starting plugin: %s ... done", plugin.Name)
		return true
	})
}

func (n *Node) Start() {
	n.execute()

	n.LogInfo("Starting background workers ...")
	n.Daemon().Start()
}

func (n *Node) Run() {
	n.execute()

	n.LogInfo("Starting background workers ...")
	n.Daemon().Run()

	n.LogInfo("Shutdown complete!")
}

func (n *Node) Shutdown() {
	n.Daemon().ShutdownAndWait()
}

func (n *Node) Daemon() daemon.Daemon {
	return n.options.daemon
}

func (n *Node) addCorePlugin(corePlugin *CorePlugin) {
	name := corePlugin.Name

	if _, exists := n.corePluginsMap[name]; exists {
		panic("duplicate core plugin - \"" + name + "\" was defined already")
	}

	n.corePluginsMap[name] = corePlugin
	n.corePlugins = append(n.corePlugins, corePlugin)
}

func (n *Node) addPlugin(plugin *Plugin) {
	name := plugin.Name

	if _, exists := n.pluginsMap[name]; exists {
		panic("duplicate plugin - \"" + name + "\" was defined already")
	}

	n.pluginsMap[name] = plugin
	n.plugins = append(n.plugins, plugin)
}

// InitConfigParsFunc gets called with a dig.Container.
type InitConfigParsFunc func(c *dig.Container)

// PreProvideFunc gets called with a dig.Container, the configs the InitPlugin brings to the node and the InitConfig.
type PreProvideFunc func(c *dig.Container, configs map[string]*configuration.Configuration, initConf *InitConfig)

// ProvideFunc gets called with a dig.Container.
type ProvideFunc func(c *dig.Container)

// InitConfig describes the result of a node initialization.
type InitConfig struct {
	EnabledPlugins          []string
	DisabledPlugins         []string
	forceDisabledPluggables []string
}

// ForceDisablePluggable is used to force disable pluggables before the provide stage.
func (ic *InitConfig) ForceDisablePluggable(identifier string) {
	exists := false
	for _, entry := range ic.forceDisabledPluggables {
		if entry == identifier {
			exists = true
			break
		}
	}

	if !exists {
		ic.forceDisabledPluggables = append(ic.forceDisabledPluggables, identifier)
	}
}

// InitFunc gets called as the initialization function of the node.
type InitFunc func(params map[string][]*flag.FlagSet, maskedKeys []string) (*InitConfig, error)

// Callback is a function called without any arguments.
type Callback func()

// CorePluginForEachFunc is used in ForEachCorePlugin.
// Returning false indicates to stop looping.
type CorePluginForEachFunc func(corePlugin *CorePlugin) bool

func forEachCorePlugin(corePlugins []*CorePlugin, f CorePluginForEachFunc) {
	for _, corePlugin := range corePlugins {
		if !f(corePlugin) {
			break
		}
	}
}

// ForEachCorePlugin calls the given CorePluginForEachFunc on each loaded core plugins.
func (n *Node) ForEachCorePlugin(f CorePluginForEachFunc) {
	forEachCorePlugin(n.corePlugins, f)
}

// PluginForEachFunc is used in ForEachPlugin.
// Returning false indicates to stop looping.
type PluginForEachFunc func(plugin *Plugin) bool

func forEachPlugin(plugins []*Plugin, f PluginForEachFunc) {
	for _, plugin := range plugins {
		if !f(plugin) {
			break
		}
	}
}

// ForEachPlugin calls the given PluginForEachFunc on each loaded plugin.
func (n *Node) ForEachPlugin(f PluginForEachFunc) {
	forEachPlugin(n.plugins, f)
}
<file_sep>package mqtt

import (
	"fmt"
	"net/url"

	"github.com/labstack/echo/v4"
	"github.com/labstack/echo/v4/middleware"
	"go.uber.org/dig"

	"github.com/gohornet/hornet/pkg/model/milestone"
	"github.com/gohornet/hornet/pkg/model/storage"
	"github.com/gohornet/hornet/pkg/model/utxo"
	mqttpkg "github.com/gohornet/hornet/pkg/mqtt"
	"github.com/gohornet/hornet/pkg/node"
	"github.com/gohornet/hornet/pkg/shutdown"
	"github.com/gohornet/hornet/pkg/tangle"
	"github.com/gohornet/hornet/plugins/restapi"
	"github.com/iotaledger/hive.go/configuration"
	"github.com/iotaledger/hive.go/events"
	"github.com/iotaledger/hive.go/workerpool"
	iotago "github.com/iotaledger/iota.go/v2"
)

func init() {
	Plugin = &node.Plugin{
		Status: node.StatusEnabled,
		Pluggable: node.Pluggable{
			Name:      "MQTT",
			DepsFunc:  func(cDeps dependencies) { deps = cDeps },
			Params:    params,
			Configure: configure,
			Run:       run,
		},
	}
}

const (
	// RouteMQTT is the route for accessing the MQTT over WebSockets.
	RouteMQTT = "/mqtt"

	workerCount     = 1
	workerQueueSize = 10000
)

var (
	Plugin *node.Plugin
	deps   dependencies

	newLatestMilestoneWorkerPool    *workerpool.WorkerPool
	newConfirmedMilestoneWorkerPool *workerpool.WorkerPool

	messagesWorkerPool        *workerpool.WorkerPool
	messageMetadataWorkerPool *workerpool.WorkerPool
	utxoOutputWorkerPool      *workerpool.WorkerPool
	receiptWorkerPool         *workerpool.WorkerPool

	topicSubscriptionWorkerPool *workerpool.WorkerPool

	wasSyncBefore = false

	mqttBroker *mqttpkg.Broker
)

type dependencies struct {
	dig.In
	Storage                               *storage.Storage
	Tangle                                *tangle.Tangle
	NodeConfig                            *configuration.Configuration `name:"nodeConfig"`
	MaxDeltaMsgYoungestConeRootIndexToCMI int                          `name:"maxDeltaMsgYoungestConeRootIndexToCMI"`
	MaxDeltaMsgOldestConeRootIndexToCMI   int                          `name:"maxDeltaMsgOldestConeRootIndexToCMI"`
	BelowMaxDepth                         int                          `name:"belowMaxDepth"`
	Bech32HRP                             iotago.NetworkPrefix         `name:"bech32HRP"`
	Echo                                  *echo.Echo                   `optional:"true"`
}

func configure() {
	// check if RestAPI plugin is disabled
	if Plugin.Node.IsSkipped(restapi.Plugin) {
		Plugin.Panic("RestAPI plugin needs to be enabled to use the MQTT plugin")
	}

	newLatestMilestoneWorkerPool = workerpool.New(func(task workerpool.Task) {
		publishLatestMilestone(task.Param(0).(*storage.CachedMilestone)) // milestone pass +1
		task.Return(nil)
	}, workerpool.WorkerCount(workerCount), workerpool.QueueSize(workerQueueSize), workerpool.FlushTasksAtShutdown(true))

	newConfirmedMilestoneWorkerPool = workerpool.New(func(task workerpool.Task) {
		publishConfirmedMilestone(task.Param(0).(*storage.CachedMilestone)) // milestone pass +1
		task.Return(nil)
	}, workerpool.WorkerCount(workerCount), workerpool.QueueSize(workerQueueSize), workerpool.FlushTasksAtShutdown(true))

	messagesWorkerPool = workerpool.New(func(task workerpool.Task) {
		publishMessage(task.Param(0).(*storage.CachedMessage)) // metadata pass +1
		task.Return(nil)
	}, workerpool.WorkerCount(workerCount), workerpool.QueueSize(workerQueueSize), workerpool.FlushTasksAtShutdown(true))

	messageMetadataWorkerPool = workerpool.New(func(task workerpool.Task) {
		publishMessageMetadata(task.Param(0).(*storage.CachedMetadata)) // metadata pass +1
		task.Return(nil)
	}, workerpool.WorkerCount(workerCount), workerpool.QueueSize(workerQueueSize), workerpool.FlushTasksAtShutdown(true))

	utxoOutputWorkerPool = workerpool.New(func(task workerpool.Task) {
		publishOutput(task.Param(0).(milestone.Index), task.Param(1).(*utxo.Output), task.Param(2).(bool))
		task.Return(nil)
	}, workerpool.WorkerCount(workerCount), workerpool.QueueSize(workerQueueSize))

	receiptWorkerPool = workerpool.New(func(task workerpool.Task) {
		publishReceipt(task.Param(0).(*iotago.Receipt))
		task.Return(nil)
	}, workerpool.WorkerCount(workerCount), workerpool.QueueSize(workerQueueSize))

	topicSubscriptionWorkerPool = workerpool.New(func(task workerpool.Task) {
		defer task.Return(nil)

		topic := task.Param(0).([]byte)
		topicName := string(topic)

		if messageID := messageIDFromTopic(topicName); messageID != nil {
			if cachedMsgMeta := deps.Storage.CachedMessageMetadataOrNil(messageID); cachedMsgMeta != nil {
				if _, added := messageMetadataWorkerPool.TrySubmit(cachedMsgMeta); added {
					return // Avoid Release (done inside workerpool task)
				}
				cachedMsgMeta.Release(true)
			}
			return
		}

		if transactionID := transactionIDFromTopic(topicName); transactionID != nil {
			// Find the first output of the transaction
			outputID := &iotago.UTXOInputID{}
			copy(outputID[:], transactionID[:])

			output, err := deps.Storage.UTXO().ReadOutputByOutputIDWithoutLocking(outputID)
			if err != nil {
				return
			}

			publishTransactionIncludedMessage(transactionID, output.MessageID())
			return
		}

		if outputID := outputIDFromTopic(topicName); outputID != nil {

			// we need to lock the ledger here to have the correct index for unspent info of the output.
			deps.Storage.UTXO().ReadLockLedger()
			defer deps.Storage.UTXO().ReadUnlockLedger()

			ledgerIndex, err := deps.Storage.UTXO().ReadLedgerIndexWithoutLocking()
			if err != nil {
				return
			}

			output, err := deps.Storage.UTXO().ReadOutputByOutputIDWithoutLocking(outputID)
			if err != nil {
				return
			}

			unspent, err := deps.Storage.UTXO().IsOutputUnspentWithoutLocking(output)
			if err != nil {
				return
			}
			utxoOutputWorkerPool.TrySubmit(ledgerIndex, output, !unspent)
			return
		}

		if topicName == topicMilestonesLatest {
			index := deps.Storage.LatestMilestoneIndex()
			if milestone := deps.Storage.CachedMilestoneOrNil(index); milestone != nil {
				publishLatestMilestone(milestone) // milestone pass +1
			}
			return
		}

		if topicName == topicMilestonesConfirmed {
			index := deps.Storage.ConfirmedMilestoneIndex()
			if milestone := deps.Storage.CachedMilestoneOrNil(index); milestone != nil {
				publishConfirmedMilestone(milestone) // milestone pass +1
			}
			return
		}

	}, workerpool.WorkerCount(workerCount), workerpool.QueueSize(workerQueueSize), workerpool.FlushTasksAtShutdown(true))

	var err error
	mqttBroker, err = mqttpkg.NewBroker(deps.NodeConfig.String(CfgMQTTBindAddress), deps.NodeConfig.Int(CfgMQTTWSPort), "/ws", deps.NodeConfig.Int(CfgMQTTWorkerCount), func(topic []byte) {
		Plugin.LogInfof("Subscribe to topic: %s", string(topic))
		topicSubscriptionWorkerPool.TrySubmit(topic)
	}, func(topic []byte) {
		Plugin.LogInfof("Unsubscribe from topic: %s", string(topic))
	})

	if err != nil {
		Plugin.LogFatalf("MQTT broker init failed! %s", err)
	}

	setupWebSocketRoute()
}

func setupWebSocketRoute() {

	// Configure MQTT WebSocket route
	mqttWSUrl, err := url.Parse(fmt.Sprintf("http://%s:%s", mqttBroker.Config().Host, mqttBroker.Config().WsPort))
	if err != nil {
		Plugin.LogFatalf("MQTT WebSocket init failed! %s", err)
	}

	wsGroup := deps.Echo.Group(RouteMQTT)
	proxyConfig := middleware.ProxyConfig{
		Skipper: middleware.DefaultSkipper,
		Balancer: middleware.NewRoundRobinBalancer([]*middleware.ProxyTarget{
			{
				URL: mqttWSUrl,
			},
		}),
		// We need to forward any calls to the MQTT route to the ws endpoint of our broker
		Rewrite: map[string]string{
			RouteMQTT: mqttBroker.Config().WsPath,
		},
	}

	wsGroup.Use(middleware.ProxyWithConfig(proxyConfig))
}

func run() {

	Plugin.LogInfof("Starting MQTT Broker (port %s) ...", mqttBroker.Config().Port)

	onLatestMilestoneChanged := events.NewClosure(func(cachedMs *storage.CachedMilestone) {
		if !wasSyncBefore {
			// Not sync
			cachedMs.Release(true)
			return
		}

		if _, added := newLatestMilestoneWorkerPool.TrySubmit(cachedMs); added {
			return // Avoid Release (done inside workerpool task)
		}
		cachedMs.Release(true)
	})

	onConfirmedMilestoneChanged := events.NewClosure(func(cachedMs *storage.CachedMilestone) {
		if !wasSyncBefore {
			if !deps.Storage.IsNodeAlmostSynced() {
				cachedMs.Release(true)
				return
			}
			wasSyncBefore = true
		}

		if _, added := newConfirmedMilestoneWorkerPool.TrySubmit(cachedMs); added {
			return // Avoid Release (done inside workerpool task)
		}
		cachedMs.Release(true)
	})

	onReceivedNewMessage := events.NewClosure(func(cachedMsg *storage.CachedMessage, _ milestone.Index, _ milestone.Index) {
		if !wasSyncBefore {
			// Not sync
			cachedMsg.Release(true)
			return
		}

		if _, added := messagesWorkerPool.TrySubmit(cachedMsg); added {
			return // Avoid Release (done inside workerpool task)
		}
		cachedMsg.Release(true)
	})

	onMessageSolid := events.NewClosure(func(cachedMetadata *storage.CachedMetadata) {
		if _, added := messageMetadataWorkerPool.TrySubmit(cachedMetadata); added {
			return // Avoid Release (done inside workerpool task)
		}
		cachedMetadata.Release(true)
	})

	onMessageReferenced := events.NewClosure(func(cachedMetadata *storage.CachedMetadata, _ milestone.Index, _ uint64) {
		if _, added := messageMetadataWorkerPool.TrySubmit(cachedMetadata); added {
			return // Avoid Release (done inside workerpool task)
		}
		cachedMetadata.Release(true)
	})

	onUTXOOutput := events.NewClosure(func(index milestone.Index, output *utxo.Output) {
		utxoOutputWorkerPool.TrySubmit(index, output, false)
	})

	onUTXOSpent := events.NewClosure(func(index milestone.Index, spent *utxo.Spent) {
		utxoOutputWorkerPool.TrySubmit(index, spent.Output(), true)
	})

	onReceipt := events.NewClosure(func(receipt *iotago.Receipt) {
		receiptWorkerPool.TrySubmit(receipt)
	})

	if err := Plugin.Daemon().BackgroundWorker("MQTT Broker", func(shutdownSignal <-chan struct{}) {
		go func() {
			mqttBroker.Start()
			Plugin.LogInfof("Starting MQTT Broker (port %s) ... done", mqttBroker.Config().Port)
		}()

		if mqttBroker.Config().Port != "" {
			Plugin.LogInfof("You can now listen to MQTT via: http://%s:%s", mqttBroker.Config().Host, mqttBroker.Config().Port)
		}

		if mqttBroker.Config().TlsPort != "" {
			Plugin.LogInfof("You can now listen to MQTT via: https://%s:%s", mqttBroker.Config().TlsHost, mqttBroker.Config().TlsPort)
		}

		<-shutdownSignal
		Plugin.LogInfo("Stopping MQTT Broker ...")
		Plugin.LogInfo("Stopping MQTT Broker ... done")
	}, shutdown.PriorityMetricsPublishers); err != nil {
		Plugin.Panicf("failed to start worker: %s", err)
	}

	if err := Plugin.Daemon().BackgroundWorker("MQTT Events", func(shutdownSignal <-chan struct{}) {
		Plugin.LogInfo("Starting MQTT Events ... done")

		deps.Tangle.Events.LatestMilestoneChanged.Attach(onLatestMilestoneChanged)
		deps.Tangle.Events.ConfirmedMilestoneChanged.Attach(onConfirmedMilestoneChanged)

		deps.Tangle.Events.ReceivedNewMessage.Attach(onReceivedNewMessage)
		deps.Tangle.Events.MessageSolid.Attach(onMessageSolid)
		deps.Tangle.Events.MessageReferenced.Attach(onMessageReferenced)

		deps.Tangle.Events.NewUTXOOutput.Attach(onUTXOOutput)
		deps.Tangle.Events.NewUTXOSpent.Attach(onUTXOSpent)

		deps.Tangle.Events.NewReceipt.Attach(onReceipt)

		messagesWorkerPool.Start()
		newLatestMilestoneWorkerPool.Start()
		newConfirmedMilestoneWorkerPool.Start()
		messageMetadataWorkerPool.Start()
		topicSubscriptionWorkerPool.Start()
		utxoOutputWorkerPool.Start()
		receiptWorkerPool.Start()

		<-shutdownSignal

		deps.Tangle.Events.LatestMilestoneChanged.Detach(onLatestMilestoneChanged)
		deps.Tangle.Events.ConfirmedMilestoneChanged.Detach(onConfirmedMilestoneChanged)

		deps.Tangle.Events.ReceivedNewMessage.Detach(onReceivedNewMessage)
		deps.Tangle.Events.MessageSolid.Detach(onMessageSolid)
		deps.Tangle.Events.MessageReferenced.Detach(onMessageReferenced)

		deps.Tangle.Events.NewUTXOOutput.Detach(onUTXOOutput)
		deps.Tangle.Events.NewUTXOSpent.Detach(onUTXOSpent)

		deps.Tangle.Events.NewReceipt.Detach(onReceipt)

		messagesWorkerPool.StopAndWait()
		newLatestMilestoneWorkerPool.StopAndWait()
		newConfirmedMilestoneWorkerPool.StopAndWait()
		messageMetadataWorkerPool.StopAndWait()
		topicSubscriptionWorkerPool.StopAndWait()
		utxoOutputWorkerPool.StopAndWait()
		receiptWorkerPool.StopAndWait()

		Plugin.LogInfo("Stopping MQTT Events ... done")
	}, shutdown.PriorityMetricsPublishers); err != nil {
		Plugin.Panicf("failed to start worker: %s", err)
	}
}
<file_sep>---
keywords:
- IOTA Node 
- Hornet Node
- Troubleshooting
- Problems
- Discord
- FAQ
- Difficulties
description: Resources to troubleshoot your node.  
image: /img/logo/HornetLogo.png
---

# Troubleshooting

Check our [Frequently asked questions](faq.md).

If you can't find your question in the FAQ, feel free to ask in the `#hornet` channel in the ([official iota discord server](https://discord.iota.org/)).

## Something went wrong?

Please open a [new issue](https://github.com/gohornet/hornet/issues/new) if you detect an error or crash. You can also submit a PR if you have already fixed it. 
<file_sep>package urts

import (
	"time"

	"go.uber.org/dig"

	"github.com/gohornet/hornet/pkg/metrics"
	"github.com/gohornet/hornet/pkg/model/storage"
	"github.com/gohornet/hornet/pkg/node"
	"github.com/gohornet/hornet/pkg/shutdown"
	"github.com/gohornet/hornet/pkg/tangle"
	"github.com/gohornet/hornet/pkg/tipselect"
	"github.com/gohornet/hornet/pkg/whiteflag"
	"github.com/iotaledger/hive.go/configuration"
	"github.com/iotaledger/hive.go/events"
)

func init() {
	Plugin = &node.Plugin{
		Status: node.StatusEnabled,
		Pluggable: node.Pluggable{
			Name:           "URTS",
			DepsFunc:       func(cDeps dependencies) { deps = cDeps },
			Params:         params,
			InitConfigPars: initConfigPars,
			Provide:        provide,
			Configure:      configure,
			Run:            run,
		},
	}
}

var (
	Plugin *node.Plugin
	deps   dependencies

	// Closures
	onMessageSolid       *events.Closure
	onMilestoneConfirmed *events.Closure
)

type dependencies struct {
	dig.In
	TipSelector *tipselect.TipSelector
	Storage     *storage.Storage
	Tangle      *tangle.Tangle
}

func initConfigPars(c *dig.Container) {

	type cfgDeps struct {
		dig.In
		NodeConfig *configuration.Configuration `name:"nodeConfig"`
	}

	type cfgResult struct {
		dig.Out
		MaxDeltaMsgYoungestConeRootIndexToCMI int `name:"maxDeltaMsgYoungestConeRootIndexToCMI"`
		MaxDeltaMsgOldestConeRootIndexToCMI   int `name:"maxDeltaMsgOldestConeRootIndexToCMI"`
		BelowMaxDepth                         int `name:"belowMaxDepth"`
	}

	if err := c.Provide(func(deps cfgDeps) cfgResult {
		return cfgResult{
			MaxDeltaMsgYoungestConeRootIndexToCMI: deps.NodeConfig.Int(CfgTipSelMaxDeltaMsgYoungestConeRootIndexToCMI),
			MaxDeltaMsgOldestConeRootIndexToCMI:   deps.NodeConfig.Int(CfgTipSelMaxDeltaMsgOldestConeRootIndexToCMI),
			BelowMaxDepth:                         deps.NodeConfig.Int(CfgTipSelBelowMaxDepth),
		}
	}); err != nil {
		Plugin.Panic(err)
	}
}

func provide(c *dig.Container) {

	type tipselDeps struct {
		dig.In
		Storage                               *storage.Storage
		ServerMetrics                         *metrics.ServerMetrics
		NodeConfig                            *configuration.Configuration `name:"nodeConfig"`
		MaxDeltaMsgYoungestConeRootIndexToCMI int                          `name:"maxDeltaMsgYoungestConeRootIndexToCMI"`
		MaxDeltaMsgOldestConeRootIndexToCMI   int                          `name:"maxDeltaMsgOldestConeRootIndexToCMI"`
		BelowMaxDepth                         int                          `name:"belowMaxDepth"`
	}

	if err := c.Provide(func(deps tipselDeps) *tipselect.TipSelector {
		return tipselect.New(
			deps.Storage,
			deps.ServerMetrics,

			deps.MaxDeltaMsgYoungestConeRootIndexToCMI,
			deps.MaxDeltaMsgOldestConeRootIndexToCMI,
			deps.BelowMaxDepth,

			deps.NodeConfig.Int(CfgTipSelNonLazy+CfgTipSelRetentionRulesTipsLimit),
			deps.NodeConfig.Duration(CfgTipSelNonLazy+CfgTipSelMaxReferencedTipAge),
			uint32(deps.NodeConfig.Int64(CfgTipSelNonLazy+CfgTipSelMaxChildren)),
			deps.NodeConfig.Int(CfgTipSelNonLazy+CfgTipSelSpammerTipsThreshold),

			deps.NodeConfig.Int(CfgTipSelSemiLazy+CfgTipSelRetentionRulesTipsLimit),
			deps.NodeConfig.Duration(CfgTipSelSemiLazy+CfgTipSelMaxReferencedTipAge),
			uint32(deps.NodeConfig.Int64(CfgTipSelSemiLazy+CfgTipSelMaxChildren)),
			deps.NodeConfig.Int(CfgTipSelSemiLazy+CfgTipSelSpammerTipsThreshold),
		)
	}); err != nil {
		Plugin.Panic(err)
	}
}

func configure() {
	configureEvents()
}

func run() {

	if err := Plugin.Daemon().BackgroundWorker("Tipselection[Events]", func(shutdownSignal <-chan struct{}) {
		attachEvents()
		<-shutdownSignal
		detachEvents()
	}, shutdown.PriorityTipselection); err != nil {
		Plugin.Panicf("failed to start worker: %s", err)
	}

	if err := Plugin.Daemon().BackgroundWorker("Tipselection[Cleanup]", func(shutdownSignal <-chan struct{}) {
		for {
			select {
			case <-shutdownSignal:
				return
			case <-time.After(time.Second):
				ts := time.Now()
				removedTipCount := deps.TipSelector.CleanUpReferencedTips()
				Plugin.LogDebugf("CleanUpReferencedTips finished, removed: %d, took: %v", removedTipCount, time.Since(ts).Truncate(time.Millisecond))
			}
		}
	}, shutdown.PriorityTipselection); err != nil {
		Plugin.Panicf("failed to start worker: %s", err)
	}
}

func configureEvents() {
	onMessageSolid = events.NewClosure(func(cachedMsgMeta *storage.CachedMetadata) {
		cachedMsgMeta.ConsumeMetadata(func(metadata *storage.MessageMetadata) { // metadata -1
			// do not add tips during syncing, because it is not needed at all
			if !deps.Storage.IsNodeAlmostSynced() {
				return
			}

			deps.TipSelector.AddTip(metadata)
		})
	})

	onMilestoneConfirmed = events.NewClosure(func(_ *whiteflag.Confirmation) {
		// do not update tip scores during syncing, because it is not needed at all
		if !deps.Storage.IsNodeAlmostSynced() {
			return
		}

		ts := time.Now()
		removedTipCount := deps.TipSelector.UpdateScores()
		Plugin.LogDebugf("UpdateScores finished, removed: %d, took: %v", removedTipCount, time.Since(ts).Truncate(time.Millisecond))
	})
}

func attachEvents() {
	deps.Tangle.Events.MessageSolid.Attach(onMessageSolid)
	deps.Tangle.Events.MilestoneConfirmed.Attach(onMilestoneConfirmed)
}

func detachEvents() {
	deps.Tangle.Events.MessageSolid.Detach(onMessageSolid)
	deps.Tangle.Events.MilestoneConfirmed.Detach(onMilestoneConfirmed)
}
<file_sep>package database

import (
	"os"

	flag "github.com/spf13/pflag"
	"go.uber.org/dig"

	"github.com/gohornet/hornet/pkg/database"
	"github.com/gohornet/hornet/pkg/keymanager"
	"github.com/gohornet/hornet/pkg/metrics"
	"github.com/gohornet/hornet/pkg/model/coordinator"
	"github.com/gohornet/hornet/pkg/model/storage"
	"github.com/gohornet/hornet/pkg/model/utxo"
	"github.com/gohornet/hornet/pkg/node"
	"github.com/gohornet/hornet/pkg/profile"
	"github.com/gohornet/hornet/pkg/shutdown"
	"github.com/gohornet/hornet/pkg/utils"
	"github.com/iotaledger/hive.go/configuration"
	"github.com/iotaledger/hive.go/events"
	"github.com/iotaledger/hive.go/kvstore/pebble"
	"github.com/iotaledger/hive.go/kvstore/rocksdb"
	"github.com/iotaledger/hive.go/logger"
)

const (
	// whether to delete the database at startup
	CfgTangleDeleteDatabase = "deleteDatabase"
	// whether to delete the database and snapshots at startup
	CfgTangleDeleteAll = "deleteAll"
)

func init() {
	CorePlugin = &node.CorePlugin{
		Pluggable: node.Pluggable{
			Name:           "Database",
			DepsFunc:       func(cDeps dependencies) { deps = cDeps },
			Params:         params,
			InitConfigPars: initConfigPars,
			Provide:        provide,
			Configure:      configure,
			Run:            run,
		},
	}
}

var (
	CorePlugin *node.CorePlugin
	deps       dependencies

	deleteDatabase = flag.Bool(CfgTangleDeleteDatabase, false, "whether to delete the database at startup")
	deleteAll      = flag.Bool(CfgTangleDeleteAll, false, "whether to delete the database and snapshots at startup")

	// Closures
	onPruningStateChanged *events.Closure
)

type dependencies struct {
	dig.In
	Database       *database.Database
	Storage        *storage.Storage
	StorageMetrics *metrics.StorageMetrics
}

func initConfigPars(c *dig.Container) {

	type cfgDeps struct {
		dig.In
		NodeConfig *configuration.Configuration `name:"nodeConfig"`
	}

	type cfgResult struct {
		dig.Out
		DatabaseEngine           database.Engine `name:"databaseEngine"`
		DatabasePath             string          `name:"databasePath"`
		DeleteDatabaseFlag       bool            `name:"deleteDatabase"`
		DeleteAllFlag            bool            `name:"deleteAll"`
		DatabaseDebug            bool            `name:"databaseDebug"`
		DatabaseAutoRevalidation bool            `name:"databaseAutoRevalidation"`
	}

	if err := c.Provide(func(deps cfgDeps) cfgResult {
		dbEngine, err := database.DatabaseEngine(deps.NodeConfig.String(CfgDatabaseEngine))
		if err != nil {
			CorePlugin.Panic(err)
		}
		return cfgResult{
			DatabaseEngine:           dbEngine,
			DatabasePath:             deps.NodeConfig.String(CfgDatabasePath),
			DeleteDatabaseFlag:       *deleteDatabase,
			DeleteAllFlag:            *deleteAll,
			DatabaseDebug:            deps.NodeConfig.Bool(CfgDatabaseDebug),
			DatabaseAutoRevalidation: deps.NodeConfig.Bool(CfgDatabaseAutoRevalidation),
		}
	}); err != nil {
		CorePlugin.Panic(err)
	}
}

func provide(c *dig.Container) {

	type dbResult struct {
		dig.Out
		StorageMetrics  *metrics.StorageMetrics
		DatabaseMetrics *metrics.DatabaseMetrics
	}

	if err := c.Provide(func() dbResult {
		return dbResult{
			StorageMetrics:  &metrics.StorageMetrics{},
			DatabaseMetrics: &metrics.DatabaseMetrics{},
		}
	}); err != nil {
		CorePlugin.Panic(err)
	}

	type databaseDeps struct {
		dig.In
		DeleteDatabaseFlag bool                         `name:"deleteDatabase"`
		DeleteAllFlag      bool                         `name:"deleteAll"`
		NodeConfig         *configuration.Configuration `name:"nodeConfig"`
		DatabaseEngine     database.Engine              `name:"databaseEngine"`
		DatabasePath       string                       `name:"databasePath"`
		Metrics            *metrics.DatabaseMetrics
	}

	if err := c.Provide(func(deps databaseDeps) *database.Database {

		events := &database.Events{
			DatabaseCleanup:    events.NewEvent(database.DatabaseCleanupCaller),
			DatabaseCompaction: events.NewEvent(events.BoolCaller),
		}

		if deps.DeleteDatabaseFlag || deps.DeleteAllFlag {
			// delete old database folder
			if err := os.RemoveAll(deps.DatabasePath); err != nil {
				CorePlugin.Panicf("deleting database folder failed: %s", err)
			}
		}

		targetEngine, err := database.CheckDatabaseEngine(deps.DatabasePath, true, deps.DatabaseEngine)
		if err != nil {
			CorePlugin.Panic(err)
		}

		switch targetEngine {
		case database.EnginePebble:
			reportCompactionRunning := func(running bool) {
				deps.Metrics.CompactionRunning.Store(running)
				if running {
					deps.Metrics.CompactionCount.Inc()
				}
				events.DatabaseCompaction.Trigger(running)
			}

			db, err := database.NewPebbleDB(deps.DatabasePath, reportCompactionRunning, true)
			if err != nil {
				CorePlugin.Panicf("database initialization failed: %s", err)
			}

			return database.New(
				CorePlugin.Logger(),
				pebble.New(db),
				events,
				true,
				func() bool { return deps.Metrics.CompactionRunning.Load() },
			)

		case database.EngineRocksDB:
			db, err := database.NewRocksDB(deps.DatabasePath)
			if err != nil {
				CorePlugin.Panicf("database initialization failed: %s", err)
			}

			return database.New(
				CorePlugin.Logger(),
				rocksdb.New(db),
				events,
				true,
				func() bool {
					if numCompactions, success := db.GetIntProperty("rocksdb.num-running-compactions"); success {
						runningBefore := deps.Metrics.CompactionRunning.Load()
						running := numCompactions != 0

						deps.Metrics.CompactionRunning.Store(running)
						if running && !runningBefore {
							// we may miss some compactions, since this is only calculated if polled.
							deps.Metrics.CompactionCount.Inc()
							events.DatabaseCompaction.Trigger(running)
						}
						return running
					}
					return false
				},
			)
		default:
			CorePlugin.Panicf("unknown database engine: %s, supported engines: pebble/rocksdb", targetEngine)
			return nil
		}
	}); err != nil {
		CorePlugin.Panic(err)
	}

	type storageDeps struct {
		dig.In
		NodeConfig                 *configuration.Configuration `name:"nodeConfig"`
		DatabasePath               string                       `name:"databasePath"`
		Database                   *database.Database
		Profile                    *profile.Profile
		BelowMaxDepth              int `name:"belowMaxDepth"`
		CoordinatorPublicKeyRanges coordinator.PublicKeyRanges
		MilestonePublicKeyCount    int `name:"milestonePublicKeyCount"`
	}

	if err := c.Provide(func(deps storageDeps) *storage.Storage {

		keyManager := keymanager.New()
		for _, keyRange := range deps.CoordinatorPublicKeyRanges {
			pubKey, err := utils.ParseEd25519PublicKeyFromString(keyRange.Key)
			if err != nil {
				CorePlugin.Panicf("can't load public key ranges: %s", err)
			}

			keyManager.AddKeyRange(pubKey, keyRange.StartIndex, keyRange.EndIndex)
		}

		store, err := storage.New(logger.NewLogger("Storage"), deps.DatabasePath, deps.Database.KVStore(), deps.Profile.Caches, deps.BelowMaxDepth, keyManager, deps.MilestonePublicKeyCount)
		if err != nil {
			CorePlugin.Panicf("can't initialize storage: %s", err)
		}
		return store
	}); err != nil {
		CorePlugin.Panic(err)
	}

	if err := c.Provide(func(storage *storage.Storage) *utxo.Manager {
		return storage.UTXO()
	}); err != nil {
		CorePlugin.Panic(err)
	}
}

func configure() {

	correctDatabaseVersion, err := deps.Storage.IsCorrectDatabaseVersion()
	if err != nil {
		CorePlugin.Panic(err)
	}

	if !correctDatabaseVersion {
		databaseVersionUpdated, err := deps.Storage.UpdateDatabaseVersion()
		if err != nil {
			CorePlugin.Panic(err)
		}

		if !databaseVersionUpdated {
			CorePlugin.Panic("HORNET database version mismatch. The database scheme was updated. Please delete the database folder and start with a new snapshot.")
		}
	}

	if err = CorePlugin.Daemon().BackgroundWorker("Close database", func(shutdownSignal <-chan struct{}) {
		<-shutdownSignal

		if err = deps.Storage.MarkDatabaseHealthy(); err != nil {
			CorePlugin.Panic(err)
		}

		CorePlugin.LogInfo("Syncing databases to disk...")
		if err = closeDatabases(); err != nil {
			CorePlugin.Panicf("Syncing databases to disk... failed: %s", err)
		}
		CorePlugin.LogInfo("Syncing databases to disk... done")
	}, shutdown.PriorityCloseDatabase); err != nil {
		CorePlugin.Panicf("failed to start worker: %s", err)
	}

	configureEvents()
}

func run() {
	if err := CorePlugin.Daemon().BackgroundWorker("Database[Events]", func(shutdownSignal <-chan struct{}) {
		attachEvents()
		<-shutdownSignal
		detachEvents()
	}, shutdown.PriorityMetricsUpdater); err != nil {
		CorePlugin.Panicf("failed to start worker: %s", err)
	}
}

func closeDatabases() error {

	if err := deps.Database.KVStore().Flush(); err != nil {
		return err
	}

	if err := deps.Database.KVStore().Close(); err != nil {
		return err
	}

	return nil
}

func configureEvents() {
	onPruningStateChanged = events.NewClosure(func(running bool) {
		deps.StorageMetrics.PruningRunning.Store(running)
		if running {
			deps.StorageMetrics.Prunings.Inc()
		}
	})
}

func attachEvents() {
	deps.Storage.Events.PruningStateChanged.Attach(onPruningStateChanged)
}

func detachEvents() {
	deps.Storage.Events.PruningStateChanged.Detach(onPruningStateChanged)
}
<file_sep>package faucet

import (
	"github.com/labstack/echo/v4"
	"github.com/pkg/errors"

	"github.com/gohornet/hornet/pkg/model/faucet"
	"github.com/gohornet/hornet/pkg/restapi"
	iotago "github.com/iotaledger/iota.go/v2"
)

func parseBech32Address(addressParam string) (*iotago.Ed25519Address, error) {

	hrp, bech32Address, err := iotago.ParseBech32(addressParam)
	if err != nil {
		return nil, errors.WithMessage(restapi.ErrInvalidParameter, "Invalid bech32 address provided!")
	}

	if hrp != deps.Faucet.NetworkPrefix() {
		return nil, errors.WithMessagef(restapi.ErrInvalidParameter, "Invalid bech32 address provided! Address does not start with \"%s\".", deps.Faucet.NetworkPrefix())
	}

	switch address := bech32Address.(type) {
	case *iotago.Ed25519Address:
		return address, nil
	default:
		return nil, errors.WithMessage(restapi.ErrInvalidParameter, "Invalid bech32 address provided! Unknown address type.")
	}
}

func getFaucetInfo(_ echo.Context) (*faucet.FaucetInfoResponse, error) {
	return deps.Faucet.Info()
}

func addFaucetOutputToQueue(c echo.Context) (*faucet.FaucetEnqueueResponse, error) {

	request := &faucetEnqueueRequest{}
	if err := c.Bind(request); err != nil {
		return nil, errors.WithMessagef(restapi.ErrInvalidParameter, "Invalid Request! Error: %s", err)
	}

	bech32Addr := request.Address
	ed25519Addr, err := parseBech32Address(bech32Addr)
	if err != nil {
		return nil, err
	}

	response, err := deps.Faucet.Enqueue(bech32Addr, ed25519Addr)
	if err != nil {
		return nil, err
	}

	return response, nil
}
<file_sep>package storage

import (
	"time"

	"github.com/gohornet/hornet/pkg/common"
	"github.com/gohornet/hornet/pkg/model/hornet"
	"github.com/gohornet/hornet/pkg/profile"
	"github.com/iotaledger/hive.go/kvstore"
	"github.com/iotaledger/hive.go/objectstorage"
	iotago "github.com/iotaledger/iota.go/v2"
)

// CachedIndexation represents a cached indexation.
type CachedIndexation struct {
	objectstorage.CachedObject
}

// Indexation retrieves the indexation, that is cached in this container.
func (c *CachedIndexation) Indexation() *Indexation {
	return c.Get().(*Indexation)
}

func indexationFactory(key []byte, data []byte) (objectstorage.StorableObject, error) {
	return &Indexation{
		index:     key[:IndexationIndexLength],
		messageID: hornet.MessageIDFromSlice(key[IndexationIndexLength : IndexationIndexLength+iotago.MessageIDLength]),
	}, nil
}

func (s *Storage) IndexationStorageSize() int {
	return s.indexationStorage.GetSize()
}

func (s *Storage) configureIndexationStorage(store kvstore.KVStore, opts *profile.CacheOpts) error {

	cacheTime, err := time.ParseDuration(opts.CacheTime)
	if err != nil {
		return err
	}

	leakDetectionMaxConsumerHoldTime, err := time.ParseDuration(opts.LeakDetectionOptions.MaxConsumerHoldTime)
	if err != nil {
		return err
	}

	s.indexationStorage = objectstorage.New(
		store.WithRealm([]byte{common.StorePrefixIndexation}),
		indexationFactory,
		objectstorage.CacheTime(cacheTime),
		objectstorage.PersistenceEnabled(true),
		objectstorage.PartitionKey(IndexationIndexLength, iotago.MessageIDLength),
		objectstorage.KeysOnly(true),
		objectstorage.StoreOnCreation(true),
		objectstorage.ReleaseExecutorWorkerCount(opts.ReleaseExecutorWorkerCount),
		objectstorage.LeakDetectionEnabled(opts.LeakDetectionOptions.Enabled,
			objectstorage.LeakDetectionOptions{
				MaxConsumersPerObject: opts.LeakDetectionOptions.MaxConsumersPerObject,
				MaxConsumerHoldTime:   leakDetectionMaxConsumerHoldTime,
			}),
	)

	return nil
}

// IndexMessageIDs returns all known message IDs for the given index.
// indexation +-0
func (s *Storage) IndexMessageIDs(index []byte, iteratorOptions ...IteratorOption) hornet.MessageIDs {
	var messageIDs hornet.MessageIDs

	indexPadded := PadIndexationIndex(index)

	s.indexationStorage.ForEachKeyOnly(func(key []byte) bool {
		messageIDs = append(messageIDs, hornet.MessageIDFromSlice(key[IndexationIndexLength:IndexationIndexLength+iotago.MessageIDLength]))
		return true
	}, append(iteratorOptions, objectstorage.WithIteratorPrefix(indexPadded[:]))...)

	return messageIDs
}

// IndexConsumer consumes the messageID during looping through all messages with given index.
type IndexConsumer func(messageID hornet.MessageID) bool

// ForEachMessageIDWithIndex loops over all messages with the given index.
func (s *Storage) ForEachMessageIDWithIndex(index []byte, consumer IndexConsumer, iteratorOptions ...IteratorOption) {
	indexPadded := PadIndexationIndex(index)

	s.indexationStorage.ForEachKeyOnly(func(key []byte) bool {
		return consumer(hornet.MessageIDFromSlice(key[IndexationIndexLength : IndexationIndexLength+iotago.MessageIDLength]))
	}, append(iteratorOptions, objectstorage.WithIteratorPrefix(indexPadded[:]))...)
}

// CachedIndexationConsumer consumes the given indexation during looping through all indexations.
type CachedIndexationConsumer func(indexation *CachedIndexation) bool

// ForEachIndexation loops over all indexations.
// indexation +1
func (s *Storage) ForEachIndexation(consumer CachedIndexationConsumer, iteratorOptions ...IteratorOption) {
	s.indexationStorage.ForEach(func(key []byte, cachedObject objectstorage.CachedObject) bool {
		return consumer(&CachedIndexation{CachedObject: cachedObject})
	}, iteratorOptions...)
}

// StoreIndexation stores the indexation in the persistence layer and returns a cached object.
// indexation +1
func (s *Storage) StoreIndexation(index []byte, messageID hornet.MessageID) *CachedIndexation {
	indexation := NewIndexation(index, messageID)
	return &CachedIndexation{CachedObject: s.indexationStorage.Store(indexation)}
}

// DeleteIndexation deletes the indexation in the cache/persistence layer.
// indexation +-0
func (s *Storage) DeleteIndexation(index []byte, messageID hornet.MessageID) {
	indexation := NewIndexation(index, messageID)
	s.indexationStorage.Delete(indexation.ObjectStorageKey())
}

// DeleteIndexationByKey deletes the indexation by key in the cache/persistence layer.
// indexation +-0
func (s *Storage) DeleteIndexationByKey(key []byte) {
	s.indexationStorage.Delete(key)
}

// ShutdownIndexationStorage shuts down the indexation storage.
func (s *Storage) ShutdownIndexationStorage() {
	s.indexationStorage.Shutdown()
}

// FlushIndexationStorage flushes the indexation storage.
func (s *Storage) FlushIndexationStorage() {
	s.indexationStorage.Flush()
}
<file_sep>package toolset

import (
	stded25519 "crypto/ed25519"
	"encoding/hex"
	"fmt"
	"os"
	"path/filepath"

	"github.com/libp2p/go-libp2p-core/crypto"
	"github.com/libp2p/go-libp2p-core/peer"
	"github.com/mr-tron/base58"

	p2pCore "github.com/gohornet/hornet/core/p2p"
	"github.com/gohornet/hornet/pkg/p2p"
	"github.com/gohornet/hornet/pkg/utils"
	"github.com/iotaledger/hive.go/configuration"
)

func generateP2PIdentity(nodeConfig *configuration.Configuration, args []string) error {

	printUsage := func() {
		println("Usage:")
		println(fmt.Sprintf("	%s [P2P_DATABASE_PATH] [P2P_PRIVATE_KEY]", ToolP2PIdentityGen))
		println()
		println("	[P2P_DATABASE_PATH] - the path to the p2p database folder (optional)")
		println("	[P2P_PRIVATE_KEY]   - the p2p private key (optional)")
		println()
		println(fmt.Sprintf("example: %s %s", ToolP2PIdentityGen, "p2pstore"))
	}

	if len(args) > 2 {
		printUsage()
		return fmt.Errorf("too many arguments for '%s'", ToolP2PIdentityGen)
	}

	p2pDatabasePath := nodeConfig.String(p2pCore.CfgP2PDatabasePath)
	if len(args) > 0 {
		p2pDatabasePath = args[0]
	}
	privKeyFilePath := filepath.Join(p2pDatabasePath, p2p.PrivKeyFileName)

	if err := os.MkdirAll(p2pDatabasePath, 0700); err != nil {
		return fmt.Errorf("could not create peer store database dir '%s': %w", p2pDatabasePath, err)
	}

	_, err := os.Stat(privKeyFilePath)
	switch {
	case err == nil || os.IsExist(err):
		// private key file already exists
		return fmt.Errorf("private key file (%s) already exists", privKeyFilePath)

	case os.IsNotExist(err):
		// private key file does not exist, create a new one

	default:
		return fmt.Errorf("unable to check private key file (%s): %w", privKeyFilePath, err)
	}

	var privateKey crypto.PrivKey
	var publicKey crypto.PubKey

	if len(args) > 1 {
		hivePrivKey, err := utils.ParseEd25519PrivateKeyFromString(args[1])
		if err != nil {
			return fmt.Errorf("invalid private key given '%s': %w", args[1], err)
		}

		stdPrvKey := stded25519.PrivateKey(hivePrivKey)
		privateKey, publicKey, err = crypto.KeyPairFromStdKey(&stdPrvKey)
		if err != nil {
			return fmt.Errorf("unable to convert given private key '%s': %w", args[1], err)
		}
	} else {
		// create identity
		privateKey, publicKey, err = crypto.GenerateKeyPair(crypto.Ed25519, -1)
		if err != nil {
			return fmt.Errorf("unable to generate Ed25519 private key for peer identity: %w", err)
		}
	}

	// obtain Peer ID from public key
	peerID, err := peer.IDFromPublicKey(publicKey)
	if err != nil {
		return fmt.Errorf("unable to get peer identity from public key: %w", err)
	}

	privKeyBytes, err := privateKey.Raw()
	if err != nil {
		return fmt.Errorf("unable to get raw private key bytes: %w", err)
	}

	pubKeyBytes, err := publicKey.Raw()
	if err != nil {
		return fmt.Errorf("unable to get raw public key bytes: %w", err)
	}

	if err := p2p.WriteEd25519PrivateKeyToPEMFile(privKeyFilePath, privateKey); err != nil {
		return fmt.Errorf("writing private key file for peer identity failed: %w", err)
	}

	fmt.Println("Your p2p private key (hex):   ", hex.EncodeToString(privKeyBytes))
	fmt.Println("Your p2p public key (hex):    ", hex.EncodeToString(pubKeyBytes))
	fmt.Println("Your p2p public key (base58): ", base58.Encode(pubKeyBytes))
	fmt.Println("Your p2p PeerID:              ", peerID.String())

	return nil
}
<file_sep># Keys for the private tangle

## Coordinator
1. private key:   <KEY>
1. public key:    ed3c3f1a319ff4e909cf2771d79fece0ac9bd9fd2ee49ea6c0885c9cb3b1248c
2. private key:   0e324c6ff069f31890d496e9004636fd73d8e8b5bea08ec58a4178ca85462325f6752f5f46a53364e2ee9c4d662d762a81efd51010282a75cd6bd03f28ef349c
2. public key:    f6752f5f46a53364e2ee9c4d662d762a81efd51010282a75cd6bd03f28ef349c

## Wallet

token mnemonic:            giant dynamic museum toddler six deny defense ostrich bomb access mercy blood explain muscle shoot shallow glad autumn author calm heavy hawk abuse rally <br>
token ed25519 private key: 52d23081a626b1eca34b63f1eaeeafcbd66bf545635befc12cd0f19926efefb0 <br>
token ed25519 public key:  31f176dadf38cdec0eadd1d571394be78f0bbee3ed594316678dffc162a095cb <br>
token ed25519 address:     60200bad8137a704216e84f8f9acfe65b972d9f4155becb4815282b03cef99fe <br>
token bech32 address:      atoi1qpszqzadsym6wpppd6z037dvlejmjuke7s24hm95s9fg9vpua7vluehe53e

## P2P Identities

### Coordinator node
p2p private key:   1f46fad4f538a031d4f87f490f6bca4319dfd0307636a5759a22b5e8874bd608f9156ba976a12918c16a481c38c88a7b5351b769adc30390e93b6c0a63b09b79 <br>
p2p public key:    f9156ba976a12918c16a481c38c88a7b5351b769adc30390e93b6c0a63b09b79 <br>
p2p PeerID:        12D3KooWSagdVaCrS14GeJhM8CbQr41AW2PiYMgptTyAybCbQuEY

### 2nd node
p2p private key:   <KEY> <br>
p2p public key:    <KEY> <br>
p2p PeerID:        12D3KooWCKwcTWevoRKa2kEBputeGASvEBuDfRDSbe8t1DWugUmL

### 3nd node
p2p private key:   <KEY> <br>
p2p public key:    <KEY> <br>
p2p PeerID:        12D3KooWGdr8M5KX8KuKaXSiKfHJstdVnRkadYmupF7tFk2HrRoA

### 4th node
p2p private key:   <KEY> <br>
p2p public key:    <KEY> <br>
p2p PeerID:        12D3KooWC7uE9w3RN4Vh1FJAZa8SbE8yMWR6wCVBajcWpyWguV73

## Dashboard

Username: admin <br>
Password: <PASSWORD>
<file_sep>package shutdown

import (
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"

	"github.com/iotaledger/hive.go/daemon"
	"github.com/iotaledger/hive.go/logger"
)

const (
	// the maximum amount of time to wait for background processes to terminate. After that the process is killed.
	waitToKillTimeInSeconds = 300
)

// ShutdownHandler waits until a shutdown signal was received or the node tried to shutdown itself,
// and shuts down all processes gracefully.
type ShutdownHandler struct {
	log              *logger.Logger
	daemon           daemon.Daemon
	gracefulStop     chan os.Signal
	nodeSelfShutdown chan string
}

// NewShutdownHandler creates a new shutdown handler.
func NewShutdownHandler(log *logger.Logger, daemon daemon.Daemon) *ShutdownHandler {

	gs := &ShutdownHandler{
		log:              log,
		daemon:           daemon,
		gracefulStop:     make(chan os.Signal, 1),
		nodeSelfShutdown: make(chan string),
	}

	signal.Notify(gs.gracefulStop, syscall.SIGTERM)
	signal.Notify(gs.gracefulStop, syscall.SIGINT)

	return gs
}

// SelfShutdown can be called in order to instruct the node to shutdown cleanly without receiving any interrupt signals.
func (gs *ShutdownHandler) SelfShutdown(msg string) {
	select {
	case gs.nodeSelfShutdown <- msg:
	default:
	}
}

// Run starts the ShutdownHandler go routine.
func (gs *ShutdownHandler) Run() {

	go func() {
		select {
		case <-gs.gracefulStop:
			gs.log.Warnf("Received shutdown request - waiting (max %d seconds) to finish processing ...", waitToKillTimeInSeconds)
		case msg := <-gs.nodeSelfShutdown:
			gs.log.Warnf("Node self-shutdown: %s; waiting (max %d seconds) to finish processing ...", msg, waitToKillTimeInSeconds)
		}

		go func() {
			start := time.Now()
			for x := range time.Tick(1 * time.Second) {
				secondsSinceStart := x.Sub(start).Seconds()

				if secondsSinceStart <= waitToKillTimeInSeconds {
					processList := ""
					runningBackgroundWorkers := gs.daemon.GetRunningBackgroundWorkers()
					if len(runningBackgroundWorkers) >= 1 {
						processList = "(" + strings.Join(runningBackgroundWorkers, ", ") + ") "
					}

					gs.log.Warnf("Received shutdown request - waiting (max %d seconds) to finish processing %s...", waitToKillTimeInSeconds-int(secondsSinceStart), processList)
				} else {
					gs.log.Fatal("Background processes did not terminate in time! Forcing shutdown ...")
				}
			}
		}()

		gs.daemon.ShutdownAndWait()
	}()
}
<file_sep>package prometheus

import (
	"context"
	"encoding/json"
	"io/ioutil"
	"net/http"
	"time"

	"github.com/pkg/errors"

	"github.com/labstack/echo/v4"
	"github.com/labstack/echo/v4/middleware"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/collectors"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"go.uber.org/dig"

	"github.com/gohornet/hornet/pkg/app"
	"github.com/gohornet/hornet/pkg/database"
	"github.com/gohornet/hornet/pkg/metrics"
	"github.com/gohornet/hornet/pkg/model/coordinator"
	"github.com/gohornet/hornet/pkg/model/migrator"
	"github.com/gohornet/hornet/pkg/model/storage"
	"github.com/gohornet/hornet/pkg/node"
	"github.com/gohornet/hornet/pkg/p2p"
	"github.com/gohornet/hornet/pkg/protocol/gossip"
	"github.com/gohornet/hornet/pkg/shutdown"
	"github.com/gohornet/hornet/pkg/snapshot"
	"github.com/gohornet/hornet/pkg/tangle"
	"github.com/gohornet/hornet/pkg/tipselect"
	"github.com/iotaledger/hive.go/configuration"
)

// RouteMetrics is the route for getting the prometheus metrics.
// GET returns metrics.
const (
	RouteMetrics = "/metrics"
)

func init() {
	Plugin = &node.Plugin{
		Status: node.StatusDisabled,
		Pluggable: node.Pluggable{
			Name:      "Prometheus",
			DepsFunc:  func(cDeps dependencies) { deps = cDeps },
			Params:    params,
			Configure: configure,
			Run:       run,
		},
	}
}

var (
	Plugin *node.Plugin
	deps   dependencies

	server   *http.Server
	registry = prometheus.NewRegistry()
	collects []func()
)

type dependencies struct {
	dig.In
	AppInfo          *app.AppInfo
	NodeConfig       *configuration.Configuration `name:"nodeConfig"`
	Database         *database.Database
	DatabasePath     string `name:"databasePath"`
	Storage          *storage.Storage
	ServerMetrics    *metrics.ServerMetrics
	DatabaseMetrics  *metrics.DatabaseMetrics
	StorageMetrics   *metrics.StorageMetrics
	RestAPIMetrics   *metrics.RestAPIMetrics `optional:"true"`
	Service          *gossip.Service
	ReceiptService   *migrator.ReceiptService `optional:"true"`
	Tangle           *tangle.Tangle
	MigratorService  *migrator.MigratorService `optional:"true"`
	Manager          *p2p.Manager
	RequestQueue     gossip.RequestQueue
	MessageProcessor *gossip.MessageProcessor
	TipSelector      *tipselect.TipSelector `optional:"true"`
	Snapshot         *snapshot.Snapshot
	Coordinator      *coordinator.Coordinator `optional:"true"`
}

func configure() {
	if deps.NodeConfig.Bool(CfgPrometheusDatabase) {
		configureDatabase()
	}
	if deps.NodeConfig.Bool(CfgPrometheusNode) {
		configureNode()
	}
	if deps.NodeConfig.Bool(CfgPrometheusGossip) {
		configureGossipPeers()
		configureGossipNode()
	}
	if deps.NodeConfig.Bool(CfgPrometheusCaches) {
		configureCaches()
	}
	if deps.NodeConfig.Bool(CfgPrometheusRestAPI) && deps.RestAPIMetrics != nil {
		configureRestAPI()
	}
	if deps.NodeConfig.Bool(CfgPrometheusMigration) {
		if deps.ReceiptService != nil {
			configureReceipts()
		}
		if deps.MigratorService != nil {
			configureMigrator()
		}
	}
	if deps.NodeConfig.Bool(CfgPrometheusCoordinator) && deps.Coordinator != nil {
		configureCoordinator()
	}
	if deps.NodeConfig.Bool(CfgPrometheusDebug) {
		configureDebug()
	}
	if deps.NodeConfig.Bool(CfgPrometheusGoMetrics) {
		registry.MustRegister(collectors.NewGoCollector())
	}
	if deps.NodeConfig.Bool(CfgPrometheusProcessMetrics) {
		registry.MustRegister(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}))
	}
}

func addCollect(collect func()) {
	collects = append(collects, collect)
}

type fileservicediscovery struct {
	Targets []string          `json:"targets"`
	Labels  map[string]string `json:"labels"`
}

func writeFileServiceDiscoveryFile() {
	path := deps.NodeConfig.String(CfgPrometheusFileServiceDiscoveryPath)
	d := []fileservicediscovery{{
		Targets: []string{deps.NodeConfig.String(CfgPrometheusFileServiceDiscoveryTarget)},
		Labels:  make(map[string]string),
	}}
	j, err := json.MarshalIndent(d, "", "  ")
	if err != nil {
		Plugin.Panic("unable to marshal file service discovery JSON:", err)
		return
	}

	// this truncates an existing file
	if err := ioutil.WriteFile(path, j, 0666); err != nil {
		Plugin.Panic("unable to write file service discovery file:", err)
	}

	Plugin.LogInfof("Wrote 'file service discovery' content to %s", path)
}

func run() {
	Plugin.LogInfo("Starting Prometheus exporter ...")

	if deps.NodeConfig.Bool(CfgPrometheusFileServiceDiscoveryEnabled) {
		writeFileServiceDiscoveryFile()
	}

	if err := Plugin.Daemon().BackgroundWorker("Prometheus exporter", func(shutdownSignal <-chan struct{}) {
		Plugin.LogInfo("Starting Prometheus exporter ... done")

		e := echo.New()
		e.HideBanner = true
		e.Use(middleware.Recover())

		e.GET(RouteMetrics, func(c echo.Context) error {
			for _, collect := range collects {
				collect()
			}
			handler := promhttp.HandlerFor(
				registry,
				promhttp.HandlerOpts{
					EnableOpenMetrics: true,
				},
			)
			if deps.NodeConfig.Bool(CfgPrometheusPromhttpMetrics) {
				handler = promhttp.InstrumentMetricHandler(registry, handler)
			}

			handler.ServeHTTP(c.Response().Writer, c.Request())
			return nil
		})

		bindAddr := deps.NodeConfig.String(CfgPrometheusBindAddress)
		server = &http.Server{Addr: bindAddr, Handler: e}

		go func() {
			Plugin.LogInfof("You can now access the Prometheus exporter using: http://%s/metrics", bindAddr)
			if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
				Plugin.LogWarnf("Stopped Prometheus exporter due to an error (%s)", err)
			}
		}()

		<-shutdownSignal
		Plugin.LogInfo("Stopping Prometheus exporter ...")

		if server != nil {
			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			err := server.Shutdown(ctx)
			if err != nil {
				Plugin.LogWarn(err)
			}
			cancel()
		}
		Plugin.LogInfo("Stopping Prometheus exporter ... done")
	}, shutdown.PriorityPrometheus); err != nil {
		Plugin.Panicf("failed to start worker: %s", err)
	}
}
<file_sep>package storage

import (
	"github.com/pkg/errors"

	"github.com/gohornet/hornet/pkg/model/hornet"
	"github.com/gohornet/hornet/pkg/model/milestone"
)

var (
	ErrSolidEntryPointsAlreadyInitialized = errors.New("solidEntryPoints already initialized")
	ErrSolidEntryPointsNotInitialized     = errors.New("solidEntryPoints not initialized")
)

func (s *Storage) ReadLockSolidEntryPoints() {
	s.solidEntryPointsLock.RLock()
}

func (s *Storage) ReadUnlockSolidEntryPoints() {
	s.solidEntryPointsLock.RUnlock()
}

func (s *Storage) WriteLockSolidEntryPoints() {
	s.solidEntryPointsLock.Lock()
}

func (s *Storage) WriteUnlockSolidEntryPoints() {
	s.solidEntryPointsLock.Unlock()
}

func (s *Storage) loadSolidEntryPoints() error {
	s.WriteLockSolidEntryPoints()
	defer s.WriteUnlockSolidEntryPoints()

	if s.solidEntryPoints != nil {
		return ErrSolidEntryPointsAlreadyInitialized
	}

	points, err := s.readSolidEntryPoints()
	if err != nil {
		return err
	}

	if points == nil {
		s.solidEntryPoints = NewSolidEntryPoints()
		return nil
	}

	s.solidEntryPoints = points
	return nil
}

func (s *Storage) SolidEntryPointsContain(messageID hornet.MessageID) bool {
	s.ReadLockSolidEntryPoints()
	defer s.ReadUnlockSolidEntryPoints()

	if s.solidEntryPoints == nil {
		// this can only happen at startup of the node, no need to return an unused error all the time
		panic(ErrSolidEntryPointsNotInitialized)
	}
	return s.solidEntryPoints.Contains(messageID)
}

// SolidEntryPointsIndex returns the index of a solid entry point and whether the message is a solid entry point or not.
func (s *Storage) SolidEntryPointsIndex(messageID hornet.MessageID) (milestone.Index, bool) {
	s.ReadLockSolidEntryPoints()
	defer s.ReadUnlockSolidEntryPoints()

	if s.solidEntryPoints == nil {
		// this can only happen at startup of the node, no need to return an unused error all the time
		panic(ErrSolidEntryPointsNotInitialized)
	}
	return s.solidEntryPoints.Index(messageID)
}

// SolidEntryPointsAddWithoutLocking adds a message to the solid entry points.
// WriteLockSolidEntryPoints must be held while entering this function.
func (s *Storage) SolidEntryPointsAddWithoutLocking(messageID hornet.MessageID, milestoneIndex milestone.Index) {
	if s.solidEntryPoints == nil {
		// this can only happen at startup of the node, no need to return an unused error all the time
		panic(ErrSolidEntryPointsNotInitialized)
	}
	s.solidEntryPoints.Add(messageID, milestoneIndex)
}

// ResetSolidEntryPointsWithoutLocking resets the solid entry points.
// WriteLockSolidEntryPoints must be held while entering this function.
func (s *Storage) ResetSolidEntryPointsWithoutLocking() {
	if s.solidEntryPoints == nil {
		// this can only happen at startup of the node, no need to return an unused error all the time
		panic(ErrSolidEntryPointsNotInitialized)
	}
	s.solidEntryPoints.Clear()
}

// StoreSolidEntryPointsWithoutLocking stores the solid entry points in the persistence layer.
// WriteLockSolidEntryPoints must be held while entering this function.
func (s *Storage) StoreSolidEntryPointsWithoutLocking() error {
	if s.solidEntryPoints == nil {
		// this can only happen at startup of the node, no need to return an unused error all the time
		panic(ErrSolidEntryPointsNotInitialized)
	}
	return s.storeSolidEntryPoints(s.solidEntryPoints)
}
<file_sep>package protocfg

import (
	flag "github.com/spf13/pflag"

	"github.com/gohornet/hornet/pkg/node"
	iotago "github.com/iotaledger/iota.go/v2"
)

const (
	// the network ID on which this node operates on.
	CfgProtocolNetworkIDName = "protocol.networkID"
	// the HRP which should be used for Bech32 addresses.
	CfgProtocolBech32HRP = "protocol.bech32HRP"
	// the minimum PoW score required by the network.
	CfgProtocolMinPoWScore = "protocol.minPoWScore"
	// the amount of public keys in a milestone.
	CfgProtocolMilestonePublicKeyCount = "protocol.milestonePublicKeyCount"
	// the ed25519 public key of the coordinator in hex representation.
	CfgProtocolPublicKeyRanges = "protocol.publicKeyRanges"
	// the ed25519 public key of the coordinator in hex representation.
	CfgProtocolPublicKeyRangesJSON = "publicKeyRanges"
)

var params = &node.PluginParams{
	Params: map[string]*flag.FlagSet{
		"nodeConfig": func() *flag.FlagSet {
			fs := flag.NewFlagSet("", flag.ContinueOnError)
			fs.String(CfgProtocolNetworkIDName, "chrysalis-mainnet", "the network ID on which this node operates on.")
			fs.String(CfgProtocolBech32HRP, string(iotago.PrefixMainnet), "the HRP which should be used for Bech32 addresses.")
			fs.Float64(CfgProtocolMinPoWScore, 4000, "the minimum PoW score required by the network.")
			fs.Int(CfgProtocolMilestonePublicKeyCount, 2, "the amount of public keys in a milestone")
			return fs
		}(),
	},
	Masked: nil,
}
<file_sep>package prometheus

import (
	"github.com/prometheus/client_golang/prometheus"
)

var (
	cacheSizes *prometheus.GaugeVec
)

func configureCaches() {
	cacheSizes = prometheus.NewGaugeVec(
		prometheus.GaugeOpts{
			Namespace: "iota",
			Subsystem: "caches",
			Name:      "size",
			Help:      "Size of the cache.",
		},
		[]string{"type"},
	)

	registry.MustRegister(cacheSizes)

	addCollect(collectCaches)
}

func collectCaches() {
	cacheSizes.WithLabelValues("children").Set(float64(deps.Storage.ChildrenStorageSize()))
	cacheSizes.WithLabelValues("indexations").Set(float64(deps.Storage.IndexationStorageSize()))
	cacheSizes.WithLabelValues("messages").Set(float64(deps.Storage.MessageStorageSize()))
	cacheSizes.WithLabelValues("messages_metadata").Set(float64(deps.Storage.MessageMetadataStorageSize()))
	cacheSizes.WithLabelValues("milestones").Set(float64(deps.Storage.MilestoneStorageSize()))
	cacheSizes.WithLabelValues("unreferenced_messages").Set(float64(deps.Storage.UnreferencedMessageStorageSize()))
	cacheSizes.WithLabelValues("message_processor_work_units").Set(float64(deps.MessageProcessor.WorkUnitsSize()))
}
