libs5-go/service/default/p2p.go

package _default

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"git.lumeweb.com/LumeWeb/libs5-go/ed25519"
	"git.lumeweb.com/LumeWeb/libs5-go/encoding"
	"git.lumeweb.com/LumeWeb/libs5-go/net"
	"git.lumeweb.com/LumeWeb/libs5-go/protocol"
	"git.lumeweb.com/LumeWeb/libs5-go/service"
	"git.lumeweb.com/LumeWeb/libs5-go/structs"
	"git.lumeweb.com/LumeWeb/libs5-go/types"
	"git.lumeweb.com/LumeWeb/libs5-go/utils"
	"github.com/vmihailenco/msgpack/v5"
	bolt "go.etcd.io/bbolt"
	"go.uber.org/zap"
	"net/url"
	"sort"
	"sync"
	"time"
)

var _ service.P2PService = (*P2PServiceDefault)(nil)

var (
	errUnsupportedProtocol       = errors.New("unsupported protocol")
	errConnectionIdMissingNodeID = errors.New("connection id missing node id")
)

const nodeBucketName = "nodes"

type P2PServiceDefault struct {
	nodeKeyPair             *ed25519.KeyPairEd25519
	localNodeID             *encoding.NodeId
	networkID               string
	nodesBucket             *bolt.Bucket
	inited                  bool
	reconnectDelay          structs.Map
	peers                   structs.Map
	peersPending            structs.Map
	selfConnectionUris      []*url.URL
	outgoingPeerBlocklist   structs.Map
	incomingPeerBlockList   structs.Map
	incomingIPBlocklist     structs.Map
	outgoingPeerFailures    structs.Map
	maxOutgoingPeerFailures uint
	connections             sync.WaitGroup
	hashQueryRoutingTable   structs.Map
	service.ServiceBase
}

func NewP2P(params service.ServiceParams) *P2PServiceDefault {
	uri, err := url.Parse(fmt.Sprintf("wss://%s:%d/s5/p2p", params.Config.HTTP.API.Domain, params.Config.HTTP.API.Port))
	if err != nil {
		params.Logger.Fatal("failed to parse HTTP API URL", zap.Error(err))
	}

	service := &P2PServiceDefault{
		nodeKeyPair:             params.Config.KeyPair,
		networkID:               params.Config.P2P.Network,
		inited:                  false,
		reconnectDelay:          structs.NewMap(),
		peers:                   structs.NewMap(),
		peersPending:            structs.NewMap(),
		selfConnectionUris:      []*url.URL{uri},
		outgoingPeerBlocklist:   structs.NewMap(),
		incomingPeerBlockList:   structs.NewMap(),
		incomingIPBlocklist:     structs.NewMap(),
		outgoingPeerFailures:    structs.NewMap(),
		maxOutgoingPeerFailures: params.Config.P2P.MaxOutgoingPeerFailures,
		hashQueryRoutingTable:   structs.NewMap(),
		ServiceBase:             service.NewServiceBase(params.Logger, params.Config, params.Db),
	}

	return service
}

func (p *P2PServiceDefault) SelfConnectionUris() []*url.URL {
	return p.selfConnectionUris
}

func (p *P2PServiceDefault) Peers() structs.Map {
	return p.peers
}

func (p *P2PServiceDefault) Start(ctx context.Context) error {
	config := p.Config()
	if len(config.P2P.Peers.Initial) > 0 {
		initialPeers := config.P2P.Peers.Initial

		for _, peer := range initialPeers {
			u, err := url.Parse(peer)
			if err != nil {
				return err
			}

			peer := peer
			go func() {
				err := p.ConnectToNode([]*url.URL{u}, false, nil)
				if err != nil {
					p.Logger().Error("failed to connect to initial peer", zap.Error(err), zap.String("peer", peer))
				}
			}()
		}
	}

	return nil
}

func (p *P2PServiceDefault) Stop(ctx context.Context) error {
	return nil
}

