feat: add first draft of replicated collection class

packages/signaldb/__tests__/ReplicatedCollection.spec.ts

@@ -0,0 +1,124 @@
+import { vi, it, describe, expect, beforeAll } from 'vitest'
+import type { ReplicatedCollectionOptions } from '../src/ReplicatedCollection'
+import ReplicatedCollection, { createReplicationAdapter } from '../src/ReplicatedCollection'
+import type { BaseItem } from '../src/Collection'
+import Collection from '../src/Collection'
+import type { Changeset, LoadResponse } from '../src/types/PersistenceAdapter'
+import createPersistenceAdapter from '../src/persistence/createPersistenceAdapter'
+import waitForEvent from './helpers/waitForEvent'
+
+describe('createReplicationAdapter', () => {
+  it('should call handleRemoteChange and register onChange', async () => {
+    const pull = vi.fn().mockResolvedValue({} as LoadResponse<any>)
+    const push = vi.fn().mockResolvedValue(undefined)
+    const handleRemoteChange = vi.fn().mockResolvedValue(undefined)
+    const onChange = vi.fn()
+
+    const options = {
+      pull,
+      push,
+      handleRemoteChange,
+    }
+
+    const adapter = createReplicationAdapter(options)
+    await adapter.register(onChange)
+
+    expect(handleRemoteChange).toHaveBeenCalledTimes(1)
+    expect(handleRemoteChange).toHaveBeenCalledWith(onChange)
+  })
+
+  it('should not call handleRemoteChange if it is not provided', async () => {
+    const pull = vi.fn().mockResolvedValue({} as LoadResponse<any>)
+    const push = vi.fn().mockResolvedValue(undefined)
+    const onChange = vi.fn()
+
+    const options = {
+      pull,
+      push,
+    }
+
+    const adapter = createReplicationAdapter(options)
+    await adapter.register(onChange)
+
+    expect(onChange).toHaveBeenCalledTimes(0)
+  })
+
+  it('should call pull when load is called', async () => {
+    const pull = vi.fn().mockResolvedValue({} as LoadResponse<any>)
+    const push = vi.fn().mockResolvedValue(undefined)
+
+    const options = {
+      pull,
+      push,
+    }
+
+    const adapter = createReplicationAdapter(options)
+    await adapter.load()
+
+    expect(pull).toHaveBeenCalledTimes(1)
+  })
+
+  it('should call push when save is called', async () => {
+    const pull = vi.fn().mockResolvedValue({} as LoadResponse<any>)
+    const push = vi.fn().mockResolvedValue(undefined)
+    const items = [{ id: 1, name: 'Item 1' }]
+    const changes: Changeset<any> = { added: [], modified: [], removed: [] }
+
+    const options = {
+      pull,
+      push,
+    }
+
+    const adapter = createReplicationAdapter(options)
+    // eslint-disable-next-line @typescript-eslint/no-unsafe-argument
+    await adapter.save(items, changes)
+
+    expect(push).toHaveBeenCalledTimes(1)
+    expect(push).toHaveBeenCalledWith(changes, items)
+  })
+})
+
+describe('ReplicatedCollection', () => {
+  beforeAll(() => {
+    vi.useFakeTimers()
+  })
+
+  it('should create a ReplicatedCollection instance', () => {
+    const options: ReplicatedCollectionOptions<BaseItem<number>, number> = {
+      pull: vi.fn().mockResolvedValue({} as LoadResponse<BaseItem<number>>),
+      push: vi.fn().mockResolvedValue(undefined),
+    }
+
+    const collection = new ReplicatedCollection(options)
+
+    expect(collection).toBeInstanceOf(ReplicatedCollection)
+    expect(collection).toBeInstanceOf(Collection)
+  })
+
+  it('should combine persistence and replication adapters', async () => {
+    const pull = vi.fn().mockResolvedValue({ items: [] } as LoadResponse<any>)
+    const push = vi.fn().mockResolvedValue(undefined)
+    const handleRemoteChange = vi.fn().mockResolvedValue(undefined)
+
+    const persistenceAdapter = createPersistenceAdapter({
+      register: vi.fn().mockResolvedValue(undefined),
+      load: vi.fn().mockResolvedValue({ items: [] } as LoadResponse<any>),
+      save: vi.fn().mockResolvedValue(undefined),
+    })
+
+    const collection = new ReplicatedCollection({
+      pull,
+      push,
+      handleRemoteChange,
+      persistence: persistenceAdapter,
+    })
+    expect(collection).toBeInstanceOf(ReplicatedCollection)
+    await waitForEvent(collection, 'persistence.init')
+    expect(persistenceAdapter.register).toHaveBeenCalledTimes(1)
+    expect(persistenceAdapter.load).toHaveBeenCalledTimes(1)
+    expect(pull).toHaveBeenCalledTimes(1)
+    expect(persistenceAdapter.save).toHaveBeenCalledTimes(0)
+    expect(push).toHaveBeenCalledTimes(0)
+    expect(handleRemoteChange).toHaveBeenCalledTimes(1)
+  })
+})

