waku-org/go-waku

package library

import (
    "context"
    "encoding/json"
    "errors"
    "time"

    "github.com/libp2p/go-libp2p/core/peer"
    "github.com/waku-org/go-waku/waku/v2/protocol"
    "github.com/waku-org/go-waku/waku/v2/protocol/filter"
    "github.com/waku-org/go-waku/waku/v2/protocol/subscription"
    "github.com/waku-org/go-waku/waku/v2/utils"
)

type filterArgument struct {
    PubsubTopic   string   `json:"pubsubTopic,omitempty"`
    ContentTopics []string `json:"contentTopics,omitempty"`
}

func toContentFilter(filterJSON string) (protocol.ContentFilter, error) {
    var f filterArgument
    err := json.Unmarshal([]byte(filterJSON), &f)
    if err != nil {
        return protocol.ContentFilter{}, err
    }

    return protocol.ContentFilter{
        PubsubTopic:   f.PubsubTopic,
        ContentTopics: protocol.NewContentTopicSet(f.ContentTopics...),
    }, nil
}

type subscribeResult struct {
    Subscriptions []*subscription.SubscriptionDetails `json:"subscriptions"`
    Error         string                              `json:"error,omitempty"`
}

// FilterSubscribe is used to create a subscription to a filter node to receive messages
func FilterSubscribe(instance *WakuInstance, filterJSON string, peerID string, ms int) (string, error) {
    cf, err := toContentFilter(filterJSON)
    if err != nil {
        return "", err
    }

    if err := validateInstance(instance, MustBeStarted); err != nil {
        return "", err
    }

    var ctx context.Context
    var cancel context.CancelFunc

    if ms > 0 {
        ctx, cancel = context.WithTimeout(instance.ctx, time.Duration(int(ms))*time.Millisecond)
        defer cancel()
    } else {
        ctx = instance.ctx
    }

    var fOptions []filter.FilterSubscribeOption
    if peerID != "" {
        p, err := peer.Decode(peerID)
        if err != nil {
            return "", err
        }
        fOptions = append(fOptions, filter.WithPeer(p))
    } else {
        fOptions = append(fOptions, filter.WithAutomaticPeerSelection())
    }

    subscriptions, err := instance.node.FilterLightnode().Subscribe(ctx, cf, fOptions...)
    if err != nil && subscriptions == nil {
        return "", err
    }

    for _, subscriptionDetails := range subscriptions {
        go func(subscriptionDetails *subscription.SubscriptionDetails) {
            defer utils.LogOnPanic()
            for envelope := range subscriptionDetails.C {
                send(instance, "message", toSubscriptionMessage(envelope))
            }
        }(subscriptionDetails)
    }
    var subResult subscribeResult
    subResult.Subscriptions = subscriptions
    if err != nil {
        subResult.Error = err.Error()
    }

    return marshalJSON(subResult)
}

// FilterPing is used to determine if a peer has an active subscription
func FilterPing(instance *WakuInstance, peerID string, ms int) error {
    if err := validateInstance(instance, MustBeStarted); err != nil {
        return err
    }

    var ctx context.Context
    var cancel context.CancelFunc

    if ms > 0 {
        ctx, cancel = context.WithTimeout(instance.ctx, time.Duration(int(ms))*time.Millisecond)
        defer cancel()
    } else {
        ctx = instance.ctx
    }

    var pID peer.ID
    var err error
    if peerID != "" {
        pID, err = peer.Decode(peerID)
        if err != nil {
            return err
        }
    } else {
        return errors.New("peerID is required")
    }

    return instance.node.FilterLightnode().Ping(ctx, pID)
}

// FilterUnsubscribe is used to remove a filter criteria from an active subscription with a filter node
func FilterUnsubscribe(instance *WakuInstance, filterJSON string, peerID string, ms int) error {
    cf, err := toContentFilter(filterJSON)
    if err != nil {
        return err
    }

    if err := validateInstance(instance, MustBeStarted); err != nil {
        return err
    }

    var ctx context.Context
    var cancel context.CancelFunc

    if ms > 0 {
        ctx, cancel = context.WithTimeout(instance.ctx, time.Duration(int(ms))*time.Millisecond)
        defer cancel()
    } else {
        ctx = instance.ctx
    }

    var fOptions []filter.FilterSubscribeOption
    if peerID != "" {
        p, err := peer.Decode(peerID)
        if err != nil {
            return err
        }
        fOptions = append(fOptions, filter.WithPeer(p))
    } else {
        return errors.New("peerID is required")
    }

    result, err := instance.node.FilterLightnode().Unsubscribe(ctx, cf, fOptions...)
    if err != nil {
        return err
    }

    errs := result.Errors()
    if len(errs) == 0 {
        return nil
    }
    return errs[0].Err
}

type unsubscribeAllResult struct {
    PeerID string `json:"peerID"`
    Error  string `json:"error"`
}

// FilterUnsubscribeAll is used to remove an active subscription to a peer. If no peerID is defined, it will stop all active filter subscriptions
func FilterUnsubscribeAll(instance *WakuInstance, peerID string, ms int) (string, error) {
    if err := validateInstance(instance, MustBeStarted); err != nil {
        return "", err
    }

    var ctx context.Context
    var cancel context.CancelFunc

    if ms > 0 {
        ctx, cancel = context.WithTimeout(instance.ctx, time.Duration(int(ms))*time.Millisecond)
        defer cancel()
    } else {
        ctx = instance.ctx
    }

    var fOptions []filter.FilterSubscribeOption
    if peerID != "" {
        p, err := peer.Decode(peerID)
        if err != nil {
            return "", err
        }
        fOptions = append(fOptions, filter.WithPeer(p))
    } else {
        fOptions = append(fOptions, filter.UnsubscribeAll())
    }

    result, err := instance.node.FilterLightnode().UnsubscribeAll(ctx, fOptions...)
    if err != nil {
        return "", err
    }

    var unsubscribeResult []unsubscribeAllResult

    for _, err := range result.Errors() {
        ur := unsubscribeAllResult{
            PeerID: err.PeerID.String(),
        }
        if err.Err != nil {
            ur.Error = err.Err.Error()
        }
        unsubscribeResult = append(unsubscribeResult, ur)
    }

    return marshalJSON(unsubscribeResult)
}