Implement Auto Fetch Collections

docs/replication/index.md

@@ -8,7 +8,10 @@ head:
 
 For seamless integration of your app with remote services, SignalDB offers robust data replication capabilities. Whether you're building a local app or sharing data across multiple clients, SignalDB's modular replication system ensures efficient data synchronization.
 
-Central to SignalDB's replication functionality is the `ReplicatedCollection` class. This specialized class streamlines the replication process, allowing you to effortlessly replicate data to any remote service.
+
+## `ReplicatedCollection`
+
+Central to SignalDB's replication functionality is the `ReplicatedCollection` class. This specialized class streamlines the replication process, allowing you to effortlessly replicate data to any remote service. It inherits from the `Collection` class, so you can use it just like any other collection.
 
 The usage of the `ReplicatedCollection` is really simple:
 
@@ -41,3 +44,48 @@ const Todos = new ReplicatedCollection({
   persistence: createLocalStorageAdapter('todos'),
 })
 ```
+
+## `AutoFetchCollection`
+
+The `AutoFetchCollection` class is a specialized variant of the `ReplicatedCollection` that automatically fetches data from the remote service when the collection is accessed. This is useful if you want to fetch specific data on demand rather than pulling the whole dataset at app start.
+
+The concept of the `AutoFetchCollection` is, that it calls the `fetchQueryItems` method everytime a query is executed on the collection. This way, you can fetch only the data that is needed for the query. The first time the query is executed, the query will return a empty dataset (if the data is not already fetched). After the data is fetched, the query will reactively update and return the loaded data.
+While the data is fetched, the you can observe the loading state with the `isLoading` function on the collection to show a loading indicator. The `ìsLoading` function will be updated reactively.
+
+The usage of the `AutoFetchCollection` is also really simple:
+
+```js
+const Todos = new AutoFetchCollection({
+  fetchQueryItems: async (selector) => {
+    // The fetchQueryItems method is for fetching data from the remote service.
+    // The selector parameter is the query that is executed on the collection.
+    // Use this to fetch only the data that is needed for the query.
+    // Also make sure that the returned data matches the query to avoid inconsistencies
+    // The return value is similar to one of the pull method of the ReplicatedCollection,
+
+    // You can return the data directly
+    // return { items: [...] }
+
+    // Or you can return only the changes
+    // return { changes: { added: [...], modified: [...], removed: [...] } }
+  },
+  push: async (changes, items) => {
+    // The push method is the same as in the ReplicatedCollection
+    // The push method is called when the local data has changed
+    // As the first parameter you get the changes in the format { added: [...], modified: [...], removed: [...] }
+    // As the second parameter you also get all items in the collection, if you need them
+    // in the push method, no return value is expected
+  },
+
+  // Like in the ReplicatedCollection, you can also optionally specify a persistence adapter
+  // If a persistence adapter is used, the data is loaded first and will be updated after the server data is fetched
+  // If the data will be updated, the data will be saved to the persistence adapter and pushed to the server simultaneously
+  persistence: createLocalStorageAdapter('todos'),
+})
+
+// You can also observe the loading state of the collection.
+const loading = Todos.isLoading()
+
+// The isLoading method takes an optional selector parameter to observe the loading state of a specific query
+const postsFromMaxLoading = Todos.isLoading({ author: 'Max' })
+```

packages/signaldb/__tests__/AutoFetchCollection.spec.ts

@@ -0,0 +1,180 @@
+import { vi, it, expect } from 'vitest'
+import { AutoFetchCollection, createReactivityAdapter } from '../src'
+import waitForEvent from './helpers/waitForEvent'
+
+it('should fetch query items when observer is created', async () => {
+  const fetchQueryItems = vi.fn()
+  const reactivity = {
+    create: () => ({
+      depend: vi.fn(),
+      notify: vi.fn(),
+    }),
+  }
+  const collection = new AutoFetchCollection({
+    push: vi.fn(),
+    fetchQueryItems,
+    reactivity,
+  })
+
+  // Mock fetchQueryItems response
+  const response = {
+    items: [{ id: 1, name: 'Item 1' }, { id: 2, name: 'Item 2' }],
+  }
+  fetchQueryItems.mockResolvedValue(response)
+
+  expect(collection.find({}).fetch()).toEqual([])
+  await waitForEvent(collection, 'persistence.received')
+
+  // Wait for fetchQueryItems to be called
+  await vi.waitFor(() => expect(fetchQueryItems).toBeCalledTimes(1))
+  await vi.waitFor(() => expect(collection.isLoading({})).toBe(false))
+  expect(collection.find({}, { reactive: false }).fetch()).toEqual(response.items)
+})
+
+it('should remove query when observer is disposed', async () => {
+  const fetchQueryItems = vi.fn()
+  const disposalCallbacks: (() => void)[] = []
+  const disposeAll = () => disposalCallbacks.forEach(callback => callback())
+  const reactivity = createReactivityAdapter({
+    create: () => ({
+      depend: vi.fn(),
+      notify: vi.fn(),
+    }),
+    onDispose(callback) {
+      disposalCallbacks.push(callback)
+    },
+  })
+  const collection = new AutoFetchCollection({
+    push: vi.fn(),
+    fetchQueryItems,
+    reactivity,
+  })
+
+  // Mock fetchQueryItems response
+  const response = {
+    items: [{ id: 1, name: 'Item 1' }, { id: 2, name: 'Item 2' }],
+  }
+  fetchQueryItems.mockResolvedValue(response)
+
+  expect(collection.find({}).fetch()).toEqual([])
+  await waitForEvent(collection, 'persistence.received')
+
+  // Wait for fetchQueryItems to be called
+  await vi.waitFor(() => expect(fetchQueryItems).toBeCalledTimes(1))
+  expect(collection.find({}).fetch()).toEqual(response.items)
+
+  disposeAll()
+  await waitForEvent(collection, 'persistence.received')
+  expect(collection.find({}, { reactive: false }).fetch()).toEqual([])
+})
+
+it('should trigger persistence.error event when fetchQueryItems fails', async () => {
+  const fetchQueryItems = vi.fn()
+  const reactivity = createReactivityAdapter({
+    create: () => ({
+      depend: vi.fn(),
+      notify: vi.fn(),
+    }),
+  })
+  const collection = new AutoFetchCollection({
+    push: vi.fn(),
+    fetchQueryItems,
+    reactivity,
+  })
+
+  const error = new Error('Failed to fetch query items')
+  fetchQueryItems.mockRejectedValue(error)
+
+  expect(collection.find({}).fetch()).toEqual([])
+
+  await waitForEvent(collection, 'persistence.error')
+  expect(collection.find({}, { reactive: false }).fetch()).toEqual([])
+})
+
+it('should handle multiple observers for the same query', async () => {
+  const fetchQueryItems = vi.fn()
+  const disposalCallbacks: (() => void)[] = []
+  const disposeAll = () => disposalCallbacks.forEach(callback => callback())
+  const reactivity = createReactivityAdapter({
+    create: () => ({
+      depend: vi.fn(),
+      notify: vi.fn(),
+    }),
+    onDispose(callback) {
+      disposalCallbacks.push(callback)
+    },
+  })
+  const collection = new AutoFetchCollection({
+    push: vi.fn(),
+    fetchQueryItems,
+    reactivity,
+  })
+
+  // Mock fetchQueryItems response
+  const response = {
+    items: [{ id: 1, name: 'Item 1' }, { id: 2, name: 'Item 2' }],
+  }
+  fetchQueryItems.mockResolvedValue(response)
+
+  expect(collection.find({}).fetch()).toEqual([])
+  expect(collection.find({}).fetch()).toEqual([])
+  expect(collection.find({}).fetch()).toEqual([])
+  expect(collection.find({}).fetch()).toEqual([])
+  expect(collection.find({}).fetch()).toEqual([])
+  await waitForEvent(collection, 'persistence.received')
+
+  // Wait for fetchQueryItems to be called
+  await vi.waitFor(() => expect(fetchQueryItems).toBeCalledTimes(1))
+  expect(collection.find({}).fetch()).toEqual(response.items)
+
+  disposeAll()
+  await waitForEvent(collection, 'persistence.received')
+  expect(collection.find({}, { reactive: false }).fetch()).toEqual([])
+})
+
+it('should handle multiple queriey', async () => {
+  const fetchQueryItems = vi.fn()
+  const disposalCallbacks: (() => void)[] = []
+  const disposeAll = () => disposalCallbacks.forEach(callback => callback())
+  const reactivity = createReactivityAdapter({
+    create: () => ({
+      depend: vi.fn(),
+      notify: vi.fn(),
+    }),
+    onDispose(callback) {
+      disposalCallbacks.push(callback)
+    },
+  })
+  const collection = new AutoFetchCollection({
+    push: vi.fn(),
+    fetchQueryItems,
+    reactivity,
+  })
+
+  const responseAllItems = {
+    items: [{ id: 1, name: 'Item 1' }, { id: 2, name: 'Item 2' }],
+  }
+  const responseFilteredItems = {
+    items: [{ id: 1, name: 'Item 1' }],
+  }
+  fetchQueryItems.mockImplementation((selector) => {
+    if (selector.name) return Promise.resolve(responseFilteredItems)
+    return Promise.resolve(responseAllItems)
+  })
+
+  expect(collection.find({ name: 'Item 1' }).fetch()).toEqual([])
+  expect(fetchQueryItems).toBeCalledWith({ name: 'Item 1' })
+  await waitForEvent(collection, 'persistence.received')
+  expect(fetchQueryItems).toBeCalledTimes(1)
+  expect(collection.find({}).fetch()).toEqual(responseFilteredItems.items)
+
+  expect(fetchQueryItems).toBeCalledWith({})
+  expect(fetchQueryItems).toBeCalledTimes(2)
+  await waitForEvent(collection, 'persistence.received')
+  await new Promise((resolve) => { setTimeout(resolve, 100) }) // wait a bit to ensure fetchQueryItems cache was updated
+  expect(collection.find({}, { reactive: false }).fetch()).toEqual(responseAllItems.items)
+
+  disposeAll()
+  await waitForEvent(collection, 'persistence.received')
+  expect(collection.find({}, { reactive: false }).fetch()).toEqual([])
+})

packages/signaldb/__tests__/persistence.spec.ts

@@ -332,4 +332,52 @@ describe('Persistence', () => {
 
     expect(fn).toHaveBeenCalledWith(new Error('test'))
   })
+
+  it('should emit all required events', async () => {
+    const persistence = memoryPersistenceAdapter([{ id: '1', name: 'John' }])
+    const collection = new Collection({ persistence })
+    await Promise.all([
+      waitForEvent(collection, 'persistence.pullStarted'),
+      waitForEvent(collection, 'persistence.received'),
+      waitForEvent(collection, 'persistence.pullCompleted'),
+      waitForEvent(collection, 'persistence.init'),
+    ])
+
+    collection.updateOne({ id: '1' }, { $set: { name: 'Johnny' } })
+    await Promise.all([
+      waitForEvent(collection, 'persistence.pushStarted'),
+      waitForEvent(collection, 'persistence.pushCompleted'),
+      waitForEvent(collection, 'persistence.transmitted'),
+    ])
+
+    const items = collection.find().fetch()
+    expect(items).toEqual([{ id: '1', name: 'Johnny' }])
+    expect((await persistence.load()).items).toEqual([{ id: '1', name: 'Johnny' }])
+  })
+
+  it('should return correct values from isPulling, isPushing and isLoading', async () => {
+    const persistence = memoryPersistenceAdapter([{ id: '1', name: 'John' }])
+    const collection = new Collection({ persistence })
+
+    const pullStarted = waitForEvent(collection, 'persistence.pullStarted')
+    const pullCompleted = waitForEvent(collection, 'persistence.pullCompleted')
+    const initialized = waitForEvent(collection, 'persistence.init')
+    await pullStarted
+    expect(collection.isPulling()).toBe(true)
+    expect(collection.isLoading()).toBe(true)
+    await pullCompleted
+    expect(collection.isPulling()).toBe(false)
+    expect(collection.isLoading()).toBe(false)
+    await initialized
+
+    const pushStarted = waitForEvent(collection, 'persistence.pushStarted')
+    const pushCompleted = waitForEvent(collection, 'persistence.pushCompleted')
+    collection.updateOne({ id: '1' }, { $set: { name: 'Johnny' } })
+    await pushStarted
+    expect(collection.isPushing()).toBe(true)
+    expect(collection.isLoading()).toBe(true)
+    await pushCompleted
+    expect(collection.isPushing()).toBe(false)
+    expect(collection.isLoading()).toBe(false)
+  })
 })