perf(parallel): run local+remote fetches concurrently across services; stream+dedupe

2025-12-18 23:24:22 +01:00 · 2025-10-12 23:13:26 +02:00
parent ea1046fe13
commit 496d1df404
9 changed files with 204 additions and 373 deletions
--- a/src/services/articleService.ts
+++ b/src/services/articleService.ts
@@ -1,11 +1,12 @@
-import { RelayPool, onlyEvents } from 'applesauce-relay'
-import { lastValueFrom, take, takeUntil, timer, toArray } from 'rxjs'
+import { RelayPool } from 'applesauce-relay'
+import { lastValueFrom, take } from 'rxjs'
 import { nip19 } from 'nostr-tools'
 import { AddressPointer } from 'nostr-tools/nip19'
 import { NostrEvent } from 'nostr-tools'
 import { Helpers } from 'applesauce-core'
 import { RELAYS } from '../config/relays'
-import { prioritizeLocalRelays, partitionRelays } from '../utils/helpers'
+import { prioritizeLocalRelays, partitionRelays, createParallelReqStreams } from '../utils/helpers'
+import { merge, toArray as rxToArray } from 'rxjs'
 import { UserSettings } from './settingsService'
 import { rebroadcastEvents } from './rebroadcastService'

@@ -112,43 +113,10 @@ export async function fetchArticleByNaddr(
      '#d': [pointer.identifier]
    }

