/**
 * tests/analytics/query-cache.test.ts
 * ====================================
 * BI-P020 — QUERY CACHE.
 *
 * DoD: "Invalidation tests prove changed dataset/metric does not serve stale
 * unauthorized result."
 *
 * Two words in that sentence carry the weight — STALE and UNAUTHORIZED — and
 * they are different failures, so they are tested separately:
 *
 *   STALE        a component of the key changed (metric version, semantic
 *                model version, data snapshot) and the cache still answered.
 *                Tested by changing each of the six components ONE AT A TIME
 *                and requiring a miss for every one; a table-driven loop, so
 *                adding a component to the key without adding it to the loop
 *                is caught by the count assertion.
 *
 *   UNAUTHORIZED another tenant's result came back. Tested three ways: a
 *                different tenant misses; a poisoned entry stored under this
 *                tenant's key but carrying another tenant's id is REFUSED and
 *                evicted rather than returned; and the same components under
 *                two tenants produce different keys.
 *
 * The poisoning test is the one that matters. Key namespacing is a naming
 * convention, and a naming convention is what ST0RE was relying on when it
 * turned out to have four cross-tenant paths rather than the one reported.
 */
import { describe, expect, it } from 'vitest';
import { SiaxError } from '@/src/platform/errors';
import {
  buildCacheKey,
  InMemoryResultCache,
  tenantKeyPrefix,
  ValkeyResultCache,
  type CacheKeyComponents,
  type CachedResult,
  type ValkeyLikeClient,
} from '@/src/analytics/cache/query-cache';

const BASE: CacheKeyComponents = {
  tenant_id: 't_alpha',
  project_id: 'p_main',
  dataset_id: 'ds_sales',
  semantic_model_version: 7,
  metric_versions: { m_revenue: 3, m_orders: 1 },
  compiled_query_hash: 'a'.repeat(64),
  data_snapshot: 'dsv_42',
  format: 'json',
};

describe('BI-P020 — the key requires all six components', () => {
  const omissions: Array<[string, Partial<CacheKeyComponents>]> = [
    ['tenant_id', { tenant_id: '' }],
    ['dataset_id', { dataset_id: '' }],
    ['semantic_model_version', { semantic_model_version: undefined as unknown as number }],
    ['metric_versions', { metric_versions: {} }],
    ['compiled_query_hash', { compiled_query_hash: '' }],
    ['data_snapshot', { data_snapshot: '' }],
    ['format', { format: '' }],
  ];

  it.each(omissions)('refuses to build a key without %s', (_name, override) => {
    expect(() => buildCacheKey({ ...BASE, ...override })).toThrowError(SiaxError);
  });

  it('refuses a metric whose version is missing', () => {
    expect(() =>
      buildCacheKey({ ...BASE, metric_versions: { m_revenue: undefined as unknown as number } }),
    ).toThrowError(SiaxError);
  });

  it('is stable under key ordering — the same query does not fragment the cache', () => {
    const a = buildCacheKey({ ...BASE, metric_versions: { m_orders: 1, m_revenue: 3 } });
    const b = buildCacheKey({ ...BASE, metric_versions: { m_revenue: 3, m_orders: 1 } });
    expect(a).toBe(b);
  });

  it('carries the tenant in plaintext so invalidation is a prefix scan', () => {
    expect(buildCacheKey(BASE).startsWith(tenantKeyPrefix('t_alpha'))).toBe(true);
  });

  it('refuses a tenant id that is not key-safe', () => {
    for (const bad of ['t alpha', 't:alpha', 't*', '../t_beta']) {
      expect(() => buildCacheKey({ ...BASE, tenant_id: bad })).toThrowError(SiaxError);
    }
  });
});

