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)
@@ -27,20 +28,28 @@ export class StoreQueryRequest {
         : undefined
     });
 
-    // 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"
-      );
+    // Validate that when not using messageHashes, pubsubTopic and contentTopics are consistent
+    if (!params.messageHashes || params.messageHashes.length === 0) {
+      // If we're doing a content-based query (not message hash), both pubsubTopic and contentTopics
+      // should be set or unset together (as per the RFC)
+      if (
+        (params.pubsubTopic && !params.contentTopics) ||
+        (!params.pubsubTopic &&
+          params.contentTopics &&
+          params.contentTopics.length > 0)
+      ) {
+        throw new Error(
+          "Both pubsubTopic and contentTopics must be set or unset for content-based queries"
+        );
+      }
     }
 
+    // If messageHashes is provided, no content filters should be included
     if (
       params.messageHashes &&
+      params.messageHashes.length > 0 &&
       (params.pubsubTopic ||
-        params.contentTopics ||
+        (params.contentTopics && params.contentTopics.length > 0) ||
         params.timeStart ||
         params.timeEnd)
     ) {

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

@@ -58,16 +58,40 @@ export class Store implements IStore {
     decoders: IDecoder<T>[],
     options?: Partial<QueryRequestParams>
   ): AsyncGenerator<Promise<T | undefined>[]> {
-    const { pubsubTopic, contentTopics, decodersAsMap } =
-      this.validateDecodersAndPubsubTopic(decoders);
-
-    const queryOpts = {
-      pubsubTopic,
-      contentTopics,
-      includeData: true,
-      paginationForward: true,
-      ...options
-    };
+    let queryOpts: QueryRequestParams;
+    let decodersAsMap: Map<string, IDecoder<T>> = new Map();
+    if (options?.messageHashes) {
+      // For message hash queries, we need a dummy pubsubTopic to satisfy the type
+      // This value will not be used for filtering since messageHashes take precedence
+      const dummyPubsubTopic =
+        options.pubsubTopic ||
+        this.protocol.pubsubTopics[0] ||
+        "/waku/2/default-waku/proto";
+
+      queryOpts = {
+        ...options,
+        pubsubTopic: dummyPubsubTopic,
+        contentTopics: [],
+        includeData: true,
+        paginationForward: true
+      };
+    } else {
+      const {
+        pubsubTopic,
+        contentTopics,
+        decodersAsMap: _decodersAsMap
+      } = this.validateDecodersAndPubsubTopic(decoders);
+
+      decodersAsMap = _decodersAsMap;
+
+      queryOpts = {
+        pubsubTopic,
+        contentTopics,
+        includeData: true,
+        paginationForward: true,
+        ...options
+      };
+    }
 
     const peer = await this.getPeerToUse();
 

packages/sdk/src/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,73 @@
+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 () {
+    // Send messages first
+    await sendMessages(
+      nwaku,
+      totalMsgs,
+      TestDecoder.contentTopic,
+      TestDecoder.pubsubTopic
+    );
+
+    // Generate message hashes for the test
+    const messageHashes: Uint8Array[] = [];
+
+    // Create message hashes for all numbers from 0 to totalMsgs-1, matching the payload pattern in sendMessages
+    for (let i = 0; i < totalMsgs; i++) {
+      // Using type assertion to handle type mismatch
+      messageHashes.push(
+        messageHash(TestDecoder.pubsubTopic, {
+          payload: new Uint8Array([i]) as any,
+          contentTopic: TestDecoder.contentTopic,
+          version: undefined,
+          timestamp: undefined,
+          meta: undefined,
+          rateLimitProof: undefined,
+          ephemeral: undefined
+        })
+      );
+    }
+
+    // Query messages by hash
+    const messages: DecodedMessage[] = [];
+    for await (const page of waku.store.queryGenerator([], {
+      messageHashes
+    })) {
+      for await (const msg of page) {
+        messages.push(msg as DecodedMessage);
+      }
+    }
+    assert.equal(messages.length, messageHashes.length);
+  });
+});

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(