#!/usr/bin/env node
/**
 * BI-P010 DoD: "Benchmark representative VehIQ-scale fixture and generic SaaS
 * fixture."
 *
 * Builds both fixtures in a THROWAWAY database on a ClickHouse the operator
 * points it at, times the query shapes each schema exists to serve, and writes
 * a machine-readable transcript to
 * evidence/bi-p002-p012/bi-p010-benchmark.json.
 *
 *   CLICKHOUSE_URL=http://localhost:18123 \
 *   CLICKHOUSE_AUTH=user:pass \
 *     node scripts/bi-p010-benchmark.mjs
 *
 * It also measures the COUNTERFACTUAL the program warns about — "one giant
 * unbounded table for every analytics use case" — by running the same
 * questions against a single wide table with a non-tenant-leading sort key.
 * Without that comparison the split is an assertion; with it, it is a number.
 *
 * NEVER point this at the b0ts ClickHouse or any quarantined market data: it
 * CREATEs and DROPs a database.
 */
import fs from "node:fs";
import path from "node:path";

const URL_ = process.env.CLICKHOUSE_URL;
const AUTH = process.env.CLICKHOUSE_AUTH;
if (!URL_) {
  console.error("CLICKHOUSE_URL is required. Refusing to invent numbers.");
  process.exit(2);
}

const DB = `opt_bench_${Date.now().toString(36)}`;
const headers = AUTH ? { Authorization: `Basic ${Buffer.from(AUTH).toString("base64")}` } : {};

async function ch(sql, body, settings = {}) {
  const u = new URL(URL_);
  u.searchParams.set("query", sql);
  for (const [k, v] of Object.entries(settings)) u.searchParams.set(k, v);
  const res = await fetch(u.toString(), { method: "POST", headers, body: body ?? "" });
  const text = await res.text();
  if (!res.ok) throw new Error(`CH ${res.status}: ${text.slice(0, 400)}`);
  return text;
}

async function json(sql, settings) {
  const t = await ch(sql, undefined, settings);
  return t ? JSON.parse(t) : { data: [] };
}

/** Run `sql` n times, return {p50, p95, min, max, rows_read, bytes_read}. */
async function timeQuery(label, sql, n = 12) {
  // One warm-up so we time the query, not the page cache miss.
  await ch(sql);
  const times = [];
  let stats = {};
  for (let i = 0; i < n; i++) {
    const t0 = process.hrtime.bigint();
    const out = await json(sql);
    times.push(Number(process.hrtime.bigint() - t0) / 1e6);
    stats = out.statistics ?? stats;
  }
  times.sort((a, b) => a - b);
  const pick = (p) => Number(times[Math.min(times.length - 1, Math.floor((times.length * p) / 100))].toFixed(2));
  return {
    label,
    runs: n,
    p50_ms: pick(50),
    p95_ms: pick(95),
    min_ms: Number(times[0].toFixed(2)),
    max_ms: Number(times[times.length - 1].toFixed(2)),
    rows_read: stats.rows_read ?? null,
    bytes_read: stats.bytes_read ?? null,
  };
}

const evidence = {
  generated_at: new Date().toISOString(),
  clickhouse_version: null,
  database: DB,
  fixtures: {},
  measurements: [],
  notes: [],
};

