From b95a74c6002216ff09895d08eefbfcc26f84de45 Mon Sep 17 00:00:00 2001
From: Ivan FB <ivansete@status.im>
Date: Fri, 26 Jun 2026 11:56:47 +0200
Subject: [PATCH 1/5] messaging: depend on the Waku kernel, not the raw
 WakuNode

MessagingClient sat one layer below where the documented layering
(Waku <- MessagingClient <- ReliableChannelManager) places it: it took a
raw WakuNode and reached around the Waku kernel to its internals. Make the
messaging layer hold the Waku kernel and read `waku.node` from there, so the
declared dependency matches the layering and the layer holds the kernel handle
it will later route through.

The health-monitor test hand-builds a WakuNode, so it now wraps it in a
minimal Waku (conf/stateInfo only satisfy {.requiresInit.}; the messaging
path reads neither).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 logos_delivery/logos_delivery.nim             |  2 +-
 logos_delivery/messaging/api/send.nim         | 14 +++++++++-----
 logos_delivery/messaging/api/subscription.nim |  5 +++--
 logos_delivery/messaging/messaging_client.nim | 16 ++++++++--------
 tests/node/test_wakunode_health_monitor.nim   | 18 ++++++++++++++++--
 5 files changed, 37 insertions(+), 18 deletions(-)

