diff --git a/logos_delivery/api/events/kernel_events.nim b/logos_delivery/api/events/kernel_events.nim new file mode 100644 index 000000000..d8dd46627 --- /dev/null +++ b/logos_delivery/api/events/kernel_events.nim @@ -0,0 +1,18 @@ +import brokers/event_broker + +import logos_delivery/api/types +import logos_delivery/waku/waku_core/topics/pubsub_topic +import logos_delivery/waku/waku_core/message + +export event_broker, pubsub_topic, message + +EventBroker: + # Internal event emitted when a message arrives from the network via any protocol + type MessageSeenEvent* = object + topic*: PubsubTopic + message*: WakuMessage + +# Emitted by the health monitor when overall node connectivity changes. +EventBroker: + type EventConnectionStatusChange* = object + connectionStatus*: ConnectionStatus diff --git a/logos_delivery/api/events/messaging_client_events.nim b/logos_delivery/api/events/messaging_client_events.nim new file mode 100644 index 000000000..b3b7a46f6 --- /dev/null +++ b/logos_delivery/api/events/messaging_client_events.nim @@ -0,0 +1,30 @@ +import brokers/event_broker + +import logos_delivery/api/types as api_types + +export event_broker, api_types + +EventBroker: + # Event emitted when a message is sent to the network + type MessageSentEvent* = object + requestId*: RequestId + messageHash*: string + +EventBroker: + # Event emitted when a message send operation fails + type MessageErrorEvent* = object + requestId*: RequestId + messageHash*: string + error*: string + +EventBroker: + # Confirmation that a message has been correctly delivered to some neighbouring nodes. + type MessagePropagatedEvent* = object + requestId*: RequestId + messageHash*: string + +EventBroker: + # Event emitted when a message is received via Waku + type MessageReceivedEvent* = object + messageHash*: string + message*: WakuMessage diff --git a/logos_delivery/api/events/reliable_channel_manager_events.nim b/logos_delivery/api/events/reliable_channel_manager_events.nim new file mode 100644 index 000000000..0f3b4e1ec --- /dev/null +++ b/logos_delivery/api/events/reliable_channel_manager_events.nim @@ -0,0 +1,27 @@ +import brokers/event_broker + +import logos_delivery/channels/types as channel_types + +export event_broker, channel_types + +EventBroker: + type ChannelMessageReceivedEvent* = object + channelId*: ChannelId + senderId*: SdsParticipantID + payload*: seq[byte] + +EventBroker: + ## Emitted when every segment of a channel-level `send()` reached + ## `Confirmed`. Channel-level analogue of `MessageSentEvent`; the + ## `requestId` is the channel-layer parent returned by `send()`. + type ChannelMessageSentEvent* = object + channelId*: ChannelId + requestId*: RequestId + +EventBroker: + ## Emitted when a channel-level `send()` finalises with at least one + ## segment in `Failed`. Channel-level analogue of `MessageErrorEvent`. + type ChannelMessageErrorEvent* = object + channelId*: ChannelId + requestId*: RequestId + error*: string diff --git a/logos_delivery/api/kernel_api.nim b/logos_delivery/api/kernel_api.nim index 35cf7251f..01b0c30c0 100644 --- a/logos_delivery/api/kernel_api.nim +++ b/logos_delivery/api/kernel_api.nim @@ -1,203 +1,81 @@ -import std/options import chronos, results -import brokers/event_broker import logos_delivery/api/types as api_types import logos_delivery/waku/waku_core/topics/pubsub_topic import logos_delivery/waku/waku_core/message +import logos_delivery/waku/waku_core/subscription/push_handler import logos_delivery/waku/waku_store/common as store_types -export event_broker export api_types, pubsub_topic, store_types -type IKernel* = ref object of RootObj +# Structural API contract for the Kernel surface, implemented by `Waku` +# (ops in `waku/api/*`). +type KernelApi* = concept w + # --- topic construction --- + buildContentTopic( + w, appName = string, appVersion = uint32, name = string, encoding = string + ) is Future[Result[ContentTopic, string]] + buildPubsubTopic(w, topicName = string) is Future[Result[PubsubTopic, string]] + defaultPubsubTopic(w) is Future[Result[PubsubTopic, string]] -EventBroker: - # Internal event emitted when a message arrives from the network via any protocol - type MessageSeenEvent* = object - topic*: PubsubTopic - message*: WakuMessage + # --- relay --- + relayPublish(w, pubsubTopic = PubsubTopic, message = WakuMessage, timeoutMs = uint32) is + Future[Result[string, string]] + relaySubscribe(w, pubsubTopic = PubsubTopic) is Future[Result[bool, string]] + relayUnsubscribe(w, pubsubTopic = PubsubTopic) is Future[Result[bool, string]] + relayAddProtectedShard(w, clusterId = uint16, shardId = uint16, publicKey = string) is + Future[Result[bool, string]] + relayConnectedPeers(w, pubsubTopic = PubsubTopic) is + Future[Result[seq[string], string]] + relayPeersInMesh(w, pubsubTopic = PubsubTopic) is Future[Result[seq[string], string]] + relayNumPeersInMesh(w, pubsubTopic = PubsubTopic) is Future[Result[int, string]] + relayNumConnectedPeers(w, pubsubTopic = PubsubTopic) is Future[Result[int, string]] -# --- topic construction --- -method buildContentTopic*( - self: IKernel, appName: string, appVersion: uint32, name: string, encoding: string -): Future[Result[ContentTopic, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.buildContentTopic not implemented") + # --- filter --- + filterSubscribe( + w, + pubsubTopic = PubsubTopic, + contentTopics = seq[ContentTopic], + pushHandler = FilterPushHandler, + ) is Future[Result[bool, string]] + filterUnsubscribe(w, pubsubTopic = PubsubTopic, contentTopics = seq[ContentTopic]) is + Future[Result[bool, string]] + filterUnsubscribeAll(w) is Future[Result[bool, string]] -method buildPubsubTopic*( - self: IKernel, topicName: string -): Future[Result[PubsubTopic, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.buildPubsubTopic not implemented") + # --- lightpush --- + lightpushPublish(w, pubsubTopic = PubsubTopic, message = WakuMessage) is + Future[Result[string, string]] -method defaultPubsubTopic*( - self: IKernel -): Future[Result[PubsubTopic, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.defaultPubsubTopic not implemented") + # --- store --- + storeQuery(w, request = StoreQueryRequest, peer = string, timeoutMs = int) is + Future[Result[StoreQueryResponse, string]] -# --- relay --- -method relayPublish*( - self: IKernel, pubsubTopic: PubsubTopic, message: WakuMessage, timeoutMs: uint32 -): Future[Result[int, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.relayPublish not implemented") + # --- peer management --- + connect(w, peers = seq[string], timeoutMs = uint32) is Future[Result[bool, string]] + disconnectPeerById(w, peerId = string) is Future[Result[bool, string]] + disconnectAllPeers(w) is Future[Result[bool, string]] + dialPeer(w, peerAddr = string, protocol = string, timeoutMs = int) is + Future[Result[bool, string]] + dialPeerById(w, peerId = string, protocol = string, timeoutMs = int) is + Future[Result[bool, string]] + peerIdsFromPeerstore(w) is Future[Result[seq[string], string]] + connectedPeersInfo(w) is Future[Result[seq[PeerConnInfo], string]] + connectedPeers(w) is Future[Result[seq[string], string]] + peerIdsByProtocol(w, protocol = string) is Future[Result[seq[string], string]] -method relaySubscribe*( - self: IKernel, pubsubTopic: PubsubTopic -): Future[Result[bool, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.relaySubscribe not implemented") + # --- discovery --- + dnsDiscovery(w, enrTreeUrl = string, nameServer = string, timeoutMs = int) is + Future[Result[seq[string], string]] + discv5UpdateBootnodes(w, bootnodes = string) is Future[Result[bool, string]] + startDiscv5(w) is Future[Result[bool, string]] + stopDiscv5(w) is Future[Result[bool, string]] + peerExchangeRequest(w, numPeers = uint64) is Future[Result[int, string]] -method relayUnsubscribe*( - self: IKernel, pubsubTopic: PubsubTopic -): Future[Result[bool, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.relayUnsubscribe not implemented") - -method relayAddProtectedShard*( - self: IKernel, clusterId: uint16, shardId: uint16, publicKey: string -): Future[Result[bool, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.relayAddProtectedShard not implemented") - -method relayConnectedPeers*( - self: IKernel, pubsubTopic: PubsubTopic -): Future[Result[seq[string], string]] {.async: (raises: []), base.} = - return err("Interface IKernel.relayConnectedPeers not implemented") - -method relayPeersInMesh*( - self: IKernel, pubsubTopic: PubsubTopic -): Future[Result[seq[string], string]] {.async: (raises: []), base.} = - return err("Interface IKernel.relayPeersInMesh not implemented") - -# --- filter --- -method filterSubscribe*( - self: IKernel, - pubsubTopic: Option[PubsubTopic], - contentTopics: seq[ContentTopic], - peer: string, -): Future[Result[bool, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.filterSubscribe not implemented") - -method filterUnsubscribe*( - self: IKernel, - pubsubTopic: Option[PubsubTopic], - contentTopics: seq[ContentTopic], - peer: string, -): Future[Result[bool, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.filterUnsubscribe not implemented") - -method filterUnsubscribeAll*( - self: IKernel, peer: string -): Future[Result[bool, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.filterUnsubscribeAll not implemented") - -# --- lightpush --- -method lightpushPublish*( - self: IKernel, pubsubTopic: PubsubTopic, message: WakuMessage, peer: string -): Future[Result[string, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.lightpushPublish not implemented") - -# --- store --- -method storeQuery*( - self: IKernel, request: StoreQueryRequest, peer: string, timeoutMs: int -): Future[Result[StoreQueryResponse, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.storeQuery not implemented") - -# --- peer management --- -method connect*( - self: IKernel, peers: seq[string], timeoutMs: uint32 -): Future[Result[bool, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.connect not implemented") - -method disconnectPeerById*( - self: IKernel, peerId: string -): Future[Result[bool, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.disconnectPeerById not implemented") - -method disconnectAllPeers*( - self: IKernel -): Future[Result[bool, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.disconnectAllPeers not implemented") - -method dialPeer*( - self: IKernel, peerAddr: string, protocol: string, timeoutMs: int -): Future[Result[bool, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.dialPeer not implemented") - -method dialPeerById*( - self: IKernel, peerId: string, protocol: string, timeoutMs: int -): Future[Result[bool, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.dialPeerById not implemented") - -method peerIdsFromPeerstore*( - self: IKernel -): Future[Result[seq[string], string]] {.async: (raises: []), base.} = - return err("Interface IKernel.peerIdsFromPeerstore not implemented") - -method connectedPeersInfo*( - self: IKernel -): Future[Result[seq[string], string]] {.async: (raises: []), base.} = - return err("Interface IKernel.connectedPeersInfo not implemented") - -method connectedPeers*( - self: IKernel -): Future[Result[seq[string], string]] {.async: (raises: []), base.} = - return err("Interface IKernel.connectedPeers not implemented") - -method peerIdsByProtocol*( - self: IKernel, protocol: string -): Future[Result[seq[string], string]] {.async: (raises: []), base.} = - return err("Interface IKernel.peerIdsByProtocol not implemented") - -# --- discovery --- -method dnsDiscovery*( - self: IKernel, enrTreeUrl: string, nameServer: string, timeoutMs: int -): Future[Result[seq[string], string]] {.async: (raises: []), base.} = - return err("Interface IKernel.dnsDiscovery not implemented") - -method discv5UpdateBootnodes*( - self: IKernel, bootnodes: seq[string] -): Future[Result[bool, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.discv5UpdateBootnodes not implemented") - -method startDiscv5*( - self: IKernel -): Future[Result[bool, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.startDiscv5 not implemented") - -method stopDiscv5*( - self: IKernel -): Future[Result[bool, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.stopDiscv5 not implemented") - -method peerExchangeRequest*( - self: IKernel, numPeers: uint64 -): Future[Result[int, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.peerExchangeRequest not implemented") - -# --- debug / info --- -method version*( - self: IKernel -): Future[Result[string, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.version not implemented") - -method listenAddresses*( - self: IKernel -): Future[Result[seq[string], string]] {.async: (raises: []), base.} = - return err("Interface IKernel.listenAddresses not implemented") - -method myEnr*( - self: IKernel -): Future[Result[string, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.myEnr not implemented") - -method myPeerId*( - self: IKernel -): Future[Result[string, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.myPeerId not implemented") - -method metrics*( - self: IKernel -): Future[Result[string, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.metrics not implemented") - -method pingPeer*( - self: IKernel, peerAddr: string, timeoutMs: int -): Future[Result[int64, string]] {.async: (raises: []), base.} = - return err("Interface IKernel.pingPeer not implemented") + # --- debug / info --- + version(w) is Future[Result[string, string]] + listenAddresses(w) is Future[Result[seq[string], string]] + myEnr(w) is Future[Result[string, string]] + myPeerId(w) is Future[Result[string, string]] + metrics(w) is Future[Result[string, string]] + isOnline(w) is Future[Result[bool, string]] + pingPeer(w, peerAddr = string, timeoutMs = int) is Future[Result[int64, string]] diff --git a/logos_delivery/api/logos_delivery_api.nim b/logos_delivery/api/logos_delivery_api.nim index 162159d8e..708484553 100644 --- a/logos_delivery/api/logos_delivery_api.nim +++ b/logos_delivery/api/logos_delivery_api.nim @@ -1,33 +1,11 @@ -## `LogosDelivery` is the project entry point. It is a pure concentrator: it -## owns exactly one instance of each API layer -## -## Waku <- MessagingClient <- ReliableChannelManager -## -## and chains them together (each layer drives the one below it). Every layer -## keeps its own, separate public API — `LogosDelivery` only wires them up and -## drives the shared `new` / `start` / `stop` lifecycle. +import chronos, results -{.push raises: [].} +# Structural API contract for the top-level entry point, implemented by +# `LogosDelivery` (the aggregate owning one instance of each API layer). +type LogosDeliveryApi* = concept ld + # --- lifecycle --- + start(ld) is Future[Result[void, string]] + stop(ld) is Future[Result[void, string]] -import results, chronos -import brokers/event_broker -import types as api_types - -export api_types, event_broker - -type - ## Entry point. Holds one instance of each API layer. - ILogosDelivery* = ref object of RootObj - -EventBroker: - type EventConnectionStatusChange* = object - connectionStatus*: ConnectionStatus - -method start*(self: ILogosDelivery): Future[Result[void, string]] {.async, base.} = - return err("ILogosDelivery.start not implemented") - -method stop*(self: ILogosDelivery): Future[Result[void, string]] {.async, base.} = - return err("ILogosDelivery.stop not implemented") - -method isOnline*(self: ILogosDelivery): Future[Result[bool, string]] {.async, base.} = - return err("ILogosDelivery.isOnline not implemented") + # --- health --- + isOnline(ld) is Future[Result[bool, string]] diff --git a/logos_delivery/api/messaging_client_api.nim b/logos_delivery/api/messaging_client_api.nim index f29a5cb41..536ce94fa 100644 --- a/logos_delivery/api/messaging_client_api.nim +++ b/logos_delivery/api/messaging_client_api.nim @@ -1,49 +1,11 @@ import chronos, results -import brokers/event_broker import logos_delivery/api/types as api_types -import logos_delivery/waku/waku_core/message -export event_broker, api_types +export api_types -type IMessagingClient* = ref object of RootObj - -EventBroker: - # Event emitted when a message is sent to the network - type MessageSentEvent* = object - requestId*: RequestId - messageHash*: string - -EventBroker: - # Event emitted when a message send operation fails - type MessageErrorEvent* = object - requestId*: RequestId - messageHash*: string - error*: string - -EventBroker: - # Confirmation that a message has been correctly delivered to some neighbouring nodes. - type MessagePropagatedEvent* = object - requestId*: RequestId - messageHash*: string - -EventBroker: - # Event emitted when a message is received via Waku - type MessageReceivedEvent* = object - messageHash*: string - message*: WakuMessage - -method subscribe*( - self: IMessagingClient, contentTopic: ContentTopic -): Future[Result[void, string]] {.async: (raises: []), base.} = - return err("Interface IMessagingClient.subscribe not implemented") - -method unsubscribe*( - self: IMessagingClient, contentTopic: ContentTopic -): Result[void, string] {.base, raises: [].} = - return err("Interface IMessagingClient.unsubscribe not implemented") - -method send*( - self: IMessagingClient, envelope: MessageEnvelope -): Future[Result[RequestId, string]] {.async: (raises: []), base.} = - return err("Interface IMessagingClient.send not implemented") +# Structural API contract for a messaging client (ops in `messaging/api/*`). +type MessagingApi* = concept c + subscribe(c, contentTopic = ContentTopic) is Future[Result[void, string]] + unsubscribe(c, contentTopic = ContentTopic) is Result[void, string] + send(c, envelope = MessageEnvelope) is Future[Result[RequestId, string]] diff --git a/logos_delivery/api/reliable_channel_manager_api.nim b/logos_delivery/api/reliable_channel_manager_api.nim index c5266972a..d50b57288 100644 --- a/logos_delivery/api/reliable_channel_manager_api.nim +++ b/logos_delivery/api/reliable_channel_manager_api.nim @@ -1,70 +1,23 @@ import chronos, results -import brokers/event_broker - import logos_delivery/api/types as api_types import logos_delivery/channels/types as channel_types -# The channel layer re-uses the messaging-layer message events (the `requestId` -# is shared across layers), so it re-exports the messaging interface's event -# surface and only adds the channel-level events that have no lower-layer -# analogue (reassembled payload / senderId / channelId). -import logos_delivery/api/messaging_client_api +export api_types, channel_types -export event_broker, api_types -export channel_types, messaging_client_api +type SendHandler* = proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {. + async: (raises: [CatchableError]), gcsafe +.} + ## Egress dispatch boundary. Typically wraps `MessagingClient.send`; + ## tests inject a fake that records calls and returns canned + ## `RequestId`s so the send state machine can be exercised end-to-end + ## without a network. -type - IReliableChannelManager* = ref object of RootObj - - SendHandler* = proc(envelope: MessageEnvelope): Future[Result[RequestId, string]] {. - async: (raises: [CatchableError]), gcsafe - .} - ## Egress dispatch boundary. Typically wraps `MessagingClient.send`; - ## tests inject a fake that records calls and returns canned - ## `RequestId`s so the send state machine can be exercised end-to-end - ## without a network. - -EventBroker: - type ChannelMessageReceivedEvent* = object - channelId*: ChannelId - senderId*: SdsParticipantID - payload*: seq[byte] - -EventBroker: - ## Emitted when every segment of a channel-level `send()` reached - ## `Confirmed`. Channel-level analogue of `MessageSentEvent`; the - ## `requestId` is the channel-layer parent returned by `send()`. - type ChannelMessageSentEvent* = object - channelId*: ChannelId - requestId*: RequestId - -EventBroker: - ## Emitted when a channel-level `send()` finalises with at least one - ## segment in `Failed`. Channel-level analogue of `MessageErrorEvent`. - type ChannelMessageErrorEvent* = object - channelId*: ChannelId - requestId*: RequestId - error*: string - -method createReliableChannel*( - self: IReliableChannelManager, - channelId: ChannelId, - contentTopic: ContentTopic, - senderId: SdsParticipantID, - sendHandler: SendHandler = nil, -): Result[ChannelId, string] {.base.} = - return err("Interface IReliableChannelManager.createReliableChannel not implemented") - -method closeChannel*( - self: IReliableChannelManager, channelId: ChannelId -): Future[Result[void, string]] {.async: (raises: []), base.} = - return err("Interface IReliableChannelManager.closeChannel not implemented") - -method send*( - self: IReliableChannelManager, - channelId: ChannelId, - appPayload: seq[byte], - ephemeral: bool = false, -): Future[Result[RequestId, string]] {.async: (raises: []), base.} = - return err("Interface IReliableChannelManager.send not implemented") +# Structural API contract for the reliable-channel surface (ops in `channels/api/*`). +type ReliableChannelApi* = concept c + createReliableChannel( + c, channelId = ChannelId, contentTopic = ContentTopic, senderId = SdsParticipantID + ) is Result[ChannelId, string] + closeChannel(c, channelId = ChannelId) is Future[Result[void, string]] + send(c, channelId = ChannelId, appPayload = seq[byte]) is + Future[Result[RequestId, string]] diff --git a/logos_delivery/api/types.nim b/logos_delivery/api/types.nim index 5757a8e82..97e7a11c9 100644 --- a/logos_delivery/api/types.nim +++ b/logos_delivery/api/types.nim @@ -26,6 +26,11 @@ type PartiallyConnected Connected + PeerConnInfo* = object ## structured connected-peer info for the api boundary + peerId*: string + protocols*: seq[string] + addresses*: seq[string] + proc new*(T: typedesc[RequestId], rng: crypto.Rng): T = ## Generate a new RequestId using the provided RNG. RequestId(request_utils.generateRequestId(rng)) diff --git a/logos_delivery/channels/reliable_channel.nim b/logos_delivery/channels/reliable_channel.nim index d06f74e36..06648a8f7 100644 --- a/logos_delivery/channels/reliable_channel.nim +++ b/logos_delivery/channels/reliable_channel.nim @@ -23,6 +23,8 @@ import libp2p/crypto/crypto as libp2p_crypto import logos_delivery/api/types import logos_delivery/api/reliable_channel_manager_api +import logos_delivery/api/events/messaging_client_events +import logos_delivery/api/events/reliable_channel_manager_events import logos_delivery/messaging/delivery_service/send_service import logos_delivery/waku/waku_core/topics diff --git a/logos_delivery/channels/reliable_channel_manager.nim b/logos_delivery/channels/reliable_channel_manager.nim index d3f671c2b..e8d579326 100644 --- a/logos_delivery/channels/reliable_channel_manager.nim +++ b/logos_delivery/channels/reliable_channel_manager.nim @@ -28,7 +28,7 @@ type ## channel API. Placeholder for now (segmentation / SDS / rate-limit defaults ## will move here in a follow-up PR); kept so each layer owns its own config. - ReliableChannelManager* = ref object of IReliableChannelManager + ReliableChannelManager* = ref object ## Implements `ReliableChannelApi`. channels*: Table[ChannelId, ReliableChannel] ## read by `channels/api.nim` messagingClient: MessagingClient ## The channel layer chains onto messaging. sendHandler*: SendHandler diff --git a/logos_delivery/logos_delivery.nim b/logos_delivery/logos_delivery.nim index d0fb33b90..0b161d6f7 100644 --- a/logos_delivery/logos_delivery.nim +++ b/logos_delivery/logos_delivery.nim @@ -11,9 +11,6 @@ import results, chronos, chronicles -import logos_delivery/api/logos_delivery_api -export logos_delivery_api - # Each layer has a core module (type + new/start/stop) and an api/ folder whose # modules each implement a differentiated set of operations, plus an events # surface. The concentrator re-exports them so library consumers get the full @@ -31,8 +28,9 @@ import export 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. +# Kernel event surface (`MessageSeenEvent`) plus the remaining waku health events. +import logos_delivery/api/events/kernel_events +export kernel_events import logos_delivery/waku/api/events/health_events export health_events @@ -41,7 +39,8 @@ import logos_delivery/messaging/messaging_client export messaging_client import logos_delivery/messaging/api/[subscription, send] export subscription, send -# Message* events are surfaced via `export messaging_client` (messaging interface). +import logos_delivery/api/events/messaging_client_events +export messaging_client_events # Reliable Channel layer import logos_delivery/channels/reliable_channel_manager @@ -50,8 +49,11 @@ import logos_delivery/channels/api/channel_lifecycle export channel_lifecycle import logos_delivery/channels/api/send as channel_send export channel_send -# ChannelMessage* events are surfaced via `export reliable_channel_manager`. +import logos_delivery/api/events/reliable_channel_manager_events +export reliable_channel_manager_events +import logos_delivery/api/logos_delivery_api +export logos_delivery_api import logos_delivery/waku/factory/waku_conf import logos_delivery/waku/factory/app_callbacks import tools/confutils/cli_args @@ -69,8 +71,7 @@ type messaging*: MessagingClientConf reliableChannel*: ReliableChannelManagerConf - LogosDelivery* = ref object of ILogosDelivery - ## Entry point. Holds one instance of each API layer. + LogosDelivery* = ref object ## Entry point. Holds one instance of each API layer. waku*: Waku messagingClient*: MessagingClient reliableChannelManager*: ReliableChannelManager @@ -114,7 +115,7 @@ proc new*( ) ) -method start*(self: LogosDelivery): Future[Result[void, string]] {.async.} = +proc start*(self: LogosDelivery): Future[Result[void, string]] {.async.} = ## Starts each layer bottom-up: transport first, then messaging, then channels. if self.waku.isNil(): return err("Waku node is not initialized") @@ -134,7 +135,7 @@ method start*(self: LogosDelivery): Future[Result[void, string]] {.async.} = return ok() -method stop*(self: LogosDelivery): Future[Result[void, string]] {.async.} = +proc stop*(self: LogosDelivery): Future[Result[void, string]] {.async.} = ## Stops in reverse order so higher layers drain before their dependencies. await self.reliableChannelManager.stop() await self.messagingClient.stop() @@ -144,7 +145,14 @@ method stop*(self: LogosDelivery): Future[Result[void, string]] {.async.} = return ok() -method isOnline*(self: LogosDelivery): Future[Result[bool, string]] {.async.} = +proc isOnline*(self: LogosDelivery): Future[Result[bool, string]] {.async.} = if self.waku.isNil(): return err("Waku node is not initialized") - return ok(self.waku.healthMonitor.onlineMonitor.amIOnline()) + return await self.waku.isOnline() + +# Compile-time check that each concrete type satisfies its API concept. +static: + doAssert Waku is KernelApi + doAssert MessagingClient is MessagingApi + doAssert ReliableChannelManager is ReliableChannelApi + doAssert LogosDelivery is LogosDeliveryApi diff --git a/logos_delivery/messaging/api/send.nim b/logos_delivery/messaging/api/send.nim index 27c86d0ab..9a01c5f35 100644 --- a/logos_delivery/messaging/api/send.nim +++ b/logos_delivery/messaging/api/send.nim @@ -2,6 +2,7 @@ import results, chronos, chronicles import logos_delivery/api/types +import logos_delivery/api/events/messaging_client_events import logos_delivery/messaging/messaging_client import logos_delivery/waku/waku import logos_delivery/waku/api/subscriptions 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 1090be223..d2c369d46 100644 --- a/logos_delivery/messaging/delivery_service/recv_service/recv_service.nim +++ b/logos_delivery/messaging/delivery_service/recv_service/recv_service.nim @@ -11,9 +11,9 @@ import logos_delivery/waku/waku, logos_delivery/waku/api/[store, subscriptions] import - logos_delivery/api/kernel_api, # MessageSeenEvent - logos_delivery/api/messaging_client_api, # MessageReceivedEvent - logos_delivery/api/logos_delivery_api # EventConnectionStatusChange + logos_delivery/api/events/kernel_events, + # MessageSeenEvent + EventConnectionStatusChange + logos_delivery/api/events/messaging_client_events # MessageReceivedEvent const MaxMessageLife = chronos.minutes(7) ## Max time we will keep track of rx messages 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 f482a2aff..037ad879b 100644 --- a/logos_delivery/messaging/delivery_service/send_service/send_service.nim +++ b/logos_delivery/messaging/delivery_service/send_service/send_service.nim @@ -10,7 +10,7 @@ import logos_delivery/waku/[waku_core, waku_store/common], logos_delivery/waku/waku, logos_delivery/waku/api/[store, subscriptions, publish] -import logos_delivery/api/messaging_client_api +import logos_delivery/api/events/messaging_client_events logScope: topics = "send service" diff --git a/logos_delivery/messaging/messaging_client.nim b/logos_delivery/messaging/messaging_client.nim index 4e02c7a82..e82c6d6fd 100644 --- a/logos_delivery/messaging/messaging_client.nim +++ b/logos_delivery/messaging/messaging_client.nim @@ -17,7 +17,7 @@ type ## follow-up PR. Today it only carries the p2p reliability toggle. useP2PReliability*: bool - MessagingClient* = ref object of IMessagingClient + MessagingClient* = ref object ## Implements `MessagingApi`. waku*: Waku ## The Waku kernel this layer drives; read by `messaging/api/*`. sendService*: SendService recvService*: RecvService diff --git a/logos_delivery/waku/api/events/health_events.nim b/logos_delivery/waku/api/events/health_events.nim index fbf00debb..9c49402f7 100644 --- a/logos_delivery/waku/api/events/health_events.nim +++ b/logos_delivery/waku/api/events/health_events.nim @@ -4,10 +4,7 @@ import logos_delivery/api/types import logos_delivery/waku/node/health_monitor/[protocol_health, topic_health] import logos_delivery/waku/waku_core/topics -export protocol_health, topic_health - -# Note: `EventConnectionStatusChange` lives in `logos_delivery/api/logos_delivery_api` -# (the top-level orchestrator interface owns the node-connectivity event). +export event_broker, protocol_health, topic_health # Notify health changes to a subscribed topic # TODO: emit content topic health change events when subscribe/unsubscribe diff --git a/logos_delivery/waku/api/peer_manager.nim b/logos_delivery/waku/api/peer_manager.nim index 26de46594..3a73051b8 100644 --- a/logos_delivery/waku/api/peer_manager.nim +++ b/logos_delivery/waku/api/peer_manager.nim @@ -8,10 +8,7 @@ import libp2p/[peerid, peerstore] import logos_delivery/waku/waku import logos_delivery/waku/[waku_core, node/waku_node, node/peer_manager] -type PeerConnInfo* = object ## structured connected-peer info for the api boundary - peerId*: string - protocols*: seq[string] - addresses*: seq[string] +# `PeerConnInfo` is defined in `api/types` (surfaced here via `import waku`). proc connect*( self: Waku, peers: seq[string], timeoutMs: uint32 diff --git a/logos_delivery/waku/node/health_monitor/node_health_monitor.nim b/logos_delivery/waku/node/health_monitor/node_health_monitor.nim index f59e0d7e6..8a4aadb4b 100644 --- a/logos_delivery/waku/node/health_monitor/node_health_monitor.nim +++ b/logos_delivery/waku/node/health_monitor/node_health_monitor.nim @@ -9,7 +9,7 @@ import libp2p/protocols/pubsub, libp2p/protocols/pubsub/rpc/messages, logos_delivery/api/types, - logos_delivery/api/logos_delivery_api, # EventConnectionStatusChange + logos_delivery/api/events/kernel_events, # EventConnectionStatusChange logos_delivery/waku/[ waku_relay, api/events/health_events, diff --git a/logos_delivery/waku/node/subscription_manager.nim b/logos_delivery/waku/node/subscription_manager.nim index d0468749d..2c18a7b5e 100644 --- a/logos_delivery/waku/node/subscription_manager.nim +++ b/logos_delivery/waku/node/subscription_manager.nim @@ -22,7 +22,7 @@ import node/health_monitor/topic_health, node/health_monitor/connection_status, ] -import logos_delivery/api/kernel_api # MessageSeenEvent +import logos_delivery/api/events/kernel_events # MessageSeenEvent {.push raises: [].} diff --git a/logos_delivery/waku/node/waku_node.nim b/logos_delivery/waku/node/waku_node.nim index 3c2e41d8a..db90ab47c 100644 --- a/logos_delivery/waku/node/waku_node.nim +++ b/logos_delivery/waku/node/waku_node.nim @@ -62,7 +62,7 @@ import api/events/health_events, api/events/peer_events, ], - logos_delivery/api/kernel_api, # MessageSeenEvent + logos_delivery/api/events/kernel_events, # MessageSeenEvent logos_delivery/waku/discovery/waku_kademlia, logos_delivery/waku/net/[bound_ports, net_config], ./peer_manager, diff --git a/logos_delivery/waku/node/waku_node/relay.nim b/logos_delivery/waku/node/waku_node/relay.nim index ed039fb49..18a3f0f59 100644 --- a/logos_delivery/waku/node/waku_node/relay.nim +++ b/logos_delivery/waku/node/waku_node/relay.nim @@ -34,7 +34,7 @@ import node/subscription_manager, node/peer_manager, ] -import logos_delivery/api/kernel_api # MessageSeenEvent +import logos_delivery/api/events/kernel_events # MessageSeenEvent export waku_relay.WakuRelayHandler diff --git a/logos_delivery/waku/waku.nim b/logos_delivery/waku/waku.nim index 1dba0f633..29290483f 100644 --- a/logos_delivery/waku/waku.nim +++ b/logos_delivery/waku/waku.nim @@ -57,8 +57,8 @@ import ./factory/waku_conf, ./factory/waku_state_info -# Surfaces the Kernel API interface (and its `MessageSeenEvent`) to consumers -# of the Waku layer. +# Surfaces the Kernel API interface to consumers of the Waku layer. +# `MessageSeenEvent` now lives in `events/kernel_events` (surfaced by the concentrator). export kernel_api logScope: @@ -67,7 +67,7 @@ logScope: # Git version in git describe format (defined at compile time) const git_version* {.strdefine.} = "n/a" -type Waku* = ref object of IKernel +type Waku* = ref object ## Implements `KernelApi` (ops in `waku/api/*`). stateInfo*: WakuStateInfo conf*: WakuConf rng*: crypto.Rng diff --git a/tests/channels/test_reliable_channel_send_receive.nim b/tests/channels/test_reliable_channel_send_receive.nim index 3a0a48c06..c20f0d0b0 100644 --- a/tests/channels/test_reliable_channel_send_receive.nim +++ b/tests/channels/test_reliable_channel_send_receive.nim @@ -10,7 +10,7 @@ import ../testlib/[common, wakucore, wakunode, testasync] import logos_delivery import logos_delivery/waku/[waku_node, waku_core] import logos_delivery/waku/factory/waku_conf -import logos_delivery/api/messaging_client_api as waku_message_events +import logos_delivery/api/events/messaging_client_events as waku_message_events import tools/confutils/cli_args import logos_delivery/channels/reliable_channel_manager