func (p *P2PServiceDefault) Init(ctx context.Context) error {
	if p.inited {
		return nil
	}
	p.localNodeID = encoding.NewNodeId(p.nodeKeyPair.PublicKey())

	err := utils.CreateBucket(nodeBucketName, p.Db())

	if err != nil {
		return err
	}

	p.inited = true

	return nil
}
func (p *P2PServiceDefault) ConnectToNode(connectionUris []*url.URL, retried bool, fromPeer net.Peer) error {
	if !p.Services().IsStarted() {
		return nil
	}

	unsupported, _ := url.Parse("http://0.0.0.0")
	unsupported.Scheme = "unsupported"

	var connectionUri *url.URL

	for _, uri := range connectionUris {
		if uri.Scheme == "ws" || uri.Scheme == "wss" {
			connectionUri = uri
			break
		}
	}

	if connectionUri == nil {
		for _, uri := range connectionUris {
			if uri.Scheme == "tcp" {
				connectionUri = uri
				break
			}
		}
	}

	if connectionUri == nil {
		connectionUri = unsupported
	}

	if connectionUri.Scheme == "unsupported" {
		return errUnsupportedProtocol
	}

	scheme := connectionUri.Scheme

	if connectionUri.User == nil {
		return errConnectionIdMissingNodeID
	}

	username := connectionUri.User.Username()
	id, err := encoding.DecodeNodeId(username)
	if err != nil {
		return err
	}

	idString, err := id.ToString()
	if err != nil {
		return err
	}

	if p.peersPending.Contains(idString) || p.peers.Contains(idString) {
		p.Logger().Debug("already connected", zap.String("node", connectionUri.String()))
		return nil
	}

	if p.outgoingPeerBlocklist.Contains(idString) {
		p.Logger().Debug("outgoing peer is on blocklist", zap.String("node", connectionUri.String()))

		var fromPeerId string

		if fromPeer != nil {
			blocked := false
			if fromPeer.Id() != nil {
				fromPeerId, err = fromPeer.Id().ToString()
				if err != nil {
					return err
				}
				if !p.incomingPeerBlockList.Contains(fromPeerId) {
					p.incomingPeerBlockList.Put(fromPeerId, true)
					blocked = true
				}
			}

			fromPeerIP := fromPeer.GetIP()

			if !p.incomingIPBlocklist.Contains(fromPeerIP) {
				p.incomingIPBlocklist.Put(fromPeerIP, true)
				blocked = true
			}
			err = fromPeer.EndForAbuse()
			if err != nil {
				return err
			}

			if blocked {
				p.Logger().Debug("blocking peer for sending peer on blocklist", zap.String("node", connectionUri.String()), zap.String("peer", fromPeerId), zap.String("ip", fromPeerIP))
			}
		}
		return nil
	}

	reconnectDelay := p.reconnectDelay.GetUInt(idString)
	if reconnectDelay == nil {
		delay := uint(1)
		reconnectDelay = &delay
	}

	if id.Equals(p.localNodeID) {
		return nil
	}

	p.Logger().Debug("connect", zap.String("node", connectionUri.String()))

	socket, err := net.CreateTransportSocket(scheme, connectionUri)
	if err != nil {
		if retried {
			p.Logger().Error("failed to connect, too many retries", zap.String("node", connectionUri.String()), zap.Error(err))
			counter := uint(0)
			if p.outgoingPeerFailures.Contains(idString) {
				tmp := *p.outgoingPeerFailures.GetUInt(idString)
				counter = tmp
			}

			counter++

			p.outgoingPeerFailures.PutUInt(idString, counter)

			if counter >= p.maxOutgoingPeerFailures {

				if fromPeer != nil {
					blocked := false
					var fromPeerId string
					if fromPeer.Id() != nil {
						fromPeerId, err = fromPeer.Id().ToString()
						if err != nil {
							return err
						}
						if !p.incomingPeerBlockList.Contains(fromPeerId) {
							p.incomingPeerBlockList.Put(fromPeerId, true)
							blocked = true
						}
					}

					fromPeerIP := fromPeer.GetIP()
					if !p.incomingIPBlocklist.Contains(fromPeerIP) {
						p.incomingIPBlocklist.Put(fromPeerIP, true)
						blocked = true
					}
					err = fromPeer.EndForAbuse()
					if err != nil {
						return err
					}

					if blocked {
						p.Logger().Debug("blocking peer for sending peer on blocklist", zap.String("node", connectionUri.String()), zap.String("peer", fromPeerId), zap.String("ip", fromPeerIP))
					}
				}
				p.outgoingPeerBlocklist.Put(idString, true)
				p.Logger().Debug("blocking peer for too many failures", zap.String("node", connectionUri.String()))
			}

			return nil
		}
		retried = true

		p.Logger().Error("failed to connect", zap.String("node", connectionUri.String()), zap.Error(err))

		delay := p.reconnectDelay.GetUInt(idString)
		if delay == nil {
			tmp := uint(1)
			delay = &tmp
		}
		delayDeref := *delay
		p.reconnectDelay.PutUInt(idString, delayDeref*2)

		time.Sleep(time.Duration(delayDeref) * time.Second)

		return p.ConnectToNode(connectionUris, retried, fromPeer)
	}

	if p.outgoingPeerFailures.Contains(idString) {
		p.outgoingPeerFailures.Remove(idString)
	}

	peer, err := net.CreateTransportPeer(scheme, &net.TransportPeerConfig{
		Socket: socket,
		Uris:   []*url.URL{connectionUri},
	})

	if err != nil {
		return err
	}

	peer.SetId(id)

	p.Services().P2P().ConnectionTracker().Add(1)

	peerId, err := peer.Id().ToString()
	if err != nil {
		return err
	}
	p.peersPending.Put(peerId, peer)

	go func() {
		err := p.OnNewPeer(peer, true)
		if err != nil && !peer.Abuser() {
			p.Logger().Error("peer error", zap.Error(err))
		}
		p.Services().P2P().ConnectionTracker().Done()
	}()

	return nil

}

