FULL OC-DISCOVERY LOGIC

2026-01-30 16:57:36 +01:00
parent d50e5d56f7
commit 562d86125e
24 changed files with 1769 additions and 929 deletions
--- a/daemons/node/stream/handler.go
+++ b/daemons/node/stream/handler.go
@@ -0,0 +1,123 @@
+package stream
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"oc-discovery/daemons/node/common"
+	"strings"
+
+	oclib "cloud.o-forge.io/core/oc-lib"
+	"cloud.o-forge.io/core/oc-lib/models/peer"
+	"cloud.o-forge.io/core/oc-lib/models/resources"
+	"cloud.o-forge.io/core/oc-lib/tools"
+)
+
+func (ps *StreamService) getTopicName(topicName string) tools.PubSubAction {
+	ns := strings.Split(topicName, ".")
+	if len(ns) > 0 {
+		return tools.GetActionString(ns[0])
+	}
+	return tools.NONE
+}
+
+func (ps *StreamService) handleEvent(topicName string, evt *common.Event) error {
+	action := ps.getTopicName(topicName)
+	if err := ps.handleEventFromPartner(evt, action); err != nil {
+		return err
+	}
+	if action == tools.PB_SEARCH_RESPONSE {
+		if err := ps.retrieveResponse(evt); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (abs *StreamService) retrieveResponse(event *common.Event) error { //
+	res, err := resources.ToResource(int(event.DataType), event.Payload)
+	if err != nil || res == nil {
+		return nil
+	}
+	b, err := json.Marshal(res.Serialize(res))
+	tools.NewNATSCaller().SetNATSPub(tools.CATALOG_SEARCH_EVENT, tools.NATSResponse{
+		FromApp:  "oc-discovery",
+		Datatype: tools.DataType(event.DataType),
+		Method:   int(tools.CATALOG_SEARCH_EVENT),
+		Payload:  b,
+	})
+	return nil
+}
+
+func (ps *StreamService) handleEventFromPartner(evt *common.Event, action tools.PubSubAction) error {
+	if !(action == tools.PB_CREATE || action == tools.PB_UPDATE || action == tools.PB_DELETE) {
+		return nil
+	}
+	resource, err := resources.ToResource(int(evt.DataType), evt.Payload)
+	if err != nil {
+		return err
+	}
+	b, err := json.Marshal(resource)
+	if err != nil {
+		return err
+	}
+	switch action {
+	case tools.PB_SEARCH:
+		access := oclib.NewRequestAdmin(oclib.LibDataEnum(oclib.PEER), nil)
+		peers := access.Search(nil, evt.From, false)
+		if len(peers.Data) > 0 {
+			p := peers.Data[0].(*peer.Peer)
+			// TODO : something if peer is missing in our side !
+			ps.SendResponse(p, evt)
+		} else if p, err := ps.Node.GetPeerRecord(context.Background(), evt.From); err == nil {
+			ps.SendResponse(p, evt)
+		}
+	case tools.PB_CREATE:
+	case tools.PB_UPDATE:
+		tools.NewNATSCaller().SetNATSPub(tools.CREATE_RESOURCE, tools.NATSResponse{
+			FromApp:  "oc-discovery",
+			Datatype: tools.DataType(evt.DataType),
+			Method:   int(tools.CREATE_RESOURCE),
+			Payload:  b,
+		})
+	case tools.PB_DELETE:
+		tools.NewNATSCaller().SetNATSPub(tools.REMOVE_RESOURCE, tools.NATSResponse{
+			FromApp:  "oc-discovery",
+			Datatype: tools.DataType(evt.DataType),
+			Method:   int(tools.REMOVE_RESOURCE),
+			Payload:  b,
+		})
+	default:
+		return errors.New("no action authorized available : " + action.String())
+	}
+	return nil
+}
+
+func (abs *StreamService) SendResponse(p *peer.Peer, event *common.Event) error {
+	dts := []oclib.LibDataEnum{oclib.LibDataEnum(event.DataType)}
+	if event.DataType == -1 { // expect all resources
+		dts = []oclib.LibDataEnum{oclib.LibDataEnum(oclib.COMPUTE_RESOURCE), oclib.LibDataEnum(oclib.STORAGE_RESOURCE),
+			oclib.LibDataEnum(oclib.PROCESSING_RESOURCE), oclib.LibDataEnum(oclib.DATA_RESOURCE), oclib.LibDataEnum(oclib.WORKFLOW_RESOURCE)}
+	}
+	var m map[string]string
+	err := json.Unmarshal(event.Payload, &m)
+	if err != nil {
+		return err
+	}
+	for _, dt := range dts {
+		access := oclib.NewRequestAdmin(oclib.LibDataEnum(event.DataType), nil)
+		peerID := p.GetID()
+		searched := access.Search(abs.FilterPeer(peerID, m["search"]), "", false)
+		for _, ss := range searched.Data {
+			if j, err := json.Marshal(ss); err == nil {
+				if event.DataType != -1 {
+					ndt := tools.DataType(dt.EnumIndex())
+					abs.PublishResources(&ndt, event.User, peerID, j)
+				} else {
+					abs.PublishResources(nil, event.User, peerID, j)
+				} // TODO : TEMP STREAM !
+			}
+		}
+	}
+	return nil
+}
--- a/daemons/node/stream/publish.go
+++ b/daemons/node/stream/publish.go
@@ -0,0 +1,150 @@
+package stream
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"oc-discovery/daemons/node/common"
+	"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/libp2p/go-libp2p/core/network"
+	pp "github.com/libp2p/go-libp2p/core/peer"
+	"github.com/libp2p/go-libp2p/core/protocol"
+)
+
+func (ps *StreamService) PublishResources(dt *tools.DataType, user string, toPeerID string, resource []byte) error {
+	access := oclib.NewRequestAdmin(oclib.LibDataEnum(oclib.PEER), nil)
+	p := access.LoadOne(toPeerID)
+	if p.Err != "" {
+		return errors.New(p.Err)
+	} else {
+		ad, err := pp.AddrInfoFromString(p.Data.(*peer.Peer).StreamAddress)
+		if err != nil {
+			return err
+		}
+		ps.write(tools.PB_SEARCH, toPeerID, ad, dt, user, resource, ProtocolSearchResource)
+	}
+	return nil
+}
+
+func (ps *StreamService) SearchKnownPublishEvent(dt *tools.DataType, user string, search string) error {
+	access := oclib.NewRequestAdmin(oclib.LibDataEnum(oclib.PEER), nil)
+	peers := access.Search(nil, search, false)
+	if peers.Err != "" {
+		return errors.New(peers.Err)
+	} else {
+		b, err := json.Marshal(map[string]string{"search": search})
+		if err != nil {
+			return err
+		}
+		for _, p := range peers.Data {
+			ad, err := pp.AddrInfoFromString(p.(*peer.Peer).StreamAddress)
+			if err != nil {
+				continue
+			}
+			ps.write(tools.PB_SEARCH, p.GetID(), ad, dt, user, b, ProtocolSearchResource)
+		}
+	}
+	return nil
+}
+
+func (ps *StreamService) SearchPartnersPublishEvent(dt *tools.DataType, user string, search string) error {
+	if peers, err := ps.searchPeer(fmt.Sprintf("%v", peer.PARTNER.EnumIndex())); err != nil {
+		return err
+	} else {
+		b, err := json.Marshal(map[string]string{"search": search})
+		if err != nil {
+			return err
+		}
+		for _, p := range peers {
+			ad, err := pp.AddrInfoFromString(p.StreamAddress)
+			if err != nil {
+				continue
+			}
+			ps.write(tools.PB_SEARCH, p.GetID(), ad, dt, user, b, ProtocolSearchResource)
+		}
+	}
+	return nil
+}
+
+func (ps *StreamService) ToPartnerPublishEvent(
+	ctx context.Context, action tools.PubSubAction, dt *tools.DataType, user string, payload []byte) error {
+	if peers, err := ps.searchPeer(fmt.Sprintf("%v", peer.PARTNER.EnumIndex())); err != nil {
+		return err
+	} else {
+		for _, p := range peers {
+			for _, protocol := range protocols {
+				ad, err := pp.AddrInfoFromString(p.StreamAddress)
+				if err != nil {
+					continue
+				}
+				ps.write(action, p.GetID(), ad, dt, user, payload, protocol)
+			}
+		}
+	}
+	return nil
+}
+
+func (s *StreamService) write(
+	action tools.PubSubAction,
+	did string,
+	peerID *pp.AddrInfo,
+	dt *tools.DataType,
+	user string,
+	payload []byte,
+	proto protocol.ID) error {
+	logger := oclib.GetLogger()
+
+	name := action.String() + "#" + peerID.ID.String()
+	if dt != nil {
+		name = action.String() + "." + (*dt).String() + "#" + peerID.ID.String()
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.Streams[proto] == nil {
+		s.Streams[proto] = map[pp.ID]*common.Stream{}
+	}
+
+	if s.Streams[proto][peerID.ID] == nil {
+		// should create a very temp stream
+		ctxTTL, err := context.WithTimeout(context.Background(), 10*time.Second)
+		if err == nil {
+			if s.Host.Network().Connectedness(peerID.ID) != network.Connected {
+				_ = s.Host.Connect(ctxTTL, *peerID)
+				str, err := s.Host.NewStream(ctxTTL, peerID.ID, ProtocolHeartbeatPartner)
+				if err == nil {
+					s.Streams[ProtocolHeartbeatPartner][peerID.ID] = &common.Stream{
+						DID:    did,
+						Stream: str,
+						Expiry: time.Now().UTC().Add(5 * time.Second),
+					}
+					str2, err := s.Host.NewStream(ctxTTL, peerID.ID, proto)
+					if err == nil {
+						s.Streams[proto][peerID.ID] = &common.Stream{
+							DID:    did,
+							Stream: str2,
+							Expiry: time.Now().UTC().Add(5 * time.Second),
+						}
+					}
+				}
+
+			}
+		}
+		return errors.New("no stream available for protocol " + fmt.Sprintf("%v", proto) + " from PID " + peerID.ID.String())
+	}
+	stream := s.Streams[proto][peerID.ID]
+
+	enc := json.NewEncoder(stream.Stream)
+
+	evt := common.NewEvent(name, peerID.ID.String(), dt, user, payload)
+	if err := enc.Encode(evt); err != nil {
+		stream.Stream.Close()
+		logger.Err(err)
+		return nil
+	}
+	return nil
+}
--- a/daemons/node/stream/service.go
+++ b/daemons/node/stream/service.go
@@ -0,0 +1,228 @@
+package stream
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"oc-discovery/daemons/node/common"
+	"sync"
+	"time"
+
+	oclib "cloud.o-forge.io/core/oc-lib"
+	"cloud.o-forge.io/core/oc-lib/dbs"
+	"cloud.o-forge.io/core/oc-lib/models/peer"
+	"github.com/libp2p/go-libp2p/core/host"
+	"github.com/libp2p/go-libp2p/core/network"
+	pp "github.com/libp2p/go-libp2p/core/peer"
+	"github.com/libp2p/go-libp2p/core/protocol"
+)
+
+const ProtocolSearchResource = "/opencloud/resource/search/1.0"
+const ProtocolCreateResource = "/opencloud/resource/create/1.0"
+const ProtocolUpdateResource = "/opencloud/resource/update/1.0"
+const ProtocolDeleteResource = "/opencloud/resource/delete/1.0"
+
+const ProtocolHeartbeatPartner = "/opencloud/resource/heartbeat/partner/1.0"
+
+var protocols = []protocol.ID{
+	ProtocolSearchResource,
+	ProtocolCreateResource,
+	ProtocolUpdateResource,
+	ProtocolDeleteResource,
+}
+
+type StreamService struct {
+	Key          pp.ID
+	Host         host.Host
+	Node         common.DiscoveryPeer
+	Streams      common.ProtocolStream
+	maxNodesConn int
+	mu           sync.Mutex
+	// Stream       map[protocol.ID]map[pp.ID]*daemons.Stream
+}
+
+func InitStream(ctx context.Context, h host.Host, key pp.ID, maxNode int, node common.DiscoveryPeer) (*StreamService, error) {
+	service := &StreamService{
+		Key:          key,
+		Node:         node,
+		Host:         h,
+		Streams:      common.ProtocolStream{},
+		maxNodesConn: maxNode,
+	}
+	service.Host.SetStreamHandler(ProtocolHeartbeatPartner, service.HandlePartnerHeartbeat)
+	service.connectToPartners() // we set up a stream
+	go service.StartGC(30 * time.Second)
+	return service, nil
+}
+
+func (s *StreamService) HandlePartnerHeartbeat(stream network.Stream) {
+	pid, hb, err := common.CheckHeartbeat(s.Host, stream, s.maxNodesConn)
+	if err != nil {
+		return
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	streams := s.Streams[ProtocolHeartbeatPartner]
+	if streams == nil {
+		s.Streams[ProtocolHeartbeatPartner] = map[pp.ID]*common.Stream{}
+		return
+	}
+	// if record already seen update last seen
+	if rec, ok := streams[*pid]; ok {
+		rec.DID = hb.DID
+		rec.Expiry = time.Now().UTC().Add(2 * time.Minute)
+	} else { // if not in stream ?
+		pid := stream.Conn().RemotePeer()
+		ai, err := pp.AddrInfoFromP2pAddr(stream.Conn().RemoteMultiaddr())
+		if err == nil {
+			s.connectToPartner(pid, ai)
+		}
+	}
+	go s.StartGC(30 * time.Second)
+}
+
+func (s *StreamService) connectToPartners() error {
+	peers, err := s.searchPeer(fmt.Sprintf("%v", peer.PARTNER.EnumIndex()))
+	if err != nil {
+		return err
+	}
+	for _, p := range peers {
+		ad, err := pp.AddrInfoFromString(p.StreamAddress)
+		if err != nil {
+			return err
+		}
+		pid, err := pp.Decode(p.PeerID)
+		if err != nil {
+			continue
+		}
+		s.connectToPartner(pid, ad)
+		// heartbeat your partner.
+	}
+	// TODO if handle... from partner then HeartBeat back
+	return nil
+}
+
+func (s *StreamService) connectToPartner(pid pp.ID, ad *pp.AddrInfo) {
+	for _, proto := range protocols {
+		f := func(ss network.Stream) {
+			if s.Streams[proto] == nil {
+				s.Streams[proto] = map[pp.ID]*common.Stream{}
+			}
+			s.Streams[proto][pid] = &common.Stream{
+				Stream: ss,
+				Expiry: time.Now().Add(2 * time.Minute),
+			}
+			go s.readLoop(s.Streams[proto][pid])
+		}
+		s.Streams = common.AddStreamProtocol(nil, s.Streams, s.Host, proto, pid, s.Key, &f)
+		if s.Streams[proto][pid] != nil {
+			go s.readLoop(s.Streams[proto][pid]) // reaaaad...
+		}
+	}
+	common.SendHeartbeat(context.Background(), ProtocolHeartbeatPartner,
+		s.Host, s.Streams, []*pp.AddrInfo{ad}, time.Minute)
+}
+
+func (s *StreamService) searchPeer(search string) ([]*peer.Peer, error) {
+	ps := []*peer.Peer{}
+	access := oclib.NewRequestAdmin(oclib.LibDataEnum(oclib.PEER), nil)
+	peers := access.Search(nil, search, false)
+	if len(peers.Data) == 0 {
+		return ps, errors.New("no self available")
+	}
+	for _, p := range peers.Data {
+		ps = append(ps, p.(*peer.Peer))
+	}
+	return ps, nil
+}
+
+func (ix *StreamService) Close() {
+	for _, s := range ix.Streams {
+		for _, ss := range s {
+			ss.Stream.Close()
+		}
+	}
+}
+
+func (s *StreamService) StartGC(interval time.Duration) {
+	go func() {
+		t := time.NewTicker(interval)
+		defer t.Stop()
+		for range t.C {
+			s.gc()
+		}
+	}()
+}
+
+func (s *StreamService) gc() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	now := time.Now()
+	streams := s.Streams[ProtocolHeartbeatPartner]
+	if streams == nil {
+		s.Streams[ProtocolHeartbeatPartner] = map[pp.ID]*common.Stream{}
+		return
+	}
+	for pid, rec := range streams {
+		if now.After(rec.Expiry) {
+			for _, sstreams := range s.Streams {
+				if sstreams[pid] != nil {
+					sstreams[pid].Stream.Close()
+					delete(sstreams, pid)
+				}
+			}
+		}
+	}
+}
+
+func (ps *StreamService) readLoop(s *common.Stream) {
+	dec := json.NewDecoder(s.Stream)
+	for {
+		var evt common.Event
+		if err := dec.Decode(&evt); err != nil {
+			s.Stream.Close()
+			return
+		}
+		ps.handleEvent(evt.Type, &evt)
+	}
+}
+
+func (abs *StreamService) FilterPeer(peerID string, search string) *dbs.Filters {
+	id, err := oclib.GetMySelf()
+	if err != nil {
+		return nil
+	}
+	filter := map[string][]dbs.Filter{
+		"creator_id": {{Operator: dbs.EQUAL.String(), Value: id}}, // is my resource...
+		"": {{Operator: dbs.OR.String(), Value: &dbs.Filters{
+			Or: map[string][]dbs.Filter{
+				"abstractobject.access_mode": {{Operator: dbs.EQUAL.String(), Value: 1}}, // if public
+				"abstractinstanciatedresource.instances": {{Operator: dbs.ELEMMATCH.String(), Value: &dbs.Filters{ // or got a partners instances
+					And: map[string][]dbs.Filter{
+						"resourceinstance.partnerships": {{Operator: dbs.ELEMMATCH.String(), Value: &dbs.Filters{
+							And: map[string][]dbs.Filter{
+								"resourcepartnership.peer_groups." + peerID: {{Operator: dbs.EXISTS.String(), Value: true}},
+							},
+						}}},
+					},
+				}}},
+			},
+		}}},
+	}
+	if search != "" {
+		filter[" "] = []dbs.Filter{{Operator: dbs.OR.String(), Value: &dbs.Filters{
+			Or: map[string][]dbs.Filter{ // filter by like name, short_description, description, owner, url if no filters are provided
+				"abstractintanciatedresource.abstractresource.abstractobject.name":       {{Operator: dbs.LIKE.String(), Value: search}},
+				"abstractintanciatedresource.abstractresource.type":                      {{Operator: dbs.LIKE.String(), Value: search}},
+				"abstractintanciatedresource.abstractresource.short_description":         {{Operator: dbs.LIKE.String(), Value: search}},
+				"abstractintanciatedresource.abstractresource.description":               {{Operator: dbs.LIKE.String(), Value: search}},
+				"abstractintanciatedresource.abstractresource.owners.name":               {{Operator: dbs.LIKE.String(), Value: search}},
+				"abstractintanciatedresource.abstractresource.abstractobject.creator_id": {{Operator: dbs.EQUAL.String(), Value: search}},
+			},
+		}}}
+	}
+	return &dbs.Filters{
+		And: filter,
+	}
+}