Various optimizations

packages/app/src/handles.ts

@@ -1,6 +1,13 @@
-import {writable, derived, Subscriber} from "svelte/store"
+import {writable, Subscriber} from "svelte/store"
 import {tryCatch, fetchJson, batcher, postJson, last} from "@welshman/lib"
-import {getter, deriveItems, makeForceLoadItem, makeLoadItem, makeDeriveItem} from "@welshman/store"
+import {
+  getter,
+  deriveItems,
+  deriveDeduplicated,
+  makeForceLoadItem,
+  makeLoadItem,
+  makeDeriveItem,
+} from "@welshman/store"
 import {deriveProfile, loadProfile} from "./profiles.js"
 import {appContext} from "./context.js"
 
@@ -123,15 +130,18 @@ export const loadHandleForPubkey = async (pubkey: string, relays: string[] = [])
 export const deriveHandleForPubkey = (pubkey: string, relays: string[] = []) => {
   loadHandleForPubkey(pubkey, relays)
 
-  return derived([handlesByNip05, deriveProfile(pubkey, relays)], ([$handlesByNip05, $profile]) => {
-    if (!$profile?.nip05) return undefined
+  return deriveDeduplicated(
+    [handlesByNip05, deriveProfile(pubkey, relays)],
+    ([$handlesByNip05, $profile]) => {
+      if (!$profile?.nip05) return undefined
 
-    const handle = $handlesByNip05.get($profile.nip05)
+      const handle = $handlesByNip05.get($profile.nip05)
 
-    if (handle?.pubkey !== pubkey) return undefined
+      if (handle?.pubkey !== pubkey) return undefined
 
-    return handle
-  })
+      return handle
+    },
+  )
 }
 
 export const displayNip05 = (nip05: string) =>

packages/app/src/relayStats.ts

