chore(sds): optimise lookups

packages/sds/src/message_channel/mem_local_history.spec.ts

@@ -7,44 +7,44 @@ describe("MemLocalHistory", () => {
   it("Cap max size when messages are pushed one at a time", () => {
     const maxSize = 2;
 
-    const hist = new MemLocalHistory(maxSize);
+    const hist = new MemLocalHistory({ maxSize: maxSize });
 
-    hist.push(
+    hist.addMessages(
       new ContentMessage("1", "c", "a", [], 1n, undefined, new Uint8Array([1]))
     );
-    expect(hist.length).to.eq(1);
-    hist.push(
+    expect(hist.size).to.eq(1);
+    hist.addMessages(
       new ContentMessage("2", "c", "a", [], 2n, undefined, new Uint8Array([2]))
     );
-    expect(hist.length).to.eq(2);
+    expect(hist.size).to.eq(2);
 
-    hist.push(
+    hist.addMessages(
       new ContentMessage("3", "c", "a", [], 3n, undefined, new Uint8Array([3]))
     );
-    expect(hist.length).to.eq(2);
+    expect(hist.size).to.eq(2);
 
-    expect(hist.findIndex((m) => m.messageId === "1")).to.eq(-1);
-    expect(hist.findIndex((m) => m.messageId === "2")).to.not.eq(-1);
-    expect(hist.findIndex((m) => m.messageId === "3")).to.not.eq(-1);
+    expect(hist.hasMessage("1")).to.eq(false);
+    expect(hist.hasMessage("2")).to.eq(true);
+    expect(hist.hasMessage("3")).to.eq(true);
   });
 
   it("Cap max size when a pushed array is exceeding the cap", () => {
     const maxSize = 2;
 
-    const hist = new MemLocalHistory(maxSize);
+    const hist = new MemLocalHistory({ maxSize: maxSize });
 
-    hist.push(
+    hist.addMessages(
       new ContentMessage("1", "c", "a", [], 1n, undefined, new Uint8Array([1]))
     );
-    expect(hist.length).to.eq(1);
-    hist.push(
+    expect(hist.size).to.eq(1);
+    hist.addMessages(
       new ContentMessage("2", "c", "a", [], 2n, undefined, new Uint8Array([2])),
       new ContentMessage("3", "c", "a", [], 3n, undefined, new Uint8Array([3]))
     );
-    expect(hist.length).to.eq(2);
+    expect(hist.size).to.eq(2);
 
-    expect(hist.findIndex((m) => m.messageId === "1")).to.eq(-1);
-    expect(hist.findIndex((m) => m.messageId === "2")).to.not.eq(-1);
-    expect(hist.findIndex((m) => m.messageId === "3")).to.not.eq(-1);
+    expect(hist.hasMessage("1")).to.eq(false);
+    expect(hist.hasMessage("2")).to.eq(true);
+    expect(hist.hasMessage("3")).to.eq(true);
   });
 });

packages/sds/src/message_channel/mem_local_history.ts

@@ -1,6 +1,13 @@
+import { Logger } from "@waku/utils";
 import _ from "lodash";
 
-import { ContentMessage, isContentMessage } from "./message.js";
+import {
+  type ChannelId,
+  ContentMessage,
+  type HistoryEntry,
+  isContentMessage
+} from "./message.js";
+import { PersistentStorage } from "./persistent_storage.js";
 
 export const DEFAULT_MAX_LENGTH = 10_000;
 
@@ -17,79 +24,111 @@ export const DEFAULT_MAX_LENGTH = 10_000;
  * If an array of items longer than `maxLength` is pushed, dropping will happen
  * at next push.
  */
-export class MemLocalHistory {
+export interface ILocalHistory {
+  readonly size: number;
+  addMessages(...messages: ContentMessage[]): void;
+  hasMessage(messageId: string): boolean;
+  getMessage(messageId: string): ContentMessage | undefined;
+  getRecentMessages(count: number): ContentMessage[];
+  getAllMessages(): ContentMessage[];
+  findMissingDependencies(entries: HistoryEntry[]): HistoryEntry[];
+}
+
+export type MemLocalHistoryOptions = {
+  storage?: ChannelId | PersistentStorage;
+  maxSize?: number;
+};
+
+const log = new Logger("sds:local-history");
+
+export class MemLocalHistory implements ILocalHistory {
   private items: ContentMessage[] = [];
+  private messageIndex: Map<string, ContentMessage> = new Map();
+  private readonly storage?: PersistentStorage;
+  private readonly maxSize: number;
 
   /**
-   * Construct a new in-memory local history
+   * Construct a new in-memory local history.
    *
-   * @param maxLength The maximum number of message to store.
+   * @param opts Configuration object.
+   *   - storage: Optional persistent storage backend for message persistence or channelId to use with PersistentStorage.
+   *   - maxSize: The maximum number of messages to store. Optional, defaults to DEFAULT_MAX_LENGTH.
    */
-  public constructor(private maxLength: number = DEFAULT_MAX_LENGTH) {}
+  public constructor(opts: MemLocalHistoryOptions = {}) {
+    const { storage, maxSize } = opts;
+    this.maxSize = maxSize ?? DEFAULT_MAX_LENGTH;
+    if (storage instanceof PersistentStorage) {
+      this.storage = storage;
+      log.info("Using explicit persistent storage");
+    } else if (typeof storage === "string") {
+      this.storage = PersistentStorage.create(storage);
+      log.info("Creating persistent storage for channel", storage);
+    } else {
+      this.storage = undefined;
+      log.info("Using in-memory storage");
+    }
 
-  public get length(): number {
+    this.load();
+  }
+
+  public get size(): number {
     return this.items.length;
   }
 
-  public push(...items: ContentMessage[]): number {
-    for (const item of items) {
-      this.validateMessage(item);
+  public addMessages(...messages: ContentMessage[]): void {
+    for (const message of messages) {
+      this.validateMessage(message);
     }
 
     // Add new items and sort by timestamp, ensuring uniqueness by messageId
     // The valueOf() method on ContentMessage enables native < operator sorting
-    const combinedItems = [...this.items, ...items];
+    const combinedItems = [...this.items, ...messages];
 
     // Sort by timestamp (using valueOf() which creates timestamp_messageId string)
     combinedItems.sort((a, b) => a.valueOf().localeCompare(b.valueOf()));
 
     // Remove duplicates by messageId while maintaining order
     this.items = _.uniqBy(combinedItems, "messageId");
 
+    this.rebuildIndex();
+
     // Let's drop older messages if max length is reached
-    if (this.length > this.maxLength) {
-      const numItemsToRemove = this.length - this.maxLength;
-      this.items.splice(0, numItemsToRemove);
+    if (this.size > this.maxSize) {
+      const numItemsToRemove = this.size - this.maxSize;
+      const removedItems = this.items.splice(0, numItemsToRemove);
+      for (const item of removedItems) {
+        this.messageIndex.delete(item.messageId);
+      }
     }
 
-    return this.items.length;
+    this.save();
   }
 
-  public some(
-    predicate: (
-      value: ContentMessage,
-      index: number,
-      array: ContentMessage[]
-    ) => unknown,
-    thisArg?: any
-  ): boolean {
-    return this.items.some(predicate, thisArg);
+  public hasMessage(messageId: string): boolean {
+    return this.messageIndex.has(messageId);
   }
 
-  public slice(start?: number, end?: number): ContentMessage[] {
-    return this.items.slice(start, end);
+  public getRecentMessages(count: number): ContentMessage[] {
+    return this.items.slice(-count);
   }
 
-  public find(
-    predicate: (
-      value: ContentMessage,
-      index: number,
-      obj: ContentMessage[]
-    ) => unknown,
-    thisArg?: any
-  ): ContentMessage | undefined {
-    return this.items.find(predicate, thisArg);
+  public getAllMessages(): ContentMessage[] {
+    return [...this.items];
   }
 
-  public findIndex(
-    predicate: (
-      value: ContentMessage,
-      index: number,
-      obj: ContentMessage[]
-    ) => unknown,
-    thisArg?: any
-  ): number {
-    return this.items.findIndex(predicate, thisArg);
+  public getMessage(messageId: string): ContentMessage | undefined {
+    return this.messageIndex.get(messageId);
+  }
+
+  public findMissingDependencies(entries: HistoryEntry[]): HistoryEntry[] {
+    return entries.filter((entry) => !this.messageIndex.has(entry.messageId));
+  }
+
+  private rebuildIndex(): void {
+    this.messageIndex.clear();
+    for (const message of this.items) {
+      this.messageIndex.set(message.messageId, message);
+    }
   }
 
   private validateMessage(message: ContentMessage): void {
@@ -99,4 +138,16 @@ export class MemLocalHistory {
       );
     }
   }
+
+  private save(): void {
+    this.storage?.save(this.items);
+  }
+
+  private load(): void {
+    const messages = this.storage?.load() ?? [];
+    if (messages.length > 0) {
+      this.items = messages;
+      this.rebuildIndex();
+    }
+  }
 }