func (h *Hub) addConnection(host string) {
	h.connectionsMtx.Lock()
	h.connections[host]++
	h.connectionsMtx.Unlock()
	atomic.AddInt32(&h.clients, 1)
}
addConnection records a new client connection for the provided host.
package pool
import (
	"context"
	"crypto/rand"
	"crypto/tls"
	"crypto/x509"
	"encoding/binary"
	"encoding/hex"
	"errors"
	"fmt"
	"math/big"
	"net/http"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"decred.org/dcrwallet/rpc/walletrpc"
	"github.com/decred/dcrd/chaincfg/chainhash"
	"github.com/decred/dcrd/chaincfg/v3"
	"github.com/decred/dcrd/dcrjson/v3"
	"github.com/decred/dcrd/dcrutil/v3"
	chainjson "github.com/decred/dcrd/rpc/jsonrpc/types/v2"
	"github.com/decred/dcrd/rpcclient/v6"
	"github.com/decred/dcrd/wire"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	errs "github.com/decred/dcrpool/errors"
)
const (
	MaxReorgLimit = 6
	getworkDataLen = (1 + ((wire.MaxBlockHeaderPayload*8 + 65) /
		(chainhash.HashBlockSize * 8))) * chainhash.HashBlockSize
	NewParent = "newparent"
	NewVotes = "newvotes"
	NewTxns = "newtxns"
)
type CacheUpdateEvent int
const (
	Confirmed CacheUpdateEvent = iota
	Unconfirmed
	ClaimedShare
	DividendsPaid
)
var (
	soloMaxGenTime = time.Second * 28
)
type WalletConnection interface {
	SignTransaction(context.Context, *walletrpc.SignTransactionRequest, ...grpc.CallOption) (*walletrpc.SignTransactionResponse, error)
	PublishTransaction(context.Context, *walletrpc.PublishTransactionRequest, ...grpc.CallOption) (*walletrpc.PublishTransactionResponse, error)
	GetTransaction(context.Context, *walletrpc.GetTransactionRequest, ...grpc.CallOption) (*walletrpc.GetTransactionResponse, error)
	Rescan(ctx context.Context, in *walletrpc.RescanRequest, opts ...grpc.CallOption) (walletrpc.WalletService_RescanClient, error)
}
type NodeConnection interface {
	GetTxOut(context.Context, *chainhash.Hash, uint32, bool) (*chainjson.GetTxOutResult, error)
	CreateRawTransaction(context.Context, []chainjson.TransactionInput, map[dcrutil.Address]dcrutil.Amount, *int64, *int64) (*wire.MsgTx, error)
	GetWorkSubmit(context.Context, string) (bool, error)
	GetWork(context.Context) (*chainjson.GetWorkResult, error)
	GetBlockVerbose(context.Context, *chainhash.Hash, bool) (*chainjson.GetBlockVerboseResult, error)
	GetBlock(context.Context, *chainhash.Hash) (*wire.MsgBlock, error)
	NotifyWork(context.Context) error
	NotifyBlocks(context.Context) error
	Shutdown()
}
type HubConfig struct {
	ActiveNet *chaincfg.Params
	DB Database
	NodeRPCConfig *rpcclient.ConnConfig
	WalletRPCCert string
	WalletTLSCert string
	WalletTLSKey string
	WalletGRPCHost string
	PoolFee float64
	MaxGenTime time.Duration
	PaymentMethod string
	LastNPeriod time.Duration
	WalletPass string
	SoloPool bool
	PoolFeeAddrs []dcrutil.Address
	AdminPass string
	NonceIterations float64
	MinerListen string
	MaxConnectionsPerHost uint32
	WalletAccount uint32
	CoinbaseConfTimeout time.Duration
	MonitorCycle time.Duration
	MaxUpgradeTries uint32
	ClientTimeout time.Duration
}
type Hub struct {
	clients int32  
	cfg            *HubConfig
	limiter        *RateLimiter
	nodeConn       NodeConnection
	walletClose    func() error
	walletConn     WalletConnection
	notifClient    walletrpc.WalletService_ConfirmationNotificationsClient
	poolDiffs      *DifficultySet
	paymentMgr     *PaymentMgr
	chainState     *ChainState
	connections    map[string]uint32
	connectionsMtx sync.RWMutex
	cancel         context.CancelFunc
	endpoint       *Endpoint
	blake256Pad    []byte
	wg             *sync.WaitGroup
	cacheCh        chan CacheUpdateEvent
}
func (h *Hub) SignalCache(event CacheUpdateEvent) {
	select {
	case h.cacheCh <- event:
	default:
	}
}
func (h *Hub) FetchCacheChannel() chan CacheUpdateEvent {
	return h.cacheCh
}
func generateBlake256Pad() []byte {
	blake256Pad := make([]byte, getworkDataLen-wire.MaxBlockHeaderPayload)
	blake256Pad[0] = 0x80
	blake256Pad[len(blake256Pad)-9] |= 0x01
	binary.BigEndian.PutUint64(blake256Pad[len(blake256Pad)-8:],
		wire.MaxBlockHeaderPayload*8)
	return blake256Pad
}
func NewHub(cancel context.CancelFunc, hcfg *HubConfig) (*Hub, error) {
	h := &Hub{
		cfg:         hcfg,
		limiter:     NewRateLimiter(),
		wg:          new(sync.WaitGroup),
		connections: make(map[string]uint32),
		cacheCh:     make(chan CacheUpdateEvent, bufferSize),
		cancel:      cancel,
	}
	h.blake256Pad = generateBlake256Pad()
	powLimit := new(big.Rat).SetInt(h.cfg.ActiveNet.PowLimit)
	maxGenTime := h.cfg.MaxGenTime
	if h.cfg.SoloPool {
		maxGenTime = soloMaxGenTime
	}
	log.Infof("Maximum work submission generation time at "+
		"pool difficulty is %s.", maxGenTime)
	h.poolDiffs = NewDifficultySet(h.cfg.ActiveNet, powLimit, maxGenTime)
	pCfg := &PaymentMgrConfig{
		db:                    h.cfg.DB,
		ActiveNet:             h.cfg.ActiveNet,
		PoolFee:               h.cfg.PoolFee,
		LastNPeriod:           h.cfg.LastNPeriod,
		SoloPool:              h.cfg.SoloPool,
		PaymentMethod:         h.cfg.PaymentMethod,
		PoolFeeAddrs:          h.cfg.PoolFeeAddrs,
		WalletAccount:         h.cfg.WalletAccount,
		WalletPass:            h.cfg.WalletPass,
		GetBlockConfirmations: h.getBlockConfirmations,
		FetchTxCreator:        func() TxCreator { return h.nodeConn },
		FetchTxBroadcaster:    func() TxBroadcaster { return h.walletConn },
		CoinbaseConfTimeout:   h.cfg.CoinbaseConfTimeout,
		SignalCache:           h.SignalCache,
		HubWg:                 h.wg,
	}
	var err error
	h.paymentMgr, err = NewPaymentMgr(pCfg)
	if err != nil {
		return nil, err
	}
	sCfg := &ChainStateConfig{
		db:                    h.cfg.DB,
		SoloPool:              h.cfg.SoloPool,
		ProcessPayments:       h.paymentMgr.processPayments,
		GeneratePayments:      h.paymentMgr.generatePayments,
		GetBlock:              h.getBlock,
		GetBlockConfirmations: h.getBlockConfirmations,
		Cancel:                h.cancel,
		SignalCache:           h.SignalCache,
		HubWg:                 h.wg,
	}
	h.chainState = NewChainState(sCfg)
	cfgMode := uint32(0)
	if h.cfg.SoloPool {
		cfgMode = 1
	}
	dbMode, err := hcfg.DB.fetchPoolMode()
	if err != nil {
		if errors.Is(err, errs.ValueNotFound) {
			err = hcfg.DB.persistPoolMode(cfgMode)
			if err != nil {
				return nil, fmt.Errorf("failed to persist pool mode: %v", err)
			}
			dbMode = cfgMode
		} else {
			return nil, err
		}
	}
	if cfgMode != dbMode {
		return nil, fmt.Errorf("database and config have differing values for "+
			"pool mode, config=%d, database=%d", cfgMode, dbMode)
	}
	if !h.cfg.SoloPool {
		log.Infof("Payment method is %s.", strings.ToUpper(hcfg.PaymentMethod))
	} else {
		log.Infof("Solo pool mode active.")
	}
	eCfg := &EndpointConfig{
		ActiveNet:             h.cfg.ActiveNet,
		db:                    h.cfg.DB,
		SoloPool:              h.cfg.SoloPool,
		Blake256Pad:           h.blake256Pad,
		NonceIterations:       h.cfg.NonceIterations,
		MaxConnectionsPerHost: h.cfg.MaxConnectionsPerHost,
		HubWg:                 h.wg,
		FetchMinerDifficulty:  h.poolDiffs.fetchMinerDifficulty,
		SubmitWork:            h.submitWork,
		FetchCurrentWork:      h.chainState.fetchCurrentWork,
		WithinLimit:           h.limiter.withinLimit,
		AddConnection:         h.addConnection,
		RemoveConnection:      h.removeConnection,
		FetchHostConnections:  h.fetchHostConnections,
		MaxGenTime:            h.cfg.MaxGenTime,
		SignalCache:           h.SignalCache,
		MonitorCycle:          h.cfg.MonitorCycle,
		MaxUpgradeTries:       h.cfg.MaxUpgradeTries,
		ClientTimeout:         h.cfg.ClientTimeout,
	}
	h.endpoint, err = NewEndpoint(eCfg, h.cfg.MinerListen)
	if err != nil {
		return nil, err
	}
	return h, nil
}
func (h *Hub) Connect(ctx context.Context) error {
	nodeConn, err := rpcclient.New(h.cfg.NodeRPCConfig,
		h.createNotificationHandlers())
	if err != nil {
		return err
	}
	if err := nodeConn.NotifyWork(ctx); err != nil {
		nodeConn.Shutdown()
		return fmt.Errorf("unable to subscribe for work "+
			"notifications: %v", err)
	}
	if err := nodeConn.NotifyBlocks(ctx); err != nil {
		nodeConn.Shutdown()
		return fmt.Errorf("unable to subscribe for block "+
			"notifications: %v", err)
	}
	h.nodeConn = nodeConn
	if !h.cfg.SoloPool {
		serverCAs := x509.NewCertPool()
		serverCert, err := os.ReadFile(h.cfg.WalletRPCCert)
		if err != nil {
			return err
		}
		if !serverCAs.AppendCertsFromPEM(serverCert) {
			return fmt.Errorf("no certificates found in %s",
				h.cfg.WalletRPCCert)
		}
		keypair, err := tls.LoadX509KeyPair(h.cfg.WalletTLSCert,
			h.cfg.WalletTLSKey)
		if err != nil {
			return fmt.Errorf("unable to read keypair: %v", err)
		}
		creds := credentials.NewTLS(&tls.Config{
			Certificates: []tls.Certificate{keypair},
			RootCAs:      serverCAs,
		})
		grpc, err := grpc.Dial(h.cfg.WalletGRPCHost,
			grpc.WithTransportCredentials(creds))
		if err != nil {
			return fmt.Errorf("unable to establish wallet "+
				"grpc connection: %v", err)
		}
		walletConn := walletrpc.NewWalletServiceClient(grpc)
		req := &walletrpc.BalanceRequest{
			AccountNumber:         h.cfg.WalletAccount,
			RequiredConfirmations: 1,
		}
		_, err = walletConn.Balance(ctx, req)
		if err != nil {
			return err
		}
		h.walletConn = walletConn
		h.walletClose = grpc.Close
		confNotifs, err := walletConn.ConfirmationNotifications(ctx)
		if err != nil {
			return fmt.Errorf("unable to create confirmation "+
				"notification client: %v", err)
		}
		h.notifClient = confNotifs
	}
	return nil
}
func (h *Hub) submitWork(ctx context.Context, data *string) (bool, error) {
	if h.nodeConn == nil {
		return false, errs.PoolError(errs.Disconnected, "node disconnected")
	}
	return h.nodeConn.GetWorkSubmit(ctx, *data)
}
func (h *Hub) getWork(ctx context.Context) (string, string, error) {
	if h.nodeConn == nil {
		return "", "", errs.PoolError(errs.Disconnected, "node disonnected")
	}
	work, err := h.nodeConn.GetWork(ctx)
	if err != nil {
		desc := fmt.Sprintf("unable to fetch current work: %v", err)
		return "", "", errs.PoolError(errs.GetWork, desc)
	}
	return work.Data, work.Target, err
}
func (h *Hub) getBlockConfirmations(ctx context.Context, hash *chainhash.Hash) (int64, error) {
	info, err := h.nodeConn.GetBlockVerbose(ctx, hash, false)
	if err != nil {
		desc := fmt.Sprintf("unable to fetch block confirmations: %v", err)
		var rErr *dcrjson.RPCError
		if errors.As(err, &rErr) {
			if rErr.Code == dcrjson.ErrRPCBlockNotFound {
				return 0, errs.PoolError(errs.BlockNotFound, desc)
			}
		}
		return 0, errs.PoolError(errs.BlockConf, desc)
	}
	return info.Confirmations, nil
}
func (h *Hub) WithinLimit(ip string, clientType int) bool {
	return h.limiter.withinLimit(ip, clientType)
}
func (h *Hub) FetchLastWorkHeight() uint32 {
	return h.chainState.fetchLastWorkHeight()
}
func (h *Hub) FetchLastPaymentInfo() (uint32, int64, int64, error) {
	height, paidOn, err := h.cfg.DB.loadLastPaymentInfo()
	if err != nil {
		return 0, 0, 0, nil
	}
	createdOn, err := h.cfg.DB.loadLastPaymentCreatedOn()
	if err != nil {
		return 0, 0, 0, nil
	}
	return height, paidOn, createdOn, nil
}
func (h *Hub) getBlock(ctx context.Context, blockHash *chainhash.Hash) (*wire.MsgBlock, error) {
	if h.nodeConn == nil {
		return nil, errs.PoolError(errs.Disconnected, "node disconnected")
	}
	block, err := h.nodeConn.GetBlock(ctx, blockHash)
	if err != nil {
		desc := fmt.Sprintf("unable to fetch block %s: %v",
			blockHash.String(), err)
		return nil, errs.PoolError(errs.GetBlock, desc)
	}
	return block, nil
}
func (h *Hub) fetchHostConnections(host string) uint32 {
	h.connectionsMtx.RLock()
	defer h.connectionsMtx.RUnlock()
	return h.connections[host]
}