Merge pull request #1718 from waku-org/adklempner/autoshard-topic-hashing

feat: add function for determining shard index from content topic

Author: adklempner

package-lock.json

@@ -67,6 +67,7 @@
     "node": ">=18"
   },
   "dependencies": {
+    "@noble/hashes": "^1.3.2",
     "chai": "^4.3.8",
     "debug": "^4.3.4",
     "uint8arrays": "^4.0.4"

packages/utils/package.json

@@ -1,6 +1,6 @@
 import { expect } from "chai";
 
-import { ensureValidContentTopic } from "./sharding";
+import { contentTopicToShardIndex, ensureValidContentTopic } from "./sharding";
 
 const testInvalidCases = (
   contentTopics: string[],
@@ -86,3 +86,15 @@ describe("ensureValidContentTopic", () => {
     testInvalidCases(["/0/myapp/1/mytopic/"], "Encoding field cannot be empty");
   });
 });
+
+describe("contentTopicToShardIndex", () => {
+  it("converts content topics to expected shard index", () => {
+    const contentTopics: [string, number][] = [
+      ["/toychat/2/huilong/proto", 3],
+      ["/myapp/1/latest/proto", 0]
+    ];
+    for (const [topic, shard] of contentTopics) {
+      expect(contentTopicToShardIndex(topic)).to.eq(shard);
+    }
+  });
+});

packages/utils/src/common/sharding.spec.ts

@@ -1,5 +1,8 @@
+import { sha256 } from "@noble/hashes/sha256";
 import type { PubsubTopic, ShardInfo } from "@waku/interfaces";
 
+import { concat, utf8ToBytes } from "../bytes/index.js";
+
 export const shardInfoToPubsubTopics = (
   shardInfo: ShardInfo
 ): PubsubTopic[] => {
@@ -19,18 +22,28 @@ export function ensurePubsubTopicIsConfigured(
   }
 }
 
+interface ContentTopic {
+  generation: number;
+  application: string;
+  version: string;
+  topicName: string;
+  encoding: string;
+}
+
 /**
  * Given a string, will throw an error if it is not formatted as a valid content topic for autosharding based on https://rfc.vac.dev/spec/51/
  * @param contentTopic String to validate
+ * @returns Object with each content topic field as an attribute
  */
-export function ensureValidContentTopic(contentTopic: string): void {
+export function ensureValidContentTopic(contentTopic: string): ContentTopic {
   const parts = contentTopic.split("/");
   if (parts.length < 5 || parts.length > 6) {
     throw Error("Content topic format is invalid");
   }
   // Validate generation field if present
+  let generation = 0;
   if (parts.length == 6) {
-    const generation = parseInt(parts[1]);
+    generation = parseInt(parts[1]);
     if (isNaN(generation)) {
       throw new Error("Invalid generation field in content topic");
     }
@@ -56,4 +69,37 @@ export function ensureValidContentTopic(contentTopic: string): void {
   if (fields[3].length == 0) {
     throw new Error("Encoding field cannot be empty");
   }
+
+  return {
+    generation,
+    application: fields[0],
+    version: fields[1],
+    topicName: fields[2],
+    encoding: fields[3]
+  };
+}
+
+/**
+ * Given a string, determines which autoshard index to use for its pubsub topic.
+ * Based on the algorithm described in the RFC: https://rfc.vac.dev/spec/51//#algorithm
+ */
+export function contentTopicToShardIndex(
+  contentTopic: string,
+  networkShards: number = 8
+): number {
+  const { application, version } = ensureValidContentTopic(contentTopic);
+  const digest = sha256(
+    concat([utf8ToBytes(application), utf8ToBytes(version)])
+  );
+  const dataview = new DataView(digest.buffer.slice(-8));
+  return Number(dataview.getBigUint64(0, false) % BigInt(networkShards));
+}
+
+export function contentTopicToPubsubTopic(
+  contentTopic: string,
+  clusterId: number = 1,
+  networkShards: number = 8
+): string {
+  const shardIndex = contentTopicToShardIndex(contentTopic, networkShards);
+  return `/waku/2/rs/${clusterId}/${shardIndex}`;
 }