PotentialStorage/test/events/kv.test.ts

import { describe, expect, it } from 'vitest'
import fc from 'fast-check'
import { KVStorage } from '../../src/kv/node.js'
import { InMemoryKVDriver } from '../../src/kv/memory.js'
import { KVEventStorage } from '../../src/events/kv.js'
import { ConcurrencyError } from '../../src/errors.js'
import { collect, CountingDriver, type Capability } from '../helpers.js'

function setup(hide: Capability[] = [], inner = new InMemoryKVDriver()) {
  const driver = new CountingDriver(inner, hide)
  let t = 1000
  const events = new KVEventStorage(new KVStorage(driver), { now: () => t++ })
  return { driver, events }
}

describe('KVEventStorage', () => {
  it('appends and reads with versions and metadata', async () => {
    const { events } = setup()
    const orders = events.child('orders')
    expect(await orders.append('o1', [{ type: 'created', total: 40 }], { expectedVersion: 0 })).toEqual({ version: 1, commit: 1 })
    expect(await orders.child('o1').append([{ type: 'paid' }, { type: 'shipped' }])).toEqual({ version: 3, commit: 2 })
    expect(await collect(orders.read('o1'))).toEqual([
      { version: 1, commit: 1, at: 1000, data: { type: 'created', total: 40 } },
      { version: 2, commit: 2, at: 1001, data: { type: 'paid' } },
      { version: 3, commit: 2, at: 1001, data: { type: 'shipped' } },
    ])
    expect(await orders.version('o1')).toBe(3)
    expect(await orders.child('o1').version()).toBe(3)
    expect(await orders.version('nope')).toBe(0)
  })

  it('stores one key per commit under <stream>/~e/<seq:12>', async () => {
    const { events, driver } = setup()
    await events.child('orders').append('o1', ['a'])
    await events.child('orders').append('o1', ['b', 'c'])
    expect(await collect(driver.inner.scan({ values: true }))).toEqual([
      ['orders/o1/~e/000000000001', { firstVersion: 1, at: 1000, events: ['a'] }],
      ['orders/o1/~e/000000000002', { firstVersion: 2, at: 1001, events: ['b', 'c'] }],
    ])
  })

  it('enforces expectedVersion', async () => {
    const { events } = setup()
    const s = events.child('s')
    await s.append(['a'], { expectedVersion: 0 })
    await expect(s.append(['b'], { expectedVersion: 0 })).rejects.toBeInstanceOf(ConcurrencyError)
    await expect(s.append(['b'], { expectedVersion: 5 })).rejects.toMatchObject({ expectedVersion: 5, actualVersion: 1 })
    expect(await s.append(['b'], { expectedVersion: 1 })).toEqual({ version: 2, commit: 2 })
    await expect(s.append([], { expectedVersion: 1 })).rejects.toBeInstanceOf(ConcurrencyError)
    expect(await s.append([])).toEqual({ version: 2, commit: 2 })
  })

  it('applies the arity rules', async () => {
    const { events } = setup()
    const orders = events.child('orders')
    const id: string | undefined = undefined
    await expect(orders.append(id as never, ['x'] as never)).rejects.toThrow(TypeError)
    await expect(orders.version(id as never)).rejects.toThrow(TypeError)
    await expect(collect(orders.read(id as never))).rejects.toThrow(TypeError)
    await expect(events.append(['x'])).rejects.toThrow(/root/)
    await expect(orders.append('o1', 'x' as never)).rejects.toThrow(/array/)
  })

  it('lists child streams and keeps nested streams separate', async () => {
    const { events } = setup()
    const orders = events.child('orders')
    await orders.append('o1', ['a'])
    await orders.append('o2', ['b'])
    await orders.child('o1').child('shipments').append('s1', ['ship'])
    await orders.child('o1').append('notes', ['n'])
    expect(await collect(orders.keys())).toEqual(['o1', 'o2'])
    expect(await collect(orders.child('o1').keys())).toEqual(['notes', 'shipments'])
    expect((await collect(orders.read('o1'))).map((e) => e.data)).toEqual(['a'])
  })
})

describe('read windows (property)', () => {
  it('matches brute force for any commit layout', async () => {
    await fc.assert(
      fc.asyncProperty(
        fc.array(fc.integer({ min: 1, max: 4 }), { maxLength: 25 }),
        fc.record({
          from: fc.option(fc.integer({ min: -1, max: 70 }), { nil: undefined }),
          to: fc.option(fc.integer({ min: 0, max: 70 }), { nil: undefined }),
          reverse: fc.boolean(),
          limit: fc.option(fc.integer({ min: 0, max: 10 }), { nil: undefined }),
        }),
        async (sizes, opts) => {
          const { events } = setup()
          const s = events.child('s')
          const all: number[] = []
          for (const n of sizes) {
            const batch = Array.from({ length: n }, (_, i) => all.length + i + 1)
            await s.append(batch)
            all.push(...batch)
          }
          let exp = all.filter((v) => v >= Math.max(1, opts.from ?? 1) && v <= (opts.to ?? Infinity))
          if (opts.reverse) exp.reverse()
          if (opts.limit !== undefined) exp = exp.slice(0, opts.limit)
          const got = await collect(s.read(opts))
          expect(got.map((e) => e.data)).toEqual(exp)
          expect(got.map((e) => e.version)).toEqual(exp)
        },
      ),
      { numRuns: 300 },
    )
  })
})

