0pt/src/analytics/engine/clickhouse-engine.ts

/**
 * src/analytics/engine/clickhouse-engine.ts
 * ==========================================
 * BI-P009 — ClickHouse as the CANONICAL `SiaxAnalyticsEngine` provider.
 *
 * PRESERVATION (BI-P009's first sentence): `ClickHouseAnalytics`
 * (src/analytics/clickhouse-analytics.ts) is untouched by this change — not
 * one line. This engine COMPOSES it:
 *
 *   * `health()` delegates to `ClickHouseAnalytics.health()`.
 *   * `executeMetric()` on the pre-existing `opt_metric_events` table
 *     delegates to `ClickHouseAnalytics.queryMetric()`, i.e. keeps the
 *     already-proven forced-tenant path exactly as it is.
 *   * everything else (the BI-P010 tables) goes through
 *     `src/analytics/engine/query-compiler.ts`, which forces the same
 *     predicate for the same reason.
 *
 * "Do not regress the existing forced tenant predicate" is enforced by
 * tests/analytics/tenant-predicate-regression.test.ts, which fails if either
 * compiler stops emitting `tenant_id = {tenant:String}`.
 *
 * BI-P012 governance is applied on EVERY statement, including the delegated
 * ones — the settings ride on the HTTP request, and `readonly=1` means the
 * query text cannot re-set them.
 *
 * LEGAL BOUNDARY (inherited): generic 0pt analytics only. Never the b0ts
 * ClickHouse pipelines, never quarantined market data.
 */
import crypto from "node:crypto";
import { SiaxError, siaxErrorCodeForUpstreamStatus } from "../../platform/errors";
import { ClickHouseAnalytics } from "../clickhouse-analytics";
import {
  compileMetricQuery,
  compileProjectionQuery,
  dimensionCatalogFrom,
  FORCED_TENANT_PREDICATE,
  type DimensionCatalog,
} from "./query-compiler";
import { analyticsSchemaDdl, type AnalyticsSchemaOptions } from "./schema";
import {
  applyQuota,
  clickhouseSettings,
  ConcurrencyGate,
  DEFAULT_BUDGETS,
  planningSettings,
  type ResourceBudget,
} from "./resource-governance";
import type {
  AnalystQueryRequest,
  AnalyticsIdentity,
  EngineResult,
  HealthReport,
  MetricRequest,
  ProjectionIngestRequest,
  ProjectionRefreshResult,
  QueryEstimate,
  QueryExplanation,
  SiaxAnalyticsEngine,
} from "./siax-analytics-engine";

export interface ClickHouseEngineOptions {
  baseUrl: string;
  /** "user:password" — from an env/secret ref, never a literal in code. */
  basicAuth?: string;
  database?: string;
  timeoutMs?: number;
  /** Dimension allow-list per tenant. Absent tenant = no dimensions allowed. */
  dimensionCatalogs?: Record<string, Record<string, string>>;
  schema?: Omit<AnalyticsSchemaOptions, "database">;
}

const EMPTY_CATALOG: DimensionCatalog = { resolve: () => undefined };

export class ClickHouseAnalyticsEngine implements SiaxAnalyticsEngine {
  readonly provider = "clickhouse";

  /** PRESERVED pre-existing implementation, composed not replaced. */
  private readonly legacy: ClickHouseAnalytics;
  private readonly baseUrl: string;
  private readonly authHeader?: string;
  private readonly database: string;
  private readonly timeoutMs: number;
  private readonly dimensionCatalogs: Record<string, Record<string, string>>;
  private readonly schemaOpts: Omit<AnalyticsSchemaOptions, "database">;
  private readonly gate: ConcurrencyGate;
  /** query_id -> AbortController, so cancelQuery can stop the client side too. */
  private readonly inFlight = new Map<string, AbortController>();

  constructor(opts: ClickHouseEngineOptions) {
    this.baseUrl = opts.baseUrl.replace(/\/+$/, "");
    this.authHeader = opts.basicAuth ? `Basic ${Buffer.from(opts.basicAuth).toString("base64")}` : undefined;
    this.database = opts.database ?? "opt_analytics";
    this.timeoutMs = opts.timeoutMs ?? 30_000;
    this.dimensionCatalogs = opts.dimensionCatalogs ?? {};
    this.schemaOpts = opts.schema ?? {};
    this.legacy = new ClickHouseAnalytics({
      baseUrl: opts.baseUrl,
      basicAuth: opts.basicAuth,
      database: this.database,
      timeoutMs: this.timeoutMs,
    });
    this.gate = new ConcurrencyGate((tenantId) => this.budgetForTenant(tenantId).maxConcurrentQueries);
  }

