feat: introduce Send API

packages/core/src/lib/light_push/light_push.ts

@@ -55,11 +55,11 @@ export class LightPushCore {
       };
     }
 
-    const { rpc, error: prepError } = await ProtocolHandler.preparePushMessage(
-      encoder,
-      message,
-      protocol
-    );
+    const {
+      rpc,
+      error: prepError,
+      message: protoMessage
+    } = await ProtocolHandler.preparePushMessage(encoder, message, protocol);
 
     if (prepError) {
       return {
@@ -117,7 +117,21 @@ export class LightPushCore {
       };
     }
 
-    return ProtocolHandler.handleResponse(bytes, protocol, peerId);
+    const processedResponse = ProtocolHandler.handleResponse(
+      bytes,
+      protocol,
+      peerId
+    );
+
+    if (processedResponse.success) {
+      return {
+        success: processedResponse.success,
+        failure: null,
+        message: protoMessage
+      };
+    }
+
+    return processedResponse;
   }
 
   private async getProtocol(

packages/core/src/lib/light_push/protocol_handler.ts

@@ -1,5 +1,10 @@
 import type { PeerId } from "@libp2p/interface";
-import type { IEncoder, IMessage, LightPushCoreResult } from "@waku/interfaces";
+import type {
+  IEncoder,
+  IMessage,
+  IProtoMessage,
+  LightPushCoreResult
+} from "@waku/interfaces";
 import { LightPushError, LightPushStatusCode } from "@waku/interfaces";
 import { PushResponse, WakuMessage } from "@waku/proto";
 import { isMessageSizeUnderCap, Logger } from "@waku/utils";
@@ -15,8 +20,8 @@ type VersionedPushRpc =
   | ({ version: "v3" } & PushRpc);
 
 type PreparePushMessageResult =
-  | { rpc: VersionedPushRpc; error: null }
-  | { rpc: null; error: LightPushError };
+  | { rpc: VersionedPushRpc; error: null; message?: IProtoMessage }
+  | { rpc: null; error: LightPushError; message?: IProtoMessage };
 
 const log = new Logger("light-push:protocol-handler");
 
@@ -47,13 +52,15 @@ export class ProtocolHandler {
         log.info("Creating v3 RPC message");
         return {
           rpc: ProtocolHandler.createV3Rpc(protoMessage, encoder.pubsubTopic),
-          error: null
+          error: null,
+          message: protoMessage
         };
       }
 
       log.info("Creating v2 RPC message");
       return {
         rpc: ProtocolHandler.createV2Rpc(protoMessage, encoder.pubsubTopic),
+        message: protoMessage,
         error: null
       };
     } catch (err) {

packages/core/src/lib/message/constants.ts

@@ -0,0 +1,2 @@
+export const OneMillion = BigInt(1_000_000);
+export const Version = 0;

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

@@ -1 +1,2 @@
 export * as version_0 from "./version_0.js";
+export { OneMillion, Version } from "./constants.js";

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

@@ -16,10 +16,10 @@ import { bytesToHex } from "@waku/utils/bytes";
 
 import { messageHash } from "../message_hash/index.js";
 
+import { OneMillion, Version } from "./constants.js";
+
 const log = new Logger("message:version-0");
-const OneMillion = BigInt(1_000_000);
 
-export const Version = 0;
 export { proto };
 
 export class DecodedMessage implements IDecodedMessage {

packages/interfaces/src/message.ts

@@ -69,6 +69,21 @@ export interface IMessage {
   rateLimitProof?: IRateLimitProof;
 }
 
+/**
+ * Send message data structure used in {@link IWaku.send}.
+ */
+export interface ISendMessage {
+  contentTopic: string;
+  payload: Uint8Array;
+  ephemeral?: boolean;
+  rateLimitProof?: boolean;
+}
+
+/**
+ * Request ID of attempt to send a message.
+ */
+export type RequestId = string;
+
 export interface IMetaSetter {
   (message: IProtoMessage & { meta: undefined }): Uint8Array;
 }

packages/interfaces/src/protocols.ts

@@ -5,7 +5,7 @@ import type { DiscoveryOptions, PeerCache } from "./discovery.js";
 import type { FilterProtocolOptions } from "./filter.js";
 import type { CreateLibp2pOptions } from "./libp2p.js";
 import type { LightPushProtocolOptions } from "./light_push.js";
-import type { IDecodedMessage } from "./message.js";
+import type { IDecodedMessage, IProtoMessage } from "./message.js";
 import type { ThisAndThat, ThisOrThat } from "./misc.js";
 import { NetworkConfig } from "./sharding.js";
 import type { StoreProtocolOptions } from "./store.js";
@@ -195,7 +195,13 @@ export type LightPushCoreResult = ThisOrThat<
   PeerId,
   "failure",
   LightPushFailure
->;
+> & {
+  /**
+   * The proto object of the message.
+   * Present only if the message was successfully pushed to the network.
+   */
+  message?: IProtoMessage;
+};
 
 export type FilterCoreResult = ThisOrThat<
   "success",
@@ -209,7 +215,13 @@ export type LightPushSDKResult = ThisAndThat<
   PeerId[],
   "failures",
   LightPushFailure[]
->;
+> & {
+  /**
+   * The proto objects of the messages.
+   * Present only if the messages were successfully pushed to the network.
+   */
+  messages?: IProtoMessage[];
+};
 
 export type FilterSDKResult = ThisAndThat<
   "successes",

packages/interfaces/src/sender.ts

@@ -20,6 +20,12 @@ export type ISendOptions = {
    * @default false
    */
   useLegacy?: boolean;
+
+  /**
+   * Amount of peers to send message to.
+   * Overrides `numPeersToUse` in {@link @waku/interfaces!CreateNodeOptions}.
+   */
+  numPeersToUse?: number;
 };
 
 export interface ISender {

packages/interfaces/src/waku.ts

@@ -10,7 +10,13 @@ import type { IFilter } from "./filter.js";
 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 {
+  IDecodedMessage,
+  IDecoder,
+  IEncoder,
+  ISendMessage,
+  RequestId
+} from "./message.js";
 import type { Protocols } from "./protocols.js";
 import type { IRelay } from "./relay.js";
 import type { ShardId } from "./sharding.js";
@@ -58,9 +64,22 @@ export type IWakuEventEmitter = TypedEventEmitter<IWakuEvents>;
 
 export interface IWaku {
   libp2p: Libp2p;
+
+  /**
+   * @deprecated should not be accessed directly, use {@link IWaku.send} and {@link IWaku.subscribe} instead
+   */
   relay?: IRelay;
+
   store?: IStore;
+
+  /**
+   * @deprecated should not be accessed directly, use {@link IWaku.subscribe} instead
+   */
   filter?: IFilter;
+
+  /**
+   * @deprecated should not be accessed directly, use {@link IWaku.send} instead
+   */
   lightPush?: ILightPush;
 
   /**
@@ -251,6 +270,14 @@ export interface IWaku {
    */
   createEncoder(params: CreateEncoderParams): IEncoder;
 
+  /**
+   * Sends a message to the Waku network.
+   *
+   * @param {ISendMessage} message - The message to send.
+   * @returns {Promise<RequestId>} A promise that resolves to the request ID
+   */
+  send(message: ISendMessage): Promise<RequestId>;
+
   /**
    * @returns {boolean} `true` if the node was started and `false` otherwise
    */

packages/sdk/package.json

@@ -75,7 +75,8 @@
     "@waku/sds": "^0.0.7",
     "@waku/utils": "0.0.27",
     "libp2p": "2.8.11",
-    "lodash.debounce": "^4.0.8"
+    "lodash.debounce": "^4.0.8",
+    "uuid": "^10.0.0"
   },
   "devDependencies": {
     "@libp2p/interface": "2.10.4",

packages/sdk/src/light_push/light_push.ts

@@ -4,6 +4,7 @@ import {
   type IEncoder,
   ILightPush,
   type IMessage,
+  IProtoMessage,
   type ISendOptions,
   type Libp2p,
   LightPushCoreResult,
@@ -82,23 +83,27 @@ export class LightPush implements ILightPush {
 
     log.info("send: attempting to send a message to pubsubTopic:", pubsubTopic);
 
-    const peerIds = await this.peerManager.getPeers({
+    let peerIds = await this.peerManager.getPeers({
       protocol: options.useLegacy ? "light-push-v2" : Protocols.LightPush,
       pubsubTopic: encoder.pubsubTopic
     });
+    peerIds = peerIds.slice(0, options.numPeersToUse);
 
     const coreResults =
       peerIds?.length > 0
         ? await Promise.all(
             peerIds.map((peerId) =>
               this.protocol
                 .send(encoder, message, peerId, options.useLegacy)
-                .catch((_e) => ({
-                  success: null,
-                  failure: {
-                    error: LightPushError.GENERIC_FAIL
-                  }
-                }))
+                .catch(
+                  (_e) =>
+                    ({
+                      success: null,
+                      failure: {
+                        error: LightPushError.GENERIC_FAIL
+                      }
+                    }) as LightPushCoreResult
+                )
             )
           )
         : [];
@@ -110,7 +115,10 @@ export class LightPush implements ILightPush {
             .map((v) => v.success) as PeerId[],
           failures: coreResults
             .filter((v) => v.failure)
-            .map((v) => v.failure) as LightPushFailure[]
+            .map((v) => v.failure) as LightPushFailure[],
+          messages: coreResults
+            .filter((v) => v.message)
+            .map((v) => v.message) as IProtoMessage[]
         }
       : {
           successes: [],