feat:add offer handle method

Signed-off-by: Chen Kai <281165273grape@gmail.com>

go.mod

@@ -134,6 +134,7 @@ require (
 	github.com/prometheus/client_model v0.2.1-0.20210607210712-147c58e9608a // indirect
 	github.com/prometheus/common v0.32.1 // indirect
 	github.com/prometheus/procfs v0.7.3 // indirect
+	github.com/prysmaticlabs/go-bitfield v0.0.0-20210809151128-385d8c5e3fb7 // indirect
 	github.com/rivo/uniseg v0.2.0 // indirect
 	github.com/rogpeppe/go-internal v1.9.0 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect

go.sum

@@ -539,6 +539,8 @@ github.com/prometheus/procfs v0.7.3 h1:4jVXhlkAyzOScmCkXBTOLRLTz8EeU+eyjrwB/EPq0
 github.com/prometheus/procfs v0.7.3/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
 github.com/protolambda/bls12-381-util v0.0.0-20220416220906-d8552aa452c7 h1:cZC+usqsYgHtlBaGulVnZ1hfKAi8iWtujBnRLQE698c=
 github.com/protolambda/bls12-381-util v0.0.0-20220416220906-d8552aa452c7/go.mod h1:IToEjHuttnUzwZI5KBSM/LOOW3qLbbrHOEfp3SbECGY=
+github.com/prysmaticlabs/go-bitfield v0.0.0-20210809151128-385d8c5e3fb7 h1:0tVE4tdWQK9ZpYygoV7+vS6QkDvQVySboMVEIxBJmXw=
+github.com/prysmaticlabs/go-bitfield v0.0.0-20210809151128-385d8c5e3fb7/go.mod h1:wmuf/mdK4VMD+jA9ThwcUKjg3a2XWM9cVfFYjDyY4j4=
 github.com/rivo/uniseg v0.2.0 h1:S1pD9weZBuJdFmowNwbpi7BJ8TNftyUImj/0WQi72jY=
 github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=

p2p/discover/portal_protocol.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"go.uber.org/zap"
 	"io"
+	"math/big"
 	"net"
 	"sort"
 	"time"
@@ -23,6 +24,7 @@ import (
 	ssz "github.com/ferranbt/fastssz"
 	"github.com/holiman/uint256"
 	"github.com/optimism-java/utp-go"
+	"github.com/prysmaticlabs/go-bitfield"
 )
 
 const (
@@ -538,6 +540,22 @@ func (p *PortalProtocol) handleTalkRequest(id enode.ID, addr *net.UDPAddr, msg [
 			return nil
 		}
 
+		return resp
+	case portalwire.OFFER:
+		offerRequest := &portalwire.Offer{}
+		err := offerRequest.UnmarshalSSZ(msg[1:])
+		if err != nil {
+			p.log.Error("failed to unmarshal offer request", "err", err)
+			return nil
+		}
+
+		p.log.Trace("received offer request", "protocol", p.protocolId, "source", id, "offerRequest", offerRequest)
+		resp, err := p.handleOffer(id, addr, offerRequest)
+		if err != nil {
+			p.log.Error("failed to handle offer request", "err", err)
+			return nil
+		}
+
 		return resp
 	}
 
@@ -747,6 +765,48 @@ func (p *PortalProtocol) handleFindContent(id enode.ID, addr *net.UDPAddr, reque
 	}
 }
 
+func (p *PortalProtocol) handleOffer(id enode.ID, addr *net.UDPAddr, request *portalwire.Offer) ([]byte, error) {
+	contentKeyBitsets := bitfield.NewBitlist(uint64(len(request.ContentKeys)))
+	contentKeys := make([][]byte, 0)
+	for i, contentKey := range request.ContentKeys {
+		contentId := p.storage.ContentId(contentKey)
+		if contentId != nil {
+			if p.inRange(p.Self().ID(), p.nodeRadius, contentId) {
+				if _, err := p.storage.Get(contentKey, contentId); err != nil {
+					contentKeyBitsets.SetBitAt(uint64(i), true)
+					contentKeys = append(contentKeys, contentKey)
+				}
+			}
+		} else {
+			return nil, nil
+		}
+	}
+
+	if contentKeyBitsets.Count() == 0 {
+		idBuffer := make([]byte, 2)
+		binary.BigEndian.PutUint16(idBuffer, uint16(0))
+		acceptMsg := &portalwire.Accept{
+			ConnectionId: idBuffer,
+			ContentKeys:  contentKeyBitsets.BytesNoTrim(),
+		}
+
+		p.log.Trace("Sending accept response", "protocol", p.protocolId, "source", addr, "accept", acceptMsg)
+		var acceptMsgBytes []byte
+		acceptMsgBytes, err := acceptMsg.MarshalSSZ()
+		if err != nil {
+			return nil, err
+		}
+
+		talkRespBytes := make([]byte, 0, len(acceptMsgBytes)+1)
+		talkRespBytes = append(talkRespBytes, portalwire.ACCEPT)
+		talkRespBytes = append(talkRespBytes, acceptMsgBytes...)
+
+		return talkRespBytes, nil
+	}
+
+	return nil, nil
+}
+
 func (p *PortalProtocol) Self() *enode.Node {
 	return p.DiscV5.LocalNode().Node()
 }
@@ -875,3 +935,9 @@ func (p *PortalProtocol) findNodesCloseToContent(contentId []byte) []*enode.Node
 
 	return allNodes
 }
+
+func (p *PortalProtocol) inRange(nodeId enode.ID, nodeRadius *uint256.Int, contentId []byte) bool {
+	distance := enode.LogDist(nodeId, enode.ID(contentId))
+	disBig := new(big.Int).SetInt64(int64(distance))
+	return nodeRadius.CmpBig(disBig) > 0
+}

p2p/discover/portalwire/messages.go

@@ -28,12 +28,24 @@ const (
 	ACCEPT      byte = 0x07
 )
 
+// Content selectors for the portal protocol.
 const (
 	ContentConnIdSelector byte = 0x00
 	ContentRawSelector    byte = 0x01
 	ContentEnrsSelector   byte = 0x02
 )
 
+// Offer request types for the portal protocol.
+const (
+	OfferRequestDirect   byte = 0x00
+	OfferRequestDatabase byte = 0x01
+)
+
+type ContentKV struct {
+	ContentKey []byte
+	Content    []byte
+}
+
 // Request messages for the portal protocol.
 type (
 	PingPongCustomData struct {