Implement Waku Store 2.0.0-beta4 (#566)

Author: D4nte

CHANGELOG.md

@@ -10,7 +10,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Changed
 
 - Replaced `rlp` dependency with `@ethersproject/rlp`.
-- **breaking**: `staticNoiseKey` changed from `Buffer` to `Uint8Array`.
+- **Breaking**: `staticNoiseKey` changed from `Buffer` to `Uint8Array`.
+- Implement Waku Store 2.0.0-beta4. The `WakuMessage.timestamp` field now stores nanoseconds over the wire.
+- **Breaking**: `HistoryRPC.createQuery` takes `Date` instead of `number` for `startTime` and `endTime`.
 
 ### Removed
 

buf.gen.yaml

@@ -3,4 +3,5 @@ version: v1beta1
 plugins:
   - name: ts_proto
     out: ./src/proto
-    opt: grpc_js,esModuleInterop=true
+    strategy: all
+    opt: grpc_js,esModuleInterop=true,forceLong=long

package.json

@@ -43,7 +43,7 @@
     "examples:test": "run-s examples:pretest; for d in examples/*/; do (cd $d && npm test;); done",
     "proto": "run-s proto:*",
     "proto:lint": "buf lint",
-    "proto:build": "buf generate",
+    "proto:build": "rimraf ./src/proto && buf generate",
     "watch:build": "tsc -p tsconfig.json -w",
     "watch:test": "nyc --silent mocha --watch",
     "doc": "run-s doc:*",

proto/waku/v2/message.proto

@@ -6,5 +6,6 @@ message WakuMessage {
   optional bytes payload = 1;
   optional string content_topic = 2;
   optional uint32 version = 3;
-  optional double timestamp  = 4;
+  optional double timestamp_deprecated  = 4;
+  optional sint64 timestamp  = 10;
 }

proto/waku/v2/store.proto renamed to proto/waku/v2/store/v2beta3/store.proto

@@ -1,6 +1,6 @@
 syntax = "proto3";
 
-package waku.v2;
+package waku.v2.store.v2beta3;
 
 import "waku/v2/message.proto";
 

proto/waku/v2/store/v2beta4/store.proto

@@ -0,0 +1,49 @@
+syntax = "proto3";
+
+package waku.v2.store.v2beta4;
+
+import "waku/v2/message.proto";
+
+message Index {
+  bytes digest = 1;
+  sint64 received_time = 2;
+  sint64 sender_time = 3;
+}
+
+message PagingInfo {
+  uint64 page_size = 1;
+  Index cursor = 2;
+  enum Direction {
+    DIRECTION_BACKWARD_UNSPECIFIED = 0;
+    DIRECTION_FORWARD = 1;
+  }
+  Direction direction = 3;
+}
+
+message ContentFilter {
+  string content_topic = 1;
+}
+
+message HistoryQuery {
+  optional string pub_sub_topic = 2;
+  repeated ContentFilter content_filters = 3;
+  optional PagingInfo paging_info = 4;
+  optional sint64 start_time = 5;
+  optional sint64 end_time = 6;
+}
+
+message HistoryResponse {
+  repeated WakuMessage messages = 2;
+  PagingInfo paging_info = 3;
+  enum Error {
+    ERROR_NONE_UNSPECIFIED = 0;
+    ERROR_INVALID_CURSOR = 1;
+  }
+  Error error = 4;
+}
+
+message HistoryRPC {
+  string request_id = 1;
+  HistoryQuery query = 2;
+  HistoryResponse response = 3;
+}

src/index.ts

@@ -30,6 +30,4 @@ export * as waku_relay from "./lib/waku_relay";
 export { WakuRelay, RelayCodecs } from "./lib/waku_relay";
 
 export * as waku_store from "./lib/waku_store";
-export { PageDirection, WakuStore, StoreCodec } from "./lib/waku_store";
-
-export * as proto from "./proto";
+export { PageDirection, WakuStore, StoreCodecs } from "./lib/waku_store";

src/lib/select_peer.ts

@@ -24,10 +24,17 @@ export async function selectRandomPeer(
  */
 export async function* getPeersForProtocol(
   libp2p: Libp2p,
-  protocol: string
+  protocols: string[]
 ): AsyncIterable<Peer> {
   for await (const peer of libp2p.peerStore.getPeers()) {
-    if (!peer.protocols.includes(protocol)) {
+    let peerFound = false;
+    for (let i = 0; i < protocols.length; i++) {
+      if (peer.protocols.includes(protocols[i])) {
+        peerFound = true;
+        break;
+      }
+    }
+    if (!peerFound) {
       continue;
     }
     yield peer;

src/lib/waku.ts

@@ -21,7 +21,7 @@ import { LightPushCodec, WakuLightPush } from "./waku_light_push";
 import { DecryptionMethod, WakuMessage } from "./waku_message";
 import { RelayCodecs, WakuRelay } from "./waku_relay";
 import { RelayPingContentTopic } from "./waku_relay/constants";
-import { StoreCodec, WakuStore } from "./waku_store";
+import { StoreCodecs, WakuStore } from "./waku_store";
 
 const websocketsTransportKey = Websockets.prototype[Symbol.toStringTag];
 
@@ -246,7 +246,9 @@ export class Waku {
       RelayCodecs.forEach((codec) => codecs.push(codec));
     }
     if (_protocols.includes(Protocols.Store)) {
-      codecs.push(StoreCodec);
+      for (const codec of Object.values(StoreCodecs)) {
+        codecs.push(codec);
+      }
     }
     if (_protocols.includes(Protocols.LightPush)) {
       codecs.push(LightPushCodec);
@@ -364,9 +366,11 @@ export class Waku {
           this.libp2p.peerStore.on(
             "change:protocols",
             ({ protocols: connectedPeerProtocols }) => {
-              if (connectedPeerProtocols.includes(StoreCodec)) {
-                dbg("Resolving for", StoreCodec, connectedPeerProtocols);
-                resolve();
+              for (const codec of Object.values(StoreCodecs)) {
+                if (connectedPeerProtocols.includes(codec)) {
+                  dbg("Resolving for", codec, connectedPeerProtocols);
+                  resolve();
+                }
               }
             }
           );

src/lib/waku_light_push/index.ts

@@ -100,7 +100,7 @@ export class WakuLightPush {
    * light push protocol. Waku may or  may not be currently connected to these peers.
    */
   get peers(): AsyncIterable<Peer> {
-    return getPeersForProtocol(this.libp2p, LightPushCodec);
+    return getPeersForProtocol(this.libp2p, [LightPushCodec]);
   }
 
   /**