func (p *P2PServiceDefault) OnNewPeer(peer net.Peer, verifyId bool) error {
	var wg sync.WaitGroup

	var pid string

	if peer.Id() != nil {
		pid, _ = peer.Id().ToString()
	} else {
		pid = "unknown"
	}

	pip := peer.GetIP()

	if p.incomingIPBlocklist.Contains(pid) {
		p.Logger().Error("peer is on identity blocklist", zap.String("peer", pid))
		err := peer.EndForAbuse()
		if err != nil {
			return err
		}
		return nil
	}
	if p.incomingPeerBlockList.Contains(pip) {
		p.Logger().Debug("peer is on ip blocklist", zap.String("peer", pid), zap.String("ip", pip))
		err := peer.EndForAbuse()
		if err != nil {
			return err
		}
		return nil
	}

	p.Logger().Debug("OnNewPeer started", zap.String("peer", pid))

	challenge := protocol.GenerateChallenge()
	peer.SetChallenge(challenge)

	wg.Add(1)
	go func() {
		defer wg.Done()
		p.OnNewPeerListen(peer, verifyId)
	}()

	handshakeOpenMsg, err := msgpack.Marshal(protocol.NewHandshakeOpen(challenge, p.networkID))
	if err != nil {
		return err
	}

	err = peer.SendMessage(handshakeOpenMsg)
	if err != nil {
		return err
	}
	p.Logger().Debug("OnNewPeer sent handshake", zap.String("peer", pid))

	p.Logger().Debug("OnNewPeer before Wait", zap.String("peer", pid))
	wg.Wait() // Wait for OnNewPeerListen goroutine to finish
	p.Logger().Debug("OnNewPeer ended", zap.String("peer", pid))
	return nil
}
func (p *P2PServiceDefault) OnNewPeerListen(peer net.Peer, verifyId bool) {
	onDone := net.CloseCallback(func() {
		if peer.Id() != nil {
			pid, err := peer.Id().ToString()
			if err != nil {
				p.Logger().Error("failed to get peer id", zap.Error(err))
				return
			}
			// Handle closure of the connection
			if p.peers.Contains(pid) {
				p.peers.Remove(pid)
			}
			if p.peersPending.Contains(pid) {
				p.peersPending.Remove(pid)
			}
		}
	})

	onError := net.ErrorCallback(func(args ...interface{}) {
		if !peer.Abuser() {
			p.Logger().Error("peer error", zap.Any("args", args))
		}
	})

	peer.ListenForMessages(func(message []byte) error {
		var reader protocol.IncomingMessageReader

		err := msgpack.Unmarshal(message, &reader)
		if err != nil {
			p.Logger().Error("Error decoding basic message info", zap.Error(err))
			return err
		}

		// Now, get the specific message handler based on the message kind
		handler, ok := protocol.GetMessageType(reader.Kind)
		if !ok {
			p.Logger().Error("Unknown message type", zap.Int("type", reader.Kind))
			return fmt.Errorf("unknown message type: %d", reader.Kind)
		}

		data := protocol.IncomingMessageData{
			Original: message,
			Data:     reader.Data,
			Ctx:      context.Background(),
			Peer:     peer,
			VerifyId: verifyId,
			Config:   p.Config(),
			Logger:   p.Logger(),
			Mediator: NewMediator(service.ServiceParams{
				Logger: p.Logger(),
				Config: p.Config(),
				Db:     p.Db(),
			}),
		}

		if mediator, ok := data.Mediator.(service.ServicesSetter); ok {
			mediator.SetServices(p.Services())
		} else {
			p.Logger().Fatal("failed to cast mediator to service.Service")
		}

		dec := msgpack.NewDecoder(bytes.NewReader(reader.Data))

		err = handler.DecodeMessage(dec, data)
		if err != nil {
			p.Logger().Error("Error decoding message", zap.Error(err))
			return err
		}

		// Directly decode and handle the specific message type
		if err = handler.HandleMessage(data); err != nil {
			p.Logger().Error("Error handling message", zap.Error(err))
			return err
		}

		return nil
	}, net.ListenerOptions{
		OnClose: &onDone,
		OnError: &onError,
		Logger:  p.Logger(),
	})
}