describe('costs', () => {
  it('cached head: a warm append is a single setIfAbsent', async () => {
    const { events, driver } = setup()
    const s = events.child('s')
    await s.append(['a'])
    driver.reset()
    await s.append(['b'])
    await s.append(['c'], { expectedVersion: 2 })
    expect(driver.calls).toEqual({ setIfAbsent: 2 })
  })

  it('cold head lookup is one reverse scan with limit 1', async () => {
    const inner = new InMemoryKVDriver()
    await setup([], inner).events.child('s').append(['a', 'b'])
    const { events, driver } = setup([], inner)
    expect(await events.child('s').append(['c'])).toEqual({ version: 3, commit: 2 })
    expect(driver.calls).toEqual({ scan: 1, setIfAbsent: 1 })
  })

  it('read({ from }) finds single-event commits in one probe', async () => {
    const { events, driver } = setup()
    const s = events.child('s')
    for (let i = 0; i < 200; i++) await s.append([i])
    driver.reset()
    expect((await collect(s.read({ from: 150, limit: 1 }))).map((e) => e.version)).toEqual([150])
    expect(driver.calls).toEqual({ getMany: 1, scan: 1 })
  })

  it('read({ from }) uses O(log commits) probes for multi-event commits', async () => {
    const { events, driver } = setup()
    const s = events.child('s')
    for (let i = 0; i < 1000; i++) await s.append(i % 3 === 0 ? [i, i] : [i])
    driver.reset()
    const [first] = await collect(s.read({ from: 777, limit: 1 }))
    expect(first?.version).toBe(777)
    // 1 guess probe + ceil(log8(1000)) search rounds, then the read scan. The guess hit an
    // existing commit, so no head scan was needed.
    expect(driver.calls.getMany).toBeLessThanOrEqual(1 + 4)
    expect(driver.calls.scan).toBe(1)
  })
})

describe('concurrency', () => {
  it('a stale writer collides with the existing commit key', async () => {
    const inner = new InMemoryKVDriver()
    const a = setup([], inner).events.child('s')
    const b = setup([], inner).events.child('s')
    await a.append(['a1'])
    await b.append(['b1']) // a's cached head is now stale
    await expect(a.append(['a2'], { expectedVersion: 1 })).rejects.toMatchObject({ actualVersion: 2 })
    // 'any' retries after refreshing the head
    expect(await a.append(['a3'])).toEqual({ version: 3, commit: 3 })
    expect((await collect(b.read())).map((e) => e.data)).toEqual(['a1', 'b1', 'a3'])
  })

  it('a stale cached head is refreshed before rejecting expectedVersion', async () => {
    const inner = new InMemoryKVDriver()
    const a = setup([], inner).events.child('s')
    const b = setup([], inner).events.child('s')
    await a.append(['a1'])
    await b.append(['b1'])
    expect(await a.append(['a2'], { expectedVersion: 2 })).toEqual({ version: 3, commit: 3 })
  })

  it('interleaved writers from two instances produce a dense, non-overlapping log', async () => {
    const inner = new InMemoryKVDriver()
    const a = setup([], inner).events.child('s')
    const b = setup([], inner).events.child('s')
    await Promise.all([a.append(['a1']), b.append(['b1']), a.append(['a2', 'a3']), b.append(['b2'])])
    const log = await collect(a.read())
    expect(log.map((e) => e.version)).toEqual([1, 2, 3, 4, 5])
    expect(new Set(log.map((e) => e.data))).toEqual(new Set(['a1', 'a2', 'a3', 'b1', 'b2']))
  })

  it('exactly one of several optimistic writers at the same version wins', async () => {
    const inner = new InMemoryKVDriver()
    const writers = [0, 1, 2, 3].map(() => setup([], inner).events.child('s'))
    const results = await Promise.allSettled(writers.map((w, i) => w.append([i], { expectedVersion: 0 })))
    expect(results.filter((r) => r.status === 'fulfilled')).toHaveLength(1)
    expect(results.filter((r) => r.status === 'rejected').every((r) => (r as PromiseRejectedResult).reason instanceof ConcurrencyError)).toBe(true)
  })

  it('without setIfAbsent, an in-process mutex serialises appends (single-writer mode)', async () => {
    const { events, driver } = setup(['setIfAbsent'])
    const s = events.child('s')
    await Promise.all(Array.from({ length: 20 }, (_, i) => s.append([i])))
    expect((await collect(s.read())).map((e) => e.version)).toEqual(Array.from({ length: 20 }, (_, i) => i + 1))
    expect(driver.calls.setIfAbsent).toBeUndefined()
    await expect(Promise.all([s.append(['x'], { expectedVersion: 20 }), s.append(['y'], { expectedVersion: 20 })])).rejects.toBeInstanceOf(ConcurrencyError)
  })
})