refactor(consumers): reuse envelope.hash in archive/filter/REST/FFI; share filter buffer

archive.handleMessage and filter.handleMessage gain envelope-native entry points
(reusing the precomputed hash) with thin (topic,msg) compat overloads for tests.
Filter pushToPeer takes a shared `ref seq[byte]` so the encoded buffer is not
deep-copied per subscribed peer. REST messageCacheHandler and the FFI relay
handler adopt the envelope; JsonMessageEvent.new reuses the passed-in hash.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

library/events/json_message_event.nim

@@ -69,9 +69,15 @@ type JsonMessageEvent* = ref object of JsonEvent
   messageHash*: string
   wakuMessage*: JsonMessage
 
-proc new*(T: type JsonMessageEvent, pubSubTopic: string, msg: WakuMessage): T =
+proc new*(
+    T: type JsonMessageEvent,
+    pubSubTopic: string,
+    msg: WakuMessage,
+    msgHash: WakuMessageHash,
+): T =
   # Returns a WakuMessage event as indicated in
   # https://github.com/vacp2p/rfc/blob/master/content/docs/rfcs/36/README.md#jsonmessageevent-type
+  # `msgHash` is the precomputed message hash (reused from the inbound envelope).
 
   var payload = newSeq[byte](len(msg.payload))
   if len(msg.payload) != 0:
@@ -85,8 +91,6 @@ proc new*(T: type JsonMessageEvent, pubSubTopic: string, msg: WakuMessage): T =
   if len(msg.proof) != 0:
     copyMem(addr proof[0], unsafeAddr msg.proof[0], len(msg.proof))
 
-  let msgHash = computeMessageHash(pubSubTopic, msg)
-
   return JsonMessageEvent(
     eventType: "message",
     pubSubTopic: pubSubTopic,
@@ -102,5 +106,9 @@ proc new*(T: type JsonMessageEvent, pubSubTopic: string, msg: WakuMessage): T =
     ),
   )
 
+proc new*(T: type JsonMessageEvent, pubSubTopic: string, msg: WakuMessage): T =
+  ## Convenience overload computing the hash for callers without one.
+  JsonMessageEvent.new(pubSubTopic, msg, computeMessageHash(pubSubTopic, msg))
+
 method `$`*(jsonMessage: JsonMessageEvent): string =
   $(%*jsonMessage)

library/kernel_api/protocols/relay_api.nim

@@ -78,9 +78,9 @@ proc waku_relay_subscribe(
     pubSubTopic: cstring,
 ) {.ffi.} =
   proc onReceivedMessage(ctx: ptr FFIContext[LogosDelivery]): WakuRelayHandler =
-    return proc(pubsubTopic: PubsubTopic, msg: WakuMessage) {.async.} =
+    return proc(envelope: WakuEnvelope) {.async.} =
       callEventCallback(ctx, "onReceivedMessage"):
