oc-catalog/infrastructure/subscribe.go

package infrastructure

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"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"
	pubsub "github.com/libp2p/go-libp2p-pubsub"
	"github.com/libp2p/go-libp2p/core/crypto"
)

func (ps *PubSubService) InitSubscribeEvents(ctx context.Context) error {
	ourPeerID, err := oclib.GenerateNodeID()
	if err != nil {
		return err
	}
	// subscribe :
	if err := ps.subscribeEvents(ctx, nil, SEARCH, "", -1); err != nil { // we subscribe at our proprer deductible search adresse.
		return err
	}
	if err := ps.subscribeEvents(ctx, nil, SEARCH, ourPeerID, -1); err != nil { // we subscribe at our proprer deductible search adresse.
		return err
	}
	if err := ps.PartnersSubscribeEvents(ctx); err != nil {
		return nil
	}
	return nil
}

func (ps *PubSubService) SearchSubscribeEvents(
	ctx context.Context,
	priv crypto.PrivKey,
	dt tools.DataType,
	id string,
	payload []byte,
) error {
	return ps.subscribeEvents(ctx, &dt, CREATE, id, 10)
}

func (ps *PubSubService) PartnersSubscribeEvents(
	ctx context.Context,
) error {
	access := oclib.NewRequestAdmin(oclib.LibDataEnum(oclib.PEER), "", "", []string{}, nil)
	peers := access.Search(nil, fmt.Sprintf("%v", peer.PARTNER.EnumIndex()), false)

	for _, p := range peers.Data {
		loadedPeer := access.LoadOne(p.GetID())
		rp := loadedPeer.ToPeer()
		if rp == nil {
			continue
		}
		if err := ps.PartnerSubscribeEvents(ctx, rp.PeerID); err != nil {
			return err
		}
	}
	return nil
}

func (ps *PubSubService) PartnerSubscribeEvents(
	ctx context.Context,
	peerID string,
) error {
	if peerID == "" {
		return errors.New("should discover a particular peer")
	}
	if err := ps.subscribeEvents(ctx, nil, CREATE, peerID, -1); err != nil {
		return err
	}
	if err := ps.subscribeEvents(ctx, nil, UPDATE, peerID, -1); err != nil {
		return err
	}
	if err := ps.subscribeEvents(ctx, nil, DELETE, peerID, -1); err != nil {
		return err
	}
	return nil
}

// generic function to subscribe to DHT flow of event
func (ps *PubSubService) subscribeEvents(
	ctx context.Context,
	dt *tools.DataType,
	action PubSubAction,
	peerID string,
	timeout int,
) error {
	// define a name app.action#peerID
	name := "oc-catalog." + action.String() + "#" + peerID
	if dt != nil { // if a datatype is precised then : app.action.datatype#peerID
		name = "oc-catalog." + action.String() + "." + (*dt).String() + "#" + peerID
	}
	topic, err := ps.PS.Join(name) // find out the topic
	if err != nil {
		return err
	}

	sub, err := topic.Subscribe() // then subscribe to it
	if err != nil {
		return err
	}
	ps.mutex.Lock() // add safely in cache your subscription.
	ps.Subscription = append(ps.Subscription, name)
	ps.mutex.Unlock()

	// launch loop waiting for results.
	go ps.waitResults(ctx, sub, name, timeout)

	return nil
}

func (ps *PubSubService) waitResults(ctx context.Context, sub *pubsub.Subscription, topicName string, timeout int) {
	logger := oclib.GetLogger()
	defer ctx.Done()
	for {
		ps.mutex.Lock()                                   // check safely if cache is actually notified subscribed to topic
		if !slices.Contains(ps.Subscription, topicName) { // if not kill the loop.
			break
		}
		ps.mutex.Unlock()
		// if still subscribed -> wait for new message
		var cancel context.CancelFunc
		if timeout != -1 {
			ctx, cancel = context.WithTimeout(ctx, time.Duration(timeout)*time.Second)
			defer cancel()
		}
		msg, err := sub.Next(ctx)
		if err != nil {
			if errors.Is(err, context.DeadlineExceeded) {
				// timeout hit, no message before deadline kill subsciption.
				ps.mutex.Lock()
				subs := []string{}
				for _, ss := range ps.Subscription {
					if ss != topicName {
						subs = append(subs, ss)
					}
				}
				ps.Subscription = subs
				ps.mutex.Unlock()
				return
			}
			continue
		}
		var evt Event
		if err := json.Unmarshal(msg.Data, &evt); err != nil { // map to event
			continue
		}

		access := oclib.NewRequestAdmin(oclib.LibDataEnum(oclib.PEER), "", "", []string{}, nil)
		peers := access.Search(nil, evt.From, false)
		if len(peers.Data) > 0 { // then we check if the peer is friendly or not : Partner or None...
			if err := ps.processEventPeerKnown(ctx, peers, evt, topicName); err != nil {
				logger.Err(err)
			}
		} else {
			tools.NewNATSCaller().SetNATSPub(tools.PEER_DISCOVERY, map[string]string{
				"peer_id": evt.From,
			})
			time.Sleep(30 * time.Second)
			peers = access.Search(nil, evt.From, false)
			if len(peers.Data) > 0 { // if found... ok... if not found ignore
				if err := ps.processEventPeerKnown(ctx, peers, evt, topicName); err != nil {
					logger.Err(err)
				}
			}
		}
	}
}

func (ps *PubSubService) processEventPeerKnown(ctx context.Context, peers oclib.LibDataShallow, event Event, topicName string) error {
	if err := ps.verifyPeer(peers, event); err != nil {
		return err
	}
	ps.handleEvent(ctx, topicName, event)
	return nil
}

func (pc *PubSubService) verifyPeer(peers oclib.LibDataShallow, event Event) error {
	access := oclib.NewRequestAdmin(oclib.LibDataEnum(oclib.PEER), "", "", []string{}, nil)
	loadedPeer := access.LoadOne(peers.Data[0].GetID())
	rp := loadedPeer.ToPeer()
	if rp == nil || rp.Relation == peer.BLACKLIST { // if peer is blacklisted... quit...
		return errors.New("peer is blacklisted")
	}
	pubKey, err := PubKeyFromString(rp.PublicKey) // extract pubkey from pubkey str
	if err != nil {
		return errors.New("pubkey is malformed")
	}
	data, _ := json.Marshal(event.rawEvent())               // extract byte from raw event excluding signature.
	if ok, _ := pubKey.Verify(data, event.Signature); !ok { // then verify if pubkey sign this message...
		return errors.New("check signature failed")
	}
	return nil
}

func PubKeyFromString(s string) (crypto.PubKey, error) {
	data, err := base64.StdEncoding.DecodeString(s)
	if err != nil {
		return nil, err
	}

	return crypto.UnmarshalPublicKey(data)
}