import { describe, expect, it } from 'vitest' import fc from 'fast-check' import { EventStorageKV } from '../../src/event-storage/event-storage-kv.js' import { KVStorageInMemory } from '../../src/kv-storage/kv-storage-in-memory.js' import { ConcurrencyError } from '../../src/utils/errors.js' import { collect, KVStorageCounting } from '../helpers.js' function setup(kv = new KVStorageCounting()) { let t = 1000 return { kv, events: new EventStorageKV(kv, { now: () => t++ }) } } describe('EventStorageKV', () => { it('appends and reads in self and child forms', async () => { const orders = setup().events.child('orders') expect(await orders.append('o1', [{ type: 'created' }], { expectedVersion: 0 })).toEqual({ version: 1 }) expect(await orders.child('o1').append([{ type: 'paid' }, { type: 'shipped' }])).toEqual({ version: 3 }) expect(await collect(orders.read('o1'))).toEqual([ { version: 1, at: 1000, data: { type: 'created' } }, { version: 2, at: 1001, data: { type: 'paid' } }, { version: 3, at: 1001, data: { type: 'shipped' } }, ]) expect(await collect(orders.child('o1').read({ from: 3 }))).toHaveLength(1) expect(await orders.version('o1')).toBe(3) expect(await orders.version('none')).toBe(0) }) it('stores one key per commit', async () => { const { kv, events } = setup() await events.child('s').append(['a']) await events.child('s').append(['b', 'c']) expect((await collect(kv.scan())).map(([k]) => k)).toEqual(['s/~e/000000000001', 's/~e/000000000002']) }) it('enforces expectedVersion and the arity rules', async () => { const { events } = setup() const s = events.child('s') await s.append(['a']) await expect(s.append(['b'], { expectedVersion: 0 })).rejects.toBeInstanceOf(ConcurrencyError) await expect(events.append(['x'])).rejects.toThrow(/root/) await expect(events.child('o').append(undefined as never, ['x'] as never)).rejects.toThrow(TypeError) await expect(events.child('o').version(undefined as never)).rejects.toThrow(TypeError) }) it('lists child streams', async () => { const orders = setup().events.child('orders') await orders.append('o1', ['a']) await orders.append('o2', ['b']) await orders.child('o1').append('notes', ['n']) expect(await collect(orders.keys())).toEqual(['o1', 'o2']) expect(await collect(orders.child('o1').keys())).toEqual(['notes']) }) it('a warm append is one writeIfAbsent; a cold one adds one reverse scan', async () => { const { kv, events } = setup() await events.child('s').append(['a']) kv.reset() await events.child('s').append(['b']) expect(kv.calls).toEqual({ writeIfAbsent: 1 }) const cold = setup(kv).events kv.reset() await cold.child('s').append(['c']) expect(kv.calls).toEqual({ scan: 1, writeIfAbsent: 1 }) }) it('a writer with a stale cache collides; with "any" it retries', async () => { const kv = new KVStorageInMemory() const a = setup(new KVStorageCounting(kv)).events.child('s') const b = setup(new KVStorageCounting(kv)).events.child('s') await a.append(['a1']) await b.append(['b1']) await expect(a.append(['a2'], { expectedVersion: 1 })).rejects.toMatchObject({ actualVersion: 2 }) expect(await a.append(['a3'])).toEqual({ version: 3 }) const results = await Promise.allSettled([a.append(['x'], { expectedVersion: 3 }), b.append(['y'], { expectedVersion: 3 })]) expect(results.filter((r) => r.status === 'fulfilled')).toHaveLength(1) }) it('reads any window correctly (property)', async () => { await fc.assert( fc.asyncProperty( fc.array(fc.integer({ min: 1, max: 4 }), { maxLength: 20 }), fc.record({ from: fc.option(fc.integer({ min: 0, max: 60 }), { nil: undefined }), to: fc.option(fc.integer({ min: 0, max: 60 }), { nil: undefined }), limit: fc.option(fc.nat(10), { nil: undefined }), }), async (sizes, options) => { const s = setup().events.child('s') let n = 0 for (const size of sizes) await s.append(Array.from({ length: size }, () => ++n)) let expected = Array.from({ length: n }, (_, i) => i + 1).filter((v) => v >= (options.from ?? 1) && v <= (options.to ?? Infinity)) if (options.limit !== undefined) expected = expected.slice(0, options.limit) expect((await collect(s.read(options))).map((e) => e.data)).toEqual(expected) }, ), { numRuns: 200 }, ) }) })