From 53c83184d0e0d84a1b053185ab1862521122aef7 Mon Sep 17 00:00:00 2001 From: Ivan FB Date: Thu, 25 Jun 2026 04:18:56 +0200 Subject: [PATCH] FFI: typed kernel_api operations (nim-ffi v0.2.0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Convert the kernel_api operations — relay/filter/lightpush/store, peer_manager, discovery, ping, debug/node-info — to typed {.ffi.} procs that pass parameters directly and ride generic CBOR (no per-op request wrappers). Co-Authored-By: Claude Opus 4.8 --- library/kernel_api/debug_node_api.nim | 73 ++---- library/kernel_api/discovery_api.nim | 117 ++------- library/kernel_api/node_info_api.nim | 34 +++ library/kernel_api/peer_manager_api.nim | 177 +++++-------- library/kernel_api/ping_api.nim | 51 +--- library/kernel_api/protocols/filter_api.nim | 136 +++------- .../kernel_api/protocols/lightpush_api.nim | 59 +---- library/kernel_api/protocols/relay_api.nim | 234 ++++++------------ library/kernel_api/protocols/store_api.nim | 57 ++--- 9 files changed, 286 insertions(+), 652 deletions(-) create mode 100644 library/kernel_api/node_info_api.nim diff --git a/library/kernel_api/debug_node_api.nim b/library/kernel_api/debug_node_api.nim index 44764a673..0fc254c35 100644 --- a/library/kernel_api/debug_node_api.nim +++ b/library/kernel_api/debug_node_api.nim @@ -1,53 +1,32 @@ -import std/json -import - chronicles, - chronos, - results, - eth/p2p/discoveryv5/enr, - strutils, - libp2p/peerid, - metrics, - ffi -import - logos_delivery/waku/waku, - logos_delivery/waku/node/waku_node, - logos_delivery/waku/node/health_monitor, - library/declare_lib +## The waku api getters are synchronous and can't fail, so the bodies just wrap +## the value; the `{.ffi.}` macro wraps it into the `Future` it must expose. -proc getMultiaddresses(node: WakuNode): seq[string] = - return node.info().listenAddresses +proc version*( + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = + return ok(self.waku.version()) -proc getMetrics(): string = - {.gcsafe.}: - return defaultRegistry.toText() ## defaultRegistry is {.global.} in metrics module +proc listen_addresses*( + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = + return ok(self.waku.listenAddresses().join(",")) -proc waku_version( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffi.} = - return ok(WakuNodeVersionString) +proc get_my_enr*( + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = + return ok(self.waku.myEnr()) -proc waku_listen_addresses( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffi.} = - ## returns a comma-separated string of the listen addresses - return ok(ctx.myLib[].waku.node.getMultiaddresses().join(",")) +proc get_my_peerid*( + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = + return ok(self.waku.myPeerId()) -proc waku_get_my_enr( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffi.} = - return ok(ctx.myLib[].waku.node.enr.toURI()) +proc get_metrics*( + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = + return ok(self.waku.metrics()) -proc waku_get_my_peerid( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffi.} = - return ok($ctx.myLib[].waku.node.peerId()) - -proc waku_get_metrics( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffi.} = - return ok(getMetrics()) - -proc waku_is_online( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffi.} = - return ok($ctx.myLib[].waku.healthMonitor.onlineMonitor.amIOnline()) +proc is_online*( + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = + return ok($self.waku.isOnline()) diff --git a/library/kernel_api/discovery_api.nim b/library/kernel_api/discovery_api.nim index 158c3a925..a5b069a25 100644 --- a/library/kernel_api/discovery_api.nim +++ b/library/kernel_api/discovery_api.nim @@ -1,96 +1,31 @@ -import logos_delivery/waku/compat/option_valueor -import std/json -import chronos, chronicles, results, strutils, libp2p/multiaddress, ffi -import - logos_delivery/waku/waku, - logos_delivery/waku/discovery/waku_dnsdisc, - logos_delivery/waku/discovery/waku_discv5, - logos_delivery/waku/waku_core/peers, - logos_delivery/waku/waku_node, - library/declare_lib +proc discv5_update_bootnodes*( + self: LogosDelivery, bootnodes: string +): Future[Result[string, string]] {.ffi.} = + ## `bootnodes` is a JSON array of ENRs, e.g. `["enr:...", "enr:..."]`. + (await self.waku.discv5UpdateBootnodes(bootnodes)).isOkOr: + return err(error) + return ok("") -proc retrieveBootstrapNodes( - enrTreeUrl: string, ipDnsServer: string -): Future[Result[seq[string], string]] {.async.} = - let dnsNameServers = @[parseIpAddress(ipDnsServer)] - let discoveredPeers: seq[RemotePeerInfo] = ( - await retrieveDynamicBootstrapNodes(enrTreeUrl, dnsNameServers) - ).valueOr: - return err("failed discovering peers from DNS: " & $error) - - var multiAddresses = newSeq[string]() - - for discPeer in discoveredPeers: - for address in discPeer.addrs: - multiAddresses.add($address & "/p2p/" & $discPeer) - - return ok(multiAddresses) - -proc updateDiscv5BootstrapNodes(nodes: string, waku: Waku): Result[void, string] = - waku.wakuDiscv5.updateBootstrapRecords(nodes).isOkOr: - return err("error in updateDiscv5BootstrapNodes: " & $error) - return ok() - -proc performPeerExchangeRequestTo*( - numPeers: uint64, waku: Waku -): Future[Result[int, string]] {.async.} = - let numPeersRecv = (await waku.node.fetchPeerExchangePeers(numPeers)).valueOr: - return err($error) - return ok(numPeersRecv) - -proc waku_discv5_update_bootnodes( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - bootnodes: cstring, -) {.ffi.} = - ## Updates the bootnode list used for discovering new peers via DiscoveryV5 - ## bootnodes - JSON array containing the bootnode ENRs i.e. `["enr:...", "enr:..."]` - - updateDiscv5BootstrapNodes($bootnodes, ctx.myLib[].waku).isOkOr: - error "UPDATE_DISCV5_BOOTSTRAP_NODES failed", error = error - return err($error) - - return ok("discovery request processed correctly") - -proc waku_dns_discovery( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - enrTreeUrl: cstring, - nameDnsServer: cstring, - timeoutMs: cint, -) {.ffi.} = - let nodes = (await retrieveBootstrapNodes($enrTreeUrl, $nameDnsServer)).valueOr: - error "GET_BOOTSTRAP_NODES failed", error = error - return err($error) - - ## returns a comma-separated string of bootstrap nodes' multiaddresses +proc dns_discovery*( + self: LogosDelivery, enrTreeUrl: string, nameDnsServer: string, timeoutMs: int +): Future[Result[string, string]] {.ffi.} = + let nodes = (await self.waku.dnsDiscovery(enrTreeUrl, nameDnsServer, timeoutMs)).valueOr: + return err(error) return ok(nodes.join(",")) -proc waku_start_discv5( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffi.} = - (await ctx.myLib[].waku.wakuDiscv5.start()).isOkOr: - error "START_DISCV5 failed", error = error - return err("error starting discv5: " & $error) +proc start_discv5*(self: LogosDelivery): Future[Result[string, string]] {.ffi.} = + (await self.waku.startDiscv5()).isOkOr: + return err(error) + return ok("") - return ok("discv5 started correctly") +proc stop_discv5*(self: LogosDelivery): Future[Result[string, string]] {.ffi.} = + (await self.waku.stopDiscv5()).isOkOr: + return err(error) + return ok("") -proc waku_stop_discv5( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffi.} = - await ctx.myLib[].waku.wakuDiscv5.stop() - return ok("discv5 stopped correctly") - -proc waku_peer_exchange_request( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - numPeers: uint64, -) {.ffi.} = - let numValidPeers = (await performPeerExchangeRequestTo(numPeers, ctx.myLib[].waku)).valueOr: - error "waku_peer_exchange_request failed", error = error - return err("failed peer exchange: " & $error) - - return ok($numValidPeers) +proc peer_exchange_request*( + self: LogosDelivery, numPeers: uint64 +): Future[Result[string, string]] {.ffi.} = + let n = (await self.waku.peerExchangeRequest(numPeers)).valueOr: + return err(error) + return ok($n) diff --git a/library/kernel_api/node_info_api.nim b/library/kernel_api/node_info_api.nim new file mode 100644 index 000000000..3c82b06fd --- /dev/null +++ b/library/kernel_api/node_info_api.nim @@ -0,0 +1,34 @@ +import std/json +import logos_delivery/waku/factory/waku_state_info +import tools/confutils/[cli_args, config_option_meta] + +proc get_available_node_info_ids*( + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = + ## All node-info item ids that can be queried with `get_node_info`. + return ok($self.waku.stateInfo.getAllPossibleInfoItemIds()) + +proc get_node_info*( + self: LogosDelivery, nodeInfoId: string +): Future[Result[string, string]] {.ffi.} = + let infoItemIdEnum = + try: + parseEnum[NodeInfoId](nodeInfoId) + except ValueError: + return err("Invalid node info id: " & nodeInfoId) + return ok(self.waku.stateInfo.getNodeInfoItem(infoItemIdEnum)) + +proc get_available_configs*( + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = + let optionMetas: seq[ConfigOptionMeta] = extractConfigOptionMeta(WakuNodeConf) + var configOptionDetails = newJArray() + for meta in optionMetas: + configOptionDetails.add( + %*{ + meta.fieldName: meta.typeName & "(" & meta.defaultValue & ")", "desc": meta.desc + } + ) + var jsonNode = newJObject() + jsonNode["configOptions"] = configOptionDetails + return ok(pretty(jsonNode)) diff --git a/library/kernel_api/peer_manager_api.nim b/library/kernel_api/peer_manager_api.nim index eeea2c63f..b5d461ae5 100644 --- a/library/kernel_api/peer_manager_api.nim +++ b/library/kernel_api/peer_manager_api.nim @@ -1,133 +1,76 @@ -import logos_delivery/waku/compat/option_valueor -import std/[sequtils, strutils, tables] -import chronicles, chronos, results, options, json, ffi -import - logos_delivery/waku/waku, - logos_delivery/waku/node/waku_node, - logos_delivery/waku/node/peer_manager, - library/declare_lib +import std/sequtils -type PeerInfo = object - protocols: seq[string] - addresses: seq[string] +type ConnectedPeersInfoResponse {.ffi.} = object + peers: seq[PeerConnInfoFFI] -proc waku_get_peerids_from_peerstore( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffi.} = - ## returns a comma-separated string of peerIDs - let peerIDs = ctx.myLib[].waku.node.peerManager.switch.peerStore - .peers() - .mapIt($it.peerId) - .join(",") - return ok(peerIDs) +proc get_peerids_from_peerstore*( + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = + let ids = (await self.waku.peerIdsFromPeerstore()).valueOr: + return err(error) + return ok(ids.join(",")) -proc waku_connect( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - peerMultiAddr: cstring, - timeoutMs: cuint, -) {.ffi.} = - let peers = ($peerMultiAddr).split(",").mapIt(strip(it)) - await ctx.myLib[].waku.node.connectToNodes(peers, source = "static") +proc connect_peers*( + self: LogosDelivery, peers: seq[string], timeoutMs: uint32 +): Future[Result[string, string]] {.ffi.} = + ## `peers` are multiaddrs. + (await self.waku.connect(peers, timeoutMs)).isOkOr: + return err(error) return ok("") -proc waku_disconnect_peer_by_id( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - peerId: cstring, -) {.ffi.} = - let pId = PeerId.init($peerId).valueOr: - error "DISCONNECT_PEER_BY_ID failed", error = $error - return err($error) - await ctx.myLib[].waku.node.peerManager.disconnectNode(pId) +proc disconnect_peer_by_id*( + self: LogosDelivery, peerId: string +): Future[Result[string, string]] {.ffi.} = + (await self.waku.disconnectPeerById(peerId)).isOkOr: + return err(error) return ok("") -proc waku_disconnect_all_peers( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffi.} = - await ctx.myLib[].waku.node.peerManager.disconnectAllPeers() +proc disconnect_all_peers*( + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = + (await self.waku.disconnectAllPeers()).isOkOr: + return err(error) return ok("") -proc waku_dial_peer( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - peerMultiAddr: cstring, - protocol: cstring, - timeoutMs: cuint, -) {.ffi.} = - let remotePeerInfo = parsePeerInfo($peerMultiAddr).valueOr: - error "DIAL_PEER failed", error = $error - return err($error) - let conn = await ctx.myLib[].waku.node.peerManager.dialPeer(remotePeerInfo, $protocol) - if conn.isNone(): - let msg = "failed dialing peer" - error "DIAL_PEER failed", error = msg, peerId = $remotePeerInfo.peerId - return err(msg) +proc dial_peer*( + self: LogosDelivery, peer: string, protocol: string, timeoutMs: int +): Future[Result[string, string]] {.ffi.} = + ## `peer` is a multiaddr. + (await self.waku.dialPeer(peer, protocol, timeoutMs)).isOkOr: + return err(error) return ok("") -proc waku_dial_peer_by_id( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - peerId: cstring, - protocol: cstring, - timeoutMs: cuint, -) {.ffi.} = - let pId = PeerId.init($peerId).valueOr: - error "DIAL_PEER_BY_ID failed", error = $error - return err($error) - let conn = await ctx.myLib[].waku.node.peerManager.dialPeer(pId, $protocol) - if conn.isNone(): - let msg = "failed dialing peer" - error "DIAL_PEER_BY_ID failed", error = msg, peerId = $peerId - return err(msg) - +proc dial_peer_by_id*( + self: LogosDelivery, peer: string, protocol: string, timeoutMs: int +): Future[Result[string, string]] {.ffi.} = + ## `peer` is a peerId. + (await self.waku.dialPeerById(peer, protocol, timeoutMs)).isOkOr: + return err(error) return ok("") -proc waku_get_connected_peers_info( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffi.} = - ## returns a JSON string mapping peerIDs to objects with protocols and addresses - - var peersMap = initTable[string, PeerInfo]() - let peers = ctx.myLib[].waku.node.peerManager.switch.peerStore.peers().filterIt( - it.connectedness == Connected +proc get_connected_peers_info*( + self: LogosDelivery +): Future[Result[ConnectedPeersInfoResponse, string]] {.ffi.} = + let infos = (await self.waku.connectedPeersInfo()).valueOr: + return err(error) + return ok( + ConnectedPeersInfoResponse( + peers: infos.mapIt( + PeerConnInfoFFI( + peerId: it.peerId, protocols: it.protocols, addresses: it.addresses + ) + ) ) + ) - # Build a map of peer IDs to peer info objects - for peer in peers: - let peerIdStr = $peer.peerId - peersMap[peerIdStr] = - PeerInfo(protocols: peer.protocols, addresses: peer.addrs.mapIt($it)) +proc get_connected_peers*(self: LogosDelivery): Future[Result[string, string]] {.ffi.} = + let ids = (await self.waku.connectedPeers()).valueOr: + return err(error) + return ok(ids.join(",")) - # Convert the map to JSON string - let jsonObj = %*peersMap - let jsonStr = $jsonObj - return ok(jsonStr) - -proc waku_get_connected_peers( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffi.} = - ## returns a comma-separated string of peerIDs - let - (inPeerIds, outPeerIds) = ctx.myLib[].waku.node.peerManager.connectedPeers() - connectedPeerids = concat(inPeerIds, outPeerIds) - - return ok(connectedPeerids.mapIt($it).join(",")) - -proc waku_get_peerids_by_protocol( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - protocol: cstring, -) {.ffi.} = - ## returns a comma-separated string of peerIDs that mount the given protocol - let connectedPeers = ctx.myLib[].waku.node.peerManager.switch.peerStore - .peers($protocol) - .filterIt(it.connectedness == Connected) - .mapIt($it.peerId) - .join(",") - return ok(connectedPeers) +proc get_peerids_by_protocol*( + self: LogosDelivery, protocol: string +): Future[Result[string, string]] {.ffi.} = + let ids = (await self.waku.peerIdsByProtocol(protocol)).valueOr: + return err(error) + return ok(ids.join(",")) diff --git a/library/kernel_api/ping_api.nim b/library/kernel_api/ping_api.nim index e6ed69dd1..58d607052 100644 --- a/library/kernel_api/ping_api.nim +++ b/library/kernel_api/ping_api.nim @@ -1,44 +1,7 @@ -import std/[json, strutils] -import chronos, results, ffi -import libp2p/[protocols/ping, switch, multiaddress, multicodec] -import logos_delivery/waku/[waku, waku_core/peers, node/waku_node], library/declare_lib - -proc waku_ping_peer( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - peerAddr: cstring, - timeoutMs: cuint, -) {.ffi.} = - let peerInfo = peers.parsePeerInfo(($peerAddr).split(",")).valueOr: - return err("PingRequest failed to parse peer addr: " & $error) - - let timeout = chronos.milliseconds(timeoutMs) - proc ping(): Future[Result[Duration, string]] {.async, gcsafe.} = - try: - let conn = await ctx.myLib[].waku.node.switch.dial( - peerInfo.peerId, peerInfo.addrs, PingCodec - ) - defer: - await conn.close() - - let pingRTT = await ctx.myLib[].waku.node.libp2pPing.ping(conn) - if pingRTT == 0.nanos: - return err("could not ping peer: rtt-0") - return ok(pingRTT) - except CatchableError as exc: - return err("could not ping peer: " & exc.msg) - - let pingFuture = ping() - let pingRTT: Duration = - if timeout == chronos.milliseconds(0): # No timeout expected - (await pingFuture).valueOr: - return err("ping failed, no timeout expected: " & error) - else: - let timedOut = not (await pingFuture.withTimeout(timeout)) - if timedOut: - return err("ping timed out") - pingFuture.read().valueOr: - return err("failed to read ping future: " & error) - - return ok($(pingRTT.nanos)) +proc ping_peer*( + self: LogosDelivery, peerAddr: string, timeoutMs: int +): Future[Result[string, string]] {.ffi.} = + ## Returns the round-trip time in nanoseconds. + let rtt = (await self.waku.pingPeer(peerAddr, timeoutMs)).valueOr: + return err(error) + return ok($rtt) diff --git a/library/kernel_api/protocols/filter_api.nim b/library/kernel_api/protocols/filter_api.nim index a070bd2c7..fd189ac79 100644 --- a/library/kernel_api/protocols/filter_api.nim +++ b/library/kernel_api/protocols/filter_api.nim @@ -1,109 +1,39 @@ -import logos_delivery/waku/compat/option_valueor -import options, std/[strutils, sequtils] -import chronicles, chronos, results, ffi -import - logos_delivery/waku/waku_filter_v2/client, - logos_delivery/waku/waku_core/message/message, - logos_delivery/waku/waku, - logos_delivery/waku/waku_relay, - logos_delivery/waku/waku_filter_v2/common, - logos_delivery/waku/waku_core/subscription/push_handler, - logos_delivery/waku/node/peer_manager/peer_manager, - logos_delivery/waku/waku_node, - logos_delivery/waku/waku_core/topics/pubsub_topic, - logos_delivery/waku/waku_core/topics/content_topic, - library/events/json_message_event, - library/declare_lib +import std/sequtils +import logos_delivery/waku/waku_core/subscription/push_handler -const FilterOpTimeout = 5.seconds - -proc checkFilterClientMounted(waku: Waku): Result[string, string] = - if waku.node.wakuFilterClient.isNil(): - let errorMsg = "wakuFilterClient is not mounted" - error "fail filter process", error = errorMsg - return err(errorMsg) +proc filter_subscribe*( + self: LogosDelivery, pubsubTopic: string, contentTopics: seq[string] +): Future[Result[string, string]] {.ffi.} = + # `filterSubscribe` re-registers the filter push handler, so it must keep + # feeding MessageSeenEvent — the single source the ctor's listener delivers + # to the foreign side (see liblogosdelivery.nim). + let brokerCtx = self.waku.brokerCtx + let pushHandler = proc(pubsubTopic: PubsubTopic, msg: WakuMessage) {.async.} = + MessageSeenEvent.emit(brokerCtx, pubsubTopic, msg) + ( + await self.waku.filterSubscribe( + PubsubTopic(pubsubTopic), + contentTopics.mapIt(ContentTopic(it)), + FilterPushHandler(pushHandler), + ) + ).isOkOr: + return err(error) return ok("") -proc waku_filter_subscribe( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - pubSubTopic: cstring, - contentTopics: cstring, -) {.ffi.} = - proc onReceivedMessage(ctx: ptr FFIContext): WakuRelayHandler = - return proc(pubsubTopic: PubsubTopic, msg: WakuMessage) {.async.} = - callEventCallback(ctx, "onReceivedMessage"): - $JsonMessageEvent.new(pubsubTopic, msg) - - checkFilterClientMounted(ctx.myLib[].waku).isOkOr: - return err($error) - - var filterPushEventCallback = FilterPushHandler(onReceivedMessage(ctx)) - ctx.myLib[].waku.node.wakuFilterClient.registerPushHandler(filterPushEventCallback) - - let peer = ctx.myLib[].waku.node.peerManager.selectPeer(WakuFilterSubscribeCodec).valueOr: - let errorMsg = "could not find peer with WakuFilterSubscribeCodec when subscribing" - error "fail filter subscribe", error = errorMsg - return err(errorMsg) - - let subFut = ctx.myLib[].waku.node.filterSubscribe( - some(PubsubTopic($pubsubTopic)), - ($contentTopics).split(",").mapIt(ContentTopic(it)), - peer, - ) - if not await subFut.withTimeout(FilterOpTimeout): - let errorMsg = "filter subscription timed out" - error "fail filter unsubscribe", error = errorMsg - - return err(errorMsg) - +proc filter_unsubscribe*( + self: LogosDelivery, pubsubTopic: string, contentTopics: seq[string] +): Future[Result[string, string]] {.ffi.} = + ( + await self.waku.filterUnsubscribe( + PubsubTopic(pubsubTopic), contentTopics.mapIt(ContentTopic(it)) + ) + ).isOkOr: + return err(error) return ok("") -proc waku_filter_unsubscribe( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - pubSubTopic: cstring, - contentTopics: cstring, -) {.ffi.} = - checkFilterClientMounted(ctx.myLib[].waku).isOkOr: - return err($error) - - let peer = ctx.myLib[].waku.node.peerManager.selectPeer(WakuFilterSubscribeCodec).valueOr: - let errorMsg = - "could not find peer with WakuFilterSubscribeCodec when unsubscribing" - error "fail filter process", error = errorMsg - return err(errorMsg) - - let subFut = ctx.myLib[].waku.node.filterUnsubscribe( - some(PubsubTopic($pubsubTopic)), - ($contentTopics).split(",").mapIt(ContentTopic(it)), - peer, - ) - if not await subFut.withTimeout(FilterOpTimeout): - let errorMsg = "filter un-subscription timed out" - error "fail filter unsubscribe", error = errorMsg - return err(errorMsg) - return ok("") - -proc waku_filter_unsubscribe_all( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffi.} = - checkFilterClientMounted(ctx.myLib[].waku).isOkOr: - return err($error) - - let peer = ctx.myLib[].waku.node.peerManager.selectPeer(WakuFilterSubscribeCodec).valueOr: - let errorMsg = - "could not find peer with WakuFilterSubscribeCodec when unsubscribing all" - error "fail filter unsubscribe all", error = errorMsg - return err(errorMsg) - - let unsubFut = ctx.myLib[].waku.node.filterUnsubscribeAll(peer) - - if not await unsubFut.withTimeout(FilterOpTimeout): - let errorMsg = "filter un-subscription all timed out" - error "fail filter unsubscribe all", error = errorMsg - - return err(errorMsg) +proc filter_unsubscribe_all*( + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = + (await self.waku.filterUnsubscribeAll()).isOkOr: + return err(error) return ok("") diff --git a/library/kernel_api/protocols/lightpush_api.nim b/library/kernel_api/protocols/lightpush_api.nim index eb0d1de09..25a4d2852 100644 --- a/library/kernel_api/protocols/lightpush_api.nim +++ b/library/kernel_api/protocols/lightpush_api.nim @@ -1,52 +1,7 @@ -import logos_delivery/waku/compat/option_valueor -import options, std/[json, strformat] -import chronicles, chronos, results, ffi -import - logos_delivery/waku/waku_core/message/message, - logos_delivery/waku/waku_core/codecs, - logos_delivery/waku/waku, - logos_delivery/waku/waku_core/message, - logos_delivery/waku/waku_core/topics/pubsub_topic, - logos_delivery/waku/waku_lightpush_legacy/client, - logos_delivery/waku/node/peer_manager/peer_manager, - library/events/json_message_event, - library/declare_lib - -proc waku_lightpush_publish( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - pubSubTopic: cstring, - jsonWakuMessage: cstring, -) {.ffi.} = - if ctx.myLib[].waku.node.wakuLightpushClient.isNil(): - let errorMsg = "LightpushRequest waku.node.wakuLightpushClient is nil" - error "PUBLISH failed", error = errorMsg - return err(errorMsg) - - var jsonMessage: JsonMessage - try: - let jsonContent = parseJson($jsonWakuMessage) - jsonMessage = JsonMessage.fromJsonNode(jsonContent).valueOr: - raise newException(JsonParsingError, $error) - except JsonParsingError as exc: - return err(fmt"Error parsing json message: {exc.msg}") - - let msg = json_message_event.toWakuMessage(jsonMessage).valueOr: - return err("Problem building the WakuMessage: " & $error) - - let peerOpt = ctx.myLib[].waku.node.peerManager.selectPeer(WakuLightPushCodec) - if peerOpt.isNone(): - let errorMsg = "failed to lightpublish message, no suitable remote peers" - error "PUBLISH failed", error = errorMsg - return err(errorMsg) - - let msgHashHex = ( - await ctx.myLib[].waku.node.wakuLegacyLightpushClient.publish( - $pubsubTopic, msg, peer = peerOpt.get() - ) - ).valueOr: - error "PUBLISH failed", error = error - return err($error) - - return ok(msgHashHex) +proc lightpush_publish*( + self: LogosDelivery, pubsubTopic: string, message: WakuMessage +): Future[Result[string, string]] {.ffi.} = + ## Returns the published message hash. + let hash = (await self.waku.lightpushPublish(PubsubTopic(pubsubTopic), message)).valueOr: + return err(error) + return ok(hash) diff --git a/library/kernel_api/protocols/relay_api.nim b/library/kernel_api/protocols/relay_api.nim index d580597ae..66c2c3103 100644 --- a/library/kernel_api/protocols/relay_api.nim +++ b/library/kernel_api/protocols/relay_api.nim @@ -1,175 +1,85 @@ -import logos_delivery/waku/compat/option_valueor -import std/[net, sequtils, strutils, json], strformat -import chronicles, chronos, stew/byteutils, results, ffi -import - logos_delivery/waku/waku_core/message/message, - logos_delivery/waku/factory/validator_signed, - logos_delivery/waku/waku, - tools/confutils/cli_args, - logos_delivery/waku/waku_core/message, - logos_delivery/waku/waku_core/topics/pubsub_topic, - logos_delivery/waku/waku_core/topics, - logos_delivery/waku/node/waku_node/relay, - logos_delivery/waku/waku_relay/protocol, - logos_delivery/waku/node/peer_manager, - library/events/json_message_event, - library/declare_lib +proc relay_get_peers_in_mesh*( + self: LogosDelivery, pubsubTopic: string +): Future[Result[string, string]] {.ffi.} = + let peers = (await self.waku.relayPeersInMesh(PubsubTopic(pubsubTopic))).valueOr: + return err(error) + return ok(peers.join(",")) -proc waku_relay_get_peers_in_mesh( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - pubSubTopic: cstring, -) {.ffi.} = - let meshPeers = ctx.myLib[].waku.node.wakuRelay.getPeersInMesh($pubsubTopic).valueOr: - error "LIST_MESH_PEERS failed", error = error - return err($error) - ## returns a comma-separated string of peerIDs - return ok(meshPeers.mapIt($it).join(",")) +proc relay_get_num_peers_in_mesh*( + self: LogosDelivery, pubsubTopic: string +): Future[Result[string, string]] {.ffi.} = + let n = (await self.waku.relayNumPeersInMesh(PubsubTopic(pubsubTopic))).valueOr: + return err(error) + return ok($n) -proc waku_relay_get_num_peers_in_mesh( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - pubSubTopic: cstring, -) {.ffi.} = - let numPeersInMesh = ctx.myLib[].waku.node.wakuRelay.getNumPeersInMesh($pubsubTopic).valueOr: - error "NUM_MESH_PEERS failed", error = error - return err($error) - return ok($numPeersInMesh) +proc relay_get_connected_peers*( + self: LogosDelivery, pubsubTopic: string +): Future[Result[string, string]] {.ffi.} = + let peers = (await self.waku.relayConnectedPeers(PubsubTopic(pubsubTopic))).valueOr: + return err(error) + return ok(peers.join(",")) -proc waku_relay_get_connected_peers( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - pubSubTopic: cstring, -) {.ffi.} = - ## Returns the list of all connected peers to an specific pubsub topic - let connPeers = ctx.myLib[].waku.node.wakuRelay.getConnectedPeers($pubsubTopic).valueOr: - error "LIST_CONNECTED_PEERS failed", error = error - return err($error) - ## returns a comma-separated string of peerIDs - return ok(connPeers.mapIt($it).join(",")) - -proc waku_relay_get_num_connected_peers( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - pubSubTopic: cstring, -) {.ffi.} = - let numConnPeers = ctx.myLib[].waku.node.wakuRelay.getNumConnectedPeers($pubsubTopic).valueOr: - error "NUM_CONNECTED_PEERS failed", error = error - return err($error) - return ok($numConnPeers) - -proc waku_relay_add_protected_shard( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - clusterId: cint, - shardId: cint, - publicKey: cstring, -) {.ffi.} = - ## Protects a shard with a public key - try: - let relayShard = RelayShard(clusterId: uint16(clusterId), shardId: uint16(shardId)) - let protectedShard = ProtectedShard.parseCmdArg($relayShard & ":" & $publicKey) - ctx.myLib[].waku.node.wakuRelay.addSignedShardsValidator( - @[protectedShard], uint16(clusterId) - ) - except ValueError as exc: - return err("ERROR in waku_relay_add_protected_shard: " & exc.msg) +proc relay_get_num_connected_peers*( + self: LogosDelivery, pubsubTopic: string +): Future[Result[string, string]] {.ffi.} = + let n = (await self.waku.relayNumConnectedPeers(PubsubTopic(pubsubTopic))).valueOr: + return err(error) + return ok($n) +proc relay_add_protected_shard*( + self: LogosDelivery, clusterId: uint16, shardId: uint16, publicKey: string +): Future[Result[string, string]] {.ffi.} = + (await self.waku.relayAddProtectedShard(clusterId, shardId, publicKey)).isOkOr: + return err(error) return ok("") -proc waku_relay_subscribe( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - pubSubTopic: cstring, -) {.ffi.} = - echo "Subscribing to topic: " & $pubSubTopic & " ..." - proc onReceivedMessage(ctx: ptr FFIContext[LogosDelivery]): WakuRelayHandler = - return proc(pubsubTopic: PubsubTopic, msg: WakuMessage) {.async.} = - callEventCallback(ctx, "onReceivedMessage"): - $JsonMessageEvent.new(pubsubTopic, msg) - - var cb = onReceivedMessage(ctx) - - ctx.myLib[].waku.node.subscribe( - (kind: SubscriptionKind.PubsubSub, topic: $pubsubTopic), - handler = WakuRelayHandler(cb), - ).isOkOr: - error "SUBSCRIBE failed", error = error - return err($error) +proc relay_subscribe*( + self: LogosDelivery, pubsubTopic: string +): Future[Result[string, string]] {.ffi.} = + # Just establishes the subscription; delivery flows through the global + # MessageSeenEvent listener (see the ctor in liblogosdelivery.nim). + (await self.waku.relaySubscribe(PubsubTopic(pubsubTopic))).isOkOr: + return err(error) return ok("") -proc waku_relay_unsubscribe( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - pubSubTopic: cstring, -) {.ffi.} = - ctx.myLib[].waku.node.unsubscribe( - (kind: SubscriptionKind.PubsubSub, topic: $pubsubTopic) - ).isOkOr: - error "UNSUBSCRIBE failed", error = error - return err($error) - +proc relay_unsubscribe*( + self: LogosDelivery, pubsubTopic: string +): Future[Result[string, string]] {.ffi.} = + (await self.waku.relayUnsubscribe(PubsubTopic(pubsubTopic))).isOkOr: + return err(error) return ok("") -proc waku_relay_publish( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - pubSubTopic: cstring, - jsonWakuMessage: cstring, - timeoutMs: cuint, -) {.ffi.} = - var - # https://rfc.vac.dev/spec/36/#extern-char-waku_relay_publishchar-messagejson-char-pubsubtopic-int-timeoutms - jsonMessage: JsonMessage - try: - let jsonContent = parseJson($jsonWakuMessage) - jsonMessage = JsonMessage.fromJsonNode(jsonContent).valueOr: - raise newException(JsonParsingError, $error) - except JsonParsingError as exc: - return err(fmt"Error parsing json message: {exc.msg}") +proc relay_publish*( + self: LogosDelivery, pubsubTopic: string, message: WakuMessage, timeoutMs: uint32 +): Future[Result[string, string]] {.ffi.} = + ## Returns the published message hash (0x-hex). + let hash = ( + await self.waku.relayPublish(PubsubTopic(pubsubTopic), message, timeoutMs) + ).valueOr: + return err(error) + return ok(hash) - let msg = json_message_event.toWakuMessage(jsonMessage).valueOr: - return err("Problem building the WakuMessage: " & $error) +proc relay_default_pubsub_topic*( + self: LogosDelivery +): Future[Result[string, string]] {.ffi.} = + return ok(string(self.waku.defaultPubsubTopic())) - (await ctx.myLib[].waku.node.wakuRelay.publish($pubsubTopic, msg)).isOkOr: - error "PUBLISH failed", error = error - return err($error) +proc relay_content_topic*( + self: LogosDelivery, + appName: string, + appVersion: uint32, + contentTopicName: string, + encoding: string, +): Future[Result[string, string]] {.ffi.} = + let contentTopic = self.waku.buildContentTopic( + appName, appVersion, contentTopicName, encoding + ).valueOr: + return err(error) + return ok(string(contentTopic)) - let msgHash = computeMessageHash($pubSubTopic, msg).to0xHex - return ok(msgHash) - -proc waku_default_pubsub_topic( - ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer -) {.ffi.} = - # https://rfc.vac.dev/spec/36/#extern-char-waku_default_pubsub_topic - return ok(DefaultPubsubTopic) - -proc waku_content_topic( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - appName: cstring, - appVersion: cuint, - contentTopicName: cstring, - encoding: cstring, -) {.ffi.} = - # https://rfc.vac.dev/spec/36/#extern-char-waku_content_topicchar-applicationname-unsigned-int-applicationversion-char-contenttopicname-char-encoding - - return ok(fmt"/{$appName}/{$appVersion}/{$contentTopicName}/{$encoding}") - -proc waku_pubsub_topic( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - topicName: cstring, -) {.ffi.} = - # https://rfc.vac.dev/spec/36/#extern-char-waku_pubsub_topicchar-name-char-encoding - return ok(fmt"/waku/2/{$topicName}") +proc relay_pubsub_topic*( + self: LogosDelivery, topicName: string +): Future[Result[string, string]] {.ffi.} = + let pubsubTopic = self.waku.buildPubsubTopic(topicName).valueOr: + return err(error) + return ok(string(pubsubTopic)) diff --git a/library/kernel_api/protocols/store_api.nim b/library/kernel_api/protocols/store_api.nim index c6356bf6a..9543772c7 100644 --- a/library/kernel_api/protocols/store_api.nim +++ b/library/kernel_api/protocols/store_api.nim @@ -1,17 +1,12 @@ -import logos_delivery/waku/compat/option_valueor -import std/[json, sugar, strutils, options] -import chronos, chronicles, results, stew/byteutils, ffi -import - logos_delivery/waku/waku, - library/utils, - logos_delivery/waku/waku_core/peers, - logos_delivery/waku/waku_core/message/digest, - logos_delivery/waku/waku_store/common, - logos_delivery/waku/waku_store/client, - logos_delivery/waku/common/paging, - library/declare_lib +## The query/response are complex types, so this keeps the JSON bridge: the +## request carries the query as a JSON string, the response is returned as JSON. +import std/[json, sugar, options] +import logos_delivery/waku/waku_core/message/digest +import logos_delivery/waku/waku_store/common +import logos_delivery/waku/common/paging +import library/utils -func fromJsonNode(jsonContent: JsonNode): Result[StoreQueryRequest, string] = +func storeQueryFromJson(jsonContent: JsonNode): Result[StoreQueryRequest, string] = var contentTopics: seq[string] if jsonContent.contains("contentTopics"): contentTopics = collect(newSeq): @@ -67,29 +62,19 @@ func fromJsonNode(jsonContent: JsonNode): Result[StoreQueryRequest, string] = ) ) -proc waku_store_query( - ctx: ptr FFIContext[LogosDelivery], - callback: FFICallBack, - userData: pointer, - jsonQuery: cstring, - peerAddr: cstring, - timeoutMs: cint, -) {.ffi.} = - let jsonContentRes = catch: - parseJson($jsonQuery) +proc store_query*( + self: LogosDelivery, queryJson: string, peer: string, timeoutMs: int +): Future[Result[string, string]] {.ffi.} = + let jsonContent = + try: + parseJson(queryJson) + except CatchableError as e: + return err("StoreRequest failed parsing store request: " & e.msg) - if jsonContentRes.isErr(): - return err("StoreRequest failed parsing store request: " & jsonContentRes.error.msg) + let storeQueryRequest = storeQueryFromJson(jsonContent).valueOr: + return err(error) - let storeQueryRequest = ?fromJsonNode(jsonContentRes.get()) + let queryResponse = (await self.waku.storeQuery(storeQueryRequest, peer, timeoutMs)).valueOr: + return err("StoreRequest failed store query: " & error) - let peer = peers.parsePeerInfo(($peerAddr).split(",")).valueOr: - return err("StoreRequest failed to parse peer addr: " & $error) - - let queryResponse = ( - await ctx.myLib[].waku.node.wakuStoreClient.query(storeQueryRequest, peer) - ).valueOr: - return err("StoreRequest failed store query: " & $error) - - let res = $(%*(queryResponse.toHex())) - return ok(res) ## returning the response in json format + return ok($(%*(queryResponse.toHex()))) ## response in json format