libs5-go/protocol/hash_query.go

187 lines
3.8 KiB
Go
Raw Normal View History

2024-01-06 11:33:46 +00:00
package protocol
import (
"git.lumeweb.com/LumeWeb/libs5-go/encoding"
"git.lumeweb.com/LumeWeb/libs5-go/net"
"git.lumeweb.com/LumeWeb/libs5-go/storage"
"git.lumeweb.com/LumeWeb/libs5-go/types"
"github.com/emirpasic/gods/sets/hashset"
2024-01-06 11:33:46 +00:00
"github.com/vmihailenco/msgpack/v5"
"go.uber.org/zap"
"log"
2024-01-06 11:33:46 +00:00
)
2024-01-30 03:28:39 +00:00
var _ EncodeableMessage = (*HashQuery)(nil)
2024-01-30 03:25:21 +00:00
var _ IncomingMessage = (*HashQuery)(nil)
2024-01-06 11:33:46 +00:00
type HashQuery struct {
hash *encoding.Multihash
kinds []types.StorageLocationType
2024-01-30 03:25:21 +00:00
HandshakeRequirement
2024-01-06 11:33:46 +00:00
}
func NewHashQuery() *HashQuery {
hq := &HashQuery{}
hq.SetRequiresHandshake(true)
return hq
}
2024-01-09 11:57:06 +00:00
func NewHashRequest(hash *encoding.Multihash, kinds []types.StorageLocationType) *HashQuery {
if len(kinds) == 0 {
kinds = []types.StorageLocationType{types.StorageLocationTypeFile}
}
return &HashQuery{
hash: hash,
kinds: kinds,
}
}
2024-01-06 11:33:46 +00:00
func (h HashQuery) Hash() *encoding.Multihash {
return h.hash
}
func (h HashQuery) Kinds() []types.StorageLocationType {
2024-01-06 11:33:46 +00:00
return h.kinds
}
2024-01-30 03:25:21 +00:00
func (h *HashQuery) DecodeMessage(dec *msgpack.Decoder, message IncomingMessageData) error {
2024-01-06 11:33:46 +00:00
hash, err := dec.DecodeBytes()
if err != nil {
return err
}
h.hash = encoding.NewMultihash(hash)
var kinds []types.StorageLocationType
2024-01-06 11:33:46 +00:00
err = dec.Decode(&kinds)
if err != nil {
return err
}
h.kinds = kinds
return nil
}
func (h HashQuery) EncodeMsgpack(enc *msgpack.Encoder) error {
err := enc.EncodeInt(int64(types.ProtocolMethodHashQuery))
if err != nil {
return err
}
err = enc.EncodeBytes(h.hash.FullBytes())
if err != nil {
return err
}
err = enc.Encode(h.kinds)
if err != nil {
return err
}
return nil
}
2024-01-30 03:25:21 +00:00
func (h *HashQuery) HandleMessage(message IncomingMessageData) error {
peer := message.Peer
mediator := message.Mediator
logger := message.Logger
config := message.Config
mapLocations, err := mediator.GetCachedStorageLocations(h.hash, h.kinds)
if err != nil {
log.Printf("Error getting cached storage locations: %v", err)
return err
}
if len(mapLocations) > 0 {
availableNodes := make([]*encoding.NodeId, 0, len(mapLocations))
for key := range mapLocations {
nodeId, err := encoding.DecodeNodeId(key)
if err != nil {
logger.Error("Error decoding node id", zap.Error(err))
continue
}
availableNodes = append(availableNodes, nodeId)
}
score, err := mediator.SortNodesByScore(availableNodes)
if err != nil {
return err
}
2024-01-07 08:58:22 +00:00
sortedNodeId, err := (*score[0]).ToString()
if err != nil {
return err
}
entry, exists := mapLocations[sortedNodeId]
if exists {
err := peer.SendMessage(entry.ProviderMessage())
if err != nil {
return err
}
}
}
if mediator.ProviderStore() != nil {
if mediator.ProviderStore().CanProvide(h.hash, h.kinds) {
location, err := mediator.ProviderStore().Provide(h.hash, h.kinds)
2024-01-24 08:01:59 +00:00
if err != nil {
return err
}
message := storage.PrepareProvideMessage(config.KeyPair, h.hash, location)
2024-01-24 08:01:59 +00:00
err = mediator.AddStorageLocation(h.hash, mediator.NodeId(), location, message)
2024-01-24 08:01:59 +00:00
if err != nil {
return err
}
err = peer.SendMessage(message)
if err != nil {
return err
}
}
}
var peers *hashset.Set
hashString, err := h.hash.ToString()
logger.Debug("HashQuery", zap.Any("hashString", hashString))
if err != nil {
return err
}
peersVal, ok := mediator.HashQueryRoutingTable().Get(hashString)
if ok {
peers = peersVal.(*hashset.Set)
if !peers.Contains(peer.Id()) {
peers.Add(peer.Id())
}
return nil
}
peerList := hashset.New()
peerList.Add(peer.Id())
mediator.HashQueryRoutingTable().Put(hashString, peerList)
for _, val := range mediator.Peers().Values() {
peerVal := val.(net.Peer)
if !peerVal.Id().Equals(peer.Id()) {
err := peerVal.SendMessage(message.Original)
if err != nil {
logger.Error("Failed to send message", zap.Error(err))
}
}
}
2024-01-06 11:33:46 +00:00
return nil
}