WIP: messaging client REST endpoints

logos_delivery/logos_delivery.nim

@@ -39,6 +39,8 @@ import logos_delivery/messaging/[messaging_client, messaging_client_lifecycle]
 export messaging_client
 import logos_delivery/messaging/api/[subscription, send]
 export subscription, send
+import logos_delivery/messaging/rest_api/handlers as messaging_rest_api
+export messaging_rest_api
 import logos_delivery/api/events/messaging_client_events
 export messaging_client_events
 import logos_delivery/api/conf/messaging_conf
@@ -165,6 +167,10 @@ proc start*(self: LogosDelivery): Future[Result[void, string]] {.async.} =
   if not self.messagingClient.isNil():
     self.messagingClient.start().isOkOr:
       return err("failed to start MessagingClient: " & error)
+    # Mount the messaging REST endpoints onto the kernel's REST router (no-op if
+    # REST is disabled). Done here rather than in MessagingClient.start so the
+    # core messaging module need not depend on the REST layer above it.
+    self.messagingClient.mountRestApi()
 
   if not self.reliableChannelManager.isNil():
     self.reliableChannelManager.start().isOkOr:

logos_delivery/messaging/rest_api/client.nim

@@ -0,0 +1,41 @@
+{.push raises: [].}
+
+import chronicles, json_serialization, presto/[route, client, common]
+import
+  logos_delivery/waku/rest_api/endpoint/serdes,
+  logos_delivery/waku/rest_api/endpoint/rest_serdes,
+  logos_delivery/api/types,
+  ./types
+
+export types
+
+logScope:
+  topics = "messaging rest client"
+
+proc encodeBytes*(
+    value: seq[ContentTopic], contentType: string
+): RestResult[seq[byte]] =
+  return encodeBytesOf(value, contentType)
+
+proc encodeBytes*(
+    value: MessagingPostMessageRequest, contentType: string
+): RestResult[seq[byte]] =
+  return encodeBytesOf(value, contentType)
+
+proc messagingPostSubscriptionsV1*(
+  body: seq[ContentTopic]
+): RestResponse[string] {.
+  rest, endpoint: "/messaging/v1/subscriptions", meth: HttpMethod.MethodPost
+.}
+
+proc messagingDeleteSubscriptionsV1*(
+  body: seq[ContentTopic]
+): RestResponse[string] {.
+  rest, endpoint: "/messaging/v1/subscriptions", meth: HttpMethod.MethodDelete
+.}
+
+proc messagingPostMessagesV1*(
+  body: MessagingPostMessageRequest
+): RestResponse[MessagingSendResponse] {.
+  rest, endpoint: "/messaging/v1/messages", meth: HttpMethod.MethodPost
+.}

logos_delivery/messaging/rest_api/handlers.nim

