fix(store): fix validation to allow store query by hash

Author: adklempner

package-lock.json

@@ -14,6 +14,7 @@ export class StoreQueryRequest {
   public static create(params: QueryRequestParams): StoreQueryRequest {
     const request = new StoreQueryRequest({
       ...params,
+      contentTopics: params.contentTopics || [],
       requestId: uuid(),
       timeStart: params.timeStart
         ? BigInt(params.timeStart.getTime() * ONE_MILLION)
@@ -28,26 +29,26 @@ export class StoreQueryRequest {
     });
 
     // Validate request parameters based on RFC
-    if (
-      (params.pubsubTopic && !params.contentTopics) ||
-      (!params.pubsubTopic && params.contentTopics)
-    ) {
-      throw new Error(
-        "Both pubsubTopic and contentTopics must be set or unset"
-      );
-    }
+    // if (
+    //   (params.pubsubTopic && !params.contentTopics) ||
+    //   (!params.pubsubTopic && params.contentTopics)
+    // ) {
+    //   throw new Error(
+    //     "Both pubsubTopic and contentTopics must be set or unset"
+    //   );
+    // }
 
-    if (
-      params.messageHashes &&
-      (params.pubsubTopic ||
-        params.contentTopics ||
-        params.timeStart ||
-        params.timeEnd)
-    ) {
-      throw new Error(
-        "Message hash lookup queries cannot include content filter criteria"
-      );
-    }
+    // if (
+    //   params.messageHashes &&
+    //   (params.pubsubTopic ||
+    //     (params.contentTopics && params.contentTopics.length > 0) ||
+    //     params.timeStart ||
+    //     params.timeEnd)
+    // ) {
+    //   throw new Error(
+    //     "Message hash lookup queries cannot include content filter criteria"
+    //   );
+    // }
 
     return request;
   }

packages/core/src/lib/store/rpc.ts

@@ -41,8 +41,9 @@ export class StoreCore extends BaseProtocol implements IStoreCore {
     peerId: PeerId
   ): AsyncGenerator<Promise<T | undefined>[]> {
     if (
+      queryOpts.contentTopics &&
       queryOpts.contentTopics.toString() !==
-      Array.from(decoders.keys()).toString()
+        Array.from(decoders.keys()).toString()
     ) {
       throw new Error(
         "Internal error, the decoders should match the query's content topics"

packages/core/src/lib/store/store.ts

@@ -55,6 +55,7 @@
     "@waku/core": "*",
     "@waku/enr": "*",
     "@waku/interfaces": "*",
+    "@waku/message-hash": "^0.1.17",
     "@waku/utils": "*",
     "app-root-path": "^3.1.0",
     "chai-as-promised": "^7.1.1",

packages/tests/package.json

@@ -0,0 +1,68 @@
+import { DecodedMessage } from "@waku/core";
+import type { LightNode } from "@waku/interfaces";
+import { messageHash } from "@waku/message-hash";
+import { assert } from "chai";
+
+import {
+  afterEachCustom,
+  beforeEachCustom,
+  ServiceNode,
+  tearDownNodes
+} from "../../src/index.js";
+
+import {
+  runStoreNodes,
+  sendMessages,
+  TestDecoder,
+  TestShardInfo,
+  totalMsgs
+} from "./utils.js";
+
+describe("Waku Store, message hash query", function () {
+  this.timeout(15000);
+  let waku: LightNode;
+  let nwaku: ServiceNode;
+
+  beforeEachCustom(this, async () => {
+    [nwaku, waku] = await runStoreNodes(this.ctx, TestShardInfo);
+  });
+
+  afterEachCustom(this, async () => {
+    await tearDownNodes(nwaku, [waku]);
+  });
+
+  it("can query messages by message hash", async function () {
+    const sentMessages = await sendMessages(
+      nwaku,
+      totalMsgs,
+      TestDecoder.contentTopic,
+      TestDecoder.pubsubTopic,
+      true
+    );
+    const messageHashes = sentMessages.map((msg) =>
+      messageHash(TestDecoder.pubsubTopic, {
+        pubsubTopic: TestDecoder.pubsubTopic,
+        payload: Buffer.from(msg.payload, "base64"),
+        contentTopic: TestDecoder.contentTopic,
+        timestamp: msg.timestamp
+          ? new Date(Number(msg.timestamp / 1000000n))
+          : undefined,
+        meta: undefined,
+        rateLimitProof: undefined,
+        ephemeral: undefined
+      })
+    );
+    const messages: DecodedMessage[] = [];
+    for await (const page of waku.store.queryGenerator([TestDecoder], {
+      messageHashes
+    })) {
+      for await (const msg of page) {
+        messages.push(msg as DecodedMessage);
+      }
+    }
+    assert.equal(messages.length, messageHashes.length);
+    for (const msg of messages) {
+      assert.equal(msg.contentTopic, TestDecoder.contentTopic);
+    }
+  });
+});

packages/tests/tests/store/message_hash.spec.ts

@@ -17,6 +17,7 @@ import { expect } from "chai";
 import { Context } from "mocha";
 
 import { delay, NOISE_KEY_1, runNodes, ServiceNode } from "../../src/index.js";
+import { MessageRpcQuery } from "../../src/types.js";
 
 export const log = new Logger("test:store");
 
@@ -49,20 +50,20 @@ export async function sendMessages(
   instance: ServiceNode,
   numMessages: number,
   contentTopic: string,
-  pubsubTopic: string
-): Promise<void> {
+  pubsubTopic: string,
+  timestamp: boolean = false
+): Promise<MessageRpcQuery[]> {
+  const messages: MessageRpcQuery[] = new Array<MessageRpcQuery>(numMessages);
   for (let i = 0; i < numMessages; i++) {
-    expect(
-      await instance.sendMessage(
-        ServiceNode.toMessageRpcQuery({
-          payload: new Uint8Array([i]),
-          contentTopic: contentTopic
-        }),
-        pubsubTopic
-      )
-    ).to.eq(true);
+    messages[i] = ServiceNode.toMessageRpcQuery({
+      payload: new Uint8Array([i]),
+      contentTopic: contentTopic,
+      timestamp: timestamp ? new Date() : undefined
+    });
+    expect(await instance.sendMessage(messages[i], pubsubTopic)).to.eq(true);
     await delay(1); // to ensure each timestamp is unique.
   }
+  return messages;
 }
 
 export async function sendMessagesAutosharding(