oc-discovery/daemons/node/indexer/handler.go

package indexer

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"oc-discovery/conf"
	"oc-discovery/daemons/node/common"
	"time"

	oclib "cloud.o-forge.io/core/oc-lib"
	pp "cloud.o-forge.io/core/oc-lib/models/peer"
	"cloud.o-forge.io/core/oc-lib/models/utils"
	"cloud.o-forge.io/core/oc-lib/tools"
	"github.com/libp2p/go-libp2p/core/crypto"
	"github.com/libp2p/go-libp2p/core/network"
	"github.com/libp2p/go-libp2p/core/peer"
)

type PeerRecord struct {
	Name          string    `json:"name"`
	DID           string    `json:"did"` // real PEER ID
	PeerID        string    `json:"peer_id"`
	PubKey        []byte    `json:"pub_key"`
	APIUrl        string    `json:"api_url"`
	StreamAddress string    `json:"stream_address"`
	NATSAddress   string    `json:"nats_address"`
	WalletAddress string    `json:"wallet_address"`
	Signature     []byte    `json:"signature"`
	ExpiryDate    time.Time `json:"expiry_date"`

	TTL   int  `json:"ttl"` // max of hop diffusion
	NoPub bool `json:"no_pub"`
}

func (p *PeerRecord) Sign() error {
	priv, err := common.LoadKeyFromFilePrivate()
	if err != nil {
		return err
	}
	dht := PeerRecord{
		Name:       p.Name,
		DID:        p.DID,
		PubKey:     p.PubKey,
		ExpiryDate: p.ExpiryDate,
	}
	payload, _ := json.Marshal(dht)
	b, err := common.Sign(priv, payload)
	p.Signature = b
	return err
}

func (p *PeerRecord) Verify() (crypto.PubKey, error) {
	fmt.Println(p.PubKey)
	pubKey, err := crypto.UnmarshalPublicKey(p.PubKey) // retrieve pub key in message
	if err != nil {
		fmt.Println("UnmarshalPublicKey")
		return pubKey, err
	}
	dht := PeerRecord{
		Name:       p.Name,
		DID:        p.DID,
		PubKey:     p.PubKey,
		ExpiryDate: p.ExpiryDate,
	}
	payload, _ := json.Marshal(dht)

	if ok, _ := common.Verify(pubKey, payload, p.Signature); !ok { // verify minimal message was sign per pubKey
		fmt.Println("Verify")
		return pubKey, errors.New("invalid signature")
	}
	return pubKey, nil
}

func (pr *PeerRecord) ExtractPeer(ourkey string, key string, pubKey crypto.PubKey) (bool, *pp.Peer, error) {
	pubBytes, err := crypto.MarshalPublicKey(pubKey)
	if err != nil {
		return false, nil, err
	}
	rel := pp.NONE
	if ourkey == key { // at this point is PeerID is same as our... we are... thats our peer INFO
		rel = pp.SELF
	}

	p := &pp.Peer{
		AbstractObject: utils.AbstractObject{
			UUID: pr.DID,
			Name: pr.Name,
		},
		State:         pp.ONLINE,
		Relation:      rel, // VERIFY.... it crush nothing
		PeerID:        pr.PeerID,
		PublicKey:     base64.StdEncoding.EncodeToString(pubBytes),
		APIUrl:        pr.APIUrl,
		StreamAddress: pr.StreamAddress,
		NATSAddress:   pr.NATSAddress,
		WalletAddress: pr.WalletAddress,
	}
	if time.Now().UTC().After(pr.ExpiryDate) { // is expired
		p.State = pp.OFFLINE // then is considers OFFLINE
	}
	b, err := json.Marshal(p)
	if err != nil {
		return pp.SELF == p.Relation, nil, err
	}
	fmt.Println("SENDPEER SELF")
	go tools.NewNATSCaller().SetNATSPub(tools.CREATE_PEER, tools.NATSResponse{
		FromApp:  "oc-discovery",
		Datatype: tools.PEER,
		Method:   int(tools.CREATE_PEER),
		Payload:  b,
	})
	if p.State == pp.OFFLINE {
		return pp.SELF == p.Relation, nil, errors.New("peer " + key + " is offline")
	}
	return pp.SELF == p.Relation, p, nil
}

type GetValue struct {
	Key string `json:"key"`
}

type GetResponse struct {
	Found   bool                  `json:"found"`
	Records map[string]PeerRecord `json:"records,omitempty"`
}

func (ix *IndexerService) initNodeHandler() {
	fmt.Println("Node activity")
	ix.Host.SetStreamHandler(common.ProtocolHeartbeat, ix.HandleNodeHeartbeat)
	ix.Host.SetStreamHandler(common.ProtocolPublish, ix.handleNodePublish)
	ix.Host.SetStreamHandler(common.ProtocolGet, ix.handleNodeGet)
}

