oc-discovery/daemons/node/node.go

package node

import (
	"context"
	"crypto/sha256"
	"encoding/json"
	"errors"
	"fmt"
	"oc-discovery/conf"
	"oc-discovery/daemons/node/common"
	"oc-discovery/daemons/node/indexer"
	"oc-discovery/daemons/node/pubsub"
	"oc-discovery/daemons/node/stream"
	"sync"
	"time"

	oclib "cloud.o-forge.io/core/oc-lib"
	"cloud.o-forge.io/core/oc-lib/models/peer"
	"cloud.o-forge.io/core/oc-lib/tools"
	"github.com/google/uuid"
	"github.com/libp2p/go-libp2p"
	pubsubs "github.com/libp2p/go-libp2p-pubsub"
	"github.com/libp2p/go-libp2p/core/crypto"
	pp "github.com/libp2p/go-libp2p/core/peer"
)

type Node struct {
	*common.LongLivedStreamRecordedService[interface{}] // change type of stream
	PS                                                  *pubsubs.PubSub
	IndexerService                                      *indexer.IndexerService
	PubSubService                                       *pubsub.PubSubService
	StreamService                                       *stream.StreamService
	PeerID                                              pp.ID
	isIndexer                                           bool

	Mu    sync.RWMutex
	Peers map[pp.ID]bool
}

func InitNode(isNode bool, isIndexer bool) (*Node, error) {
	if !isNode && !isIndexer {
		return nil, errors.New("wait... what ? your node need to  at least something. Retry we can't be friend in that case")
	}
	logger := oclib.GetLogger()
	logger.Info().Msg("retrieving private key...")
	priv, err := tools.LoadKeyFromFilePrivate() // your node private key
	if err != nil {
		return nil, err
	}
	logger.Info().Msg("retrieving psk file...")
	psk, err := common.LoadPSKFromFile() // network common private Network. Public OC PSK is Public Network
	if err != nil {
		return nil, nil
	}
	logger.Info().Msg("open a host...")
	h, err := libp2p.New(
		libp2p.PrivateNetwork(psk),
		libp2p.Identity(priv),
		libp2p.ListenAddrStrings(
			fmt.Sprintf("/ip4/0.0.0.0/tcp/%d", conf.GetConfig().NodeEndpointPort),
		),
	)
	logger.Info().Msg("Host open on " + h.ID().String())
	if err != nil {
		return nil, errors.New("no host no node")
	}
	node := &Node{
		Peers:                          map[pp.ID]bool{},
		PeerID:                         h.ID(),
		isIndexer:                      isIndexer,
		LongLivedStreamRecordedService: common.NewStreamRecordedService[interface{}](h, 1000, false),
	}
	var ps *pubsubs.PubSub
	if isNode {
		logger.Info().Msg("generate opencloud node...")
		ps, err = pubsubs.NewGossipSub(context.Background(), node.Host)
		if err != nil {
			panic(err) // can't run your node without a propalgation pubsub, of state of node.
		}
		node.PS = ps
		logger.Info().Msg("connect to indexers...")
		common.ConnectToIndexers(node.Host, 0, 5, node.PeerID) // TODO : make var to change how many indexers are allowed.
		logger.Info().Msg("claims my node...")
		if _, err := node.claimInfo(conf.GetConfig().Name, conf.GetConfig().Hostname); err != nil {
			panic(err)
		}
		logger.Info().Msg("subscribe to decentralized search flow...")
		logger.Info().Msg("run garbage collector...")
		node.StartGC(30 * time.Second)

		if node.StreamService, err = stream.InitStream(context.Background(), node.Host, node.PeerID, 1000, node); err != nil {
			panic(err)
		}

		if node.PubSubService, err = pubsub.InitPubSub(context.Background(), node.Host, node.PS, node, node.StreamService); err != nil {
			panic(err)
		}
		f := func(ctx context.Context, evt common.Event, topic string) {
			if p, err := node.GetPeerRecord(ctx, evt.From); err == nil && len(p) > 0 {
				node.StreamService.SendResponse(p[0], &evt)
			}
		}
		node.SubscribeToSearch(node.PS, &f)
		ff := func(ctx context.Context, evt common.TopicNodeActivityPub, _ string) {
			node.Mu.Lock()
			defer node.Mu.Unlock()
			if pid, err := pp.Decode(evt.PeerID); err == nil {
				if _, ok := node.Peers[pid]; !ok {
					node.Peers[pid] = evt.NodeActivity == peer.ONLINE.EnumIndex()
					m := map[string]interface{}{
						"id":    evt.DID,
						"state": evt.NodeActivity,
					}
					if b, err := json.Marshal(m); err == nil {
						go tools.NewNATSCaller().SetNATSPub(tools.CREATE_PEER, tools.NATSResponse{
							FromApp:  "oc-discovery",
							Datatype: tools.PEER,
							Method:   int(tools.CREATE_PEER),
							Payload:  b,
						})
					}
				}
			}
		}
		node.SubscribeToNodeActivity(node.PS, &ff)
		access := oclib.NewRequestAdmin(oclib.LibDataEnum(oclib.PEER), nil)
		peers := access.LoadAll(false)
		for _, p := range peers.Data { // fill cache.
			if pid, err := pp.Decode(p.(*peer.Peer).PeerID); err == nil {
				node.Peers[pid] = p.(*peer.Peer).State == peer.ONLINE
			}
		}
	}
	if isIndexer {
		logger.Info().Msg("generate opencloud indexer...")
		node.IndexerService = indexer.NewIndexerService(node.Host, ps, 5)
	}
	logger.Info().Msg("connect to NATS")
	ListenNATS(node)
	logger.Info().Msg("Node is actually running.")
	return node, nil
}

