diff --git a/js/nostr/client.js b/js/nostr/client.js index 2b7e5ae3..b7aed262 100644 --- a/js/nostr/client.js +++ b/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 = {}) { diff --git a/js/services/dmNotificationManager.js b/js/services/dmNotificationManager.js index 0e45ed83..736d73d0 100644 --- a/js/services/dmNotificationManager.js +++ b/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() {