Chore add api interfaces, phase3

apps/liteprotocoltester/statistics.nim

@@ -10,6 +10,7 @@ import
 
 from std/sugar import `=>`
 
+import logos_delivery/waku/compat/option_valueor
 import ./tester_message, ./lpt_metrics
 
 type

library/logos_delivery_api/node_api.nim

@@ -3,9 +3,9 @@ import chronos, chronicles, results, ffi
 import
   logos_delivery,
   logos_delivery/waku/node/waku_node,
-  logos_delivery/waku/events/message_events,
+  logos_delivery/api/messaging_client_api,
   logos_delivery/api/types,
-  logos_delivery/waku/events/[message_events, health_events],
+  logos_delivery/waku/events/health_events,
   tools/confutils/conf_from_json,
   ../declare_lib,
   ../json_event
@@ -116,7 +116,7 @@ proc logosdelivery_start_node(
     chronicles.error "MessageReceivedEvent.listen failed", err = $error
     return err("MessageReceivedEvent.listen failed: " & $error)
 
-  let ConnectionStatusChangeListener = EventConnectionStatusChange.listen(
+  discard EventConnectionStatusChange.listen(
     ctx.myLib[].waku.brokerCtx,
     proc(event: EventConnectionStatusChange) {.async: (raises: []).} =
       callEventCallback(ctx, "onConnectionStatusChange"):

logos_delivery/api/kernel_api.nim

@@ -0,0 +1,202 @@
+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_store/common as store_types
+
+export event_broker
+export api_types, pubsub_topic, store_types
+
+type IKernel* = ref object of RootObj
+
+EventBroker:
+  # Internal event emitted when a message arrives from the network via any protocol
+  type MessageSeenEvent* = object
+    topic*: PubsubTopic
+    message*: WakuMessage
+
+# --- 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")
+
+method buildPubsubTopic*(
+    self: IKernel, topicName: string
+): Future[Result[PubsubTopic, string]] {.async: (raises: []), base.} =
+  return err("Interface IKernel.buildPubsubTopic not implemented")
+
+method defaultPubsubTopic*(
+    self: IKernel
+): Future[Result[PubsubTopic, string]] {.async: (raises: []), base.} =
+  return err("Interface IKernel.defaultPubsubTopic not implemented")
+
+# --- 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")
+
+method relaySubscribe*(
+    self: IKernel, pubsubTopic: PubsubTopic
+): Future[Result[bool, string]] {.async: (raises: []), base.} =
+  return err("Interface IKernel.relaySubscribe not implemented")
+
+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")

logos_delivery/api/logos_delivery_api.nim

@@ -0,0 +1,33 @@
+## `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.
+
+{.push raises: [].}
+
+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")

logos_delivery/api/messaging_client_api.nim

@@ -0,0 +1,48 @@
+import chronos, results
+import brokers/event_broker
+
+import logos_delivery/api/types as api_types
+
+export event_broker, 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")

logos_delivery/api/reliable_channel_manager_api.nim

@@ -0,0 +1,62 @@
+import chronos, results
+
+import brokers/event_broker
+
+import logos_delivery/api/types as api_types
+
+export event_broker, api_types
+
+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")