feat: REST APIs discovery handlers (#2109)

apps/wakunode2/app.nim

@@ -31,7 +31,7 @@ import
   ../../waku/node/peer_manager/peer_store/waku_peer_storage,
   ../../waku/node/peer_manager/peer_store/migrations as peer_store_sqlite_migrations,
   ../../waku/waku_api/message_cache,
-  ../../waku/waku_api/cache_handlers,
+  ../../waku/waku_api/handlers,
   ../../waku/waku_api/rest/server,
   ../../waku/waku_api/rest/debug/handlers as rest_debug_api,
   ../../waku/waku_api/rest/relay/handlers as rest_relay_api,
@@ -679,18 +679,38 @@ proc startRestServer(app: App, address: ValidIpAddress, port: Port, conf: WakuNo
     rest_legacy_filter_api.installLegacyFilterRestApiHandlers(server.router, app.node, legacyFilterCache)
 
     let filterCache = rest_filter_api.MessageCache.init()
-    rest_filter_api.installFilterRestApiHandlers(server.router, app.node, filterCache)
+
+    let filterDiscoHandler = 
+      if app.wakuDiscv5.isSome():
+        some(defaultDiscoveryHandler(app.wakuDiscv5.get(), Filter))
+      else: none(DiscoveryHandler)
+
+    rest_filter_api.installFilterRestApiHandlers(
+      server.router,
+      app.node,
+      filterCache,
+      filterDiscoHandler,
+    )
   else:
     notInstalledTab["filter"] = "/filter endpoints are not available. Please check your configuration: --filternode"
 
-
   ## Store REST API
-  installStoreApiHandlers(server.router, app.node)
+  let storeDiscoHandler = 
+    if app.wakuDiscv5.isSome():
+      some(defaultDiscoveryHandler(app.wakuDiscv5.get(), Store))
+    else: none(DiscoveryHandler)
+
+  installStoreApiHandlers(server.router, app.node, storeDiscoHandler)
 
   ## Light push API
   if conf.lightpushnode  != "" and
      app.node.wakuLightpushClient != nil:
-    rest_lightpush_api.installLightPushRequestHandler(server.router, app.node)
+    let lightDiscoHandler = 
+      if app.wakuDiscv5.isSome():
+        some(defaultDiscoveryHandler(app.wakuDiscv5.get(), Lightpush))
+      else: none(DiscoveryHandler)
+
+    rest_lightpush_api.installLightPushRequestHandler(server.router, app.node, lightDiscoHandler)
   else:
     notInstalledTab["lightpush"] = "/lightpush endpoints are not available. Please check your configuration: --lightpushnode"
 

tests/wakunode_rest/test_rest_store.nim

@@ -513,7 +513,7 @@ procSuite "Waku v2 Rest API - Store":
       response.status == 412
       $response.contentType == $MIMETYPE_TEXT
       response.data.messages.len == 0
-      response.data.error_message.get == "Missing known store-peer node"
+      response.data.error_message.get == NoPeerNoDiscError.errobj.message
 
     # Now add the storenode from "config"
     node.peerManager.addServicePeer(remotePeerInfo,

waku/waku_api/cache_handlers.nim

@@ -1,23 +0,0 @@
-when (NimMajor, NimMinor) < (1, 4):
-  {.push raises: [Defect].}
-else:
-  {.push raises: [].}
-
-import
-  chronos,
-  chronicles
-import
-  ../waku_relay,
-  ../waku_core,
-  ./message_cache
-
-##### Message handler
-
-proc messageCacheHandler*(cache: MessageCache[string]): WakuRelayHandler =
-  return proc(pubsubTopic: string, msg: WakuMessage): Future[void] {.async, closure.} =
-    cache.addMessage(PubSubTopic(pubsubTopic), msg)
-
-proc autoMessageCacheHandler*(cache: MessageCache[string]): WakuRelayHandler =
-  return proc(pubsubTopic: string, msg: WakuMessage): Future[void] {.async, closure.} =
-    if cache.isSubscribed(msg.contentTopic):
-      cache.addMessage(msg.contentTopic, msg)

waku/waku_api/handlers.nim

@@ -0,0 +1,50 @@
+when (NimMajor, NimMinor) < (1, 4):
+  {.push raises: [Defect].}
+else:
+  {.push raises: [].}
+
+import
+  chronos,
+  chronicles,
+  std/[options, sequtils],
+  stew/results
+import
+  ../waku_discv5,
+  ../waku_relay,
+  ../waku_core,
+  ./message_cache
+
+### Discovery
+
+type DiscoveryHandler* = proc(): Future[Result[Option[RemotePeerInfo], string]] {.async, closure.}
+
+proc defaultDiscoveryHandler*(discv5: WakuDiscoveryV5, cap: Capabilities): DiscoveryHandler =
+  proc(): Future[Result[Option[RemotePeerInfo], string]] {.async, closure.} =
+    #Discv5 is already filtering peers by shards no need to pass a predicate.
+    let findPeers = discv5.findRandomPeers()
+    
+    if not await findPeers.withTimeout(60.seconds):
+      return err("discovery process timed out!")
+
+    var peers = findPeers.read()
+
+    peers.keepItIf(it.supportsCapability(cap))
+
+    if peers.len == 0:
+      return ok(none(RemotePeerInfo))
+
+    let remotePeerInfo = peers[0].toRemotePeerInfo().valueOr:
+      return err($error)
+
+    return ok(some(remotePeerInfo))
+
+### Message Cache
+
+proc messageCacheHandler*(cache: MessageCache[string]): WakuRelayHandler =
+  return proc(pubsubTopic: string, msg: WakuMessage): Future[void] {.async, closure.} =
+    cache.addMessage(PubSubTopic(pubsubTopic), msg)
+
+proc autoMessageCacheHandler*(cache: MessageCache[string]): WakuRelayHandler =
+  return proc(pubsubTopic: string, msg: WakuMessage): Future[void] {.async, closure.} =
+    if cache.isSubscribed(msg.contentTopic):
+      cache.addMessage(msg.contentTopic, msg)

waku/waku_api/jsonrpc/relay/handlers.nim

@@ -17,7 +17,7 @@ import
   ../../../waku_rln_relay/rln/wrappers,
   ../../../waku_node,
   ../../message_cache,
-  ../../cache_handlers,
+  ../../handlers,
   ../message
 
 from std/times import getTime

waku/waku_api/rest/filter/handlers.nim

@@ -21,6 +21,7 @@ import
   ../../../waku_filter_v2/client as filter_protocol_client,
   ../../../waku_filter_v2/common as filter_protocol_type,
   ../../message_cache,
+  ../../handlers,
   ../serdes,
   ../responses,
   ./types
@@ -145,11 +146,18 @@ proc makeRestResponse(requestId: string, protocolClientRes: filter_protocol_type
 
   return resp.get()
 
-proc filterPostPutSubscriptionRequestHandler(node: WakuNode,
+const NoPeerNoDiscoError = FilterSubscribeError.serviceUnavailable(
+  "No suitable service peer & no discovery method")
+
+const NoPeerNoneFoundError = FilterSubscribeError.serviceUnavailable(
+  "No suitable service peer & none discovered")
+
+proc filterPostPutSubscriptionRequestHandler(
+  node: WakuNode,
   contentBody: Option[ContentBody],
-                                             cache: MessageCache):
+  cache: MessageCache,
-                                          Future[RestApiResponse]
+  discHandler: Option[DiscoveryHandler] = none(DiscoveryHandler),
-                                          {.async.} =
+  ): Future[RestApiResponse] {.async.} =
   ## handles any filter subscription requests, adds or modifies.
 
   let decodedBody = decodeRequestBody[FilterSubscribeRequest](contentBody)
@@ -159,14 +167,17 @@ proc filterPostPutSubscriptionRequestHandler(node: WakuNode,
 
   let req: FilterSubscribeRequest = decodedBody.value()
 
-  let peerOpt = node.peerManager.selectPeer(WakuFilterSubscribeCodec)
+  let peer = node.peerManager.selectPeer(WakuFilterSubscribeCodec).valueOr:
+    let handler = discHandler.valueOr:
+      return makeRestResponse(req.requestId, NoPeerNoDiscoError) 
 
-  if peerOpt.isNone():
+    let peerOp = (await handler()).valueOr:
-    return makeRestResponse(req.requestId, FilterSubscribeError.serviceUnavailable("No suitable peers"))
+      return RestApiResponse.internalServerError($error)
 
-  let subFut = node.filterSubscribe(req.pubsubTopic,
+    peerOp.valueOr:
-                                      req.contentFilters,
+      return makeRestResponse(req.requestId, NoPeerNoneFoundError)
-                                      peerOpt.get())
+
+  let subFut = node.filterSubscribe(req.pubsubTopic, req.contentFilters, peer)
 
   if not await subFut.withTimeout(futTimeoutForSubscriptionProcessing):
     error "Failed to subscribe to contentFilters do to timeout!"
@@ -178,29 +189,36 @@ proc filterPostPutSubscriptionRequestHandler(node: WakuNode,
 
   return makeRestResponse(req.requestId, subFut.read())
 
-proc installFilterPostSubscriptionsHandler(router: var RestRouter,
+proc installFilterPostSubscriptionsHandler(
+  router: var RestRouter,
   node: WakuNode,
-                                           cache: MessageCache) =
+  cache: MessageCache,
+  discHandler: Option[DiscoveryHandler] = none(DiscoveryHandler),
+  ) =
   router.api(MethodPost, ROUTE_FILTER_SUBSCRIPTIONS) do (contentBody: Option[ContentBody]) -> RestApiResponse:
     ## Subscribes a node to a list of contentTopics of a pubsubTopic
     debug "post", ROUTE_FILTER_SUBSCRIPTIONS, contentBody
 
-    let response = await filterPostPutSubscriptionRequestHandler(node, contentBody, cache)
+    return await filterPostPutSubscriptionRequestHandler(node, contentBody, cache, discHandler)
-    return response
 
-proc installFilterPutSubscriptionsHandler(router: var RestRouter,
+proc installFilterPutSubscriptionsHandler(
+  router: var RestRouter,
   node: WakuNode,
-                                          cache: MessageCache) =
+  cache: MessageCache,
+  discHandler: Option[DiscoveryHandler] = none(DiscoveryHandler),
+  ) =
   router.api(MethodPut, ROUTE_FILTER_SUBSCRIPTIONS) do (contentBody: Option[ContentBody]) -> RestApiResponse:
     ## Modifies a subscribtion of a node to a list of contentTopics of a pubsubTopic
     debug "put", ROUTE_FILTER_SUBSCRIPTIONS, contentBody
 
-    let response = await filterPostPutSubscriptionRequestHandler(node, contentBody, cache)
+    return await filterPostPutSubscriptionRequestHandler(node, contentBody, cache, discHandler)
-    return response
 
-proc installFilterDeleteSubscriptionsHandler(router: var RestRouter,
+proc installFilterDeleteSubscriptionsHandler(
+  router: var RestRouter,
   node: WakuNode,
-                                             cache: MessageCache) =
+  cache: MessageCache,
+  discHandler: Option[DiscoveryHandler] = none(DiscoveryHandler),
+  ) =
   router.api(MethodDelete, ROUTE_FILTER_SUBSCRIPTIONS) do (contentBody: Option[ContentBody]) -> RestApiResponse:
     ## Subscribes a node to a list of contentTopics of a PubSub topic
     debug "delete", ROUTE_FILTER_SUBSCRIPTIONS, contentBody
@@ -213,13 +231,18 @@ proc installFilterDeleteSubscriptionsHandler(router: var RestRouter,
 
     let req: FilterUnsubscribeRequest = decodedBody.value()
 
-    let peerOpt = node.peerManager.selectPeer(WakuFilterSubscribeCodec)
+    let peer = node.peerManager.selectPeer(WakuFilterSubscribeCodec).valueOr:
+      let handler = discHandler.valueOr:
+        return makeRestResponse(req.requestId, NoPeerNoDiscoError) 
 
-    if peerOpt.isNone():
+      let peerOp = (await handler()).valueOr:
-      return makeRestResponse(req.requestId,
+        return RestApiResponse.internalServerError($error)
-                FilterSubscribeError.serviceUnavailable("No suitable peers"))
+
+      peerOp.valueOr:
+        return makeRestResponse(req.requestId, NoPeerNoneFoundError)
+
+    let unsubFut = node.filterUnsubscribe(req.pubsubTopic, req.contentFilters, peer)
 
-    let unsubFut = node.filterUnsubscribe(req.pubsubTopic, req.contentFilters, peerOpt.get())
     if not await unsubFut.withTimeout(futTimeoutForSubscriptionProcessing):
       error "Failed to unsubscribe from contentFilters due to timeout!"
       return makeRestResponse(req.requestId,
@@ -233,9 +256,12 @@ proc installFilterDeleteSubscriptionsHandler(router: var RestRouter,
     # Successfully unsubscribed from all requested contentTopics
     return makeRestResponse(req.requestId, unsubFut.read())
 
-proc installFilterDeleteAllSubscriptionsHandler(router: var RestRouter,
+proc installFilterDeleteAllSubscriptionsHandler(
+  router: var RestRouter,
   node: WakuNode,
-                                             cache: MessageCache) =
+  cache: MessageCache,
+  discHandler: Option[DiscoveryHandler] = none(DiscoveryHandler),
+  ) =
   router.api(MethodDelete, ROUTE_FILTER_ALL_SUBSCRIPTIONS) do (contentBody: Option[ContentBody]) -> RestApiResponse:
     ## Subscribes a node to a list of contentTopics of a PubSub topic
     debug "delete", ROUTE_FILTER_ALL_SUBSCRIPTIONS, contentBody
@@ -248,13 +274,18 @@ proc installFilterDeleteAllSubscriptionsHandler(router: var RestRouter,
 
     let req: FilterUnsubscribeAllRequest = decodedBody.value()
 
-    let peerOpt = node.peerManager.selectPeer(WakuFilterSubscribeCodec)
+    let peer = node.peerManager.selectPeer(WakuFilterSubscribeCodec).valueOr:
+      let handler = discHandler.valueOr:
+        return makeRestResponse(req.requestId, NoPeerNoDiscoError) 
 
-    if peerOpt.isNone():
+      let peerOp = (await handler()).valueOr:
-      return makeRestResponse(req.requestId,
+        return RestApiResponse.internalServerError($error)
-                FilterSubscribeError.serviceUnavailable("No suitable peers"))
+
+      peerOp.valueOr:
+        return makeRestResponse(req.requestId, NoPeerNoneFoundError)
+
+    let unsubFut = node.filterUnsubscribeAll(peer)
     
-    let unsubFut = node.filterUnsubscribeAll(peerOpt.get())
     if not await unsubFut.withTimeout(futTimeoutForSubscriptionProcessing):
       error "Failed to unsubscribe from contentFilters due to timeout!"
       return makeRestResponse(req.requestId,
@@ -268,18 +299,26 @@ proc installFilterDeleteAllSubscriptionsHandler(router: var RestRouter,
 
 const ROUTE_FILTER_SUBSCRIBER_PING* = "/filter/v2/subscriptions/{requestId}"
 
-proc installFilterPingSubscriberHandler(router: var RestRouter,
+proc installFilterPingSubscriberHandler(
-                                        node: WakuNode) =
+  router: var RestRouter,
+  node: WakuNode,
+  discHandler: Option[DiscoveryHandler] = none(DiscoveryHandler),
+  ) =
   router.api(MethodGet, ROUTE_FILTER_SUBSCRIBER_PING) do (requestId: string) -> RestApiResponse:
     ## Checks if a node has valid subscription or not.
     debug "get", ROUTE_FILTER_SUBSCRIBER_PING, requestId
 
-    let peerOpt = node.peerManager.selectPeer(WakuFilterSubscribeCodec)
+    let peer = node.peerManager.selectPeer(WakuFilterSubscribeCodec).valueOr:
-    if peerOpt.isNone():
+      let handler = discHandler.valueOr:
-      return makeRestResponse(requestId.get(),
+        return makeRestResponse(requestId.get(), NoPeerNoDiscoError) 
-                FilterSubscribeError.serviceUnavailable("No suitable remote filter peers"))
 
-    let pingFutRes = node.wakuFilterClient.ping(peerOpt.get())
+      let peerOp = (await handler()).valueOr:
+        return RestApiResponse.internalServerError($error)
+
+      peerOp.valueOr:
+        return makeRestResponse(requestId.get(), NoPeerNoneFoundError)
+
+    let pingFutRes = node.wakuFilterClient.ping(peer)
 
     if not await pingFutRes.withTimeout(futTimeoutForSubscriptionProcessing):
       error "Failed to ping filter service peer due to timeout!"
@@ -325,12 +364,15 @@ proc installFilterGetMessagesHandler(router: var RestRouter,
 
     return resp.get()
 
-proc installFilterRestApiHandlers*(router: var RestRouter,
+proc installFilterRestApiHandlers*(
+  router: var RestRouter,
   node: WakuNode,
-                                   cache: MessageCache) =
+  cache: MessageCache,
-  installFilterPingSubscriberHandler(router, node)
+  discHandler: Option[DiscoveryHandler] = none(DiscoveryHandler),
-  installFilterPostSubscriptionsHandler(router, node, cache)
+  ) =
-  installFilterPutSubscriptionsHandler(router, node, cache)
+  installFilterPingSubscriberHandler(router, node, discHandler)
-  installFilterDeleteSubscriptionsHandler(router, node, cache)
+  installFilterPostSubscriptionsHandler(router, node, cache, discHandler)
-  installFilterDeleteAllSubscriptionsHandler(router, node, cache)
+  installFilterPutSubscriptionsHandler(router, node, cache, discHandler)
+  installFilterDeleteSubscriptionsHandler(router, node, cache, discHandler)
+  installFilterDeleteAllSubscriptionsHandler(router, node, cache, discHandler)
   installFilterGetMessagesHandler(router, node, cache)

waku/waku_api/rest/lightpush/handlers.nim

@@ -18,6 +18,7 @@ import
   ../../waku/node/peer_manager,
   ../../../waku_node,
   ../../waku/waku_lightpush,
+  ../../handlers,
   ../serdes,
   ../responses,
   ./types
@@ -29,6 +30,12 @@ logScope:
 
 const futTimeoutForPushRequestProcessing* = 5.seconds
 
+const NoPeerNoDiscoError = RestApiResponse.serviceUnavailable(
+  "No suitable service peer & no discovery method")
+
+const NoPeerNoneFoundError = RestApiResponse.serviceUnavailable(
+  "No suitable service peer & none discovered")
+
 #### Request handlers
 
 const ROUTE_LIGHTPUSH* = "/lightpush/v1/message"
@@ -50,8 +57,11 @@ func decodeRequestBody[T](contentBody: Option[ContentBody]) : Result[T, RestApiR
 
   return ok(requestResult.get())
 
-proc installLightPushRequestHandler*(router: var RestRouter,
+proc installLightPushRequestHandler*(
-                                             node: WakuNode) =
+  router: var RestRouter,
+  node: WakuNode,
+  discHandler: Option[DiscoveryHandler] = none(DiscoveryHandler),
+  ) =
 
   router.api(MethodPost, ROUTE_LIGHTPUSH) do (contentBody: Option[ContentBody]) -> RestApiResponse:
     ## Send a request to push a waku message
@@ -63,24 +73,29 @@ proc installLightPushRequestHandler*(router: var RestRouter,
       return decodedBody.error()
 
     let req: PushRequest = decodedBody.value()
-    let msg = req.message.toWakuMessage()
     
-    if msg.isErr():
+    let msg = req.message.toWakuMessage().valueOr:
       return RestApiResponse.badRequest("Invalid message: {msg.error}")
 
-    let peerOpt = node.peerManager.selectPeer(WakuLightPushCodec)
+    let peer = node.peerManager.selectPeer(WakuLightPushCodec).valueOr:
-    if peerOpt.isNone():
+      let handler = discHandler.valueOr:
-      return RestApiResponse.serviceUnavailable("No suitable remote lightpush peers")
+        return NoPeerNoDiscoError
 
-    let subFut = node.lightpushPublish(req.pubsubTopic,
+      let peerOp = (await handler()).valueOr:
-                                       msg.value(),
+        return RestApiResponse.internalServerError($error)
-                                       peerOpt.get())
+
+      peerOp.valueOr:
+        return NoPeerNoneFoundError
+
+    let subFut = node.lightpushPublish(req.pubsubTopic, msg, peer)
 
     if not await subFut.withTimeout(futTimeoutForPushRequestProcessing):
       error "Failed to request a message push due to timeout!"
       return RestApiResponse.serviceUnavailable("Push request timed out")
 
     if subFut.value().isErr():
-      return RestApiResponse.serviceUnavailable(fmt("Failed to request a message push: {subFut.value().error}"))
+      return RestApiResponse.serviceUnavailable(
+        fmt("Failed to request a message push: {subFut.value().error}")
+      )
 
     return RestApiResponse.ok()

waku/waku_api/rest/relay/handlers.nim

@@ -17,7 +17,7 @@ import
   ../../../waku_rln_relay,
   ../../../node/waku_node,
   ../../message_cache,
-  ../../cache_handlers,
+  ../../handlers,
   ../serdes,
   ../responses,
   ./types

waku/waku_api/rest/store/handlers.nim

@@ -15,6 +15,7 @@ import
   ../../../waku_store/common,
   ../../../waku_node,
   ../../../node/peer_manager,
+  ../../handlers,
   ../responses,
   ../serdes,
   ./types
@@ -26,6 +27,9 @@ logScope:
 
 const futTimeout* = 5.seconds # Max time to wait for futures
 
+const NoPeerNoDiscError* = RestApiResponse.preconditionFailed(
+            "No suitable service peer & no discovery method")
+
 # Queries the store-node with the query parameters and
 # returns a RestApiResponse that is sent back to the api client.
 proc performHistoryQuery(selfNode: WakuNode,
@@ -182,10 +186,12 @@ proc toOpt(self: Option[Result[string, cstring]]): Option[string] =
   if self.isSome() and self.get().value != "":
     return some(self.get().value)
 
-
 # Subscribes the rest handler to attend "/store/v1/messages" requests
-proc installStoreV1Handler(router: var RestRouter,
+proc installStoreApiHandlers*(
-                           node: WakuNode) =
+  router: var RestRouter,
+  node: WakuNode,
+  discHandler: Option[DiscoveryHandler] = none(DiscoveryHandler),
+  ) =
 
   # Handles the store-query request according to the passed parameters
   router.api(MethodGet,
@@ -209,18 +215,20 @@ proc installStoreV1Handler(router: var RestRouter,
     # /store/v1/messages?peerAddr=%2Fip4%2F127.0.0.1%2Ftcp%2F60001%2Fp2p%2F16Uiu2HAmVFXtAfSj4EiR7mL2KvL4EE2wztuQgUSBoj2Jx2KeXFLN\&pubsubTopic=my-waku-topic
 
     # Parse the peer address parameter
-    var parsedPeerAddr = parseUrlPeerAddr(peerAddr.toOpt())
+    let parsedPeerAddr = parseUrlPeerAddr(peerAddr.toOpt()).valueOr:
-    if not parsedPeerAddr.isOk():
+      return RestApiResponse.badRequest(error)
-      return RestApiResponse.badRequest(parsedPeerAddr.error)
 
-    var peerOpt = none(RemotePeerInfo)
+    let peerAddr = parsedPeerAddr.valueOr:
-    if parsedPeerAddr.value.isSome():
+      node.peerManager.selectPeer(WakuStoreCodec).valueOr:
-      peerOpt = parsedPeerAddr.value
+        let handler = discHandler.valueOr:
-    else:
+          return NoPeerNoDiscError
-      # The user didn't specify any store peer address.
+
-      peerOpt = node.peerManager.selectPeer(WakuStoreCodec)
+        let peerOp = (await handler()).valueOr:
-      if peerOpt.isNone():
+          return RestApiResponse.internalServerError($error)
-        return RestApiResponse.preconditionFailed("Missing known store-peer node")
+
+        peerOp.valueOr:
+          return RestApiResponse.preconditionFailed(
+            "No suitable service peer & none discovered")
 
     # Parse the rest of the parameters and create a HistoryQuery
     let histQuery = createHistoryQuery(
@@ -238,10 +246,4 @@ proc installStoreV1Handler(router: var RestRouter,
     if not histQuery.isOk():
       return RestApiResponse.badRequest(histQuery.error)
 
-    return await node.performHistoryQuery(histQuery.value,
+    return await node.performHistoryQuery(histQuery.value, peerAddr)
-                                          peerOpt.get())
-
-# Registers the Api Handlers
-proc installStoreApiHandlers*(router: var RestRouter,
-                              node: WakuNode) =
-  installStoreV1Handler(router, node)