-        $JsonMessageEvent.new(pubsubTopic, msg)
+        $JsonMessageEvent.new(envelope.pubsubTopic, envelope.msg, envelope.hash)
 
   (
     await ctx.myLib[].waku.relaySubscribe(

logos_delivery/waku/rest_api/handlers.nim

@@ -34,5 +34,5 @@ proc defaultDiscoveryHandler*(
 ### Message Cache
 
 proc messageCacheHandler*(cache: MessageCache): WakuRelayHandler =
-  return proc(pubsubTopic: string, msg: WakuMessage): Future[void] {.async, closure.} =
-    cache.addMessage(pubsubTopic, msg)
+  return proc(envelope: WakuEnvelope): Future[void] {.async, closure.} =
+    cache.addMessage(envelope.pubsubTopic, envelope.msg)

logos_delivery/waku/waku_archive/archive.nim

@@ -95,10 +95,10 @@ proc new*(
 
   return ok(archive)
 
-proc handleMessage*(
-    self: WakuArchive, pubsubTopic: PubsubTopic, msg: WakuMessage
-) {.async.} =
-  let msgHash = computeMessageHash(pubsubTopic, msg)
+proc handleMessage*(self: WakuArchive, envelope: WakuEnvelope) {.async.} =
+  let pubsubTopic = envelope.pubsubTopic
+  let msg = envelope.msg
+  let msgHash = envelope.hash
   let msgHashHex = msgHash.to0xHex()
 
   trace "handling message",
@@ -144,6 +144,14 @@ proc handleMessage*(
     timestamp = msg.timestamp,
     insertDuration = insertDuration
 
+proc handleMessage*(
+    self: WakuArchive, pubsubTopic: PubsubTopic, msg: WakuMessage
+) {.async.} =
+  ## Convenience overload building the envelope (and its hash) for callers that
+  ## don't already have one (e.g. tests, REST). The relay dispatch path uses the
+  ## envelope overload directly to avoid re-hashing.
+  await self.handleMessage(WakuEnvelope.init(pubsubTopic, msg))
+
 proc syncMessageIngress*(
     self: WakuArchive,
     msgHash: WakuMessageHash,

logos_delivery/waku/waku_filter_v2/protocol.nim

@@ -168,8 +168,10 @@ proc handleSubscribeRequest*(
   return FilterSubscribeResponse.ok(request.requestId)
 
 proc pushToPeer(
-    wf: WakuFilter, peerId: PeerId, buffer: seq[byte]
+    wf: WakuFilter, peerId: PeerId, buffer: ref seq[byte]
 ): Future[Result[void, string]] {.async.} =
+  ## `buffer` is shared by reference across all target peers (encoded once in
+  ## `pushToPeers`) so no per-peer async-closure copy of the payload happens.
   info "pushing message to subscribed peer", peerId = shortLog(peerId)
 
   let stream = (
@@ -178,21 +180,21 @@ proc pushToPeer(
     error "pushToPeer failed", error
     return err("pushToPeer failed: " & $error)
 
-  await stream.writeLp(buffer)
+  await stream.writeLp(buffer[])
 
   info "published successful", peerId = shortLog(peerId), stream
   waku_service_network_bytes.inc(
-    amount = buffer.len().int64, labelValues = [WakuFilterPushCodec, "out"]
+    amount = buffer[].len().int64, labelValues = [WakuFilterPushCodec, "out"]
   )
 
   return ok()
 
 proc pushToPeers(
-    wf: WakuFilter, peers: seq[PeerId], messagePush: MessagePush
+    wf: WakuFilter, peers: seq[PeerId], messagePush: MessagePush, msgHash: string
 ) {.async.} =
+  ## `msgHash` is the precomputed 0x-hex hash of the pushed message (reused from
+  ## the inbound envelope; not recomputed here).
   let targetPeerIds = peers.mapIt(shortLog(it))
-  let msgHash =
-    messagePush.pubsubTopic.computeMessageHash(messagePush.wakuMessage).to0xHex()
 
   ## it's also refresh expire of msghash, that's why update cache every time, even if it has a value.
   if wf.messageCache.put(msgHash, Moment.now()):
@@ -210,7 +212,8 @@ proc pushToPeers(
       target_peer_ids = targetPeerIds,
       msg_hash = msgHash
 
-    let bufferToPublish = messagePush.encode().buffer
+    let bufferToPublish = new(seq[byte])
+    bufferToPublish[] = messagePush.encode().buffer
     var pushFuts: seq[Future[Result[void, string]]]
 
     for peerId in peers:
@@ -240,10 +243,10 @@ proc maintainSubscriptions*(wf: WakuFilter) {.async.} =
   waku_filter_subscriptions.set(wf.subscriptions.peersSubscribed.len.float64)
 
 const MessagePushTimeout = 20.seconds
-proc handleMessage*(
-    wf: WakuFilter, pubsubTopic: PubsubTopic, message: WakuMessage
-) {.async.} =
-  let msgHash = computeMessageHash(pubsubTopic, message).to0xHex()
+proc handleMessage*(wf: WakuFilter, envelope: WakuEnvelope) {.async.} =
+  let pubsubTopic = envelope.pubsubTopic
+  let message = envelope.msg
+  let msgHash = envelope.hash.to0xHex()
 
   info "handling message",
     pubsubTopic = pubsubTopic, contentTopic = message.contentTopic, msg_hash = msgHash
@@ -263,7 +266,7 @@ proc handleMessage*(
 
     let messagePush = MessagePush(pubsubTopic: pubsubTopic, wakuMessage: message)
 
-    if not await wf.pushToPeers(subscribedPeers, messagePush).withTimeout(
+    if not await wf.pushToPeers(subscribedPeers, messagePush, msgHash).withTimeout(
       MessagePushTimeout
     ):
       error "timed out pushing message to peers",
@@ -287,6 +290,14 @@ proc handleMessage*(
       # Duration in seconds with millisecond precision floating point
   waku_filter_handle_message_duration_seconds.observe(handleMessageDurationSec)
 
+proc handleMessage*(
+    wf: WakuFilter, pubsubTopic: PubsubTopic, message: WakuMessage
+) {.async.} =
+  ## Convenience overload building the envelope (and its hash) for callers that
+  ## don't already have one (e.g. tests). The relay dispatch path uses the
+  ## envelope overload directly to avoid re-hashing.
+  await wf.handleMessage(WakuEnvelope.init(pubsubTopic, message))
+
 proc initProtocolHandler(wf: WakuFilter) =
   proc handler(conn: Connection, proto: string) {.async: (raises: [CancelledError]).} =
     info "filter subscribe request handler triggered",