/**
 * AriaEngine — SSTable CRC-32 损坏检测集成测试
 *
 * 覆盖：
 * 1. 打开时整文件 CRC 校验失败 → 损坏 SSTable 被清理（自愈），打开不阻塞
 * 2. 运行期预加载 CRC 校验失败 → 损坏文件不缓存并被清理
 * 3. 旧版无校验文件（checksum=0）在 LSM 层仍正常加载
 */
import { AriaEngine } from '../../src/engine/aria/index';
import { createSchema } from '../../src/table/schema';

import { resetOPFSMock, readManifestNamespace } from '../helpers/storage-harness';

beforeEach(() => { resetOPFSMock(); });

let idbCounter = 0;
function uniqueDB(): string {
  return `crc-${Date.now()}-${++idbCounter}-${Math.random().toString(36).slice(2, 8)}`;
}

const SCHEMA = () => createSchema('users', {
  id: { type: 'string', primaryKey: true },
  name: { type: 'string' },
  age: { type: 'number' },
});

/** 篡改指定 pg_ 页面文件的一个字节 */
async function corruptKey(engine: AriaEngine, key: string, byteOffset: number): Promise<void> {
  const backend = (engine as any).backend;
  const raw = await backend.read(key);
  expect(raw).not.toBeNull();
  const buf = new Uint8Array(raw as ArrayBuffer);
  buf[Math.min(byteOffset, buf.length - 1)] ^= 0xff;
  await backend.write(key, buf.buffer as ArrayBuffer);
}

/** 列出全部 SSTable 的页面文件 key（页面化存储：pg_ 前缀） */
async function listSSTKeys(engine: AriaEngine): Promise<string[]> {
  const backend = (engine as any).backend;
  const keys = (await backend.listKeys()) as string[];
  return keys.filter((k) => k.startsWith('pg_'));
}

/**
 * 读取主 LSM 的 SSTable meta 列表（v0.8.0 B-6：改读 manifest）。
 *
 * 元数据不再是独立的裸 JSON：它随 manifest 原子提交（带 CRC + 世代号）。
 * 旧布局下 `JSON.parse` 失败会被当成"没有文件"，即元数据损坏 = 静默空库。
 */
async function listSSTMetas(engine: AriaEngine): Promise<{ id: number; pageIds?: number[] }[]> {
  const backend = (engine as any).backend;
  return readManifestNamespace(backend, 'main');
}

