fix: ensure that some fields are not omitted by `Message` implementation

CHANGELOG.md

@@ -11,6 +11,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - `RateLimitProof` field in Waku Message protobuf for RLN.
 
+### Changed
+
+- `Message` interface changed to ensure implementations do not omit fields.
+
 ## [0.29.0] - 2022-09-21
 
 ### Changed

src/lib/interfaces.ts

@@ -62,24 +62,24 @@ export interface RateLimitProof {
 }
 
 export interface ProtoMessage {
-  payload?: Uint8Array;
-  contentTopic?: string;
-  version?: number;
-  timestamp?: bigint;
-  rateLimitProof?: RateLimitProof;
+  payload: Uint8Array | undefined;
+  contentTopic: string | undefined;
+  version: number | undefined;
+  timestamp: bigint | undefined;
+  rateLimitProof: RateLimitProof | undefined;
 }
 
 export interface Message {
-  payload?: Uint8Array;
-  contentTopic?: string;
-  timestamp?: Date;
-  rateLimitProof?: RateLimitProof;
+  payload: Uint8Array | undefined;
+  contentTopic: string | undefined;
+  timestamp: Date | undefined;
+  rateLimitProof: RateLimitProof | undefined;
 }
 
 export interface Encoder {
   contentTopic: string;
-  encode: (message: Message) => Promise<Uint8Array | undefined>;
-  encodeProto: (message: Message) => Promise<ProtoMessage | undefined>;
+  encode: (message: Partial<Message>) => Promise<Uint8Array | undefined>;
+  encodeProto: (message: Partial<Message>) => Promise<ProtoMessage | undefined>;
 }
 
 export interface Decoder<T extends Message> {

src/lib/to_proto_message.spec.ts

@@ -0,0 +1,24 @@
+import { expect } from "chai";
+
+import { WakuMessage as WakuMessageProto } from "../proto/message";
+
+import { toProtoMessage } from "./to_proto_message";
+
+describe("to proto message", () => {
+  it("Fields are not dropped", () => {
+    const wire: WakuMessageProto = {
+      contentTopic: "foo",
+    };
+
+    const protoMessage = toProtoMessage(wire);
+
+    expect(protoMessage.contentTopic).to.eq("foo");
+
+    const keys = Object.keys(protoMessage);
+    expect(keys).to.contain("payload");
+    expect(keys).to.contain("contentTopic");
+    expect(keys).to.contain("version");
+    expect(keys).to.contain("timestamp");
+    expect(keys).to.contain("rateLimitProof");
+  });
+});

src/lib/to_proto_message.ts

@@ -0,0 +1,15 @@
+import { WakuMessage as WakuMessageProto } from "../proto/message";
+
+import { ProtoMessage } from "./interfaces";
+
+const EmptyMessage: ProtoMessage = {
+  payload: undefined,
+  contentTopic: undefined,
+  version: undefined,
+  timestamp: undefined,
+  rateLimitProof: undefined,
+};
+
+export function toProtoMessage(wire: WakuMessageProto): ProtoMessage {
+  return { ...EmptyMessage, ...wire };
+}

src/lib/waku_filter/index.ts

@@ -18,6 +18,7 @@ import {
   selectPeerForProtocol,
   selectRandomPeer,
 } from "../select_peer";
+import { toProtoMessage } from "../to_proto_message";
 
 import { ContentFilter, FilterRPC } from "./filter_rpc";
 export { ContentFilter };
@@ -198,7 +199,7 @@ export class WakuFilter {
       // noinspection ES6MissingAwait
       decoders.forEach(async (dec) => {
         if (msg) return;
-        const decoded = await dec.decode(protoMessage);
+        const decoded = await dec.decode(toProtoMessage(protoMessage));
         if (!decoded) {
           log("Not able to decode message");
           return;

src/lib/waku_light_push/index.ts

@@ -53,7 +53,7 @@ export class WakuLightPush {
 
   async push(
     encoder: Encoder,
-    message: Message,
+    message: Partial<Message>,
     opts?: PushOptions
   ): Promise<SendResult> {
     const pubSubTopic = opts?.pubSubTopic ? opts.pubSubTopic : this.pubSubTopic;

src/lib/waku_message/topic_only_message.ts

@@ -6,6 +6,10 @@ import type { Decoder, Message, ProtoMessage } from "../interfaces";
 const log = debug("waku:message:topic-only");
 
 export class TopicOnlyMessage implements Message {
+  public payload: undefined;
+  public rateLimitProof: undefined;
+  public timestamp: undefined;
+
   constructor(private proto: proto.TopicOnlyMessage) {}
 
   get contentTopic(): string {
@@ -19,7 +23,13 @@ export class TopicOnlyDecoder implements Decoder<TopicOnlyMessage> {
   decodeProto(bytes: Uint8Array): Promise<ProtoMessage | undefined> {
     const protoMessage = proto.TopicOnlyMessage.decode(bytes);
     log("Message decoded", protoMessage);
-    return Promise.resolve(protoMessage);
+    return Promise.resolve({
+      contentTopic: protoMessage.contentTopic,
+      payload: undefined,
+      rateLimitProof: undefined,
+      timestamp: undefined,
+      version: undefined,
+    });
   }
 
   async decode(proto: ProtoMessage): Promise<TopicOnlyMessage | undefined> {

src/lib/waku_message/version_0.ts

@@ -1,7 +1,7 @@
 import debug from "debug";
 
 import * as proto from "../../proto/message";
-import { Decoder, Message, ProtoMessage } from "../interfaces";
+import { Decoder, Message, ProtoMessage, RateLimitProof } from "../interfaces";
 import { Encoder } from "../interfaces";
 
 const log = debug("waku:message:version-0");
@@ -54,16 +54,20 @@ export class MessageV0 implements Message {
     // https://github.com/status-im/js-waku/issues/921
     return this.proto.version ?? 0;
   }
+
+  get rateLimitProof(): RateLimitProof | undefined {
+    return this.proto.rateLimitProof;
+  }
 }
 
 export class EncoderV0 implements Encoder {
   constructor(public contentTopic: string) {}
 
-  async encode(message: Message): Promise<Uint8Array> {
+  async encode(message: Partial<Message>): Promise<Uint8Array> {
     return proto.WakuMessage.encode(await this.encodeProto(message));
   }
 
-  async encodeProto(message: Message): Promise<ProtoMessage> {
+  async encodeProto(message: Partial<Message>): Promise<ProtoMessage> {
     const timestamp = message.timestamp ?? new Date();
 
     return {
@@ -71,6 +75,7 @@ export class EncoderV0 implements Encoder {
       version: Version,
       contentTopic: message.contentTopic ?? this.contentTopic,
       timestamp: BigInt(timestamp.valueOf()) * OneMillion,
+      rateLimitProof: message.rateLimitProof,
     };
   }
 }
@@ -81,7 +86,13 @@ export class DecoderV0 implements Decoder<MessageV0> {
   decodeProto(bytes: Uint8Array): Promise<ProtoMessage | undefined> {
     const protoMessage = proto.WakuMessage.decode(bytes);
     log("Message decoded", protoMessage);
-    return Promise.resolve(protoMessage);
+    return Promise.resolve({
+      payload: protoMessage.payload ?? undefined,
+      contentTopic: protoMessage.contentTopic ?? undefined,
+      version: protoMessage.version ?? undefined,
+      timestamp: protoMessage.timestamp ?? undefined,
+      rateLimitProof: protoMessage.rateLimitProof ?? undefined,
+    });
   }
 
   async decode(proto: ProtoMessage): Promise<MessageV0 | undefined> {

src/lib/waku_message/version_1.ts

@@ -52,14 +52,16 @@ export class AsymEncoder implements Encoder {
     private sigPrivKey?: Uint8Array
   ) {}
 
-  async encode(message: Message): Promise<Uint8Array | undefined> {
+  async encode(message: Partial<Message>): Promise<Uint8Array | undefined> {
     const protoMessage = await this.encodeProto(message);
     if (!protoMessage) return;
 
     return proto.WakuMessage.encode(protoMessage);
   }
 
-  async encodeProto(message: Message): Promise<ProtoMessage | undefined> {
+  async encodeProto(
+    message: Partial<Message>
+  ): Promise<ProtoMessage | undefined> {
     const timestamp = message.timestamp ?? new Date();
     if (!message.payload) {
       log("No payload to encrypt, skipping: ", message);
@@ -74,6 +76,7 @@ export class AsymEncoder implements Encoder {
       version: Version,
       contentTopic: this.contentTopic,
       timestamp: BigInt(timestamp.valueOf()) * OneMillion,
+      rateLimitProof: message.rateLimitProof,
     };
   }
 }
@@ -85,14 +88,16 @@ export class SymEncoder implements Encoder {
     private sigPrivKey?: Uint8Array
   ) {}
 
-  async encode(message: Message): Promise<Uint8Array | undefined> {
+  async encode(message: Partial<Message>): Promise<Uint8Array | undefined> {
     const protoMessage = await this.encodeProto(message);
     if (!protoMessage) return;
 
     return proto.WakuMessage.encode(protoMessage);
   }
 
-  async encodeProto(message: Message): Promise<ProtoMessage | undefined> {
+  async encodeProto(
+    message: Partial<Message>
+  ): Promise<ProtoMessage | undefined> {
     const timestamp = message.timestamp ?? new Date();
     if (!message.payload) {
       log("No payload to encrypt, skipping: ", message);
@@ -106,6 +111,7 @@ export class SymEncoder implements Encoder {
       version: Version,
       contentTopic: this.contentTopic,
       timestamp: BigInt(timestamp.valueOf()) * OneMillion,
+      rateLimitProof: message.rateLimitProof,
     };
   }
 }

src/lib/waku_relay/index.ts

@@ -92,7 +92,10 @@ export class WakuRelay extends GossipSub {
   /**
    * Send Waku message.
    */
-  public async send(encoder: Encoder, message: Message): Promise<SendResult> {
+  public async send(
+    encoder: Encoder,
+    message: Partial<Message>
+  ): Promise<SendResult> {
     const msg = await encoder.encode(message);
     if (!msg) {
       log("Failed to encode message, aborting publish");

src/lib/waku_store/index.ts

@@ -14,6 +14,7 @@ import { DefaultPubSubTopic, StoreCodecs } from "../constants";
 import { Decoder, Message } from "../interfaces";
 import { selectConnection } from "../select_connection";
 import { getPeersForProtocol, selectPeerForProtocol } from "../select_peer";
+import { toProtoMessage } from "../to_proto_message";
 
 import { HistoryRPC, PageDirection, Params } from "./history_rpc";
 
@@ -341,7 +342,7 @@ async function* paginate<T extends Message>(
       if (typeof contentTopic !== "undefined") {
         const decoder = decoders.get(contentTopic);
         if (decoder) {
-          return decoder.decode(protoMsg);
+          return decoder.decode(toProtoMessage(protoMsg));
         }
       }
       return Promise.resolve(undefined);