0pt/tests/integration/analytics-engine-clickhouse.integration.test.ts

/**
 * BI-P009 (canonical OLAP provider) + BI-P010 (production schema), against a
 * LIVE ClickHouse.
 *
 * Every one of the eight `SiaxAnalyticsEngine` capabilities is exercised for
 * real here. A capability that only has a unit test with a stubbed fetch is a
 * capability whose SQL has never been parsed by ClickHouse — `EXPLAIN
 * ESTIMATE`, `KILL QUERY` and `OPTIMIZE ... PARTITION` in particular are
 * syntax the server either accepts or does not, and no mock can tell you
 * which.
 *
 * Gated on CLICKHOUSE_URL; skips honestly when unset. CI's `integration` job
 * supplies it and hard-fails on a skipped suite.
 */
import { describe, it, expect, beforeAll, afterAll } from "vitest";
import { ClickHouseAnalyticsEngine } from "@/src/analytics/engine/clickhouse-engine";
import { ANALYTICS_MATERIALIZED_VIEWS, ANALYTICS_TABLES } from "@/src/analytics/engine/schema";
import type { AnalyticsIdentity } from "@/src/analytics/engine/siax-analytics-engine";


/**
 * Raw events carry a 30-day TTL (src/analytics/engine/schema.ts). Fixed
 * historical timestamps silently expire and make live-ClickHouse suites read
 * 0 rows, so seed rows are stamped relative to now (UTC, ClickHouse
 * DateTime64(3) text form).
 */
const SEED_EVENT_TIME = new Date(Date.now() - 2 * 3600_000).toISOString().replace("T", " ").replace("Z", "");

const CLICKHOUSE_URL = process.env.CLICKHOUSE_URL;
const CLICKHOUSE_AUTH = process.env.CLICKHOUSE_AUTH;
const run = CLICKHOUSE_URL ? describe : describe.skip;

const DB = `opt_engine_${Date.now().toString(36)}`;
const ID: AnalyticsIdentity = { tenant_id: "t_alpha", project_id: "p_main", principal_id: "usr_alpha" };

let engine: ClickHouseAnalyticsEngine;

async function raw(sql: string, body?: string): Promise<string> {
  const url = new URL(CLICKHOUSE_URL ?? "http://skipped.invalid");
  url.searchParams.set("query", sql);
  const headers: Record<string, string> = {};
  if (CLICKHOUSE_AUTH) headers["Authorization"] = `Basic ${Buffer.from(CLICKHOUSE_AUTH).toString("base64")}`;
  const res = await fetch(url.toString(), { method: "POST", headers, body: body ?? "" });
  const text = await res.text();
  if (!res.ok) throw new Error(`CH ${res.status}: ${text.slice(0, 300)}`);
  return text;
}