@@ -0,0 +1,98 @@
+{.push raises: [].}
+
+import chronos, chronicles, results, json_serialization, json_serialization/std/options
+import presto/[route, common]
+import
+  logos_delivery/waku/waku,
+  logos_delivery/waku/rest_api/endpoint/serdes,
+  logos_delivery/waku/rest_api/endpoint/responses,
+  logos_delivery/waku/rest_api/endpoint/rest_serdes,
+  logos_delivery/messaging/messaging_client,
+  logos_delivery/messaging/api/subscription,
+  logos_delivery/messaging/api/send,
+  logos_delivery/api/types,
+  ./types
+
+export types
+
+logScope:
+  topics = "messaging rest api"
+
+#### Routes
+
+const ROUTE_MESSAGING_SUBSCRIPTIONSV1* = "/messaging/v1/subscriptions"
+const ROUTE_MESSAGING_MESSAGESV1* = "/messaging/v1/messages"
+
+proc installMessagingApiHandlers*(router: var RestRouter, client: MessagingClient) =
+  ## Mounts the MessagingClient subscribe / unsubscribe / send operations as
+  ## REST endpoints onto the given (kernel-owned) router. Subscriptions are
+  ## keyed by content topic, matching the messaging layer's content-topic API.
+
+  router.api(MethodOptions, ROUTE_MESSAGING_SUBSCRIPTIONSV1) do() -> RestApiResponse:
+    return RestApiResponse.ok()
+
+  router.api(MethodPost, ROUTE_MESSAGING_SUBSCRIPTIONSV1) do(
+    contentBody: Option[ContentBody]
+  ) -> RestApiResponse:
+    ## Subscribes the messaging client to a list of content topics.
+    let req: seq[ContentTopic] = decodeRequestBody[seq[ContentTopic]](contentBody).valueOr:
+      return error
+
+    for contentTopic in req:
+      (await client.subscribe(contentTopic)).isOkOr:
+        let errorMsg = "Subscribe failed: " & error
+        error "messaging SUBSCRIBE failed", error = errorMsg
+        return RestApiResponse.internalServerError(errorMsg)
+
+    return RestApiResponse.ok()
+
+  router.api(MethodDelete, ROUTE_MESSAGING_SUBSCRIPTIONSV1) do(
+    contentBody: Option[ContentBody]
+  ) -> RestApiResponse:
+    ## Unsubscribes the messaging client from a list of content topics.
+    let req: seq[ContentTopic] = decodeRequestBody[seq[ContentTopic]](contentBody).valueOr:
+      return error
+
+    for contentTopic in req:
+      client.unsubscribe(contentTopic).isOkOr:
+        let errorMsg = "Unsubscribe failed: " & error
+        error "messaging UNSUBSCRIBE failed", error = errorMsg
+        return RestApiResponse.internalServerError(errorMsg)
+
+    return RestApiResponse.ok()
+
+  router.api(MethodOptions, ROUTE_MESSAGING_MESSAGESV1) do() -> RestApiResponse:
+    return RestApiResponse.ok()
+
+  router.api(MethodPost, ROUTE_MESSAGING_MESSAGESV1) do(
+    contentBody: Option[ContentBody]
+  ) -> RestApiResponse:
+    ## Sends a message through the messaging client, returning the request id.
+    let req: MessagingMessage = decodeRequestBody[MessagingMessage](contentBody).valueOr:
+      return error
+
+    let envelope = req.toMessageEnvelope().valueOr:
+      return RestApiResponse.badRequest("Invalid message: " & error)
+
+    let requestId = (await client.send(envelope)).valueOr:
+      error "messaging SEND failed", error = error
+      return RestApiResponse.internalServerError("Send failed: " & error)
+
+    let data = MessagingSendResponse(requestId: $requestId)
+    return RestApiResponse.jsonResponse(data, status = Http200).valueOr:
+      error "An error occurred while building the json response", error = error
+      return RestApiResponse.internalServerError($error)
+
+proc mountRestApi*(client: MessagingClient) =
+  ## Mounts the messaging REST endpoints onto the kernel-owned REST router, if
+  ## the REST server is enabled. Called by the `LogosDelivery` concentrator
+  ## after the messaging layer has started. Lives here (not in the core
+  ## `messaging_client` module) so the core need not depend on the REST layer
+  ## above it — that would form an import cycle.
+  if client.waku.restServer.isNil():
+    return
+  # The BTree route table is ref-backed, so mutating the copied router persists
+  # (same pattern as the waku REST builder).
+  var router = client.waku.restServer.router
+  installMessagingApiHandlers(router, client)
+  info "Mounted messaging REST API endpoints"

logos_delivery/messaging/rest_api/types.nim

