func (d *PrunedBlockDispatcher) handleResp(req, resp wire.Message,
	peer string) query.Progress {
	block, ok := resp.(*wire.MsgBlock)
	if !ok {
		return query.Progress{
			Progressed: false,
			Finished:   false,
		}
	}
	getData, ok := req.(*wire.MsgGetData)
	if !ok {
		return query.Progress{
			Progressed: false,
			Finished:   false,
		}
	}
	blockHash := block.BlockHash()
	d.blockMtx.Lock()
	blockChans, ok := d.blocksQueried[blockHash]
	if !ok {
		d.blockMtx.Unlock()
		return query.Progress{
			Progressed: false,
			Finished:   false,
		}
	}
	err := blockchain.CheckBlockSanity(
		btcutil.NewBlock(block), d.cfg.ChainParams.PowLimit,
		d.timeSource,
	)
	if err != nil {
		d.blockMtx.Unlock()
		log.Warnf("Received invalid block %v from peer %v: %v",
			blockHash, peer, err)
		d.banPeer(peer)
		return query.Progress{
			Progressed: false,
			Finished:   false,
		}
	}
	delete(d.blocksQueried, blockHash)
	progress := query.Progress{Progressed: true, Finished: true}
	for _, inv := range getData.InvList {
		if _, ok := d.blocksQueried[inv.Hash]; ok {
			progress.Finished = false
			break
		}
	}
	d.blockMtx.Unlock()
	d.wg.Add(1)
	go func() {
		defer d.wg.Done()
		for _, blockChan := range blockChans {
			select {
			case blockChan <- block:
			case <-d.quit:
				return
			}
		}
	}()
	return progress
}
handleResp is a response handler that will be called for every message
received from the peer that the request was made to. It should validate the
response against the request made, and return a Progress indicating whether
the request was answered by this particular response.