run("BI-P009/BI-P010 ClickHouse provider against a LIVE server", () => {
  beforeAll(async () => {
    engine = new ClickHouseAnalyticsEngine({
      baseUrl: CLICKHOUSE_URL!,
      basicAuth: CLICKHOUSE_AUTH,
      database: DB,
      dimensionCatalogs: { t_alpha: { severity: "severity", make: "vehicle_make" } },
    });
    await engine.ensureSchema();
    await engine.ensureSchema(); // idempotent, as the DDL claims
  }, 120_000);

  afterAll(async () => {
    if (CLICKHOUSE_URL) await raw(`DROP DATABASE IF EXISTS ${DB}`);
  }, 60_000);

  // ── capability 1: health ────────────────────────────────────────────────

  it("health() reports the live server's version", async () => {
    const h = await engine.health();
    expect(h.healthy).toBe(true);
    expect(h.provider).toBe("clickhouse");
    expect(h.version).toMatch(/^\d+\.\d+/);
  });

  // ── BI-P010: the schema is what it claims to be ─────────────────────────

  it("BI-P010: four purpose-built tables exist, NOT one giant table", async () => {
    const listed = await raw(
      `SELECT name FROM system.tables WHERE database = '${DB}' AND engine != 'MaterializedView' ORDER BY name FORMAT TSV`,
    );
    const names = listed.trim().split("\n").filter(Boolean);
    // The three this module owns, PLUS the preserved opt_metric_events.
    for (const t of ANALYTICS_TABLES) expect(names).toContain(t);
    expect(names).toContain("opt_metric_events");
    expect(names.length).toBeGreaterThanOrEqual(4);
  });

  it("BI-P010: the tables use different engines and different sort keys — the split is real", async () => {
    const meta = await raw(
      `SELECT name, engine, sorting_key FROM system.tables WHERE database = '${DB}' ORDER BY name FORMAT JSON`,
    );
    const rows = (JSON.parse(meta) as { data: { name: string; engine: string; sorting_key: string }[] }).data;
    const by = Object.fromEntries(rows.map((r) => [r.name, r]));
    expect(by["opt_analytics_event"].engine).toBe("MergeTree");
    expect(by["opt_metric_daily"].engine).toBe("AggregatingMergeTree");
    expect(by["opt_dataset_projection"].engine).toBe("ReplacingMergeTree");
    // Different ORDER BY keys — the reason one table cannot serve both jobs.
    expect(by["opt_analytics_event"].sorting_key).not.toBe(by["opt_dataset_projection"].sorting_key);
    // Every one still leads with tenant_id.
    for (const t of ANALYTICS_TABLES) expect(by[t].sorting_key.startsWith("tenant_id")).toBe(true);
  });

  it("BI-P010: TTL and LowCardinality are actually applied, not just written in a comment", async () => {
    const ddl = await raw(`SHOW CREATE TABLE ${DB}.opt_analytics_event FORMAT TSVRaw`);
    expect(ddl).toContain("TTL");
    expect(ddl).toContain("LowCardinality(String)");
    expect(ddl).toContain("PARTITION BY toYYYYMM(event_time)");
    const rollup = await raw(`SHOW CREATE TABLE ${DB}.opt_metric_daily FORMAT TSVRaw`);
    expect(rollup).toContain("AggregateFunction(sum, Float64)");
  });

  it("BI-P010: the materialized view actually populates the rollup", async () => {
    const views = await raw(
      `SELECT name FROM system.tables WHERE database='${DB}' AND engine='MaterializedView' FORMAT TSV`,
    );
    for (const v of ANALYTICS_MATERIALIZED_VIEWS) expect(views).toContain(v);

    await engine.ingestProjection(ID, { dataset_id: "warmup", rows: [{ entity_key: "w1" }] });
    const rows = Array.from({ length: 20 }, (_, i) =>
      JSON.stringify({
        tenant_id: "t_alpha",
        project_id: "p_main",
        dataset_id: "ds_mv",
        metric: "mv_metric",
        event_id: `mv${i}`,
        event_time: SEED_EVENT_TIME,
        value: 2,
        dim_names: [],
        dim_values: [],
      }),
    ).join("\n");
    await raw(`INSERT INTO ${DB}.opt_analytics_event FORMAT JSONEachRow`, rows);

    const rolled = await raw(
      `SELECT sumMerge(sum_state) AS s, countMerge(count_state) AS c FROM ${DB}.opt_metric_daily ` +
        `WHERE tenant_id = 't_alpha' AND metric = 'mv_metric' FORMAT JSON`,
    );
    const d = (JSON.parse(rolled) as { data: { s: number | string; c: number | string }[] }).data[0];
    expect(Number(d.s)).toBe(40); // 20 rows x 2
    expect(Number(d.c)).toBe(20);
  });

  // ── capability 2: executeMetric (already covered by the isolation suite) ─

  it("executeMetric returns grouped rows keyed by the CALLER-FACING dimension name", async () => {
    const rows = ["warning", "critical", "critical"].map((sev, i) =>
      JSON.stringify({
        tenant_id: "t_alpha",
        project_id: "p_main",
        dataset_id: "ds_g",
        metric: "grouped",
        event_id: `g${i}`,
        event_time: SEED_EVENT_TIME,
        value: 5,
        dim_names: ["severity"],
        dim_values: [sev],
      }),
    );
    await raw(`INSERT INTO ${DB}.opt_analytics_event FORMAT JSONEachRow`, rows.join("\n"));
    const result = await engine.executeMetric(ID, {
      metric: "grouped",
      aggregate: "sum",
      group_by: ["severity"],
    });
    const critical = result.rows.find((r) => r.groups?.severity === "critical");
    expect(critical?.value).toBe(10);
    expect(result.stats.result_rows).toBe(2);
    expect(result.query_id.startsWith("opt_")).toBe(true);
  });

  // ── capability 4: estimateQuery ─────────────────────────────────────────

  it("estimateQuery gets a real EXPLAIN ESTIMATE from the server and compares it to the budget", async () => {
    const est = await engine.estimateQuery(ID, { metric: "grouped", aggregate: "sum" });
    expect(est.estimated_rows_to_read).toBeGreaterThan(0);
    expect(est.within_budget).toBe(true);
    expect(est.budget.maxRowsToRead).toBeGreaterThan(0);
  });

  it("estimateQuery REFUSES a query whose estimate exceeds a tightened quota", async () => {
    const tiny: AnalyticsIdentity = { ...ID, quota: { maxRowsToRead: 1 } };
    const est = await engine.estimateQuery(tiny, { metric: "grouped", aggregate: "sum" });
    expect(est.within_budget).toBe(false);
    expect(est.reason).toMatch(/over budget/);
  });

  // ── capability 6: explain ───────────────────────────────────────────────

  it("explain() proves the forced predicate reaches the PRIMARY KEY, pruning parts", async () => {
    const ex = await engine.explain(ID, { metric: "grouped", aggregate: "sum", group_by: ["severity"] });
    expect(ex.forced_tenant_predicate).toBe("tenant_id = {tenant:String}");
    expect(ex.table).toBe("opt_analytics_event");
    // This is the payoff of BI-P010's tenant-first ORDER BY: the server
    // reports tenant_id as a PrimaryKey key with a condition on it, i.e. the
    // isolation predicate is an index range scan, not a post-read filter.
    expect(ex.plan).toContain("ReadFromMergeTree");
    expect(ex.plan).toContain("PrimaryKey");
    expect(ex.plan).toMatch(/Keys:[\s\S]*tenant_id/);
    expect(ex.plan).toMatch(/tenant_id in \['t_alpha', 't_alpha'\]/);
    // The plan describes only the CALLER's own tenant; no other tenant id
    // may appear in a plan handed back to them.
    expect(ex.plan).not.toContain("t_beta");
  });

  // ── capability 5: cancelQuery ───────────────────────────────────────────

  it("cancelQuery issues a real KILL QUERY the server accepts", async () => {
    // Nothing is running under this id, so the honest outcome is "the KILL
    // statement was valid and matched nothing" -> true, with no exception.
    const ok = await engine.cancelQuery(ID, `${(await engine.executeMetric(ID, { metric: "grouped", aggregate: "count" })).query_id}`);
    expect(ok).toBe(true);
  });

  // ── capabilities 7 + 8: ingestProjection / refreshProjection ────────────

  it("ingestProjection writes the SERVING tier with the identity's tenant, ignoring any row-level claim", async () => {
    const res = await engine.ingestProjection(ID, {
      dataset_id: "ds_vehicles",
      rows: [
        { entity_key: "VIN001", attributes: { make: "volvo" }, measures: { price: 189000 } },
        { entity_key: "VIN002", attributes: { make: "saab" }, measures: { price: 42000 } },
      ],
    });
    expect(res.ingested).toBe(2);

    const stored = await raw(
      `SELECT uniqExact(tenant_id) AS t FROM ${DB}.opt_dataset_projection WHERE dataset_id = 'ds_vehicles' FORMAT JSON`,
    );
    expect(Number((JSON.parse(stored) as { data: { t: number | string }[] }).data[0].t)).toBe(1);
  });

  it("refreshProjection makes a re-ingest idempotent (ReplacingMergeTree + OPTIMIZE FINAL)", async () => {
    // Re-ingest the SAME entity keys at a higher version.
    await engine.ingestProjection(ID, {
      dataset_id: "ds_vehicles",
      rows: [
        { entity_key: "VIN001", version: 9_999_999_999_999, measures: { price: 175000 } },
        { entity_key: "VIN002", version: 9_999_999_999_999, measures: { price: 41000 } },
      ],
    });
    const refreshed = await engine.refreshProjection(ID, "ds_vehicles");
    expect(refreshed.row_count).toBe(2); // 2 entities, not 4 rows
    expect(refreshed.dataset_id).toBe("ds_vehicles");

    const winner = await raw(
      `SELECT measures['price'] AS p FROM ${DB}.opt_dataset_projection FINAL ` +
        `WHERE tenant_id='t_alpha' AND dataset_id='ds_vehicles' AND entity_key='VIN001' FORMAT JSON`,
    );
    expect(Number((JSON.parse(winner) as { data: { p: number | string }[] }).data[0].p)).toBe(175000);
  });

  // ── the PRESERVED path still works through the engine ───────────────────

  it("the PRESERVED ClickHouseAnalytics path still ingests and answers through the engine", async () => {
    await engine.legacyAnalytics.ingest([
      { tenant_id: "t_alpha", dataset_id: "ds_legacy", metric: "legacy_metric", value: 7 },
      { tenant_id: "t_beta", dataset_id: "ds_legacy", metric: "legacy_metric", value: 700 },
    ]);
    const res = await engine.executeLegacyMetric(ID, { metric: "legacy_metric", aggregate: "sum" });
    expect(res.rows[0].value).toBe(7); // NOT 707
  });
});