diff --git a/docs/replication/index.md b/docs/replication/index.md
index 16697fda..7b7b4453 100644
--- a/docs/replication/index.md
+++ b/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' })
+```
diff --git a/packages/signaldb/__tests__/AutoFetchCollection.spec.ts b/packages/signaldb/__tests__/AutoFetchCollection.spec.ts
new file mode 100644
index 00000000..58f4d5c5
--- /dev/null
+++ b/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([])
+})
diff --git a/packages/signaldb/__tests__/persistence.spec.ts b/packages/signaldb/__tests__/persistence.spec.ts
index c09260ec..ba093303 100644
--- a/packages/signaldb/__tests__/persistence.spec.ts
+++ b/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)
+  })
 })
diff --git a/packages/signaldb/src/AutoFetchCollection.ts b/packages/signaldb/src/AutoFetchCollection.ts
new file mode 100644
index 00000000..658d3ebe
--- /dev/null
+++ b/packages/signaldb/src/AutoFetchCollection.ts
@@ -0,0 +1,132 @@
+import type { BaseItem } from './Collection'
+import type { ReplicatedCollectionOptions } from './ReplicatedCollection'
+import ReplicatedCollection from './ReplicatedCollection'
+import type ReactivityAdapter from './types/ReactivityAdapter'
+import type Signal from './types/Signal'
+import type Selector from './types/Selector'
+import uniqueBy from './utils/uniqueBy'
+import createSignal from './utils/createSignal'
+import isEqual from './utils/isEqual'
+
+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<string, 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
+  private reactivityAdapter: ReactivityAdapter | null = null
+  private loadingSignals = new Map<string, Signal<boolean>>()
+
+  constructor(options: AutoFetchCollectionOptions<T, I, U>) {
+    let triggerRemoteChange: (() => Promise<void> | void) | undefined
+    super({
+      ...options,
+      pull: () => Promise.resolve({ items: this.itemsCache }),
+      registerRemoteChange: async (onChange) => {
+        triggerRemoteChange = onChange
+        if (options.registerRemoteChange) await options.registerRemoteChange(onChange)
+      },
+    })
+    if (!triggerRemoteChange) throw new Error('No triggerRemoteChange method found. Looks like your persistence adapter was not registered')
+    this.triggerReload = triggerRemoteChange
+    this.reactivityAdapter = options.reactivity ?? null
+
+    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(JSON.stringify(selector)) ?? 0
+    // increment the count of observers for this query
+    this.activeObservers.set(JSON.stringify(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')
+
+          // 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)
+          })
+
+          this.setLoading(selector, true)
+          this.once('persistence.received', () => {
+            this.setLoading(selector, false)
+          })
+          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(JSON.stringify(selector)) ?? 0
+    if (activeObservers > 1) {
+      // decrement the count of observers for this query
+      this.activeObservers.set(JSON.stringify(selector), activeObservers - 1)
+      return
+    }
+
+    // if this is the last observer for this query, remove the query from the cache
+    this.activeObservers.delete(JSON.stringify(selector))
+
+    // remove the query from the cache
+    this.idQueryCache.forEach((queries, id) => {
+      const updatedQueries = queries.filter(query => !isEqual(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()
+  }
+
+  private ensureSignal(selector: Selector<T>) {
+    if (!this.reactivityAdapter) throw new Error('No reactivity adapter found')
+    if (!this.loadingSignals.has(JSON.stringify(selector))) {
+      this.loadingSignals.set(
+        JSON.stringify(selector),
+        createSignal(this.reactivityAdapter.create(), false),
+      )
+    }
+    return this.loadingSignals.get(JSON.stringify(selector)) as Signal<boolean>
+  }
+
+  private setLoading(selector: Selector<T>, value: boolean) {
+    const signal = this.ensureSignal(selector)
+    signal.set(value)
+  }
+
+  public isLoading(selector?: Selector<T>) {
+    if (!selector) return super.isLoading()
+    const signal = this.ensureSignal(selector)
+    return signal.get()
+  }
+}
diff --git a/packages/signaldb/src/Collection/index.ts b/packages/signaldb/src/Collection/index.ts
index de796d7c..b117f128 100644
--- a/packages/signaldb/src/Collection/index.ts
+++ b/packages/signaldb/src/Collection/index.ts
@@ -13,6 +13,8 @@ import randomId from '../utils/randomId'
 import type { Changeset } from '../types/PersistenceAdapter'
 import executeOncePerTick from '../utils/executeOncePerTick'
 import serializeValue from '../utils/serializeValue'
+import type Signal from '../types/Signal'
+import createSignal from '../utils/createSignal'
 import Cursor from './Cursor'
 import type { BaseItem, FindOptions, Transform } from './types'
 import getIndexInfo from './getIndexInfo'
@@ -41,6 +43,13 @@ interface CollectionEvents<T extends BaseItem, U = T> {
   'persistence.error': (error: Error) => void,
   'persistence.transmitted': () => void,
   'persistence.received': () => void,
+  'persistence.pullStarted': () => void,
+  'persistence.pullCompleted': () => void,
+  'persistence.pushStarted': () => void,
+  'persistence.pushCompleted': () => 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,
@@ -91,6 +100,8 @@ export default class Collection<T extends BaseItem<I> = BaseItem, I = any, U = T
 
   private options: CollectionOptions<T, I, U>
   private persistenceAdapter: PersistenceAdapter<T, I> | null = null
+  private isPullingSignal: Signal<boolean>
+  private isPushingSignal: Signal<boolean>
   private indexProviders: (IndexProvider<T, I> | LowLevelIndexProvider<T, I>)[] = []
   private indicesOutdated = false
   private idIndex = new Map<string, Set<number>>()
@@ -106,17 +117,33 @@ export default class Collection<T extends BaseItem<I> = BaseItem, I = any, U = T
       ...(this.options.indices || []),
     ]
     this.rebuildIndices()
-    if (this.options.persistence) {
-      const persistenceAdapter = this.options.persistence
-      this.persistenceAdapter = persistenceAdapter
 
+    this.isPullingSignal = createSignal(this.options.reactivity?.create(), false)
+    this.isPushingSignal = createSignal(this.options.reactivity?.create(), false)
+    this.on('persistence.pullStarted', () => {
+      this.isPullingSignal?.set(true)
+    })
+    this.on('persistence.pullCompleted', () => {
+      this.isPullingSignal?.set(false)
+    })
+    this.on('persistence.pushStarted', () => {
+      this.isPushingSignal?.set(true)
+    })
+    this.on('persistence.pushCompleted', () => {
+      this.isPushingSignal?.set(false)
+    })
+
+    this.persistenceAdapter = this.options.persistence ?? null
+    if (this.persistenceAdapter) {
       let ongoingSaves = 0
       let isInitialized = false
       const pendingUpdates: Changeset<T> = { added: [], modified: [], removed: [] }
 
       const loadPersistentData = async () => {
+        if (!this.persistenceAdapter) throw new Error('Persistence adapter not found')
+        this.emit('persistence.pullStarted')
         // load items from persistence adapter and push them into memory
-        const { items, changes } = await persistenceAdapter.load()
+        const { items, changes } = await this.persistenceAdapter.load()
 
         if (items) {
           // as we overwrite all items, we need to discard if there are ongoing saves
@@ -149,6 +176,7 @@ export default class Collection<T extends BaseItem<I> = BaseItem, I = any, U = T
         this.rebuildIndices()
 
         this.emit('persistence.received')
+        this.emit('persistence.pullCompleted')
       }
 
       const saveQueue = {
@@ -158,6 +186,8 @@ export default class Collection<T extends BaseItem<I> = BaseItem, I = any, U = T
       } as Changeset<T>
       let isFlushing = false
       const flushQueue = () => {
+        if (!this.persistenceAdapter) throw new Error('Persistence adapter not found')
+        if (ongoingSaves <= 0) this.emit('persistence.pushStarted')
         if (isFlushing) return
         if (!hasPendingUpdates(saveQueue)) return
         isFlushing = true
@@ -167,7 +197,7 @@ export default class Collection<T extends BaseItem<I> = BaseItem, I = any, U = T
         saveQueue.added = []
         saveQueue.modified = []
         saveQueue.removed = []
-        persistenceAdapter.save(currentItems, changes)
+        this.persistenceAdapter.save(currentItems, changes)
           .then(() => {
             this.emit('persistence.transmitted')
           }).catch((error) => {
@@ -176,6 +206,7 @@ export default class Collection<T extends BaseItem<I> = BaseItem, I = any, U = T
             ongoingSaves -= 1
             isFlushing = false
             flushQueue()
+            if (ongoingSaves <= 0) this.emit('persistence.pushCompleted')
           })
       }
 
@@ -204,8 +235,9 @@ export default class Collection<T extends BaseItem<I> = BaseItem, I = any, U = T
         flushQueue()
       })
 
-      persistenceAdapter.register(() => loadPersistentData())
+      this.persistenceAdapter.register(() => loadPersistentData())
         .then(async () => {
+          if (!this.persistenceAdapter) throw new Error('Persistence adapter not found')
           let currentItems = this.memoryArray()
           await loadPersistentData()
           while (hasPendingUpdates(pendingUpdates)) {
@@ -214,7 +246,7 @@ export default class Collection<T extends BaseItem<I> = BaseItem, I = any, U = T
             const removed = pendingUpdates.removed.splice(0)
             currentItems = applyUpdates(currentItems, { added, modified, removed })
             // eslint-disable-next-line no-await-in-loop
-            await persistenceAdapter.save(currentItems, { added, modified, removed })
+            await this.persistenceAdapter.save(currentItems, { added, modified, removed })
               .then(() => {
                 this.emit('persistence.transmitted')
               })
@@ -230,6 +262,20 @@ export default class Collection<T extends BaseItem<I> = BaseItem, I = any, U = T
     }
   }
 
+  public isPulling() {
+    return this.isPullingSignal.get() ?? false
+  }
+
+  public isPushing() {
+    return this.isPushingSignal.get() ?? false
+  }
+
+  public isLoading() {
+    const isPulling = this.isPulling()
+    const isPushing = this.isPushing()
+    return isPulling || isPushing
+  }
+
   public getDebugMode() {
     return this.debugMode
   }
@@ -338,11 +384,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)
         }
       },
     })
diff --git a/packages/signaldb/src/createReactivityAdapter.ts b/packages/signaldb/src/createReactivityAdapter.ts
index 0ade3fdc..ef8b39b4 100644
--- a/packages/signaldb/src/createReactivityAdapter.ts
+++ b/packages/signaldb/src/createReactivityAdapter.ts
@@ -1,5 +1,5 @@
+import type Dependency from './types/Dependency'
 import type ReactivityAdapter from './types/ReactivityAdapter'
-import type { Dependency } from './types/ReactivityAdapter'
 
 export default function createReactivityAdapter<T extends Dependency = Dependency>(
   definition: ReactivityAdapter<T>,
diff --git a/packages/signaldb/src/index.ts b/packages/signaldb/src/index.ts
index 8ae5a21a..cdde66c7 100644
--- a/packages/signaldb/src/index.ts
+++ b/packages/signaldb/src/index.ts
@@ -13,6 +13,7 @@ export type {
 export { default as Collection, createIndex } from './Collection'
 export { default as PersistentCollection } from './PersistentCollection'
 export { default as ReplicatedCollection } from './ReplicatedCollection'
+export { default as AutoFetchCollection } from './AutoFetchCollection'
 export { default as createLocalStorageAdapter } from './persistence/createLocalStorageAdapter'
 export { default as createFilesystemAdapter } from './persistence/createFilesystemAdapter'
 export { default as createPersistenceAdapter } from './persistence/createPersistenceAdapter'
diff --git a/packages/signaldb/src/types/Dependency.ts b/packages/signaldb/src/types/Dependency.ts
new file mode 100644
index 00000000..bc40d714
--- /dev/null
+++ b/packages/signaldb/src/types/Dependency.ts
@@ -0,0 +1,4 @@
+export default interface Dependency {
+  depend(): void,
+  notify(): void,
+}
diff --git a/packages/signaldb/src/types/ReactivityAdapter.ts b/packages/signaldb/src/types/ReactivityAdapter.ts
index 3666e1f2..fc9020be 100644
--- a/packages/signaldb/src/types/ReactivityAdapter.ts
+++ b/packages/signaldb/src/types/ReactivityAdapter.ts
@@ -1,7 +1,4 @@
-export interface Dependency {
-  depend(): void,
-  notify(): void,
-}
+import type Dependency from './Dependency'
 
 export default interface ReactivityAdapter<T extends Dependency = Dependency> {
   create(): T,
diff --git a/packages/signaldb/src/types/Signal.ts b/packages/signaldb/src/types/Signal.ts
new file mode 100644
index 00000000..0d2203fa
--- /dev/null
+++ b/packages/signaldb/src/types/Signal.ts
@@ -0,0 +1,4 @@
+export default interface Signal<T> {
+  get(): T,
+  set(value: T): void,
+}
diff --git a/packages/signaldb/src/utils/createSignal.spec.ts b/packages/signaldb/src/utils/createSignal.spec.ts
new file mode 100644
index 00000000..b66c87f0
--- /dev/null
+++ b/packages/signaldb/src/utils/createSignal.spec.ts
@@ -0,0 +1,62 @@
+// @vitest-environment happy-dom
+import { vi, it, expect } from 'vitest'
+import type Dependency from '../types/Dependency'
+import createSignal from './createSignal'
+
+it('should return the initial value', () => {
+  const dependency = {
+    depend: vi.fn(),
+    notify: vi.fn(),
+  } as Dependency
+  const signal = createSignal(dependency, 'initialValue')
+  expect(signal.get()).toBe('initialValue')
+  expect(dependency.depend).toHaveBeenCalled()
+  expect(dependency.notify).not.toHaveBeenCalled()
+})
+
+it('should update the value', () => {
+  const dependency = {
+    depend: vi.fn(),
+    notify: vi.fn(),
+  } as Dependency
+  const signal = createSignal(dependency, 'initialValue')
+  signal.set('newValue')
+  expect(dependency.notify).toHaveBeenCalled()
+  expect(signal.get()).toBe('newValue')
+  expect(dependency.depend).toHaveBeenCalled()
+})
+
+it('should not notify dependency if value is equal', () => {
+  const dependency = {
+    depend: vi.fn(),
+    notify: vi.fn(),
+  } as Dependency
+  const signal = createSignal(dependency, 'initialValue')
+  signal.set('initialValue')
+  expect(dependency.depend).not.toHaveBeenCalled()
+  expect(dependency.notify).not.toHaveBeenCalled()
+})
+
+it('should notify dependency if value is not equal', () => {
+  const dependency = {
+    depend: vi.fn(),
+    notify: vi.fn(),
+  } as Dependency
+  const signal = createSignal(dependency, 'initialValue')
+  signal.set('newValue')
+  expect(dependency.depend).not.toHaveBeenCalled()
+  expect(dependency.notify).toHaveBeenCalled()
+})
+
+it('should use custom isEqual function', () => {
+  const dependency = {
+    depend: vi.fn(),
+    notify: vi.fn(),
+  } as Dependency
+  const isEqual = (a: string, b: string) => a.toLowerCase() === b.toLowerCase()
+  const signal = createSignal<string>(dependency, 'initialValue', isEqual)
+  signal.set('INITIALVALUE')
+  expect(dependency.notify).not.toHaveBeenCalled()
+  expect(signal.get()).toBe('initialValue')
+  expect(dependency.depend).toHaveBeenCalled()
+})
diff --git a/packages/signaldb/src/utils/createSignal.ts b/packages/signaldb/src/utils/createSignal.ts
new file mode 100644
index 00000000..f41420a7
--- /dev/null
+++ b/packages/signaldb/src/utils/createSignal.ts
@@ -0,0 +1,22 @@
+import type Dependency from '../types/Dependency'
+import type Signal from '../types/Signal'
+
+export default function createSignal<T>(
+  dependency: Dependency | undefined,
+  initialValue: T,
+  isEqual: (a: T, b: T) => boolean = Object.is,
+) {
+  let value = initialValue
+  const signal: Signal<T> = {
+    get() {
+      if (dependency) dependency.depend()
+      return value
+    },
+    set(newValue) {
+      if (isEqual(value, newValue)) return
+      value = newValue
+      if (dependency) dependency.notify()
+    },
+  }
+  return signal
+}
diff --git a/packages/signaldb/src/utils/uniqueBy.spec.ts b/packages/signaldb/src/utils/uniqueBy.spec.ts
new file mode 100644
index 00000000..7819c507
--- /dev/null
+++ b/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])
+})
diff --git a/packages/signaldb/src/utils/uniqueBy.ts b/packages/signaldb/src/utils/uniqueBy.ts
new file mode 100644
index 00000000..552e719f
--- /dev/null
+++ b/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)
+  })
+}