packages/signaldb/src/ReplicatedCollection.ts

@@ -0,0 +1,46 @@
+import Collection from './Collection'
+import type { BaseItem, CollectionOptions } from './Collection'
+import type { Changeset, LoadResponse } from './types/PersistenceAdapter'
+import combinePersistenceAdapters from './persistence/combinePersistenceAdapters'
+import createPersistenceAdapter from './persistence/createPersistenceAdapter'
+
+interface ReplicationOptions<T extends { id: I } & Record<string, any>, I> {
+  pull: () => Promise<LoadResponse<T>>,
+  push(changes: Changeset<T>, items: T[]): Promise<void>,
+  handleRemoteChange?: (onChange: () => void | Promise<void>) => Promise<void>,
+}
+export function createReplicationAdapter<T extends { id: I } & Record<string, any>, I>(
+  options: ReplicationOptions<T, I>,
+) {
+  return createPersistenceAdapter({
+    async register(onChange) {
+      if (!options.handleRemoteChange) return
+      await options.handleRemoteChange(onChange)
+    },
+    load: () => options.pull(),
+    save: (items, changes) => options.push(changes, items),
+  })
+}
+
+export type ReplicatedCollectionOptions<
+  T extends BaseItem<I>,
+  I,
+  U = T,
+> = CollectionOptions<T, I, U> & ReplicationOptions<T, I>
+
+export default class ReplicatedCollection<
+  T extends BaseItem<I> = BaseItem,
+  I = any,
+  U = T,
+> extends Collection<T, I, U> {
+  constructor(options: ReplicatedCollectionOptions<T, I, U>) {
+    const replicationAdapter = createReplicationAdapter(options)
+    const persistenceAdapter = options?.persistence
+      ? combinePersistenceAdapters(replicationAdapter, options.persistence)
+      : replicationAdapter
+    super({
+      ...options,
+      persistence: persistenceAdapter,
+    })
+  }
+}

packages/signaldb/src/index.ts

@@ -12,6 +12,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 createLocalStorageAdapter } from './persistence/createLocalStorageAdapter'
 export { default as createFilesystemAdapter } from './persistence/createFilesystemAdapter'
 export { default as createPersistenceAdapter } from './persistence/createPersistenceAdapter'

packages/signaldb/src/types/PersistenceAdapter.ts

@@ -4,7 +4,13 @@ export interface Changeset<T> {
   removed: T[],
 }
 
-type LoadResponse<T> = { items: T[], changes?: never } | { changes: Changeset<T>, items?: never }
+export type LoadResponse<T> = {
+  items: T[],
+  changes?: never,
+} | {
+  changes: Changeset<T>,
+  items?: never,
+}
 
 // eslint-disable-next-line max-len
 export default interface PersistenceAdapter<T extends { id: I } & Record<string, any>, I> {