spb/cancerindex

1import { mkdtempSync, rmSync } from 'node:fs';2import os from 'node:os';3import path from 'node:path';4import { afterAll, describe, expect, it } from 'vitest';5import { RawLake } from './lake.js';67const root = mkdtempSync(path.join(os.tmpdir(), 'ci-lake-'));8afterAll(() => rmSync(root, { recursive: true, force: true }));910describe('RawLake', () => {11  it('serialises 5,000 concurrent writes without piling up drain listeners and reads any line back', async () => {12    const warnings: string[] = [];13    const onWarning = (w: Error) => warnings.push(`${w.name}: ${w.message}`);14    process.on('warning', onWarning);15    try {16      const lake = new RawLake('fake-source', 'ING-FAKE-20260908-000001', new Date('2026-09-08T00:00:00Z'), root);17      // Large-ish payloads so the gzip stream hits back-pressure (highWaterMark 16 KB) many times.18      const refs = await Promise.all(Array.from({ length: 5000 }, (_, i) => lake.put('record', { i, id: `REC-${i}`, text: 'x'.repeat(400), nested: { odd: i % 2 === 1 } })));19      await lake.close();20      expect(refs).toHaveLength(5000);21      expect(refs[0]).toMatch(/fake-source\/2026-09-08\/record\/ING-FAKE-20260908-000001-001\.jsonl\.gz#0$/);22      expect(refs[4999]!.endsWith('#4999')).toBe(true);23      // Same file for all refs (well below the 64 MB part limit), line numbers unique and in call order.24      expect(new Set(refs.map((r) => r.split('#')[0])).size).toBe(1);25      expect(new Set(refs).size).toBe(5000);2627      const one = (await RawLake.read(refs[4242]!)) as { i: number; id: string };28      expect(one.i).toBe(4242);29      expect(one.id).toBe('REC-4242');30      const last = (await RawLake.read(refs[4999]!)) as { i: number };31      expect(last.i).toBe(4999);32      expect(await RawLake.read(`${refs[0]!.split('#')[0]}#5000`)).toBeNull();33    } finally {34      // Let any pending 'warning' events flush before asserting.35      await new Promise((r) => setImmediate(r));36      process.off('warning', onWarning);37    }38    expect(warnings.filter((w) => w.includes('MaxListenersExceededWarning'))).toEqual([]);39  });4041  it('flush() makes already-written lines readable before close()', async () => {42    const lake = new RawLake('fake-source', 'ING-FAKE-20260908-000002', new Date('2026-09-08T00:00:00Z'), root);43    const refs = await Promise.all(Array.from({ length: 50 }, (_, i) => lake.put('entity', { i })));44    await lake.flush();45    // The trailer is missing until close(), but the flushed block decodes: the target line is found.46    const row = (await RawLake.read(refs[10]!).catch(() => null)) as { i: number } | null;47    expect(row?.i).toBe(10);48    await lake.close();49    expect(((await RawLake.read(refs[49]!)) as { i: number }).i).toBe(49);50  });5152  it('keeps separate writers per entity', async () => {53    const lake = new RawLake('fake-source', 'ING-FAKE-20260908-000003', new Date('2026-09-08T00:00:00Z'), root);54    const [a, b] = await Promise.all([lake.put('alpha', { a: 1 }), lake.put('beta', { b: 2 })]);55    await lake.close();56    expect(a).toContain('/alpha/');57    expect(b).toContain('/beta/');58    expect(await RawLake.read(a)).toEqual({ a: 1 });59    expect(await RawLake.read(b)).toEqual({ b: 2 });60  });61});62