async function main() {
  evidence.clickhouse_version = (await ch("SELECT version() FORMAT TSVRaw")).trim();
  await ch(`CREATE DATABASE ${DB}`);

  // ── the BI-P010 schema under test ──────────────────────────────────────
  await ch(`CREATE TABLE ${DB}.opt_analytics_event (
      tenant_id LowCardinality(String), project_id LowCardinality(String),
      dataset_id LowCardinality(String), metric LowCardinality(String),
      event_id String, event_time DateTime64(3,'UTC'), value Float64,
      dim_names Array(LowCardinality(String)), dim_values Array(String),
      ingested_at DateTime64(3,'UTC') DEFAULT now64(3),
      INDEX idx_dim_values dim_values TYPE bloom_filter(0.01) GRANULARITY 4)
    ENGINE = MergeTree() PARTITION BY toYYYYMM(event_time)
    ORDER BY (tenant_id, project_id, dataset_id, metric, event_time)`);

  await ch(`CREATE TABLE ${DB}.opt_metric_daily (
      tenant_id LowCardinality(String), project_id LowCardinality(String),
      dataset_id LowCardinality(String), metric LowCardinality(String), day Date,
      sum_state AggregateFunction(sum, Float64), count_state AggregateFunction(count, Float64),
      min_state AggregateFunction(min, Float64), max_state AggregateFunction(max, Float64))
    ENGINE = AggregatingMergeTree() PARTITION BY toYYYYMM(day)
    ORDER BY (tenant_id, project_id, dataset_id, metric, day)`);

  await ch(`CREATE MATERIALIZED VIEW ${DB}.opt_metric_daily_mv TO ${DB}.opt_metric_daily AS
    SELECT tenant_id, project_id, dataset_id, metric, toDate(event_time) AS day,
           sumState(value) AS sum_state, countState(value) AS count_state,
           minState(value) AS min_state, maxState(value) AS max_state
    FROM ${DB}.opt_analytics_event
    GROUP BY tenant_id, project_id, dataset_id, metric, day`);

  await ch(`CREATE TABLE ${DB}.opt_dataset_projection (
      tenant_id LowCardinality(String), project_id LowCardinality(String),
      dataset_id LowCardinality(String), entity_key String,
      as_of DateTime64(3,'UTC'), version UInt64,
      attributes Map(String,String), measures Map(String,Float64),
      INDEX idx_entity entity_key TYPE bloom_filter(0.01) GRANULARITY 4)
    ENGINE = ReplacingMergeTree(version) PARTITION BY (tenant_id)
    ORDER BY (tenant_id, project_id, dataset_id, entity_key)`);

  // ── the COUNTERFACTUAL the program warns about ─────────────────────────
  // One table for everything, sorted by time (the shape people reach for when
  // "we'll just put it all in events").
  await ch(`CREATE TABLE ${DB}.giant_everything (
      tenant_id String, project_id String, dataset_id String, metric String,
      entity_key String, event_time DateTime64(3,'UTC'), value Float64,
      dim_names Array(String), dim_values Array(String),
      attributes Map(String,String), measures Map(String,Float64))
    ENGINE = MergeTree() PARTITION BY toYYYYMM(event_time)
    ORDER BY (event_time, tenant_id)`);

  // ── FIXTURE 1: generic SaaS event stream ───────────────────────────────
  // 200 tenants x 12 metrics x 90 days x 25 events/day ~= 5.4M events.
  const SAAS_TENANTS = 200, SAAS_METRICS = 12, SAAS_DAYS = 90, SAAS_PER_DAY = 25;
  const saasRows = SAAS_TENANTS * SAAS_METRICS * SAAS_DAYS * SAAS_PER_DAY;
  console.log(`seeding SaaS fixture: ${saasRows.toLocaleString()} events...`);
  const t0saas = Date.now();
  await ch(`INSERT INTO ${DB}.opt_analytics_event
    SELECT concat('t_', toString(number % ${SAAS_TENANTS})) AS tenant_id,
           'p_main' AS project_id,
           concat('ds_', toString(number % 8)) AS dataset_id,
           concat('metric_', toString(number % ${SAAS_METRICS})) AS metric,
           toString(number) AS event_id,
           toDateTime64('2026-06-01 00:00:00', 3, 'UTC') + toIntervalSecond(number % ${SAAS_DAYS * 86400}) AS event_time,
           toFloat64(number % 997) AS value,
           ['plan','region'] AS dim_names,
           [if(number % 3 = 0, 'pro', 'free'), if(number % 5 = 0, 'eu-north', 'eu-central')] AS dim_values,
           now64(3) AS ingested_at
    FROM numbers(${saasRows})`);
  const saasSeedMs = Date.now() - t0saas;

  // ── FIXTURE 2: VehIQ-scale vehicle projections ─────────────────────────
  // 2M vehicles for the tenant under test, plus 2M belonging to other
  // tenants — so a tenant-scoped lookup has to prove it prunes.
  const VEHICLES = 2_000_000, DECOYS = 2_000_000;
  console.log(`seeding VehIQ fixture: ${(VEHICLES + DECOYS).toLocaleString()} vehicle projections...`);
  const t0veh = Date.now();
  await ch(`INSERT INTO ${DB}.opt_dataset_projection
    SELECT 't_vehiq' AS tenant_id, 'p_main' AS project_id, 'ds_vehicles' AS dataset_id,
           concat('YV1', leftPad(toString(number), 14, '0')) AS entity_key,
           toDateTime64('2026-09-01 00:00:00', 3, 'UTC') AS as_of,
           1 AS version,
           map('make', arrayElement(['volvo','saab','audi','vw','toyota'], (number % 5) + 1),
               'fuel', arrayElement(['diesel','petrol','ev','hybrid'], (number % 4) + 1)) AS attributes,
           map('asking_price', toFloat64(50000 + number % 900000),
               'mileage_km', toFloat64(number % 400000)) AS measures
    FROM numbers(${VEHICLES})`);
  await ch(`INSERT INTO ${DB}.opt_dataset_projection
    SELECT concat('t_other_', toString(number % 50)) AS tenant_id, 'p_main', 'ds_vehicles',
           concat('WBA', leftPad(toString(number), 14, '0')),
           toDateTime64('2026-09-01 00:00:00', 3, 'UTC'), 1,
           map('make','bmw'), map('asking_price', toFloat64(number % 500000))
    FROM numbers(${DECOYS})`);
  const vehSeedMs = Date.now() - t0veh;

  // ── the same data in the giant table ───────────────────────────────────
  console.log("seeding the counterfactual giant table...");
  await ch(`INSERT INTO ${DB}.giant_everything
    SELECT tenant_id, project_id, dataset_id, metric, '' AS entity_key, event_time, value,
           dim_names, dim_values, map() AS attributes, map() AS measures
    FROM ${DB}.opt_analytics_event`);
  await ch(`INSERT INTO ${DB}.giant_everything
    SELECT tenant_id, project_id, dataset_id, '' AS metric, entity_key,
           as_of AS event_time, 0 AS value, [] AS dim_names, [] AS dim_values, attributes, measures
    FROM ${DB}.opt_dataset_projection`);

  await ch(`OPTIMIZE TABLE ${DB}.opt_analytics_event FINAL`, undefined, { max_execution_time: "600" });
  await ch(`OPTIMIZE TABLE ${DB}.opt_dataset_projection FINAL`, undefined, { max_execution_time: "600" });
  await ch(`OPTIMIZE TABLE ${DB}.giant_everything FINAL`, undefined, { max_execution_time: "900" });

  const sizes = await json(`SELECT table, sum(rows) AS rows, sum(bytes_on_disk) AS bytes
    FROM system.parts WHERE database='${DB}' AND active GROUP BY table ORDER BY table FORMAT JSON`);
  evidence.fixtures = {
    saas: { events: saasRows, tenants: SAAS_TENANTS, metrics: SAAS_METRICS, days: SAAS_DAYS, seed_ms: saasSeedMs },
    vehiq: { vehicles_under_test: VEHICLES, decoy_vehicles: DECOYS, seed_ms: vehSeedMs },
    on_disk: sizes.data.map((r) => ({ table: r.table, rows: Number(r.rows), bytes: Number(r.bytes) })),
  };

  // ── measurements ───────────────────────────────────────────────────────
  const T = "t_57"; // one representative SaaS tenant

  evidence.measurements.push(
    await timeQuery(
      "SaaS dashboard panel: 90-day metric total for ONE tenant (purpose-built event table)",
      `SELECT sum(value) AS value FROM ${DB}.opt_analytics_event
       WHERE tenant_id = '${T}' AND metric = 'metric_3' FORMAT JSON`,
    ),
    await timeQuery(
      "SAME question against ONE GIANT TABLE (time-leading sort key)",
      `SELECT sum(value) AS value FROM ${DB}.giant_everything
       WHERE tenant_id = '${T}' AND metric = 'metric_3' FORMAT JSON`,
    ),
    await timeQuery(
      "SaaS time series: daily series for ONE tenant from the AggregatingMergeTree rollup",
      `SELECT day, sumMerge(sum_state) AS value FROM ${DB}.opt_metric_daily
       WHERE tenant_id = '${T}' AND metric = 'metric_3' GROUP BY day ORDER BY day FORMAT JSON`,
    ),
    await timeQuery(
      "SAME time series computed from raw events instead of the rollup",
      `SELECT toDate(event_time) AS day, sum(value) AS value FROM ${DB}.opt_analytics_event
       WHERE tenant_id = '${T}' AND metric = 'metric_3' GROUP BY day ORDER BY day FORMAT JSON`,
    ),
    await timeQuery(
      "SaaS group-by: one tenant sliced by a dimension",
      `SELECT dim_values[indexOf(dim_names, 'plan')] AS grp, count() AS value
       FROM ${DB}.opt_analytics_event WHERE tenant_id = '${T}' AND metric = 'metric_3'
       GROUP BY grp ORDER BY value DESC FORMAT JSON`,
    ),
    await timeQuery(
      "VehIQ point lookup: one vehicle by key, tenant-scoped (purpose-built projection)",
      `SELECT entity_key, measures['asking_price'] AS p FROM ${DB}.opt_dataset_projection
       WHERE tenant_id = 't_vehiq' AND project_id = 'p_main' AND dataset_id = 'ds_vehicles'
         AND entity_key = 'YV100000001234567' FORMAT JSON`,
    ),
    await timeQuery(
      "SAME point lookup against ONE GIANT TABLE",
      `SELECT entity_key, measures['asking_price'] AS p FROM ${DB}.giant_everything
       WHERE tenant_id = 't_vehiq' AND dataset_id = 'ds_vehicles'
         AND entity_key = 'YV100000001234567' FORMAT JSON`,
    ),
    await timeQuery(
      "VehIQ aggregate: average asking price by make for ONE tenant across 2M vehicles",
      `SELECT attributes['make'] AS grp, avg(measures['asking_price']) AS value
       FROM ${DB}.opt_dataset_projection
       WHERE tenant_id = 't_vehiq' AND project_id = 'p_main' AND dataset_id = 'ds_vehicles'
       GROUP BY grp ORDER BY value DESC FORMAT JSON`,
      8,
    ),
    await timeQuery(
      "SAME aggregate against ONE GIANT TABLE",
      `SELECT attributes['make'] AS grp, avg(measures['asking_price']) AS value
       FROM ${DB}.giant_everything
       WHERE tenant_id = 't_vehiq' AND dataset_id = 'ds_vehicles' AND entity_key != ''
       GROUP BY grp ORDER BY value DESC FORMAT JSON`,
      8,
    ),
  );

  // Index pruning evidence: how many granules the tenant predicate skips.
  const plan = await ch(
    `EXPLAIN indexes = 1 SELECT sum(value) FROM ${DB}.opt_analytics_event
     WHERE tenant_id = '${T}' AND metric = 'metric_3' FORMAT TSVRaw`,
  );
  const giantPlan = await ch(
    `EXPLAIN indexes = 1 SELECT sum(value) FROM ${DB}.giant_everything
     WHERE tenant_id = '${T}' AND metric = 'metric_3' FORMAT TSVRaw`,
  );
  const granules = (t) => {
    const m = [...t.matchAll(/Granules: (\d+)\/(\d+)/g)].pop();
    return m ? { read: Number(m[1]), total: Number(m[2]) } : null;
  };
  evidence.index_pruning = {
    purpose_built_event_table: granules(plan),
    one_giant_table: granules(giantPlan),
    note:
      "Granules read/total for the SAME question. The tenant-leading ORDER BY is what turns the " +
      "forced tenant predicate (BI-P011) into a primary-key range scan instead of a full scan.",
  };

  evidence.notes.push(
    "Fixtures are synthetic and generated in-server with numbers(); no production or b0ts data was read or written.",
    "Each timing is the p50/p95 of repeated runs after one warm-up, measured client-side over HTTP on the same host.",
    "The 'ONE GIANT TABLE' rows are the counterfactual BI-P010 warns against, measured rather than asserted.",
  );

  const out = path.join(process.cwd(), "evidence", "bi-p002-p012", "bi-p010-benchmark.json");
  fs.writeFileSync(out, JSON.stringify(evidence, null, 2) + "\n");
  console.log(`wrote ${out}`);
  for (const m of evidence.measurements) {
    console.log(`  ${m.p50_ms.toString().padStart(9)} ms p50  ${String(m.rows_read).padStart(10)} rows  ${m.label}`);
  }
  console.log("  pruning:", JSON.stringify(evidence.index_pruning));
}

main()
  .catch((e) => {
    console.error(e);
    process.exitCode = 1;
  })
  .finally(async () => {
    try {
      await ch(`DROP DATABASE IF EXISTS ${DB}`);
      console.log(`dropped ${DB}`);
    } catch (e) {
      console.error("cleanup failed:", e.message);
    }
  });