feat: add first draft of auto fetch collection class

resolves #394

packages/signaldb/src/AutoFetchCollection.ts

@@ -0,0 +1,101 @@
+import type { BaseItem } from './Collection'
+import type { ReplicatedCollectionOptions } from './ReplicatedCollection'
+import ReplicatedCollection from './ReplicatedCollection'
+import type Selector from './types/Selector'
+import uniqueBy from './utils/uniqueBy'
+
+interface AutoFetchOptions<T extends { id: I } & Record<string, any>, I> {
+  fetchQueryItems: (selector: Selector<T>) => ReturnType<ReplicatedCollectionOptions<T, I>['pull']>,
+}
+export type AutoFetchCollectionOptions<
+  T extends BaseItem<I>,
+  I,
+  U = T,
+> = Omit<ReplicatedCollectionOptions<T, I, U>, 'pull'> & AutoFetchOptions<T, I>
+
+export default class AutoFetchCollection<
+  T extends BaseItem<I> = BaseItem,
+  I = any,
+  U = T,
+> extends ReplicatedCollection<T, I, U> {
+  private activeObservers = new Map<Selector<T>, number>()
+  private idQueryCache = new Map<I, Selector<T>[]>()
+  private itemsCache: T[] = []
+  private fetchQueryItems: (selector: Selector<T>) => ReturnType<ReplicatedCollectionOptions<T, I>['pull']>
+  private triggerReload: null | (() => void | Promise<void>) = null
+
+  constructor(options: AutoFetchCollectionOptions<T, I, U>) {
+    super({
+      ...options,
+      pull: () => Promise.resolve({ items: this.itemsCache }),
+      registerRemoteChange: (onChange) => {
+        this.triggerReload = onChange
+        return Promise.resolve()
+      },
+    })
+
+    this.fetchQueryItems = options.fetchQueryItems
+    this.on('observer.created', selector => this.handleObserverCreation(selector ?? {}))
+    this.on('observer.disposed', selector => this.handleObserverDisposal(selector ?? {}))
+  }
+
+  private handleObserverCreation(selector: Selector<T>) {
+    const activeObservers = this.activeObservers.get(selector) ?? 0
+    // increment the count of observers for this query
+    this.activeObservers.set(selector, activeObservers + 1)
+
+    // if this is the first observer for this query, fetch the data
+    if (activeObservers === 0) {
+      this.fetchQueryItems(selector)
+        .then((response) => {
+          if (!response.items) throw new Error('AutoFetchCollection currently only works with a full item response')
+
+          if (!this.itemsCache) {
+            // if this is the first response, cache it
+            this.itemsCache = response.items
+          } else if (response.items && this.itemsCache) {
+            // merge the response into the cache
+            this.itemsCache = uniqueBy([...this.itemsCache, ...response.items], 'id')
+          }
+
+          response.items.forEach((item) => {
+            const queries = this.idQueryCache.get(item.id) ?? []
+            queries.push(selector)
+            this.idQueryCache.set(item.id, queries)
+          })
+
+          if (!this.triggerReload) throw new Error('No triggerReload method found. Looks like your persistence adapter was not registered')
+          void this.triggerReload()
+        })
+        .catch((error: Error) => {
+          this.emit('persistence.error', error)
+        })
+    }
+  }
+
+  private handleObserverDisposal(selector: Selector<T>) {
+    const activeObservers = this.activeObservers.get(selector) ?? 0
+    if (activeObservers > 1) {
+      // decrement the count of observers for this query
+      this.activeObservers.set(selector, activeObservers - 1)
+      return
+    }
+
+    // if this is the last observer for this query, remove the query from the cache
+    this.activeObservers.delete(selector)
+
+    // remove the query from the cache
+    this.idQueryCache.forEach((queries, id) => {
+      const updatedQueries = queries.filter(query => query !== selector)
+      if (updatedQueries.length === 0) {
+        this.idQueryCache.delete(id)
+        this.itemsCache = this.itemsCache.filter(item => item.id !== id)
+      } else {
+        this.idQueryCache.set(id, updatedQueries)
+      }
+    })
+
+    if (!this.triggerReload) throw new Error('No triggerReload method found. Looks like your persistence adapter was not registered')
+    void this.triggerReload()
+  }
+}