func (ix *IndexerService) handleNodePublish(s network.Stream) {
	defer s.Close()
	logger := oclib.GetLogger()
	for {
		var rec PeerRecord
		if err := json.NewDecoder(s).Decode(&rec); err != nil {
			continue
		}
		rec2 := PeerRecord{
			Name:   rec.Name,
			DID:    rec.DID, // REAL PEER ID
			PubKey: rec.PubKey,
			PeerID: rec.PeerID,
		}
		if _, err := rec2.Verify(); err != nil {
			logger.Err(err)
			continue
		}
		if rec.PeerID == "" || rec.ExpiryDate.Before(time.Now().UTC()) { // already expired
			logger.Warn().Msg(rec.PeerID + " is expired.")
			continue
		}
		pid, err := peer.Decode(rec.PeerID)
		if err != nil {
			continue
		}

		ix.StreamMU.Lock()

		if ix.StreamRecords[common.ProtocolPublish] == nil {
			ix.StreamRecords[common.ProtocolPublish] = map[peer.ID]*common.StreamRecord[PeerRecord]{}
		}
		streams := ix.StreamRecords[common.ProtocolPublish]

		if srec, ok := streams[pid]; ok {
			fmt.Println("UPDATE PUBLISH", pid)
			srec.DID = rec.DID
			srec.Record = rec
			srec.LastSeen = time.Now().UTC()
		} else {
			fmt.Println("CREATE PUBLISH", pid)
			streams[pid] = &common.StreamRecord[PeerRecord]{ // HeartBeat wil
				DID:      rec.DID,
				Record:   rec,
				LastSeen: time.Now().UTC(),
			}
		}

		if ix.LongLivedPubSubs[common.TopicPubSubNodeActivity] != nil && !rec.NoPub {
			ad, err := peer.AddrInfoFromString("/ip4/" + conf.GetConfig().Hostname + "/tcp/" + fmt.Sprintf("%v", conf.GetConfig().NodeEndpointPort) + "/p2p/" + ix.Host.ID().String())
			if err == nil {
				if b, err := json.Marshal(common.TopicNodeActivityPub{
					Disposer:     *ad,
					DID:          rec.DID,
					Name:         rec.Name,
					PeerID:       pid.String(),
					NodeActivity: pp.ONLINE,
				}); err == nil {
					ix.LongLivedPubSubs[common.TopicPubSubNodeActivity].Publish(context.Background(), b)
				}
			}
		}

		if rec.TTL > 0 {
			rec.NoPub = true
			for _, ad := range common.StaticIndexers {
				if ad.ID == s.Conn().RemotePeer() {
					continue
				}
				if common.StreamIndexers[common.ProtocolPublish][ad.ID] == nil {
					continue
				}
				stream := common.StreamIndexers[common.ProtocolPublish][ad.ID]
				rec.TTL -= 1
				if err := json.NewEncoder(stream.Stream).Encode(&rec); err != nil { // then publish on stream
					continue
				}
			}
		}
		ix.StreamMU.Unlock()
	}
}

func (ix *IndexerService) handleNodeGet(s network.Stream) {
	defer s.Close()
	logger := oclib.GetLogger()
	for {
		var req GetValue
		if err := json.NewDecoder(s).Decode(&req); err != nil {
			logger.Err(err)
			continue
		}
		ix.StreamMU.Lock()

		if ix.StreamRecords[common.ProtocolGet] == nil {
			ix.StreamRecords[common.ProtocolGet] = map[peer.ID]*common.StreamRecord[PeerRecord]{}
		}
		resp := GetResponse{
			Found:   false,
			Records: map[string]PeerRecord{},
		}
		streams := ix.StreamRecords[common.ProtocolPublish]
		// simple lookup by PeerID (or DID)
		for _, rec := range streams {
			if rec.Record.DID == req.Key || rec.Record.PeerID == req.Key || rec.Record.Name == req.Key { // OK
				resp.Found = true
				resp.Records[rec.Record.PeerID] = rec.Record
				if rec.Record.DID == req.Key || rec.Record.PeerID == req.Key { // there unique... no need to proceed more...
					_ = json.NewEncoder(s).Encode(resp)
					ix.StreamMU.Unlock()
					return
				}
				continue
			}
		}
		// if not found ask to my neighboor indexers
		for pid, dsp := range ix.DisposedPeers {
			if _, ok := resp.Records[dsp.PeerID]; !ok && (dsp.Name == req.Key || dsp.DID == req.Key || dsp.PeerID == req.Key) {
				ctxTTL, err := context.WithTimeout(context.Background(), 120*time.Second)
				if err != nil {
					continue
				}
				if ix.Host.Network().Connectedness(pid) != network.Connected {
					_ = ix.Host.Connect(ctxTTL, dsp.Disposer)
					str, err := ix.Host.NewStream(ctxTTL, pid, common.ProtocolGet)
					if err != nil {
						continue
					}
					for {
						if ctxTTL.Err() == context.DeadlineExceeded {
							break
						}
						var subResp GetResponse
						if err := json.NewDecoder(str).Decode(&resp); err != nil {
							continue
						}
						if subResp.Found {
							for k, v := range subResp.Records {
								if _, ok := resp.Records[k]; !ok {
									resp.Records[k] = v
								}
							}
							break
						}
					}
				}
			}
		}
		// Not found
		_ = json.NewEncoder(s).Encode(resp)
		ix.StreamMU.Unlock()
	}
}