  /** Access to the preserved implementation, for callers that still use it. */
  get legacyAnalytics(): ClickHouseAnalytics {
    return this.legacy;
  }

  private budgetForTenant(_tenantId: string): ResourceBudget {
    return DEFAULT_BUDGETS.INTERACTIVE;
  }

  private budgetFor(identity: AnalyticsIdentity): ResourceBudget {
    const tier = identity.budget_tier ?? "INTERACTIVE";
    return applyQuota(DEFAULT_BUDGETS[tier], identity.quota);
  }

  private catalogFor(identity: AnalyticsIdentity): DimensionCatalog {
    const pairs = this.dimensionCatalogs[identity.tenant_id];
    return pairs ? dimensionCatalogFrom(pairs) : EMPTY_CATALOG;
  }

  /** Fail closed: a missing/blank verified tenant is never a wildcard. */
  private assertIdentity(identity: AnalyticsIdentity): void {
    if (!identity?.tenant_id || !identity.tenant_id.trim()) {
      throw new SiaxError(
        "UNAUTHENTICATED",
        "0pt.analytics",
        "analytics requires a verified CL0UD identity carrying tenant_id",
      );
    }
    if (!identity.principal_id || !identity.principal_id.trim()) {
      throw new SiaxError("UNAUTHENTICATED", "0pt.analytics", "analytics requires a verified principal_id");
    }
  }

  // ── schema ───────────────────────────────────────────────────────────────

  /** Apply BI-P010 DDL (idempotent) plus the preserved legacy table. */
  async ensureSchema(): Promise<void> {
    await this.legacy.ensureSchema();
    for (const stmt of analyticsSchemaDdl({ database: this.database, ...this.schemaOpts })) {
      await this.ddl(stmt);
    }
  }

  // ── BI-P009 capability 1: health ─────────────────────────────────────────

  async health(): Promise<HealthReport> {
    try {
      const legacyOk = await this.legacy.health();
      if (!legacyOk) return { healthy: false, provider: this.provider, detail: "legacy health probe returned false" };
      const v = await this.raw("SELECT version() AS v FORMAT JSON", {}, {});
      const version = (v as { data?: { v?: string }[] })?.data?.[0]?.v;
      return { healthy: true, provider: this.provider, version };
    } catch (err) {
      return {
        healthy: false,
        provider: this.provider,
        detail: err instanceof SiaxError ? err.code : "unreachable",
      };
    }
  }

  // ── BI-P009 capability 2: executeMetric ──────────────────────────────────

  async executeMetric(identity: AnalyticsIdentity, request: MetricRequest): Promise<EngineResult> {
    this.assertIdentity(identity);
    const budget = this.budgetFor(identity);
    const release = this.gate.acquire(identity.tenant_id);
    const started = Date.now();
    const queryId = this.newQueryId(identity);
    try {
      const compiled = compileMetricQuery(
        {
          tenant_id: identity.tenant_id, // FORCED from the verified identity
          project_id: identity.project_id,
          dataset_id: request.dataset_id,
          metric: request.metric,
          aggregate: request.aggregate,
          group_by: request.group_by,
          filters: request.filters,
          from_time: request.from_time,
          to_time: request.to_time,
          limit: request.limit,
          table: request.table,
        },
        this.database,
        this.catalogFor(identity),
        budget.maxResultRows,
      );
      const json = (await this.raw(compiled.sql, compiled.params, clickhouseSettings(budget), queryId)) as
        | ClickHouseJsonResponse
        | undefined;
      const groupNames = request.group_by ?? [];
      const rows = (json?.data ?? []).map((r) => {
        const groups: Record<string, string> = {};
        groupNames.forEach((name, i) => {
          const v = (r as Record<string, unknown>)[`grp${i}`];
          groups[name] = v === undefined || v === null ? "" : String(v);
        });
        return {
          ...(groupNames.length ? { groups } : {}),
          value: Number((r as { value: number | string }).value),
        };
      });
      const limit = Math.max(1, Math.min(Math.floor(request.limit ?? 1000), budget.maxResultRows));
      return {
        rows,
        query_id: queryId,
        truncated: rows.length >= limit,
        stats: {
          duration_ms: Date.now() - started,
          rows_read: json?.statistics?.rows_read,
          bytes_read: json?.statistics?.bytes_read,
          result_rows: rows.length,
        },
      };
    } finally {
      release();
      this.inFlight.delete(queryId);
    }
  }

