Improve DVM feeds

package-lock.json

@@ -3061,10 +3061,10 @@
     },
     "packages/feeds": {
       "name": "@welshman/feeds",
-      "version": "0.0.1",
+      "version": "0.0.2",
       "license": "MIT",
       "dependencies": {
-        "@welshman/util": "0.0.1"
+        "@welshman/util": "0.0.2"
       },
       "devDependencies": {
         "gts": "^5.0.1",
@@ -3074,7 +3074,7 @@
     },
     "packages/lib": {
       "name": "@welshman/lib",
-      "version": "0.0.1",
+      "version": "0.0.2",
       "license": "MIT",
       "dependencies": {
         "@scure/base": "^1.1.6",
@@ -3098,11 +3098,11 @@
     },
     "packages/net": {
       "name": "@welshman/net",
-      "version": "0.0.1",
+      "version": "0.0.2",
       "license": "MIT",
       "dependencies": {
-        "@welshman/lib": "0.0.1",
-        "@welshman/util": "0.0.1",
+        "@welshman/lib": "0.0.2",
+        "@welshman/util": "0.0.2",
         "isomorphic-ws": "^5.0.0",
         "ws": "^8.16.0"
       },
@@ -3115,10 +3115,10 @@
     },
     "packages/util": {
       "name": "@welshman/util",
-      "version": "0.0.1",
+      "version": "0.0.2",
       "license": "MIT",
       "dependencies": {
-        "@welshman/lib": "0.0.1",
+        "@welshman/lib": "0.0.2",
         "nostr-tools": "^2.3.2"
       },
       "devDependencies": {

packages/feeds/compiler.ts

@@ -1,4 +1,4 @@
-import {uniq, now, isNil} from '@welshman/lib'
+import {uniq, tryCatch, now, isNil} from '@welshman/lib'
 import type {Rumor, Filter} from '@welshman/util'
 import {Tags, getIdFilters, mergeFilters} from '@welshman/util'
 import type {RequestItem, DVMItem, Scope, Feed, DynamicFilter, FeedOptions} from './core'
@@ -82,9 +82,10 @@ export class FeedCompiler<E extends Rumor> {
     await this.options.request({
       relays: [],
       filters: getIdFilters(addresses),
-      onEvent: events.push,
+      onEvent: (e: E) => events.push(e),
     })
 
+
     return {
       relays: [],
       filters: this._getFiltersFromTags(Tags.fromEvents(events)),
@@ -92,21 +93,27 @@ export class FeedCompiler<E extends Rumor> {
   }
 
   async _compileDvms(requests: DVMItem[]): Promise<RequestItem> {
-    const events: E[] = []
+    const responseTags: Tags[] = []
 
     await Promise.all(
       requests.map(request =>
         this.options.requestDvm({
-          tags: [],
           ...request,
-          onEvent: events.push,
+          onEvent: async (e: E) => {
+            const tags = Tags.fromEvent(e)
+            const {id, pubkey} = await tryCatch(() => JSON.parse(tags.get("request")?.value())) || {}
+
+            responseTags.push(tags.filterByKey(["t", "p", "e", "a"]).rejectByValue([id, pubkey]))
+          },
         })
       )
     )
 
+    const mergedTags = Tags.from(responseTags.flatMap(tags => tags.valueOf()))
+
     return {
-      relays: [],
-      filters: this._getFiltersFromTags(Tags.fromEvents(events)),
+      relays: mergedTags.relays().valueOf(),
+      filters: this._getFiltersFromTags(mergedTags),
     }
   }
 
@@ -160,6 +167,11 @@ export class FeedCompiler<E extends Rumor> {
       }
     }
 
+    // If we don't have any filters, return nothing instead of everything
+    if (filters.length === 0) {
+      filters.push({authors: []})
+    }
+
     return filters
   }
 }

packages/feeds/core.ts

@@ -90,6 +90,7 @@ export type RequestOpts<E> = RequestItem & {
 export type DVMItem = {
   kind: number
   tags?: string[][]
+  relays?: string[]
 }
 
 export type DVMOpts<E> = DVMItem & {

packages/feeds/loader.ts

@@ -39,6 +39,11 @@ export class FeedLoader<E extends Rumor> {
   }
 
   async _getRequestLoader({relays, filters}: RequestItem, {onEvent, onExhausted}: LoadOpts<E>) {
+    // Make sure we have some kind of filter to send if we've been given an empty one, as happens with relay feeds
+    if (filters.length === 0) {
+      filters = [{}]
+    }
+
     const untils = filters.flatMap((filter: Filter) => filter.until ? [filter.until] : [])
     const sinces = filters.flatMap((filter: Filter) => filter.since ? [filter.since] : [])
     const maxUntil = untils.length === filters.length ? max(untils) : now()

packages/feeds/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@welshman/feeds",
-  "version": "0.0.1",
+  "version": "0.0.2",
   "author": "hodlbod",
   "license": "MIT",
   "description": "Utilities for building dynamic nostr feeds.",
@@ -31,6 +31,6 @@
     "typescript": "~5.1.6"
   },
   "dependencies": {
-    "@welshman/util": "0.0.1"
+    "@welshman/util": "0.0.2"
   }
 }

packages/lib/Tools.ts

@@ -67,6 +67,24 @@ export const randomId = (): string => Math.random().toString().slice(2)
 
 export const stripProtocol = (url: string) => url.replace(/.*:\/\//, "")
 
+export const sleep = (t: number) => new Promise(resolve => setTimeout(resolve, t))
+
+export const concat = <T>(...xs: (T | Nil)[][]) => xs.flatMap(x => x || [])
+
+export const append = <T>(xs: (T | Nil)[], x: T) => concat(xs, [x])
+
+export const clamp = ([min, max]: [number, number], n: number) => Math.min(max, Math.max(min, n))
+
+export const tryCatch = async <T>(f: () => Promise<T | void> | T | void, onError?: (e: Error) => void): Promise<T | void> => {
+  try {
+    return await f()
+  } catch (e) {
+    onError?.(e as Error)
+  }
+
+  return undefined
+}
+
 // Curried utils
 
 export const nth = (i: number) => <T>(xs: T[], ...args: unknown[]) => xs[i]

packages/lib/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@welshman/lib",
-  "version": "0.0.1",
+  "version": "0.0.2",
   "author": "hodlbod",
   "license": "MIT",
   "description": "A collection of utilities.",

packages/net/Subscribe.ts

@@ -37,6 +37,7 @@ export type SubscribeRequest = {
   relays: string[]
   filters: Filter[]
   timeout?: number
+  tracker?: Tracker
   immediate?: boolean
   closeOnEose?: boolean
 }
@@ -53,8 +54,8 @@ export type Subscription = {
 export const makeSubscription = (request: SubscribeRequest) => {
   const id = randomId()
   const emitter = new Emitter()
-  const tracker = new Tracker()
   const result = defer<Event[]>()
+  const tracker = request.tracker || new Tracker()
   const close = () => emitter.emit('abort')
 
   emitter.setMaxListeners(100)

packages/net/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@welshman/net",
-  "version": "0.0.1",
+  "version": "0.0.2",
   "author": "hodlbod",
   "license": "MIT",
   "description": "Utilities for connecting with nostr relays.",
@@ -32,8 +32,8 @@
     "typescript": "~5.1.6"
   },
   "dependencies": {
-    "@welshman/lib": "0.0.1",
-    "@welshman/util": "0.0.1",
+    "@welshman/lib": "0.0.2",
+    "@welshman/util": "0.0.2",
     "isomorphic-ws": "^5.0.0",
     "ws": "^8.16.0"
   }

packages/util/Relay.ts

@@ -1,4 +1,4 @@
-import {Emitter, uniq, omit, now, range, identity} from '@welshman/lib'
+import {Emitter, chunk, sleep, uniq, omit, now, range, identity} from '@welshman/lib'
 import {matchFilters, matchFilter} from './Filters'
 import {encodeAddress, addressFromEvent} from './Address'
 import {isReplaceable} from './Events'
@@ -24,9 +24,15 @@ export class Relay<E extends Rumor> extends Emitter {
     return Array.from(this.eventsById.values())
   }
 
-  load(events: E[]) {
-    for (const event of events) {
-      this._addEvent(event)
+  async load(events: E[], chunkSize = 1000) {
+    for (const eventsChunk of chunk(chunkSize, events)) {
+      for (const event of eventsChunk) {
+        this._addEvent(event)
+      }
+
+      if (eventsChunk.length === chunkSize) {
+        await sleep(1)
+      }
     }
   }
 

packages/util/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@welshman/util",
-  "version": "0.0.1",
+  "version": "0.0.2",
   "author": "hodlbod",
   "license": "MIT",
   "description": "A collection of nostr-related utilities.",
@@ -32,7 +32,7 @@
     "typescript": "~5.1.6"
   },
   "dependencies": {
-    "@welshman/lib": "0.0.1",
+    "@welshman/lib": "0.0.2",
     "nostr-tools": "^2.3.2"
   }
 }