Merge branch 'master' into danisharora/cursor-support-store

.github/workflows/ci.yml

@@ -9,7 +9,7 @@ on:
   pull_request:
 
 env:
-  NWAKU_VERSION: "v0.11"
+  NWAKU_VERSION: "v0.13.0"
   NODE_JS: "16"
 
 jobs:

nwaku

@@ -1 +1 @@
-Subproject commit fec13974836149c2b81cab8d4178dee7bfc56db1
+Subproject commit 9debd44e2aebf07eaf96b4525a4497c69aaf4445

packages/core/CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Waku Message `ephemeral` field to mark messages as do-not-store.
+
 ## @waku/core [0.0.5](https://github.com/waku-org/js-waku/compare/@waku/core@0.0.4...@waku/core@0.0.5) (2022-11-11)
 
 ### Changed

packages/core/src/lib/to_proto_message.spec.ts

@@ -20,5 +20,6 @@ describe("to proto message", () => {
     expect(keys).to.contain("version");
     expect(keys).to.contain("timestamp");
     expect(keys).to.contain("rateLimitProof");
+    expect(keys).to.contain("ephemeral");
   });
 });

packages/core/src/lib/to_proto_message.ts

@@ -8,6 +8,7 @@ const EmptyMessage: ProtoMessage = {
   version: undefined,
   timestamp: undefined,
   rateLimitProof: undefined,
+  ephemeral: undefined,
 };
 
 export function toProtoMessage(wire: WakuMessageProto): ProtoMessage {

packages/core/src/lib/waku.ts

@@ -13,8 +13,8 @@ import { FilterCodec, WakuFilter } from "./waku_filter";
 import { LightPushCodec, WakuLightPush } from "./waku_light_push";
 import { EncoderV0 } from "./waku_message/version_0";
 import { WakuRelay } from "./waku_relay";
-import { RelayCodecs, RelayPingContentTopic } from "./waku_relay/constants";
 import * as relayConstants from "./waku_relay/constants";
+import { RelayCodecs, RelayPingContentTopic } from "./waku_relay/constants";
 import { StoreCodec, WakuStore } from "./waku_store";
 
 export const DefaultPingKeepAliveValueSecs = 0;
@@ -109,13 +109,20 @@ export class WakuNode implements Waku {
    * Dials to the provided peer.
    *
    * @param peer The peer to dial
-   * @param protocols Waku protocols we expect from the peer; Default to Relay
+   * @param protocols Waku protocols we expect from the peer; Defaults to mounted protocols
    */
   async dial(
     peer: PeerId | Multiaddr,
     protocols?: Protocols[]
   ): Promise<Stream> {
-    const _protocols = protocols ?? [Protocols.Relay];
+    const _protocols = protocols ?? [];
+
+    if (typeof protocols === "undefined") {
+      this.relay && _protocols.push(Protocols.Relay);
+      this.store && _protocols.push(Protocols.Store);
+      this.filter && _protocols.push(Protocols.Filter);
+      this.lightPush && _protocols.push(Protocols.LightPush);
+    }
 
     const codecs: string[] = [];
     if (_protocols.includes(Protocols.Relay)) {
@@ -131,8 +138,6 @@ export class WakuNode implements Waku {
       codecs.push(FilterCodec);
     }
 
-    // eslint-disable-next-line @typescript-eslint/ban-ts-comment
-    // @ts-ignore: new Multiaddr is not backward compatible
     return this.libp2p.dialProtocol(peer, codecs);
   }
 

packages/core/src/lib/waku_message/topic_only_message.ts

@@ -1,14 +1,15 @@
-import type { Decoder, Message, ProtoMessage } from "@waku/interfaces";
+import type { DecodedMessage, Decoder, ProtoMessage } from "@waku/interfaces";
 import debug from "debug";
 
 import * as proto from "../../proto/topic_only_message";
 
 const log = debug("waku:message:topic-only");
 
-export class TopicOnlyMessage implements Message {
+export class TopicOnlyMessage implements DecodedMessage {
   public payload: undefined;
   public rateLimitProof: undefined;
   public timestamp: undefined;
+  public ephemeral: undefined;
 
   constructor(private proto: proto.TopicOnlyMessage) {}
 
@@ -29,6 +30,7 @@ export class TopicOnlyDecoder implements Decoder<TopicOnlyMessage> {
       rateLimitProof: undefined,
       timestamp: undefined,
       version: undefined,
+      ephemeral: undefined,
     });
   }
 

packages/core/src/lib/waku_message/version_0.spec.ts

@@ -17,6 +17,25 @@ describe("Waku Message version 0", function () {
 
         expect(result.contentTopic).to.eq(TestContentTopic);
         expect(result.version).to.eq(0);
+        expect(result.ephemeral).to.be.false;
+        expect(result.payload).to.deep.eq(payload);
+        expect(result.timestamp).to.not.be.undefined;
+      })
+    );
+  });
+
+  it("Ephemeral", async function () {
+    await fc.assert(
+      fc.asyncProperty(fc.uint8Array({ minLength: 1 }), async (payload) => {
+        const encoder = new EncoderV0(TestContentTopic, true);
+        const bytes = await encoder.toWire({ payload });
+        const decoder = new DecoderV0(TestContentTopic);
+        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);
+        expect(result.ephemeral).to.be.true;
         expect(result.payload).to.deep.eq(payload);
         expect(result.timestamp).to.not.be.undefined;
       })

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

@@ -26,6 +26,10 @@ export class MessageV0 implements DecodedMessage {
     return;
   }
 
+  get ephemeral(): boolean {
+    return Boolean(this.proto.ephemeral);
+  }
+
   get payload(): Uint8Array | undefined {
     return this._rawPayload;
   }
@@ -68,7 +72,7 @@ export class MessageV0 implements DecodedMessage {
 }
 
 export class EncoderV0 implements Encoder {
-  constructor(public contentTopic: string) {}
+  constructor(public contentTopic: string, public ephemeral: boolean = false) {}
 
   async toWire(message: Partial<Message>): Promise<Uint8Array> {
     return proto.WakuMessage.encode(await this.toProtoObj(message));
@@ -83,12 +87,13 @@ export class EncoderV0 implements Encoder {
       contentTopic: this.contentTopic,
       timestamp: BigInt(timestamp.valueOf()) * OneMillion,
       rateLimitProof: message.rateLimitProof,
+      ephemeral: this.ephemeral,
     };
   }
 }
 
 export class DecoderV0 implements Decoder<MessageV0> {
-  constructor(public contentTopic: string) {}
+  constructor(public contentTopic: string, public ephemeral: boolean = false) {}
 
   fromWireToProtoObj(bytes: Uint8Array): Promise<ProtoMessage | undefined> {
     const protoMessage = proto.WakuMessage.decode(bytes);
@@ -99,6 +104,7 @@ export class DecoderV0 implements Decoder<MessageV0> {
       version: protoMessage.version ?? undefined,
       timestamp: protoMessage.timestamp ?? undefined,
       rateLimitProof: protoMessage.rateLimitProof ?? undefined,
+      ephemeral: protoMessage.ephemeral ?? false,
     });
   }
 

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

@@ -233,8 +233,8 @@ export class WakuStore {
     );
 
     log("Querying history with the following options", {
-      peerId: options?.peerId?.toString(),
       ...options,
+      peerId: options?.peerId?.toString(),
     });
 
     const res = await selectPeerForProtocol(

packages/core/src/proto/filter.ts

@@ -501,6 +501,7 @@ export interface WakuMessage {
   timestampDeprecated?: number;
   timestamp?: bigint;
   rateLimitProof?: RateLimitProof;
+  ephemeral?: boolean;
 }
 
 export namespace WakuMessage {
@@ -544,6 +545,11 @@ export namespace WakuMessage {
             RateLimitProof.codec().encode(obj.rateLimitProof, writer);
           }
 
+          if (obj.ephemeral != null) {
+            writer.uint32(248);
+            writer.bool(obj.ephemeral);
+          }
+
           if (opts.lengthDelimited !== false) {
             writer.ldelim();
           }
@@ -578,6 +584,9 @@ export namespace WakuMessage {
                   reader.uint32()
                 );
                 break;
+              case 31:
+                obj.ephemeral = reader.bool();
+                break;
               default:
                 reader.skipType(tag & 7);
                 break;

packages/core/src/proto/light_push.ts

@@ -425,6 +425,7 @@ export interface WakuMessage {
   timestampDeprecated?: number;
   timestamp?: bigint;
   rateLimitProof?: RateLimitProof;
+  ephemeral?: boolean;
 }
 
 export namespace WakuMessage {
@@ -468,6 +469,11 @@ export namespace WakuMessage {
             RateLimitProof.codec().encode(obj.rateLimitProof, writer);
           }
 
+          if (obj.ephemeral != null) {
+            writer.uint32(248);
+            writer.bool(obj.ephemeral);
+          }
+
           if (opts.lengthDelimited !== false) {
             writer.ldelim();
           }
@@ -502,6 +508,9 @@ export namespace WakuMessage {
                   reader.uint32()
                 );
                 break;
+              case 31:
+                obj.ephemeral = reader.bool();
+                break;
               default:
                 reader.skipType(tag & 7);
                 break;

packages/core/src/proto/message.proto

@@ -17,5 +17,6 @@ message WakuMessage {
   optional double timestamp_deprecated  = 4;
   optional sint64 timestamp  = 10;
   optional RateLimitProof rate_limit_proof = 21;
+  optional bool ephemeral = 31;
 }
 

packages/core/src/proto/message.ts

@@ -203,6 +203,7 @@ export interface WakuMessage {
   timestampDeprecated?: number;
   timestamp?: bigint;
   rateLimitProof?: RateLimitProof;
+  ephemeral?: boolean;
 }
 
 export namespace WakuMessage {
@@ -246,6 +247,11 @@ export namespace WakuMessage {
             RateLimitProof.codec().encode(obj.rateLimitProof, writer);
           }
 
+          if (obj.ephemeral != null) {
+            writer.uint32(248);
+            writer.bool(obj.ephemeral);
+          }
+
           if (opts.lengthDelimited !== false) {
             writer.ldelim();
           }
@@ -280,6 +286,9 @@ export namespace WakuMessage {
                   reader.uint32()
                 );
                 break;
+              case 31:
+                obj.ephemeral = reader.bool();
+                break;
               default:
                 reader.skipType(tag & 7);
                 break;

packages/core/src/proto/store.ts

@@ -743,6 +743,7 @@ export interface WakuMessage {
   timestampDeprecated?: number;
   timestamp?: bigint;
   rateLimitProof?: RateLimitProof;
+  ephemeral?: boolean;
 }
 
 export namespace WakuMessage {
@@ -786,6 +787,11 @@ export namespace WakuMessage {
             RateLimitProof.codec().encode(obj.rateLimitProof, writer);
           }
 
+          if (obj.ephemeral != null) {
+            writer.uint32(248);
+            writer.bool(obj.ephemeral);
+          }
+
           if (opts.lengthDelimited !== false) {
             writer.ldelim();
           }
@@ -820,6 +826,9 @@ export namespace WakuMessage {
                   reader.uint32()
                 );
                 break;
+              case 31:
+                obj.ephemeral = reader.bool();
+                break;
               default:
                 reader.skipType(tag & 7);
                 break;

packages/interfaces/src/index.ts

@@ -175,6 +175,7 @@ export interface ProtoMessage {
   version: number | undefined;
   timestamp: bigint | undefined;
   rateLimitProof: RateLimitProof | undefined;
+  ephemeral: boolean | undefined;
 }
 
 /**
@@ -188,6 +189,7 @@ export interface Message {
 
 export interface Encoder {
   contentTopic: string;
+  ephemeral: boolean;
   toWire: (message: Message) => Promise<Uint8Array | undefined>;
   toProtoObj: (message: Message) => Promise<ProtoMessage | undefined>;
 }
@@ -197,6 +199,7 @@ export interface DecodedMessage {
   contentTopic: string | undefined;
   timestamp: Date | undefined;
   rateLimitProof: RateLimitProof | undefined;
+  ephemeral: boolean | undefined;
 }
 
 export interface Decoder<T extends DecodedMessage> {

packages/message-encryption/src/index.ts

@@ -68,7 +68,8 @@ export class AsymEncoder implements Encoder {
   constructor(
     public contentTopic: string,
     private publicKey: Uint8Array,
-    private sigPrivKey?: Uint8Array
+    private sigPrivKey?: Uint8Array,
+    public ephemeral: boolean = false
   ) {}
 
   async toWire(message: Partial<Message>): Promise<Uint8Array | undefined> {
@@ -96,6 +97,7 @@ export class AsymEncoder implements Encoder {
       contentTopic: this.contentTopic,
       timestamp: BigInt(timestamp.valueOf()) * OneMillion,
       rateLimitProof: message.rateLimitProof,
+      ephemeral: this.ephemeral,
     };
   }
 }
@@ -104,7 +106,8 @@ export class SymEncoder implements Encoder {
   constructor(
     public contentTopic: string,
     private symKey: Uint8Array,
-    private sigPrivKey?: Uint8Array
+    private sigPrivKey?: Uint8Array,
+    public ephemeral: boolean = false
   ) {}
 
   async toWire(message: Partial<Message>): Promise<Uint8Array | undefined> {
@@ -131,6 +134,7 @@ export class SymEncoder implements Encoder {
       contentTopic: this.contentTopic,
       timestamp: BigInt(timestamp.valueOf()) * OneMillion,
       rateLimitProof: message.rateLimitProof,
+      ephemeral: this.ephemeral,
     };
   }
 }

packages/tests/package.json

@@ -48,7 +48,7 @@
     "check:spelling": "cspell \"{README.md,{tests,src}/**/*.ts}\"",
     "check:tsc": "tsc -p tsconfig.dev.json",
     "test": "run-s test:*",
-    "test:node": "TS_NODE_PROJECT=./tsconfig.json mocha",
+    "test:node": "TS_NODE_PROJECT=./tsconfig.dev.json mocha",
     "reset-hard": "git clean -dfx -e .idea && git reset --hard && npm i && npm run build"
   },
   "engines": {

packages/tests/src/nwaku.ts

@@ -44,10 +44,10 @@ export interface Args {
   nodekey?: string;
   portsShift?: number;
   logLevel?: LogLevel;
-  persistMessages?: boolean;
   lightpush?: boolean;
   filter?: boolean;
   store?: boolean;
+  storeMessageDbUrl?: string;
   topics?: string;
   rpcPrivate?: boolean;
   websocketSupport?: boolean;
@@ -407,7 +407,7 @@ export class Nwaku {
       headers: new Headers({ "Content-Type": "application/json" }),
     });
     const json = await res.json();
-    log(`RPC Response: `, res, JSON.stringify(json));
+    log(`RPC Response: `, JSON.stringify(json));
     return json.result;
   }
 
@@ -423,7 +423,7 @@ export function argsToArray(args: Args): Array<string> {
 
   for (const [key, value] of Object.entries(args)) {
     // Change the key from camelCase to kebab-case
-    const kebabKey = key.replace(/([A-Z])/, (_, capital) => {
+    const kebabKey = key.replace(/([A-Z])/g, (_, capital) => {
       return "-" + capital.toLowerCase();
     });
 
@@ -442,6 +442,7 @@ export function defaultArgs(): Args {
     rpc: true,
     rpcAdmin: true,
     websocketSupport: true,
+    storeMessageDbUrl: "sqlite://:memory:",
     logLevel: LogLevel.Debug,
   };
 }

packages/tests/tests/nwaku.node.spec.ts

@@ -16,6 +16,7 @@ describe("nwaku", () => {
       "--rpc=true",
       "--rpc-admin=true",
       "--websocket-support=true",
+      "--store-message-db-url=sqlite://:memory:",
       "--log-level=DEBUG",
       "--ports-shift=42",
     ];

packages/tests/tests/store.node.spec.ts

@@ -3,7 +3,7 @@ import { createCursor, PageDirection } from "@waku/core";
 import { waitForRemotePeer } from "@waku/core/lib/wait_for_remote_peer";
 import { DecoderV0, EncoderV0 } from "@waku/core/lib/waku_message/version_0";
 import { createFullNode } from "@waku/create";
-import type { Message, WakuFull } from "@waku/interfaces";
+import { DecodedMessage, Message, WakuFull } from "@waku/interfaces";
 import { Protocols } from "@waku/interfaces";
 import {
   AsymDecoder,
@@ -32,7 +32,7 @@ describe("Waku Store", () => {
   beforeEach(async function () {
     this.timeout(15_000);
     nwaku = new Nwaku(makeLogFileName(this));
-    await nwaku.start({ persistMessages: true, store: true, lightpush: true });
+    await nwaku.start({ store: true, lightpush: true });
   });
 
   afterEach(async function () {
@@ -48,7 +48,7 @@ describe("Waku Store", () => {
       expect(
         await nwaku.sendMessage(
           Nwaku.toMessageRpcQuery({
-            payload: utf8ToBytes(`Message ${i}`),
+            payload: new Uint8Array([i]),
             contentTopic: TestContentTopic,
           })
         )
@@ -78,7 +78,7 @@ describe("Waku Store", () => {
 
     expect(messages?.length).eq(totalMsgs);
     const result = messages?.findIndex((msg) => {
-      return bytesToUtf8(msg.payload!) === "Message 0";
+      return msg.payload![0]! === 0;
     });
     expect(result).to.not.eq(-1);
   });
@@ -180,7 +180,7 @@ describe("Waku Store", () => {
       expect(
         await nwaku.sendMessage(
           Nwaku.toMessageRpcQuery({
-            payload: utf8ToBytes(`Message ${i}`),
+            payload: new Uint8Array([i]),
             contentTopic: TestContentTopic,
           })
         )
@@ -207,7 +207,7 @@ describe("Waku Store", () => {
 
     expect(messages?.length).eq(totalMsgs);
     const result = messages?.findIndex((msg) => {
-      return bytesToUtf8(msg.payload!) === "Message 0";
+      return msg.payload![0]! === 0;
     });
     expect(result).to.not.eq(-1);
   });
@@ -221,7 +221,7 @@ describe("Waku Store", () => {
       expect(
         await nwaku.sendMessage(
           Nwaku.toMessageRpcQuery({
-            payload: utf8ToBytes(`Message ${i}`),
+            payload: new Uint8Array([i]),
             contentTopic: TestContentTopic,
           })
         )
@@ -260,7 +260,7 @@ describe("Waku Store", () => {
       expect(
         await nwaku.sendMessage(
           Nwaku.toMessageRpcQuery({
-            payload: utf8ToBytes(`Message ${i}`),
+            payload: new Uint8Array([i]),
             contentTopic: TestContentTopic,
           })
         )
@@ -286,13 +286,8 @@ describe("Waku Store", () => {
     );
 
     expect(messages?.length).eq(totalMsgs);
-    for (let index = 0; index < totalMsgs; index++) {
+    const payloads = messages.map((msg) => msg.payload![0]!);
-      expect(
+    expect(payloads).to.deep.eq(Array.from(Array(totalMsgs).keys()));
-        messages?.findIndex((msg) => {
-          return bytesToUtf8(msg.payload!) === `Message ${index}`;
-        })
-      ).to.eq(index);
-    }
   });
 
   it("Ordered Callback - Backward", async function () {
@@ -303,7 +298,7 @@ describe("Waku Store", () => {
       expect(
         await nwaku.sendMessage(
           Nwaku.toMessageRpcQuery({
-            payload: utf8ToBytes(`Message ${i}`),
+            payload: new Uint8Array([i]),
             contentTopic: TestContentTopic,
           })
         )
@@ -331,13 +326,8 @@ describe("Waku Store", () => {
     messages = messages.reverse();
 
     expect(messages?.length).eq(totalMsgs);
-    for (let index = 0; index < totalMsgs; index++) {
+    const payloads = messages.map((msg) => msg.payload![0]!);
-      expect(
+    expect(payloads).to.deep.eq(Array.from(Array(totalMsgs).keys()));
-        messages?.findIndex((msg) => {
-          return bytesToUtf8(msg.payload!) === `Message ${index}`;
-        })
-      ).to.eq(index);
-    }
   });
 
   it("Generator, with asymmetric & symmetric encrypted messages", async function () {
@@ -414,7 +404,7 @@ describe("Waku Store", () => {
 
     await waitForRemotePeer(waku2, [Protocols.Store]);
 
-    const messages: Message[] = [];
+    const messages: DecodedMessage[] = [];
     log("Retrieve messages from store");
 
     for await (const msgPromises of waku2.store.queryGenerator([
@@ -440,33 +430,170 @@ describe("Waku Store", () => {
     !!waku2 && waku2.stop().catch((e) => console.log("Waku failed to stop", e));
   });
 
+  it.skip("Ephemeral support", async function () {
+    this.timeout(15_000);
+
+    const asymText = "This message is encrypted for me using asymmetric";
+    const asymTopic = "/test/1/asymmetric/proto";
+
+    const symText =
+      "This message is encrypted for me using symmetric encryption";
+    const symTopic = "/test/1/symmetric/proto";
+
+    const clearText = "This is a clear text message for everyone to read";
+
+    const storeReadableText = "This message is readable by the store";
+    const storeUnreadableText = "This message is not readable by the store";
+
+    const timestamp = new Date();
+
+    const asymMsg = { payload: utf8ToBytes(asymText), timestamp };
+    const symMsg = {
+      payload: utf8ToBytes(symText),
+      timestamp: new Date(timestamp.valueOf() + 1),
+    };
+    const clearMsg = {
+      payload: utf8ToBytes(clearText),
+      timestamp: new Date(timestamp.valueOf() + 2),
+    };
+
+    const storeReadableMsg = {
+      payload: utf8ToBytes(storeReadableText),
+    };
+    const storeUnreadableMsg = {
+      payload: utf8ToBytes(storeUnreadableText),
+    };
+
+    const privateKey = generatePrivateKey();
+    const symKey = generateSymmetricKey();
+    const publicKey = getPublicKey(privateKey);
+
+    const storeWithAsymEncoder = new AsymEncoder(
+      asymTopic,
+      publicKey,
+      undefined,
+      false
+    );
+    const storeWithSymEncoder = new SymEncoder(
+      symTopic,
+      symKey,
+      undefined,
+      false
+    );
+
+    const dontStoreWithAsymEncoder = new AsymEncoder(
+      asymTopic,
+      publicKey,
+      undefined,
+      true
+    );
+    const dontStoreWithSymEncoder = new SymEncoder(
+      symTopic,
+      symKey,
+      undefined,
+      true
+    );
+
+    const storeEncoder = new EncoderV0(TestContentTopic, false);
+    const storeUnreadableEncoder = new EncoderV0(TestContentTopic, true);
+
+    const asymDecoder = new AsymDecoder(asymTopic, privateKey);
+    const symDecoder = new SymDecoder(symTopic, symKey);
+
+    const [waku1, waku2, nimWakuMultiaddr] = await Promise.all([
+      createFullNode({
+        staticNoiseKey: NOISE_KEY_1,
+      }).then((waku) => waku.start().then(() => waku)),
+      createFullNode({
+        staticNoiseKey: NOISE_KEY_2,
+      }).then((waku) => waku.start().then(() => waku)),
+      nwaku.getMultiaddrWithId(),
+    ]);
+
+    log("Waku nodes created");
+
+    await Promise.all([
+      waku1.dial(nimWakuMultiaddr),
+      waku2.dial(nimWakuMultiaddr),
+    ]);
+
+    log("Waku nodes connected to nwaku");
+
+    await waitForRemotePeer(waku1, [Protocols.LightPush]);
+
+    log("Sending messages using light push");
+    await Promise.all([
+      waku1.lightPush.push(storeWithAsymEncoder, asymMsg),
+      waku1.lightPush.push(storeWithSymEncoder, symMsg),
+      waku1.lightPush.push(dontStoreWithAsymEncoder, asymMsg),
+      waku1.lightPush.push(dontStoreWithSymEncoder, symMsg),
+      waku1.lightPush.push(TestEncoder, clearMsg),
+      waku1.lightPush.push(storeEncoder, storeReadableMsg),
+      waku1.lightPush.push(storeUnreadableEncoder, storeUnreadableMsg),
+    ]);
+
+    await waitForRemotePeer(waku2, [Protocols.Store]);
+
+    const messages: DecodedMessage[] = [];
+    log("Retrieve messages from store");
+
+    for await (const msgPromises of waku2.store.queryGenerator([
+      asymDecoder,
+      symDecoder,
+      TestDecoder,
+    ])) {
+      for (const promise of msgPromises) {
+        const msg = await promise;
+        if (msg) {
+          messages.push(msg);
+        }
+      }
+    }
+
+    // Messages are ordered from oldest to latest within a page (1 page query)
+    expect(bytesToUtf8(messages[0].payload!)).to.eq(asymText);
+    expect(bytesToUtf8(messages[1].payload!)).to.eq(symText);
+    expect(bytesToUtf8(messages[2].payload!)).to.eq(clearText);
+    expect(bytesToUtf8(messages[3].payload!)).to.eq(storeReadableText);
+    expect(messages?.length).eq(4);
+
+    // check for ephemeral
+    expect(messages[0].ephemeral).to.be.false;
+    expect(messages[1].ephemeral).to.be.false;
+    expect(messages[2].ephemeral).to.be.false;
+    expect(messages[3].ephemeral).to.be.false;
+
+    !!waku1 && waku1.stop().catch((e) => console.log("Waku failed to stop", e));
+    !!waku2 && waku2.stop().catch((e) => console.log("Waku failed to stop", e));
+  });
+
   it("Ordered callback, using start and end time", async function () {
     this.timeout(20000);
 
     const now = new Date();
 
     const startTime = new Date();
-    // Set start time 5 minutes in the past
+    // Set start time 15 seconds in the past
-    startTime.setTime(now.getTime() - 5 * 60 * 1000);
+    startTime.setTime(now.getTime() - 15 * 1000);
 
     const message1Timestamp = new Date();
-    // Set first message was 4 minutes in the past
+    // Set first message was 10 seconds in the past
-    message1Timestamp.setTime(now.getTime() - 4 * 60 * 1000);
+    message1Timestamp.setTime(now.getTime() - 10 * 1000);
 
     const message2Timestamp = new Date();
-    // Set second message 2 minutes in the past
+    // Set second message 2 seconds in the past
-    message2Timestamp.setTime(now.getTime() - 2 * 60 * 1000);
+    message2Timestamp.setTime(now.getTime() - 2 * 1000);
     const messageTimestamps = [message1Timestamp, message2Timestamp];
 
     const endTime = new Date();
-    // Set end time 1 minute in the past
+    // Set end time 1 second in the past
-    endTime.setTime(now.getTime() - 60 * 1000);
+    endTime.setTime(now.getTime() - 1000);
 
     for (let i = 0; i < 2; i++) {
       expect(
         await nwaku.sendMessage(
           Nwaku.toMessageRpcQuery({
-            payload: utf8ToBytes(`Message ${i}`),
+            payload: new Uint8Array([i]),
             contentTopic: TestContentTopic,
             timestamp: messageTimestamps[i],
           })
@@ -514,7 +641,7 @@ describe("Waku Store", () => {
 
     expect(firstMessages?.length).eq(1);
 
-    expect(bytesToUtf8(firstMessages[0].payload!)).eq("Message 0");
+    expect(firstMessages[0].payload![0]!).eq(0);
 
     expect(bothMessages?.length).eq(2);
   });
@@ -528,7 +655,7 @@ describe("Waku Store", () => {
       expect(
         await nwaku.sendMessage(
           Nwaku.toMessageRpcQuery({
-            payload: utf8ToBytes(`Message ${i}`),
+            payload: new Uint8Array([i]),
             contentTopic: TestContentTopic,
           })
         )
@@ -566,7 +693,6 @@ describe("Waku Store, custom pubsub topic", () => {
     this.timeout(15_000);
     nwaku = new Nwaku(makeLogFileName(this));
     await nwaku.start({
-      persistMessages: true,
       store: true,
       topics: customPubSubTopic,
     });
@@ -585,7 +711,7 @@ describe("Waku Store, custom pubsub topic", () => {
       expect(
         await nwaku.sendMessage(
           Nwaku.toMessageRpcQuery({
-            payload: utf8ToBytes(`Message ${i}`),
+            payload: new Uint8Array([i]),
             contentTopic: TestContentTopic,
           }),
           customPubSubTopic
@@ -617,7 +743,7 @@ describe("Waku Store, custom pubsub topic", () => {
 
     expect(messages?.length).eq(totalMsgs);
     const result = messages?.findIndex((msg) => {
-      return bytesToUtf8(msg.payload!) === "Message 0";
+      return msg.payload![0]! === 0;
     });
     expect(result).to.not.eq(-1);
   });

packages/tests/tests/wait_for_remote_peer.node.spec.ts

@@ -101,7 +101,6 @@ describe("Wait for remote peer", function () {
       relay: false,
       lightpush: false,
       filter: false,
-      persistMessages: true,
     });
     const multiAddrWithId = await nwaku.getMultiaddrWithId();
 
@@ -128,7 +127,6 @@ describe("Wait for remote peer", function () {
       relay: false,
       lightpush: false,
       filter: false,
-      persistMessages: true,
     });
     const multiAddrWithId = await nwaku.getMultiaddrWithId();
 
@@ -213,7 +211,6 @@ describe("Wait for remote peer", function () {
       lightpush: true,
       relay: false,
       store: true,
-      persistMessages: true,
     });
     const multiAddrWithId = await nwaku.getMultiaddrWithId();
 

packages/tests/tests/waku.node.spec.ts

@@ -38,7 +38,6 @@ describe("Waku Dial [node only]", function () {
         filter: true,
         store: true,
         lightpush: true,
-        persistMessages: true,
       });
       const multiAddrWithId = await nwaku.getMultiaddrWithId();