describe('BI-P020 — a changed component never serves a STALE result', () => {
  /**
   * Each entry changes ONE component. Every one must miss. The count is
   * asserted so that adding a component to `CacheKeyComponents` without adding
   * it here is caught rather than silently untested.
   */
  const mutations: Array<[string, Partial<CacheKeyComponents>]> = [
    ['semantic model version bumped', { semantic_model_version: 8 }],
    ['a metric definition versioned', { metric_versions: { m_revenue: 4, m_orders: 1 } }],
    ['a metric removed from the query', { metric_versions: { m_revenue: 3 } }],
    ['the data snapshot advanced', { data_snapshot: 'dsv_43' }],
    ['the compiled query changed (different filters)', { compiled_query_hash: 'b'.repeat(64) }],
    ['a different dataset', { dataset_id: 'ds_other' }],
    ['a different result format', { format: 'arrow_ipc' }],
    ['a different project', { project_id: 'p_other' }],
    ['a different tenant', { tenant_id: 't_beta' }],
  ];

  it('covers every component of the key', () => {
    const covered = new Set(mutations.flatMap(([, m]) => Object.keys(m)));
    const componentKeys = Object.keys(BASE);
    expect([...covered].sort()).toEqual([...componentKeys].sort());
  });

  it.each(mutations)('misses after %s', async (_label, override) => {
    const cache = new InMemoryResultCache();
    await cache.set(BASE, { rows: [{ value: 100 }] }, { ttlSeconds: 300, rowCount: 1 });
    expect(await cache.get(BASE)).toBeDefined();
    const changed = { ...BASE, ...override };
    expect(await cache.get(changed), 'a changed component served the old result').toBeUndefined();
  });

  it('expires an entry rather than serving it past its TTL', async () => {
    let clock = 1_000_000;
    const cache = new InMemoryResultCache(() => clock);
    await cache.set(BASE, { rows: [] }, { ttlSeconds: 10, rowCount: 0 });
    expect(await cache.get(BASE)).toBeDefined();
    clock += 11_000;
    expect(await cache.get(BASE)).toBeUndefined();
  });

  it('refuses a non-positive TTL rather than caching forever', async () => {
    const cache = new InMemoryResultCache();
    for (const ttl of [0, -1, Number.NaN, Number.POSITIVE_INFINITY]) {
      await expect(cache.set(BASE, {}, { ttlSeconds: ttl, rowCount: 0 })).rejects.toBeInstanceOf(SiaxError);
    }
  });
});

describe('BI-P020 — invalidation', () => {
  async function seed() {
    const cache = new InMemoryResultCache();
    await cache.set(BASE, { v: 1 }, { ttlSeconds: 300, rowCount: 1 });
    await cache.set(
      { ...BASE, compiled_query_hash: 'c'.repeat(64) },
      { v: 2 },
      { ttlSeconds: 300, rowCount: 1 },
    );
    await cache.set(
      { ...BASE, dataset_id: 'ds_other', metric_versions: { m_other: 1 } },
      { v: 3 },
      { ttlSeconds: 300, rowCount: 1 },
    );
    await cache.set({ ...BASE, tenant_id: 't_beta' }, { v: 4 }, { ttlSeconds: 300, rowCount: 1 });
    return cache;
  }

  it("invalidateDataset drops that dataset's entries and nothing else", async () => {
    const cache = await seed();
    const dropped = await cache.invalidateDataset('t_alpha', 'ds_sales');
    expect(dropped).toBe(2);
    expect(await cache.get(BASE)).toBeUndefined();
    expect(
      await cache.get({ ...BASE, dataset_id: 'ds_other', metric_versions: { m_other: 1 } }),
    ).toBeDefined();
    // The other tenant is untouched — invalidation is tenant-scoped too.
    expect(await cache.get({ ...BASE, tenant_id: 't_beta' })).toBeDefined();
  });

  it('invalidateMetric drops every entry that used the metric', async () => {
    const cache = await seed();
    const dropped = await cache.invalidateMetric('t_alpha', 'm_revenue');
    expect(dropped).toBe(2);
    expect(await cache.get(BASE)).toBeUndefined();
    expect(
      await cache.get({ ...BASE, dataset_id: 'ds_other', metric_versions: { m_other: 1 } }),
    ).toBeDefined();
  });

  it('invalidateTenant drops only that tenant', async () => {
    const cache = await seed();
    const dropped = await cache.invalidateTenant('t_alpha');
    expect(dropped).toBe(3);
    expect(await cache.get({ ...BASE, tenant_id: 't_beta' })).toBeDefined();
  });
});

