diff --git a/logos_delivery/logos_delivery.nim b/logos_delivery/logos_delivery.nim index 35e9f6a97..463b0b343 100644 --- a/logos_delivery/logos_delivery.nim +++ b/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: diff --git a/logos_delivery/messaging/rest_api/client.nim b/logos_delivery/messaging/rest_api/client.nim new file mode 100644 index 000000000..51de21952 --- /dev/null +++ b/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 +.} diff --git a/logos_delivery/messaging/rest_api/handlers.nim b/logos_delivery/messaging/rest_api/handlers.nim new file mode 100644 index 000000000..e083e6600 --- /dev/null +++ b/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" diff --git a/logos_delivery/messaging/rest_api/types.nim b/logos_delivery/messaging/rest_api/types.nim new file mode 100644 index 000000000..5145635cc --- /dev/null +++ b/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())