feat!: use ShardingParams on subscriptions, make Decoder/Encoder auto sharding friendly by default

packages/core/src/lib/message/version_0.spec.ts

@@ -1,108 +1,97 @@
 import type { IProtoMessage } from "@waku/interfaces";
+import { contentTopicToPubsubTopic } from "@waku/utils";
 import { expect } from "chai";
 import fc from "fast-check";
 
 import { createDecoder, createEncoder, DecodedMessage } from "./version_0.js";
 
+const contentTopic = "/js-waku/1/tests/bytes";
+const pubsubTopic = contentTopicToPubsubTopic(contentTopic);
+
 describe("Waku Message version 0", function () {
   it("Round trip binary serialization", async function () {
     await fc.assert(
-      fc.asyncProperty(
-        fc.string({ minLength: 1 }),
-        fc.string({ minLength: 1 }),
-        fc.uint8Array({ minLength: 1 }),
-        async (contentTopic, pubsubTopic, payload) => {
-          const encoder = createEncoder({
-            contentTopic
-          });
-          const bytes = await encoder.toWire({ payload });
-          const decoder = createDecoder(contentTopic);
-          const protoResult = await decoder.fromWireToProtoObj(bytes);
-          const result = (await decoder.fromProtoObj(
-            pubsubTopic,
-            protoResult!
-          )) as DecodedMessage;
+      fc.asyncProperty(fc.uint8Array({ minLength: 1 }), async (payload) => {
+        const encoder = createEncoder({
+          contentTopic
+        });
+        const bytes = await encoder.toWire({ payload });
+        const decoder = createDecoder(contentTopic);
+        const protoResult = await decoder.fromWireToProtoObj(bytes);
+        const result = (await decoder.fromProtoObj(
+          pubsubTopic,
+          protoResult!
+        )) as DecodedMessage;
 
-          expect(result.contentTopic).to.eq(contentTopic);
-          expect(result.pubsubTopic).to.eq(pubsubTopic);
-          expect(result.version).to.eq(0);
-          expect(result.ephemeral).to.be.false;
-          expect(result.payload).to.deep.eq(payload);
-          expect(result.timestamp).to.not.be.undefined;
-        }
-      )
+        expect(result.contentTopic).to.eq(contentTopic);
+        expect(result.pubsubTopic).to.eq(pubsubTopic);
+        expect(result.version).to.eq(0);
+        expect(result.ephemeral).to.be.false;
+        expect(result.payload).to.deep.eq(payload);
+        expect(result.timestamp).to.not.be.undefined;
+      })
     );
   });
 
   it("Ephemeral field set to true", async function () {
     await fc.assert(
-      fc.asyncProperty(
-        fc.string({ minLength: 1 }),
-        fc.string({ minLength: 1 }),
-        fc.uint8Array({ minLength: 1 }),
-        async (contentTopic, pubsubTopic, payload) => {
-          const encoder = createEncoder({
-            contentTopic,
-            ephemeral: true
-          });
-          const bytes = await encoder.toWire({ payload });
-          const decoder = createDecoder(contentTopic);
-          const protoResult = await decoder.fromWireToProtoObj(bytes);
-          const result = (await decoder.fromProtoObj(
-            pubsubTopic,
-            protoResult!
-          )) as DecodedMessage;
+      fc.asyncProperty(fc.uint8Array({ minLength: 1 }), async (payload) => {
+        const encoder = createEncoder({
+          contentTopic,
+          ephemeral: true
+        });
+        const bytes = await encoder.toWire({ payload });
+        const decoder = createDecoder(contentTopic);
+        const protoResult = await decoder.fromWireToProtoObj(bytes);
+        const result = (await decoder.fromProtoObj(
+          pubsubTopic,
+          protoResult!
+        )) as DecodedMessage;
 
-          expect(result.ephemeral).to.be.true;
-        }
-      )
+        expect(result.ephemeral).to.be.true;
+      })
     );
   });
 
   it("Meta field set when metaSetter is specified", async function () {
     await fc.assert(
-      fc.asyncProperty(
-        fc.string({ minLength: 1 }),
-        fc.string({ minLength: 1 }),
-        fc.uint8Array({ minLength: 1 }),
-        async (contentTopic, pubsubTopic, payload) => {
-          // Encode the length of the payload
-          // Not a relevant real life example
-          const metaSetter = (
-            msg: IProtoMessage & { meta: undefined }
-          ): Uint8Array => {
-            const buffer = new ArrayBuffer(4);
-            const view = new DataView(buffer);
-            view.setUint32(0, msg.payload.length, false);
-            return new Uint8Array(buffer);
-          };
+      fc.asyncProperty(fc.uint8Array({ minLength: 1 }), async (payload) => {
+        // Encode the length of the payload
+        // Not a relevant real life example
+        const metaSetter = (
+          msg: IProtoMessage & { meta: undefined }
+        ): Uint8Array => {
+          const buffer = new ArrayBuffer(4);
+          const view = new DataView(buffer);
+          view.setUint32(0, msg.payload.length, false);
+          return new Uint8Array(buffer);
+        };
 
-          const encoder = createEncoder({
-            contentTopic,
-            ephemeral: true,
-            metaSetter
-          });
-          const bytes = await encoder.toWire({ payload });
-          const decoder = createDecoder(contentTopic);
-          const protoResult = await decoder.fromWireToProtoObj(bytes);
-          const result = (await decoder.fromProtoObj(
-            pubsubTopic,
-            protoResult!
-          )) as DecodedMessage;
+        const encoder = createEncoder({
+          contentTopic,
+          ephemeral: true,
+          metaSetter
+        });
+        const bytes = await encoder.toWire({ payload });
+        const decoder = createDecoder(contentTopic);
+        const protoResult = await decoder.fromWireToProtoObj(bytes);
+        const result = (await decoder.fromProtoObj(
+          pubsubTopic,
+          protoResult!
+        )) as DecodedMessage;
 
-          const expectedMeta = metaSetter({
-            payload,
-            timestamp: undefined,
-            contentTopic: "",
-            ephemeral: undefined,
-            meta: undefined,
-            rateLimitProof: undefined,
-            version: undefined
-          });
+        const expectedMeta = metaSetter({
+          payload,
+          timestamp: undefined,
+          contentTopic: "",
+          ephemeral: undefined,
+          meta: undefined,
+          rateLimitProof: undefined,
+          version: undefined
+        });
 
-          expect(result.meta).to.deep.eq(expectedMeta);
-        }
-      )
+        expect(result.meta).to.deep.eq(expectedMeta);
+      })
     );
   });
 });