func (d *Node) Close() {
	if d.isIndexer && d.IndexerService != nil {
		d.IndexerService.Close()
	}
	d.PubSubService.Close()
	d.StreamService.Close()
	d.Host.Close()
}

func (d *Node) publishPeerRecord(
	rec *indexer.PeerRecord,
) error {
	priv, err := tools.LoadKeyFromFilePrivate() // your node private key
	if err != nil {
		return err
	}
	if common.StreamIndexers[common.ProtocolPublish] == nil {
		return errors.New("no protocol Publish is set up on the node")
	}
	for _, ad := range common.StaticIndexers {
		if common.StreamIndexers[common.ProtocolPublish][ad.ID] == nil {
			return errors.New("no protocol Publish for peer " + ad.ID.String() + " is set up on the node")
		}
		stream := common.StreamIndexers[common.ProtocolPublish][ad.ID]
		base := indexer.PeerRecord{
			Name:       rec.Name,
			DID:        rec.DID,
			PubKey:     rec.PubKey,
			ExpiryDate: time.Now().UTC().Add(2 * time.Minute),
		}
		payload, _ := json.Marshal(base)
		hash := sha256.Sum256(payload)

		rec.ExpiryDate = base.ExpiryDate
		rec.Signature, err = priv.Sign(hash[:])
		rec.TTL = 2
		if err := json.NewEncoder(stream.Stream).Encode(&rec); err != nil { // then publish on stream
			return err
		}
	}
	return nil
}

func (d *Node) GetPeerRecord(
	ctx context.Context,
	key string,
) ([]*peer.Peer, error) {
	var err error
	var info map[string]indexer.PeerRecord
	if common.StreamIndexers[common.ProtocolPublish] == nil {
		return nil, errors.New("no protocol Publish is set up on the node")
	}
	for _, ad := range common.StaticIndexers {
		if common.StreamIndexers[common.ProtocolPublish][ad.ID] == nil {
			return nil, errors.New("no protocol Publish for peer " + ad.ID.String() + " is set up on the node")
		}
		stream := common.StreamIndexers[common.ProtocolPublish][ad.ID]
		if err := json.NewEncoder(stream.Stream).Encode(indexer.GetValue{Key: key}); err != nil {
			return nil, err
		}

		for {
			var resp indexer.GetResponse
			if err := json.NewDecoder(stream.Stream).Decode(&resp); err != nil {
				return nil, err
			}
			if resp.Found {
				info = resp.Records
				break
			}
		}
	}
	var ps []*peer.Peer
	for _, pr := range info {
		if pk, err := pr.Verify(); err != nil {
			return nil, err
		} else if ok, p, err := pr.ExtractPeer(d.PeerID.String(), key, pk); err != nil {
			return nil, err
		} else {
			if ok {
				d.publishPeerRecord(&pr)
			}
			ps = append(ps, p)
		}
	}

	return ps, err
}

func (d *Node) claimInfo(
	name string,
	endPoint string, // TODO : endpoint is not necesserry StreamAddress
) (*peer.Peer, error) {
	if endPoint == "" {
		return nil, errors.New("no endpoint found for peer")
	}
	peerID := uuid.New().String()
	priv, err := tools.LoadKeyFromFilePrivate()
	if err != nil {
		return nil, err
	}
	pub, err := tools.LoadKeyFromFilePublic()
	if err != nil {
		return nil, err
	}
	pubBytes, err := crypto.MarshalPublicKey(pub)
	if err != nil {
		return nil, err
	}

	now := time.Now().UTC()
	expiry := now.Add(150 * time.Second)

	rec := &indexer.PeerRecord{
		Name:   name,
		DID:    peerID, // REAL PEER ID
		PubKey: pubBytes,
	}

	rec.PeerID = d.Host.ID().String()
	d.PeerID = d.Host.ID()

	payload, _ := json.Marshal(rec)
	hash := sha256.Sum256(payload)

	rec.Signature, err = priv.Sign(hash[:])
	if err != nil {
		return nil, err
	}

	rec.APIUrl = endPoint
	rec.StreamAddress = "/ip4/" + conf.GetConfig().Hostname + "/tcp/" + fmt.Sprintf("%v", conf.GetConfig().NodeEndpointPort) + "/p2p/" + rec.PeerID
	rec.NATSAddress = oclib.GetConfig().NATSUrl
	rec.WalletAddress = "my-wallet"
	rec.ExpiryDate = expiry

	if err := d.publishPeerRecord(rec); err != nil {
		return nil, err
	}
	/*if pk, err := rec.Verify(); err != nil {
		fmt.Println("Verify")
		return nil, err
	} else {*/
	_, p, err := rec.ExtractPeer(peerID, peerID, pub)
	return p, err
	//}
}