/**
 * SqlMonitoringStorage
 *
 * Implements MonitoringStorage using SQL via QueryEngine.
 * Supports two SQL dialects:
 * - "clickhouse": ClickHouse SQL for production (remote ClickHouse instance)
 * - "duckdb": DuckDB SQL for local dev (embedded DuckDB querying NDJSON files)
 *
 * Writes flow through the OTel pipeline (NDJSONLogExporter).
 * This adapter is read-only — it queries NDJSON files on disk or a ClickHouse table.
 */

import type {
  MonitoringDateRange,
  QueryEngine,
} from "../monitoring/query-engine";
import type { MonitoringLog } from "./types";
import type {
  AggregationParams,
  AggregationResult,
  GroupByColumn,
  MonitoringStorage,
  PropertyFilters,
} from "./ports";

// ---------------------------------------------------------------------------
// Types
// ---------------------------------------------------------------------------

export type SqlDialect = "clickhouse" | "duckdb";

// ---------------------------------------------------------------------------
// Validation helpers
// ---------------------------------------------------------------------------

/**
 * Normalise a ClickHouse / DuckDB bucket timestamp to ISO 8601 UTC.
 *
 * ClickHouse JSONEachRow returns DateTime as "YYYY-MM-DD HH:MM:SS" (no TZ).
 * Browsers parse that as *local* time, which shifts charts by the user's
 * UTC offset and breaks the bucket-key lookup on the frontend.
 */
function toISOBucket(bucket: unknown): string {
  const raw = String(bucket ?? "");
  if (raw && !raw.includes("T")) {
    return raw.replace(" ", "T") + "Z";
  }
  return raw;
}

