Refactor DM loading to be opportunistic and serialize notification updates

Co-authored-by: PR0M3TH3AN <[email protected]>

js/nostr/client.js

@@ -5248,65 +5248,73 @@ export class NostrClient {
       options,
     );
 
-    let events = [];
-    try {
-      events = await this.pool.list(relaysToUse, filters);
-    } catch (error) {
-      devLogger.warn("[nostr] Failed to list DM events.", error);
-      throw error;
-    }
-
-    const deduped = new Map();
-    for (const event of Array.isArray(events) ? events : []) {
-      if (!event || typeof event !== "object") {
-        continue;
-      }
-      const key =
-        typeof event.id === "string" && event.id
-          ? event.id
-          : `${event.kind || ""}:${event.pubkey || ""}:${event.created_at || ""}`;
-      if (!deduped.has(key)) {
-        deduped.set(key, event);
-      }
-    }
-
-    let decryptTargets = Array.from(deduped.values());
-    decryptTargets.sort((a, b) => (b?.created_at || 0) - (a?.created_at || 0));
-
-    if (decryptLimit && decryptTargets.length > decryptLimit) {
-      devLogger.debug("[nostr] Limiting DM decrypt workload", {
-        requested: decryptTargets.length,
-        decryptLimit,
-      });
-      decryptTargets = decryptTargets.slice(0, decryptLimit);
-    }
-
+    const dedupedIds = new Set();
     const messages = [];
