Feat : New Relay rest API for autosharding

cmd/waku/server/rest/debug.go

@@ -21,7 +21,7 @@ type InfoReply struct {
 }
 
 const routeDebugInfoV1 = "/debug/v1/info"
-const routeDebugVersionV1 = "/debug/v1/info"
+const routeDebugVersionV1 = "/debug/v1/version"
 
 func NewDebugService(node *node.WakuNode, m *chi.Mux) *DebugService {
  d := &DebugService{

cmd/waku/server/rest/relay.go

@@ -5,6 +5,7 @@
  "encoding/json"
  "errors"
  "net/http"
+ "net/url"
  "strings"
  "sync"
 
@@ -20,6 +21,9 @@
 const routeRelayV1Subscriptions = "/relay/v1/subscriptions"
 const routeRelayV1Messages = "/relay/v1/messages/{topic}"
 
+const routeRelayV1AutoSubscriptions = "/relay/v1/auto/subscriptions"
+const routeRelayV1AutoMessages = "/relay/v1/auto/messages"
+
 // RelayService represents the REST service for WakuRelay
 type RelayService struct {
  node *node.WakuNode
@@ -50,6 +54,14 @@
  m.Get(routeRelayV1Messages, s.getV1Messages)
  m.Post(routeRelayV1Messages, s.postV1Message)
 
+ m.Post(routeRelayV1AutoSubscriptions, s.postV1AutoSubscriptions)
+ m.Delete(routeRelayV1AutoSubscriptions, s.deleteV1AutoSubscriptions)
+
+ m.Route(routeRelayV1AutoMessages, func(r chi.Router) {
+ r.Get("/{contentTopic}", s.getV1AutoMessages)
+ r.Post("/", s.postV1AutoMessage)
+ })
+
  return s
 }
 
@@ -215,3 +227,110 @@
 
  writeErrOrResponse(w, err, true)
 }
+
+func (r *RelayService) deleteV1AutoSubscriptions(w http.ResponseWriter, req *http.Request) {
+ var cTopics []string
+ decoder := json.NewDecoder(req.Body)
+ if err := decoder.Decode(&cTopics); err != nil {
+ w.WriteHeader(http.StatusBadRequest)
+ return
+ }
+ defer req.Body.Close()
+
+ err := r.node.Relay().Unsubscribe(req.Context(), protocol.NewContentFilter("", cTopics...))
+ if err != nil {
+ r.log.Error("unsubscribing from topics", zap.Strings("contentTopics", cTopics), zap.Error(err))
+ }
+
+ writeErrOrResponse(w, err, true)
+}
+
+func (r *RelayService) postV1AutoSubscriptions(w http.ResponseWriter, req *http.Request) {
+ var cTopics []string
+ decoder := json.NewDecoder(req.Body)
+ if err := decoder.Decode(&cTopics); err != nil {
+ w.WriteHeader(http.StatusBadRequest)
+ return
+ }
+ defer req.Body.Close()
+
+ var err error
+ _, err = r.node.Relay().Subscribe(r.node.Relay().Context(), protocol.NewContentFilter("", cTopics...))
+ if err != nil {
+ r.log.Error("subscribing to topics", zap.Strings("contentTopics", cTopics), zap.Error(err))
+ }
+
+ if err != nil {
+ w.WriteHeader(http.StatusBadRequest)
+ _, err := w.Write([]byte(err.Error()))
+ r.log.Error("writing response", zap.Error(err))
+ } else {
+ w.WriteHeader(http.StatusOK)
+ }
+
+}
+
+func (r *RelayService) getV1AutoMessages(w http.ResponseWriter, req *http.Request) {
+ cTopic := chi.URLParam(req, "contentTopic")
+ if cTopic == "" {
+ w.WriteHeader(http.StatusBadRequest)
+ _, err := w.Write([]byte("contentTopic is required"))
+ r.log.Error("writing response", zap.Error(err))
+ return
+ }
+ cTopic, err := url.QueryUnescape(cTopic)
+ if err != nil {
+ w.WriteHeader(http.StatusBadRequest)
+ _, err = w.Write([]byte("invalid contentTopic format"))
+ r.log.Error("writing response", zap.Error(err))
+ return
+ }
+
+ sub, err := r.node.Relay().GetSubscription(cTopic)
+ if err != nil {
+ w.WriteHeader(http.StatusNotFound)
+ _, err = w.Write([]byte("not subscribed to topic"))
+ r.log.Error("writing response", zap.Error(err))
+ return
+ }
+ var response []*pb.WakuMessage
+ select {
+ case msg := <-sub.Ch:
+ response = append(response, msg.Message())
+ default:
+ break
+ }
+
+ writeErrOrResponse(w, nil, response)
+}
+
+func (r *RelayService) postV1AutoMessage(w http.ResponseWriter, req *http.Request) {
+
+ var message *pb.WakuMessage
+ decoder := json.NewDecoder(req.Body)
+ if err := decoder.Decode(&message); err != nil {
+ r.log.Error("decoding message failure", zap.Error(err))
+ w.WriteHeader(http.StatusBadRequest)
+ return
+ }
+ defer req.Body.Close()
+ var err error
+ if err = server.AppendRLNProof(r.node, message); err != nil {
+ writeErrOrResponse(w, err, nil)
+ return
+ }
+
+ _, err = r.node.Relay().Publish(req.Context(), message)
+ if err != nil {
+ r.log.Error("publishing message", zap.Error(err))
+ }
+
+ if err != nil {
+ w.WriteHeader(http.StatusBadRequest)
+ _, err := w.Write([]byte(err.Error()))
+ r.log.Error("writing response", zap.Error(err))
+ } else {
+ w.WriteHeader(http.StatusOK)
+ }
+
+}

cmd/waku/server/rest/relay_api.yaml

@@ -106,6 +106,103 @@ paths:
  '5XX':
  description: Unexpected error.
 
+ /relay/v1/auto/messages/{contentTopic}: # Note the plural in messages
+ get: # get_waku_v2_relay_v1_auto_messages
+ summary: Get the latest messages on the polled topic
+ description: Get a list of messages that were received on a subscribed Content topic after the last time this method was called.
+ operationId: getMessagesByTopic
+ tags:
+ - relay
+ parameters:
+ - in: path
+ name: contentTopic # Note the name is the same as in the path
+ required: true
+ schema:
+ type: string
+ description: The user ID
+ responses:
+ '200':
+ description: The latest messages on the polled topic.
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/RelayGetMessagesResponse'
+ '4XX':
+ description: Bad request.
+ '5XX':
+ description: Unexpected error.
+
+ /relay/v1/auto/messages: # Note the plural in messages
+ post: # post_waku_v2_relay_v1_auto_message
+ summary: Publish a message to be relayed
+ description: Publishes a message to be relayed on a Content topic.
+ operationId: postMessagesToTopic
+ tags:
+ - relay
+ requestBody:
+ content:
+ application/json:
+ schema:
+ $ref: '#/components/schemas/RelayPostMessagesRequest'
+ responses:
+ '200':
+ description: OK
+ '4XX':
+ description: Bad request.
+ '5XX':
+ description: Unexpected error.
+
+ /relay/v1/auto/subscriptions:
+ post: # post_waku_v2_relay_v1_auto_subscriptions
+ summary: Subscribe a node to an array of topics
+ description: Subscribe a node to an array of Content topics.
+ operationId: postSubscriptions
+ tags:
+ - relay
+ requestBody:
+ content:
+ application/json:
+ schema:
+ type array:
+ items:
+ $ref: '#/components/schemas/ContentTopic'
+ responses:
+ '200':
+ description: OK
+ content:
+ text/plain:
+ schema:
+ type: string
+ '4XX':
+ description: Bad request.
+ '5XX':
+ description: Unexpected error.
+
+ delete: # delete_waku_v2_relay_v1_auto_subscriptions
+ summary: Unsubscribe a node from an array of topics
+ description: Unsubscribe a node from an array of Content topics.
+ operationId: deleteSubscriptions
+ tags:
+ - relay
+ requestBody:
+ content:
+ application/json:
+ schema:
+ type array:
+ items:
+ $ref: '#/components/schemas/ContentTopic'
+ responses:
+ '200':
+ description: OK
+ content:
+ text/plain:
+ schema:
+ type: string
+ '4XX':
+ description: Bad request.
+ '5XX':
+ description: Unexpected error.
+
 components:
  schemas:
  PubSubTopic:
@@ -145,4 +242,4 @@ components:
  type: array
  items:
  $ref: '#/components/schemas/PubSubTopic'
-  
+

cmd/waku/server/rpc/relay.go

@@ -164,11 +164,7 @@ func (r *RelayService) PostV1AutoMessage(req *http.Request, args *RelayAutoMessa
  r.log.Error("publishing message", zap.Error(err))
  return err
  }
- if msg.ContentTopic == "" {
- err := fmt.Errorf("content-topic cannot be empty")
- r.log.Error("publishing message", zap.Error(err))
- return err
- }
+
  if err = server.AppendRLNProof(r.node, msg); err != nil {
  return err
  }