make PeerManager use only ConnectionManager, move getPeers to Connect…

…ionManager, remove not needed code

packages/core/src/index.ts

@@ -15,11 +15,10 @@ export { LightPushCodec, LightPushCore } from "./lib/light_push/index.js";
 export * as waku_store from "./lib/store/index.js";
 export { StoreCore, StoreCodec } from "./lib/store/index.js";
 
-export { ConnectionManager } from "./lib/connection_manager.js";
+export { ConnectionManager } from "./lib/connection_manager/index.js";
 
 export { getHealthManager } from "./lib/health_manager.js";
 
-export { KeepAliveManager } from "./lib/keep_alive_manager.js";
 export { StreamManager } from "./lib/stream_manager/index.js";
 
 export { MetadataCodec, wakuMetadata } from "./lib/metadata/index.js";

packages/core/src/lib/base_protocol.ts

@@ -5,10 +5,8 @@ import type {
   Libp2pComponents,
   PubsubTopic
 } from "@waku/interfaces";
-import { Logger } from "@waku/utils";
-import { getPeersForProtocol, sortPeersByLatency } from "@waku/utils/libp2p";
+import { getPeersForProtocol } from "@waku/utils/libp2p";
 
-import { filterPeersByDiscovery } from "./filterPeers.js";
 import { StreamManager } from "./stream_manager/index.js";
 
 /**
@@ -23,7 +21,6 @@ export class BaseProtocol implements IBaseProtocolCore {
   protected constructor(
     public multicodec: string,
     protected components: Libp2pComponents,
-    private log: Logger,
     public readonly pubsubTopics: PubsubTopic[]
   ) {
     this.addLibp2pEventListener = components.events.addEventListener.bind(
@@ -64,54 +61,4 @@ export class BaseProtocol implements IBaseProtocolCore {
       return connections.length > 0;
     });
   }
-
-  /**
-   * Retrieves a list of connected peers that support the protocol. The list is sorted by latency.
-   *
-   * @param numPeers - The total number of peers to retrieve. If 0, all peers are returned.
-   * @param maxBootstrapPeers - The maximum number of bootstrap peers to retrieve.
-   * @returns A list of peers that support the protocol sorted by latency. By default, returns all peers available, including bootstrap.
-   */
-  public async getPeers(
-    {
-      numPeers,
-      maxBootstrapPeers
-    }: {
-      numPeers: number;
-      maxBootstrapPeers: number;
-    } = {
-      maxBootstrapPeers: 0,
-      numPeers: 0
-    }
-  ): Promise<Peer[]> {
-    // Retrieve all connected peers that support the protocol & shard (if configured)
-    const allAvailableConnectedPeers = await this.connectedPeers();
-
-    // Filter the peers based on discovery & number of peers requested
-    const filteredPeers = filterPeersByDiscovery(
-      allAvailableConnectedPeers,
-      numPeers,
-      maxBootstrapPeers
-    );
-
-    // Sort the peers by latency
-    const sortedFilteredPeers = await sortPeersByLatency(
-      this.components.peerStore,
-      filteredPeers
-    );
-
-    if (sortedFilteredPeers.length === 0) {
-      this.log.warn(
-        "No peers found. Ensure you have a connection to the network."
-      );
-    }
-
-    if (sortedFilteredPeers.length < numPeers) {
-      this.log.warn(
-        `Only ${sortedFilteredPeers.length} peers found. Requested ${numPeers}.`
-      );
-    }
-
-    return sortedFilteredPeers;
-  }
 }

packages/core/src/lib/connection_manager.ts → packages/core/src/lib/connection_manager/connection_manager.ts

@@ -19,6 +19,7 @@ import { decodeRelayShard, shardInfoToPubsubTopics } from "@waku/utils";
 import { Logger } from "@waku/utils";
 
 import { KeepAliveManager } from "./keep_alive_manager.js";
+import { getPeerPing } from "./utils.js";
 
 const log = new Logger("connection-manager");
 
@@ -180,6 +181,29 @@ export class ConnectionManager
     );
   }
 
