feat: Waku API's subscribe

packages/interfaces/src/filter.ts

@@ -1,5 +1,6 @@
-import type { IDecodedMessage, IDecoder } from "./message.js";
-import type { Callback } from "./protocols.js";
+import type { IDecodedMessage } from "./message.js";
+import { ContentTopic } from "./misc.js";
+import { IRoutingInfo } from "./sharding.js";
 
 export type IFilter = {
   readonly multicodec: string;
@@ -38,9 +39,10 @@ export type IFilter = {
    *   console.error("Failed to subscribe");
    * }
    */
-  subscribe<T extends IDecodedMessage>(
-    decoders: IDecoder<T> | IDecoder<T>[],
-    callback: Callback<T>
+  subscribe(
+    contentTopics: ContentTopic[],
+    routingInfo: IRoutingInfo,
+    callback: (msg: IDecodedMessage) => void | Promise<void>
   ): Promise<boolean>;
 
   /**
@@ -64,8 +66,9 @@ export type IFilter = {
    *   console.error("Failed to unsubscribe");
    * }
    */
-  unsubscribe<T extends IDecodedMessage>(
-    decoders: IDecoder<T> | IDecoder<T>[]
+  unsubscribe(
+    contentTopics: ContentTopic[],
+    routingInfo: IRoutingInfo
   ): Promise<boolean>;
 
   /**

packages/interfaces/src/message.ts

@@ -105,6 +105,7 @@ export interface IEncoder {
 export interface IDecoder<T extends IDecodedMessage> {
   contentTopic: string;
   pubsubTopic: PubsubTopic;
+  routingInfo: IRoutingInfo;
   fromWireToProtoObj: (bytes: Uint8Array) => Promise<IProtoMessage | undefined>;
   fromProtoObj: (
     pubsubTopic: string,

packages/interfaces/src/waku.ts

@@ -11,6 +11,7 @@ import type { HealthStatus } from "./health_status.js";
 import type { Libp2p } from "./libp2p.js";
 import type { ILightPush } from "./light_push.js";
 import { IDecodedMessage, IDecoder, IEncoder } from "./message.js";
+import { ContentTopic } from "./misc.js";
 import type { Protocols } from "./protocols.js";
 import type { IRelay } from "./relay.js";
 import type { ShardId } from "./sharding.js";
@@ -54,7 +55,12 @@ export interface IWakuEvents {
   [WakuEvent.Health]: CustomEvent<HealthStatus>;
 }
 
+export interface IMessageEmitterEvents {
+  [contentTopic: string]: CustomEvent<Uint8Array>;
+}
+
 export type IWakuEventEmitter = TypedEventEmitter<IWakuEvents>;
+export type IMessageEmitter = TypedEventEmitter<IMessageEmitterEvents>;
 
 export interface IWaku {
   libp2p: Libp2p;
@@ -78,6 +84,20 @@ export interface IWaku {
    */
   events: IWakuEventEmitter;
 
+  /**
+   * Emits messages on their content topic. Messages may be coming from subscriptions
+   * or store queries (TODO). The payload is directly emitted
+   *
+   * @example
+   * ```typescript
+   * waku.messageEmitter.addEventListener("/some/0/content-topic/proto", (event) => {
+   *  const payload: UInt8Array = event.detail
+   *   MyDecoder.decode(payload);
+   * });
+   * ```
+   */
+  messageEmitter: IMessageEmitter;
+
   /**
    * Returns a unique identifier for a node on the network.
    *
@@ -251,6 +271,8 @@ export interface IWaku {
    */
   createEncoder(params: CreateEncoderParams): IEncoder;
 
+  subscribe(contentTopics: ContentTopic[]): Promise<void>;
+
   /**
    * @returns {boolean} `true` if the node was started and `false` otherwise
    */

packages/message-encryption/src/crypto/utils.ts

@@ -6,8 +6,8 @@
 
 import { Asymmetric, Symmetric } from "../misc.js";
 
 declare const self: Record<string, any> | undefined;
 const crypto: { node?: any; web?: any } = {
   node: nodeCrypto,
   web: typeof self === "object" && "crypto" in self ? self.crypto : undefined
 };
@@ -74,3 +74,32 @@
 export function keccak256(input: Uint8Array): Uint8Array {
   return new Uint8Array(sha3.keccak256.arrayBuffer(input));
 }
+
+/**
+ * Compare two public keys, can be used to verify that a given signature matches
+ * expectations.
+ *
+ * @param publicKeyA - The first public key to compare
+ * @param publicKeyB - The second public key to compare
+ * @returns true if the public keys are the same
+ */
+export function comparePublicKeys(
+  publicKeyA: Uint8Array | undefined,
+  publicKeyB: Uint8Array | undefined
+): boolean {
+  if (!publicKeyA || !publicKeyB) {
+    return false;
+  }
+
+  if (publicKeyA.length !== publicKeyB.length) {
+    return false;
+  }
+
+  for (let i = 0; i < publicKeyA.length; i++) {
+    if (publicKeyA[i] !== publicKeyB[i]) {
+      return false;
+    }
+  }
+
+  return true;
+}

packages/message-encryption/src/index.ts

@@ -1,11 +1,17 @@
 import {
+  comparePublicKeys,
   generatePrivateKey,
   generateSymmetricKey,
   getPublicKey
 } from "./crypto/index.js";
 import { DecodedMessage } from "./decoded_message.js";
 
-export { generatePrivateKey, generateSymmetricKey, getPublicKey };
+export {
+  generatePrivateKey,
+  generateSymmetricKey,
+  getPublicKey,
+  comparePublicKeys
+};
 export type { DecodedMessage };
 
 export * as ecies from "./ecies.js";

packages/message-encryption/src/symmetric.ts

@@ -206,3 +206,105 @@ export function createDecoder(
 ): Decoder {
   return new Decoder(contentTopic, routingInfo, symKey);
 }
+
+/**
+ * Result of decrypting a message with AES symmetric encryption.
+ */
+export interface SymmetricDecryptionResult {
+  /** The decrypted payload */
+  payload: Uint8Array;
+  /** The signature if the message was signed */
+  signature?: Uint8Array;
+  /** The recovered public key if the message was signed */
+  signaturePublicKey?: Uint8Array;
+}
+
+/**
+ * AES symmetric encryption.
+ *
+ *
+ * Follows [26/WAKU2-PAYLOAD](https://rfc.vac.dev/spec/26/) encryption standard.
+ */
+export class SymmetricEncryption {
+  /**
+   * Creates an AES Symmetric encryption instance.
+   *
+   * @param symKey - The symmetric key for encryption (32 bytes recommended)
+   * @param sigPrivKey - Optional private key to sign messages before encryption
+   */
+  public constructor(
+    private symKey: Uint8Array,
+    private sigPrivKey?: Uint8Array
+  ) {}
+
+  /**
+   * Encrypts a byte array payload.
+   *
+   * The encryption process:
+   * 1. Optionally signs the payload with the private key
+   * 2. Adds padding to obscure payload size
+   * 3. Encrypts using AES-256-GCM
+   *
+   * @param payload - The data to encrypt
+   * @returns The encrypted payload
+   */
+  public async encrypt(payload: Uint8Array): Promise<Uint8Array> {
+    const preparedPayload = await preCipher(payload, this.sigPrivKey);
+    return encryptSymmetric(preparedPayload, this.symKey);
+  }
+}
+
+/**
+ * AES symmetric decryption.
+ *
+ * Follows [26/WAKU2-PAYLOAD](https://rfc.vac.dev/spec/26/) encryption standard.
+ */
+export class SymmetricDecryption {
+  /**
+   * Creates an AES Symmetric decryption instance.
+   *
+   * @param symKey - The symmetric key for decryption (must match encryption key)
+   */
+  public constructor(private symKey: Uint8Array) {}
+
+  /**
+   * Decrypts an encrypted byte array payload.
+   *
+   * The decryption process:
+   * 1. Decrypts using AES-256-GCM
+   * 2. Removes padding
+   * 3. Verifies and recovers signature if present
+   *
+   * @param encryptedPayload - The encrypted data (from [[SymmetricEncryption.encrypt]])
+   * @returns Object containing the decrypted payload and signature info, or undefined if decryption fails
+   */
+  public async decrypt(
+    encryptedPayload: Uint8Array
+  ): Promise<SymmetricDecryptionResult | undefined> {
+    try {
+      const decryptedData = await decryptSymmetric(
+        encryptedPayload,
+        this.symKey
+      );
+
+      if (!decryptedData) {
+        return undefined;
+      }
+
+      const result = postCipher(decryptedData);
+
+      if (!result) {
+        return undefined;
+      }
+
+      return {
+        payload: result.payload,
+        signature: result.sig?.signature,
+        signaturePublicKey: result.sig?.publicKey
+      };
+    } catch (error) {
+      log.error("Failed to decrypt payload", error);
+      return undefined;
+    }
+  }
+}

packages/sdk/src/filter/filter.spec.ts

@@ -49,15 +49,22 @@ describe("Filter SDK", () => {
     const addStub = sinon.stub(Subscription.prototype, "add").resolves(true);
     const startStub = sinon.stub(Subscription.prototype, "start");
 
-    const result = await filter.subscribe(decoder, callback);
+    const result = await filter.subscribe(
+      [testContentTopic],
+      testRoutingInfo,
+      callback
+    );
 
     expect(result).to.be.true;
     expect(addStub.calledOnce).to.be.true;
     expect(startStub.calledOnce).to.be.true;
   });
 
   it("should return false when unsubscribing from a non-existing subscription", async () => {
-    const result = await filter.unsubscribe(decoder);
+    const result = await filter.unsubscribe(
+      [testContentTopic],
+      testRoutingInfo
+    );
     expect(result).to.be.false;
   });