Add event poll for messaging rest with cache mechanism

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
+.}

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.}

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"

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
+  )