Feat/sds command queue

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

@@ -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/store.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/sds/src/bloom.ts → packages/sds/src/bloom_filter/bloom.ts

@@ -1,5 +1,5 @@
-import { hashN } from "./nim_hashn/nim_hashn.mjs";
-import { getMOverNBitsForK } from "./probabilities.js";
+import { hashN } from "../nim_hashn/nim_hashn.mjs";
+import { getMOverNBitsForK } from "../probabilities.js";
 
 export interface BloomFilterOptions {
   // The expected maximum number of elements for which this BloomFilter is sized.

packages/sds/src/index.ts

@@ -1,3 +1,5 @@
-import { BloomFilter } from "./bloom.js";
+import { BloomFilter } from "./bloom_filter/bloom.js";
+
+export * from "./message_channel/index.js";
 
 export { BloomFilter };

packages/sds/src/message_channel/command_queue.ts

@@ -0,0 +1,34 @@
+import { Message } from "./events.js";
+
+export enum Command {
+  Send = "send",
+  Receive = "receive",
+  SendEphemeral = "sendEphemeral"
+}
+
+export interface ParamsByAction {
+  [Command.Send]: {
+    payload: Uint8Array;
+    callback?: (message: Message) => Promise<{
+      success: boolean;
+      retrievalHint?: Uint8Array;
+    }>;
+  };
+  [Command.Receive]: {
+    message: Message;
+  };
+  [Command.SendEphemeral]: {
+    payload: Uint8Array;
+    callback?: (message: Message) => Promise<boolean>;
+  };
+}
+
+export type Task<A extends Command = Command> = {
+  command: A;
+  params: ParamsByAction[A];
+};
+
+// Define a mapping for handlers based on action type
+export type Handlers = {
+  [A in Command]: (params: ParamsByAction[A]) => Promise<void>;
+};

packages/sds/src/message_channel/events.ts

@@ -0,0 +1,41 @@
+import { proto_sds_message } from "@waku/proto";
+
+export enum MessageChannelEvent {
+  MessageSent = "messageSent",
+  MessageDelivered = "messageDelivered",
+  MessageReceived = "messageReceived",
+  MessageAcknowledged = "messageAcknowledged",
+  PartialAcknowledgement = "partialAcknowledgement",
+  MissedMessages = "missedMessages",
+  SyncSent = "syncSent",
+  SyncReceived = "syncReceived"
+}
+
+export type Message = proto_sds_message.SdsMessage;
+export type HistoryEntry = proto_sds_message.HistoryEntry;
+export type ChannelId = string;
+
+export function encodeMessage(message: Message): Uint8Array {
+  return proto_sds_message.SdsMessage.encode(message);
+}
+
+export function decodeMessage(data: Uint8Array): Message {
+  return proto_sds_message.SdsMessage.decode(data);
+}
+
+export type MessageChannelEvents = {
+  [MessageChannelEvent.MessageSent]: CustomEvent<Message>;
+  [MessageChannelEvent.MessageDelivered]: CustomEvent<{
+    messageId: string;
+    sentOrReceived: "sent" | "received";
+  }>;
+  [MessageChannelEvent.MessageReceived]: CustomEvent<Message>;
+  [MessageChannelEvent.MessageAcknowledged]: CustomEvent<string>;
+  [MessageChannelEvent.PartialAcknowledgement]: CustomEvent<{
+    messageId: string;
+    count: number;
+  }>;
+  [MessageChannelEvent.MissedMessages]: CustomEvent<HistoryEntry[]>;
+  [MessageChannelEvent.SyncSent]: CustomEvent<Message>;
+  [MessageChannelEvent.SyncReceived]: CustomEvent<Message>;
+};

packages/sds/src/message_channel/index.ts

@@ -0,0 +1,3 @@
+export * from "./command_queue.js";
+export * from "./events.js";
+export * from "./message_channel.js";