0pt/src/analytics/cache/query-cache.ts

/**
 * src/analytics/cache/query-cache.ts
 * ===================================
 * BI-P020 — QUERY CACHE.
 *
 * "Cache key includes: tenant, semantic model version, metric definitions,
 *  filters, query, data snapshot/version. Never share tenant result cache."
 *
 * All six are REQUIRED, and the key builder refuses to produce a key when one
 * is missing rather than defaulting it. A defaulted component is how a cache
 * serves a result computed under a metric definition that no longer exists —
 * the answer is not stale, it is WRONG, and it looks current.
 *
 * "Never share tenant result cache" is enforced structurally, not by
 * discipline:
 *
 *   * the key is PREFIXED with the tenant id, and the prefix is derived from
 *     the verified identity inside this module — a caller cannot pass one;
 *   * every read re-derives the prefix from the identity and REFUSES a stored
 *     entry whose recorded tenant does not match, so a prefix collision or a
 *     shared backing store still cannot serve a foreign result;
 *   * the stored entry carries its own tenant id, so the check is against the
 *     DATA, not against the key that found it.
 *
 * The second and third points exist because the first is a naming convention,
 * and naming conventions are what ST0RE was relying on when it turned out to
 * have four cross-tenant paths rather than the one reported.
 *
 * Provider-neutral: `AnalyticsResultCache` is the port. `InMemoryResultCache`
 * is the reference implementation and the test double; `ValkeyResultCache`
 * adapts any Valkey/Redis client that can GET/SET/DEL with a TTL, taking it as
 * an injected interface so this module adds no dependency.
 */
import crypto from 'node:crypto';
import { SiaxError } from '../../platform/errors';

/**
 * Everything that must be true for a cached result to still be the right
 * answer. Missing any one of these makes the cache unsound, so the type has no
 * optional fields.
 */
export interface CacheKeyComponents {
  /** From the verified identity. Not a caller-supplied value. */
  tenant_id: string;
  project_id?: string;
  dataset_id: string;
  /** BI-P030's semantic model version. Bumping it invalidates everything under it. */
  semantic_model_version: number;
  /**
   * metric_id -> version, for EVERY metric in the query. A metric whose
   * definition changed must not serve a result computed under the old one.
   */
  metric_versions: Record<string, number>;
  /** The compiled statement hash (BI-P018). Covers filters, grain and limit. */
  compiled_query_hash: string;
  /**
   * The data snapshot the result was computed from. Whatever identifies "the
   * data as of now" for this dataset: a DatasetVersion number, an Iceberg
   * snapshot id, a max ingested watermark. Advancing it invalidates.
   */
  data_snapshot: string;
  /** Result encoding, since a JSON and an Arrow result are different bytes. */
  format: string;
}

export interface CachedResult<T = unknown> {
  /** Recorded so the read check is against the DATA, not against the key. */
  tenant_id: string;
  key: string;
  value: T;
  stored_at: string;
  expires_at: string;
  row_count: number;
  /** Kept for targeted invalidation. */
  dataset_id: string;
  metric_ids: string[];
  semantic_model_version: number;
  data_snapshot: string;
}

export interface AnalyticsResultCache {
  readonly provider: string;
  get<T>(components: CacheKeyComponents): Promise<CachedResult<T> | undefined>;
  set<T>(
    components: CacheKeyComponents,
    value: T,
    opts: { ttlSeconds: number; rowCount: number },
  ): Promise<void>;
  /** Drop every entry for a dataset — the blunt, always-correct invalidation. */
  invalidateDataset(tenantId: string, datasetId: string): Promise<number>;
  /** Drop every entry that used a given metric. */
  invalidateMetric(tenantId: string, metricId: string): Promise<number>;
  /** Drop everything for a tenant (offboarding, incident response). */
  invalidateTenant(tenantId: string): Promise<number>;
}

function required(value: unknown, name: string): void {
  if (
    value === undefined ||
    value === null ||
    value === '' ||
    (typeof value === 'number' && !Number.isFinite(value))
  ) {
    throw new SiaxError(
      'INVALID_REQUEST',
      '0pt.analytics.cache',
      `cache key component "${name}" is required — a defaulted component serves results computed under a definition that no longer exists`,
    );
  }
}