  /**
   * Delegating path to the PRESERVED implementation for the pre-existing
   * `opt_metric_events` table. Kept so no existing caller has to migrate and
   * the already-proven forced-predicate path stays in service.
   */
  async executeLegacyMetric(
    identity: AnalyticsIdentity,
    request: { metric: string; aggregate: "sum" | "avg" | "min" | "max" | "count"; dataset_id?: string; group_by?: string; from_time?: string; to_time?: string; limit?: number },
  ): Promise<EngineResult> {
    this.assertIdentity(identity);
    const started = Date.now();
    const rows = await this.legacy.queryMetric({
      tenant_id: identity.tenant_id, // FORCED from the verified identity
      metric: request.metric,
      aggregate: request.aggregate,
      dataset_id: request.dataset_id,
      group_by: request.group_by,
      from_time: request.from_time,
      to_time: request.to_time,
      limit: request.limit,
    });
    return {
      rows: rows.map((r) => ({
        ...(r.group !== undefined && request.group_by ? { groups: { [request.group_by]: r.group } } : {}),
        value: r.value,
      })),
      query_id: this.newQueryId(identity),
      truncated: false,
      stats: { duration_ms: Date.now() - started, result_rows: rows.length },
    };
  }

  // ── BI-P009 capability 3: executeQuery (analyst SQL) ─────────────────────

  async executeQuery(identity: AnalyticsIdentity, _request: AnalystQueryRequest): Promise<EngineResult> {
    this.assertIdentity(identity);
    // Deliberate, documented refusal — see AnalystQueryRequest's doc comment.
    // Opening a raw-SQL surface before BI-P017 (SQL dialect contract) and
    // BI-P033 (analytics authorization) exist would create precisely the
    // "raw SQL attempt" hole BI-P011's adversarial list enumerates.
    throw new SiaxError(
      "CAPABILITY_UNAVAILABLE",
      "0pt.analytics",
      "analyst SQL is not enabled on the ClickHouse provider until BI-P017/BI-P033 land",
    );
  }

  // ── BI-P009 capability 4: estimateQuery ──────────────────────────────────

  async estimateQuery(identity: AnalyticsIdentity, request: MetricRequest): Promise<QueryEstimate> {
    this.assertIdentity(identity);
    const budget = this.budgetFor(identity);
    const compiled = compileMetricQuery(
      {
        tenant_id: identity.tenant_id,
        project_id: identity.project_id,
        dataset_id: request.dataset_id,
        metric: request.metric,
        aggregate: request.aggregate,
        group_by: request.group_by,
        filters: request.filters,
        from_time: request.from_time,
        to_time: request.to_time,
        limit: request.limit,
        table: request.table,
      },
      this.database,
      this.catalogFor(identity),
      budget.maxResultRows,
    );
    // `EXPLAIN ESTIMATE` reports the parts/rows/marks the planner will read
    // AFTER primary-key pruning — which is why tenant_id leads every ORDER BY
    // in schema.ts: the estimate for one tenant does not include everyone.
    const estimateSql = compiled.sql.replace(/ FORMAT JSON$/, "");
    let rowsToRead = 0;
    let partsRead = 0;
    try {
      const res = (await this.raw(
        `EXPLAIN ESTIMATE ${estimateSql} FORMAT JSON`,
        compiled.params,
        planningSettings(budget),
      )) as { data?: { rows?: number | string; parts?: number | string }[] } | undefined;
      for (const row of res?.data ?? []) {
        rowsToRead += Number(row.rows ?? 0);
        partsRead += Number(row.parts ?? 0);
      }
    } catch (err) {
      // An estimate that cannot be produced must not read as "cheap".
      throw err instanceof SiaxError
        ? err
        : new SiaxError("DEPENDENCY_UNAVAILABLE", "clickhouse", "could not estimate query");
    }
    // Rough byte estimate: ClickHouse's ESTIMATE reports rows, not bytes.
    // 128 B/row is a deliberately PESSIMISTIC constant for these tables
    // (measured mean in the BI-P010 benchmark is lower), so the gate errs
    // toward refusing rather than admitting an oversized query.
    const bytesToRead = rowsToRead * 128;
    const within = rowsToRead <= budget.maxRowsToRead && bytesToRead <= budget.maxBytesToRead;
    return {
      estimated_rows_to_read: rowsToRead,
      estimated_bytes_to_read: bytesToRead,
      within_budget: within,
      budget,
      ...(within
        ? {}
        : {
            reason:
              `estimate ${rowsToRead} rows / ${bytesToRead} bytes over budget ` +
              `(${budget.maxRowsToRead} rows / ${budget.maxBytesToRead} bytes), parts=${partsRead}`,
          }),
    };
  }

