feat: replace waitForRemotePeers() with waku.waitForPeer() method

packages/interfaces/src/waku.ts

@@ -10,36 +10,107 @@ import { Protocols } from "./protocols.js";
 import type { IRelay } from "./relay.js";
 import type { IStore } from "./store.js";
 
-export interface Waku {
+export interface IWaku {
   libp2p: Libp2p;
   relay?: IRelay;
   store?: IStore;
   filter?: IFilter;
   lightPush?: ILightPush;
 
+  health: IHealthManager;
   connectionManager: IConnectionManager;
 
+  /**
+   * Dials to the provided peer
+   *
+   * @param {PeerId | MultiaddrInput} peer information to use for dialing
+   * @param {Protocols[]} [protocols] array of Waku protocols to be used for dialing. If no provided - will be derived from mounted protocols.
+   *
+   * @returns {Promise<Stream>} `Promise` that will resolve to a `Stream` to a dialed peer
+   *
+   * @example
+   * ```typescript
+   * await waku.dial(remotePeerId, [Protocols.LightPush]);
+   *
+   * waku.isConnected() === true;
+   * ```
+   */
   dial(peer: PeerId | MultiaddrInput, protocols?: Protocols[]): Promise<Stream>;
 
+  /**
+   * Starts all services and components related to functionality of Waku node.
+   *
+   * @returns {Promise<boolean>} `Promise` that will resolve when started.
+   *
+   * @example
+   * ```typescript
+   * await waku.start();
+   *
+   * waku.isStarted() === true;
+   * ```
+   */
   start(): Promise<void>;
 
+  /**
+   * Stops all recurring processes and services that are needed for functionality of Waku node.
+   *
+   * @returns {Promise<boolean>} `Promise` that resolves when stopped.
+   *
+   * @example
+   * ```typescript
+   * await waku.stop();
+   *
+   * waku.isStarted === false;
+   * ```
+   */
   stop(): Promise<void>;
 
+  /**
+   * Resolves when Waku successfully gains connection to a remote peers that fits provided requirements.
+   * Must be used after attempting to connect to nodes, using {@link IWaku.dial} or
+   * if was bootstrapped by using {@link IPeerExchange} or {@link DnsDiscoveryComponents}.
+   *
+   * @param {Protocols[]} [protocols] Protocols that need to be enabled by remote peers
+   * @param {number} [timeoutMs] Timeout value in milliseconds after which promise rejects
+   *
+   * @returns {Promise<void>} `Promise` that **resolves** if all desired protocols are fulfilled by
+   * at least one remote peer, **rejects** if the timeoutMs is reached
+   * @throws If passing a protocol that is not mounted or Waku node is not started
+   *
+   * @example
+   * ```typescript
+   * try {
+   *  // let's wait for at least one LightPush node and timeout in 1 second
+   *  await waku.connect([Protocols.LightPush], 1000);
+   * } catch(e) {
+   *  waku.isConnected() === false;
+   *  console.error("Failed to connect due to", e);
+   * }
+   *
+   * waku.isConnected() === true;
+   * ```
+   */
+  connect(protocols?: Protocols[], timeoutMs?: number): Promise<void>;
+
+  /**
+   * @returns {boolean} `true` if the node was started and `false` otherwise
+   */
   isStarted(): boolean;
 
+  /**
+   * @returns {boolean} `true` if the node has working connection and `false` otherwise
+   */
   isConnected(): boolean;
-
-  health: IHealthManager;
 }
 
-export interface LightNode extends Waku {
+export interface LightNode extends IWaku {
   relay: undefined;
   store: IStore;
   filter: IFilter;
   lightPush: ILightPush;
 }
 
-export interface RelayNode extends Waku {
+export interface RelayNode extends IWaku {
   relay: IRelay;
   store: undefined;
   filter: undefined;

packages/sdk/src/wait_for_remote_peer.ts

@@ -1,6 +1,6 @@
 import type { IdentifyResult } from "@libp2p/interface";
 import { FilterCodecs, LightPushCodec, StoreCodec } from "@waku/core";
-import type { IRelay, Libp2p, Waku } from "@waku/interfaces";
+import type { IRelay, IWaku, Libp2p } from "@waku/interfaces";
 import { Protocols } from "@waku/interfaces";
 import { Logger } from "@waku/utils";
 import { pEvent } from "p-event";
@@ -27,19 +27,20 @@ const log = new Logger("wait-for-remote-peer");
  * @default Wait for remote peers with protocols enabled locally and no time out is applied.
  */
 export async function waitForRemotePeer(
-  waku: Waku,
+  waku: IWaku,
   protocols?: Protocols[],
   timeoutMs?: number
 ): Promise<void> {
-  protocols = protocols ?? getEnabledProtocols(waku);
+  // if no protocols or empty array passed - try to derive from mounted
+  protocols = protocols?.length ? protocols : getEnabledProtocols(waku);
   const connections = waku.libp2p.getConnections();
 
   if (!waku.isStarted()) {
     throw Error("Waku node is not started");
   }
 
   if (connections.length > 0 && !protocols.includes(Protocols.Relay)) {
-    const success = await waitForMetadata(waku.libp2p);
+    const success = await waitForMetadata(waku, protocols);
 
     if (success) {
       return;
@@ -135,33 +136,55 @@ async function waitForConnectedPeer(
 /**
  * Waits for the metadata from the remote peer.
  */
-async function waitForMetadata(libp2p: Libp2p): Promise<boolean> {
-  const connections = libp2p.getConnections();
-  const metadataService = libp2p.services.metadata;
-
-  if (!connections.length || !metadataService) {
+async function waitForMetadata(
+  waku: IWaku,
+  protocols: Protocols[]
+): Promise<boolean> {
+  const connectedPeers = waku.libp2p.getPeers();
+  const metadataService = waku.libp2p.services.metadata;
+  const enabledCodes = mapProtocolsToCodecs(protocols);
+
+  if (!connectedPeers.length || !metadataService) {
     log.info(
-      `Skipping waitForMetadata due to missing connections:${connections.length} or metadataService:${!!metadataService}`
+      `Skipping waitForMetadata due to missing connections:${connectedPeers.length} or metadataService:${!!metadataService}`
     );
     return false;
   }
 
-  try {
-    // confirm at least with one connected peer
-    await Promise.any(
-      connections
-        .map((c) => c.remotePeer)
-        .map((peer) => metadataService.confirmOrAttemptHandshake(peer))
+  for (const peerId of connectedPeers) {
+    const confirmedAllCodecs = Array.from(enabledCodes.values()).every(
+      (v) => v
     );
 
-    return true;
-  } catch (e) {
-    // eslint-disable-next-line @typescript-eslint/no-explicit-any
-    if ((e as any).code === "ERR_CONNECTION_BEING_CLOSED") {
-      log.error("Connection closed. Some peers can be on different shard.");
+    if (confirmedAllCodecs) {
+      return true;
     }
 
-    log.error(`Error waiting for metadata: ${e}`);
+    try {
+      const peer = await waku.libp2p.peerStore.get(peerId);
+      const hasSomeCodes = peer.protocols.some((c) => enabledCodes.has(c));
+
+      if (hasSomeCodes) {
+        const response =
+          await metadataService.confirmOrAttemptHandshake(peerId);
+
+        if (!response.error) {
+          peer.protocols.forEach((c) => {
+            if (enabledCodes.has(c)) {
+              enabledCodes.set(c, true);
+            }
+          });
+        }
+      }
+    } catch (e) {
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      if ((e as any).code === "ERR_CONNECTION_BEING_CLOSED") {
+        log.error("Connection closed. Some peers can be on different shard.");
+      }
+
+      log.error(`Error while iterating through peers: ${e}`);
+      continue;
+    }
   }
 
   return false;
@@ -195,7 +218,7 @@ async function rejectOnTimeout<T>(
   await Promise.race([promise, awaitTimeout(timeoutMs, rejectReason)]);
 }
 
-function getEnabledProtocols(waku: Waku): Protocols[] {
+function getEnabledProtocols(waku: IWaku): Protocols[] {
   const protocols = [];
 
   if (waku.relay) {
@@ -216,3 +239,21 @@ function getEnabledProtocols(waku: Waku): Protocols[] {
 
   return protocols;
 }
+
+function mapProtocolsToCodecs(protocols: Protocols[]): Map<string, boolean> {
+  const codecs: Map<string, boolean> = new Map();
+
+  const protocolToCodec: Record<string, string> = {
+    [Protocols.Filter]: FilterCodecs.SUBSCRIBE,
+    [Protocols.LightPush]: LightPushCodec,
+    [Protocols.Store]: StoreCodec
+  };
+
+  for (const protocol of protocols) {
+    if (protocolToCodec[protocol]) {
+      codecs.set(protocolToCodec[protocol], false);
+    }
+  }
+
+  return codecs;
+}

packages/sdk/src/waku.ts

@@ -8,10 +8,10 @@ import type {
   ILightPush,
   IRelay,
   IStore,
+  IWaku,
   Libp2p,
   ProtocolCreateOptions,
-  PubsubTopic,
-  Waku
+  PubsubTopic
 } from "@waku/interfaces";
 import { Protocols } from "@waku/interfaces";
 import { Logger } from "@waku/utils";
@@ -20,6 +20,7 @@ import { wakuFilter } from "./protocols/filter/index.js";
 import { wakuLightPush } from "./protocols/light_push/index.js";
 import { wakuStore } from "./protocols/store/index.js";
 import { ReliabilityMonitorManager } from "./reliability_monitor/index.js";
+import { waitForRemotePeer } from "./wait_for_remote_peer.js";
 
 export const DefaultPingKeepAliveValueSecs = 5 * 60;
 export const DefaultRelayKeepAliveValueSecs = 5 * 60;
@@ -59,7 +60,7 @@ type ProtocolsEnabled = {
   store?: boolean;
 };
 
-export class WakuNode implements Waku {
+export class WakuNode implements IWaku {
   public libp2p: Libp2p;
   public relay?: IRelay;
   public store?: IStore;
@@ -126,12 +127,6 @@ export class WakuNode implements Waku {
     );
   }
 
-  /**
-   * Dials to the provided peer.
-   *
-   * @param peer The peer to dial
-   * @param protocols Waku protocols we expect from the peer; Defaults to mounted protocols
-   */
   public async dial(
     peer: PeerId | MultiaddrInput,
     protocols?: Protocols[]
@@ -201,6 +196,13 @@ export class WakuNode implements Waku {
     await this.libp2p.stop();
   }
 
+  public async connect(
+    protocols?: Protocols[],
+    timeoutMs?: number
+  ): Promise<void> {
+    return waitForRemotePeer(this, protocols, timeoutMs);
+  }
+
   public isStarted(): boolean {
     return this.libp2p.status == "started";
   }

packages/tests/src/utils/nodes.ts

@@ -1,10 +1,10 @@
 import {
   DefaultNetworkConfig,
+  IWaku,
   LightNode,
   NetworkConfig,
   ProtocolCreateOptions,
-  Protocols,
-  Waku
+  Protocols
 } from "@waku/interfaces";
 import { createLightNode, waitForRemotePeer } from "@waku/sdk";
 import { derivePubsubTopicsFromNetworkConfig, isDefined } from "@waku/utils";
@@ -79,7 +79,7 @@ export async function runMultipleNodes(
 
 export async function teardownNodesWithRedundancy(
   serviceNodes: ServiceNodesFleet,
-  wakuNodes: Waku | Waku[]
+  wakuNodes: IWaku | IWaku[]
 ): Promise<void> {
   const wNodes = Array.isArray(wakuNodes) ? wakuNodes : [wakuNodes];
 

packages/tests/src/utils/teardown.ts

@@ -1,4 +1,4 @@
-import { Waku } from "@waku/interfaces";
+import { IWaku } from "@waku/interfaces";
 import { Logger } from "@waku/utils";
 import pRetry from "p-retry";
 
@@ -8,7 +8,7 @@ const log = new Logger("test:teardown");
 
 export async function tearDownNodes(
   nwakuNodes: ServiceNode | ServiceNode[],
-  wakuNodes: Waku | Waku[]
+  wakuNodes: IWaku | IWaku[]
 ): Promise<void> {
   const nNodes = Array.isArray(nwakuNodes) ? nwakuNodes : [nwakuNodes];
   const wNodes = Array.isArray(wakuNodes) ? wakuNodes : [wakuNodes];