diff --git a/cmd/waku/rest/relay.go b/cmd/waku/rest/relay.go index 4cce1daa9..484fef0b2 100644 --- a/cmd/waku/rest/relay.go +++ b/cmd/waku/rest/relay.go @@ -3,6 +3,7 @@ package rest import ( "context" "encoding/json" + "errors" "net/http" "strings" "sync" @@ -15,9 +16,10 @@ import ( "go.uber.org/zap" ) -const ROUTE_RELAY_SUBSCRIPTIONSV1 = "/relay/v1/subscriptions" -const ROUTE_RELAY_MESSAGESV1 = "/relay/v1/messages/{topic}" +const routeRelayV1Subscriptions = "/relay/v1/subscriptions" +const routeRelayV1Messages = "/relay/v1/messages/{topic}" +// RelayService represents the REST service for WakuRelay type RelayService struct { node *node.WakuNode cancel context.CancelFunc @@ -31,6 +33,7 @@ type RelayService struct { runner *runnerService } +// NewRelayService returns an instance of RelayService func NewRelayService(node *node.WakuNode, m *chi.Mux, cacheCapacity int, log *zap.Logger) *RelayService { s := &RelayService{ node: node, @@ -41,10 +44,10 @@ func NewRelayService(node *node.WakuNode, m *chi.Mux, cacheCapacity int, log *za s.runner = newRunnerService(node.Broadcaster(), s.addEnvelope) - m.Post(ROUTE_RELAY_SUBSCRIPTIONSV1, s.postV1Subscriptions) - m.Delete(ROUTE_RELAY_SUBSCRIPTIONSV1, s.deleteV1Subscriptions) - m.Get(ROUTE_RELAY_MESSAGESV1, s.getV1Messages) - m.Post(ROUTE_RELAY_MESSAGESV1, s.postV1Message) + m.Post(routeRelayV1Subscriptions, s.postV1Subscriptions) + m.Delete(routeRelayV1Subscriptions, s.deleteV1Subscriptions) + m.Get(routeRelayV1Messages, s.getV1Messages) + m.Post(routeRelayV1Messages, s.postV1Message) return s } @@ -65,6 +68,7 @@ func (r *RelayService) addEnvelope(envelope *protocol.Envelope) { r.messages[envelope.PubsubTopic()] = append(r.messages[envelope.PubsubTopic()], envelope.Message()) } +// Start starts the RelayService func (r *RelayService) Start(ctx context.Context) { ctx, cancel := context.WithCancel(ctx) r.cancel = cancel @@ -80,6 +84,7 @@ func (r *RelayService) Start(ctx context.Context) { r.runner.Start(ctx) } +// Stop stops the RelayService func (r *RelayService) Stop() { if r.cancel == nil { return @@ -187,11 +192,23 @@ func (d *RelayService) postV1Message(w http.ResponseWriter, r *http.Request) { var err error if topic == "" { - _, err = d.node.Relay().Publish(r.Context(), message) - } else { - _, err = d.node.Relay().PublishToTopic(r.Context(), message, strings.Replace(topic, "\n", "", -1)) + topic = relay.DefaultWakuTopic } + if !d.node.Relay().IsSubscribed(topic) { + writeErrOrResponse(w, errors.New("not subscribed to pubsubTopic"), nil) + return + } + + if node.SupportsRLN { + err = d.node.RLNRelay().AppendRLNProof(message, d.node.Timesource().Now()) + if err != nil { + writeErrOrResponse(w, errors.New("not subscribed to pubsubTopic"), nil) + return + } + } + + _, err = d.node.Relay().PublishToTopic(r.Context(), message, strings.Replace(topic, "\n", "", -1)) if err != nil { d.log.Error("publishing message", zap.Error(err)) } diff --git a/cmd/waku/rpc/relay.go b/cmd/waku/rpc/relay.go index 8fce74cae..9f14d4e9e 100644 --- a/cmd/waku/rpc/relay.go +++ b/cmd/waku/rpc/relay.go @@ -1,6 +1,7 @@ package rpc import ( + "errors" "fmt" "net/http" "sync" @@ -12,6 +13,7 @@ import ( "go.uber.org/zap" ) +// RelayService represents the JSON RPC service for WakuRelay type RelayService struct { node *node.WakuNode @@ -24,19 +26,23 @@ type RelayService struct { runner *runnerService } +// RelayMessageArgs represents the requests used for posting messages type RelayMessageArgs struct { Topic string `json:"topic,omitempty"` Message *RPCWakuMessage `json:"message,omitempty"` } +// TopicArgs represents the lists of topics to use when subscribing / unsubscribing type TopicsArgs struct { Topics []string `json:"topics,omitempty"` } +// TopicArgs represents a request that contains a single topic type TopicArgs struct { Topic string `json:"topic,omitempty"` } +// NewRelayService returns an instance of RelayService func NewRelayService(node *node.WakuNode, cacheCapacity int, log *zap.Logger) *RelayService { s := &RelayService{ node: node, @@ -66,6 +72,7 @@ func (r *RelayService) addEnvelope(envelope *protocol.Envelope) { r.messages[envelope.PubsubTopic()] = append(r.messages[envelope.PubsubTopic()], envelope.Message()) } +// Start starts the RelayService func (r *RelayService) Start() { r.messagesMutex.Lock() // Node may already be subscribed to some topics when Relay API handlers are installed. Let's add these @@ -78,18 +85,34 @@ func (r *RelayService) Start() { r.runner.Start() } +// Stop stops the RelayService func (r *RelayService) Stop() { r.runner.Stop() } +// PostV1Message is invoked when the json rpc request uses the post_waku_v2_relay_v1_message method func (r *RelayService) PostV1Message(req *http.Request, args *RelayMessageArgs, reply *SuccessReply) error { var err error - if args.Topic == "" { - _, err = r.node.Relay().Publish(req.Context(), args.Message.toProto()) - } else { - _, err = r.node.Relay().PublishToTopic(req.Context(), args.Message.toProto(), args.Topic) + topic := relay.DefaultWakuTopic + if args.Topic != "" { + topic = args.Topic } + + if !r.node.Relay().IsSubscribed(topic) { + return errors.New("not subscribed to pubsubTopic") + } + + msg := args.Message.toProto() + + if node.SupportsRLN { + err = r.node.RLNRelay().AppendRLNProof(msg, r.node.Timesource().Now()) + if err != nil { + return err + } + } + + _, err = r.node.Relay().PublishToTopic(req.Context(), args.Message.toProto(), topic) if err != nil { r.log.Error("publishing message", zap.Error(err)) return err @@ -99,6 +122,7 @@ func (r *RelayService) PostV1Message(req *http.Request, args *RelayMessageArgs, return nil } +// PostV1Subscription is invoked when the json rpc request uses the post_waku_v2_relay_v1_subscription method func (r *RelayService) PostV1Subscription(req *http.Request, args *TopicsArgs, reply *SuccessReply) error { ctx := req.Context() for _, topic := range args.Topics { @@ -129,6 +153,7 @@ func (r *RelayService) PostV1Subscription(req *http.Request, args *TopicsArgs, r return nil } +// DeleteV1Subscription is invoked when the json rpc request uses the delete_waku_v2_relay_v1_subscription method func (r *RelayService) DeleteV1Subscription(req *http.Request, args *TopicsArgs, reply *SuccessReply) error { ctx := req.Context() for _, topic := range args.Topics { @@ -145,6 +170,7 @@ func (r *RelayService) DeleteV1Subscription(req *http.Request, args *TopicsArgs, return nil } +// GetV1Messages is invoked when the json rpc request uses the get_waku_v2_relay_v1_messages method func (r *RelayService) GetV1Messages(req *http.Request, args *TopicArgs, reply *MessagesReply) error { r.messagesMutex.Lock() defer r.messagesMutex.Unlock() diff --git a/waku/v2/node/wakunode2_no_rln.go b/waku/v2/node/wakunode2_no_rln.go index 53620cda5..6356fc4fe 100644 --- a/waku/v2/node/wakunode2_no_rln.go +++ b/waku/v2/node/wakunode2_no_rln.go @@ -5,6 +5,9 @@ package node import "context" +const SupportsRLN = false + +// RLNRelay is used to access any operation related to Waku RLN protocol func (w *WakuNode) RLNRelay() RLNRelay { return nil } diff --git a/waku/v2/node/wakunode2_rln.go b/waku/v2/node/wakunode2_rln.go index 456ce396a..bcf69330a 100644 --- a/waku/v2/node/wakunode2_rln.go +++ b/waku/v2/node/wakunode2_rln.go @@ -16,6 +16,8 @@ import ( r "github.com/waku-org/go-zerokit-rln/rln" ) +const SupportsRLN = true + // RLNRelay is used to access any operation related to Waku RLN protocol func (w *WakuNode) RLNRelay() RLNRelay { return w.rlnRelay