Merge pull request #975 from waku-org/proof-attached

CHANGELOG.md

@@ -11,6 +11,11 @@ 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.
+- `Decoder` and `Encoder` interfaces change to better express what the function members do.
+
 ## [0.29.0] - 2022-09-21
 
 ### Changed

src/lib/interfaces.ts

@@ -62,30 +62,30 @@ 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>;
+  toWire: (message: Partial<Message>) => Promise<Uint8Array | undefined>;
+  toProtoObj: (message: Partial<Message>) => Promise<ProtoMessage | undefined>;
 }
 
 export interface Decoder<T extends Message> {
   contentTopic: string;
-  decodeProto: (bytes: Uint8Array) => Promise<ProtoMessage | undefined>;
-  decode: (proto: ProtoMessage) => Promise<T | undefined>;
+  fromWireToProtoObj: (bytes: Uint8Array) => Promise<ProtoMessage | undefined>;
+  fromProtoObj: (proto: ProtoMessage) => Promise<T | undefined>;
 }
 
 export interface SendResult {

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.fromProtoObj(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;
@@ -79,7 +79,7 @@ export class WakuLightPush {
     const recipients: PeerId[] = [];
 
     try {
-      const protoMessage = await encoder.encodeProto(message);
+      const protoMessage = await encoder.toProtoObj(message);
       if (!protoMessage) {
         log("Failed to encode to protoMessage, aborting push");
         return { recipients };

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 {
@@ -16,13 +20,21 @@ export class TopicOnlyMessage implements Message {
 export class TopicOnlyDecoder implements Decoder<TopicOnlyMessage> {
   public contentTopic = "";
 
-  decodeProto(bytes: Uint8Array): Promise<ProtoMessage | undefined> {
+  fromWireToProtoObj(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> {
+  async fromProtoObj(
+    proto: ProtoMessage
+  ): Promise<TopicOnlyMessage | undefined> {
     return new TopicOnlyMessage(proto);
   }
 }

src/lib/waku_message/version_0.spec.ts

@@ -10,10 +10,10 @@ describe("Waku Message version 0", function () {
     await fc.assert(
       fc.asyncProperty(fc.uint8Array({ minLength: 1 }), async (payload) => {
         const encoder = new EncoderV0(TestContentTopic);
-        const bytes = await encoder.encode({ payload });
+        const bytes = await encoder.toWire({ payload });
         const decoder = new DecoderV0(TestContentTopic);
-        const protoResult = await decoder.decodeProto(bytes);
-        const result = (await decoder.decode(protoResult!)) as MessageV0;
+        const protoResult = await decoder.fromWireToProtoObj(bytes);
+        const result = (await decoder.fromProtoObj(protoResult!)) as MessageV0;
 
         expect(result.contentTopic).to.eq(TestContentTopic);
         expect(result.version).to.eq(0);

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> {
-    return proto.WakuMessage.encode(await this.encodeProto(message));
+  async toWire(message: Partial<Message>): Promise<Uint8Array> {
+    return proto.WakuMessage.encode(await this.toProtoObj(message));
   }
 
-  async encodeProto(message: Message): Promise<ProtoMessage> {
+  async toProtoObj(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,
     };
   }
 }
@@ -78,13 +83,19 @@ export class EncoderV0 implements Encoder {
 export class DecoderV0 implements Decoder<MessageV0> {
   constructor(public contentTopic: string) {}
 
-  decodeProto(bytes: Uint8Array): Promise<ProtoMessage | undefined> {
+  fromWireToProtoObj(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> {
+  async fromProtoObj(proto: ProtoMessage): Promise<MessageV0 | undefined> {
     // https://github.com/status-im/js-waku/issues/921
     if (proto.version === undefined) {
       proto.version = 0;

src/lib/waku_message/version_1.spec.ts

@@ -28,12 +28,12 @@ describe("Waku Message version 1", function () {
           const publicKey = getPublicKey(privateKey);
 
           const encoder = new AsymEncoder(TestContentTopic, publicKey);
-          const bytes = await encoder.encode({ payload });
+          const bytes = await encoder.toWire({ payload });
 
           const decoder = new AsymDecoder(TestContentTopic, privateKey);
-          const protoResult = await decoder.decodeProto(bytes!);
+          const protoResult = await decoder.fromWireToProtoObj(bytes!);
           if (!protoResult) throw "Failed to proto decode";
-          const result = await decoder.decode(protoResult);
+          const result = await decoder.fromProtoObj(protoResult);
           if (!result) throw "Failed to decode";
 
           expect(result.contentTopic).to.equal(TestContentTopic);
@@ -63,12 +63,12 @@ describe("Waku Message version 1", function () {
             bobPublicKey,
             alicePrivateKey
           );
-          const bytes = await encoder.encode({ payload });
+          const bytes = await encoder.toWire({ payload });
 
           const decoder = new AsymDecoder(TestContentTopic, bobPrivateKey);
-          const protoResult = await decoder.decodeProto(bytes!);
+          const protoResult = await decoder.fromWireToProtoObj(bytes!);
           if (!protoResult) throw "Failed to proto decode";
-          const result = await decoder.decode(protoResult);
+          const result = await decoder.fromProtoObj(protoResult);
           if (!result) throw "Failed to decode";
 
           expect(result.contentTopic).to.equal(TestContentTopic);
@@ -88,12 +88,12 @@ describe("Waku Message version 1", function () {
         fc.uint8Array({ min: 1, minLength: 32, maxLength: 32 }),
         async (payload, symKey) => {
           const encoder = new SymEncoder(TestContentTopic, symKey);
-          const bytes = await encoder.encode({ payload });
+          const bytes = await encoder.toWire({ payload });
 
           const decoder = new SymDecoder(TestContentTopic, symKey);
-          const protoResult = await decoder.decodeProto(bytes!);
+          const protoResult = await decoder.fromWireToProtoObj(bytes!);
           if (!protoResult) throw "Failed to proto decode";
-          const result = await decoder.decode(protoResult);
+          const result = await decoder.fromProtoObj(protoResult);
           if (!result) throw "Failed to decode";
 
           expect(result.contentTopic).to.equal(TestContentTopic);
@@ -116,12 +116,12 @@ describe("Waku Message version 1", function () {
           const sigPubKey = getPublicKey(sigPrivKey);
 
           const encoder = new SymEncoder(TestContentTopic, symKey, sigPrivKey);
-          const bytes = await encoder.encode({ payload });
+          const bytes = await encoder.toWire({ payload });
 
           const decoder = new SymDecoder(TestContentTopic, symKey);
-          const protoResult = await decoder.decodeProto(bytes!);
+          const protoResult = await decoder.fromWireToProtoObj(bytes!);
           if (!protoResult) throw "Failed to proto decode";
-          const result = await decoder.decode(protoResult);
+          const result = await decoder.fromProtoObj(protoResult);
           if (!result) throw "Failed to decode";
 
           expect(result.contentTopic).to.equal(TestContentTopic);

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> {
-    const protoMessage = await this.encodeProto(message);
+  async toWire(message: Partial<Message>): Promise<Uint8Array | undefined> {
+    const protoMessage = await this.toProtoObj(message);
     if (!protoMessage) return;
 
     return proto.WakuMessage.encode(protoMessage);
   }
 
-  async encodeProto(message: Message): Promise<ProtoMessage | undefined> {
+  async toProtoObj(
+    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> {
-    const protoMessage = await this.encodeProto(message);
+  async toWire(message: Partial<Message>): Promise<Uint8Array | undefined> {
+    const protoMessage = await this.toProtoObj(message);
     if (!protoMessage) return;
 
     return proto.WakuMessage.encode(protoMessage);
   }
 
-  async encodeProto(message: Message): Promise<ProtoMessage | undefined> {
+  async toProtoObj(
+    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,
     };
   }
 }
@@ -115,7 +121,9 @@ export class AsymDecoder extends DecoderV0 implements Decoder<MessageV1> {
     super(contentTopic);
   }
 
-  async decode(protoMessage: ProtoMessage): Promise<MessageV1 | undefined> {
+  async fromProtoObj(
+    protoMessage: ProtoMessage
+  ): Promise<MessageV1 | undefined> {
     const cipherPayload = protoMessage.payload;
 
     if (protoMessage.version !== Version) {
@@ -171,7 +179,9 @@ export class SymDecoder extends DecoderV0 implements Decoder<MessageV1> {
     super(contentTopic);
   }
 
-  async decode(protoMessage: ProtoMessage): Promise<MessageV1 | undefined> {
+  async fromProtoObj(
+    protoMessage: ProtoMessage
+  ): Promise<MessageV1 | undefined> {
     const cipherPayload = protoMessage.payload;
 
     if (protoMessage.version !== Version) {

src/lib/waku_relay/index.ts

@@ -92,8 +92,11 @@ export class WakuRelay extends GossipSub {
   /**
    * Send Waku message.
    */
-  public async send(encoder: Encoder, message: Message): Promise<SendResult> {
-    const msg = await encoder.encode(message);
+  public async send(
+    encoder: Encoder,
+    message: Partial<Message>
+  ): Promise<SendResult> {
+    const msg = await encoder.toWire(message);
     if (!msg) {
       log("Failed to encode message, aborting publish");
       return { recipients: [] };
@@ -136,7 +139,7 @@ export class WakuRelay extends GossipSub {
         if (event.detail.msg.topic !== pubSubTopic) return;
         log(`Message received on ${pubSubTopic}`);
 
-        const topicOnlyMsg = await this.defaultDecoder.decodeProto(
+        const topicOnlyMsg = await this.defaultDecoder.fromWireToProtoObj(
           event.detail.msg.data
         );
         if (!topicOnlyMsg || !topicOnlyMsg.contentTopic) {
@@ -150,14 +153,16 @@ export class WakuRelay extends GossipSub {
         }
         await Promise.all(
           Array.from(observers).map(async ({ decoder, callback }) => {
-            const protoMsg = await decoder.decodeProto(event.detail.msg.data);
+            const protoMsg = await decoder.fromWireToProtoObj(
+              event.detail.msg.data
+            );
             if (!protoMsg) {
               log(
                 "Internal error: message previously decoded failed on 2nd pass."
               );
               return;
             }
-            const msg = await decoder.decode(protoMsg);
+            const msg = await decoder.fromProtoObj(protoMsg);
             if (msg) {
               callback(msg);
             } else {

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.fromProtoObj(toProtoMessage(protoMsg));
         }
       }
       return Promise.resolve(undefined);