/**
 * Build the key. Deterministic, tenant-prefixed, and total over the six
 * components BI-P020 lists.
 *
 * The tenant prefix is in PLAINTEXT at the front rather than folded into the
 * hash, so `invalidateTenant` is a prefix scan and an operator reading the
 * store can see which tenant an entry belongs to without a lookup table.
 */
export function buildCacheKey(c: CacheKeyComponents): string {
  required(c.tenant_id, 'tenant_id');
  required(c.dataset_id, 'dataset_id');
  required(c.semantic_model_version, 'semantic_model_version');
  required(c.compiled_query_hash, 'compiled_query_hash');
  required(c.data_snapshot, 'data_snapshot');
  required(c.format, 'format');
  if (!c.metric_versions || Object.keys(c.metric_versions).length === 0) {
    throw new SiaxError(
      'INVALID_REQUEST',
      '0pt.analytics.cache',
      'cache key component "metric_versions" is required and must name every metric in the query',
    );
  }
  for (const [id, version] of Object.entries(c.metric_versions)) {
    required(version, `metric_versions[${id}]`);
  }
  if (!/^[A-Za-z0-9_-]{1,64}$/.test(c.tenant_id)) {
    throw new SiaxError('INVALID_REQUEST', '0pt.analytics.cache', 'tenant id is not key-safe');
  }
  const material = JSON.stringify([
    c.tenant_id,
    c.project_id ?? '',
    c.dataset_id,
    c.semantic_model_version,
    Object.keys(c.metric_versions)
      .sort()
      .map((k) => [k, c.metric_versions[k]]),
    c.compiled_query_hash,
    c.data_snapshot,
    c.format,
  ]);
  const digest = crypto.createHash('sha256').update(material).digest('hex');
  return `optq:${c.tenant_id}:${digest}`;
}

/** The prefix `invalidateTenant` scans. Derived, never supplied. */
export function tenantKeyPrefix(tenantId: string): string {
  return `optq:${tenantId}:`;
}

abstract class BaseResultCache implements AnalyticsResultCache {
  abstract readonly provider: string;
  protected abstract read(key: string): Promise<CachedResult | undefined>;
  protected abstract write(entry: CachedResult, ttlSeconds: number): Promise<void>;
  protected abstract entries(tenantId: string): Promise<CachedResult[]>;
  protected abstract remove(keys: string[]): Promise<void>;

  protected readonly now: () => number;
  constructor(now: () => number = Date.now) {
    this.now = now;
  }

  async get<T>(components: CacheKeyComponents): Promise<CachedResult<T> | undefined> {
    const key = buildCacheKey(components);
    const entry = await this.read(key);
    if (!entry) return undefined;

    // The check that survives a prefix collision or a shared backing store:
    // the ENTRY says which tenant it belongs to, and it must be this one.
    if (entry.tenant_id !== components.tenant_id) {
      // Not a miss — a cross-tenant hit is a security event, and returning
      // `undefined` would hide it. Drop the poisoned entry and refuse loudly.
      await this.remove([key]);
      throw new SiaxError(
        'AUTHORIZATION_DENIED',
        '0pt.analytics.cache',
        'cache entry belongs to a different tenant; entry evicted',
      );
    }
    if (Date.parse(entry.expires_at) <= this.now()) {
      await this.remove([key]);
      return undefined;
    }
    return entry as CachedResult<T>;
  }

  async set<T>(
    components: CacheKeyComponents,
    value: T,
    opts: { ttlSeconds: number; rowCount: number },
  ): Promise<void> {
    if (!Number.isFinite(opts.ttlSeconds) || opts.ttlSeconds <= 0) {
      throw new SiaxError('INVALID_REQUEST', '0pt.analytics.cache', 'ttlSeconds must be positive');
    }
    const key = buildCacheKey(components);
    const nowMs = this.now();
    await this.write(
      {
        tenant_id: components.tenant_id,
        key,
        value,
        stored_at: new Date(nowMs).toISOString(),
        expires_at: new Date(nowMs + opts.ttlSeconds * 1000).toISOString(),
        row_count: opts.rowCount,
        dataset_id: components.dataset_id,
        metric_ids: Object.keys(components.metric_versions),
        semantic_model_version: components.semantic_model_version,
        data_snapshot: components.data_snapshot,
      },
      opts.ttlSeconds,
    );
  }