@@ -109,7 +109,7 @@ export const getRelayQuality = (url: string) => {
 
 type RelayStatsUpdate = [string, (stats: RelayStats) => void]
 
-const updateRelayStats = batch(500, (updates: RelayStatsUpdate[]) => {
+const updateRelayStats = batch(1000, (updates: RelayStatsUpdate[]) => {
   relayStatsByUrl.update($relayStatsByUrl => {
     for (const [url, items] of groupBy(([url]) => url, updates)) {
       if (!url || !isRelayUrl(url)) {

packages/app/src/thunk.ts

@@ -372,7 +372,8 @@ export const waitForThunkCompletion = (thunk: Thunk) =>
 export const thunks = writable<Thunk[]>([])
 
 export const thunkQueue = new TaskQueue<Thunk>({
-  batchSize: 50,
+  batchSize: 10,
+  batchDelay: 100,
   processItem: (thunk: Thunk) => {
     thunk.publish()
   },

packages/app/src/user.ts

@@ -1,5 +1,5 @@
 import {derived, Readable} from "svelte/store"
-import {ItemsByKey} from "@welshman/store"
+import {ItemsByKey, deriveDeduplicated} from "@welshman/store"
 import {pubkey} from "./session.js"
 import {profilesByPubkey, forceLoadProfile, loadProfile} from "./profiles.js"
 import {followListsByPubkey, forceLoadFollowList, loadFollowList} from "./follows.js"
@@ -27,7 +27,7 @@ export const makeUserData = <T>(
   itemsByKey: Readable<ItemsByKey<T>>,
   onDerive?: (key: string, ...args: any[]) => void,
 ) =>
-  derived([itemsByKey, pubkey], ([$itemsByKey, $pubkey]) => {
+  deriveDeduplicated([itemsByKey, pubkey], ([$itemsByKey, $pubkey]) => {
     if (!$pubkey) return undefined
 
     onDerive?.($pubkey)

packages/app/src/zappers.ts

@@ -1,4 +1,4 @@
-import {writable, derived, Subscriber} from "svelte/store"
+import {writable, Subscriber} from "svelte/store"
 import {Zapper, TrustedEvent, Zap, getTagValues, getLnUrl, zapFromEvent} from "@welshman/util"
 import {
   removeUndefined,
@@ -9,7 +9,14 @@ import {
   batcher,
   postJson,
 } from "@welshman/lib"
-import {getter, deriveItems, makeForceLoadItem, makeLoadItem, makeDeriveItem} from "@welshman/store"
+import {
+  getter,
+  deriveItems,
+  deriveDeduplicated,
+  makeForceLoadItem,
+  makeLoadItem,
+  makeDeriveItem,
+} from "@welshman/store"
 import {deriveProfile, loadProfile} from "./profiles.js"
 import {appContext} from "./context.js"
 
@@ -103,9 +110,12 @@ export const loadZapperForPubkey = async (pubkey: string, relays: string[] = [])
 export const deriveZapperForPubkey = (pubkey: string, relays: string[] = []) => {
   loadZapperForPubkey(pubkey, relays)
 
-  return derived([zappersByLnurl, deriveProfile(pubkey, relays)], ([$zappersByLnurl, $profile]) => {
-    return $profile?.lnurl ? $zappersByLnurl.get($profile.lnurl) : undefined
-  })
+  return deriveDeduplicated(
+    [zappersByLnurl, deriveProfile(pubkey, relays)],
+    ([$zappersByLnurl, $profile]) => {
+      return $profile?.lnurl ? $zappersByLnurl.get($profile.lnurl) : undefined
+    },
+  )
 }
 
 export const getLnUrlsForEvent = async (event: TrustedEvent) => {

packages/lib/src/TaskQueue.ts

@@ -1,7 +1,8 @@
-import {remove, yieldThread} from "./Tools.js"
+import {remove} from "./Tools.js"
 
 export type TaskQueueOptions<Item> = {
   batchSize: number
+  batchDelay: number
   processItem: (item: Item) => unknown
 }
 
@@ -30,32 +31,32 @@ export class TaskQueue<Item> {
     }
   }
 
-  async process() {
+  process() {
     if (this.isProcessing || this.isPaused || this.items.length === 0) {
       return
     }
 
     this.isProcessing = true
 
-    await yieldThread()
+    setTimeout(async () => {
+      for (const item of this.items.splice(0, this.options.batchSize)) {
+        try {
+          for (const subscriber of this._subs) {
+            subscriber(item)
+          }
 
-    for (const item of this.items.splice(0, this.options.batchSize)) {
-      try {
-        for (const subscriber of this._subs) {
-          subscriber(item)
+          await this.options.processItem(item)
+        } catch (e) {
+          console.error(e)
         }
-
-        await this.options.processItem(item)
-      } catch (e) {
-        console.error(e)
       }
-    }
 
-    this.isProcessing = false
+      this.isProcessing = false
 
-    if (this.items.length > 0) {
-      this.process()
-    }
+      if (this.items.length > 0) {
+        this.process()
+      }
+    }, this.options.batchDelay)
   }
 
   stop() {

packages/lib/src/Tools.ts

@@ -1257,11 +1257,9 @@ export const batch = <T>(t: number, f: (xs: T[]) => void) => {
   }
 
   return (x: T) => {
-    const shouldFlush = timeoutId === undefined
-
     xs.push(x)
 
-    if (shouldFlush) {
+    if (!timeoutId) {
       f(xs.splice(0))
       timeoutId = setTimeout(later, t)
     }

packages/net/src/socket.ts

@@ -32,6 +32,7 @@ export type SocketEvents = {
 
 export class Socket extends EventEmitter {
   static batchSize = 10
+  static batchDelay = 50
 
   auth: AuthState
   status = SocketStatus.Closed
@@ -47,6 +48,7 @@ export class Socket extends EventEmitter {
 
     this._sendQueue = new TaskQueue<ClientMessage>({
       batchSize: Socket.batchSize,
+      batchDelay: Socket.batchDelay,
       processItem: (message: ClientMessage) => {
         this._ws?.send(JSON.stringify(message))
         this.emit(SocketEvent.Send, message, this.url)
@@ -55,6 +57,7 @@ export class Socket extends EventEmitter {
 
     this._recvQueue = new TaskQueue<RelayMessage>({
       batchSize: Socket.batchSize,
+      batchDelay: Socket.batchDelay,
       processItem: (message: RelayMessage) => {
         this.emit(SocketEvent.Receive, message, this.url)
       },

packages/store/src/repository.ts

@@ -455,27 +455,27 @@ export const makeLoadItem = <T>(
   const pending = new Map<string, Promise<Maybe<T>>>()
   const attempts = new Map<string, number>()
 
-  return async (key: string, ...args: any[]): Promise<Maybe<T>> => {
+  return (key: string, ...args: any[]): Promise<Maybe<T>> => {
     const stale = getItem(key)
     const fetched = getFetched(key)
 
     // If we have an item, reload if it's relatively recent
     if (stale && fetched > now() - timeout) {
-      return stale
+      return Promise.resolve(stale)
     }
 
     const pendingItem = pending.get(key)
 
     // If we already are loading, await and return
     if (pendingItem) {
-      return pendingItem
+      return Promise.resolve(pendingItem)
     }
 
     const attempt = attempts.get(key) || 0
 
     // Use exponential backoff to throttle attempts
     if (fetched > now() - Math.pow(2, attempt)) {
-      return stale
+      return Promise.resolve(stale)
     }
 
     attempts.set(key, attempt + 1)
@@ -486,20 +486,22 @@ export const makeLoadItem = <T>(
 
     pending.set(key, promise)
 
-    let item
-    try {
-      item = await promise
-    } catch (e) {
-      console.warn(`Failed to load item ${key}`, e)
-    } finally {
-      pending.delete(key)
-    }
+    return call(async () => {
+      let item
+      try {
+        item = await promise
+      } catch (e) {
+        console.warn(`Failed to load item ${key}`, e)
+      } finally {
+        pending.delete(key)
+      }
 
-    if (item) {
-      attempts.delete(key)
-    }
+      if (item) {
+        attempts.delete(key)
+      }
 
-    return item
+      return item
+    })
   }
 }