diff --git a/logos_delivery/logos_delivery.nim b/logos_delivery/logos_delivery.nim
index 435bb968d..e23766dba 100644
--- a/logos_delivery/logos_delivery.nim
+++ b/logos_delivery/logos_delivery.nim
@@ -97,7 +97,7 @@ proc new*(
   let waku = (await Waku.new(layerConf.waku, appCallbacks)).valueOr:
     return err("failed to create Waku: " & error)
 
-  let messagingClient = MessagingClient.new(layerConf.messaging, waku.node).valueOr:
+  let messagingClient = MessagingClient.new(layerConf.messaging, waku).valueOr:
     return err("failed to create MessagingClient: " & error)
 
   let reliableChannelManager = ReliableChannelManager.new(
diff --git a/logos_delivery/messaging/api/send.nim b/logos_delivery/messaging/api/send.nim
index cc6312471..ed909b6d0 100644
--- a/logos_delivery/messaging/api/send.nim
+++ b/logos_delivery/messaging/api/send.nim
@@ -3,6 +3,7 @@ import results, chronos, chronicles
 
 import logos_delivery/api/types
 import logos_delivery/messaging/messaging_client
+import logos_delivery/waku/waku
 import logos_delivery/waku/node/[waku_node, subscription_manager]
 import logos_delivery/messaging/delivery_service/send_service
 import logos_delivery/messaging/delivery_service/send_service/delivery_task
@@ -16,17 +17,20 @@ proc send*(
   ## id the caller can correlate with `MessageSentEvent` / `MessageErrorEvent`.
   ?self.checkApiAvailability()
 
-  let isSubbed =
-    self.node.subscriptionManager.isSubscribed(envelope.contentTopic).valueOr(false)
+  let isSubbed = self.waku.node.subscriptionManager.isSubscribed(
+    envelope.contentTopic
+  ).valueOr(false)
   if not isSubbed:
     info "Auto-subscribing to topic on send", contentTopic = envelope.contentTopic
-    self.node.subscriptionManager.subscribe(envelope.contentTopic).isOkOr:
+    self.waku.node.subscriptionManager.subscribe(envelope.contentTopic).isOkOr:
       warn "Failed to auto-subscribe", error = error
       return err("Failed to auto-subscribe before sending: " & error)
 
-  let requestId = RequestId.new(self.node.rng)
+  let requestId = RequestId.new(self.waku.node.rng)
 
-  let deliveryTask = DeliveryTask.new(requestId, envelope, self.node.brokerCtx).valueOr:
+  let deliveryTask = DeliveryTask.new(
+    requestId, envelope, self.waku.node.brokerCtx
+  ).valueOr:
     return err("MessagingClient.send: Failed to create delivery task: " & error)
 
   asyncSpawn self.sendService.send(deliveryTask)
diff --git a/logos_delivery/messaging/api/subscription.nim b/logos_delivery/messaging/api/subscription.nim
index 35b8c7e53..f8fe0bfe6 100644
--- a/logos_delivery/messaging/api/subscription.nim
+++ b/logos_delivery/messaging/api/subscription.nim
@@ -3,16 +3,17 @@ import results, chronos
 
 import logos_delivery/api/types
 import logos_delivery/messaging/messaging_client
+import logos_delivery/waku/waku
 import logos_delivery/waku/node/[waku_node, subscription_manager]
 
 proc subscribe*(
     self: MessagingClient, contentTopic: ContentTopic
 ): Future[Result[void, string]] {.async.} =
   ?self.checkApiAvailability()
-  return self.node.subscriptionManager.subscribe(contentTopic)
+  return self.waku.node.subscriptionManager.subscribe(contentTopic)
 
 proc unsubscribe*(
     self: MessagingClient, contentTopic: ContentTopic
 ): Result[void, string] =
   ?self.checkApiAvailability()
-  return self.node.subscriptionManager.unsubscribe(contentTopic)
+  return self.waku.node.subscriptionManager.unsubscribe(contentTopic)
diff --git a/logos_delivery/messaging/messaging_client.nim b/logos_delivery/messaging/messaging_client.nim
index 69c6c520b..cf2db623b 100644
--- a/logos_delivery/messaging/messaging_client.nim
+++ b/logos_delivery/messaging/messaging_client.nim
@@ -4,7 +4,7 @@
 import results, chronos
 import
   logos_delivery/api/messaging_client_api,
-  logos_delivery/waku/node/waku_node,
+  logos_delivery/waku/waku,
   logos_delivery/messaging/delivery_service/[recv_service, send_service]
 
 # Surfaces the messaging API interface (and its Message* events) to consumers.
@@ -18,19 +18,19 @@ type
     useP2PReliability*: bool
 
   MessagingClient* = ref object of IMessagingClient
-    node*: WakuNode ## Waku core driven by this layer; read by `messaging/api.nim`.
+    waku*: Waku ## The Waku kernel this layer drives; read by `messaging/api/*`.
     sendService*: SendService
     recvService*: RecvService
     started: bool
 
 proc new*(
-    T: type MessagingClient, conf: MessagingClientConf, node: WakuNode
+    T: type MessagingClient, conf: MessagingClientConf, waku: Waku
 ): Result[T, string] =
-  ## The messaging layer chains onto Waku: it drives the underlying
-  ## `WakuNode` (Waku's core) for transport while exposing its own send/recv API.
-  let sendService = ?SendService.new(conf.useP2PReliability, node)
-  let recvService = RecvService.new(node)
-  ok(T(node: node, sendService: sendService, recvService: recvService))
+  ## The messaging layer chains onto Waku: it drives the underlying Waku kernel
+  ## for transport while exposing its own send/recv API.
+  let sendService = ?SendService.new(conf.useP2PReliability, waku.node)
+  let recvService = RecvService.new(waku.node)
+  ok(T(waku: waku, sendService: sendService, recvService: recvService))
 
 proc start*(self: MessagingClient): Result[void, string] =
   if self.started:
diff --git a/tests/node/test_wakunode_health_monitor.nim b/tests/node/test_wakunode_health_monitor.nim
index acf0bc09a..148c93330 100644
--- a/tests/node/test_wakunode_health_monitor.nim
+++ b/tests/node/test_wakunode_health_monitor.nim
@@ -26,6 +26,8 @@ import
 
 import ../testlib/[wakunode, wakucore], ../waku_archive/archive_utils
 import logos_delivery/waku/node/subscription_manager
+import logos_delivery/waku/waku
+import logos_delivery/waku/factory/waku_state_info
 import logos_delivery/messaging/messaging_client
 
 const MockDLow = 4 # Mocked GossipSub DLow value
@@ -228,8 +230,14 @@ suite "Health Monitor - events":
       nodeA.mountMetadata(1, @[0'u16]).expect("Node A failed to mount metadata")
       await nodeA.start()
 
+    # MessagingClient now depends on the Waku kernel, not the raw node. Only
+    # `waku.node` is read on the messaging path; `conf`/`stateInfo` are supplied
+    # solely to satisfy Waku's {.requiresInit.} fields.
+    let waku = Waku(
+      node: nodeA, conf: defaultTestWakuConf(), stateInfo: WakuStateInfo.init(nodeA)
+    )
     let ds = MessagingClient
-      .new(MessagingClientConf(useP2PReliability: false), nodeA)
+      .new(MessagingClientConf(useP2PReliability: false), waku)
       .expect("Failed to create MessagingClient")
     ds.start().expect("Failed to start MessagingClient")
 
@@ -333,8 +341,14 @@ suite "Health Monitor - events":
       nodeA.mountMetadata(1, @[0'u16]).expect("Node A failed to mount metadata")
       await nodeA.start()
 
+    # MessagingClient now depends on the Waku kernel, not the raw node. Only
+    # `waku.node` is read on the messaging path; `conf`/`stateInfo` are supplied
+    # solely to satisfy Waku's {.requiresInit.} fields.
+    let waku = Waku(
+      node: nodeA, conf: defaultTestWakuConf(), stateInfo: WakuStateInfo.init(nodeA)
+    )
     let ds = MessagingClient
-      .new(MessagingClientConf(useP2PReliability: false), nodeA)
+      .new(MessagingClientConf(useP2PReliability: false), waku)
       .expect("Failed to create MessagingClient")
     ds.start().expect("Failed to start MessagingClient")
     let subMgr = nodeA.subscriptionManager

From dcdc958c8e3f0c18b5b00f7ec58a9b14583f829d Mon Sep 17 00:00:00 2001
From: Ivan FB <ivansete@status.im>
Date: Fri, 26 Jun 2026 12:43:31 +0200
Subject: [PATCH 2/5] adjust nph

---
 library/kernel_api/protocols/relay_api.nim   |  4 +++-
 logos_delivery/channels/reliable_channel.nim |  4 ++--
 logos_delivery/messaging/api/send.nim        | 10 ++++------
 3 files changed, 9 insertions(+), 9 deletions(-)

diff --git a/library/kernel_api/protocols/relay_api.nim b/library/kernel_api/protocols/relay_api.nim
index 528ab7ba2..7a1fe446f 100644
--- a/library/kernel_api/protocols/relay_api.nim
+++ b/library/kernel_api/protocols/relay_api.nim
@@ -122,7 +122,9 @@ proc waku_relay_publish(
     return err("Problem building the WakuMessage: " & $error)
 
   let msgHash = (
-    await ctx.myLib[].waku.relayPublish(PubsubTopic($pubSubTopic), msg, uint32(timeoutMs))
+    await ctx.myLib[].waku.relayPublish(
+      PubsubTopic($pubSubTopic), msg, uint32(timeoutMs)
+    )
   ).valueOr:
     error "PUBLISH failed", error = error
     return err(error)
diff --git a/logos_delivery/channels/reliable_channel.nim b/logos_delivery/channels/reliable_channel.nim
index dfaf96f94..d06f74e36 100644
--- a/logos_delivery/channels/reliable_channel.nim
+++ b/logos_delivery/channels/reliable_channel.nim
@@ -32,8 +32,8 @@ import ./rate_limit_manager/rate_limit_manager
 import ./encryption/encryption
 
 export
-  types, reliable_channel_manager_api, send_service, segmentation,
-  scalable_data_sync, rate_limit_manager, encryption
+  types, reliable_channel_manager_api, send_service, segmentation, scalable_data_sync,
+  rate_limit_manager, encryption
 
 const LipWireReliableChannelVersion* = "RELIABLE-CHANNEL-API/1"
   ## Wire-format spec marker for the Reliable Channel layer, as defined
diff --git a/logos_delivery/messaging/api/send.nim b/logos_delivery/messaging/api/send.nim
index ed909b6d0..ca7256b3e 100644
--- a/logos_delivery/messaging/api/send.nim
+++ b/logos_delivery/messaging/api/send.nim
@@ -17,9 +17,9 @@ proc send*(
   ## id the caller can correlate with `MessageSentEvent` / `MessageErrorEvent`.
   ?self.checkApiAvailability()
 
-  let isSubbed = self.waku.node.subscriptionManager.isSubscribed(
-    envelope.contentTopic
-  ).valueOr(false)
+  let isSubbed = self.waku.node.subscriptionManager
+    .isSubscribed(envelope.contentTopic)
+    .valueOr(false)
   if not isSubbed:
     info "Auto-subscribing to topic on send", contentTopic = envelope.contentTopic
     self.waku.node.subscriptionManager.subscribe(envelope.contentTopic).isOkOr:
@@ -28,9 +28,7 @@ proc send*(
 
   let requestId = RequestId.new(self.waku.node.rng)
 
-  let deliveryTask = DeliveryTask.new(
-    requestId, envelope, self.waku.node.brokerCtx
-  ).valueOr:
+  let deliveryTask = DeliveryTask.new(requestId, envelope, self.waku.node.brokerCtx).valueOr:
     return err("MessagingClient.send: Failed to create delivery task: " & error)
 
   asyncSpawn self.sendService.send(deliveryTask)

From 0590b2bf68b211ad5d6ffb29f5908df3e9051205 Mon Sep 17 00:00:00 2001
From: Ivan FB <ivansete@status.im>
Date: Fri, 26 Jun 2026 13:40:40 +0200
Subject: [PATCH 3/5] messaging: route api send/subscription through the Waku
 kernel

The messaging api layer reached around the Waku kernel into the libp2p
node (`waku.node.subscriptionManager`, `waku.node.rng`,
`waku.node.brokerCtx`). That breaks the declared layering: messaging
depends on the Waku *kernel* abstraction, not the raw node.

Add a content-topic subscription api on the Waku layer
(`waku/api/subscriptions.nim`) and switch the messaging send/subscription
paths onto it, plus the kernel's own `rng`/`brokerCtx` fields, so the
layer no longer touches `node` internals.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 logos_delivery/logos_delivery.nim             |  7 +++---
 logos_delivery/messaging/api/send.nim         | 12 ++++------
 logos_delivery/messaging/api/subscription.nim |  6 ++---
 logos_delivery/waku/api/subscriptions.nim     | 23 +++++++++++++++++++
 4 files changed, 35 insertions(+), 13 deletions(-)
 create mode 100644 logos_delivery/waku/api/subscriptions.nim

diff --git a/logos_delivery/logos_delivery.nim b/logos_delivery/logos_delivery.nim
index e23766dba..d0fb33b90 100644
--- a/logos_delivery/logos_delivery.nim
+++ b/logos_delivery/logos_delivery.nim
@@ -25,11 +25,12 @@ import logos_delivery/waku/waku
 export waku
 import
   logos_delivery/waku/api/[
-    topics, relay, filter, lightpush, store, peer_manager, discovery, debug, health,
-    ping,
+    topics, relay, subscriptions, filter, lightpush, store, peer_manager, discovery,
+    debug, health, ping,
   ]
 export
-  topics, relay, filter, lightpush, store, peer_manager, discovery, debug, health, ping
+  topics, relay, subscriptions, filter, lightpush, store, peer_manager, discovery,
+  debug, health, ping
 # `MessageSeenEvent` is surfaced via `export waku` (Kernel interface); the
 # remaining waku health events live here.
 import logos_delivery/waku/api/events/health_events
diff --git a/logos_delivery/messaging/api/send.nim b/logos_delivery/messaging/api/send.nim
index ca7256b3e..27c86d0ab 100644
--- a/logos_delivery/messaging/api/send.nim
+++ b/logos_delivery/messaging/api/send.nim
@@ -4,7 +4,7 @@ import results, chronos, chronicles
 import logos_delivery/api/types
 import logos_delivery/messaging/messaging_client
 import logos_delivery/waku/waku
-import logos_delivery/waku/node/[waku_node, subscription_manager]
+import logos_delivery/waku/api/subscriptions
 import logos_delivery/messaging/delivery_service/send_service
 import logos_delivery/messaging/delivery_service/send_service/delivery_task
 
@@ -17,18 +17,16 @@ proc send*(
   ## id the caller can correlate with `MessageSentEvent` / `MessageErrorEvent`.
   ?self.checkApiAvailability()
 
-  let isSubbed = self.waku.node.subscriptionManager
-    .isSubscribed(envelope.contentTopic)
-    .valueOr(false)
+  let isSubbed = self.waku.isSubscribed(envelope.contentTopic).valueOr(false)
   if not isSubbed:
     info "Auto-subscribing to topic on send", contentTopic = envelope.contentTopic
-    self.waku.node.subscriptionManager.subscribe(envelope.contentTopic).isOkOr:
+    self.waku.subscribe(envelope.contentTopic).isOkOr:
       warn "Failed to auto-subscribe", error = error
       return err("Failed to auto-subscribe before sending: " & error)
 
-  let requestId = RequestId.new(self.waku.node.rng)
+  let requestId = RequestId.new(self.waku.rng)
 
-  let deliveryTask = DeliveryTask.new(requestId, envelope, self.waku.node.brokerCtx).valueOr:
+  let deliveryTask = DeliveryTask.new(requestId, envelope, self.waku.brokerCtx).valueOr:
     return err("MessagingClient.send: Failed to create delivery task: " & error)
 
   asyncSpawn self.sendService.send(deliveryTask)
diff --git a/logos_delivery/messaging/api/subscription.nim b/logos_delivery/messaging/api/subscription.nim
index f8fe0bfe6..8893af7f1 100644
--- a/logos_delivery/messaging/api/subscription.nim
+++ b/logos_delivery/messaging/api/subscription.nim
@@ -4,16 +4,16 @@ import results, chronos
 import logos_delivery/api/types
 import logos_delivery/messaging/messaging_client
 import logos_delivery/waku/waku
-import logos_delivery/waku/node/[waku_node, subscription_manager]
+import logos_delivery/waku/api/subscriptions
 
 proc subscribe*(
     self: MessagingClient, contentTopic: ContentTopic
 ): Future[Result[void, string]] {.async.} =
   ?self.checkApiAvailability()
-  return self.waku.node.subscriptionManager.subscribe(contentTopic)
+  return self.waku.subscribe(contentTopic)
 
 proc unsubscribe*(
     self: MessagingClient, contentTopic: ContentTopic
 ): Result[void, string] =
   ?self.checkApiAvailability()
-  return self.waku.node.subscriptionManager.unsubscribe(contentTopic)
+  return self.waku.unsubscribe(contentTopic)
diff --git a/logos_delivery/waku/api/subscriptions.nim b/logos_delivery/waku/api/subscriptions.nim
new file mode 100644
index 000000000..ec73bd9bf
--- /dev/null
+++ b/logos_delivery/waku/api/subscriptions.nim
@@ -0,0 +1,23 @@
+## Waku layer API — content-topic subscription operations.
+##
+## These wrap the node's `SubscriptionManager`, which resolves each content
+## topic to its autosharding shard. They give the layers above (messaging) a
+## kernel-level entry point so they never reach into `waku.node` internals.
+{.push raises: [].}
+
+import results
+
+import logos_delivery/waku/waku
+import logos_delivery/waku/[waku_core, node/waku_node, node/subscription_manager]
+
+proc subscribe*(self: Waku, contentTopic: ContentTopic): Result[void, string] =
+  ## Subscribes to `contentTopic`, resolving its shard via autosharding.
+  return self.node.subscriptionManager.subscribe(contentTopic)
+
+proc unsubscribe*(self: Waku, contentTopic: ContentTopic): Result[void, string] =
+  ## Unsubscribes from `contentTopic`, resolving its shard via autosharding.
+  return self.node.subscriptionManager.unsubscribe(contentTopic)
+
+proc isSubscribed*(self: Waku, contentTopic: ContentTopic): Result[bool, string] =
+  ## True if the node already subscribes to `contentTopic`.
+  return self.node.subscriptionManager.isSubscribed(contentTopic)

From e4c89917279bf5b83e53b695fda0dc9a0b0fae95 Mon Sep 17 00:00:00 2001
From: Ivan FB <ivansete@status.im>
Date: Fri, 26 Jun 2026 14:03:26 +0200
Subject: [PATCH 4/5] messaging: drive delivery services through the Waku
 kernel

`SendService`/`RecvService` took a raw `WakuNode` and reached into its
internals (`wakuStoreClient`, `subscriptionManager`, `peerManager`),
which breaks the layering: the messaging layer should depend on the Waku
kernel, not the node.

Widen the Waku api surface with the operations these services need
(`storeQueryToAny`, `isStoreMounted`, `hasStorePeer`, `isContentSubscribed`,
`subscribedContentTopics`) and switch both services to hold `Waku` and
call that surface instead. The send-processor chain still pulls raw
publish handles (relay/lightpush/RLN/peer manager) from `waku.node`,
since the kernel API does not expose publishing primitives yet; this is
isolated to the constructor and flagged with a comment.

Also make `MessagingClient.new` return explicitly.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 .../recv_service/recv_service.nim             | 32 +++++++------------
 .../send_service/send_service.nim             | 31 ++++++++++--------
 logos_delivery/messaging/messaging_client.nim |  6 ++--
 logos_delivery/waku/api/store.nim             | 27 +++++++++++++++-
 logos_delivery/waku/api/subscriptions.nim     | 14 ++++++++
 5 files changed, 73 insertions(+), 37 deletions(-)

diff --git a/logos_delivery/messaging/delivery_service/recv_service/recv_service.nim b/logos_delivery/messaging/delivery_service/recv_service/recv_service.nim
index 25b35222d..1090be223 100644
--- a/logos_delivery/messaging/delivery_service/recv_service/recv_service.nim
+++ b/logos_delivery/messaging/delivery_service/recv_service/recv_service.nim
@@ -7,15 +7,9 @@ import std/[tables, sequtils, options, sets]
 import chronos, chronicles, libp2p/utility
 import brokers/broker_context
 import
-  logos_delivery/waku/[
-    waku_core,
-    waku_core/topics,
-    waku_store/client,
-    waku_store/common,
-    waku_filter_v2/client,
-    waku_node,
-    node/subscription_manager,
-  ]
+  logos_delivery/waku/[waku_core, waku_core/topics, waku_store/common],
+  logos_delivery/waku/waku,
+  logos_delivery/waku/api/[store, subscriptions]
 import
   logos_delivery/api/kernel_api, # MessageSeenEvent
   logos_delivery/api/messaging_client_api, # MessageReceivedEvent
@@ -38,7 +32,7 @@ type RecvMessage = object
 
 type RecvService* = ref object of RootObj
   brokerCtx: BrokerContext
-  node: WakuNode
+  waku: Waku
   seenMsgListener: MessageSeenEventListener
   connStatusListener: EventConnectionStatusChangeListener
 
@@ -60,7 +54,7 @@ proc getMissingMsgsFromStore(
     self: RecvService, msgHashes: seq[WakuMessageHash]
 ): Future[Result[seq[TupleHashAndMsg], string]] {.async.} =
   let storeResp: StoreQueryResponse = (
-    await self.node.wakuStoreClient.queryToAny(
+    await self.waku.storeQueryToAny(
       StoreQueryRequest(includeData: true, messageHashes: msgHashes)
     )
   ).valueOr:
@@ -85,9 +79,7 @@ proc processIncomingMessage(
   ## or if the message is a duplicate (recently-seen). Otherwise, save it as
   ## recently-seen, emit a MessageReceivedEvent, and return true.
 
-  if not self.node.subscriptionManager.isContentSubscribed(
-    pubsubTopic, message.contentTopic
-  ):
+  if not self.waku.isContentSubscribed(pubsubTopic, message.contentTopic):
     trace "skipping message as I am not subscribed",
       shard = pubsubTopic, contentTopic = message.contentTopic
     return false
@@ -108,16 +100,16 @@ proc processIncomingMessage(
 proc checkStore*(self: RecvService) {.async.} =
   ## Checks the store for messages that were not received directly and
   ## delivers them via MessageReceivedEvent.
-  if self.node.wakuStoreClient.isNil():
+  if not self.waku.isStoreMounted():
     error "recv service has no store client mounted, skipping store check"
     return
 
   self.endTimeToCheck = getNowInNanosecondTime()
 
   ## query store and deliver new recovered messages per subscribed topic
-  for pubsubTopic, contentTopics in self.node.subscriptionManager.subscribedContentTopics:
+  for (pubsubTopic, contentTopics) in self.waku.subscribedContentTopics():
     let storeResp: StoreQueryResponse = (
-      await self.node.wakuStoreClient.queryToAny(
+      await self.waku.storeQueryToAny(
         StoreQueryRequest(
           includeData: false,
           pubsubTopic: some(pubsubTopic),
@@ -171,14 +163,14 @@ proc onConnectionStatusChange(self: RecvService, status: ConnectionStatus) =
     info "recv service backfilling missed messages after coming back online"
     self.backfillHandler = self.checkStore()
 
-proc new*(T: typedesc[RecvService], node: WakuNode): T =
+proc new*(T: typedesc[RecvService], waku: Waku): T =
   ## The storeClient will help to acquire any possible missed messages
 
   let now = getNowInNanosecondTime()
   var recvService = RecvService(
-    node: node,
+    waku: waku,
     startTimeToCheck: now,
-    brokerCtx: node.brokerCtx,
+    brokerCtx: waku.brokerCtx,
     recentReceivedMsgs: @[],
   )
 
diff --git a/logos_delivery/messaging/delivery_service/send_service/send_service.nim b/logos_delivery/messaging/delivery_service/send_service/send_service.nim
index f77c6858c..026acee98 100644
--- a/logos_delivery/messaging/delivery_service/send_service/send_service.nim
+++ b/logos_delivery/messaging/delivery_service/send_service/send_service.nim
@@ -10,15 +10,15 @@ import
   logos_delivery/waku/[
     waku_core,
     node/waku_node,
-    node/subscription_manager,
     node/peer_manager,
-    waku_store/client,
     waku_store/common,
     waku_relay/protocol,
     waku_rln_relay/rln_relay,
     waku_lightpush/client,
     waku_lightpush/callbacks,
-  ]
+  ],
+  logos_delivery/waku/waku,
+  logos_delivery/waku/api/[store, subscriptions]
 import logos_delivery/api/messaging_client_api
 
 logScope:
@@ -57,7 +57,7 @@ type SendService* = ref object of RootObj
   serviceLoopHandle: Future[void] ## handle that allows to stop the async task
   sendProcessor: BaseSendProcessor
 
-  node: WakuNode
+  waku: Waku
   checkStoreForMessages: bool
   lastStoreCheckTime: Moment ## throttles store validation queries to ArchiveTime cadence
 
@@ -97,26 +97,31 @@ proc setupSendProcessorChain(
   return ok(processors[0])
 
 proc new*(
-    T: typedesc[SendService], preferP2PReliability: bool, w: WakuNode
+    T: typedesc[SendService], preferP2PReliability: bool, waku: Waku
 ): Result[T, string] =
-  if w.wakuRelay.isNil() and w.wakuLightpushClient.isNil():
+  # The send-processor chain needs raw publish handles (relay, lightpush client,
+  # RLN, peer manager) that the kernel API does not expose yet, so it is built
+  # from `waku.node`. Everything else goes through the Waku api surface.
+  let node = waku.node
+  if node.wakuRelay.isNil() and node.wakuLightpushClient.isNil():
     return err(
       "Could not create SendService. wakuRelay or wakuLightpushClient should be set"
     )
 
-  let checkStoreForMessages = preferP2PReliability and not w.wakuStoreClient.isNil()
+  let checkStoreForMessages = preferP2PReliability and waku.isStoreMounted()
 
   let sendProcessorChain = setupSendProcessorChain(
-    w.peerManager, w.wakuLightPushClient, w.wakuRelay, w.wakuRlnRelay, w.brokerCtx
+    node.peerManager, node.wakuLightPushClient, node.wakuRelay, node.wakuRlnRelay,
+    waku.brokerCtx,
   ).valueOr:
     return err("failed to setup SendProcessorChain: " & $error)
 
   let sendService = SendService(
-    brokerCtx: w.brokerCtx,
+    brokerCtx: waku.brokerCtx,
     taskCache: newSeq[DeliveryTask](),
     serviceLoopHandle: nil,
     sendProcessor: sendProcessorChain,
-    node: w,
+    waku: waku,
     checkStoreForMessages: checkStoreForMessages,
     lastStoreCheckTime: Moment.now(),
   )
@@ -127,7 +132,7 @@ proc addTask(self: SendService, task: DeliveryTask) =
   self.taskCache.addUnique(task)
 
 proc isStorePeerAvailable*(sendService: SendService): bool =
-  return sendService.node.peerManager.selectPeer(WakuStoreCodec).isSome()
+  return sendService.waku.hasStorePeer()
 
 proc checkMsgsInStore(self: SendService, tasksToValidate: seq[DeliveryTask]) {.async.} =
   if tasksToValidate.len() == 0:
@@ -142,7 +147,7 @@ proc checkMsgsInStore(self: SendService, tasksToValidate: seq[DeliveryTask]) {.a
   # TODO: confirm hash format for store query!!!
 
   let storeResp: StoreQueryResponse = (
-    await self.node.wakuStoreClient.queryToAny(
+    await self.waku.storeQueryToAny(
       StoreQueryRequest(includeData: false, messageHashes: hashesToValidate)
     )
   ).valueOr:
@@ -292,7 +297,7 @@ proc send*(self: SendService, task: DeliveryTask) {.async.} =
   info "SendService.send: processing delivery task",
     requestId = task.requestId, msgHash = task.msgHash.to0xHex()
 
-  self.node.subscriptionManager.subscribe(task.msg.contentTopic).isOkOr:
+  self.waku.subscribe(task.msg.contentTopic).isOkOr:
     error "SendService.send: failed to subscribe to content topic",
       contentTopic = task.msg.contentTopic, error = error
 
diff --git a/logos_delivery/messaging/messaging_client.nim b/logos_delivery/messaging/messaging_client.nim
index cf2db623b..4e02c7a82 100644
--- a/logos_delivery/messaging/messaging_client.nim
+++ b/logos_delivery/messaging/messaging_client.nim
@@ -28,9 +28,9 @@ proc new*(
 ): Result[T, string] =
   ## The messaging layer chains onto Waku: it drives the underlying Waku kernel
   ## for transport while exposing its own send/recv API.
-  let sendService = ?SendService.new(conf.useP2PReliability, waku.node)
-  let recvService = RecvService.new(waku.node)
-  ok(T(waku: waku, sendService: sendService, recvService: recvService))
+  let sendService = ?SendService.new(conf.useP2PReliability, waku)
+  let recvService = RecvService.new(waku)
+  return ok(T(waku: waku, sendService: sendService, recvService: recvService))
 
 proc start*(self: MessagingClient): Result[void, string] =
   if self.started:
diff --git a/logos_delivery/waku/api/store.nim b/logos_delivery/waku/api/store.nim
index 5095dbb0d..a54ebe1f3 100644
--- a/logos_delivery/waku/api/store.nim
+++ b/logos_delivery/waku/api/store.nim
@@ -1,11 +1,36 @@
 ## Waku layer API — store (historical query) operations.
 {.push raises: [].}
 
+import std/options
 import results, chronos, chronicles
 
 import logos_delivery/waku/waku
 import
-  logos_delivery/waku/[waku_core, node/waku_node, waku_store/common, waku_store/client]
+  logos_delivery/waku/
+    [waku_core, node/waku_node, node/peer_manager, waku_store/common, waku_store/client]
+
+proc isStoreMounted*(self: Waku): bool =
+  ## True if a store client is mounted (the node can run store queries).
+  return not self.node.wakuStoreClient.isNil()
+
+proc hasStorePeer*(self: Waku): bool =
+  ## True if at least one store service peer is available to query.
+  return self.node.peerManager.selectPeer(WakuStoreCodec).isSome()
+
+proc storeQueryToAny*(
+    self: Waku, request: StoreQueryRequest
+): Future[Result[StoreQueryResponse, string]] {.async.} =
+  ## Runs a store query against any available store peer (retries across peers).
+  try:
+    if self.node.wakuStoreClient.isNil():
+      return err("wakuStoreClient is not mounted")
+
+    let queryResponse = (await self.node.wakuStoreClient.queryToAny(request)).valueOr:
+      return err($error)
+
+    return ok(queryResponse)
+  except CatchableError as e:
+    return err(e.msg)
 
 proc storeQuery*(
     self: Waku, request: StoreQueryRequest, peer: string, timeoutMs: int
diff --git a/logos_delivery/waku/api/subscriptions.nim b/logos_delivery/waku/api/subscriptions.nim
index ec73bd9bf..7638abf3e 100644
--- a/logos_delivery/waku/api/subscriptions.nim
+++ b/logos_delivery/waku/api/subscriptions.nim
@@ -5,6 +5,7 @@
 ## kernel-level entry point so they never reach into `waku.node` internals.
 {.push raises: [].}
 
+import std/sets
 import results
 
 import logos_delivery/waku/waku
@@ -21,3 +22,16 @@ proc unsubscribe*(self: Waku, contentTopic: ContentTopic): Result[void, string]
 proc isSubscribed*(self: Waku, contentTopic: ContentTopic): Result[bool, string] =
   ## True if the node already subscribes to `contentTopic`.
   return self.node.subscriptionManager.isSubscribed(contentTopic)
+
+proc isContentSubscribed*(
+    self: Waku, shard: PubsubTopic, contentTopic: ContentTopic
+): bool =
+  ## True if `contentTopic` is subscribed on the given `shard` (pubsub topic).
+  return self.node.subscriptionManager.isContentSubscribed(shard, contentTopic)
+
+proc subscribedContentTopics*(self: Waku): seq[(PubsubTopic, HashSet[ContentTopic])] =
+  ## Snapshot of every shard with its non-empty content-topic set.
+  var res: seq[(PubsubTopic, HashSet[ContentTopic])]
+  for shard, contentTopics in self.node.subscriptionManager.subscribedContentTopics:
+    res.add((shard, contentTopics))
+  return res

From e2448115b3e6644b3140a65057fa46975907fa4c Mon Sep 17 00:00:00 2001
From: Ivan FB <ivansete@status.im>
Date: Fri, 26 Jun 2026 14:21:30 +0200
Subject: [PATCH 5/5] messaging: route the send-processor chain through the
 Waku kernel

The send pipeline still reached into `waku.node` for its publish handles
(`wakuRelay`, `wakuRlnRelay`, `wakuLightpushClient`, `peerManager`) when
building the processor chain -- the last node coupling the previous step
had to leave in place with a comment.

Add `waku/api/publish.nim`, a Waku-layer surface for the send pipeline:
relay/lightpush availability (`hasRelay`/`hasLightpush`), the relay push
handler factory (`relayPushHandler`, which folds in the RLN proof), and
lightpush peer selection + publish (`lightpushPeerAvailable`,
`lightpushPublishToAny`). These keep the rich `WakuLightPushResult` the
processors branch on for retry decisions.

`SendService` and `LightpushSendProcessor` now drive that surface and
hold only a `Waku`; no part of the send pipeline inspects `waku.node`
anymore. "No lightpush peer" now surfaces as SERVICE_NOT_AVAILABLE, which
the processor already maps to NextRoundRetry, so behaviour is unchanged.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 .../send_service/lightpush_processor.nim      | 32 ++--------
 .../send_service/send_service.nim             | 44 +++----------
 logos_delivery/waku/api/publish.nim           | 64 +++++++++++++++++++
 3 files changed, 80 insertions(+), 60 deletions(-)
 create mode 100644 logos_delivery/waku/api/publish.nim

diff --git a/logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim b/logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim
index ae5e775aa..fa199774b 100644
--- a/logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim
+++ b/logos_delivery/messaging/delivery_service/send_service/lightpush_processor.nim
@@ -2,10 +2,8 @@ import logos_delivery/waku/compat/option_valueor
 import chronicles, chronos, results
 import std/options
 import brokers/broker_context
-import
-  logos_delivery/waku/node/peer_manager,
-  logos_delivery/waku/waku_core,
-  logos_delivery/waku/waku_lightpush/[common, client, rpc]
+import logos_delivery/waku/waku_core, logos_delivery/waku/waku
+import logos_delivery/waku/api/publish
 
 import ./[delivery_task, send_processor]
 
@@ -13,27 +11,17 @@ logScope:
   topics = "send service lightpush processor"
 
 type LightpushSendProcessor* = ref object of BaseSendProcessor
-  peerManager: PeerManager
-  lightpushClient: WakuLightPushClient
+  waku: Waku
 
 proc new*(
-    T: typedesc[LightpushSendProcessor],
-    peerManager: PeerManager,
-    lightpushClient: WakuLightPushClient,
-    brokerCtx: BrokerContext,
+    T: typedesc[LightpushSendProcessor], waku: Waku, brokerCtx: BrokerContext
 ): T =
-  return
-    T(peerManager: peerManager, lightpushClient: lightpushClient, brokerCtx: brokerCtx)
-
-proc isLightpushPeerAvailable(
-    self: LightpushSendProcessor, pubsubTopic: PubsubTopic
-): bool =
-  return self.peerManager.selectPeer(WakuLightPushCodec, some(pubsubTopic)).isSome()
+  return T(waku: waku, brokerCtx: brokerCtx)
 
 method isValidProcessor*(
     self: LightpushSendProcessor, task: DeliveryTask
 ): bool {.gcsafe.} =
-  return self.isLightpushPeerAvailable(task.pubsubTopic)
+  return self.waku.lightpushPeerAvailable(task.pubsubTopic)
 
 method sendImpl*(
     self: LightpushSendProcessor, task: DeliveryTask
@@ -44,14 +32,8 @@ method sendImpl*(
     msgHash = task.msgHash.to0xHex(),
     tryCount = task.tryCount
 
-  let peer = self.peerManager.selectPeer(WakuLightPushCodec, some(task.pubsubTopic)).valueOr:
-    debug "No peer available for Lightpush, request pushed back for next round",
-      requestId = task.requestId
-    task.state = DeliveryState.NextRoundRetry
-    return
-
   let numLightpushServers = (
-    await self.lightpushClient.publish(some(task.pubsubTopic), task.msg, peer)
+    await self.waku.lightpushPublishToAny(task.pubsubTopic, task.msg)
   ).valueOr:
     error "LightpushSendProcessor.sendImpl failed", error = error.desc.get($error.code)
     case error.code
diff --git a/logos_delivery/messaging/delivery_service/send_service/send_service.nim b/logos_delivery/messaging/delivery_service/send_service/send_service.nim
index 026acee98..f482a2aff 100644
--- a/logos_delivery/messaging/delivery_service/send_service/send_service.nim
+++ b/logos_delivery/messaging/delivery_service/send_service/send_service.nim
@@ -7,18 +7,9 @@ import chronos, chronicles, libp2p/utility
 import brokers/broker_context
 import
   ./[send_processor, relay_processor, lightpush_processor, delivery_task],
-  logos_delivery/waku/[
-    waku_core,
-    node/waku_node,
-    node/peer_manager,
-    waku_store/common,
-    waku_relay/protocol,
-    waku_rln_relay/rln_relay,
-    waku_lightpush/client,
-    waku_lightpush/callbacks,
-  ],
+  logos_delivery/waku/[waku_core, waku_store/common],
   logos_delivery/waku/waku,
-  logos_delivery/waku/api/[store, subscriptions]
+  logos_delivery/waku/api/[store, subscriptions, publish]
 import logos_delivery/api/messaging_client_api
 
 logScope:
@@ -62,14 +53,10 @@ type SendService* = ref object of RootObj
   lastStoreCheckTime: Moment ## throttles store validation queries to ArchiveTime cadence
 
 proc setupSendProcessorChain(
-    peerManager: PeerManager,
-    lightpushClient: WakuLightPushClient,
-    relay: WakuRelay,
-    rlnRelay: WakuRLNRelay,
-    brokerCtx: BrokerContext,
+    waku: Waku, brokerCtx: BrokerContext
 ): Result[BaseSendProcessor, string] =
-  let isRelayAvail = not relay.isNil()
-  let isLightPushAvail = not lightpushClient.isNil()
+  let isRelayAvail = waku.hasRelay()
+  let isLightPushAvail = waku.hasLightpush()
 
   if not isRelayAvail and not isLightPushAvail:
     return err("No valid send processor found for the delivery task")
@@ -77,16 +64,10 @@ proc setupSendProcessorChain(
   var processors = newSeq[BaseSendProcessor]()
 
   if isRelayAvail:
-    let rln: Option[WakuRLNRelay] =
-      if rlnRelay.isNil():
-        none[WakuRLNRelay]()
-      else:
-        some(rlnRelay)
-    let publishProc = getRelayPushHandler(relay, rln)
-
+    let publishProc = waku.relayPushHandler()
     processors.add(RelaySendProcessor.new(isLightPushAvail, publishProc, brokerCtx))
   if isLightPushAvail:
-    processors.add(LightpushSendProcessor.new(peerManager, lightpushClient, brokerCtx))
+    processors.add(LightpushSendProcessor.new(waku, brokerCtx))
 
   var currentProcessor: BaseSendProcessor = processors[0]
   for i in 1 ..< processors.len:
@@ -99,21 +80,14 @@ proc setupSendProcessorChain(
 proc new*(
     T: typedesc[SendService], preferP2PReliability: bool, waku: Waku
 ): Result[T, string] =
-  # The send-processor chain needs raw publish handles (relay, lightpush client,
-  # RLN, peer manager) that the kernel API does not expose yet, so it is built
-  # from `waku.node`. Everything else goes through the Waku api surface.
-  let node = waku.node
-  if node.wakuRelay.isNil() and node.wakuLightpushClient.isNil():
+  if not waku.hasRelay() and not waku.hasLightpush():
     return err(
       "Could not create SendService. wakuRelay or wakuLightpushClient should be set"
     )
 
   let checkStoreForMessages = preferP2PReliability and waku.isStoreMounted()
 
-  let sendProcessorChain = setupSendProcessorChain(
-    node.peerManager, node.wakuLightPushClient, node.wakuRelay, node.wakuRlnRelay,
-    waku.brokerCtx,
-  ).valueOr:
+  let sendProcessorChain = setupSendProcessorChain(waku, waku.brokerCtx).valueOr:
     return err("failed to setup SendProcessorChain: " & $error)
 
   let sendService = SendService(
diff --git a/logos_delivery/waku/api/publish.nim b/logos_delivery/waku/api/publish.nim
new file mode 100644
index 000000000..10d518c03
--- /dev/null
+++ b/logos_delivery/waku/api/publish.nim
@@ -0,0 +1,64 @@
+## Waku layer API — message publish primitives used by the messaging send
+## pipeline.
+##
+## Unlike `relay.nim`/`lightpush.nim`, these preserve the rich
+## `WakuLightPushResult` (status code + description) that the send processors
+## branch on for their retry decisions, and expose relay/lightpush availability
+## so the messaging layer never inspects `waku.node` directly.
+import logos_delivery/waku/compat/option_valueor
+{.push raises: [].}
+
+import std/options
+import results, chronos
+
+import logos_delivery/waku/waku
+import
+  logos_delivery/waku/[
+    waku_core,
+    node/waku_node,
+    node/peer_manager,
+    waku_relay/protocol,
+    waku_rln_relay/rln_relay,
+    waku_lightpush/common,
+    waku_lightpush/rpc,
+    waku_lightpush/client,
+    waku_lightpush/callbacks,
+  ]
+
+# WakuLightPushResult, PushMessageHandler, LightPushErrorCode (common) plus the
+# LightPushStatusCode `$`/`==` the send processors branch on (rpc).
+export common, rpc
+
+proc hasRelay*(self: Waku): bool =
+  ## True if relay (gossipsub publishing) is mounted.
+  return not self.node.wakuRelay.isNil()
+
+proc hasLightpush*(self: Waku): bool =
+  ## True if a lightpush client is mounted.
+  return not self.node.wakuLightpushClient.isNil()
+
+proc relayPushHandler*(self: Waku): PushMessageHandler =
+  ## Builds the relay publish handler (appending an RLN proof when RLN is
+  ## mounted) used by the send pipeline. Caller ensures relay is mounted.
+  let rln =
+    if self.node.wakuRlnRelay.isNil():
+      none[WakuRLNRelay]()
+    else:
+      some(self.node.wakuRlnRelay)
+  return getRelayPushHandler(self.node.wakuRelay, rln)
+
+proc lightpushPeerAvailable*(self: Waku, shard: PubsubTopic): bool =
+  ## True if a lightpush service peer is available for `shard`.
+  return self.node.peerManager.selectPeer(WakuLightPushCodec, some(shard)).isSome()
+
+proc lightpushPublishToAny*(
+    self: Waku, shard: PubsubTopic, message: WakuMessage
+): Future[WakuLightPushResult] {.async.} =
+  ## Selects a lightpush service peer for `shard` and publishes `message`.
+  ## Returns SERVICE_NOT_AVAILABLE when no peer is available.
+  let peer = self.node.peerManager.selectPeer(WakuLightPushCodec, some(shard)).valueOr:
+    return lightpushResultServiceUnavailable("no lightpush peer available for shard")
+  try:
+    return await self.node.wakuLightpushClient.publish(some(shard), message, peer)
+  except CatchableError as e:
+    return lightpushResultInternalError(e.msg)