describe('BI-P020 — never share a tenant result cache', () => {
  it('two tenants asking the identical question get different keys', () => {
    expect(buildCacheKey(BASE)).not.toBe(buildCacheKey({ ...BASE, tenant_id: 't_beta' }));
  });

  it("REFUSES and evicts an entry whose recorded tenant is not the caller's", async () => {
    // The scenario key namespacing alone does not cover: a poisoned or
    // mis-migrated entry sitting under the right key with the wrong owner.
    // Returning `undefined` here would hide a security event; this refuses.
    const client = new FakeValkey();
    const cache = new ValkeyResultCache(client);
    await cache.set(BASE, { secret: 'alpha data' }, { ttlSeconds: 300, rowCount: 1 });
    const key = buildCacheKey(BASE);
    const poisoned = JSON.parse(client.data.get(key) as string) as CachedResult;
    poisoned.tenant_id = 't_beta';
    poisoned.value = { secret: 'BETA DATA' };
    client.data.set(key, JSON.stringify(poisoned));

    await expect(cache.get(BASE)).rejects.toBeInstanceOf(SiaxError);
    // ...and the poisoned entry is gone, so the next read is a clean miss
    // rather than a repeated alarm nobody can clear.
    expect(client.data.has(key)).toBe(false);
    expect(await cache.get(BASE)).toBeUndefined();
  });

  it('a tenant-scoped scan confirms ownership against the entry, not the glob', async () => {
    const client = new FakeValkey();
    const cache = new ValkeyResultCache(client);
    await cache.set(BASE, { v: 1 }, { ttlSeconds: 300, rowCount: 1 });
    // An entry whose KEY matches t_alpha's prefix but whose owner is t_beta.
    // SCAN MATCH is a glob; the glob would include it, the ownership check
    // must not.
    client.data.set(
      `${tenantKeyPrefix('t_alpha')}forged`,
      JSON.stringify({ ...JSON.parse(client.data.get(buildCacheKey(BASE)) as string), tenant_id: 't_beta' }),
    );
    const dropped = await cache.invalidateTenant('t_alpha');
    expect(dropped).toBe(1);
  });
});

describe('BI-P020 — the Valkey adapter', () => {
  it('sets a store-side TTL, not only a recorded timestamp', async () => {
    const client = new FakeValkey();
    const cache = new ValkeyResultCache(client);
    await cache.set(BASE, { v: 1 }, { ttlSeconds: 42, rowCount: 1 });
    expect(client.ttls.get(buildCacheKey(BASE))).toBe(42);
  });

  it('treats a corrupt entry as a miss AND removes it', async () => {
    const client = new FakeValkey();
    const cache = new ValkeyResultCache(client);
    client.data.set(buildCacheKey(BASE), '{not json');
    expect(await cache.get(BASE)).toBeUndefined();
    expect(client.data.has(buildCacheKey(BASE))).toBe(false);
  });

  it('pages through SCAN rather than blocking on KEYS', async () => {
    const client = new FakeValkey();
    const cache = new ValkeyResultCache(client);
    for (let i = 0; i < 1200; i += 1) {
      await cache.set(
        { ...BASE, compiled_query_hash: String(i).padStart(64, '0') },
        { i },
        { ttlSeconds: 300, rowCount: 1 },
      );
    }
    expect(await cache.invalidateTenant('t_alpha')).toBe(1200);
    expect(client.scanCalls).toBeGreaterThan(1);
  });
});

class FakeValkey implements ValkeyLikeClient {
  readonly data = new Map<string, string>();
  readonly ttls = new Map<string, number>();
  scanCalls = 0;

  async get(key: string): Promise<string | null> {
    return this.data.get(key) ?? null;
  }
  async set(key: string, value: string, opts: { EX: number }): Promise<unknown> {
    this.data.set(key, value);
    this.ttls.set(key, opts.EX);
    return 'OK';
  }
  async del(keys: string[]): Promise<unknown> {
    for (const k of keys) {
      this.data.delete(k);
      this.ttls.delete(k);
    }
    return keys.length;
  }
  async scan(
    cursor: string,
    opts: { MATCH: string; COUNT: number },
  ): Promise<{ cursor: string; keys: string[] }> {
    this.scanCalls += 1;
    const prefix = opts.MATCH.replace(/\*$/, '');
    const all = [...this.data.keys()].filter((k) => k.startsWith(prefix));
    const start = Number(cursor);
    const page = all.slice(start, start + opts.COUNT);
    const next = start + opts.COUNT >= all.length ? '0' : String(start + opts.COUNT);
    return { cursor: next, keys: page };
  }
}