NOTE: Since the worker's job queue will be stalled while this method is
running, it should not be doing any expensive operations. It should validate
the response and immediately return the progress. The response should be
handed off to another goroutine for processing.
package chain
import (
	"encoding/binary"
	"encoding/hex"
	"errors"
	"fmt"
	"math/rand"
	"net"
	"sync"
	"time"
	"github.com/btcsuite/btcd/blockchain"
	"github.com/btcsuite/btcd/btcjson"
	"github.com/btcsuite/btcd/chaincfg"
	"github.com/btcsuite/btcd/chaincfg/chainhash"
	"github.com/btcsuite/btcd/peer"
	"github.com/btcsuite/btcd/wire"
	"github.com/btcsuite/btcutil"
	"github.com/lightninglabs/neutrino/query"
	"github.com/lightningnetwork/lnd/ticker"
)
const (
	defaultRefreshPeersInterval = 30 * time.Second
	defaultPeerReadyTimeout = 15 * time.Second
	requiredServices = wire.SFNodeNetwork | wire.SFNodeWitness
	prunedNodeService wire.ServiceFlag = 1 << 10
)
type queryPeer struct {
	*peer.Peer
	ready     chan struct{}
	msgsRecvd chan wire.Message
	quit      chan struct{}
}
func (p *queryPeer) signalUponDisconnect(f func()) {
	go func() {
		p.WaitForDisconnect()
		close(p.quit)
		f()
	}()
}
func (p *queryPeer) SubscribeRecvMsg() (<-chan wire.Message, func()) {
	return p.msgsRecvd, func() {}
}
func (p *queryPeer) OnDisconnect() <-chan struct{} {
	return p.quit
}
type PrunedBlockDispatcherConfig struct {
	ChainParams *chaincfg.Params
	NumTargetPeers int
	Dial func(string) (net.Conn, error)
	GetPeers func() ([]btcjson.GetPeerInfoResult, error)
	GetNodeAddresses func(*int32) ([]btcjson.GetNodeAddressesResult, error)
	PeerReadyTimeout time.Duration
	RefreshPeersTicker ticker.Ticker
	AllowSelfPeerConns bool
	MaxRequestInvs int
}
type PrunedBlockDispatcher struct {
	cfg PrunedBlockDispatcherConfig
	workManager *query.WorkManager
	blocksQueried map[chainhash.Hash][]chan *wire.MsgBlock
	blockMtx      sync.Mutex
	currentPeers map[string]*peer.Peer
	bannedPeers map[string]struct{}
	peerMtx     sync.Mutex
	peersConnected chan query.Peer
	timeSource blockchain.MedianTimeSource
	quit chan struct{}
	wg   sync.WaitGroup
}
func NewPrunedBlockDispatcher(cfg *PrunedBlockDispatcherConfig) (
	*PrunedBlockDispatcher, error) {
	if cfg.NumTargetPeers < 1 {
		return nil, errors.New("config option NumTargetPeer must be >= 1")
	}
	if cfg.MaxRequestInvs > wire.MaxInvPerMsg {
		return nil, fmt.Errorf("config option MaxRequestInvs must be "+
			"<= %v", wire.MaxInvPerMsg)
	}
	peersConnected := make(chan query.Peer)
	return &PrunedBlockDispatcher{
		cfg: *cfg,
		workManager: query.New(&query.Config{
			ConnectedPeers: func() (<-chan query.Peer, func(), error) {
				return peersConnected, func() {}, nil
			},
			NewWorker: query.NewWorker,
			Ranking:   query.NewPeerRanking(),
		}),
		blocksQueried:  make(map[chainhash.Hash][]chan *wire.MsgBlock),
		currentPeers:   make(map[string]*peer.Peer),
		bannedPeers:    make(map[string]struct{}),
		peersConnected: peersConnected,
		timeSource:     blockchain.NewMedianTime(),
		quit:           make(chan struct{}),
	}, nil
}
func (d *PrunedBlockDispatcher) Start() error {
	log.Tracef("Starting pruned block dispatcher")
	if err := d.workManager.Start(); err != nil {
		return err
	}
	d.wg.Add(1)
	go d.pollPeers()
	return nil
}
func (d *PrunedBlockDispatcher) Stop() {
	log.Tracef("Stopping pruned block dispatcher")
	close(d.quit)
	d.wg.Wait()
	_ = d.workManager.Stop()
}
func (d *PrunedBlockDispatcher) pollPeers() {
	defer d.wg.Done()
	if err := d.connectToPeers(); err != nil {
		log.Warnf("Unable to establish peer connections: %v", err)
	}
	d.cfg.RefreshPeersTicker.Resume()
	defer d.cfg.RefreshPeersTicker.Stop()
	for {
		select {
		case <-d.cfg.RefreshPeersTicker.Ticks():
			d.peerMtx.Lock()
			peersNeeded := d.cfg.NumTargetPeers - len(d.currentPeers)
			d.peerMtx.Unlock()
			if peersNeeded <= 0 {
				continue
			}
			if err := d.connectToPeers(); err != nil {
				log.Warnf("Failed to establish peer "+
					"connections: %v", err)
				continue
			}
		case <-d.quit:
			return
		}
	}
}
func (d *PrunedBlockDispatcher) connectToPeers() error {
	peers, err := d.cfg.GetPeers()
	if err != nil {
		return err
	}
	addrs, err := filterPeers(peers)
	if err != nil {
		return err
	}
	rand.Shuffle(len(addrs), func(i, j int) {
		addrs[i], addrs[j] = addrs[j], addrs[i]
	})
	for _, addr := range addrs {
		needMore, err := d.connectToPeer(addr)
		if err != nil {
			log.Debugf("Failed connecting to peer %v: %v", addr, err)
			continue
		}
		if !needMore {
			return nil
		}
	}
	nodeAddrs, err := d.cfg.GetNodeAddresses(nil)
	if err != nil {
		return err
	}
	addrs = filterNodeAddrs(nodeAddrs)
	for _, addr := range addrs {
		if _, err := d.connectToPeer(addr); err != nil {
			log.Debugf("Failed connecting to peer %v: %v", addr, err)
		}
	}
	return nil
}
func (d *PrunedBlockDispatcher) connectToPeer(addr string) (bool, error) {
	d.peerMtx.Lock()
	_, isBanned := d.bannedPeers[addr]
	_, isConnected := d.currentPeers[addr]
	d.peerMtx.Unlock()
	if isBanned || isConnected {
		return true, nil
	}
	peer, err := d.newQueryPeer(addr)
	if err != nil {
		return true, fmt.Errorf("unable to configure query peer %v: "+
			"%v", addr, err)
	}
	conn, err := d.cfg.Dial(addr)
	if err != nil {
		return true, err
	}
	peer.AssociateConnection(conn)
	select {
	case <-peer.ready:
	case <-time.After(d.cfg.PeerReadyTimeout):
		peer.Disconnect()
		return true, errors.New("timed out waiting for protocol negotiation")
	case <-d.quit:
		return false, errors.New("shutting down")
	}
	peer.signalUponDisconnect(func() {
		d.peerMtx.Lock()
		delete(d.currentPeers, peer.Addr())
		d.peerMtx.Unlock()
	})
	d.peerMtx.Lock()
	d.currentPeers[addr] = peer.Peer
	numPeers := len(d.currentPeers)
	d.peerMtx.Unlock()
	select {
	case d.peersConnected <- peer:
	case <-d.quit:
		return false, errors.New("shutting down")
	}
	return numPeers < d.cfg.NumTargetPeers, nil
}
func filterPeers(peers []btcjson.GetPeerInfoResult) ([]string, error) {
	var eligible []string
	for _, peer := range peers {
		rawServices, err := hex.DecodeString(peer.Services)
		if err != nil {
			return nil, err
		}
		services := wire.ServiceFlag(binary.BigEndian.Uint64(rawServices))
		if !satisfiesRequiredServices(services) {
			continue
		}
		eligible = append(eligible, peer.Addr)
	}
	return eligible, nil
}
func filterNodeAddrs(nodeAddrs []btcjson.GetNodeAddressesResult) []string {
	var eligible []string
	for _, nodeAddr := range nodeAddrs {
		services := wire.ServiceFlag(nodeAddr.Services)
		if !satisfiesRequiredServices(services) {
			continue
		}
		eligible = append(eligible, nodeAddr.Address)
	}
	return eligible
}
func satisfiesRequiredServices(services wire.ServiceFlag) bool {
	return services&requiredServices == requiredServices &&
		services&prunedNodeService != prunedNodeService
}
func (d *PrunedBlockDispatcher) newQueryPeer(addr string) (*queryPeer, error) {
	ready := make(chan struct{})
	msgsRecvd := make(chan wire.Message)
	cfg := &peer.Config{
		ChainParams: d.cfg.ChainParams,
		DisableRelayTx: true,
		Listeners: peer.MessageListeners{
			OnVersion: func(p *peer.Peer, msg *wire.MsgVersion) *wire.MsgReject {
				d.timeSource.AddTimeSample(p.Addr(), msg.Timestamp)
				return nil
			},
			OnVerAck: func(*peer.Peer, *wire.MsgVerAck) {
				close(ready)
			},
			OnRead: func(p *peer.Peer, _ int, msg wire.Message, err error) {
				if err != nil {
					return
				}
				var block *wire.MsgBlock
				switch msg := msg.(type) {
				case *wire.MsgBlock:
					block = msg
				case *wire.MsgVersion, *wire.MsgVerAck,
					*wire.MsgPing, *wire.MsgPong:
					return
				default:
					log.Debugf("Received unexpected message "+
						"%T from peer %v", msg, p.Addr())
					return
				}
				select {
				case msgsRecvd <- block:
				case <-d.quit:
				}
			},
		},
		AllowSelfConns: true,
	}
	p, err := peer.NewOutboundPeer(cfg, addr)
	if err != nil {
		return nil, err
	}
	return &queryPeer{
		Peer:      p,
		ready:     ready,
		msgsRecvd: msgsRecvd,
		quit:      make(chan struct{}),
	}, nil
}
func (d *PrunedBlockDispatcher) banPeer(peer string) {
	d.peerMtx.Lock()
	defer d.peerMtx.Unlock()
	d.bannedPeers[peer] = struct{}{}
	if p, ok := d.currentPeers[peer]; ok {
		p.Disconnect()
	}
}
func (d *PrunedBlockDispatcher) Query(blocks []*chainhash.Hash,
	opts ...query.QueryOption) (<-chan *wire.MsgBlock, <-chan error) {
	reqs, blockChan, err := d.newRequest(blocks)
	if err != nil {
		errChan := make(chan error, 1)
		errChan <- err
		return nil, errChan
	}
	var errChan chan error
	if len(reqs) > 0 {
		errChan = d.workManager.Query(reqs, opts...)
	}
	return blockChan, errChan
}
func (d *PrunedBlockDispatcher) newRequest(blocks []*chainhash.Hash) (
	[]*query.Request, <-chan *wire.MsgBlock, error) {
	blockChan := make(chan *wire.MsgBlock, len(blocks))
	d.blockMtx.Lock()
	defer d.blockMtx.Unlock()
	var (
		reqs    []*query.Request
		getData *wire.MsgGetData
	)
	for i, block := range blocks {
		if getData == nil {
			getData = wire.NewMsgGetData()
		}
		if _, ok := d.blocksQueried[*block]; !ok {
			log.Debugf("Queuing new block %v for request", *block)
			inv := wire.NewInvVect(wire.InvTypeBlock, block)
			if err := getData.AddInvVect(inv); err != nil {
				return nil, nil, err
			}
		} else {
			log.Debugf("Received new request for pending query of "+
				"block %v", *block)
		}
		d.blocksQueried[*block] = append(
			d.blocksQueried[*block], blockChan,
		)
		if (len(getData.InvList) > 0 && i == len(blocks)-1) ||
			len(getData.InvList) == d.cfg.MaxRequestInvs {
			reqs = append(reqs, &query.Request{
				Req:        getData,
				HandleResp: d.handleResp,
			})
			getData = nil
		}
	}
	return reqs, blockChan, nil
}