PotentialStorage/test/kv-storage/kv-storage-composite.test.ts

import { afterEach, describe, expect, it, vi } from 'vitest'
import { KVStorageComposite } from '../../src/kv-storage/kv-storage-composite.js'
import { KVStorageInMemory } from '../../src/kv-storage/kv-storage-in-memory.js'
import type { BaseKVStorage } from '../../src/kv-storage/base-kv-storage.js'
import { DocumentStorageKV } from '../../src/document-storage/document-storage-kv.js'
import { collect, KVStorageCounting } from '../helpers.js'
import { runKVStorageConformance } from './kv-storage.conformance.js'
import { TestNetwork } from './sync-helpers.js'

runKVStorageConformance(
  'KVStorageComposite (in-memory source, 1 replica)',
  () => new KVStorageComposite({ source: new KVStorageInMemory(), replicas: [new KVStorageInMemory()] }),
)
runKVStorageConformance('KVStorageComposite (http source, 2 replicas)', () => {
  network = new TestNetwork()
  return new KVStorageComposite({ source: network.client(), replicas: [new KVStorageInMemory(), new KVStorageInMemory()] })
})

let network: TestNetwork
const composites: KVStorageComposite[] = []
afterEach(async () => {
  await Promise.all(composites.splice(0).map((c) => c.close()))
  await network?.close()
})

function composite(persistent: BaseKVStorage = new KVStorageInMemory()) {
  const c = new KVStorageComposite({ source: network.client(), replicas: [new KVStorageInMemory(), persistent], retry: { baseDelay: 5, maxDelay: 20 } })
  composites.push(c)
  return c
}

describe('KVStorageComposite sync', () => {
  it('loads existing data with one full sync, then serves reads locally', async () => {
    network = new TestNetwork()
    await network.store.writeMany([
      ['users/1', { name: 'Ada' }],
      ['users/2', { name: 'Bob' }],
    ])
    const db = composite()
    await db.synced()
    network.store.reset()
    expect(await db.child('users').get('1')).toEqual({ name: 'Ada' })
    expect(await collect(db.child('users').keys())).toEqual(['1', '2'])
    expect(network.store.calls).toEqual({})
  })

  it('applies changes pushed from other clients to every replica', async () => {
    network = new TestNetwork()
    const memory = new KVStorageInMemory()
    const db = composite(memory)
    await db.synced()
    const seen = db.waitFor('change', (e) => e.origin === 'remote')
    await network.client().child('users').set('9', { name: 'Remote' })
    expect((await seen).changes).toEqual([{ key: 'users/9', value: { name: 'Remote' } }])
    expect(await db.child('users').get('9')).toEqual({ name: 'Remote' })
    expect(await memory.readMany(['users/9'])).toEqual([{ name: 'Remote' }])
  })

  it('queues writes while offline and replays them in order when back online', async () => {
    network = new TestNetwork()
    const db = composite()
    await db.synced()
    network.online = false
    await db.child('users').set('1', { v: 1 })
    await db.child('users').set('1', { v: 2 })
    await db.child('users').delete('1')
    await db.child('users').set('2', { v: 3 })
    expect(db.queued).toBe(4)
    expect(await db.child('users').get('2')).toEqual({ v: 3 }) // served locally
    await expect(db.writeIfAbsent([['x', 1]])).rejects.toThrow(/Cannot reach/)
    network.online = true
    await vi.waitFor(() => expect(db.queued).toBe(0))
    expect(await network.store.readMany(['users/1', 'users/2'])).toEqual([undefined, { v: 3 }])
  })

  it('keeps queued local values when the server pushes an older value for the same key', async () => {
    network = new TestNetwork()
    const db = composite()
    await db.synced()
    network.online = false
    await db.child('k').set({ from: 'local' })
    await network.store.writeMany([['k', { from: 'remote' }]]) // pushed over the still-open stream
    await new Promise((r) => setTimeout(r, 20))
    expect(await db.child('k').get()).toEqual({ from: 'local' })
    network.online = true
    await vi.waitFor(() => expect(db.queued).toBe(0))
    expect(await network.store.readMany(['k'])).toEqual([{ from: 'local' }])
  })

  it('restarts from the persistent replica and resumes without a full sync', async () => {
    network = new TestNetwork()
    const persistent = new KVStorageInMemory()
    const first = composite(persistent)
    await first.child('a').set(1)
    await first.synced()
    await first.close()
    await network.client().writeMany([['b', 2]]) // missed while "closed"

    network.store.reset()
    const second = composite(persistent)
    expect(await second.child('a').get()).toBe(1) // available before contacting the server
    await vi.waitFor(async () => expect(await second.child('b').get()).toBe(2))
    expect(network.store.calls.scan).toBeUndefined() // resumed via Last-Event-ID, no full scan
  })

  it('falls back to a full sync when the server cannot resume', async () => {
    network = new TestNetwork({ historySize: 1 })
    const persistent = new KVStorageInMemory()
    const first = composite(persistent)
    await first.child('a').set(1)
    await first.synced()
    await first.close()
    await network.client().writeMany([['b', 2]])
    await network.client().writeMany([['c', 3]])
    await network.client().removeMany(['a'])
    const second = composite(persistent)
    await vi.waitFor(async () => expect(await collect(second.keys())).toEqual(['b', 'c']))
  })

  it('replays an outbox left by a previous session', async () => {
    network = new TestNetwork()
    const persistent = new KVStorageInMemory()
    const first = composite(persistent)
    await first.synced()
    network.online = false
    await first.child('offline').set('queued')
    await first.close()
    network.online = true
    const second = composite(persistent)
    await second.start()
    await vi.waitFor(async () => expect(await network.store.readMany(['offline'])).toEqual(['queued']))
  })

  it('serves document queries from the local replica, offline too', async () => {
    network = new TestNetwork()
    const db = composite()
    const users = new DocumentStorageKV(db).child('users')
    await users.set('1', { age: 30 })
    await users.set('2', { age: 40 })
    await db.synced()
    network.online = false
    expect((await collect(users.find({ age: { $gt: 35 } }))).map((r) => r.key)).toEqual(['2'])
  })

  it('does not undo writes made while a full sync is scanning', async () => {
    // A source whose scan reads ahead (like the HTTP stream) and pauses after its first entry.
    let release!: () => void
    let reached!: () => void
    const gate = new Promise<void>((r) => (release = r))
    const paused = new Promise<void>((r) => (reached = r))
    const inner = new KVStorageInMemory()
    await inner.writeMany([
      ['a', 1],
      ['c', 3],
    ])
    const source = new KVStorageCounting(inner)
    const scan = source.scan.bind(source)
    source.scan = async function* (options) {
      let first = true
      for (const entry of await collect(scan(options))) {
        yield entry
        if (first && options?.values) ((first = false), reached(), await gate)
      }
    }
    const persistent = new KVStorageInMemory()
    await persistent.writeMany([['a0', 'stale']]) // a replica copy from before, without sync metadata
    const db = new KVStorageComposite({ source, replicas: [persistent] })
    void db.start()
    await paused // the diff has passed 'b' in the source but not in the replica
    await db.writeMany([['b', 2]])
    release()
    await db.synced()
    expect(await collect(db.scan({ values: true }))).toEqual([
      ['a', 1],
      ['b', 2],
      ['c', 3],
    ])
  })
})