Merge branch 'feat/metadata-protocol' of github.com:waku-org/js-waku into feat/metadata-protocol

Author: danisharora099

packages/core/src/index.ts

@@ -27,3 +27,4 @@ export { ConnectionManager } from "./lib/connection_manager.js";
 
 export { KeepAliveManager } from "./lib/keep_alive_manager.js";
 export { StreamManager } from "./lib/stream_manager.js";
+export { wakuMetadata, MetadataCodec } from "./lib/metadata/index.js";

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

@@ -1,10 +1,12 @@
 import type { PeerId } from "@libp2p/interface/peer-id";
+import type { IncomingStreamData } from "@libp2p/interface-internal/registrar";
 import type {
   IMetadata,
-  Libp2p,
+  Libp2pComponents,
   MetadataQueryParams,
   ShardInfo
 } from "@waku/interfaces";
+import { proto_metadata as proto } from "@waku/proto";
 import { Logger } from "@waku/utils";
 import all from "it-all";
 import * as lp from "it-length-prefixed";
@@ -13,29 +15,77 @@ import { Uint8ArrayList } from "uint8arraylist";
 
 import { BaseProtocol } from "../base_protocol.js";
 
-import { MetadataRpc } from "./rpc.js";
-
 const log = new Logger("metadata");
 
 export const MetadataCodec = "/vac/waku/metadata/1.0.0";
 
 class Metadata extends BaseProtocol {
-  constructor(libp2p: Libp2p) {
-    super(MetadataCodec, libp2p.components);
+  constructor(
+    libp2pComponents: Libp2pComponents,
+    private shardInfo: ShardInfo
+  ) {
+    super(MetadataCodec, libp2pComponents);
+
+    libp2pComponents.registrar
+      .handle(MetadataCodec, (streamData) => {
+        return void this.handle(streamData);
+      })
+      .then(() => "yes")
+      .catch(() => "no");
+  }
+
+  private async handle(streamData: IncomingStreamData): Promise<ShardInfo> {
+    const encodedRpcQuery = proto.WakuMetadataRequest.encode(this.shardInfo);
+
+    const res = await pipe(
+      [encodedRpcQuery],
+      lp.encode,
+      streamData.stream,
+      lp.decode,
+      async (source) => await all(source)
+    );
+
+    try {
+      const bytes = new Uint8ArrayList();
+      res.forEach((chunk) => {
+        bytes.append(chunk);
+      });
+
+      const response = proto.WakuMetadataResponse.decode(bytes);
+      if (!response) {
+        throw new Error("WakuMetadata response is undefined");
+      }
+
+      const { shards, clusterId } = response;
+
+      if (!clusterId) {
+        throw new Error("WakuMetadata response clusterId is undefined");
+      }
+
+      return {
+        clusterId,
+        shards
+      };
+    } catch (e) {
+      log.error(`Error decoding response: ${e}`);
+      throw e;
+    }
   }
 
   /**
    * Make a metadata query to a peer
    */
   async query(params: MetadataQueryParams, peerId: PeerId): Promise<ShardInfo> {
-    const rpcQuery = MetadataRpc.createRequest(params.clusterId, params.shards);
+    const encodedRpcQuery = proto.WakuMetadataRequest.encode({
+      clusterId: params.clusterId,
+      shards: params.shards
+    });
 
     const peer = await this.getPeer(peerId);
-
     const stream = await this.getStream(peer);
 
     const res = await pipe(
-      [rpcQuery.encode()],
+      [encodedRpcQuery],
       lp.encode,
       stream,
       lp.decode,
@@ -48,23 +98,30 @@ class Metadata extends BaseProtocol {
         bytes.append(chunk);
       });
 
-      const { response } = MetadataRpc.decode(bytes);
+      const response = proto.WakuMetadataResponse.decode(bytes);
       if (!response) {
-        throw new Error("No response in query");
+        throw new Error("WakuMetadata response is undefined");
       }
 
       const { shards, clusterId } = response;
+      if (!clusterId) {
+        throw new Error("WakuMetadata response clusterId is undefined");
+      }
+
       return {
-        cluster: clusterId,
-        indexList: shards
-      } as ShardInfo;
+        clusterId,
+        shards
+      };
     } catch (e) {
       log.error(`Error decoding response: ${e}`);
       throw e;
     }
   }
 }
 
-export function wakuMetadata(): (libp2p: Libp2p) => IMetadata {
-  return (libp2p: Libp2p) => new Metadata(libp2p);
+export function wakuMetadata(
+  shardInfo: ShardInfo
+): (components: Libp2pComponents) => IMetadata {
+  return (libp2pComponents: Libp2pComponents) =>
+    new Metadata(libp2pComponents, shardInfo);
 }

packages/interfaces/src/libp2p.ts

@@ -4,10 +4,13 @@ import type { Libp2pInit } from "libp2p";
 import type { identifyService } from "libp2p/identify";
 import type { PingService } from "libp2p/ping";
 
+import type { IMetadata } from "./metadata";
+
 export type Libp2pServices = {
   ping: PingService;
   pubsub?: GossipSub;
   identify: ReturnType<ReturnType<typeof identifyService>>;
+  metadata: IMetadata;
 };
 
 // TODO: Get libp2p to export this.

packages/sdk/src/create.ts

@@ -9,6 +9,7 @@ import {
   DefaultUserAgent,
   wakuFilter,
   wakuLightPush,
+  wakuMetadata,
   WakuNode,
   WakuOptions,
   wakuStore
@@ -203,6 +204,7 @@ export async function defaultLibp2p(
         agentVersion: userAgent ?? DefaultUserAgent
       }),
       ping: pingService(),
+      metadata: wakuMetadata(),
       ...pubsubService,
       ...options?.services
     }

packages/tests/src/node/interfaces.ts

@@ -24,6 +24,7 @@ export interface Args {
   discv5UdpPort?: number;
   // `legacyFilter` is required to enable filter v1 with go-waku
   legacyFilter?: boolean;
+  clusterId?: number;
 }
 
 export enum LogLevel {

packages/tests/tests/peer_exchange.node.spec.ts

@@ -52,6 +52,18 @@ describe("Peer Exchange", () => {
 
       waku = await createLightNode();
       await waku.start();
+
+      waku.libp2p.addEventListener("connection:open", (event) => {
+        console.log("conn open");
+        console.log(event.detail.streams);
+      });
+
+      waku.libp2p.addEventListener("connection:close", (event) => {
+        console.log("conn close");
+        console.log(event.detail);
+        console.log("curr conns", waku.libp2p.getConnections().length);
+      });
+
       await waku.libp2p.dialProtocol(nwaku2Ma, PeerExchangeCodec);
 
       const components = waku.libp2p.components as unknown as Libp2pComponents;