/** Escape a string value for safe use in SQL single-quoted literals. */
function esc(value: string): string {
  // oxlint-disable-next-line no-control-regex
  return value.replace(/\0/g, "").replace(/\\/g, "\\\\").replace(/'/g, "''");
}

/**
 * Coerce a query value to a finite number, defaulting to 0.
 * ClickHouse quantile()/avg() over an empty set return NaN, which would
 * otherwise leak to the UI as NaN.
 */
function finiteOrZero(value: unknown): number {
  const n = Number(value ?? 0);
  return Number.isFinite(n) ? n : 0;
}

/** Allowed groupByColumn values. */
const ALLOWED_GROUP_BY_COLUMNS = new Set<GroupByColumn>([
  "connection_id",
  "user_id",
  "tool_name",
  "virtual_mcp_id",
]);

/** Validate a groupByColumn identifier. */
function validateGroupByColumn(col: string): GroupByColumn {
  if (!ALLOWED_GROUP_BY_COLUMNS.has(col as GroupByColumn)) {
    throw new Error(`Invalid groupByColumn: ${col}`);
  }
  return col as GroupByColumn;
}

/** Strict JSONPath regex: allows $.key.subkey or key.subkey forms. */
const JSONPATH_REGEX = /^(\$\.)?[a-zA-Z0-9_]+(\.[a-zA-Z0-9_]+)*$/;

/** Validate and normalise a JSONPath expression. Returns the $.key.subkey form. */
function validateJsonPath(path: string): string {
  if (!JSONPATH_REGEX.test(path)) {
    throw new Error(`Invalid JSONPath: ${path}`);
  }
  // Normalise to $.key form
  if (path.startsWith("$.")) return path;
  if (path.startsWith(".")) return `$${path}`;
  return `$.${path}`;
}

/**
 * Convert a validated JSONPath ($.key or $.key.subkey) to ClickHouse
 * JSONExtract key arguments: 'key' or 'key', 'subkey'
 */
function jsonPathToChKeys(jsonPath: string): string {
  // Remove leading $. then split by .
  const keys = jsonPath.replace(/^\$\./, "").split(".");
  return keys.map((k) => `'${esc(k)}'`).join(", ");
}

/** Interval regex and max amounts. */
const INTERVAL_REGEX = /^(\d+)([mhd])$/;
const MAX_INTERVAL_AMOUNTS: Record<string, number> = {
  m: 525960,
  h: 8760,
  d: 365,
};

function parseInterval(interval: string): { amount: number; unit: string } {
  const match = INTERVAL_REGEX.exec(interval);
  if (!match || !match[1] || !match[2]) {
    throw new Error(`Invalid interval: ${interval}`);
  }
  const amount = parseInt(match[1], 10);
  const unit = match[2];
  const max = MAX_INTERVAL_AMOUNTS[unit]!;
  if (amount < 1 || amount > max) {
    throw new Error(
      `Invalid interval amount: ${amount}${unit} (max ${max}${unit})`,
    );
  }
  return { amount, unit };
}

function intervalToSQL(
  interval: string,
  dialect: SqlDialect,
  column: string = "timestamp",
): string {
  const { amount, unit } = parseInterval(interval);
  const unitMap: Record<string, string> = {
    m: "MINUTE",
    h: "HOUR",
    d: "DAY",
  };
  const sqlUnit = unitMap[unit];
  if (dialect === "duckdb") {
    return `time_bucket(INTERVAL '${amount} ${sqlUnit}', CAST(${column} AS TIMESTAMP))`;
  }
  return `toStartOfInterval(${column}, INTERVAL ${amount} ${sqlUnit})`;
}

// ---------------------------------------------------------------------------
// Dialect-aware JSON extraction helpers
// ---------------------------------------------------------------------------

function jsonExtractString(
  col: string,
  jsonPath: string,
  dialect: SqlDialect,
): string {
  if (dialect === "duckdb") {
    return `json_extract_string(${col}, '${jsonPath}')`;
  }
  const chKeys = jsonPathToChKeys(jsonPath);
  return `JSONExtractString(${col}, ${chKeys})`;
}

function jsonExtractFloat(
  col: string,
  jsonPath: string,
  dialect: SqlDialect,
): string {
  if (dialect === "duckdb") {
    return `CAST(json_extract_string(${col}, '${jsonPath}') AS DOUBLE)`;
  }
  const chKeys = jsonPathToChKeys(jsonPath);
  return `JSONExtractFloat(${col}, ${chKeys})`;
}

/**
 * Extract a number stored as a JSON *string* (e.g. properties values, which are
 * always strings). Returns NULL/0 when absent or unparseable — never throws.
 */
function jsonExtractNumericString(
  col: string,
  jsonPath: string,
  dialect: SqlDialect,
): string {
  if (dialect === "duckdb") {
    return `TRY_CAST(json_extract_string(${col}, '${jsonPath}') AS DOUBLE)`;
  }
  const chKeys = jsonPathToChKeys(jsonPath);
  return `toFloat64OrZero(JSONExtractString(${col}, ${chKeys}))`;
}

// ---------------------------------------------------------------------------
// Row mapping
// ---------------------------------------------------------------------------

function safeJsonParse(val: unknown): Record<string, unknown> {
  if (val === null || val === undefined) return {};
  if (typeof val === "object") return val as Record<string, unknown>;
  const str = String(val);
  try {
    return JSON.parse(str);
  } catch {
    // Truncated JSON strings (from truncateString) are not valid JSON.
    // Wrap the raw string so the UI can still display it.
    return { _decocms_truncated: str };
  }
}

function toMonitoringLog(row: Record<string, unknown>): MonitoringLog {
  return {
    id: String(row.id ?? ""),
    organizationId: String(row.organization_id ?? ""),
    connectionId: String(row.connection_id ?? ""),
    toolName: String(row.tool_name ?? ""),
    input: safeJsonParse(row.input),
    output: safeJsonParse(row.output),
    isError:
      row.is_error === 1 ||
      row.is_error === true ||
      row.is_error === "1" ||
      (typeof row.is_error === "bigint" && row.is_error === 1n),
    errorMessage: row.error_message != null ? String(row.error_message) : null,
    durationMs: Number(row.duration_ms ?? 0),
    timestamp: (() => {
      if (row.timestamp instanceof Date) return row.timestamp.toISOString();
      const str = String(row.timestamp ?? "");
      // chdb returns "YYYY-MM-DD HH:MM:SS.nnnnnnnnn" — convert to ISO 8601
      const d = new Date(str.includes("T") ? str : str.replace(" ", "T") + "Z");
      return Number.isNaN(d.getTime()) ? str : d.toISOString();
    })(),
    userId: row.user_id != null ? String(row.user_id) : null,
    requestId: String(row.request_id ?? ""),
    userAgent: row.user_agent != null ? String(row.user_agent) : null,
    virtualMcpId:
      row.virtual_mcp_id != null ? String(row.virtual_mcp_id) : null,
    properties: row.properties
      ? (safeJsonParse(row.properties) as Record<string, string>)
      : null,
  };
}

// ---------------------------------------------------------------------------
// Property filter SQL builder (dialect-aware)
// ---------------------------------------------------------------------------

function buildPropertyFilterClauses(
  filters: PropertyFilters,
  dialect: SqlDialect,
): string[] {
  const clauses: string[] = [];

  if (filters.properties) {
    for (const [key, value] of Object.entries(filters.properties)) {
      const k = esc(key);
      const v = esc(value);
      if (dialect === "duckdb") {
        clauses.push(`json_extract_string(properties, '$.${k}') = '${v}'`);
      } else {
        clauses.push(`JSONExtractString(properties, '${k}') = '${v}'`);
      }
    }
  }

  if (filters.propertyKeys) {
    for (const key of filters.propertyKeys) {
      const k = esc(key);
      if (dialect === "duckdb") {
        clauses.push(
          `json_extract_string(properties, '$.${k}') IS NOT NULL AND json_extract_string(properties, '$.${k}') != ''`,
        );
      } else {
        clauses.push(
          `JSONExtractString(properties, '${k}') IS NOT NULL AND JSONExtractString(properties, '${k}') != ''`,
        );
      }
    }
  }

  if (filters.propertyPatterns) {
    for (const [key, pattern] of Object.entries(filters.propertyPatterns)) {
      const k = esc(key);
      const p = esc(pattern);
      if (dialect === "duckdb") {
        clauses.push(`json_extract_string(properties, '$.${k}') ILIKE '${p}'`);
      } else {
        clauses.push(`JSONExtractString(properties, '${k}') ILIKE '${p}'`);
      }
    }
  }

  if (filters.propertyInValues) {
    for (const [key, value] of Object.entries(filters.propertyInValues)) {
      const k = esc(key);
      const v = esc(value);
      if (dialect === "duckdb") {
        clauses.push(
          `list_contains(string_split(json_extract_string(properties, '$.${k}'), ','), '${v}')`,
        );
      } else {
        clauses.push(
          `has(splitByChar(',', JSONExtractString(properties, '${k}')), '${v}')`,
        );
      }
    }
  }

  return clauses;
}

// ---------------------------------------------------------------------------
// ClickHouse column list — truncate large JSON blobs to cap memory usage
// ---------------------------------------------------------------------------

const LIST_COLUMNS = [
  "id",
  "organization_id",
  "connection_id",
  "tool_name",
  "is_error",
  "error_message",
  "duration_ms",
  "timestamp",
  "user_id",
  "request_id",
  "user_agent",
  "virtual_mcp_id",
  "properties",
].join(", ");

// ---------------------------------------------------------------------------
// Timestamp filter helper (dialect-aware)
// ---------------------------------------------------------------------------

function buildCommonFilterClauses(
  dialect: SqlDialect,
  filters?: {
    connectionIds?: string[];
    toolNames?: string[];
    virtualMcpIds?: string[];
    startDate?: Date;
    endDate?: Date;
    propertyFilters?: PropertyFilters;
  },
): string[] {
  const clauses: string[] = [];
  applyStartDateBound(clauses, filters?.startDate, dialect);
  if (!filters) return clauses;
  if (filters.connectionIds?.length) {
    const ids = filters.connectionIds.map((id) => `'${esc(id)}'`).join(",");
    clauses.push(`connection_id IN (${ids})`);
  }
  if (filters.toolNames?.length) {
    const names = filters.toolNames.map((n) => `'${esc(n)}'`).join(",");
    clauses.push(`tool_name IN (${names})`);
  }
  if (filters.virtualMcpIds?.length) {
    const ids = filters.virtualMcpIds.map((id) => `'${esc(id)}'`).join(",");
    clauses.push(`virtual_mcp_id IN (${ids})`);
  }
  if (filters.endDate) {
    clauses.push(tsLte(filters.endDate, dialect));
  }
  if (filters.propertyFilters) {
    clauses.push(
      ...buildPropertyFilterClauses(filters.propertyFilters, dialect),
    );
  }
  return clauses;
}

function tsGte(
  date: Date,
  dialect: SqlDialect,
  column: string = "timestamp",
): string {
  if (dialect === "duckdb") {
    return `CAST(${column} AS TIMESTAMP) >= TIMESTAMP '${date.toISOString()}'`;
  }
  if (column === "bucket") {
    return `${column} >= parseDateTimeBestEffort('${date.toISOString()}')`;
  }
  return `${column} >= parseDateTime64BestEffort('${date.toISOString()}', 9)`;
}

function tsLte(
  date: Date,
  dialect: SqlDialect,
  column: string = "timestamp",
): string {
  if (dialect === "duckdb") {
    return `CAST(${column} AS TIMESTAMP) <= TIMESTAMP '${date.toISOString()}'`;
  }
  if (column === "bucket") {
    return `${column} <= parseDateTimeBestEffort('${date.toISOString()}')`;
  }
  return `${column} <= parseDateTime64BestEffort('${date.toISOString()}', 9)`;
}

/**
 * Default lookback window (30 days) applied when no startDate is provided and
 * the dialect is ClickHouse. Prevents unbounded full-table scans that can OOM
 * the server. DuckDB scans local NDJSON files which are naturally bounded.
 */
const DEFAULT_LOOKBACK_MS = 30 * 24 * 60 * 60 * 1000;

/**
 * Tighter lookback for thread-usage queries. These always target a specific,
 * recent set of decopilot thread IDs, so the 30-day default just scans (and
 * materializes the wide `LogAttributes`/`properties` column for) weeks of
 * irrelevant rows — the dominant cost behind the queryThreadUsage OOMs.
 */
const THREAD_USAGE_LOOKBACK_MS = 7 * 24 * 60 * 60 * 1000;

function applyStartDateBound(
  where: string[],
  startDate: Date | undefined,
  dialect: SqlDialect,
  column: string = "timestamp",
): void {
  if (startDate) {
    where.push(tsGte(startDate, dialect, column));
  } else if (dialect === "clickhouse") {
    where.push(
      tsGte(new Date(Date.now() - DEFAULT_LOOKBACK_MS), dialect, column),
    );
  }
}

// ---------------------------------------------------------------------------
// SqlMonitoringStorage
// ---------------------------------------------------------------------------

export class SqlMonitoringStorage implements MonitoringStorage {
  constructor(
    private engine: QueryEngine,
    // The range lets the GCS OTLP source prune its bucket glob to the relevant
    // Hive day partitions; other source factories ignore it.
    private sourceFactory: (
      organizationId: string,
      range?: MonitoringDateRange,
    ) => string,
    private metricEngine: QueryEngine,
    private metricSourceFactory: (
      organizationId: string,
      range?: MonitoringDateRange,
    ) => string,
    private dialect: SqlDialect = "clickhouse",
    // When true (DuckDB reading OTLP flat-log rows from GCS), metrics are
    // derived from the log rows (count/avg/quantile over duration_ms + is_error)
    // instead of the pre-aggregated histogram MetricRow columns the local-NDJSON
    // path uses. Default false preserves the local histogram path.
    private metricsFromLogs: boolean = false,
  ) {}

  async query(filters: {
    organizationId: string;
    connectionId?: string;
    excludeConnectionIds?: string[];
    virtualMcpId?: string;
    toolName?: string;
    isError?: boolean;
    startDate?: Date;
    endDate?: Date;
    limit?: number;
    offset?: number;
    propertyFilters?: PropertyFilters;
  }): Promise<{ logs: MonitoringLog[]; total: number }> {
    if (!filters.organizationId) {
      throw new Error("organizationId is required");
    }

    const source = this.sourceFactory(filters.organizationId, {
      startDate: filters.startDate,
      endDate: filters.endDate,
    });

    const where: string[] = [
      `organization_id = '${esc(filters.organizationId)}'`,
    ];

    if (filters.connectionId) {
      where.push(`connection_id = '${esc(filters.connectionId)}'`);
    }
    if (filters.excludeConnectionIds?.length) {
      const ids = filters.excludeConnectionIds
        .map((id) => `'${esc(id)}'`)
        .join(",");
      where.push(`connection_id NOT IN (${ids})`);
    }
    if (filters.virtualMcpId) {
      where.push(`virtual_mcp_id = '${esc(filters.virtualMcpId)}'`);
    }
    if (filters.toolName) {
      where.push(`tool_name = '${esc(filters.toolName)}'`);
    }
    if (filters.isError !== undefined) {
      where.push(`is_error = ${filters.isError ? 1 : 0}`);
    }
    applyStartDateBound(where, filters.startDate, this.dialect);
    if (filters.endDate) {
      where.push(tsLte(filters.endDate, this.dialect));
    }
    if (filters.propertyFilters) {
      where.push(
        ...buildPropertyFilterClauses(filters.propertyFilters, this.dialect),
      );
    }

    const MAX_LIMIT = 1000;
    const limit = Math.min(
      Math.max(1, Math.floor(filters.limit ?? 50)),
      MAX_LIMIT,
    );
    const offset = Math.max(0, Math.floor(filters.offset ?? 0));

    const whereClause = where.join(" AND ");

    if (this.dialect === "duckdb") {
      const sql = `SELECT ${LIST_COLUMNS}, count(*) OVER () AS _total FROM ${source} WHERE ${whereClause} ORDER BY timestamp DESC LIMIT ${limit} OFFSET ${offset}`;
      const rows = await this.engine.query(sql);
      if (rows.length === 0) {
        return { logs: [], total: 0 };
      }
      const total = Number(rows[0]!._total ?? 0);
      return { logs: rows.map(toMonitoringLog), total };
    }

    const countSql = `SELECT count(*) AS total FROM ${source} WHERE ${whereClause}`;
    const dataSql = `SELECT ${LIST_COLUMNS} FROM ${source} WHERE ${whereClause} ORDER BY timestamp DESC LIMIT ${limit} OFFSET ${offset}`;

    const [countRows, dataRows] = await Promise.all([
      this.engine.query(countSql),
      this.engine.query(dataSql),
    ]);

    const total = Number(countRows[0]?.total ?? 0);
    return { logs: dataRows.map(toMonitoringLog), total };
  }

  async getById(
    organizationId: string,
    id: string,
  ): Promise<MonitoringLog | null> {
    if (!organizationId || !id) return null;

    // No date range available (lookup is by random span id), so the GCS OTLP
    // path can't partition-prune here and reads the whole prefix. Acceptable:
    // single-row detail open, far rarer than the date-ranged list/timeseries.
    const source = this.sourceFactory(organizationId);
    const sql = `SELECT * FROM ${source} WHERE organization_id = '${esc(organizationId)}' AND id = '${esc(id)}' LIMIT 1`;
    const rows = await this.engine.query(sql);
    if (rows.length === 0) return null;
    return toMonitoringLog(rows[0]!);
  }

  async getStats(filters: {
    organizationId: string;
    startDate?: Date;
    endDate?: Date;
  }): Promise<{
    totalCalls: number;
    errorRate: number;
    avgDurationMs: number;
  }> {
    if (!filters.organizationId) {
      throw new Error("organizationId is required");
    }

    const source = this.sourceFactory(filters.organizationId, {
      startDate: filters.startDate,
      endDate: filters.endDate,
    });

    const where: string[] = [
      `organization_id = '${esc(filters.organizationId)}'`,
    ];

    applyStartDateBound(where, filters.startDate, this.dialect);
    if (filters.endDate) {
      where.push(tsLte(filters.endDate, this.dialect));
    }

    const sql = `SELECT count(*) AS total_calls, coalesce(sum(CASE WHEN is_error = 1 THEN 1.0 ELSE 0.0 END) / NULLIF(count(*), 0), 0) AS error_rate, coalesce(avg(duration_ms), 0) AS avg_duration_ms FROM ${source} WHERE ${where.join(" AND ")}`;

    const rows = await this.engine.query(sql);
    const row = rows[0] ?? {};

    return {
      totalCalls: Number(row.total_calls ?? 0),
      errorRate: Number(row.error_rate ?? 0),
      avgDurationMs: Number(row.avg_duration_ms ?? 0),
    };
  }

  /**
   * Aggregate LLM-call usage (tokens + USD cost) from the raw log rows.
   *
   * Each LLM completion is a single log row whose `output` JSON holds the
   * token usage and (for ai-gateway / OpenRouter) cost. Reading from the log
   * source — rather than the pre-aggregated metric histograms — lets us SUM
   * those JSON values and filter by `user_id`. Works for both dialects since
   * the log source carries `output` + `user_id` in either case.
   */
  async queryLlmUsageStats(params: {
    organizationId: string;
    interval: string;
    connectionId: string;
    startDate?: Date;
    endDate?: Date;
    userIds?: string[];
    topN?: number;
  }): Promise<{
    totalCalls: number;
    totalErrors: number;
    avgDurationMs: number;
    p50DurationMs: number;
    p95DurationMs: number;
    totalInputTokens: number;
    totalOutputTokens: number;
    totalTokens: number;
    totalCostUsd: number;
    topTools: Array<{
      toolName: string;
      connectionId: string | null;
      calls: number;
      inputTokens: number;
      outputTokens: number;
      costUsd: number;
    }>;
    timeseries: Array<{
      timestamp: string;
      calls: number;
      errors: number;
      errorRate: number;
      avg: number;
      p50: number;
      p95: number;
      inputTokens: number;
      outputTokens: number;
      totalTokens: number;
      costUsd: number;
    }>;
  }> {
    const emptyResult = {
      totalCalls: 0,
      totalErrors: 0,
      avgDurationMs: 0,
      p50DurationMs: 0,
      p95DurationMs: 0,
      totalInputTokens: 0,
      totalOutputTokens: 0,
      totalTokens: 0,
      totalCostUsd: 0,
      topTools: [] as Array<{
        toolName: string;
        connectionId: string | null;
        calls: number;
        inputTokens: number;
        outputTokens: number;
        costUsd: number;
      }>,
      timeseries: [] as Array<{
        timestamp: string;
        calls: number;
        errors: number;
        errorRate: number;
        avg: number;
        p50: number;
        p95: number;
        inputTokens: number;
        outputTokens: number;
        totalTokens: number;
        costUsd: number;
      }>,
    };

    if (!params.organizationId || !params.connectionId) return emptyResult;

    const dialect = this.dialect;
    const isCh = dialect === "clickhouse";
    const source = this.sourceFactory(params.organizationId, {
      startDate: params.startDate,
      endDate: params.endDate,
    });
    const bucketExpr = intervalToSQL(params.interval, dialect, "timestamp");

    // Tokens/cost are read from `properties` (small, never truncated) rather
    // than `output` (capped at 64KB → malformed JSON on large responses).
    const num = (key: string) =>
      `coalesce(${jsonExtractNumericString("properties", `$.${key}`, dialect)}, 0)`;
    const callsExpr = isCh ? "count()" : "count(*)";
    const errExpr = isCh
      ? "countIf(is_error = 1)"
      : "count(*) FILTER (WHERE is_error = 1)";
    const pExpr = (q: number) =>
      isCh ? `quantile(${q})(duration_ms)` : `quantile_cont(duration_ms, ${q})`;

    const where: string[] = [
      `organization_id = '${esc(params.organizationId)}'`,
      `connection_id = '${esc(params.connectionId)}'`,
    ];
    applyStartDateBound(where, params.startDate, dialect);
    if (params.endDate) {
      where.push(tsLte(params.endDate, dialect));
    }
    if (params.userIds?.length) {
      const ids = params.userIds
        .slice(0, 100)
        .map((id) => `'${esc(id)}'`)
        .join(",");
      where.push(`user_id IN (${ids})`);
    }
    const whereClause = where.join(" AND ");
    const topN = Math.min(Math.max(1, Math.floor(params.topN ?? 5)), 20);

    const aggCols = `${callsExpr} AS calls,
  ${errExpr} AS errors,
  avg(duration_ms) AS avg,
  sum(${num("input_tokens")}) AS in_tok,
  sum(${num("output_tokens")}) AS out_tok,
  sum(${num("total_tokens")}) AS tot_tok,
  sum(${num("cost")}) AS cost`;

    const totalsSql = `SELECT ${aggCols},
  ${pExpr(0.5)} AS p50,
  ${pExpr(0.95)} AS p95
FROM ${source}
WHERE ${whereClause}`;

    const timeseriesSql = `SELECT ${bucketExpr} AS bucket,
  ${aggCols},
  ${pExpr(0.5)} AS p50,
  ${pExpr(0.95)} AS p95
FROM ${source}
WHERE ${whereClause}
GROUP BY bucket
ORDER BY bucket ASC`;

    const topToolsSql = `SELECT tool_name,
  ${callsExpr} AS calls,
  sum(${num("input_tokens")}) AS in_tok,
  sum(${num("output_tokens")}) AS out_tok,
  sum(${num("cost")}) AS cost
FROM ${source}
WHERE ${whereClause} AND tool_name != ''
GROUP BY tool_name
ORDER BY calls DESC
LIMIT ${topN}`;

    try {
      const [totalsRows, tsRows, topToolRows] = await Promise.all([
        this.engine.query(totalsSql),
        this.engine.query(timeseriesSql),
        this.engine.query(topToolsSql),
      ]);

      const totals = totalsRows[0] ?? {};

      return {
        totalCalls: Number(totals.calls ?? 0),
        totalErrors: Number(totals.errors ?? 0),
        avgDurationMs: finiteOrZero(totals.avg),
        p50DurationMs: finiteOrZero(totals.p50),
        p95DurationMs: finiteOrZero(totals.p95),
        totalInputTokens: finiteOrZero(totals.in_tok),
        totalOutputTokens: finiteOrZero(totals.out_tok),
        totalTokens: finiteOrZero(totals.tot_tok),
        totalCostUsd: finiteOrZero(totals.cost),
        topTools: topToolRows.map((row) => ({
          toolName: String(row.tool_name ?? ""),
          connectionId: params.connectionId,
          calls: Number(row.calls ?? 0),
          inputTokens: finiteOrZero(row.in_tok),
          outputTokens: finiteOrZero(row.out_tok),
          costUsd: finiteOrZero(row.cost),
        })),
        timeseries: tsRows.map((row) => {
          const calls = Number(row.calls ?? 0);
          const errors = Number(row.errors ?? 0);
          return {
            timestamp: toISOBucket(row.bucket),
            calls,
            errors,
            errorRate: calls > 0 ? errors / calls : 0,
            avg: finiteOrZero(row.avg),
            p50: finiteOrZero(row.p50),
            p95: finiteOrZero(row.p95),
            inputTokens: finiteOrZero(row.in_tok),
            outputTokens: finiteOrZero(row.out_tok),
            totalTokens: finiteOrZero(row.tot_tok),
            costUsd: finiteOrZero(row.cost),
          };
        }),
      };
    } catch (err) {
      console.error("queryLlmUsageStats failed:", err);
      throw new Error("Monitoring stats unavailable", { cause: err });
    }
  }

  async queryThreadUsage(params: {
    organizationId: string;
    connectionId: string;
    threadIds: string[];
    startDate?: Date;
    endDate?: Date;
  }): Promise<
    Array<{
      threadId: string;
      calls: number;
      inputTokens: number;
      outputTokens: number;
      totalTokens: number;
      costUsd: number;
    }>
  > {
    if (
      !params.organizationId ||
      !params.connectionId ||
      params.threadIds.length === 0
    ) {
      return [];
    }

    const dialect = this.dialect;
    const isCh = dialect === "clickhouse";
    const source = this.sourceFactory(params.organizationId, {
      startDate: params.startDate,
      endDate: params.endDate,
    });
    const threadExpr = jsonExtractString("properties", "$.thread_id", dialect);
    // Tokens/cost from `properties` (never truncated), not `output` (64KB cap).
    const num = (key: string) =>
      `coalesce(${jsonExtractNumericString("properties", `$.${key}`, dialect)}, 0)`;
    const callsExpr = isCh ? "count()" : "count(*)";

    const ids = params.threadIds
      .slice(0, 500)
      .map((id) => `'${esc(id)}'`)
      .join(",");

    const where: string[] = [
      `organization_id = '${esc(params.organizationId)}'`,
      `connection_id = '${esc(params.connectionId)}'`,
      `${threadExpr} IN (${ids})`,
    ];
    // ClickHouse-only tighter default: applyStartDateBound otherwise falls back
    // to the 30-day DEFAULT_LOOKBACK_MS, which scans far more of otel_logs than
    // these recent-thread lookups ever need. DuckDB stays unbounded (local files).
    const startDate =
      params.startDate ??
      (isCh ? new Date(Date.now() - THREAD_USAGE_LOOKBACK_MS) : undefined);
    applyStartDateBound(where, startDate, dialect);
    if (params.endDate) {
      where.push(tsLte(params.endDate, dialect));
    }

    const sql = `SELECT ${threadExpr} AS thread_id,
  ${callsExpr} AS calls,
  sum(${num("input_tokens")}) AS in_tok,
  sum(${num("output_tokens")}) AS out_tok,
  sum(${num("total_tokens")}) AS tot_tok,
  sum(${num("cost")}) AS cost
FROM ${source}
WHERE ${where.join(" AND ")}
GROUP BY thread_id`;

    try {
      const rows = await this.engine.query(sql);
      return rows
        .filter((row) => row.thread_id)
        .map((row) => ({
          threadId: String(row.thread_id),
          calls: Number(row.calls ?? 0),
          inputTokens: finiteOrZero(row.in_tok),
          outputTokens: finiteOrZero(row.out_tok),
          totalTokens: finiteOrZero(row.tot_tok),
          costUsd: finiteOrZero(row.cost),
        }));
    } catch (err) {
      console.error("queryThreadUsage failed:", err);
      throw new Error("Monitoring stats unavailable", { cause: err });
    }
  }

  async aggregate(params: AggregationParams): Promise<AggregationResult> {
    if (!params.organizationId) {
      throw new Error("organizationId is required");
    }

    const source = this.sourceFactory(params.organizationId, {
      startDate: params.filters?.startDate,
      endDate: params.filters?.endDate,
    });
    const jsonPath = validateJsonPath(params.path);
    const sourceCol = params.from === "input" ? "input" : "output";

    // Build the value expression using dialect-aware JSON extraction
    const valueExpr =
      params.aggregation === "count" || params.aggregation === "count_all"
        ? "1"
        : jsonExtractFloat(sourceCol, jsonPath, this.dialect);

    // Build aggregation expression
    const aggExpr = buildAggExpr(params.aggregation, valueExpr);

    // Build WHERE clause
    const where: string[] = [
      `organization_id = '${esc(params.organizationId)}'`,
    ];

    // For count (not count_all), only count rows where the path exists and is non-null
    if (params.aggregation === "count") {
      const extractExpr = jsonExtractString(sourceCol, jsonPath, this.dialect);
      where.push(`${extractExpr} IS NOT NULL`);
      where.push(`${extractExpr} != ''`);
    }

    // Exclude count_all from null filtering — it counts all rows
    if (params.aggregation !== "count" && params.aggregation !== "count_all") {
      const extractExpr = jsonExtractString(sourceCol, jsonPath, this.dialect);
      where.push(`${extractExpr} IS NOT NULL`);
      where.push(`${extractExpr} != ''`);
    }

    where.push(...buildCommonFilterClauses(this.dialect, params.filters));

    const whereClause = where.join(" AND ");

    const rowLimit = Math.min(
      Math.max(1, Math.floor(params.limit ?? 10000)),
      10000,
    );

    // --- groupByColumn (takes priority) ---
    if (params.groupByColumn) {
      const col = validateGroupByColumn(params.groupByColumn);
      const sql = `SELECT ${col} AS group_key, ${aggExpr} AS value FROM ${source} WHERE ${whereClause} GROUP BY ${col} ORDER BY value DESC LIMIT ${rowLimit}`;
      const rows = await this.engine.query(sql);
      return {
        value: null,
        groups: rows.map((r) => ({
          key: String(r.group_key ?? ""),
          value: Number(r.value ?? 0),
        })),
      };
    }

    // --- groupBy (JSONPath) ---
    if (params.groupBy) {
      const groupPath = validateJsonPath(params.groupBy);
      const groupSourceCol = params.from === "input" ? "input" : "output";
      const groupExpr = jsonExtractString(
        groupSourceCol,
        groupPath,
        this.dialect,
      );
      const sql = `SELECT ${groupExpr} AS group_key, ${aggExpr} AS value FROM ${source} WHERE ${whereClause} AND ${groupExpr} != '' GROUP BY ${groupExpr} ORDER BY value DESC LIMIT ${rowLimit}`;
      const rows = await this.engine.query(sql);
      return {
        value: null,
        groups: rows.map((r) => ({
          key: String(r.group_key ?? ""),
          value: Number(r.value ?? 0),
        })),
      };
    }

    // --- interval (timeseries) ---
    if (params.interval) {
      const bucketExpr = intervalToSQL(params.interval, this.dialect);
      const sql = `SELECT ${bucketExpr} AS bucket, ${aggExpr} AS value FROM ${source} WHERE ${whereClause} GROUP BY bucket ORDER BY bucket ASC LIMIT ${rowLimit}`;
      const rows = await this.engine.query(sql);
      return {
        value: null,
        timeseries: rows.map((r) => ({
          timestamp: String(r.bucket ?? ""),
          value: Number(r.value ?? 0),
        })),
      };
    }

    // --- simple aggregation ---
    const sql = `SELECT ${aggExpr} AS value FROM ${source} WHERE ${whereClause}`;
    const rows = await this.engine.query(sql);
    return {
      value: Number(rows[0]?.value ?? 0),
    };
  }

  async countMatched(params: {
    organizationId: string;
    path: string;
    from: "input" | "output";
    filters?: {
      connectionIds?: string[];
      toolNames?: string[];
      virtualMcpIds?: string[];
      startDate?: Date;
      endDate?: Date;
      propertyFilters?: PropertyFilters;
    };
  }): Promise<number> {
    if (!params.organizationId) {
      throw new Error("organizationId is required");
    }

    const source = this.sourceFactory(params.organizationId, {
      startDate: params.filters?.startDate,
      endDate: params.filters?.endDate,
    });
    const jsonPath = validateJsonPath(params.path);
    const sourceCol = params.from === "input" ? "input" : "output";
    const extractExpr = jsonExtractString(sourceCol, jsonPath, this.dialect);

    const where: string[] = [
      `organization_id = '${esc(params.organizationId)}'`,
      `${extractExpr} IS NOT NULL`,
      `${extractExpr} != ''`,
    ];

    where.push(...buildCommonFilterClauses(this.dialect, params.filters));

    const sql = `SELECT count(*) AS cnt FROM ${source} WHERE ${where.join(" AND ")}`;
    const rows = await this.engine.query(sql);
    return Number(rows[0]?.cnt ?? 0);
  }
  async queryMetricTimeseries(params: {
    organizationId: string;
    interval: string;
    startDate?: Date;
    endDate?: Date;
    filters?: {
      connectionIds?: string[];
      excludeConnectionIds?: string[];
      toolNames?: string[];
      status?: "success" | "error";
    };
  }): Promise<{
    totalCalls: number;
    totalErrors: number;
    avgDurationMs: number;
    p50DurationMs: number;
    p95DurationMs: number;
    connectionBreakdown: Array<{
      connectionId: string;
      calls: number;
      errors: number;
      errorRate: number;
      avgDurationMs: number;
    }>;
    timeseries: Array<{
      timestamp: string;
      calls: number;
      errors: number;
      errorRate: number;
      avg: number;
      p50: number;
      p95: number;
    }>;
  }> {
    const emptyResult = {
      totalCalls: 0,
      totalErrors: 0,
      avgDurationMs: 0,
      p50DurationMs: 0,
      p95DurationMs: 0,
      connectionBreakdown: [],
      timeseries: [],
    };

    if (!params.organizationId) {
      return emptyResult;
    }

    // ClickHouse derives metrics from the otel_logs view (count/avg/quantile
    // over duration_ms + is_error) — no separate metrics table. The DuckDB path
    // below still reads pre-aggregated histograms from local NDJSON files.
    if (this.dialect === "clickhouse") {
      return this.queryMetricTimeseriesClickHouse(params, emptyResult);
    }
    if (this.metricsFromLogs) {
      return this.queryMetricTimeseriesDuckDBLogs(params, emptyResult);
    }

    // ── DuckDB path (local NDJSON, pre-aggregated histograms) ──────────────
    const source = this.metricSourceFactory(params.organizationId, {
      startDate: params.startDate,
      endDate: params.endDate,
    });
    const bucketExpr = intervalToSQL(
      params.interval,
      this.dialect,
      "timestamp",
    );

    const where: string[] = [
      `organization_id = '${esc(params.organizationId)}'`,
    ];

    applyStartDateBound(where, params.startDate, this.dialect);
    if (params.endDate) {
      where.push(tsLte(params.endDate, this.dialect));
    }
    if (params.filters?.toolNames?.length) {
      const names = params.filters.toolNames
        .slice(0, 100)
        .map((n) => `'${esc(n)}'`)
        .join(",");
      where.push(`tool_name IN (${names})`);
    }
    if (params.filters?.connectionIds?.length) {
      const ids = params.filters.connectionIds
        .slice(0, 100)
        .map((id) => `'${esc(id)}'`)
        .join(",");
      where.push(`connection_id IN (${ids})`);
    }
    if (params.filters?.excludeConnectionIds?.length) {
      const ids = params.filters.excludeConnectionIds
        .slice(0, 100)
        .map((id) => `'${esc(id)}'`)
        .join(",");
      where.push(`connection_id NOT IN (${ids})`);
    }
    // NOTE: status is intentionally NOT added to the WHERE clause here.
    // The errors and errorRate aggregations use SUM FILTER (WHERE ...),
    // which would always return 0 if WHERE already filters to status = 'success'.

    const whereClause = where.join(" AND ");

    try {
      const sql = `SELECT
  ${bucketExpr} AS bucket,
  SUM(value) FILTER (WHERE name = 'tool.execution.count') AS calls,
  SUM(value) FILTER (WHERE name = 'tool.execution.count' AND status = 'error') AS errors,
  SUM(hist_sum) FILTER (WHERE name = 'tool.execution.duration') AS total_hist_sum,
  SUM(hist_count) FILTER (WHERE name = 'tool.execution.duration') AS total_hist_count,
  LIST(hist_boundaries) FILTER (WHERE name = 'tool.execution.duration') AS boundaries_arr,
  LIST(hist_bucket_counts) FILTER (WHERE name = 'tool.execution.duration') AS bucket_counts_arr
FROM ${source}
WHERE ${whereClause}
GROUP BY bucket
ORDER BY bucket ASC`;

      const rows = await this.metricEngine.query(sql);

      const breakdownSql = `SELECT
  connection_id,
  SUM(value) FILTER (WHERE name = 'tool.execution.count') AS calls,
  SUM(value) FILTER (WHERE name = 'tool.execution.count' AND status = 'error') AS errors,
  SUM(hist_sum) FILTER (WHERE name = 'tool.execution.duration') AS total_hist_sum,
  SUM(hist_count) FILTER (WHERE name = 'tool.execution.duration') AS total_hist_count
FROM ${source}
WHERE ${whereClause} AND connection_id != ''
GROUP BY connection_id
ORDER BY calls DESC
LIMIT 1000`;

      const connectionBreakdownRows =
        await this.metricEngine.query(breakdownSql);

      if (rows.length === 0) {
        return {
          ...emptyResult,
          connectionBreakdown: connectionBreakdownRows.map((row) => {
            const calls = Number(row.calls ?? 0);
            const errors = Number(row.errors ?? 0);
            const totalHistSum = Number(row.total_hist_sum ?? 0);
            const totalHistCount = Number(row.total_hist_count ?? 0);

            return {
              connectionId: String(row.connection_id ?? ""),
              calls,
              errors,
              errorRate: calls > 0 ? (errors / calls) * 100 : 0,
              avgDurationMs:
                totalHistCount > 0 ? totalHistSum / totalHistCount : 0,
            };
          }),
        };
      }

      // Accumulate totals across all buckets
      let totalCalls = 0;
      let totalErrors = 0;
      let totalHistSum = 0;
      let totalHistCount = 0;
      const allBoundaries: number[][] = [];
      const allBucketCounts: number[][] = [];

      const timeseries = rows.map((row) => {
        const calls = Number(row.calls ?? 0);
        const errors = Number(row.errors ?? 0);
        const histSum = Number(row.total_hist_sum ?? 0);
        const histCount = Number(row.total_hist_count ?? 0);
        const avg = histCount > 0 ? histSum / histCount : 0;
        const errorRate = calls > 0 ? errors / calls : 0;

        totalCalls += calls;
        totalErrors += errors;
        totalHistSum += histSum;
        totalHistCount += histCount;

        // Merge per-group histogram arrays in JS, then compute percentiles.
        const boundariesArr = parseGroupedArrays(row.boundaries_arr);
        const bucketCountsArr = parseGroupedArrays(row.bucket_counts_arr);
        const merged = mergeHistogramBuckets(boundariesArr, bucketCountsArr);
        const mergedBounds = merged.boundaries;
        const mergedCounts = merged.counts;

        // Accumulate for global percentiles
        if (mergedBounds.length > 0) {
          allBoundaries.push(mergedBounds);
          allBucketCounts.push(mergedCounts);
        }

        const p50 = computePercentileFromHistogramBuckets(
          mergedBounds,
          mergedCounts,
          0.5,
        );
        const p95 = computePercentileFromHistogramBuckets(
          mergedBounds,
          mergedCounts,
          0.95,
        );

        return {
          timestamp: toISOBucket(row.bucket),
          calls,
          errors,
          errorRate,
          avg,
          p50,
          p95,
        };
      });

      // Global percentiles from merged histogram data
      const { boundaries: globalBounds, counts: globalCounts } =
        mergeHistogramBuckets(allBoundaries, allBucketCounts);

      return {
        totalCalls,
        totalErrors,
        avgDurationMs: totalHistCount > 0 ? totalHistSum / totalHistCount : 0,
        p50DurationMs: computePercentileFromHistogramBuckets(
          globalBounds,
          globalCounts,
          0.5,
        ),
        p95DurationMs: computePercentileFromHistogramBuckets(
          globalBounds,
          globalCounts,
          0.95,
        ),
        connectionBreakdown: connectionBreakdownRows.map((row) => {
          const calls = Number(row.calls ?? 0);
          const errors = Number(row.errors ?? 0);
          const totalHistSum = Number(row.total_hist_sum ?? 0);
          const totalHistCount = Number(row.total_hist_count ?? 0);

          return {
            connectionId: String(row.connection_id ?? ""),
            calls,
            errors,
            errorRate: calls > 0 ? (errors / calls) * 100 : 0,
            avgDurationMs:
              totalHistCount > 0 ? totalHistSum / totalHistCount : 0,
          };
        }),
        timeseries,
      };
    } catch (err) {
      console.error("queryMetricTimeseries failed:", err);
      throw new Error("Monitoring stats unavailable", { cause: err });
    }
  }

  /**
   * ClickHouse metric timeseries, derived directly from the otel_logs view.
   * Each tool_call log row carries duration_ms + is_error, so counts, averages
   * and exact percentiles come straight from the rows — no histogram buckets.
   */
  private async queryMetricTimeseriesClickHouse(
    params: {
      organizationId: string;
      interval: string;
      startDate?: Date;
      endDate?: Date;
      filters?: {
        connectionIds?: string[];
        excludeConnectionIds?: string[];
        toolNames?: string[];
        status?: "success" | "error";
      };
    },
    emptyResult: {
      totalCalls: number;
      totalErrors: number;
      avgDurationMs: number;
      p50DurationMs: number;
      p95DurationMs: number;
      connectionBreakdown: Array<{
        connectionId: string;
        calls: number;
        errors: number;
        errorRate: number;
        avgDurationMs: number;
      }>;
      timeseries: Array<{
        timestamp: string;
        calls: number;
        errors: number;
        errorRate: number;
        avg: number;
        p50: number;
        p95: number;
      }>;
    },
  ): Promise<typeof emptyResult> {
    const source = this.metricSourceFactory(params.organizationId, {
      startDate: params.startDate,
      endDate: params.endDate,
    });
    const bucketExpr = intervalToSQL(
      params.interval,
      "clickhouse",
      "timestamp",
    );

    const where: string[] = [
      `organization_id = '${esc(params.organizationId)}'`,
    ];
    applyStartDateBound(where, params.startDate, "clickhouse");
    if (params.endDate) {
      where.push(tsLte(params.endDate, "clickhouse"));
    }
    if (params.filters?.toolNames?.length) {
      const names = params.filters.toolNames
        .slice(0, 100)
        .map((n) => `'${esc(n)}'`)
        .join(",");
      where.push(`tool_name IN (${names})`);
    }
    if (params.filters?.connectionIds?.length) {
      const ids = params.filters.connectionIds
        .slice(0, 100)
        .map((id) => `'${esc(id)}'`)
        .join(",");
      where.push(`connection_id IN (${ids})`);
    }
    if (params.filters?.excludeConnectionIds?.length) {
      const ids = params.filters.excludeConnectionIds
        .slice(0, 100)
        .map((id) => `'${esc(id)}'`)
        .join(",");
      where.push(`connection_id NOT IN (${ids})`);
    }
    const whereClause = where.join(" AND ");

    const timeseriesSql = `SELECT
  ${bucketExpr} AS bucket,
  count() AS calls,
  countIf(is_error = 1) AS errors,
  avg(duration_ms) AS avg,
  quantile(0.5)(duration_ms) AS p50,
  quantile(0.95)(duration_ms) AS p95
FROM ${source}
WHERE ${whereClause}
GROUP BY bucket
ORDER BY bucket ASC`;

    const breakdownSql = `SELECT
  connection_id,
  count() AS calls,
  countIf(is_error = 1) AS errors,
  avg(duration_ms) AS avg
FROM ${source}
WHERE ${whereClause} AND connection_id != ''
GROUP BY connection_id
ORDER BY calls DESC
LIMIT 1000`;

    const totalsSql = `SELECT
  count() AS calls,
  countIf(is_error = 1) AS errors,
  avg(duration_ms) AS avg,
  quantile(0.5)(duration_ms) AS p50,
  quantile(0.95)(duration_ms) AS p95
FROM ${source}
WHERE ${whereClause}`;

    try {
      const [tsRows, breakdownRows, totalsRows] = await Promise.all([
        this.metricEngine.query(timeseriesSql),
        this.metricEngine.query(breakdownSql),
        this.metricEngine.query(totalsSql),
      ]);

      const totals = totalsRows[0] ?? {};

      return {
        totalCalls: Number(totals.calls ?? 0),
        totalErrors: Number(totals.errors ?? 0),
        avgDurationMs: finiteOrZero(totals.avg),
        p50DurationMs: finiteOrZero(totals.p50),
        p95DurationMs: finiteOrZero(totals.p95),
        connectionBreakdown: breakdownRows.map((row) => {
          const calls = Number(row.calls ?? 0);
          const errors = Number(row.errors ?? 0);
          return {
            connectionId: String(row.connection_id ?? ""),
            calls,
            errors,
            errorRate: calls > 0 ? (errors / calls) * 100 : 0,
            avgDurationMs: finiteOrZero(row.avg),
          };
        }),
        timeseries: tsRows.map((row) => {
          const calls = Number(row.calls ?? 0);
          const errors = Number(row.errors ?? 0);
          return {
            timestamp: toISOBucket(row.bucket),
            calls,
            errors,
            errorRate: calls > 0 ? errors / calls : 0,
            avg: finiteOrZero(row.avg),
            p50: finiteOrZero(row.p50),
            p95: finiteOrZero(row.p95),
          };
        }),
      };
    } catch (err) {
      console.error("queryMetricTimeseries failed:", err);
      throw new Error("Monitoring stats unavailable", { cause: err });
    }
  }

  /**
   * DuckDB metric timeseries derived from OTLP flat-log rows (count/avg/quantile
   * over duration_ms + is_error). Same shape as the ClickHouse variant; DuckDB
   * syntax (count(*) / FILTER / quantile_cont / time_bucket).
   */
  private async queryMetricTimeseriesDuckDBLogs(
    params: {
      organizationId: string;
      interval: string;
      startDate?: Date;
      endDate?: Date;
      filters?: {
        connectionIds?: string[];
        excludeConnectionIds?: string[];
        toolNames?: string[];
        status?: "success" | "error";
      };
    },
    emptyResult: {
      totalCalls: number;
      totalErrors: number;
      avgDurationMs: number;
      p50DurationMs: number;
      p95DurationMs: number;
      connectionBreakdown: Array<{
        connectionId: string;
        calls: number;
        errors: number;
        errorRate: number;
        avgDurationMs: number;
      }>;
      timeseries: Array<{
        timestamp: string;
        calls: number;
        errors: number;
        errorRate: number;
        avg: number;
        p50: number;
        p95: number;
      }>;
    },
  ): Promise<typeof emptyResult> {
    const source = this.metricSourceFactory(params.organizationId, {
      startDate: params.startDate,
      endDate: params.endDate,
    });
    const bucketExpr = intervalToSQL(params.interval, "duckdb", "timestamp");

    const where: string[] = [
      `organization_id = '${esc(params.organizationId)}'`,
    ];
    applyStartDateBound(where, params.startDate, "duckdb");
    if (params.endDate) {
      where.push(tsLte(params.endDate, "duckdb"));
    }
    if (params.filters?.toolNames?.length) {
      const names = params.filters.toolNames
        .slice(0, 100)
        .map((n) => `'${esc(n)}'`)
        .join(",");
      where.push(`tool_name IN (${names})`);
    }
    if (params.filters?.connectionIds?.length) {
      const ids = params.filters.connectionIds
        .slice(0, 100)
        .map((id) => `'${esc(id)}'`)
        .join(",");
      where.push(`connection_id IN (${ids})`);
    }
    if (params.filters?.excludeConnectionIds?.length) {
      const ids = params.filters.excludeConnectionIds
        .slice(0, 100)
        .map((id) => `'${esc(id)}'`)
        .join(",");
      where.push(`connection_id NOT IN (${ids})`);
    }
    const whereClause = where.join(" AND ");

    const timeseriesSql = `SELECT
  ${bucketExpr} AS bucket,
  count(*) AS calls,
  count(*) FILTER (WHERE is_error = 1) AS errors,
  avg(duration_ms) AS avg,
  quantile_cont(duration_ms, 0.5) AS p50,
  quantile_cont(duration_ms, 0.95) AS p95
FROM ${source}
WHERE ${whereClause}
GROUP BY bucket
ORDER BY bucket ASC`;

    const breakdownSql = `SELECT
  connection_id,
  count(*) AS calls,
  count(*) FILTER (WHERE is_error = 1) AS errors,
  avg(duration_ms) AS avg
FROM ${source}
WHERE ${whereClause} AND connection_id != ''
GROUP BY connection_id
ORDER BY calls DESC
LIMIT 1000`;

    const totalsSql = `SELECT
  count(*) AS calls,
  count(*) FILTER (WHERE is_error = 1) AS errors,
  avg(duration_ms) AS avg,
  quantile_cont(duration_ms, 0.5) AS p50,
  quantile_cont(duration_ms, 0.95) AS p95
FROM ${source}
WHERE ${whereClause}`;

    try {
      const [tsRows, breakdownRows, totalsRows] = await Promise.all([
        this.metricEngine.query(timeseriesSql),
        this.metricEngine.query(breakdownSql),
        this.metricEngine.query(totalsSql),
      ]);

      const totals = totalsRows[0] ?? {};

      return {
        totalCalls: Number(totals.calls ?? 0),
        totalErrors: Number(totals.errors ?? 0),
        avgDurationMs: finiteOrZero(totals.avg),
        p50DurationMs: finiteOrZero(totals.p50),
        p95DurationMs: finiteOrZero(totals.p95),
        connectionBreakdown: breakdownRows.map((row) => {
          const calls = Number(row.calls ?? 0);
          const errors = Number(row.errors ?? 0);
          return {
            connectionId: String(row.connection_id ?? ""),
            calls,
            errors,
            errorRate: calls > 0 ? (errors / calls) * 100 : 0,
            avgDurationMs: finiteOrZero(row.avg),
          };
        }),
        timeseries: tsRows.map((row) => {
          const calls = Number(row.calls ?? 0);
          const errors = Number(row.errors ?? 0);
          return {
            timestamp: toISOBucket(row.bucket),
            calls,
            errors,
            errorRate: calls > 0 ? errors / calls : 0,
            avg: finiteOrZero(row.avg),
            p50: finiteOrZero(row.p50),
            p95: finiteOrZero(row.p95),
          };
        }),
      };
    } catch (err) {
      console.error("queryMetricTimeseries failed:", err);
      throw new Error("Monitoring stats unavailable", { cause: err });
    }
  }

  async queryMetricTopToolsTimeseries(params: {
    organizationId: string;
    interval: string;
    startDate?: Date;
    endDate?: Date;
    topN?: number;
    filters?: {
      connectionIds?: string[];
      excludeConnectionIds?: string[];
      toolNames?: string[];
      status?: "success" | "error";
    };
  }): Promise<{
    topTools: Array<{
      toolName: string;
      connectionId: string | null;
      calls: number;
    }>;
    timeseries: Array<{
      timestamp: string;
      toolName: string;
      calls: number;
      errors: number;
      avg: number;
      p95: number;
    }>;
  }> {
    const emptyResult = {
      topTools: [],
      timeseries: [],
    };

    if (!params.organizationId) {
      return emptyResult;
    }

    // ClickHouse derives metrics from the otel_logs view; DuckDB path below
    // reads pre-aggregated histograms from local NDJSON files.
    if (this.dialect === "clickhouse") {
      return this.queryMetricTopToolsTimeseriesClickHouse(params, emptyResult);
    }
    if (this.metricsFromLogs) {
      return this.queryMetricTopToolsTimeseriesDuckDBLogs(params, emptyResult);
    }

    // ── DuckDB path (local NDJSON, pre-aggregated histograms) ──────────────
    const source = this.metricSourceFactory(params.organizationId, {
      startDate: params.startDate,
      endDate: params.endDate,
    });
    const bucketExpr = intervalToSQL(
      params.interval,
      this.dialect,
      "timestamp",
    );
    const where: string[] = [
      `organization_id = '${esc(params.organizationId)}'`,
    ];

    applyStartDateBound(where, params.startDate, this.dialect);
    if (params.endDate) {
      where.push(tsLte(params.endDate, this.dialect));
    }
    if (params.filters?.toolNames?.length) {
      const names = params.filters.toolNames
        .slice(0, 100)
        .map((n) => `'${esc(n)}'`)
        .join(",");
      where.push(`tool_name IN (${names})`);
    }
    if (params.filters?.connectionIds?.length) {
      const ids = params.filters.connectionIds
        .slice(0, 100)
        .map((id) => `'${esc(id)}'`)
        .join(",");
      where.push(`connection_id IN (${ids})`);
    }
    if (params.filters?.excludeConnectionIds?.length) {
      const ids = params.filters.excludeConnectionIds
        .slice(0, 100)
        .map((id) => `'${esc(id)}'`)
        .join(",");
      where.push(`connection_id NOT IN (${ids})`);
    }
    // NOTE: status is intentionally NOT added to the WHERE clause here.
    // The errors aggregation uses SUM FILTER (WHERE ...), which would always
    // return 0 if WHERE already filters to status = 'success'.

    const whereClause = where.join(" AND ");
    const topN = Math.min(Math.max(1, Math.floor(params.topN ?? 10)), 20);

    try {
      const topToolsSql = `SELECT
  tool_name,
  argMax(connection_id, connection_calls) AS connection_id,
  sum(connection_calls) AS calls
FROM (
  SELECT
    tool_name,
    connection_id,
    SUM(value) FILTER (WHERE name = 'tool.execution.count') AS connection_calls
  FROM ${source}
  WHERE ${whereClause} AND tool_name != ''
  GROUP BY tool_name, connection_id
)
GROUP BY tool_name
ORDER BY calls DESC
LIMIT ${topN}`;

      const topToolRows = await this.metricEngine.query(topToolsSql);

      if (topToolRows.length === 0) {
        return emptyResult;
      }

      const topToolNames = topToolRows
        .map((row) => String(row.tool_name ?? ""))
        .filter(Boolean);
      const toolNamesSql = topToolNames
        .map((name) => `'${esc(name)}'`)
        .join(",");

      const timeseriesSql = `SELECT
  ${bucketExpr} AS bucket,
  tool_name,
  SUM(value) FILTER (WHERE name = 'tool.execution.count') AS calls,
  SUM(value) FILTER (WHERE name = 'tool.execution.count' AND status = 'error') AS errors,
  SUM(hist_sum) FILTER (WHERE name = 'tool.execution.duration') AS total_hist_sum,
  SUM(hist_count) FILTER (WHERE name = 'tool.execution.duration') AS total_hist_count,
  LIST(hist_boundaries) FILTER (WHERE name = 'tool.execution.duration') AS boundaries_arr,
  LIST(hist_bucket_counts) FILTER (WHERE name = 'tool.execution.duration') AS bucket_counts_arr
FROM ${source}
WHERE ${whereClause} AND tool_name IN (${toolNamesSql})
GROUP BY bucket, tool_name
ORDER BY bucket ASC, tool_name ASC`;

      const rows = await this.metricEngine.query(timeseriesSql);

      return {
        topTools: topToolRows.map((row) => ({
          toolName: String(row.tool_name ?? ""),
          connectionId:
            row.connection_id != null ? String(row.connection_id) : null,
          calls: Number(row.calls ?? 0),
        })),
        timeseries: rows.map((row) => {
          const calls = Number(row.calls ?? 0);
          const errors = Number(row.errors ?? 0);
          const histSum = Number(row.total_hist_sum ?? 0);
          const histCount = Number(row.total_hist_count ?? 0);
          const boundariesArr = parseGroupedArrays(row.boundaries_arr);
          const bucketCountsArr = parseGroupedArrays(row.bucket_counts_arr);
          const merged = mergeHistogramBuckets(boundariesArr, bucketCountsArr);

          return {
            timestamp: toISOBucket(row.bucket),
            toolName: String(row.tool_name ?? ""),
            calls,
            errors,
            avg: histCount > 0 ? histSum / histCount : 0,
            p95: computePercentileFromHistogramBuckets(
              merged.boundaries,
              merged.counts,
              0.95,
            ),
          };
        }),
      };
    } catch (err) {
      console.error("queryMetricTopToolsTimeseries failed:", err);
      throw new Error("Monitoring stats unavailable", { cause: err });
    }
  }

  /**
   * ClickHouse top-tools metric timeseries, derived from the otel_logs view
   * (count/avg/quantile over duration_ms + is_error).
   */
  private async queryMetricTopToolsTimeseriesClickHouse(
    params: {
      organizationId: string;
      interval: string;
      startDate?: Date;
      endDate?: Date;
      topN?: number;
      filters?: {
        connectionIds?: string[];
        excludeConnectionIds?: string[];
        toolNames?: string[];
        status?: "success" | "error";
      };
    },
    emptyResult: {
      topTools: Array<{
        toolName: string;
        connectionId: string | null;
        calls: number;
      }>;
      timeseries: Array<{
        timestamp: string;
        toolName: string;
        calls: number;
        errors: number;
        avg: number;
        p95: number;
      }>;
    },
  ): Promise<typeof emptyResult> {
    const source = this.metricSourceFactory(params.organizationId, {
      startDate: params.startDate,
      endDate: params.endDate,
    });
    const bucketExpr = intervalToSQL(
      params.interval,
      "clickhouse",
      "timestamp",
    );
    const where: string[] = [
      `organization_id = '${esc(params.organizationId)}'`,
    ];
    applyStartDateBound(where, params.startDate, "clickhouse");
    if (params.endDate) {
      where.push(tsLte(params.endDate, "clickhouse"));
    }
    if (params.filters?.toolNames?.length) {
      const names = params.filters.toolNames
        .slice(0, 100)
        .map((n) => `'${esc(n)}'`)
        .join(",");
      where.push(`tool_name IN (${names})`);
    }
    if (params.filters?.connectionIds?.length) {
      const ids = params.filters.connectionIds
        .slice(0, 100)
        .map((id) => `'${esc(id)}'`)
        .join(",");
      where.push(`connection_id IN (${ids})`);
    }
    if (params.filters?.excludeConnectionIds?.length) {
      const ids = params.filters.excludeConnectionIds
        .slice(0, 100)
        .map((id) => `'${esc(id)}'`)
        .join(",");
      where.push(`connection_id NOT IN (${ids})`);
    }
    const whereClause = where.join(" AND ");
    const topN = Math.min(Math.max(1, Math.floor(params.topN ?? 10)), 20);

    try {
      const topToolsSql = `SELECT
  tool_name,
  argMax(connection_id, conn_calls) AS connection_id,
  sum(conn_calls) AS calls
FROM (
  SELECT tool_name, connection_id, count() AS conn_calls
  FROM ${source}
  WHERE ${whereClause} AND tool_name != ''
  GROUP BY tool_name, connection_id
)
GROUP BY tool_name
ORDER BY calls DESC
LIMIT ${topN}`;

      const topToolRows = await this.metricEngine.query(topToolsSql);
      if (topToolRows.length === 0) {
        return emptyResult;
      }

      const topToolNames = topToolRows
        .map((row) => String(row.tool_name ?? ""))
        .filter(Boolean);
      const toolNamesSql = topToolNames
        .map((name) => `'${esc(name)}'`)
        .join(",");

      const timeseriesSql = `SELECT
  ${bucketExpr} AS bucket,
  tool_name,
  count() AS calls,
  countIf(is_error = 1) AS errors,
  avg(duration_ms) AS avg,
  quantile(0.95)(duration_ms) AS p95
FROM ${source}
WHERE ${whereClause} AND tool_name IN (${toolNamesSql})
GROUP BY bucket, tool_name
ORDER BY bucket ASC, tool_name ASC`;

      const rows = await this.metricEngine.query(timeseriesSql);

      return {
        topTools: topToolRows.map((row) => ({
          toolName: String(row.tool_name ?? ""),
          connectionId:
            row.connection_id != null ? String(row.connection_id) : null,
          calls: Number(row.calls ?? 0),
        })),
        timeseries: rows.map((row) => ({
          timestamp: toISOBucket(row.bucket),
          toolName: String(row.tool_name ?? ""),
          calls: Number(row.calls ?? 0),
          errors: Number(row.errors ?? 0),
          avg: finiteOrZero(row.avg),
          p95: finiteOrZero(row.p95),
        })),
      };
    } catch (err) {
      console.error("queryMetricTopToolsTimeseries failed:", err);
      throw new Error("Monitoring stats unavailable", { cause: err });
    }
  }

  /**
   * DuckDB top-tools metric timeseries derived from OTLP flat-log rows. Same
   * shape as the ClickHouse variant; DuckDB syntax (count(*) / FILTER /
   * quantile_cont / arg_max / time_bucket).
   */
  private async queryMetricTopToolsTimeseriesDuckDBLogs(
    params: {
      organizationId: string;
      interval: string;
      startDate?: Date;
      endDate?: Date;
      topN?: number;
      filters?: {
        connectionIds?: string[];
        excludeConnectionIds?: string[];
        toolNames?: string[];
        status?: "success" | "error";
      };
    },
    emptyResult: {
      topTools: Array<{
        toolName: string;
        connectionId: string | null;
        calls: number;
      }>;
      timeseries: Array<{
        timestamp: string;
        toolName: string;
        calls: number;
        errors: number;
        avg: number;
        p95: number;
      }>;
    },
  ): Promise<typeof emptyResult> {
    const source = this.metricSourceFactory(params.organizationId, {
      startDate: params.startDate,
      endDate: params.endDate,
    });
    const bucketExpr = intervalToSQL(params.interval, "duckdb", "timestamp");
    const where: string[] = [
      `organization_id = '${esc(params.organizationId)}'`,
    ];
    applyStartDateBound(where, params.startDate, "duckdb");
    if (params.endDate) {
      where.push(tsLte(params.endDate, "duckdb"));
    }
    if (params.filters?.toolNames?.length) {
      const names = params.filters.toolNames
        .slice(0, 100)
        .map((n) => `'${esc(n)}'`)
        .join(",");
      where.push(`tool_name IN (${names})`);
    }
    if (params.filters?.connectionIds?.length) {
      const ids = params.filters.connectionIds
        .slice(0, 100)
        .map((id) => `'${esc(id)}'`)
        .join(",");
      where.push(`connection_id IN (${ids})`);
    }
    if (params.filters?.excludeConnectionIds?.length) {
      const ids = params.filters.excludeConnectionIds
        .slice(0, 100)
        .map((id) => `'${esc(id)}'`)
        .join(",");
      where.push(`connection_id NOT IN (${ids})`);
    }
    const whereClause = where.join(" AND ");
    const topN = Math.min(Math.max(1, Math.floor(params.topN ?? 10)), 20);

    try {
      const topToolsSql = `SELECT
  tool_name,
  arg_max(connection_id, conn_calls) AS connection_id,
  sum(conn_calls) AS calls
FROM (
  SELECT tool_name, connection_id, count(*) AS conn_calls
  FROM ${source}
  WHERE ${whereClause} AND tool_name != ''
  GROUP BY tool_name, connection_id
)
GROUP BY tool_name
ORDER BY calls DESC
LIMIT ${topN}`;

      const topToolRows = await this.metricEngine.query(topToolsSql);
      if (topToolRows.length === 0) {
        return emptyResult;
      }

      const topToolNames = topToolRows
        .map((row) => String(row.tool_name ?? ""))
        .filter(Boolean);
      const toolNamesSql = topToolNames
        .map((name) => `'${esc(name)}'`)
        .join(",");

      const timeseriesSql = `SELECT
  ${bucketExpr} AS bucket,
  tool_name,
  count(*) AS calls,
  count(*) FILTER (WHERE is_error = 1) AS errors,
  avg(duration_ms) AS avg,
  quantile_cont(duration_ms, 0.95) AS p95
FROM ${source}
WHERE ${whereClause} AND tool_name IN (${toolNamesSql})
GROUP BY bucket, tool_name
ORDER BY bucket ASC, tool_name ASC`;

      const rows = await this.metricEngine.query(timeseriesSql);

      return {
        topTools: topToolRows.map((row) => ({
          toolName: String(row.tool_name ?? ""),
          connectionId:
            row.connection_id != null ? String(row.connection_id) : null,
          calls: Number(row.calls ?? 0),
        })),
        timeseries: rows.map((row) => ({
          timestamp: toISOBucket(row.bucket),
          toolName: String(row.tool_name ?? ""),
          calls: Number(row.calls ?? 0),
          errors: Number(row.errors ?? 0),
          avg: finiteOrZero(row.avg),
          p95: finiteOrZero(row.p95),
        })),
      };
    } catch (err) {
      console.error("queryMetricTopToolsTimeseries failed:", err);
      throw new Error("Monitoring stats unavailable", { cause: err });
    }
  }
}

// ---------------------------------------------------------------------------
// Histogram helpers
// ---------------------------------------------------------------------------

/**
 * Parse grouped array results from DuckDB.
 * LIST returns an array of JSON strings (each being a serialized array).
 */
export function parseGroupedArrays(val: unknown): number[][] {
  if (!val) return [];
  if (Array.isArray(val)) {
    return val.map((item) => {
      if (typeof item === "string") {
        try {
          return JSON.parse(item) as number[];
        } catch {
          return [];
        }
      }
      if (Array.isArray(item)) return item as number[];
      return [];
    });
  }
  if (typeof val === "string") {
    try {
      const parsed = JSON.parse(val);
      if (Array.isArray(parsed)) {
        return parsed.map((item: unknown) => {
          if (typeof item === "string") {
            try {
              return JSON.parse(item) as number[];
            } catch {
              return [];
            }
          }
          if (Array.isArray(item)) return item as number[];
          return [];
        });
      }
    } catch {
      return [];
    }
  }
  return [];
}

/**
 * Merge multiple histogram bucket arrays into one.
 * All histograms must share the same boundaries (OTel SDK uses consistent boundaries).
 * Bucket counts are additive (delta temporality).
 */
export function mergeHistogramBuckets(
  boundariesArrays: number[][],
  countsArrays: number[][],
): { boundaries: number[]; counts: number[] } {
  if (boundariesArrays.length === 0 || countsArrays.length === 0) {
    return { boundaries: [], counts: [] };
  }

  // Use the first non-empty boundaries as reference
  let refBoundaries: number[] = [];
  for (const b of boundariesArrays) {
    if (b.length > 0) {
      refBoundaries = b;
      break;
    }
  }

  if (refBoundaries.length === 0) {
    return { boundaries: [], counts: [] };
  }

  // Expected bucket count = boundaries + 1 (one bucket per boundary gap, plus overflow)
  const expectedBuckets = refBoundaries.length + 1;
  const mergedCounts = new Array<number>(expectedBuckets).fill(0);

  for (const counts of countsArrays) {
    if (counts.length === 0) continue;
    const len = Math.min(counts.length, expectedBuckets);
    for (let i = 0; i < len; i++) {
      mergedCounts[i]! += counts[i]!;
    }
  }

  return { boundaries: refBoundaries, counts: mergedCounts };
}

/**
 * Compute a percentile from histogram bucket boundaries and counts
 * using linear interpolation.
 *
 * @param boundaries - Sorted upper boundaries of histogram buckets (N values)
 * @param counts - Counts for each bucket (N+1 values: one per boundary gap + overflow)
 * @param p - Percentile to compute (0-1, e.g. 0.5 for p50)
 */
export function computePercentileFromHistogramBuckets(
  boundaries: number[],
  counts: number[],
  p: number,
): number {
  if (boundaries.length === 0 || counts.length === 0) return 0;

  const totalCount = counts.reduce((a, b) => a + b, 0);
  if (totalCount === 0) return 0;

  const targetCount = p * totalCount;
  let cumulativeCount = 0;

  // Buckets: (-inf, b[0]], (b[0], b[1]], ..., (b[n-1], +inf)
  for (let i = 0; i < counts.length; i++) {
    const prevCumulative = cumulativeCount;
    const bucketCount = counts[i] ?? 0;
    cumulativeCount += bucketCount;

    if (cumulativeCount >= targetCount) {
      // The percentile falls in this bucket
      const bucketLower = i === 0 ? 0 : (boundaries[i - 1] ?? 0);
      const bucketUpper =
        i < boundaries.length
          ? (boundaries[i] ?? 0)
          : (boundaries[boundaries.length - 1] ?? 0) * 2;

      // Linear interpolation within the bucket
      const bucketFraction =
        bucketCount > 0 ? (targetCount - prevCumulative) / bucketCount : 0;
      return bucketLower + bucketFraction * (bucketUpper - bucketLower);
    }
  }

  // Fallback: return the last boundary
  return boundaries[boundaries.length - 1] ?? 0;
}

// ---------------------------------------------------------------------------
// Aggregation expression builder
// ---------------------------------------------------------------------------

function buildAggExpr(fn: string, valueExpr: string): string {
  switch (fn) {
    case "sum":
      return `coalesce(sum(${valueExpr}), 0)`;
    case "avg":
      return `coalesce(avg(${valueExpr}), 0)`;
    case "min":
      return `min(${valueExpr})`;
    case "max":
      return `max(${valueExpr})`;
    case "count":
      return `count(*)`;
    case "count_all":
      return `count(*)`;
    case "last":
      return `argMax(${valueExpr}, timestamp)`;
    default:
      throw new Error(`Unknown aggregation function: ${fn}`);
  }
}
