feat(filter)!: new simpler filter API

packages/interfaces/src/filter.ts

@@ -1,15 +1,14 @@
 import type { PeerId } from "@libp2p/interface";
 
 import type { IDecodedMessage, IDecoder } from "./message.js";
-import type { ContentTopic, PubsubTopic, ThisOrThat } from "./misc.js";
+import type { ContentTopic, ThisOrThat } from "./misc.js";
 import type {
   Callback,
   IBaseProtocolCore,
   IBaseProtocolSDK,
   ProtocolError,
   ProtocolUseOptions,
-  SDKProtocolResult,
-  ShardingParams
+  SDKProtocolResult
 } from "./protocols.js";
 import type { IReceiver } from "./receiver.js";
 
@@ -37,12 +36,28 @@ export interface ISubscriptionSDK {
 
 export type IFilterSDK = IReceiver &
   IBaseProtocolSDK & { protocol: IBaseProtocolCore } & {
-    createSubscription(
-      pubsubTopicShardInfo?: ShardingParams | PubsubTopic,
-      options?: ProtocolUseOptions
-    ): Promise<CreateSubscriptionResult>;
+    subscribe<T extends IDecodedMessage>(
+      decoders: IDecoder<T> | IDecoder<T>[],
+      callback: Callback<T>,
+      protocolUseOptions?: ProtocolUseOptions,
+      subscribeOptions?: SubscribeOptions
+    ): Promise<SubscribeResult>;
   };
 
+export type SubscribeResult = SubscriptionSuccess | SubscriptionError;
+
+type SubscriptionSuccess = {
+  subscription: ISubscriptionSDK;
+  error: null;
+  results: SDKProtocolResult;
+};
+
+type SubscriptionError = {
+  subscription: null;
+  error: ProtocolError;
+  results: null;
+};
+
 export type CreateSubscriptionResult = ThisOrThat<
   "subscription",
   ISubscriptionSDK,

packages/interfaces/src/protocols.ts

@@ -173,6 +173,11 @@ export enum ProtocolError {
    * Ensure that the pubsub topic used for decoder creation is the same as the one used for protocol.
    */
   TOPIC_DECODER_MISMATCH = "Topic decoder mismatch",
+  /**
+   * The topics passed in the decoders do not match each other, or don't exist at all.
+   * Ensure that all the pubsub topics used in the decoders are valid and match each other.
+   */
+  INVALID_DECODER_TOPICS = "Invalid decoder topics",
   /**
    * Failure to find a peer with suitable protocols. This may due to a connection issue.
    * Mitigation can be: retrying after a given time period, display connectivity issue

packages/interfaces/src/receiver.ts

@@ -13,8 +13,10 @@ export interface IReceiver {
   toSubscriptionIterator: <T extends IDecodedMessage>(
     decoders: IDecoder<T> | IDecoder<T>[]
   ) => Promise<IAsyncIterator<T>>;
-  subscribe: <T extends IDecodedMessage>(
-    decoders: IDecoder<T> | IDecoder<T>[],
-    callback: Callback<T>
-  ) => Unsubscribe | Promise<Unsubscribe>;
+  subscribeWithUnsubscribe: SubscribeWithUnsubscribe;
 }
+
+type SubscribeWithUnsubscribe = <T extends IDecodedMessage>(
+  decoders: IDecoder<T> | IDecoder<T>[],
+  callback: Callback<T>
+) => Unsubscribe | Promise<Unsubscribe>;

packages/relay/src/index.ts

@@ -148,7 +148,7 @@ class Relay implements IRelay {
     };
   }
 
-  public subscribe<T extends IDecodedMessage>(
+  public subscribeWithUnsubscribe<T extends IDecodedMessage>(
     decoders: IDecoder<T> | IDecoder<T>[],
     callback: Callback<T>
   ): () => void {
@@ -171,6 +171,8 @@ class Relay implements IRelay {
     };
   }
 
+  public subscribe = this.subscribeWithUnsubscribe;
+
   private removeObservers<T extends IDecodedMessage>(
     observers: Array<[PubsubTopic, Observer<T>]>
   ): void {

packages/sdk/src/protocols/filter.ts

@@ -21,6 +21,7 @@ import {
   type SDKProtocolResult,
   type ShardingParams,
   type SubscribeOptions,
+  SubscribeResult,
   type Unsubscribe
 } from "@waku/interfaces";
 import { messageHashStr } from "@waku/message-hash";
@@ -448,19 +449,89 @@ class FilterSDK extends BaseProtocolSDK implements IFilterSDK {
     this.activeSubscriptions = new Map();
   }
 
-  //TODO: move to SubscriptionManager
-  private getActiveSubscription(
-    pubsubTopic: PubsubTopic
-  ): SubscriptionManager | undefined {
-    return this.activeSubscriptions.get(pubsubTopic);
-  }
+  /**
+   * Opens a subscription with the Filter protocol using the provided decoders and callback.
+   * This method combines the functionality of creating a subscription and subscribing to it.
+   *
+   * @param {IDecoder<T> | IDecoder<T>[]} decoders - A single decoder or an array of decoders to use for decoding messages.
+   * @param {Callback<T>} callback - The callback function to be invoked with decoded messages.
+   * @param {ProtocolUseOptions} [protocolUseOptions] - Optional settings for using the protocol.
+   * @param {SubscribeOptions} [subscribeOptions=DEFAULT_SUBSCRIBE_OPTIONS] - Options for the subscription.
+   *
+   * @returns {Promise<SubscribeResult>} A promise that resolves to an object containing:
+   *   - subscription: The created subscription object if successful, or null if failed.
+   *   - error: A ProtocolError if the subscription creation failed, or null if successful.
+   *   - results: An object containing arrays of failures and successes from the subscription process.
+   *     Only present if the subscription was created successfully.
+   *
+   * @throws {Error} If there's an unexpected error during the subscription process.
+   *
+   * @remarks
+   * This method attempts to create a subscription using the pubsub topic derived from the provided decoders,
+   * then tries to subscribe using the created subscription. The return value should be interpreted as follows:
+   * - If `subscription` is null and `error` is non-null, a critical error occurred and the subscription failed completely.
+   * - If `subscription` is non-null and `error` is null, the subscription was created successfully.
+   *   In this case, check the `results` field for detailed information about successes and failures during the subscription process.
+   * - Even if the subscription was created successfully, there might be some failures in the results.
+   *
+   * @example
+   * ```typescript
+   * const {subscription, error, results} = await waku.filter.subscribe(decoders, callback);
+   * if (!subscription || error) {
+   *   console.error("Failed to create subscription:", error);
+   * }
+   * console.log("Subscription created successfully");
+   * if (results.failures.length > 0) {
+   *   console.warn("Some errors occurred during subscription:", results.failures);
+   * }
+   * console.log("Successful subscriptions:", results.successes);
+   *
+   * ```
+   */
+  public async subscribe<T extends IDecodedMessage>(
+    decoders: IDecoder<T> | IDecoder<T>[],
+    callback: Callback<T>,
+    protocolUseOptions?: ProtocolUseOptions,
+    subscribeOptions: SubscribeOptions = DEFAULT_SUBSCRIBE_OPTIONS
+  ): Promise<SubscribeResult> {
+    const uniquePubsubTopics = this.getUniquePubsubTopics(decoders);
 
-  private setActiveSubscription(
-    pubsubTopic: PubsubTopic,
-    subscription: SubscriptionManager
-  ): SubscriptionManager {
-    this.activeSubscriptions.set(pubsubTopic, subscription);
-    return subscription;
+    if (uniquePubsubTopics.length !== 1) {
+      return {
+        subscription: null,
+        error: ProtocolError.INVALID_DECODER_TOPICS,
+        results: null
+      };
+    }
+
+    const pubsubTopic = uniquePubsubTopics[0];
+
+    const { subscription, error } = await this.createSubscription(
+      pubsubTopic,
+      protocolUseOptions
+    );
+
+    if (error) {
+      return {
+        subscription: null,
+        error: error,
+        results: null
+      };
+    }
+
+    const { failures, successes } = await subscription.subscribe(
+      decoders,
+      callback,
+      subscribeOptions
+    );
+    return {
+      subscription,
+      error: null,
+      results: {
+        failures: failures,
+        successes: successes
+      }
+    };
   }
 
   /**
@@ -469,7 +540,7 @@ class FilterSDK extends BaseProtocolSDK implements IFilterSDK {
    * @param pubsubTopicShardInfo The pubsub topic to subscribe to.
    * @returns The subscription object.
    */
-  public async createSubscription(
+  private async createSubscription(
     pubsubTopicShardInfo: ShardingParams | PubsubTopic,
     options?: ProtocolUseOptions
   ): Promise<CreateSubscriptionResult> {
@@ -516,7 +587,6 @@ class FilterSDK extends BaseProtocolSDK implements IFilterSDK {
     };
   }
 
-  //TODO: remove this dependency on IReceiver
   /**
    * This method is used to satisfy the `IReceiver` interface.
    *
@@ -532,7 +602,7 @@ class FilterSDK extends BaseProtocolSDK implements IFilterSDK {
    * This method should not be used directly.
    * Instead, use `createSubscription` to create a new subscription.
    */
-  public async subscribe<T extends IDecodedMessage>(
+  public async subscribeWithUnsubscribe<T extends IDecodedMessage>(
     decoders: IDecoder<T> | IDecoder<T>[],
     callback: Callback<T>,
     options: SubscribeOptions = DEFAULT_SUBSCRIBE_OPTIONS
@@ -578,6 +648,21 @@ class FilterSDK extends BaseProtocolSDK implements IFilterSDK {
     return toAsyncIterator(this, decoders);
   }
 
+  //TODO: move to SubscriptionManager
+  private getActiveSubscription(
+    pubsubTopic: PubsubTopic
+  ): SubscriptionManager | undefined {
+    return this.activeSubscriptions.get(pubsubTopic);
+  }
+
+  private setActiveSubscription(
+    pubsubTopic: PubsubTopic,
+    subscription: SubscriptionManager
+  ): SubscriptionManager {
+    this.activeSubscriptions.set(pubsubTopic, subscription);
+    return subscription;
+  }
+
   private getUniquePubsubTopics<T extends IDecodedMessage>(
     decoders: IDecoder<T> | IDecoder<T>[]
   ): string[] {

packages/tests/tests/ephemeral.node.spec.ts

@@ -4,7 +4,7 @@ import {
   DecodedMessage,
   waitForRemotePeer
 } from "@waku/core";
-import { ISubscriptionSDK, Protocols } from "@waku/interfaces";
+import { Protocols } from "@waku/interfaces";
 import type { LightNode } from "@waku/interfaces";
 import {
   generatePrivateKey,
@@ -83,8 +83,6 @@ describe("Waku Message Ephemeral field", function () {
   let waku: LightNode;
   let nwaku: ServiceNode;
 
-  let subscription: ISubscriptionSDK;
-
   afterEachCustom(this, async () => {
     await tearDownNodes(nwaku, waku);
   });
@@ -122,11 +120,6 @@ describe("Waku Message Ephemeral field", function () {
       Protocols.LightPush,
       Protocols.Store
     ]);
-
-    const { error, subscription: _subscription } =
-      await waku.filter.createSubscription(TestEncoder.pubsubTopic);
-    if (error) throw error;
-    subscription = _subscription;
   });
 
   it("Ephemeral messages are not stored", async function () {
@@ -218,7 +211,7 @@ describe("Waku Message Ephemeral field", function () {
     const callback = (msg: DecodedMessage): void => {
       messages.push(msg);
     };
-    await subscription.subscribe([TestDecoder], callback);
+    await waku.filter.subscribe([TestDecoder], callback);
 
     await delay(200);
     const normalTxt = "Normal message";
@@ -265,7 +258,7 @@ describe("Waku Message Ephemeral field", function () {
     const callback = (msg: DecodedMessage): void => {
       messages.push(msg);
     };
-    await subscription.subscribe([decoder], callback);
+    await waku.filter.subscribe([decoder], callback);
 
     await delay(200);
     const normalTxt = "Normal message";
@@ -316,7 +309,7 @@ describe("Waku Message Ephemeral field", function () {
     const callback = (msg: DecodedMessage): void => {
       messages.push(msg);
     };
-    await subscription.subscribe([decoder], callback);
+    await waku.filter.subscribe([decoder], callback);
 
     await delay(200);
     const normalTxt = "Normal message";