func (p *P2PServiceDefault) readNodeVotes(nodeId *encoding.NodeId) (service.NodeVotes, error) {
	var value []byte
	var found bool
	err := p.Db().View(func(tx *bolt.Tx) error {
		b := tx.Bucket([]byte(nodeBucketName))
		if b == nil {
			return fmt.Errorf("Bucket %s not found", nodeBucketName)
		}
		value = b.Get(nodeId.Raw())
		if value == nil {
			return nil
		}
		found = true
		return nil
	})
	if err != nil {
		return nil, err
	}
	if !found {
		return service.NewNodeVotes(), nil
	}

	var score service.NodeVotes
	err = msgpack.Unmarshal(value, &score)
	if err != nil {
		return nil, err
	}

	return score, nil
}

func (p *P2PServiceDefault) saveNodeVotes(nodeId *encoding.NodeId, votes service.NodeVotes) error {
	// Marshal the votes into data
	data, err := msgpack.Marshal(votes)
	if err != nil {
		return err
	}

	// Use a transaction to save the data
	err = p.Db().Update(func(tx *bolt.Tx) error {
		// Get or create the bucket
		b := tx.Bucket([]byte(nodeBucketName))

		// Put the data into the bucket
		return b.Put(nodeId.Raw(), data)
	})

	return err
}

