diff --git a/logos_delivery/messaging/rest_api/client.nim b/logos_delivery/messaging/rest_api/client.nim index 51de21952..c202b3f1d 100644 --- a/logos_delivery/messaging/rest_api/client.nim +++ b/logos_delivery/messaging/rest_api/client.nim @@ -39,3 +39,17 @@ proc messagingPostMessagesV1*( ): RestResponse[MessagingSendResponse] {. rest, endpoint: "/messaging/v1/messages", meth: HttpMethod.MethodPost .} + +proc messagingGetSendEventsV1*(): RestResponse[seq[SendStatus]] {. + rest, endpoint: "/messaging/v1/events/send", meth: HttpMethod.MethodGet +.} + +proc messagingGetSendEventsByIdV1*( + requestId: string +): RestResponse[SendStatus] {. + rest, endpoint: "/messaging/v1/events/send/{requestId}", meth: HttpMethod.MethodGet +.} + +proc messagingGetReceivedMessagesV1*(): RestResponse[seq[ReceivedMessageRecord]] {. + rest, endpoint: "/messaging/v1/events/received", meth: HttpMethod.MethodGet +.} diff --git a/logos_delivery/messaging/rest_api/event_cache.nim b/logos_delivery/messaging/rest_api/event_cache.nim new file mode 100644 index 000000000..3f6cb914a --- /dev/null +++ b/logos_delivery/messaging/rest_api/event_cache.nim @@ -0,0 +1,117 @@ +## In-memory cache backing the messaging event REST endpoints. +## +## REST is a poll-based client/server surface, so the interactive MessagingClient +## events must be buffered here for later observation: +## * send-related events (sent / propagated / error) grouped by request id +## * received messages +## +## Both surfaces are evict-after-poll: a GET returns the buffered data and clears +## it. A generous overflow cap bounds memory if nobody polls. All access happens +## on the single chronos event loop (broker listeners + REST handlers), so the +## synchronous (no-await) ops below need no locking. + +{.push raises: [].} + +import std/[tables, deques, options] +import results +import logos_delivery/waku/waku_core/time +import ./types + +const + DefaultMaxReceived* = 50 ## Received messages kept between polls (spec default). + DefaultMaxSendRequests* = 10_000 + ## Overflow guard: distinct request ids retained between polls. + +type MessagingEventCache* = ref object + # Send events grouped by request id, with FIFO insertion order for overflow + # eviction. Cleared on poll. + sendByReqId: Table[string, SendStatus] + sendOrder: Deque[string] + maxSendRequests: int + # Received messages, bounded ring. Cleared on poll. + received: Deque[ReceivedMessageRecord] + maxReceived: int + +proc new*( + T: type MessagingEventCache, + maxReceived = DefaultMaxReceived, + maxSendRequests = DefaultMaxSendRequests, +): MessagingEventCache = + MessagingEventCache( + sendByReqId: initTable[string, SendStatus](), + sendOrder: initDeque[string](), + maxSendRequests: maxSendRequests, + received: initDeque[ReceivedMessageRecord](), + maxReceived: maxReceived, + ) + +proc recordSend*( + self: MessagingEventCache, + requestId: string, + messageHash: string, + kind: SendEventKind, + error = "", +) = + ## Append a send event to its request id's timeline, creating the entry (and + ## evicting the oldest request id past the overflow cap) as needed. + let record = SendEventRecord( + kind: kind, + messageHash: messageHash, + error: error, + timestamp: getNowInNanosecondTime(), + ) + + if not self.sendByReqId.hasKey(requestId): + self.sendByReqId[requestId] = SendStatus(requestId: requestId, events: @[]) + self.sendOrder.addLast(requestId) + + while self.sendOrder.len > self.maxSendRequests: + let evicted = self.sendOrder.popFirst() + self.sendByReqId.del(evicted) + + self.sendByReqId.withValue(requestId, status): + status.events.add(record) + +proc recordReceived*( + self: MessagingEventCache, messageHash: string, message: MessagingMessage +) = + ## Buffer a received message, dropping the oldest past the ring capacity. + self.received.addLast( + ReceivedMessageRecord( + messageHash: messageHash, message: message, timestamp: getNowInNanosecondTime() + ) + ) + + while self.received.len > self.maxReceived: + discard self.received.popFirst() + +proc pollAllSend*(self: MessagingEventCache): seq[SendStatus] = + ## Return all buffered send statuses and clear the store (evict-after-poll). + for reqId in self.sendOrder: + self.sendByReqId.withValue(reqId, status): + result.add(status[]) + self.sendByReqId.clear() + self.sendOrder.clear() + +proc pollSend*(self: MessagingEventCache, requestId: string): Opt[SendStatus] = + ## Return one request id's send status and remove it (evict-after-poll). + var status: SendStatus + if not self.sendByReqId.pop(requestId, status): + return Opt.none(SendStatus) + + # Deque has no random removal; rebuild order without the polled id. + var rebuilt = initDeque[string]() + for reqId in self.sendOrder: + if reqId != requestId: + rebuilt.addLast(reqId) + self.sendOrder = rebuilt + + return Opt.some(status) + +proc pollReceived*(self: MessagingEventCache): seq[ReceivedMessageRecord] = + ## Return buffered received messages (oldest first) and clear (evict-after-poll). + for record in self.received: + result.add(record) + self.received.clear() + +{.pop.} diff --git a/logos_delivery/messaging/rest_api/handlers.nim b/logos_delivery/messaging/rest_api/handlers.nim index e083e6600..3bfdf8ca4 100644 --- a/logos_delivery/messaging/rest_api/handlers.nim +++ b/logos_delivery/messaging/rest_api/handlers.nim @@ -11,7 +11,9 @@ import logos_delivery/messaging/api/subscription, logos_delivery/messaging/api/send, logos_delivery/api/types, - ./types + logos_delivery/api/events/messaging_client_events, + ./types, + ./event_cache export types @@ -22,12 +24,47 @@ logScope: const ROUTE_MESSAGING_SUBSCRIPTIONSV1* = "/messaging/v1/subscriptions" const ROUTE_MESSAGING_MESSAGESV1* = "/messaging/v1/messages" +const ROUTE_MESSAGING_EVENTS_SENDV1* = "/messaging/v1/events/send" +const ROUTE_MESSAGING_EVENTS_SEND_BY_IDV1* = "/messaging/v1/events/send/{requestId}" +const ROUTE_MESSAGING_EVENTS_RECEIVEDV1* = "/messaging/v1/events/received" + +proc installEventListeners(brokerCtx: BrokerContext, cache: MessagingEventCache) = + ## Buffers the MessagingClient events into `cache` so the poll-based REST + ## endpoints can observe them. Listeners live for the node's lifetime (the + ## captured `cache` keeps them and their data alive); no teardown is wired. + discard MessageSentEvent.listen( + brokerCtx, + proc(evt: MessageSentEvent): Future[void] {.async: (raises: []).} = + cache.recordSend($evt.requestId, evt.messageHash, SendEventKind.Sent), + ) + + discard MessagePropagatedEvent.listen( + brokerCtx, + proc(evt: MessagePropagatedEvent): Future[void] {.async: (raises: []).} = + cache.recordSend($evt.requestId, evt.messageHash, SendEventKind.Propagated), + ) + + discard MessageErrorEvent.listen( + brokerCtx, + proc(evt: MessageErrorEvent): Future[void] {.async: (raises: []).} = + cache.recordSend($evt.requestId, evt.messageHash, SendEventKind.Error, evt.error), + ) + + discard MessageReceivedEvent.listen( + brokerCtx, + proc(evt: MessageReceivedEvent): Future[void] {.async: (raises: []).} = + cache.recordReceived(evt.messageHash, toMessagingMessage(evt.message)), + ) 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. + # Event observability: buffer send/received events for the poll-based GETs. + let eventCache = MessagingEventCache.new() + installEventListeners(client.waku.brokerCtx, eventCache) + router.api(MethodOptions, ROUTE_MESSAGING_SUBSCRIPTIONSV1) do() -> RestApiResponse: return RestApiResponse.ok() @@ -83,16 +120,57 @@ proc installMessagingApiHandlers*(router: var RestRouter, client: MessagingClien error "An error occurred while building the json response", error = error return RestApiResponse.internalServerError($error) + #### Event observability endpoints (poll-based, evict-after-poll) + + router.api(MethodOptions, ROUTE_MESSAGING_EVENTS_SENDV1) do() -> RestApiResponse: + return RestApiResponse.ok() + + router.api(MethodGet, ROUTE_MESSAGING_EVENTS_SENDV1) do() -> RestApiResponse: + ## Returns all buffered send events grouped by request id, then clears them. + let data = eventCache.pollAllSend() + return RestApiResponse.jsonResponse(data, status = Http200).valueOr: + error "An error occurred while building the json response", error = error + return RestApiResponse.internalServerError($error) + + router.api(MethodOptions, ROUTE_MESSAGING_EVENTS_SEND_BY_IDV1) do( + requestId: string + ) -> RestApiResponse: + return RestApiResponse.ok() + + router.api(MethodGet, ROUTE_MESSAGING_EVENTS_SEND_BY_IDV1) do( + requestId: string + ) -> RestApiResponse: + ## Returns the buffered send events for one request id, then removes them. + let reqId = requestId.valueOr: + return RestApiResponse.badRequest("Invalid requestId") + + let status = eventCache.pollSend(reqId).valueOr: + return RestApiResponse.notFound("No send events for requestId: " & reqId) + + return RestApiResponse.jsonResponse(status, status = Http200).valueOr: + error "An error occurred while building the json response", error = error + return RestApiResponse.internalServerError($error) + + router.api(MethodOptions, ROUTE_MESSAGING_EVENTS_RECEIVEDV1) do() -> RestApiResponse: + return RestApiResponse.ok() + + router.api(MethodGet, ROUTE_MESSAGING_EVENTS_RECEIVEDV1) do() -> RestApiResponse: + ## Returns buffered received messages (up to the cache capacity, oldest + ## first), then clears them — optimized for polling. + let data = eventCache.pollReceived() + 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" + if not client.waku.restServer.isNil(): + # 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 index 5145635cc..31067c48e 100644 --- a/logos_delivery/messaging/rest_api/types.nim +++ b/logos_delivery/messaging/rest_api/types.nim @@ -138,3 +138,156 @@ proc readValue*( reader.raiseUnexpectedValue("Field `requestId` is missing") value = MessagingSendResponse(requestId: requestId.get()) + +#### Event observability DTOs +## +## Send-related events (sent / propagated / error) are grouped per request id; +## received messages are cached for polling. Both are populated by the broker +## listeners installed in the messaging REST handlers. + +type + SendEventKind* {.pure.} = enum + Sent = "sent" + Propagated = "propagated" + Error = "error" + + SendEventRecord* = object + kind*: SendEventKind + messageHash*: string + error*: string ## populated only for `Error` + timestamp*: int64 ## nanoseconds, stamped when cached + + SendStatus* = object ## All send events observed so far for a single request id. + requestId*: string + events*: seq[SendEventRecord] + + ReceivedMessageRecord* = object + messageHash*: string + message*: MessagingMessage + timestamp*: int64 ## nanoseconds, stamped when cached + +proc toMessagingMessage*(msg: WakuMessage): MessagingMessage = + MessagingMessage( + payload: base64.encode(msg.payload), + contentTopic: msg.contentTopic, + ephemeral: some(msg.ephemeral), + meta: + if msg.meta.len > 0: + some(base64.encode(msg.meta)) + else: + none(Base64String), + ) + +#### Event DTO serialization + +proc writeValue*( + writer: var JsonWriter[RestJson], value: SendEventRecord +) {.raises: [IOError].} = + writer.beginRecord() + writer.writeField("kind", $value.kind) + writer.writeField("messageHash", value.messageHash) + if value.error.len > 0: + writer.writeField("error", value.error) + writer.writeField("timestamp", value.timestamp) + writer.endRecord() + +proc writeValue*( + writer: var JsonWriter[RestJson], value: SendStatus +) {.raises: [IOError].} = + writer.beginRecord() + writer.writeField("requestId", value.requestId) + writer.writeField("events", value.events) + writer.endRecord() + +proc writeValue*( + writer: var JsonWriter[RestJson], value: ReceivedMessageRecord +) {.raises: [IOError].} = + writer.beginRecord() + writer.writeField("messageHash", value.messageHash) + writer.writeField("message", value.message) + writer.writeField("timestamp", value.timestamp) + writer.endRecord() + +proc readValue*( + reader: var JsonReader[RestJson], value: var SendEventKind +) {.raises: [SerializationError, IOError].} = + let s = reader.readValue(string) + case s + of "sent": + value = SendEventKind.Sent + of "propagated": + value = SendEventKind.Propagated + of "error": + value = SendEventKind.Error + else: + reader.raiseUnexpectedValue("Invalid send event kind: " & s) + +proc readValue*( + reader: var JsonReader[RestJson], value: var SendEventRecord +) {.raises: [SerializationError, IOError].} = + var + kind = none(SendEventKind) + messageHash = "" + error = "" + timestamp = int64(0) + + for fieldName in readObjectFields(reader): + case fieldName + of "kind": + kind = some(reader.readValue(SendEventKind)) + of "messageHash": + messageHash = reader.readValue(string) + of "error": + error = reader.readValue(string) + of "timestamp": + timestamp = reader.readValue(int64) + else: + unrecognizedFieldWarning(value) + + if kind.isNone(): + reader.raiseUnexpectedValue("Field `kind` is missing") + + value = SendEventRecord( + kind: kind.get(), messageHash: messageHash, error: error, timestamp: timestamp + ) + +proc readValue*( + reader: var JsonReader[RestJson], value: var SendStatus +) {.raises: [SerializationError, IOError].} = + var + requestId = "" + events: seq[SendEventRecord] = @[] + + for fieldName in readObjectFields(reader): + case fieldName + of "requestId": + requestId = reader.readValue(string) + of "events": + events = reader.readValue(seq[SendEventRecord]) + else: + unrecognizedFieldWarning(value) + + value = SendStatus(requestId: requestId, events: events) + +proc readValue*( + reader: var JsonReader[RestJson], value: var ReceivedMessageRecord +) {.raises: [SerializationError, IOError].} = + var + messageHash = "" + message = MessagingMessage() + timestamp = int64(0) + + for fieldName in readObjectFields(reader): + case fieldName + of "messageHash": + messageHash = reader.readValue(string) + of "message": + message = reader.readValue(MessagingMessage) + of "timestamp": + timestamp = reader.readValue(int64) + else: + unrecognizedFieldWarning(value) + + value = ReceivedMessageRecord( + messageHash: messageHash, message: message, timestamp: timestamp + )