packages/signaldb/src/Collection/index.ts

@@ -42,6 +42,9 @@ interface CollectionEvents<T extends BaseItem, U = T> {
   'persistence.transmitted': () => void,
   'persistence.received': () => void,
 
+  'observer.created': <O extends FindOptions<T>>(selector?: Selector<T>, options?: O) => void,
+  'observer.disposed': <O extends FindOptions<T>>(selector?: Selector<T>, options?: O) => void,
+
   '_debug.getItems': (callstack: string, selector: Selector<T> | undefined, measuredTime: number) => void,
   '_debug.find': <O extends FindOptions<T>>(callstack: string, selector: Selector<T> | undefined, options: O | undefined, cursor: Cursor<T, U>) => void,
   '_debug.findOne': <O extends FindOptions<T>>(callstack: string, selector: Selector<T>, options: O | undefined, item: U | undefined) => void,
@@ -338,11 +341,13 @@ export default class Collection<T extends BaseItem<I> = BaseItem, I = any, U = T
         this.addListener('added', requeryOnce)
         this.addListener('changed', requeryOnce)
         this.addListener('removed', requeryOnce)
+        this.emit('observer.created', selector, options)
         return () => {
           this.removeListener('persistence.received', requeryOnce)
           this.removeListener('added', requeryOnce)
           this.removeListener('changed', requeryOnce)
           this.removeListener('removed', requeryOnce)
+          this.emit('observer.disposed', selector, options)
         }
       },
     })

packages/signaldb/src/utils/uniqueBy.spec.ts

@@ -0,0 +1,49 @@
+import { it, expect } from 'vitest'
+import uniqueBy from './uniqueBy'
+
+it('uniqueBy should return an array with unique items based on the provided key', () => {
+  const arr = [
+    { id: 1, name: 'John' },
+    { id: 2, name: 'Jane' },
+    { id: 3, name: 'John' },
+    { id: 4, name: 'Jane' },
+  ]
+
+  const result = uniqueBy(arr, 'name')
+  expect(result).toEqual([
+    { id: 1, name: 'John' },
+    { id: 2, name: 'Jane' },
+  ])
+})
+
+it('uniqueBy should return an array with unique items based on the provided function', () => {
+  const arr = [
+    { id: 1, name: 'John' },
+    { id: 2, name: 'Jane' },
+    { id: 3, name: 'John' },
+    { id: 4, name: 'Jane' },
+  ]
+
+  const result = uniqueBy(arr, item => item.name)
+  expect(result).toEqual([
+    { id: 1, name: 'John' },
+    { id: 2, name: 'Jane' },
+  ])
+})
+
+it('uniqueBy should preserve the order of the unique items', () => {
+  const arr = [
+    { id: 1, name: 'John' },
+    { id: 2, name: 'Jane' },
+    { id: 3, name: 'John' },
+    { id: 4, name: 'Jane' },
+  ]
+
+  const result = uniqueBy(arr, 'name')
+  expect(result).toEqual([
+    { id: 1, name: 'John' },
+    { id: 2, name: 'Jane' },
+  ])
+  expect(result[0]).toBe(arr[0])
+  expect(result[1]).toBe(arr[1])
+})

packages/signaldb/src/utils/uniqueBy.ts

@@ -0,0 +1,7 @@
+export default function uniqueBy<T>(arr: T[], fn: keyof T | ((item: T) => any)) {
+  const set = new Set<any>()
+  return arr.filter((el) => {
+    const value = typeof fn === 'function' ? fn(el) : el[fn]
+    return !set.has(value) && set.add(value)
+  })
+}