  // ── BI-P009 capability 5: cancelQuery ────────────────────────────────────

  /**
   * Two-sided cancellation. The client-side abort stops us waiting; the
   * server-side KILL QUERY stops ClickHouse burning CPU. Cancelling is
   * tenant-scoped: the query id embeds the tenant and is checked, so tenant A
   * cannot cancel tenant B's work.
   */
  async cancelQuery(identity: AnalyticsIdentity, queryId: string): Promise<boolean> {
    this.assertIdentity(identity);
    if (!queryId.startsWith(this.queryIdPrefix(identity))) {
      throw new SiaxError("AUTHORIZATION_DENIED", "0pt.analytics", "query does not belong to this tenant");
    }
    this.inFlight.get(queryId)?.abort();
    this.inFlight.delete(queryId);
    try {
      // KILL QUERY is a mutation of server state, so it cannot ride the
      // readonly=1 settings the SELECT path uses.
      await this.raw(
        `KILL QUERY WHERE query_id = {killId:String} SYNC FORMAT JSON`,
        { param_killId: queryId },
        { max_execution_time: "10" },
      );
      return true;
    } catch {
      return false;
    }
  }

  // ── BI-P009 capability 6: explain ────────────────────────────────────────

  async explain(identity: AnalyticsIdentity, request: MetricRequest): Promise<QueryExplanation> {
    this.assertIdentity(identity);
    const budget = this.budgetFor(identity);
    const compiled = compileMetricQuery(
      {
        tenant_id: identity.tenant_id,
        project_id: identity.project_id,
        dataset_id: request.dataset_id,
        metric: request.metric,
        aggregate: request.aggregate,
        group_by: request.group_by,
        filters: request.filters,
        from_time: request.from_time,
        to_time: request.to_time,
        limit: request.limit,
        table: request.table,
      },
      this.database,
      this.catalogFor(identity),
      budget.maxResultRows,
    );
    const sql = compiled.sql.replace(/ FORMAT JSON$/, "");
    const res = (await this.raw(
      `EXPLAIN PLAN indexes = 1 ${sql} FORMAT TSVRaw`,
      compiled.params,
      planningSettings(budget),
    )) as string | undefined;
    return {
      plan: typeof res === "string" ? res : JSON.stringify(res ?? ""),
      forced_tenant_predicate: FORCED_TENANT_PREDICATE,
      table: compiled.table,
    };
  }

  // ── BI-P009 capability 7: ingestProjection ───────────────────────────────

  async ingestProjection(
    identity: AnalyticsIdentity,
    request: ProjectionIngestRequest,
  ): Promise<{ ingested: number }> {
    this.assertIdentity(identity);
    if (request.rows.length === 0) return { ingested: 0 };
    const now = new Date();
    const version = Math.floor(now.getTime());
    const body = request.rows
      .map((r) => {
        if (!r.entity_key || !r.entity_key.trim()) {
          throw new SiaxError("INVALID_REQUEST", "0pt.analytics", "every projection row needs an entity_key");
        }
        return JSON.stringify({
          // tenant/project come from the VERIFIED identity, never the row —
          // a row-supplied tenant_id is the INF0 Wave-2 bug class.
          tenant_id: identity.tenant_id,
          project_id: identity.project_id ?? "",
          dataset_id: request.dataset_id,
          entity_key: r.entity_key,
          as_of: (r.as_of ?? now.toISOString()).replace("T", " ").replace(/Z$/, ""),
          version: r.version ?? version,
          attributes: r.attributes ?? {},
          measures: r.measures ?? {},
        });
      })
      .join("\n");
    await this.raw(
      `INSERT INTO ${this.database}.opt_dataset_projection FORMAT JSONEachRow`,
      {},
      {},
      undefined,
      body,
    );
    return { ingested: request.rows.length };
  }

  // ── BI-P009 capability 8: refreshProjection ──────────────────────────────