+  public async getConnectedPeers(codec?: string): Promise<Peer[]> {
+    const peerIDs = this.libp2p.getPeers();
+
+    if (peerIDs.length === 0) {
+      return [];
+    }
+
+    const peers = await Promise.all(
+      peerIDs.map(async (id) => {
+        try {
+          return await this.libp2p.peerStore.get(id);
+        } catch (e) {
+          return null;
+        }
+      })
+    );
+
+    return peers
+      .filter((p) => !!p)
+      .filter((p) => (codec ? (p as Peer).protocols.includes(codec) : true))
+      .sort((left, right) => getPeerPing(left) - getPeerPing(right)) as Peer[];
+  }
+
   private async dialPeerStorePeers(): Promise<void> {
     const peerInfos = await this.libp2p.peerStore.all();
     const dialPromises = [];

packages/core/src/lib/connection_manager/index.ts

@@ -0,0 +1 @@
+export { ConnectionManager } from "./connection_manager.js";

packages/core/src/lib/keep_alive_manager.ts → packages/core/src/lib/connection_manager/keep_alive_manager.ts

@@ -3,7 +3,7 @@ import type { IRelay, Libp2p, PeerIdStr } from "@waku/interfaces";
 import { Logger, pubsubTopicToSingleShardInfo } from "@waku/utils";
 import { utf8ToBytes } from "@waku/utils/bytes";
 
-import { createEncoder } from "./message/version_0.js";
+import { createEncoder } from "../message/version_0.js";
 
 const RelayPingContentTopic = "/relay-ping/1/ping/null";
 const log = new Logger("keep-alive");

packages/core/src/lib/connection_manager/utils.ts

@@ -0,0 +1,25 @@
+import type { Peer } from "@libp2p/interface";
+import { bytesToUtf8 } from "@waku/utils/bytes";
+
+/**
+ * Reads peer's metadata and retrieves ping value.
+ * @param peer Peer or null
+ * @returns -1 if no ping attached, otherwise returns ping value
+ */
+export const getPeerPing = (peer: Peer | null): number => {
+  if (!peer) {
+    return -1;
+  }
+
+  try {
+    const bytes = peer.metadata.get("ping");
+
+    if (!bytes) {
+      return -1;
+    }
+
+    return Number(bytesToUtf8(bytes));
+  } catch (e) {
+    return -1;
+  }
+};

packages/core/src/lib/filter/index.ts

@@ -40,7 +40,7 @@ export class FilterCore extends BaseProtocol implements IBaseProtocolCore {
     public readonly pubsubTopics: PubsubTopic[],
     libp2p: Libp2p
   ) {
-    super(FilterCodecs.SUBSCRIBE, libp2p.components, log, pubsubTopics);
+    super(FilterCodecs.SUBSCRIBE, libp2p.components, pubsubTopics);
 
     libp2p
       .handle(FilterCodecs.PUSH, this.onRequest.bind(this), {

packages/core/src/lib/light_push/index.ts

@@ -37,7 +37,7 @@ export class LightPushCore extends BaseProtocol implements IBaseProtocolCore {
     public readonly pubsubTopics: PubsubTopic[],
     libp2p: Libp2p
   ) {
-    super(LightPushCodec, libp2p.components, log, pubsubTopics);
+    super(LightPushCodec, libp2p.components, pubsubTopics);
   }
 
   private async preparePushMessage(

packages/core/src/lib/metadata/index.ts

@@ -30,7 +30,7 @@ class Metadata extends BaseProtocol implements IMetadata {
     public pubsubTopics: PubsubTopic[],
     libp2p: Libp2pComponents
   ) {
-    super(MetadataCodec, libp2p.components, log, pubsubTopics);
+    super(MetadataCodec, libp2p.components, pubsubTopics);
     this.libp2pComponents = libp2p;
     void libp2p.registrar.handle(MetadataCodec, (streamData) => {
       void this.onRequest(streamData);

packages/core/src/lib/store/index.ts

@@ -32,7 +32,7 @@ export class StoreCore extends BaseProtocol implements IStoreCore {
     public readonly pubsubTopics: PubsubTopic[],
     libp2p: Libp2p
   ) {
-    super(StoreCodec, libp2p.components, log, pubsubTopics);
+    super(StoreCodec, libp2p.components, pubsubTopics);
   }
 
   public async *queryPerPage<T extends IDecodedMessage>(