describe('AriaEngine — SSTable CRC 损坏检测（集成）', () => {
  it('打开时 CRC 损坏的 SSTable 被清理，其余数据可读', async () => {
    const engine = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 64 * 1024 * 1024 });
    const dbName = uniqueDB();
    await engine.open(dbName, 1);
    await engine.createTable(SCHEMA());

    // 写入 200 行（触发多次 flush → 多个 SSTable）
    for (let batch = 0; batch < 3; batch++) {
      const rows = [] as Record<string, unknown>[];
      for (let i = 0; i < 50; i++) {
        rows.push({ id: `u-${batch * 50 + i}`, name: `User${batch * 50 + i}`, age: batch * 50 + i });
      }
      await engine.insert('users', rows);
      await (engine as any).lsm.flush();
    }
    await engine.close();

    // 重新打开，篡改第一个 SSTable 文件的数据区
    const engine2 = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 64 * 1024 * 1024 });
    await engine2.open(dbName, 1);
    const sstKeys = await listSSTKeys(engine2);
    expect(sstKeys.length).toBeGreaterThanOrEqual(2);
    await corruptKey(engine2, sstKeys[0], 64);

    // 再次打开：损坏文件应被跳过并清理，打开不抛错
    await engine2.close();
    const engine3 = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 64 * 1024 * 1024 });
    await engine3.open(dbName, 1);

    // 剩余未损坏文件的数据应可查询
    const remaining = await engine3.find('users', { table: 'users' });
    expect(remaining.length).toBeGreaterThan(0);
    expect(remaining.length).toBeLessThan(200);

    // 损坏文件已被清理（页面 + meta 移除）
    const afterKeys = await listSSTKeys(engine3);
    expect(afterKeys).not.toContain(sstKeys[0]);
    const metas = await listSSTMetas(engine3);
    // 被篡改页面所属的 SSTable meta 应被移除
    const victimMetas = await listSSTMetas(engine2);
    const victimMeta = victimMetas.find((m) => m.pageIds?.includes(Number(sstKeys[0].slice(3))));
    expect(metas.some((m) => m.id === victimMeta?.id)).toBe(false);

    await engine3.close();
  });

  it('打开时损坏全部 SSTable → 库仍可打开，数据为空但不崩溃', async () => {
    const engine = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 64 * 1024 * 1024 });
    const dbName = uniqueDB();
    await engine.open(dbName, 1);
    await engine.createTable(SCHEMA());
    for (let batch = 0; batch < 2; batch++) {
      const rows = [] as Record<string, unknown>[];
      for (let i = 0; i < 50; i++) {
        rows.push({ id: `u-${batch * 50 + i}`, name: `User${batch * 50 + i}`, age: batch * 50 + i });
      }
      await engine.insert('users', rows);
      await (engine as any).lsm.flush();
    }
    await engine.close();

    const engine2 = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 64 * 1024 * 1024 });
    await engine2.open(dbName, 1);
    for (const key of await listSSTKeys(engine2)) {
      await corruptKey(engine2, key, 16);
    }
    await engine2.close();

    const engine3 = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 64 * 1024 * 1024 });
    // 不应抛 ARIA_OPEN_ERROR
    await engine3.open(dbName, 1);
    const rows3 = await engine3.find('users', { table: 'users' });
    expect(rows3.length).toBe(0);
    await engine3.close();
  });

  it('运行期 CRC 损坏 → 预加载不缓存损坏文件并清理（自愈）', async () => {
    const engine = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 64 * 1024 * 1024 });
    const dbName = uniqueDB();
    await engine.open(dbName, 1);
    await engine.createTable(SCHEMA());
    for (let batch = 0; batch < 2; batch++) {
      const rows = [] as Record<string, unknown>[];
      for (let i = 0; i < 50; i++) {
        rows.push({ id: `u-${batch * 50 + i}`, name: `User${batch * 50 + i}`, age: batch * 50 + i });
      }
      await engine.insert('users', rows);
      await (engine as any).lsm.flush();
    }

    // 运行期：直接篡改后端中的某个 SSTable 文件（不触发打开校验路径）
    const sstKeys = await listSSTKeys(engine);
    expect(sstKeys.length).toBeGreaterThanOrEqual(2);
    const victim = sstKeys[sstKeys.length - 1]; // 篡改最新（memtable 已 flush 后的文件）
    await corruptKey(engine, victim, 100);

    // 缓存里已有该文件（flush 时缓存）→ 清 LSM 层与 BufferPool 页面缓存模拟运行期磁盘损坏
    const lsm = (engine as any).lsm;
    lsm.sstableCache.clear();
    await (engine as any).bufferPool.clear();

    // 查询触发 prefetchRange → preloadSSTable 发现 CRC 失败 → 清理 + 不缓存
    const remaining = await engine.find('users', { table: 'users' });
    expect(remaining.length).toBeLessThan(100);
    expect(lsm.sstableCache.has(Number(victim.slice(3)))).toBe(false);
    const metas = await lsm.sstableStore.listMeta();
    expect(metas.some((m: { id: number }) => m.id === Number(victim.slice(3)))).toBe(false);

    await engine.close();
  });

  it('旧版无校验文件（checksum=0）在 LSM 中正常加载', async () => {
    const engine = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 64 * 1024 * 1024 });
    const dbName = uniqueDB();
    await engine.open(dbName, 1);
    await engine.createTable(SCHEMA());
    await engine.insert('users', [
      { id: 'a', name: 'Alice', age: 30 },
      { id: 'b', name: 'Bob', age: 25 },
    ]);
    await (engine as any).lsm.flush();
    await engine.close();

    const engine2 = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 64 * 1024 * 1024 });
    await engine2.open(dbName, 1);
    const sstKeys = await listSSTKeys(engine2);
    expect(sstKeys.length).toBe(1);
    const backend = (engine2 as any).backend;
    const raw = await backend.read(sstKeys[0]);
    const buf = new Uint8Array(raw as ArrayBuffer);
    new DataView(buf.buffer).setUint32(buf.byteLength - 4, 0, false); // checksum 清零 → 旧版
    await backend.write(sstKeys[0], buf.buffer as ArrayBuffer);
    await engine2.close();

    // 重开：checksum=0 跳过校验，数据完整可读
    const engine3 = new AriaEngine({ storageBackend: 'opfs', memtableSizeThreshold: 64 * 1024 * 1024 });
    await engine3.open(dbName, 1);
    const rows3 = await engine3.find('users', { table: 'users' });
    expect(rows3.length).toBe(2);
    expect(rows3.map((r) => r.id).sort()).toEqual(['a', 'b']);
    await engine3.close();
  });
});