feat(store): allow using a specific node

packages/core/src/lib/connection_manager.ts

@@ -1,5 +1,6 @@
 import type { Peer, PeerId, PeerInfo, PeerStore } from "@libp2p/interface";
 import { TypedEventEmitter } from "@libp2p/interface";
+import { multiaddr } from "@multiformats/multiaddr";
 import {
   ConnectionManagerOptions,
   DiscoveryTrigger,
@@ -11,6 +12,7 @@
   IPeersByDiscoveryEvents,
   IRelay,
   KeepAliveOptions,
+  MultiaddrStr,
   PeersByDiscoveryResult,
   PubsubTopic,
   ShardInfo
@@ -36,7 +38,7 @@
   private options: ConnectionManagerOptions;
   private libp2p: Libp2p;
   private dialAttemptsForPeer: Map<string, number> = new Map();
   private dialErrorsForPeer: Map<string, any> = new Map();
 
   private currentActiveParallelDialCount = 0;
   private pendingPeerDialQueue: Array<PeerId> = [];
@@ -219,15 +221,39 @@
     this.startNetworkStatusListener();
   }
 
-  private async dialPeer(peerId: PeerId): Promise<void> {
+  public async dialPeer(
+    peer: PeerId | MultiaddrStr,
+    protocols?: string[]
+  ): Promise<void> {
+    let peerId: PeerId;
+    if (typeof peer === "string") {
+      const ma = multiaddr(peer);
+      const peerIdStr = ma.getPeerId();
+      if (!peerIdStr) {
+        throw new Error("Failed to dial multiaddr: missing peer ID");
+      }
+      const conn = this.libp2p
+        .getConnections()
+        .find((conn) => conn.remotePeer.toString() === peerIdStr);
+      if (conn) {
+        peerId = conn.remotePeer;
+      } else {
+        throw new Error("Failed to dial multiaddr: no connection found");
+      }
+    } else {
+      peerId = peer;
+    }
+
     this.currentActiveParallelDialCount += 1;
     let dialAttempt = 0;
     while (dialAttempt < this.options.maxDialAttemptsForPeer) {
       try {
         log.info(
           `Dialing peer ${peerId.toString()} on attempt ${dialAttempt + 1}`
         );
-        await this.libp2p.dial(peerId);
+        protocols
+          ? await this.libp2p.dialProtocol(peerId, protocols)
+          : await this.libp2p.dial(peerId);
 
         const tags = await this.getTagNamesForPeer(peerId);
         // add tag to connection describing discovery mechanism
@@ -253,7 +279,7 @@
           // Handle generic error
           log.error(
             `Error dialing peer ${peerId.toString()} - ${
               (error as any).message
             }`
           );
         }

packages/interfaces/src/libp2p.ts

@@ -36,3 +36,5 @@ export type CreateLibp2pOptions = Libp2pOptions & {
    */
   filterMultiaddrs?: boolean;
 };
+
+export type MultiaddrStr = string;

packages/interfaces/src/protocols.ts

@@ -28,6 +28,10 @@ export type IBaseProtocolSDK = {
   readonly numPeersToUse: number;
 };
 
+export type StoreProtocolOptions = {
+  peer: string;
+};
+
 export type NetworkConfig = StaticSharding | AutoSharding;
 
 //TODO: merge this with ProtocolCreateOptions or establish distinction: https://github.com/waku-org/js-waku/issues/2048
@@ -106,6 +110,10 @@ export type ProtocolCreateOptions = {
    * List of peers to use to bootstrap the node. Ignored if defaultBootstrap is set to true.
    */
   bootstrapPeers?: string[];
+  /**
+   * Options for the Store protocol.
+   */
+  store?: Partial<StoreProtocolOptions>;
 };
 
 export type Callback<T extends IDecodedMessage> = (

packages/sdk/src/protocols/store/index.ts

@@ -1,11 +1,13 @@
+import type { Peer } from "@libp2p/interface";
 import { ConnectionManager, StoreCore } from "@waku/core";
 import {
   IDecodedMessage,
   IDecoder,
   IStore,
   Libp2p,
   QueryRequestParams,
-  StoreCursor
+  StoreCursor,
+  StoreProtocolOptions
 } from "@waku/interfaces";
 import { messageHash } from "@waku/message-hash";
 import { ensurePubsubTopicIsConfigured, isDefined, Logger } from "@waku/utils";
@@ -23,7 +25,11 @@ const log = new Logger("waku:store:sdk");
 export class Store extends BaseProtocolSDK implements IStore {
   public readonly protocol: StoreCore;
 
-  public constructor(connectionManager: ConnectionManager, libp2p: Libp2p) {
+  public constructor(
+    connectionManager: ConnectionManager,
+    libp2p: Libp2p,
+    private options?: Partial<StoreProtocolOptions>
+  ) {
     super(
       new StoreCore(connectionManager.configuredPubsubTopics, libp2p),
       connectionManager,
@@ -58,12 +64,15 @@ export class Store extends BaseProtocolSDK implements IStore {
       ...options
     };
 
-    const peer = (
-      await this.protocol.getPeers({
-        numPeers: this.numPeersToUse,
-        maxBootstrapPeers: 1
-      })
-    )[0];
+    const peer =
+      (await this.getPeerToUse()) ??
+      (
+        await this.protocol.getPeers({
+          numPeers: this.numPeersToUse,
+          maxBootstrapPeers: 1
+        })
+      )[0];
+
     if (!peer) {
       log.error("No peers available to query");
       throw new Error("No peers available to query");
@@ -228,6 +237,20 @@ export class Store extends BaseProtocolSDK implements IStore {
       decodersAsMap
     };
   }
+
+  private async getPeerToUse(): Promise<Peer | null> {
+    const peer = this.connectedPeers.find(
+      (p) => p.id.toString() === this.options?.peer
+    );
+    if (peer) {
+      return peer;
+    }
+
+    log.warn(
+      `Passed node to use for Store not found: ${this.options?.peer}. Attempting to use random peers.`
+    );
+    return null;
+  }
 }
 
 /**
@@ -237,9 +260,10 @@ export class Store extends BaseProtocolSDK implements IStore {
  * @returns A function that takes a Libp2p instance and returns a StoreSDK instance.
  */
 export function wakuStore(
-  connectionManager: ConnectionManager
+  connectionManager: ConnectionManager,
+  options?: Partial<StoreProtocolOptions>
 ): (libp2p: Libp2p) => IStore {
   return (libp2p: Libp2p) => {
-    return new Store(connectionManager, libp2p);
+    return new Store(connectionManager, libp2p, options);
   };
 }

packages/sdk/src/waku/waku.ts

@@ -1,7 +1,7 @@
 import type { Stream } from "@libp2p/interface";
 import { isPeerId, PeerId } from "@libp2p/interface";
 import { multiaddr, Multiaddr, MultiaddrInput } from "@multiformats/multiaddr";
-import { ConnectionManager, getHealthManager } from "@waku/core";
+import { ConnectionManager, getHealthManager, StoreCodec } from "@waku/core";
 import type {
   IFilter,
   IHealthManager,
@@ -10,6 +10,7 @@ import type {
   IStore,
   IWaku,
   Libp2p,
+  PeerIdStr,
   ProtocolCreateOptions,
   PubsubTopic
 } from "@waku/interfaces";
@@ -106,7 +107,16 @@ export class WakuNode implements IWaku {
     this.health = getHealthManager();
 
     if (protocolsEnabled.store) {
-      const store = wakuStore(this.connectionManager);
+      let peerIdStr: PeerIdStr | undefined;
+      if (options.store?.peer) {
+        this.connectionManager
+          .dialPeer(options.store.peer, [StoreCodec])
+          .catch((e) => {
+            log.error("Failed to dial store peer", e);
+          });
+      }
+
+      const store = wakuStore(this.connectionManager, { peer: peerIdStr });
       this.store = store(libp2p);
     }