import { mkdtempSync, rmSync } from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import { afterAll, describe, expect, it } from 'vitest';
import { RawLake } from './lake.js';

const root = mkdtempSync(path.join(os.tmpdir(), 'ci-lake-'));
afterAll(() => rmSync(root, { recursive: true, force: true }));

describe('RawLake', () => {
  it('serialises 5,000 concurrent writes without piling up drain listeners and reads any line back', async () => {
    const warnings: string[] = [];
    const onWarning = (w: Error) => warnings.push(`${w.name}: ${w.message}`);
    process.on('warning', onWarning);
    try {
      const lake = new RawLake('fake-source', 'ING-FAKE-20260908-000001', new Date('2026-09-08T00:00:00Z'), root);
      // Large-ish payloads so the gzip stream hits back-pressure (highWaterMark 16 KB) many times.
      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 } })));
      await lake.close();
      expect(refs).toHaveLength(5000);
      expect(refs[0]).toMatch(/fake-source\/2026-09-08\/record\/ING-FAKE-20260908-000001-001\.jsonl\.gz#0$/);
      expect(refs[4999]!.endsWith('#4999')).toBe(true);
      // Same file for all refs (well below the 64 MB part limit), line numbers unique and in call order.
      expect(new Set(refs.map((r) => r.split('#')[0])).size).toBe(1);
      expect(new Set(refs).size).toBe(5000);

      const one = (await RawLake.read(refs[4242]!)) as { i: number; id: string };
      expect(one.i).toBe(4242);
      expect(one.id).toBe('REC-4242');
      const last = (await RawLake.read(refs[4999]!)) as { i: number };
      expect(last.i).toBe(4999);
      expect(await RawLake.read(`${refs[0]!.split('#')[0]}#5000`)).toBeNull();
    } finally {
      // Let any pending 'warning' events flush before asserting.
      await new Promise((r) => setImmediate(r));
      process.off('warning', onWarning);
    }
    expect(warnings.filter((w) => w.includes('MaxListenersExceededWarning'))).toEqual([]);
  });

  it('flush() makes already-written lines readable before close()', async () => {
    const lake = new RawLake('fake-source', 'ING-FAKE-20260908-000002', new Date('2026-09-08T00:00:00Z'), root);
    const refs = await Promise.all(Array.from({ length: 50 }, (_, i) => lake.put('entity', { i })));
    await lake.flush();
    // The trailer is missing until close(), but the flushed block decodes: the target line is found.
    const row = (await RawLake.read(refs[10]!).catch(() => null)) as { i: number } | null;
    expect(row?.i).toBe(10);
    await lake.close();
    expect(((await RawLake.read(refs[49]!)) as { i: number }).i).toBe(49);
  });

  it('keeps separate writers per entity', async () => {
    const lake = new RawLake('fake-source', 'ING-FAKE-20260908-000003', new Date('2026-09-08T00:00:00Z'), root);
    const [a, b] = await Promise.all([lake.put('alpha', { a: 1 }), lake.put('beta', { b: 2 })]);
    await lake.close();
    expect(a).toContain('/alpha/');
    expect(b).toContain('/beta/');
    expect(await RawLake.read(a)).toEqual({ a: 1 });
    expect(await RawLake.read(b)).toEqual({ b: 2 });
  });
});