func (p *P2PServiceDefault) GetNodeScore(nodeId *encoding.NodeId) (float64, error) {
	if nodeId.Equals(p.localNodeID) {
		return 1, nil
	}

	score, err := p.readNodeVotes(nodeId)
	if err != nil {
		return 0.5, err
	}

	return protocol.CalculateNodeScore(score.Good(), score.Bad()), nil

}
func (p *P2PServiceDefault) SortNodesByScore(nodes []*encoding.NodeId) ([]*encoding.NodeId, error) {
	scores := make(map[encoding.NodeIdCode]float64)
	var errOccurred error

	for _, nodeId := range nodes {
		score, err := p.GetNodeScore(nodeId)
		if err != nil {
			errOccurred = err
			scores[nodeId.HashCode()] = 0 // You may choose a different default value for error cases
		} else {
			scores[nodeId.HashCode()] = score
		}
	}

	sort.Slice(nodes, func(i, j int) bool {
		return scores[nodes[i].HashCode()] > scores[nodes[j].HashCode()]
	})

	return nodes, errOccurred
}
func (p *P2PServiceDefault) SignMessageSimple(message []byte) ([]byte, error) {
	signedMessage := protocol.NewSignedMessageRequest(message)
	signedMessage.SetNodeId(p.localNodeID)

	err := signedMessage.Sign(p.Config())

	if err != nil {
		return nil, err
	}

	result, err := msgpack.Marshal(signedMessage)

	if err != nil {
		return nil, err
	}

	return result, nil
}

func (p *P2PServiceDefault) AddPeer(peer net.Peer) error {
	peerId, err := peer.Id().ToString()
	if err != nil {
		return err
	}
	p.peers.Put(peerId, peer)
	p.reconnectDelay.PutUInt(peerId, 1)

	if p.peersPending.Contains(peerId) {
		p.peersPending.Remove(peerId)
	}

	return nil
}
func (p *P2PServiceDefault) SendPublicPeersToPeer(peer net.Peer, peersToSend []net.Peer) error {
	announceRequest := protocol.NewAnnounceRequest(peer, peersToSend)

	message, err := msgpack.Marshal(announceRequest)

	if err != nil {
		return err
	}

	signedMessage, err := p.SignMessageSimple(message)

	if err != nil {
		return err
	}

	err = peer.SendMessage(signedMessage)

	return nil
}
func (p *P2PServiceDefault) SendHashRequest(hash *encoding.Multihash, kinds []types.StorageLocationType) error {
	hashRequest := protocol.NewHashRequest(hash, kinds)
	message, err := msgpack.Marshal(hashRequest)
	if err != nil {
		return err
	}

	for _, peer := range p.peers.Values() {
		peerValue, ok := peer.(net.Peer)
		if !ok {
			p.Logger().Error("failed to cast peer to net.Peer")
			continue
		}
		err = peerValue.SendMessage(message)
	}

	return nil
}

func (p *P2PServiceDefault) UpVote(nodeId *encoding.NodeId) error {
	err := p.vote(nodeId, true)
	if err != nil {
		return err
	}

	return nil
}

func (p *P2PServiceDefault) DownVote(nodeId *encoding.NodeId) error {
	err := p.vote(nodeId, false)
	if err != nil {
		return err
	}

	return nil
}

func (p *P2PServiceDefault) vote(nodeId *encoding.NodeId, upvote bool) error {
	votes, err := p.readNodeVotes(nodeId)
	if err != nil {
		return err
	}

	if upvote {
		votes.Upvote()
	} else {
		votes.Downvote()
	}

	err = p.saveNodeVotes(nodeId, votes)
	if err != nil {
		return err
	}

	return nil
}
func (p *P2PServiceDefault) NodeId() *encoding.NodeId {
	return p.localNodeID
}

func (p *P2PServiceDefault) WaitOnConnectedPeers() {
	p.connections.Wait()
}

func (p *P2PServiceDefault) ConnectionTracker() *sync.WaitGroup {
	return &p.connections
}

func (p *P2PServiceDefault) NetworkId() string {
	return p.Config().P2P.Network
}
func (n *P2PServiceDefault) HashQueryRoutingTable() structs.Map {
	return n.hashQueryRoutingTable
}