Merge pull request #762 from status-im/waku-filter

Implement Waku Filter protocol

Author: D4nte

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - `waitForRemotePeer` now accepts a `timeoutMs` parameter that rejects the promise if it is reached. By default, no timeout is applied.
+- **Experimental** support for the [Waku Filter](https://rfc.vac.dev/spec/12/) protocol (client side) added, currently only works in NodeJS.
 
 ### Changed
 

proto/waku/v2/filter.proto

@@ -0,0 +1,25 @@
+syntax = "proto3";
+
+package waku.v2;
+
+import "waku/v2/message.proto";
+
+message FilterRequest {
+  bool subscribe = 1;
+  string topic = 2;
+  repeated ContentFilter content_filters = 3;
+
+  message ContentFilter {
+    string content_topic = 1;
+  }
+}
+
+message MessagePush {
+  repeated WakuMessage messages = 1;
+}
+
+message FilterRPC {
+  string request_id = 1;
+  FilterRequest request = 2;
+  MessagePush push = 3;
+}

src/lib/waku.node.spec.ts

@@ -305,4 +305,27 @@ describe("Wait for remote peer / get peers", function () {
     expect(nimPeerId).to.not.be.undefined;
     expect(peers.includes(nimPeerId as string)).to.be.true;
   });
+
+  it("Filter", async function () {
+    this.timeout(20_000);
+    nwaku = new Nwaku(makeLogFileName(this));
+    await nwaku.start({ filter: true });
+    const multiAddrWithId = await nwaku.getMultiaddrWithId();
+
+    waku = await Waku.create({
+      staticNoiseKey: NOISE_KEY_1,
+    });
+    await waku.dial(multiAddrWithId);
+    await waku.waitForRemotePeer([Protocols.Filter]);
+
+    const peers = [];
+    for await (const peer of waku.filter.peers) {
+      peers.push(peer.id.toB58String());
+    }
+
+    const nimPeerId = multiAddrWithId.getPeerId();
+
+    expect(nimPeerId).to.not.be.undefined;
+    expect(peers.includes(nimPeerId as string)).to.be.true;
+  });
 });

src/lib/waku.ts

@@ -17,6 +17,7 @@ import { Multiaddr, multiaddr } from "multiaddr";
 import PeerId from "peer-id";
 
 import { Bootstrap, BootstrapOptions } from "./discovery";
+import { FilterCodec, WakuFilter } from "./waku_filter";
 import { LightPushCodec, WakuLightPush } from "./waku_light_push";
 import { DecryptionMethod, WakuMessage } from "./waku_message";
 import { RelayCodecs, WakuRelay } from "./waku_relay";
