Add parameter to optimize repository.query, auto-connect when sending, optimize storage adapters

package-lock.json

@@ -1487,6 +1487,12 @@
       "version": "1.0.6",
       "license": "MIT"
     },
+    "node_modules/@types/events": {
+      "version": "3.0.3",
+      "resolved": "https://registry.npmjs.org/@types/events/-/events-3.0.3.tgz",
+      "integrity": "sha512-trOc4AAUThEz9hapPtSd7wf5tiQKvTtu5b371UxXdTuqzIh0ArcRspRP0i0Viu+LXstIQ1z96t1nsPxT9ol01g==",
+      "license": "MIT"
+    },
     "node_modules/@types/hast": {
       "version": "3.0.4",
       "dev": true,
@@ -2649,6 +2655,15 @@
         "node": ">=0.10.0"
       }
     },
+    "node_modules/events": {
+      "version": "3.3.0",
+      "resolved": "https://registry.npmjs.org/events/-/events-3.3.0.tgz",
+      "integrity": "sha512-mQw+2fkQbALzQ7V0MY0IqdnXNOeTtP4r0lN9z7AAawCXgqea7bDii20AYrIBrFd/Hx0M2Ocz6S111CaFkUcb0Q==",
+      "license": "MIT",
+      "engines": {
+        "node": ">=0.8.x"
+      }
+    },
     "node_modules/execa": {
       "version": "5.1.1",
       "dev": true,
@@ -5783,7 +5798,7 @@
     },
     "packages/editor": {
       "name": "@welshman/editor",
-      "version": "0.0.1",
+      "version": "0.0.3",
       "devDependencies": {
         "@sveltejs/kit": "^2.0.0",
         "@sveltejs/package": "^2.0.0",
@@ -5802,7 +5817,7 @@
         "@tiptap/pm": "^2.9.1",
         "@welshman/lib": "^0.0.36",
         "@welshman/util": "^0.0.53",
-        "nostr-editor": "github:cesardeazevedo/nostr-editor#069fdde",
+        "nostr-editor": "github:cesardeazevedo/nostr-editor#a211491c",
         "nostr-tools": "^2.10.4",
         "publint": "^0.2.0",
         "svelte": "^4.0.0",
@@ -5828,7 +5843,7 @@
         "@tiptap/suggestion": "^2.9.1",
         "@welshman/lib": "^0.0.36",
         "@welshman/util": "^0.0.53",
-        "nostr-editor": "github:cesardeazevedo/nostr-editor#069fdde",
+        "nostr-editor": "github:cesardeazevedo/nostr-editor#a211491c",
         "nostr-tools": "^2.8.1",
         "svelte": "^4.0.0",
         "svelte-tiptap": "^1.0.0"
@@ -6352,7 +6367,9 @@
       "version": "0.0.36",
       "license": "MIT",
       "dependencies": {
-        "@scure/base": "^1.1.6"
+        "@scure/base": "^1.1.6",
+        "@types/events": "^3.0.3",
+        "events": "^3.3.0"
       }
     },
     "packages/lib/node_modules/@scure/base": {

packages/app/src/storage.ts

@@ -2,14 +2,20 @@ import {openDB, deleteDB} from "idb"
 import type {IDBPDatabase} from "idb"
 import {writable} from "svelte/store"
 import type {Unsubscriber, Writable} from "svelte/store"
-import {indexBy, throttle, fromPairs} from "@welshman/lib"
+import {indexBy, equals, throttle, fromPairs} from "@welshman/lib"
 import type {TrustedEvent, Repository} from "@welshman/util"
 import type {Tracker} from "@welshman/net"
 import {withGetter, adapter, throttled, custom} from "@welshman/store"
 
-export type IndexedDbAdapter = {
+export type StorageAdapterOptions = {
+  throttle?: number
+  migrate?: (items: any[]) => any[]
+}
+
+export type StorageAdapter = {
   keyPath: string
   store: Writable<any[]>
+  options: StorageAdapterOptions
 }
 
 export let db: IDBPDatabase
@@ -44,41 +50,45 @@ export const bulkDelete = async (name: string, ids: string[]) => {
   await tx.done
 }
 
-export const initIndexedDbAdapter = async (name: string, adapter: IndexedDbAdapter) => {
+export const initIndexedDbAdapter = async (name: string, adapter: StorageAdapter) => {
   let prevRecords = await getAll(name)
 
   adapter.store.set(prevRecords)
 
-  adapter.store.subscribe(async (currentRecords: any[]) => {
+  setTimeout(() => {
-    if (dead.get()) {
+    adapter.store.subscribe(async (currentRecords: any[]) => {
-      return
+      if (dead.get()) {
-    }
+        return
+      }
 
-    const currentIds = new Set(currentRecords.map(item => item[adapter.keyPath]))
+      const currentIds = new Set(currentRecords.map(item => item[adapter.keyPath]))
-    const removedRecords = prevRecords.filter(r => !currentIds.has(r[adapter.keyPath]))
+      const removedRecords = prevRecords.filter(r => !currentIds.has(r[adapter.keyPath]))
 
-    const prevRecordsById = indexBy(item => item[adapter.keyPath], prevRecords)
+      const prevRecordsById = indexBy(item => item[adapter.keyPath], prevRecords)
-    const updatedRecords = currentRecords.filter(r => r !== prevRecordsById.get(r[adapter.keyPath]))
+      const updatedRecords = currentRecords.filter(
-
+        r => !equals(r, prevRecordsById.get(r[adapter.keyPath])),
-    prevRecords = currentRecords
-
-    if (updatedRecords.length > 0) {
-      await bulkPut(name, updatedRecords)
-    }
-
-    if (removedRecords.length > 0) {
-      await bulkDelete(
-        name,
-        removedRecords.map(item => item[adapter.keyPath]),
       )
-    }
+
-  })
+      prevRecords = currentRecords
+
+      if (updatedRecords.length > 0) {
+        await bulkPut(name, updatedRecords)
+      }
+
+      if (removedRecords.length > 0) {
+        await bulkDelete(
+          name,
+          removedRecords.map(item => item[adapter.keyPath]),
+        )
+      }
+    })
+  }, adapter.options.throttle || 0)
 }
 
 export const initStorage = async (
   name: string,
   version: number,
-  adapters: Record<string, IndexedDbAdapter>,
+  adapters: Record<string, StorageAdapter>,
 ) => {
   if (!window.indexedDB) return
 
@@ -124,15 +134,19 @@ export const clearStorage = async () => {
   await deleteDB(db.name)
 }
 
-export type StorageAdapterOptions = {
-  throttle?: number
-  migrate?: (items: any[]) => any[]
-}
-
 const migrate = (data: any[], options: StorageAdapterOptions) =>
   options.migrate ? options.migrate(data) : data
 
 export const storageAdapters = {
+  fromCollectionStore: <T>(
+    keyPath: string,
+    store: Writable<T[]>,
+    options: StorageAdapterOptions = {},
+  ) => ({
+    options,
+    keyPath,
+    store: throttled(options.throttle || 0, store),
+  }),
   fromObjectStore: <T>(
     store: Writable<Record<string, T>>,
     options: StorageAdapterOptions = {},

packages/app/src/sync.ts

@@ -11,6 +11,9 @@ import {
 import {repository} from "./core.js"
 import {relaysByUrl} from "./relays.js"
 
+const query = (filters: Filter[]) =>
+  repository.query(filters, {shouldSort: filters.every(f => f.limit === undefined)})
+
 export const hasNegentropy = (url: string) => {
   const p = relaysByUrl.get().get(url)?.profile
 
@@ -26,7 +29,7 @@ export type AppSyncOpts = {
 }
 
 export const pull = async ({relays, filters}: AppSyncOpts) => {
-  const events = repository.query(filters)
+  const events = query(filters)
 
   await Promise.all(
     relays.map(async relay => {
@@ -38,7 +41,7 @@ export const pull = async ({relays, filters}: AppSyncOpts) => {
 }
 
 export const push = async ({relays, filters}: AppSyncOpts) => {
-  const events = repository.query(filters).filter(isSignedEvent)
+  const events = query(filters).filter(isSignedEvent)
 
   await Promise.all(
     relays.map(async relay => {
@@ -50,7 +53,7 @@ export const push = async ({relays, filters}: AppSyncOpts) => {
 }
 
 export const sync = async ({relays, filters}: AppSyncOpts) => {
-  const events = repository.query(filters).filter(isSignedEvent)
+  const events = query(filters).filter(isSignedEvent)
 
   await Promise.all(
     relays.map(async relay => {

packages/lib/package.json

@@ -26,6 +26,8 @@
     "fix": "gts fix"
   },
   "dependencies": {
-    "@scure/base": "^1.1.6"
+    "@scure/base": "^1.1.6",
+    "@types/events": "^3.0.3",
+    "events": "^3.3.0"
   }
 }

packages/net/src/Connection.ts

@@ -47,6 +47,7 @@ export class Connection extends Emitter {
       throw new Error(`Attempted to send message on ${this.status} connection`)
     }
 
+    this.socket.open()
     this.sender.push(message)
   }
 

packages/util/src/Repository.ts

@@ -111,9 +111,13 @@ export class Repository<E extends HashedEvent = TrustedEvent> extends Emitter {
     }
   }
 
-  query = (filters: Filter[], {includeDeleted = false} = {}) => {
+  query = (filters: Filter[], {includeDeleted = false, shouldSort = true} = {}) => {
     const result: E[][] = []
     for (let filter of filters) {
+      if (filter.limit !== undefined && !shouldSort) {
+        throw new Error("Unable to skip sorting if limit is defined")
+      }
+
       let events: E[] = Array.from(this.eventsById.values())
 
       if (filter.ids) {
@@ -146,8 +150,10 @@ export class Repository<E extends HashedEvent = TrustedEvent> extends Emitter {
         }
       }
 
+      const sorted = shouldSort ? sortBy((e: E) => -e.created_at, events) : events
+
       const chunk: E[] = []
-      for (const event of sortBy((e: E) => -e.created_at, events)) {
+      for (const event of sorted) {
         if (filter.limit && chunk.length >= filter.limit) {
           break
         }