Reqmeister: improve API and fetching logic (untested)

src/pipeline.ts

@@ -59,7 +59,7 @@ async function handleEvent(event: DittoEvent, signal: AbortSignal): Promise<void
     processDeletions(event, signal),
     DVM.event(event),
     trackHashtags(event),
-    fetchRelatedEvents(event, signal),
+    fetchRelatedEvents(event),
     processMedia(event),
     payZap(event, signal),
     streamOut(event),
@@ -182,19 +182,21 @@ async function trackHashtags(event: NostrEvent): Promise<void> {
 }
 
 /** Queue related events to fetch. */
-async function fetchRelatedEvents(event: DittoEvent, signal: AbortSignal) {
+async function fetchRelatedEvents(event: DittoEvent) {
   if (!event.author) {
-    Storages.reqmeister.req({ kinds: [0], authors: [event.pubkey] }, { signal })
+    const signal = AbortSignal.timeout(3000);
-      .then((event) => handleEvent(event, AbortSignal.timeout(1000)))
+    Storages.reqmeister.query([{ kinds: [0], authors: [event.pubkey] }], { signal })
+      .then((events) => events.forEach((event) => handleEvent(event, signal)))
       .catch(() => {});
   }
 
-  for (const [name, id, relay] of event.tags) {
+  for (const [name, id] of event.tags) {
     if (name === 'e') {
       const { count } = await Storages.cache.count([{ ids: [id] }]);
       if (!count) {
-        Storages.reqmeister.req({ ids: [id] }, { relays: [relay] })
+        const signal = AbortSignal.timeout(3000);
-          .then((event) => handleEvent(event, AbortSignal.timeout(1000)))
+        Storages.reqmeister.query([{ ids: [id] }], { signal })
+          .then((events) => events.forEach((event) => handleEvent(event, signal)))
           .catch(() => {});
       }
     }

src/storages/reqmeister.ts

@@ -82,7 +82,7 @@ class Reqmeister extends EventEmitter<{ [filterId: string]: (event: NostrEvent)
     this.#perform();
   }
 
-  req(filter: MicroFilter, opts: ReqmeisterReqOpts = {}): Promise<NostrEvent> {
+  private fetch(filter: MicroFilter, opts: ReqmeisterReqOpts = {}): Promise<NostrEvent> {
     const {
       relays = [],
       signal = AbortSignal.timeout(this.#opts.timeout ?? 1000),
@@ -120,12 +120,7 @@ class Reqmeister extends EventEmitter<{ [filterId: string]: (event: NostrEvent)
     return Promise.resolve();
   }
 
-  isWanted(event: NostrEvent): boolean {
+  async query(filters: NostrFilter[], opts?: { signal?: AbortSignal }): Promise<NostrEvent[]> {
-    const filterId = getFilterId(eventToMicroFilter(event));
-    return this.#queue.some(([id]) => id === filterId);
-  }
-
-  query(filters: NostrFilter[], opts?: { signal?: AbortSignal }): Promise<NostrEvent[]> {
     if (opts?.signal?.aborted) return Promise.reject(abortError());
 
     this.#debug('REQ', JSON.stringify(filters));
@@ -133,12 +128,16 @@ class Reqmeister extends EventEmitter<{ [filterId: string]: (event: NostrEvent)
 
     const promises = filters.reduce<Promise<NostrEvent>[]>((result, filter) => {
       if (isMicrofilter(filter)) {
-        result.push(this.req(filter, opts));
+        result.push(this.fetch(filter, opts));
       }
       return result;
     }, []);
 
-    return Promise.all(promises);
+    const results = await Promise.allSettled(promises);
+
+    return results
+      .filter((result): result is PromiseFulfilledResult<NostrEvent> => result.status === 'fulfilled')
+      .map((result) => result.value);
   }
 }