packages/interfaces/src/waku.ts

@@ -1,5 +1,5 @@
 import type { PeerId, Stream } from "@libp2p/interface";
-import type { Multiaddr } from "@multiformats/multiaddr";
+import type { MultiaddrInput } from "@multiformats/multiaddr";
 
 import { IConnectionManager } from "./connection_manager.js";
 import type { IFilterSDK } from "./filter.js";
@@ -18,7 +18,7 @@ export interface Waku {
 
   connectionManager: IConnectionManager;
 
-  dial(peer: PeerId | Multiaddr, protocols?: Protocols[]): Promise<Stream>;
+  dial(peer: PeerId | MultiaddrInput, protocols?: Protocols[]): Promise<Stream>;
 
   start(): Promise<void>;
 

packages/message-encryption/src/ecies.spec.ts

@@ -1,20 +1,22 @@
 import { IProtoMessage } from "@waku/interfaces";
+import { contentTopicToPubsubTopic } from "@waku/utils";
 import { expect } from "chai";
 import fc from "fast-check";
 
 import { getPublicKey } from "./crypto/index.js";
 import { createDecoder, createEncoder } from "./ecies.js";
 
+const contentTopic = "/js-waku/1/tests/bytes";
+const pubsubTopic = contentTopicToPubsubTopic(contentTopic);
+
 describe("Ecies Encryption", function () {
   this.timeout(20000);
   it("Round trip binary encryption [ecies, no signature]", async function () {
     await fc.assert(
       fc.asyncProperty(
-        fc.string({ minLength: 1 }),
-        fc.string({ minLength: 1 }),
         fc.uint8Array({ minLength: 1 }),
         fc.uint8Array({ min: 1, minLength: 32, maxLength: 32 }),
-        async (pubsubTopic, contentTopic, payload, privateKey) => {
+        async (payload, privateKey) => {
           const publicKey = getPublicKey(privateKey);
 
           const encoder = createEncoder({
@@ -46,18 +48,10 @@ describe("Ecies Encryption", function () {
 
     await fc.assert(
       fc.asyncProperty(
-        fc.string({ minLength: 1 }),
-        fc.string({ minLength: 1 }),
         fc.uint8Array({ minLength: 1 }),
         fc.uint8Array({ min: 1, minLength: 32, maxLength: 32 }),
         fc.uint8Array({ min: 1, minLength: 32, maxLength: 32 }),
-        async (
-          pubsubTopic,
-          contentTopic,
-          payload,
-          alicePrivateKey,
-          bobPrivateKey
-        ) => {
+        async (payload, alicePrivateKey, bobPrivateKey) => {
           const alicePublicKey = getPublicKey(alicePrivateKey);
           const bobPublicKey = getPublicKey(bobPrivateKey);
 
@@ -89,11 +83,9 @@ describe("Ecies Encryption", function () {
   it("Check meta is set [ecies]", async function () {
     await fc.assert(
       fc.asyncProperty(
-        fc.string({ minLength: 1 }),
-        fc.string({ minLength: 1 }),
         fc.uint8Array({ minLength: 1 }),
         fc.uint8Array({ min: 1, minLength: 32, maxLength: 32 }),
-        async (pubsubTopic, contentTopic, payload, privateKey) => {
+        async (payload, privateKey) => {
           const publicKey = getPublicKey(privateKey);
           const metaSetter = (
             msg: IProtoMessage & { meta: undefined }

packages/message-encryption/src/symmetric.spec.ts

@@ -1,19 +1,21 @@
 import { IProtoMessage } from "@waku/interfaces";
+import { contentTopicToPubsubTopic } from "@waku/utils";
 import { expect } from "chai";
 import fc from "fast-check";
 
 import { getPublicKey } from "./crypto/index.js";
 import { createDecoder, createEncoder } from "./symmetric.js";
 
+const contentTopic = "/js-waku/1/tests/bytes";
+const pubsubTopic = contentTopicToPubsubTopic(contentTopic);
+
 describe("Symmetric Encryption", function () {
   it("Round trip binary encryption [symmetric, no signature]", async function () {
     await fc.assert(
       fc.asyncProperty(
-        fc.string({ minLength: 1 }),
-        fc.string({ minLength: 1 }),
         fc.uint8Array({ minLength: 1 }),
         fc.uint8Array({ min: 1, minLength: 32, maxLength: 32 }),
-        async (pubsubTopic, contentTopic, payload, symKey) => {
+        async (payload, symKey) => {
           const encoder = createEncoder({
             contentTopic,
             symKey
@@ -41,12 +43,10 @@ describe("Symmetric Encryption", function () {
   it("Round trip binary encryption [symmetric, signature]", async function () {
     await fc.assert(
       fc.asyncProperty(
-        fc.string({ minLength: 1 }),
-        fc.string({ minLength: 1 }),
         fc.uint8Array({ minLength: 1 }),
         fc.uint8Array({ min: 1, minLength: 32, maxLength: 32 }),
         fc.uint8Array({ min: 1, minLength: 32, maxLength: 32 }),
-        async (pubsubTopic, contentTopic, payload, sigPrivKey, symKey) => {
+        async (payload, sigPrivKey, symKey) => {
           const sigPubKey = getPublicKey(sigPrivKey);
 
           const encoder = createEncoder({
@@ -77,11 +77,9 @@ describe("Symmetric Encryption", function () {
   it("Check meta is set [symmetric]", async function () {
     await fc.assert(
       fc.asyncProperty(
-        fc.string({ minLength: 1 }),
-        fc.string({ minLength: 1 }),
         fc.uint8Array({ minLength: 1 }),
         fc.uint8Array({ min: 1, minLength: 32, maxLength: 32 }),
-        async (pubsubTopic, contentTopic, payload, symKey) => {
+        async (payload, symKey) => {
           const metaSetter = (
             msg: IProtoMessage & { meta: undefined }
           ): Uint8Array => {

packages/sdk/src/protocols/filter.ts

@@ -294,7 +294,21 @@
     decoders: IDecoder<T> | IDecoder<T>[],
     callback: Callback<T>
   ): Promise<Unsubscribe> {
-    const subscription = await this.createSubscription();
+    const pubsubTopics = this.getPubsubTopics(decoders);
+
+    if (pubsubTopics.length === 0) {
+      throw Error(
+        "Failed to subscribe: no pubsubTopic found on decoders provided."
+      );
+    }
+
+    if (pubsubTopics.length > 1) {
+      throw Error(
+        "Failed to subscribe: all decoders should have the same pubsub topic. Use createSubscription to be more agile."
+      );
+    }
+
+    const subscription = await this.createSubscription(pubsubTopics[0]);
 
     await subscription.subscribe(decoders, callback);
 
@@ -314,6 +328,20 @@
   ): Promise<IAsyncIterator<T>> {
     return toAsyncIterator(this, decoders);
   }
+
+  private getPubsubTopics(decoders: IDecoder<any> | IDecoder<any>[]): string[] {
+    if (!Array.isArray(decoders)) {
+      return [decoders.pubsubTopic];
+    }
+
+    if (decoders.length === 0) {
+      return [];
+    }
+
+    const pubsubTopics = new Set(decoders.map((d) => d.pubsubTopic));
+
+    return [...pubsubTopics];
+  }
 }
 
 export function wakuFilter(

packages/tests/tests/connection-mananger/connection_state.spec.ts

@@ -87,7 +87,7 @@ describe("Connection state", function () {
     expect(eventCount).to.be.eq(1);
   });
 
-  it("`waku:online` bwtween 2 js-waku relay nodes", async function () {
+  it("`waku:online` between 2 js-waku relay nodes", async function () {
     const waku1 = await createRelayNode({
       staticNoiseKey: NOISE_KEY_1
     });
@@ -159,7 +159,7 @@ describe("Connection state", function () {
     expect(waku.isConnected()).to.be.false;
   });
 
-  it("isConnected bwtween 2 js-waku relay nodes", async function () {
+  it("isConnected between 2 js-waku relay nodes", async function () {
     const waku1 = await createRelayNode({
       staticNoiseKey: NOISE_KEY_1
     });

packages/tests/tests/filter/utils.ts

@@ -26,7 +26,7 @@ import {
 
 // Constants for test configuration.
 export const log = new Logger("test:filter");
-export const TestContentTopic = "/test/1/waku-filter";
+export const TestContentTopic = "/test/1/waku-filter/default";
 export const TestEncoder = createEncoder({ contentTopic: TestContentTopic });
 export const TestDecoder = createDecoder(TestContentTopic);
 export const messageText = "Filtering works!";

packages/tests/tests/utils.spec.ts

@@ -22,7 +22,7 @@ import {
 
 chai.use(chaiAsPromised);
 
-const TestContentTopic = "/test/1/waku-filter";
+const TestContentTopic = "/test/1/waku-filter/default";
 const TestEncoder = createEncoder({ contentTopic: TestContentTopic });
 const TestDecoder = createDecoder(TestContentTopic);