func (l *Logic) receiveLayerContent(ctx context.Context, layerID types.LayerID, peer p2p.Peer, expectedResults int, data []byte, peerErr error) {
	l.log.WithContext(ctx).With().Debug("received layer content from peer", log.String("peer", peer.String()))
	peerRes := extractPeerResult(l.log.WithContext(ctx), layerID, data, peerErr)
	if peerRes.err == nil {
		if err := l.fetchLayerBlocks(ctx, layerID, peerRes.data); err != nil {
			peerRes.err = ErrBlockNotFetched
		}
	}
	l.mutex.Lock()
	defer l.mutex.Unlock()
	result := l.layerBlocksRes[layerID]
	result.responses[peer] = &peerRes
	if len(result.responses) < expectedResults {
		l.log.WithContext(ctx).With().Debug("not yet received layer blocks from all peers",
			log.Int("received", len(result.responses)),
			log.Int("expected", expectedResults))
		return
	}
	go notifyLayerBlocksResult(ctx, layerID, l.layerDB, l.layerBlocksChs[layerID], result, l.log.WithContext(ctx).WithFields(layerID))
	delete(l.layerBlocksChs, layerID)
	delete(l.layerBlocksRes, layerID)
}
receiveLayerContent is called when response of block IDs for a layer hash is received from remote peer.
if enough responses are received, it notifies the channels waiting for the layer blocks result.
package layerfetcher
import (
	"context"
	"errors"
	"fmt"
	"sync"
	"github.com/spacemeshos/go-spacemesh/common/types"
	"github.com/spacemeshos/go-spacemesh/common/util"
	"github.com/spacemeshos/go-spacemesh/database"
	"github.com/spacemeshos/go-spacemesh/fetch"
	"github.com/spacemeshos/go-spacemesh/log"
	"github.com/spacemeshos/go-spacemesh/p2p"
	"github.com/spacemeshos/go-spacemesh/p2p/bootstrap"
	"github.com/spacemeshos/go-spacemesh/p2p/server"
)
type atxHandler interface {
	HandleAtxData(context.Context, []byte) error
}
type blockHandler interface {
	HandleBlockData(context.Context, []byte) error
}
type TxProcessor interface {
	HandleTxSyncData(data []byte) error
}
type layerDB interface {
	GetLayerHash(types.LayerID) types.Hash32
	GetAggregatedLayerHash(types.LayerID) types.Hash32
	LayerBlockIds(types.LayerID) ([]types.BlockID, error)
	GetLayerInputVectorByID(types.LayerID) ([]types.BlockID, error)
	SaveLayerInputVectorByID(context.Context, types.LayerID, []types.BlockID) error
	ProcessedLayer() types.LayerID
	SetZeroBlockLayer(types.LayerID) error
}
type atxIDsDB interface {
	GetEpochAtxs(epochID types.EpochID) ([]types.ATXID, error)
}
type poetDB interface {
	HasProof(proofRef []byte) bool
	ValidateAndStore(proofMessage *types.PoetProofMessage) error
	ValidateAndStoreMsg(data []byte) error
}
type network interface {
	bootstrap.Provider
	server.Host
}
var (
	ErrNoPeers = errors.New("no peers")
	ErrInternal = errors.New("unspecified error returned by peer")
	ErrBlockNotFetched = errors.New("block not fetched")
	errLayerNotProcessed = errors.New("requested layer is not yet processed")
)
type peerResult struct {
	data *layerBlocks
	err  error
}
type layerResult struct {
	layerID     types.LayerID
	blocks      map[types.BlockID]struct{}
	inputVector []types.BlockID
	responses   map[p2p.Peer]*peerResult
}
type LayerPromiseResult struct {
	Err   error
	Layer types.LayerID
}
type Logic struct {
	log              log.Log
	fetcher          fetch.Fetcher
	atxsrv, blocksrv server.Requestor
	host             network
	mutex            sync.Mutex
	layerBlocksRes   map[types.LayerID]*layerResult
	layerBlocksChs   map[types.LayerID][]chan LayerPromiseResult
	poetProofs       poetDB
	atxs             atxHandler
	blockHandler     blockHandler
	txs              TxProcessor
	layerDB          layerDB
	atxIds           atxIDsDB
	goldenATXID      types.ATXID
}
type Config struct {
	RequestTimeout int
	GoldenATXID    types.ATXID
}
func DefaultConfig() Config {
	return Config{RequestTimeout: 10}
}
const (
	blockProtocol = "/block/1.0.0"
	atxProtocol   = "/atx/1.0.0"
)
func NewLogic(ctx context.Context, cfg Config, blocks blockHandler, atxs atxHandler, poet poetDB, atxIDs atxIDsDB, txs TxProcessor,
	host network, fetcher fetch.Fetcher, layers layerDB, log log.Log) *Logic {
	l := &Logic{
		log:            log,
		fetcher:        fetcher,
		host:           host,
		layerBlocksRes: make(map[types.LayerID]*layerResult),
		layerBlocksChs: make(map[types.LayerID][]chan LayerPromiseResult),
		poetProofs:     poet,
		atxs:           atxs,
		layerDB:        layers,
		blockHandler:   blocks,
		atxIds:         atxIDs,
		txs:            txs,
		goldenATXID:    cfg.GoldenATXID,
	}
	l.atxsrv = server.New(host, atxProtocol, l.epochATXsReqReceiver, server.WithLog(log))
	l.blocksrv = server.New(host, blockProtocol, l.layerBlocksReqReceiver, server.WithLog(log))
	return l
}
func (l *Logic) Start() {
	l.fetcher.Start()
}
func (l *Logic) Close() {
	l.fetcher.Stop()
}
func (l *Logic) AddDBs(blockDB, AtxDB, TxDB, poetDB database.Getter) {
	l.fetcher.AddDB(fetch.BlockDB, blockDB)
	l.fetcher.AddDB(fetch.ATXDB, AtxDB)
	l.fetcher.AddDB(fetch.TXDB, TxDB)
	l.fetcher.AddDB(fetch.POETDB, poetDB)
}
func (l *Logic) epochATXsReqReceiver(ctx context.Context, msg []byte) ([]byte, error) {
	epoch := types.EpochID(util.BytesToUint32(msg))
	atxs, err := l.atxIds.GetEpochAtxs(epoch)
	if err != nil {
		return nil, fmt.Errorf("get epoch ATXs: %w", err)
	}
	l.log.WithContext(ctx).With().Debug("responded to epoch atxs request",
		epoch,
		log.Int("count", len(atxs)))
	bts, err := types.InterfaceToBytes(atxs)
	if err != nil {
		l.log.WithContext(ctx).With().Panic("failed to serialize epoch atxs", epoch, log.Err(err))
		return bts, fmt.Errorf("serialize: %w", err)
	}
	return bts, nil
}
func (l *Logic) layerBlocksReqReceiver(ctx context.Context, req []byte) ([]byte, error) {
	lyrID := types.BytesToLayerID(req)
	processed := l.layerDB.ProcessedLayer()
	if lyrID.After(processed) {
		return nil, fmt.Errorf("%w: requested layer %v is higher than processed %v", errLayerNotProcessed, lyrID, processed)
	}
	b := &layerBlocks{
		ProcessedLayer: processed,
		Hash:           l.layerDB.GetLayerHash(lyrID),
		AggregatedHash: l.layerDB.GetAggregatedLayerHash(lyrID),
	}
	var err error
	b.Blocks, err = l.layerDB.LayerBlockIds(lyrID)
	if err != nil {
		l.log.WithContext(ctx).With().Warning("failed to get layer content", lyrID, log.Err(err))
		return nil, ErrInternal
	}
	if b.InputVector, err = l.layerDB.GetLayerInputVectorByID(lyrID); err != nil {
		l.log.WithContext(ctx).With().Warning("failed to get input vector for layer", lyrID, log.Err(err))
		return nil, ErrInternal
	}
	out, err := types.InterfaceToBytes(b)
	if err != nil {
		l.log.WithContext(ctx).With().Panic("failed to serialize layer blocks response", log.Err(err))
	}
	return out, nil
}
func (l *Logic) initLayerPolling(layerID types.LayerID, ch chan LayerPromiseResult) bool {
	l.mutex.Lock()
	defer l.mutex.Unlock()
	l.layerBlocksChs[layerID] = append(l.layerBlocksChs[layerID], ch)
	if _, ok := l.layerBlocksRes[layerID]; ok {
		return false
	}
	l.layerBlocksRes[layerID] = &layerResult{
		layerID:   layerID,
		blocks:    make(map[types.BlockID]struct{}),
		responses: make(map[p2p.Peer]*peerResult),
	}
	return true
}
func (l *Logic) PollLayerContent(ctx context.Context, layerID types.LayerID) chan LayerPromiseResult {
	resChannel := make(chan LayerPromiseResult, 1)
	remotePeers := l.host.GetPeers()
	numPeers := len(remotePeers)
	if numPeers == 0 {
		resChannel <- LayerPromiseResult{Layer: layerID, Err: ErrNoPeers}
		return resChannel
	}
	if !l.initLayerPolling(layerID, resChannel) {
		return resChannel
	}
	for _, p := range remotePeers {
		peer := p
		receiveForPeerFunc := func(data []byte) {
			l.receiveLayerContent(ctx, layerID, peer, numPeers, data, nil)
		}
		errFunc := func(err error) {
			l.receiveLayerContent(ctx, layerID, peer, numPeers, nil, err)
		}
		if err := l.blocksrv.Request(ctx, peer, layerID.Bytes(), receiveForPeerFunc, errFunc); err != nil {
			errFunc(err)
		}
	}
	return resChannel
}
func (l *Logic) fetchLayerBlocks(ctx context.Context, layerID types.LayerID, blocks *layerBlocks) error {
	logger := l.log.WithContext(ctx).WithFields(layerID, log.Int("num_blocks", len(blocks.Blocks)))
	l.mutex.Lock()
	lyrResult := l.layerBlocksRes[layerID]
	var toFetch []types.BlockID
	for _, blkID := range blocks.Blocks {
		if _, ok := lyrResult.blocks[blkID]; !ok {
			lyrResult.blocks[blkID] = struct{}{}
			toFetch = append(toFetch, blkID)
		}
	}
	if len(blocks.InputVector) > len(lyrResult.inputVector) {
		lyrResult.inputVector = blocks.InputVector
	}
	l.mutex.Unlock()
	logger.With().Debug("fetching new blocks", log.Int("to_fetch", len(toFetch)))
	if err := l.GetBlocks(ctx, toFetch); err != nil {
		return err
	}
	return nil
}
func extractPeerResult(logger log.Log, layerID types.LayerID, data []byte, peerErr error) (result peerResult) {
	result.err = peerErr
	if peerErr != nil {
		logger.With().Debug("received peer error for layer content", layerID, log.Err(peerErr))
		return
	}
	var blocks layerBlocks
	if err := types.BytesToInterface(data, &blocks); err != nil {
		logger.With().Debug("error converting bytes to layerBlocks", log.Err(err))
		result.err = err
		return
	}
	result.data = &blocks
	return
}