-    // Local-first: try local relays quickly, then ALWAYS query remote to merge
-    let events = [] as NostrEvent[]
-    if (localRelays.length > 0) {
-      try {
-        events = await lastValueFrom(
-          relayPool
-            .req(localRelays, filter)
-            .pipe(
-              onlyEvents(),
-              take(1),
-              takeUntil(timer(1200)),
-              toArray()
-            )
-        )
-      } catch {
-        events = []
-      }
-    }
-
-    // Always query remote to ensure we have the latest from the wider network
-    if (remoteRelays.length > 0) {
-      try {
-        const remoteEvents = await lastValueFrom(
-          relayPool
-            .req(remoteRelays, filter)
-            .pipe(
-              onlyEvents(),
-              take(1),
-              takeUntil(timer(6000)),
-              toArray()
-            )
-        )
-        events = events.concat(remoteEvents)
-      } catch {
-        // ignore
-      }
-    }
+    // Parallel local+remote, stream immediate, collect up to first from each
+    const { local$, remote$ } = createParallelReqStreams(relayPool, localRelays, remoteRelays, filter, 1200, 6000)
+    const collected = await lastValueFrom(merge(local$.pipe(take(1)), remote$.pipe(take(1))).pipe(rxToArray()))
+    const events = collected as NostrEvent[]

    if (events.length === 0) {
      throw new Error('Article not found')
--- a/src/services/articleTitleResolver.ts
+++ b/src/services/articleTitleResolver.ts
@@ -1,10 +1,11 @@
-import { RelayPool, onlyEvents } from 'applesauce-relay'
-import { lastValueFrom, take, takeUntil, timer, toArray } from 'rxjs'
+import { RelayPool } from 'applesauce-relay'
+import { lastValueFrom, take } from 'rxjs'
 import { nip19 } from 'nostr-tools'
 import { AddressPointer } from 'nostr-tools/nip19'
 import { Helpers } from 'applesauce-core'
 import { RELAYS } from '../config/relays'
-import { prioritizeLocalRelays, partitionRelays } from '../utils/helpers'
+import { prioritizeLocalRelays, partitionRelays, createParallelReqStreams } from '../utils/helpers'
+import { merge, toArray as rxToArray } from 'rxjs'

 const { getArticleTitle } = Helpers

@@ -39,28 +40,11 @@ export async function fetchArticleTitle(
      '#d': [pointer.identifier]
    }

-    // Try to get the first event quickly from local relays
-    let events: { created_at: number }[] = []
-    if (localRelays.length > 0) {
-      try {
-        events = await lastValueFrom(
-          relayPool
-            .req(localRelays, filter)
-            .pipe(onlyEvents(), take(1), takeUntil(timer(1200)), toArray())
-        )
-      } catch {
-        events = []
-      }
-    }
-    // Always follow up with remote relays to ensure we have latest network data
-    if (remoteRelays.length > 0) {
-      const remoteEvents = await lastValueFrom(
-        relayPool
-          .req(remoteRelays, filter)
-          .pipe(onlyEvents(), take(1), takeUntil(timer(5000)), toArray())
-      )
-      events = events.concat(remoteEvents as unknown as { created_at: number }[])
-    }
+    // Parallel local+remote: collect up to one event from each
+    const { local$, remote$ } = createParallelReqStreams(relayPool, localRelays, remoteRelays, filter, 1200, 5000)
+    const events = await lastValueFrom(
+      merge(local$.pipe(take(1)), remote$.pipe(take(1))).pipe(rxToArray())
+    ) as unknown as { created_at: number }[]

    if (events.length === 0) {
      return null
--- a/src/services/bookmarkService.ts
+++ b/src/services/bookmarkService.ts
@@ -1,5 +1,5 @@
 import { RelayPool, completeOnEose } from 'applesauce-relay'
-import { lastValueFrom, takeUntil, timer, toArray } from 'rxjs'
+import { lastValueFrom, merge, Observable, takeUntil, timer, toArray } from 'rxjs'
 import {
  AccountWithExtension,
  NostrEvent,
@@ -37,30 +37,17 @@ export const fetchBookmarks = async (
    // Fetch bookmark events - NIP-51 standards, legacy formats, and web bookmarks (NIP-B0)
    console.log('🔍 Fetching bookmark events from relays:', relayUrls)
    // Try local-first quickly, then full set fallback
-    let rawEvents = [] as NostrEvent[]
-    if (localRelays.length > 0) {
-      try {
-        rawEvents = await lastValueFrom(
-          relayPool
-            .req(localRelays, { kinds: [10003, 30003, 30001, 39701], authors: [activeAccount.pubkey] })
-            .pipe(completeOnEose(), takeUntil(timer(1200)), toArray())
-        )
-      } catch {
-        rawEvents = []
-      }
-    }
-    if (remoteRelays.length > 0) {
-      try {
-        const remoteEvents = await lastValueFrom(
-          relayPool
-            .req(remoteRelays, { kinds: [10003, 30003, 30001, 39701], authors: [activeAccount.pubkey] })
-            .pipe(completeOnEose(), takeUntil(timer(6000)), toArray())
-        )
-        rawEvents = rawEvents.concat(remoteEvents)
-      } catch {
-        // ignore
-      }
-    }
+    const local$ = localRelays.length > 0
+      ? relayPool
+          .req(localRelays, { kinds: [10003, 30003, 30001, 39701], authors: [activeAccount.pubkey] })
+          .pipe(completeOnEose(), takeUntil(timer(1200)))
+      : new Observable<NostrEvent>((sub) => sub.complete())
+    const remote$ = remoteRelays.length > 0
+      ? relayPool
+          .req(remoteRelays, { kinds: [10003, 30003, 30001, 39701], authors: [activeAccount.pubkey] })
+          .pipe(completeOnEose(), takeUntil(timer(6000)))
+      : new Observable<NostrEvent>((sub) => sub.complete())
+    const rawEvents = await lastValueFrom(merge(local$, remote$).pipe(toArray()))
    console.log('📊 Raw events fetched:', rawEvents.length, 'events')
    
    // Rebroadcast bookmark events to local/all relays based on settings
@@ -125,30 +112,13 @@ export const fetchBookmarks = async (
    if (noteIds.length > 0) {
      try {
        const { local: localHydrate, remote: remoteHydrate } = partitionRelays(relayUrls)
-        let events: NostrEvent[] = []
-        if (localHydrate.length > 0) {
-          try {
-            events = await lastValueFrom(
-              relayPool
-                .req(localHydrate, { ids: noteIds })
-                .pipe(completeOnEose(), takeUntil(timer(800)), toArray())
-            )
-          } catch {
-            events = []
-          }
-        }
-        if (remoteHydrate.length > 0) {
-          try {
-            const remote = await lastValueFrom(
-              relayPool
-                .req(remoteHydrate, { ids: noteIds })
-                .pipe(completeOnEose(), takeUntil(timer(2500)), toArray())
-            )
-            events = events.concat(remote)
-          } catch {
-            // ignore
-          }
-        }
+        const localHydrate$ = localHydrate.length > 0
+          ? relayPool.req(localHydrate, { ids: noteIds }).pipe(completeOnEose(), takeUntil(timer(800)))
+          : new Observable<NostrEvent>((sub) => sub.complete())
+        const remoteHydrate$ = remoteHydrate.length > 0
+          ? relayPool.req(remoteHydrate, { ids: noteIds }).pipe(completeOnEose(), takeUntil(timer(2500)))
+          : new Observable<NostrEvent>((sub) => sub.complete())
+        const events: NostrEvent[] = await lastValueFrom(merge(localHydrate$, remoteHydrate$).pipe(toArray()))
        idToEvent = new Map(events.map((e: NostrEvent) => [e.id, e]))
      } catch (error) {
        console.warn('Failed to fetch events for hydration:', error)
--- a/src/services/contactService.ts
+++ b/src/services/contactService.ts
@@ -1,5 +1,5 @@
 import { RelayPool, completeOnEose } from 'applesauce-relay'
-import { lastValueFrom, takeUntil, timer, toArray } from 'rxjs'
+import { lastValueFrom, merge, Observable, takeUntil, timer, toArray } from 'rxjs'
 import { prioritizeLocalRelays } from '../utils/helpers'

 /**
@@ -20,19 +20,20 @@ export const fetchContacts = async (
    
    // Local-first quick attempt
    const localRelays = relayUrls.filter(url => url.includes('localhost') || url.includes('127.0.0.1'))
-    let events: Array<{ created_at: number; tags: string[][] }> = []
-    if (localRelays.length > 0) {
-      try {
-        const localEvents = await lastValueFrom(
-          relayPool
-            .req(localRelays, { kinds: [3], authors: [pubkey] })
-            .pipe(completeOnEose(), takeUntil(timer(1200)), toArray())
-        )
-        events = localEvents as Array<{ created_at: number; tags: string[][] }>
-      } catch {
-        events = []
-      }
-    }
+    const remoteRelays = relayUrls.filter(url => !url.includes('localhost') && !url.includes('127.0.0.1'))
+    const local$ = localRelays.length > 0
+      ? relayPool
+          .req(localRelays, { kinds: [3], authors: [pubkey] })
+          .pipe(completeOnEose(), takeUntil(timer(1200)))
+      : new Observable<{ created_at: number; tags: string[][] }>((sub) => sub.complete())
+    const remote$ = remoteRelays.length > 0
+      ? relayPool
+          .req(remoteRelays, { kinds: [3], authors: [pubkey] })
+          .pipe(completeOnEose(), takeUntil(timer(6000)))
+      : new Observable<{ created_at: number; tags: string[][] }>((sub) => sub.complete())
+    const events = await lastValueFrom(
+      merge(local$, remote$).pipe(toArray())
+    )
    const followed = new Set<string>()
    if (events.length > 0) {
      // Get the most recent contact list
@@ -46,29 +47,7 @@ export const fetchContacts = async (
      }
      if (onPartial) onPartial(new Set(followed))
    }
-    // Always fetch remote to merge more contacts
-    const remoteRelays = relayUrls.filter(url => !url.includes('localhost') && !url.includes('127.0.0.1'))
-    if (remoteRelays.length > 0) {
-      try {
-        const remoteEvents = await lastValueFrom(
-          relayPool
-            .req(remoteRelays, { kinds: [3], authors: [pubkey] })
-            .pipe(completeOnEose(), takeUntil(timer(6000)), toArray())
-        )
-        if (remoteEvents.length > 0) {
-          const sortedRemote = (remoteEvents as Array<{ created_at: number; tags: string[][] }>).
-            sort((a, b) => b.created_at - a.created_at)
-          const contactList = sortedRemote[0]
-          for (const tag of contactList.tags) {
-            if (tag[0] === 'p' && tag[1]) {
-              followed.add(tag[1])
-            }
-          }
-        }
-      } catch {
-        // ignore
-      }
-    }
+    // merged already via streams
    
    console.log('📊 Contact events fetched:', events.length)
    
--- a/src/services/exploreService.ts
+++ b/src/services/exploreService.ts
@@ -1,5 +1,5 @@
 import { RelayPool, completeOnEose } from 'applesauce-relay'
-import { lastValueFrom, takeUntil, timer, toArray } from 'rxjs'
+import { lastValueFrom, merge, Observable, takeUntil, timer, toArray } from 'rxjs'
 import { prioritizeLocalRelays, partitionRelays } from '../utils/helpers'
 import { NostrEvent } from 'nostr-tools'
 import { Helpers } from 'applesauce-core'
@@ -66,49 +66,18 @@ export const fetchBlogPostsFromAuthors = async (
      }
    }

-    // Phase 1: local relays fast path
-    if (localRelays.length > 0) {
-      try {
-        const localEvents = await lastValueFrom(
-          relayPool
-            .req(localRelays, {
-              kinds: [30023],
-              authors: pubkeys,
-              limit: 100
-            })
-            .pipe(
-              completeOnEose(),
-              takeUntil(timer(1200)),
-              toArray()
-            )
-        )
-        processEvents(localEvents)
-      } catch {
-        // ignore
-      }
-    }
-
-    // Phase 2: always query remote relays to fill in missing content
-    if (remoteRelays.length > 0) {
-      try {
-        const remoteEvents = await lastValueFrom(
-          relayPool
-            .req(remoteRelays, {
-              kinds: [30023],
-              authors: pubkeys,
-              limit: 100
-            })
-            .pipe(
-              completeOnEose(),
-              takeUntil(timer(6000)),
-              toArray()
-            )
-        )
-        processEvents(remoteEvents)
-      } catch {
-        // ignore
-      }
-    }
+    const local$ = localRelays.length > 0
+      ? relayPool
+          .req(localRelays, { kinds: [30023], authors: pubkeys, limit: 100 })
+          .pipe(completeOnEose(), takeUntil(timer(1200)))
+      : new Observable<NostrEvent>((sub) => sub.complete())
+    const remote$ = remoteRelays.length > 0
+      ? relayPool
+          .req(remoteRelays, { kinds: [30023], authors: pubkeys, limit: 100 })
+          .pipe(completeOnEose(), takeUntil(timer(6000)))
+      : new Observable<NostrEvent>((sub) => sub.complete())
+    const events = await lastValueFrom(merge(local$, remote$).pipe(toArray()))
+    processEvents(events)

    console.log('📊 Blog post events fetched (unique):', uniqueEvents.size)
    
--- a/src/services/highlights/fetchByAuthor.ts
+++ b/src/services/highlights/fetchByAuthor.ts
@@ -1,5 +1,5 @@
 import { RelayPool, completeOnEose, onlyEvents } from 'applesauce-relay'
-import { lastValueFrom, takeUntil, timer, tap, toArray } from 'rxjs'
+import { lastValueFrom, merge, Observable, takeUntil, timer, tap, toArray } from 'rxjs'
 import { NostrEvent } from 'nostr-tools'
 import { Highlight } from '../../types/highlights'
 import { prioritizeLocalRelays, partitionRelays } from '../../utils/helpers'
@@ -19,52 +19,37 @@ export const fetchHighlights = async (
    const { local: localRelays, remote: remoteRelays } = partitionRelays(ordered)

    const seenIds = new Set<string>()
-    let rawEvents: NostrEvent[] = []
-    if (localRelays.length > 0) {
-      try {
-        rawEvents = await lastValueFrom(
-          relayPool
-            .req(localRelays, { kinds: [9802], authors: [pubkey] })
-            .pipe(
-              onlyEvents(),
-              tap((event: NostrEvent) => {
-                if (!seenIds.has(event.id)) {
-                  seenIds.add(event.id)
-                  if (onHighlight) onHighlight(eventToHighlight(event))
-                }
-              }),
-              completeOnEose(),
-              takeUntil(timer(1200)),
-              toArray()
-            )
-        )
-      } catch {
-        rawEvents = []
-      }
-    }
-    if (remoteRelays.length > 0) {
-      try {
-        const remoteEvents = await lastValueFrom(
-          relayPool
-            .req(remoteRelays, { kinds: [9802], authors: [pubkey] })
-            .pipe(
-              onlyEvents(),
-              tap((event: NostrEvent) => {
-                if (!seenIds.has(event.id)) {
-                  seenIds.add(event.id)
-                  if (onHighlight) onHighlight(eventToHighlight(event))
-                }
-              }),
-              completeOnEose(),
-              takeUntil(timer(6000)),
-              toArray()
-            )
-        )
-        rawEvents = rawEvents.concat(remoteEvents)
-      } catch {
-        // ignore
-      }
-    }
+    const local$ = localRelays.length > 0
+      ? relayPool
+          .req(localRelays, { kinds: [9802], authors: [pubkey] })
+          .pipe(
+            onlyEvents(),
+            tap((event: NostrEvent) => {
+              if (!seenIds.has(event.id)) {
+                seenIds.add(event.id)
+                if (onHighlight) onHighlight(eventToHighlight(event))
+              }
+            }),
+            completeOnEose(),
+            takeUntil(timer(1200))
+          )
+      : new Observable<NostrEvent>((sub) => sub.complete())
+    const remote$ = remoteRelays.length > 0
+      ? relayPool
+          .req(remoteRelays, { kinds: [9802], authors: [pubkey] })
+          .pipe(
+            onlyEvents(),
+            tap((event: NostrEvent) => {
+              if (!seenIds.has(event.id)) {
+                seenIds.add(event.id)
+                if (onHighlight) onHighlight(eventToHighlight(event))
+              }
+            }),
+            completeOnEose(),
+            takeUntil(timer(6000))
+          )
+      : new Observable<NostrEvent>((sub) => sub.complete())
+    const rawEvents: NostrEvent[] = await lastValueFrom(merge(local$, remote$).pipe(toArray()))

    await rebroadcastEvents(rawEvents, relayPool, settings)
    const uniqueEvents = dedupeHighlights(rawEvents)
--- a/src/services/highlights/fetchForArticle.ts
+++ b/src/services/highlights/fetchForArticle.ts
@@ -1,5 +1,5 @@
 import { RelayPool, completeOnEose, onlyEvents } from 'applesauce-relay'
-import { lastValueFrom, takeUntil, timer, tap, toArray } from 'rxjs'
+import { lastValueFrom, merge, Observable, takeUntil, timer, tap, toArray } from 'rxjs'
 import { NostrEvent } from 'nostr-tools'
 import { Highlight } from '../../types/highlights'
 import { RELAYS } from '../../config/relays'
@@ -26,96 +26,63 @@ export const fetchHighlightsForArticle = async (
    const orderedRelays = prioritizeLocalRelays(RELAYS)
    const { local: localRelays, remote: remoteRelays } = partitionRelays(orderedRelays)

-    let aTagEvents: NostrEvent[] = []
-    if (localRelays.length > 0) {
-      try {
-        aTagEvents = await lastValueFrom(
-          relayPool
-            .req(localRelays, { kinds: [9802], '#a': [articleCoordinate] })
-            .pipe(
-              onlyEvents(),
-              tap((event: NostrEvent) => {
-                const highlight = processEvent(event)
-                if (highlight && onHighlight) onHighlight(highlight)
-              }),
-              completeOnEose(),
-              takeUntil(timer(1200)),
-              toArray()
-            )
-        )
-      } catch {
-        aTagEvents = []
-      }
-    }
-
-    // Always query remote relays to merge additional highlights
-    if (remoteRelays.length > 0) {
-      try {
-        const aRemote = await lastValueFrom(
-          relayPool
-            .req(remoteRelays, { kinds: [9802], '#a': [articleCoordinate] })
-            .pipe(
-              onlyEvents(),
-              tap((event: NostrEvent) => {
-                const highlight = processEvent(event)
-                if (highlight && onHighlight) onHighlight(highlight)
-              }),
-              completeOnEose(),
-              takeUntil(timer(6000)),
-              toArray()
-            )
-        )
-        aTagEvents = aTagEvents.concat(aRemote)
-      } catch {
-        // ignore
-      }
-    }
+    const aLocal$ = localRelays.length > 0
+      ? relayPool
+          .req(localRelays, { kinds: [9802], '#a': [articleCoordinate] })
+          .pipe(
+            onlyEvents(),
+            tap((event: NostrEvent) => {
+              const highlight = processEvent(event)
+              if (highlight && onHighlight) onHighlight(highlight)
+            }),
+            completeOnEose(),
+            takeUntil(timer(1200))
+          )
+      : new Observable<NostrEvent>((sub) => sub.complete())
+    const aRemote$ = remoteRelays.length > 0
+      ? relayPool
+          .req(remoteRelays, { kinds: [9802], '#a': [articleCoordinate] })
+          .pipe(
+            onlyEvents(),
+            tap((event: NostrEvent) => {
+              const highlight = processEvent(event)
+              if (highlight && onHighlight) onHighlight(highlight)
+            }),
+            completeOnEose(),
+            takeUntil(timer(6000))
+          )
+      : new Observable<NostrEvent>((sub) => sub.complete())
+    const aTagEvents: NostrEvent[] = await lastValueFrom(merge(aLocal$, aRemote$).pipe(toArray()))

    let eTagEvents: NostrEvent[] = []
    if (eventId) {
-      if (localRelays.length > 0) {
-        try {
-          eTagEvents = await lastValueFrom(
-            relayPool
-              .req(localRelays, { kinds: [9802], '#e': [eventId] })
-              .pipe(
-                onlyEvents(),
-                tap((event: NostrEvent) => {
-                  const highlight = processEvent(event)
-                  if (highlight && onHighlight) onHighlight(highlight)
-                }),
-                completeOnEose(),
-                takeUntil(timer(1200)),
-                toArray()
-              )
-          )
-        } catch {
-          eTagEvents = []
-        }
-      }
-
-      // Always query remote for e-tag too
-      if (remoteRelays.length > 0) {
-        try {
-          const eRemote = await lastValueFrom(
-            relayPool
-              .req(remoteRelays, { kinds: [9802], '#e': [eventId] })
-              .pipe(
-                onlyEvents(),
-                tap((event: NostrEvent) => {
-                  const highlight = processEvent(event)
-                  if (highlight && onHighlight) onHighlight(highlight)
-                }),
-                completeOnEose(),
-                takeUntil(timer(6000)),
-                toArray()
-              )
-          )
-          eTagEvents = eTagEvents.concat(eRemote)
-        } catch {
-          // ignore
-        }
-      }
+      const eLocal$ = localRelays.length > 0
+        ? relayPool
+            .req(localRelays, { kinds: [9802], '#e': [eventId] })
+            .pipe(
+              onlyEvents(),
+              tap((event: NostrEvent) => {
+                const highlight = processEvent(event)
+                if (highlight && onHighlight) onHighlight(highlight)
+              }),
+              completeOnEose(),
+              takeUntil(timer(1200))
+            )
+        : new Observable<NostrEvent>((sub) => sub.complete())
+      const eRemote$ = remoteRelays.length > 0
+        ? relayPool
+            .req(remoteRelays, { kinds: [9802], '#e': [eventId] })
+            .pipe(
+              onlyEvents(),
+              tap((event: NostrEvent) => {
+                const highlight = processEvent(event)
+                if (highlight && onHighlight) onHighlight(highlight)
+              }),
+              completeOnEose(),
+              takeUntil(timer(6000))
+            )
+        : new Observable<NostrEvent>((sub) => sub.complete())
+      eTagEvents = await lastValueFrom(merge(eLocal$, eRemote$).pipe(toArray()))
    }

    const rawEvents = [...aTagEvents, ...eTagEvents]
--- a/src/services/highlights/fetchForUrl.ts
+++ b/src/services/highlights/fetchForUrl.ts
@@ -1,5 +1,5 @@
 import { RelayPool, completeOnEose, onlyEvents } from 'applesauce-relay'
-import { lastValueFrom, takeUntil, timer, tap, toArray } from 'rxjs'
+import { lastValueFrom, merge, Observable, takeUntil, timer, tap, toArray } from 'rxjs'
 import { NostrEvent } from 'nostr-tools'
 import { Highlight } from '../../types/highlights'
 import { RELAYS } from '../../config/relays'
@@ -18,48 +18,33 @@ export const fetchHighlightsForUrl = async (
    const seenIds = new Set<string>()
    const orderedRelaysUrl = prioritizeLocalRelays(RELAYS)
    const { local: localRelaysUrl, remote: remoteRelaysUrl } = partitionRelays(orderedRelaysUrl)
-    let rawEvents: NostrEvent[] = []
-    if (localRelaysUrl.length > 0) {
-      try {
-        rawEvents = await lastValueFrom(
-          relayPool
-            .req(localRelaysUrl, { kinds: [9802], '#r': [url] })
-            .pipe(
-              onlyEvents(),
-              tap((event: NostrEvent) => {
-                seenIds.add(event.id)
-                if (onHighlight) onHighlight(eventToHighlight(event))
-              }),
-              completeOnEose(),
-              takeUntil(timer(1200)),
-              toArray()
-            )
-        )
-      } catch {
-        rawEvents = []
-      }
-    }
-    if (remoteRelaysUrl.length > 0) {
-      try {
-        const remote = await lastValueFrom(
-          relayPool
-            .req(remoteRelaysUrl, { kinds: [9802], '#r': [url] })
-            .pipe(
-              onlyEvents(),
-              tap((event: NostrEvent) => {
-                seenIds.add(event.id)
-                if (onHighlight) onHighlight(eventToHighlight(event))
-              }),
-              completeOnEose(),
-              takeUntil(timer(6000)),
-              toArray()
-            )
-        )
-        rawEvents = rawEvents.concat(remote)
-      } catch {
-        // ignore
-      }
-    }
+    const local$ = localRelaysUrl.length > 0
+      ? relayPool
+          .req(localRelaysUrl, { kinds: [9802], '#r': [url] })
+          .pipe(
+            onlyEvents(),
+            tap((event: NostrEvent) => {
+              seenIds.add(event.id)
+              if (onHighlight) onHighlight(eventToHighlight(event))
+            }),
+            completeOnEose(),
+            takeUntil(timer(1200))
+          )
+      : new Observable<NostrEvent>((sub) => sub.complete())
+    const remote$ = remoteRelaysUrl.length > 0
+      ? relayPool
+          .req(remoteRelaysUrl, { kinds: [9802], '#r': [url] })
+          .pipe(
+            onlyEvents(),
+            tap((event: NostrEvent) => {
+              seenIds.add(event.id)
+              if (onHighlight) onHighlight(eventToHighlight(event))
+            }),
+            completeOnEose(),
+            takeUntil(timer(6000))
+          )
+      : new Observable<NostrEvent>((sub) => sub.complete())
+    const rawEvents: NostrEvent[] = await lastValueFrom(merge(local$, remote$).pipe(toArray()))
    await rebroadcastEvents(rawEvents, relayPool, settings)
    const uniqueEvents = dedupeHighlights(rawEvents)
    const highlights: Highlight[] = uniqueEvents.map(eventToHighlight)
--- a/src/utils/helpers.ts
+++ b/src/utils/helpers.ts
@@ -94,3 +94,27 @@ export const prioritizeLocalRelays = (relayUrls: string[]): string[] => {
  return out
 }

+// Parallel request helper
+import { completeOnEose, onlyEvents, RelayPool } from 'applesauce-relay'
+import { Observable, takeUntil, timer } from 'rxjs'
+
+export function createParallelReqStreams(
+  relayPool: RelayPool,
+  localRelays: string[],
+  remoteRelays: string[],
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  filter: any,
+  localTimeoutMs = 1200,
+  remoteTimeoutMs = 6000
+): { local$: Observable<unknown>; remote$: Observable<unknown> } {
+  const local$ = (localRelays.length > 0)
+    ? relayPool.req(localRelays, filter).pipe(onlyEvents(), completeOnEose(), takeUntil(timer(localTimeoutMs)))
+    : new Observable<unknown>((sub) => { sub.complete() })
+
+  const remote$ = (remoteRelays.length > 0)
+    ? relayPool.req(remoteRelays, filter).pipe(onlyEvents(), completeOnEose(), takeUntil(timer(remoteTimeoutMs)))
+    : new Observable<unknown>((sub) => { sub.complete() })
+
+  return { local$, remote$ }
+}
+