feat(sds)_: add command queue architecture and improve message handling

- Introduce command queue system for sequential task processing
- Add comprehensive event system for message lifecycle tracking
- Restructure codebase with separate bloom_filter directory
- Export encode/decode helpers for SDS proto messages
- Use Set for deduplication in missing message detection
- Fix sync message handling for empty content messages
- Always emit MissedMessages event even with empty array
- Improve duplicate message detection logic

Author: adklempner

packages/sds/src/bloom.spec.ts renamed to packages/sds/src/bloom_filter/bloom.spec.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/bloom.ts renamed to packages/sds/src/bloom_filter/bloom.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/index.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/command_queue.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/events.ts

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