@@ -39,6 +40,7 @@ export enum Protocols {
   Relay = "relay",
   Store = "store",
   LightPush = "lightpush",
+  Filter = "filter",
 }
 
 export interface CreateOptions {
@@ -102,6 +104,7 @@ export class Waku {
   public libp2p: Libp2p;
   public relay: WakuRelay;
   public store: WakuStore;
+  public filter: WakuFilter;
   public lightPush: WakuLightPush;
 
   private pingKeepAliveTimers: {
@@ -115,11 +118,13 @@ export class Waku {
     options: CreateOptions,
     libp2p: Libp2p,
     store: WakuStore,
-    lightPush: WakuLightPush
+    lightPush: WakuLightPush,
+    filter: WakuFilter
   ) {
     this.libp2p = libp2p;
     this.relay = libp2p.pubsub as unknown as WakuRelay;
     this.store = store;
+    this.filter = filter;
     this.lightPush = lightPush;
     this.pingKeepAliveTimers = {};
     this.relayKeepAliveTimers = {};
@@ -220,10 +225,17 @@ export class Waku {
       pubSubTopic: options?.pubSubTopic,
     });
     const wakuLightPush = new WakuLightPush(libp2p);
+    const wakuFilter = new WakuFilter(libp2p);
 
     await libp2p.start();
 
-    return new Waku(options ? options : {}, libp2p, wakuStore, wakuLightPush);
+    return new Waku(
+      options ? options : {},
+      libp2p,
+      wakuStore,
+      wakuLightPush,
+      wakuFilter
+    );
   }
 
   /**
@@ -253,6 +265,9 @@ export class Waku {
     if (_protocols.includes(Protocols.LightPush)) {
       codecs.push(LightPushCodec);
     }
+    if (_protocols.includes(Protocols.Filter)) {
+      codecs.push(FilterCodec);
+    }
 
     return this.libp2p.dialProtocol(peer, codecs);
   }
@@ -297,6 +312,7 @@ export class Waku {
   ): void {
     this.relay.addDecryptionKey(key, options);
     this.store.addDecryptionKey(key, options);
+    this.filter.addDecryptionKey(key, options);
   }
 
   /**
@@ -308,6 +324,7 @@ export class Waku {
   deleteDecryptionKey(key: Uint8Array | string): void {
     this.relay.deleteDecryptionKey(key);
     this.store.deleteDecryptionKey(key);
+    this.filter.deleteDecryptionKey(key);
   }
 
   /**
@@ -381,6 +398,16 @@ export class Waku {
       promises.push(lightPushPromise);
     }
 
+    if (protocols.includes(Protocols.Filter)) {
+      const filterPromise = (async (): Promise<void> => {
+        for await (const peer of this.filter.peers) {
+          dbg("Filter peer found", peer.id.toB58String());
+          break;
+        }
+      })();
+      promises.push(filterPromise);
+    }
+
     if (timeoutMs) {
       await rejectOnTimeout(
         Promise.all(promises),

src/lib/waku_filter/filter_rpc.ts

@@ -0,0 +1,58 @@
+import { Reader } from "protobufjs/minimal";
+import { v4 as uuid } from "uuid";
+
+import * as proto from "../../proto/waku/v2/filter";
+
+export type ContentFilter = {
+  contentTopic: string;
+};
+
+/**
+ * FilterRPC represents a message conforming to the Waku Filter protocol
+ */
+export class FilterRPC {
+  public constructor(public proto: proto.FilterRPC) {}
+
+  static createRequest(
+    topic: string,
+    contentFilters: ContentFilter[],
+    requestId?: string,
+    subscribe = true
+  ): FilterRPC {
+    return new FilterRPC({
+      requestId: requestId || uuid(),
+      request: {
+        subscribe,
+        topic,
+        contentFilters,
+      },
+      push: undefined,
+    });
+  }
+
+  /**
+   *
+   * @param bytes Uint8Array of bytes from a FilterRPC message
+   * @returns FilterRPC
+   */
+  static decode(bytes: Uint8Array): FilterRPC {
+    const res = proto.FilterRPC.decode(Reader.create(bytes));
+    return new FilterRPC(res);
+  }
+
+  /**
+   * Encode the current FilterRPC request to bytes
+   * @returns Uint8Array
+   */
+  encode(): Uint8Array {
+    return proto.FilterRPC.encode(this.proto).finish();
+  }
+
+  get push(): proto.MessagePush | undefined {
+    return this.proto.push;
+  }
+
+  get requestId(): string {
+    return this.proto.requestId;
+  }
+}

src/lib/waku_filter/index.node.spec.ts

@@ -0,0 +1,106 @@
+import { expect } from "chai";
+import debug from "debug";
+
+import { makeLogFileName, NOISE_KEY_1, Nwaku } from "../../test_utils";
+import { delay } from "../../test_utils/delay";
+import { Protocols, Waku } from "../waku";
+import { WakuMessage } from "../waku_message";
+
+const log = debug("waku:test");
+
+const TestContentTopic = "/test/1/waku-filter";
+
+describe("Waku Filter", () => {
+  let waku: Waku;
+  let nwaku: Nwaku;
+
+  afterEach(async function () {
+    !!nwaku && nwaku.stop();
+    !!waku && waku.stop().catch((e) => console.log("Waku failed to stop", e));
+  });
+
+  beforeEach(async function () {
+    this.timeout(10000);
+    nwaku = new Nwaku(makeLogFileName(this));
+    await nwaku.start({ filter: true });
+    waku = await Waku.create({
+      staticNoiseKey: NOISE_KEY_1,
+      libp2p: { addresses: { listen: ["/ip4/0.0.0.0/tcp/0/ws"] } },
+    });
+    await waku.dial(await nwaku.getMultiaddrWithId());
+    await waku.waitForRemotePeer([Protocols.Filter, Protocols.Relay]);
+  });
+
+  it("creates a subscription", async function () {
+    this.timeout(10000);
+
+    let messageCount = 0;
+    const messageText = "Filtering works!";
+    const callback = (msg: WakuMessage): void => {
+      log("Got a message");
+      messageCount++;
+      expect(msg.contentTopic).to.eq(TestContentTopic);
+      expect(msg.payloadAsUtf8).to.eq(messageText);
+    };
+    await waku.filter.subscribe(callback, [TestContentTopic]);
+    const message = await WakuMessage.fromUtf8String(
+      messageText,
+      TestContentTopic
+    );
+    await waku.relay.send(message);
+    while (messageCount === 0) {
+      await delay(250);
+    }
+    expect(messageCount).to.eq(1);
+  });
+
+  it("handles multiple messages", async function () {
+    this.timeout(10000);
+
+    let messageCount = 0;
+    const callback = (msg: WakuMessage): void => {
+      messageCount++;
+      expect(msg.contentTopic).to.eq(TestContentTopic);
+    };
+    await waku.filter.subscribe(callback, [TestContentTopic]);
+    await waku.relay.send(
+      await WakuMessage.fromUtf8String("Filtering works!", TestContentTopic)
+    );
+    await waku.relay.send(
+      await WakuMessage.fromUtf8String(
+        "Filtering still works!",
+        TestContentTopic
+      )
+    );
+    while (messageCount < 2) {
+      await delay(250);
+    }
+    expect(messageCount).to.eq(2);
+  });
+
+  it("unsubscribes", async function () {
+    let messageCount = 0;
+    const callback = (): void => {
+      messageCount++;
+    };
+    const unsubscribe = await waku.filter.subscribe(callback, [
+      TestContentTopic,
+    ]);
+    await waku.relay.send(
+      await WakuMessage.fromUtf8String(
+        "This should be received",
+        TestContentTopic
+      )
+    );
+    await delay(100);
+    await unsubscribe();
+    await waku.relay.send(
+      await WakuMessage.fromUtf8String(
+        "This should not be received",
+        TestContentTopic
+      )
+    );
+    await delay(100);
+    expect(messageCount).to.eq(1);
+  });
+});