@@ -0,0 +1,140 @@
+{.push raises: [].}
+
+import
+  std/[sets, strformat],
+  chronicles,
+  results,
+  json_serialization,
+  json_serialization/std/options,
+  presto/[route, client, common]
+import
+  logos_delivery/waku/common/base64,
+  logos_delivery/waku/rest_api/endpoint/serdes,
+  logos_delivery/api/types
+
+export types
+
+#### Types
+
+type MessagingMessage* = object
+  ## REST wire representation of a `MessageEnvelope`. `payload` is base64.
+  payload*: Base64String
+  contentTopic*: ContentTopic
+  ephemeral*: Option[bool]
+  meta*: Option[Base64String]
+
+type
+  MessagingPostMessageRequest* = MessagingMessage
+
+  MessagingSendResponse* = object
+    ## Returned by the send endpoint; correlates with `MessageSentEvent` /
+    ## `MessageErrorEvent`.
+    requestId*: string
+
+#### Type conversion
+
+proc toMessageEnvelope*(msg: MessagingMessage): Result[MessageEnvelope, string] =
+  let
+    payload = ?msg.payload.decode()
+    meta = ?msg.meta.get(Base64String("")).decode()
+
+  return ok(
+    MessageEnvelope(
+      contentTopic: msg.contentTopic,
+      payload: payload,
+      ephemeral: msg.ephemeral.get(false),
+      meta: meta,
+    )
+  )
+
+#### Serialization and deserialization
+
+proc writeValue*(
+    writer: var JsonWriter[RestJson], value: MessagingMessage
+) {.raises: [IOError].} =
+  writer.beginRecord()
+  writer.writeField("payload", value.payload)
+  writer.writeField("contentTopic", value.contentTopic)
+  if value.ephemeral.isSome():
+    writer.writeField("ephemeral", value.ephemeral.get())
+  if value.meta.isSome():
+    writer.writeField("meta", value.meta.get())
+  writer.endRecord()
+
+proc readValue*(
+    reader: var JsonReader[RestJson], value: var MessagingMessage
+) {.raises: [SerializationError, IOError].} =
+  var
+    payload = none(Base64String)
+    contentTopic = none(ContentTopic)
+    ephemeral = none(bool)
+    meta = none(Base64String)
+
+  var keys = initHashSet[string]()
+  for fieldName in readObjectFields(reader):
+    # Check for repeated keys
+    if keys.containsOrIncl(fieldName):
+      let err =
+        try:
+          fmt"Multiple `{fieldName}` fields found"
+        except CatchableError:
+          "Multiple fields with the same name found"
+      reader.raiseUnexpectedField(err, "MessagingMessage")
+
+    case fieldName
+    of "payload":
+      payload = some(reader.readValue(Base64String))
+    of "contentTopic":
+      contentTopic = some(reader.readValue(ContentTopic))
+    of "ephemeral":
+      ephemeral = some(reader.readValue(bool))
+    of "meta":
+      meta = some(reader.readValue(Base64String))
+    else:
+      unrecognizedFieldWarning(value)
+
+  if payload.isNone() or isEmptyOrWhitespace(string(payload.get())):
+    reader.raiseUnexpectedValue("Field `payload` is missing or empty")
+
+  if contentTopic.isNone() or contentTopic.get().isEmptyOrWhitespace():
+    reader.raiseUnexpectedValue("Field `contentTopic` is missing or empty")
+
+  value = MessagingMessage(
+    payload: payload.get(),
+    contentTopic: contentTopic.get(),
+    ephemeral: ephemeral,
+    meta: meta,
+  )
+
+proc writeValue*(
+    writer: var JsonWriter[RestJson], value: MessagingSendResponse
+) {.raises: [IOError].} =
+  writer.beginRecord()
+  writer.writeField("requestId", value.requestId)
+  writer.endRecord()
+
+proc readValue*(
+    reader: var JsonReader[RestJson], value: var MessagingSendResponse
+) {.raises: [SerializationError, IOError].} =
+  var requestId = none(string)
+
+  var keys = initHashSet[string]()
+  for fieldName in readObjectFields(reader):
+    if keys.containsOrIncl(fieldName):
+      let err =
+        try:
+          fmt"Multiple `{fieldName}` fields found"
+        except CatchableError:
+          "Multiple fields with the same name found"
+      reader.raiseUnexpectedField(err, "MessagingSendResponse")
+
+    case fieldName
+    of "requestId":
+      requestId = some(reader.readValue(string))
+    else:
+      unrecognizedFieldWarning(value)
+
+  if requestId.isNone():
+    reader.raiseUnexpectedValue("Field `requestId` is missing")
+
+  value = MessagingSendResponse(requestId: requestId.get())