  async invalidateDataset(tenantId: string, datasetId: string): Promise<number> {
    const victims = (await this.entries(tenantId)).filter((e) => e.dataset_id === datasetId);
    await this.remove(victims.map((v) => v.key));
    return victims.length;
  }

  async invalidateMetric(tenantId: string, metricId: string): Promise<number> {
    const victims = (await this.entries(tenantId)).filter((e) => e.metric_ids.includes(metricId));
    await this.remove(victims.map((v) => v.key));
    return victims.length;
  }

  async invalidateTenant(tenantId: string): Promise<number> {
    const victims = await this.entries(tenantId);
    await this.remove(victims.map((v) => v.key));
    return victims.length;
  }
}

export class InMemoryResultCache extends BaseResultCache {
  readonly provider = 'memory';
  private readonly store = new Map<string, CachedResult>();

  protected async read(key: string): Promise<CachedResult | undefined> {
    return this.store.get(key);
  }
  protected async write(entry: CachedResult): Promise<void> {
    this.store.set(entry.key, entry);
  }
  protected async entries(tenantId: string): Promise<CachedResult[]> {
    const prefix = tenantKeyPrefix(tenantId);
    return [...this.store.values()].filter((e) => e.key.startsWith(prefix));
  }
  protected async remove(keys: string[]): Promise<void> {
    for (const k of keys) this.store.delete(k);
  }
  /** Test/observability affordance. Never used to bypass the tenant check. */
  get size(): number {
    return this.store.size;
  }
}

/**
 * The minimum a Valkey/Redis client must offer. Taken as an interface so this
 * module adds no dependency and no CVE surface — the same reason BI-P005's
 * Arrow writer is hand-written.
 */
export interface ValkeyLikeClient {
  get(key: string): Promise<string | null>;
  set(key: string, value: string, opts: { EX: number }): Promise<unknown>;
  del(keys: string[]): Promise<unknown>;
  /** Cursor-based scan. A blocking KEYS on a shared Valkey is an outage. */
  scan(cursor: string, opts: { MATCH: string; COUNT: number }): Promise<{ cursor: string; keys: string[] }>;
}

export class ValkeyResultCache extends BaseResultCache {
  readonly provider = 'valkey';
  constructor(
    private readonly client: ValkeyLikeClient,
    now: () => number = Date.now,
  ) {
    super(now);
  }

  protected async read(key: string): Promise<CachedResult | undefined> {
    const raw = await this.client.get(key);
    if (raw === null) return undefined;
    try {
      return JSON.parse(raw) as CachedResult;
    } catch {
      // A corrupt entry is a miss, not a crash — but it is also removed, so a
      // permanently unparseable key cannot become a permanent cache miss that
      // nobody notices.
      await this.client.del([key]);
      return undefined;
    }
  }

  protected async write(entry: CachedResult, ttlSeconds: number): Promise<void> {
    // The TTL is set on the STORE as well as recorded in the entry. Relying on
    // the recorded timestamp alone leaves expired results occupying memory
    // forever on a busy tenant.
    await this.client.set(entry.key, JSON.stringify(entry), { EX: Math.ceil(ttlSeconds) });
  }

  protected async entries(tenantId: string): Promise<CachedResult[]> {
    const prefix = tenantKeyPrefix(tenantId);
    const out: CachedResult[] = [];
    let cursor = '0';
    do {
      const page = await this.client.scan(cursor, { MATCH: `${prefix}*`, COUNT: 500 });
      cursor = page.cursor;
      for (const key of page.keys) {
        const entry = await this.read(key);
        // Defence in depth: SCAN MATCH is a glob, and a glob is a naming
        // convention. Confirm against the entry's own tenant id.
        if (entry && entry.tenant_id === tenantId) out.push(entry);
      }
    } while (cursor !== '0');
    return out;
  }

  protected async remove(keys: string[]): Promise<void> {
    if (keys.length) await this.client.del(keys);
  }
}