-    for (const event of decryptTargets) {
-      try {
-        const decrypted = await this.decryptDirectMessageEvent(event, {
-          actorPubkey: context.actorPubkey || actorPubkeyInput,
-        });
-        if (decrypted?.ok) {
-          messages.push(decrypted);
-          if (typeof options.onMessage === "function") {
-            try {
-              options.onMessage(decrypted);
-            } catch (error) {
-              devLogger.warn("[nostr] onMessage callback threw:", error);
+    let decryptedCount = 0;
+
+    return new Promise((resolve) => {
+      const sub = this.pool.sub(relaysToUse, filters);
+      const timeoutMs = options.timeoutMs || 10000;
+      let settled = false;
+      let timeoutId = null;
+
+      const finish = () => {
+        if (settled) return;
+        settled = true;
+        if (timeoutId) clearTimeout(timeoutId);
+        sub.unsub();
+        messages.sort((a, b) => (b?.timestamp || 0) - (a?.timestamp || 0));
+        resolve(messages);
+      };
+
+      timeoutId = setTimeout(() => {
+        devLogger.warn("[nostr] listDirectMessages timed out.");
+        finish();
+      }, timeoutMs);
+
+      sub.on("event", async (event) => {
+        if (!event || typeof event !== "object") return;
+
+        const eventId = typeof event.id === "string" ? event.id : "";
+        if (eventId && dedupedIds.has(eventId)) return;
+        if (eventId) dedupedIds.add(eventId);
+
+        if (decryptLimit && decryptedCount >= decryptLimit) {
+          // We reached the limit, but since relays may send out of order,
+          // strictly stopping here might miss newer messages if they arrive late.
+          // However, to respect the limit and avoid over-decrypting massive histories,
+          // we stop processing new decrypts.
+          return;
+        }
+
+        try {
+          const decrypted = await this.decryptDirectMessageEvent(event, {
+            actorPubkey: context.actorPubkey || actorPubkeyInput,
+          });
+
+          if (decrypted?.ok) {
+            decryptedCount++;
+            messages.push(decrypted);
+            if (typeof options.onMessage === "function") {
+              try {
+                options.onMessage(decrypted);
+              } catch (error) {
+                devLogger.warn("[nostr] onMessage callback threw:", error);
+              }
             }
           }
+        } catch (error) {
+          devLogger.warn("[nostr] Failed to decrypt DM event during list.", {
+            error: sanitizeDecryptError(error),
+            event: summarizeDmEventForLog(event),
+          });
         }
-      } catch (error) {
-        devLogger.warn("[nostr] Failed to decrypt DM event during list.", {
-          error: sanitizeDecryptError(error),
-          event: summarizeDmEventForLog(event),
-        });
-      }
-    }
+      });
 
-    messages.sort((a, b) => (b?.timestamp || 0) - (a?.timestamp || 0));
-    return messages;
+      sub.on("eose", () => {
+        finish();
+      });
+    });
   }
 
   subscribeDirectMessages(actorPubkeyInput = null, options = {}) {

js/services/dmNotificationManager.js

@@ -39,6 +39,7 @@ export class DmNotificationManager {
   constructor({ logger } = {}) {
     this.logger = logger || userLogger;
     this.conversationCache = new Map();
+    this.updateQueue = new Map();
     this.focusedConversationId = "";
   }
 
@@ -46,8 +47,24 @@ export class DmNotificationManager {
     return this.focusedConversationId;
   }
 
+  enqueueUpdate(conversationId, task) {
+    const key = conversationId || "global";
+    const previous = this.updateQueue.get(key) || Promise.resolve();
+    const next = previous
+      .then(() => task())
+      .catch((error) => {
+        this.logger.warn(
+          `[dmNotifications] Update task failed for ${key}:`,
+          error,
+        );
+      });
+    this.updateQueue.set(key, next);
+    return next;
+  }
+
   reset() {
     this.conversationCache.clear();
+    this.updateQueue.clear();
     this.focusedConversationId = "";
   }
 
@@ -97,9 +114,9 @@ export class DmNotificationManager {
     }
   }
 
-  async recordMessage({ record, preview = "", direction = "" } = {}) {
+  recordMessage({ record, preview = "", direction = "" } = {}) {
     if (!record || typeof record !== "object") {
-      return null;
+      return Promise.resolve(null);
     }
 
     const conversationId =
@@ -107,84 +124,89 @@ export class DmNotificationManager {
         ? record.conversation_id.trim()
         : "";
     if (!conversationId) {
-      return null;
+      return Promise.resolve(null);
     }
 
-    const createdAt = sanitizeTimestamp(record.created_at);
-    const conversationState = await this.ensureConversationState(conversationId);
-    if (!conversationState) {
-      return null;
-    }
+    return this.enqueueUpdate(conversationId, async () => {
+      const createdAt = sanitizeTimestamp(record.created_at);
+      const conversationState = await this.ensureConversationState(conversationId);
+      if (!conversationState) {
+        return null;
+      }
 
-    const isIncoming = direction === "incoming";
-    const isFocused = this.focusedConversationId === conversationId;
-    const shouldNotify =
-      isIncoming && !isFocused && createdAt > conversationState.opened_until;
-    const unseenDelta =
-      isIncoming && createdAt > conversationState.opened_until ? 1 : 0;
+      const isIncoming = direction === "incoming";
+      const isFocused = this.focusedConversationId === conversationId;
+      const shouldNotify =
+        isIncoming && !isFocused && createdAt > conversationState.opened_until;
+      const unseenDelta =
+        isIncoming && createdAt > conversationState.opened_until ? 1 : 0;
 
-    let updated = null;
-    try {
-      updated = await updateConversationFromMessage(record, {
-        preview,
-        unseenDelta,
-        downloadedUntil: createdAt,
+      let updated = null;
+      try {
+        updated = await updateConversationFromMessage(record, {
+          preview,
+          unseenDelta,
+          downloadedUntil: createdAt,
+        });
+      } catch (error) {
+        this.logger.warn("[dmNotifications] Failed to update conversation", error);
+      }
+
+      const normalizedUpdate = normalizeConversationState({
+        conversation_id: conversationId,
+        ...(updated || conversationState),
+        opened_until: updated?.opened_until ?? conversationState.opened_until,
       });
-    } catch (error) {
-      this.logger.warn("[dmNotifications] Failed to update conversation", error);
-    }
 
-    const normalizedUpdate = normalizeConversationState({
-      conversation_id: conversationId,
-      ...updated,
-      opened_until: conversationState.opened_until,
+      this.conversationCache.set(conversationId, normalizedUpdate);
+
+      return {
+        conversation: normalizedUpdate,
+        shouldNotify,
+        isFocused,
+      };
     });
-
-    this.conversationCache.set(conversationId, normalizedUpdate);
-
-    return {
-      conversation: normalizedUpdate,
-      shouldNotify,
-      isFocused,
-    };
   }
 
-  async acknowledgeRenderedMessages({ conversationId, renderedUntil } = {}) {
+  acknowledgeRenderedMessages({ conversationId, renderedUntil } = {}) {
     if (typeof conversationId !== "string" || !conversationId.trim()) {
-      return null;
+      return Promise.resolve(null);
     }
 
     const normalized = conversationId.trim();
-    const conversationState = await this.ensureConversationState(normalized);
-    if (!conversationState) {
-      return null;
-    }
 
-    const nextOpenedUntil = Math.max(
-      conversationState.opened_until || 0,
-      sanitizeTimestamp(renderedUntil),
-    );
+    return this.enqueueUpdate(normalized, async () => {
+      const conversationState = await this.ensureConversationState(normalized);
+      if (!conversationState) {
+        return null;
+      }
 
-    let updated = null;
-    try {
-      updated = await updateConversationOpenState(normalized, {
-        openedUntil: nextOpenedUntil,
-        unseenCount: 0,
+      const nextOpenedUntil = Math.max(
+        conversationState.opened_until || 0,
+        sanitizeTimestamp(renderedUntil),
+      );
+
+      let updated = null;
+      try {
+        updated = await updateConversationOpenState(normalized, {
+          openedUntil: nextOpenedUntil,
+          unseenCount: 0,
+        });
+      } catch (error) {
+        this.logger.warn("[dmNotifications] Failed to update opened_until", error);
+      }
+
+      const normalizedUpdate = normalizeConversationState({
+        conversation_id: normalized,
+        ...conversationState,
+        ...updated,
+        opened_until: nextOpenedUntil,
+        unseen_count: 0,
       });
-    } catch (error) {
-      this.logger.warn("[dmNotifications] Failed to update opened_until", error);
-    }
 
-    const normalizedUpdate = normalizeConversationState({
-      conversation_id: normalized,
-      ...conversationState,
-      ...updated,
-      opened_until: nextOpenedUntil,
-      unseen_count: 0,
+      this.conversationCache.set(normalized, normalizedUpdate);
+      return normalizedUpdate;
     });
-
-    this.conversationCache.set(normalized, normalizedUpdate);
-    return normalizedUpdate;
   }
 
   async listConversationSummaries() {