  /**
   * ReplacingMergeTree deduplicates on a background merge, so a read straight
   * after an ingest can still see superseded rows. `OPTIMIZE ... FINAL` forces
   * the merge for THIS tenant's partition only — the table is
   * `PARTITION BY (tenant_id)` precisely so one tenant's refresh does not
   * rewrite another tenant's parts (BI-P012's "without degrading unrelated
   * tenant" applies to maintenance too, not only to queries).
   */
  async refreshProjection(identity: AnalyticsIdentity, dataset_id: string): Promise<ProjectionRefreshResult> {
    this.assertIdentity(identity);
    await this.raw(
      `OPTIMIZE TABLE ${this.database}.opt_dataset_projection PARTITION {tenant:String} FINAL`,
      { param_tenant: identity.tenant_id },
      { max_execution_time: "120" },
    );
    const compiled = compileProjectionQuery(
      { tenant_id: identity.tenant_id, project_id: identity.project_id, dataset_id, limit: 1 },
      this.database,
      1,
    );
    const countSql =
      `SELECT count() AS value FROM ${this.database}.opt_dataset_projection FINAL ` +
      `WHERE ${FORCED_TENANT_PREDICATE} AND dataset_id = {dataset:String} FORMAT JSON`;
    const res = (await this.raw(countSql, compiled.params, clickhouseSettings(this.budgetFor(identity)))) as
      | { data?: { value: number | string }[] }
      | undefined;
    return {
      dataset_id,
      row_count: Number(res?.data?.[0]?.value ?? 0),
      refreshed_at: new Date().toISOString(),
    };
  }

  // ── transport ────────────────────────────────────────────────────────────

  private queryIdPrefix(identity: AnalyticsIdentity): string {
    return `opt_${crypto.createHash("sha256").update(identity.tenant_id).digest("hex").slice(0, 12)}_`;
  }

  private newQueryId(identity: AnalyticsIdentity): string {
    return `${this.queryIdPrefix(identity)}${crypto.randomUUID().replace(/-/g, "")}`;
  }

  /** DDL path: no readonly, no result parsing beyond errors. */
  private async ddl(sql: string): Promise<void> {
    await this.raw(sql, {}, { max_execution_time: "60" });
  }

  private async raw(
    sql: string,
    params: Record<string, string>,
    settings: Record<string, string>,
    queryId?: string,
    body?: string,
  ): Promise<unknown> {
    const url = new URL(this.baseUrl);
    url.searchParams.set("query", sql);
    for (const [k, v] of Object.entries(settings)) url.searchParams.set(k, v);
    for (const [k, v] of Object.entries(params)) url.searchParams.set(k, v);
    if (queryId) url.searchParams.set("query_id", queryId);

    const headers: Record<string, string> = {};
    if (this.authHeader) headers["Authorization"] = this.authHeader;

    const controller = new AbortController();
    if (queryId) this.inFlight.set(queryId, controller);
    const timeout = setTimeout(() => controller.abort(), this.timeoutMs);

    let res: Response;
    try {
      res = await fetch(url.toString(), {
        method: "POST",
        headers,
        body: body ?? "",
        signal: controller.signal,
      });
    } catch (cause) {
      throw new SiaxError("DEPENDENCY_UNAVAILABLE", "clickhouse", "clickhouse unreachable", { cause });
    } finally {
      clearTimeout(timeout);
    }
    const text = await res.text();
    if (!res.ok) {
      // ClickHouse error text can echo the query; keep only the numeric code
      // so a caller never receives another tenant's identifiers back.
      const code = /Code: (\d+)/.exec(text)?.[1] ?? String(res.status);
      throw new SiaxError(
        siaxErrorCodeForUpstreamStatus(res.status),
        "clickhouse",
        `query failed (CH code ${code})`,
      );
    }
    if (!text) return undefined;
    let parsed: unknown;
    try {
      parsed = JSON.parse(text);
    } catch {
      return text;
    }
    // DEFENCE IN DEPTH against a partial result that looks complete.
    // ClickHouse normally answers a breached limit with an HTTP error status
    // (measured on 24.8: 408 for max_execution_time, 500 for
    // max_rows_to_read), which `res.ok` above already catches. But when rows
    // have already been flushed, the JSON body can carry an `exception` key
    // under a 200 — a truncated answer indistinguishable from a whole one.
    // Treat any body carrying an exception as a failure, whatever the status.
    if (parsed && typeof parsed === "object" && "exception" in parsed) {
      const ex = String((parsed as { exception?: unknown }).exception ?? "");
      const code = /Code: (\d+)/.exec(ex)?.[1] ?? "unknown";
      throw new SiaxError("PERMANENT_FAILURE", "clickhouse", `query failed mid-stream (CH code ${code})`);
    }
    return parsed;
  }
}

interface ClickHouseJsonResponse {
  data?: Record<string, unknown>[];
  statistics?: { rows_read?: number; bytes_read?: number; elapsed?: number };
}