diff --git a/apps/api/src/routes/queries.ts b/apps/api/src/routes/queries.ts new file mode 100644 index 000000000..072b13cdc --- /dev/null +++ b/apps/api/src/routes/queries.ts @@ -0,0 +1,197 @@ +import * as CH from "@maple/query-engine/ch" +import * as Integrations from "@maple/query-engine-integrations" +import { defineQuery } from "@maple/query-engine/registry" +import { Queries as Core } from "@maple/query-engine/registry" +import type { + CloudflareInfraWorkerTimeseriesRequest, + CloudflareInfraZoneTimeseriesRequest, + FleetUtilizationTimeseriesRequest, + GetLogRequest, + NodeInfraTimeseriesRequest, + PodInfraTimeseriesRequest, + SpanDetailRequest, + WorkloadInfraTimeseriesRequest, +} from "@maple/domain/http" +import { + nodeMetricSpec, + partitionWindowAround, + podMetricSpec, + toCloudflareFilters, + workloadMetricSpec, +} from "@/routes/query-helpers" +import { traceCacheTtlSeconds } from "@/services/warehouse/trace-detail-cache" + +/** + * App-side half of the warehouse query registry. + * + * Most entries live in `@maple/query-engine/registry`. These do not, for two + * reasons that are both about dependency direction rather than taste: + * + * * The Cloudflare and PlanetScale queries are built by + * `@maple/query-engine-integrations`, which itself depends on + * `@maple/query-engine`. Declaring them in the core registry would invert + * that edge. + * * A few queries need helpers or services that belong to the API app — + * `partitionWindowAround`, `traceCacheTtlSeconds` — and pulling those down + * into the query-engine package would drag app concerns into a shared lib. + * + * Handlers import `Queries` from here, so the split is invisible at the call + * site and an entry can move between the two halves without touching handlers. + */ +export const Queries = { + ...Core, + + /** + * Bounded to a ±1h window around the requested log so ClickHouse can prune + * partitions instead of reading every retained daily partition for an + * exact-timestamp match. That window used to be computed in the handler. + */ + getLog: defineQuery({ + id: "getLog", + profile: "list", + cache: undefined, + compile: (payload: GetLogRequest, orgId: string) => { + const { startTime, endTime } = partitionWindowAround(payload.timestamp) + return CH.compile( + CH.getLogByKeyQuery({ + serviceName: payload.serviceName, + traceId: payload.traceId, + spanId: payload.spanId, + }), + { orgId, startTime, endTime, timestamp: payload.timestamp }, + ) + }, + }), + + /** + * A finished trace is immutable and cacheable; one still receiving spans is + * not. `traceCacheTtlSeconds` decides from the requested end time against + * now, which is why this def takes the dynamic-cache form. + */ + spanDetail: defineQuery({ + id: "spanDetail", + profile: "discovery", + cache: (payload: SpanDetailRequest, nowMs: number) => traceCacheTtlSeconds(payload.endTime, nowMs), + compile: (payload: SpanDetailRequest, orgId: string) => { + // Without both bounds there is no window to narrow to, and passing a + // half-open range would widen the scan rather than prune it. + const narrowByTime = payload.startTime != null && payload.endTime != null + return CH.compile( + CH.spanDetailQuery({ + traceId: payload.traceId, + spanId: payload.spanId, + narrowByTime, + }), + narrowByTime ? { orgId, startTime: payload.startTime, endTime: payload.endTime } : { orgId }, + ) + }, + }), + + fleetUtilizationTimeseries: defineQuery({ + id: "fleetUtilizationTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: FleetUtilizationTimeseriesRequest, orgId: string) => + CH.compile(CH.fleetUtilizationTimeseriesQuery(), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + bucketSeconds: payload.bucketSeconds ?? 300, + }), + }), + + podInfraTimeseries: defineQuery({ + id: "podInfraTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: PodInfraTimeseriesRequest, orgId: string) => + CH.compile( + CH.podGaugeTimeseriesQuery({ + podName: payload.podName, + namespace: payload.namespace, + metricName: podMetricSpec(payload.metric).metricName, + }), + { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + bucketSeconds: payload.bucketSeconds ?? 60, + }, + ), + }), + + nodeInfraTimeseries: defineQuery({ + id: "nodeInfraTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: NodeInfraTimeseriesRequest, orgId: string) => + CH.compile( + CH.nodeGaugeTimeseriesQuery({ + nodeName: payload.nodeName, + metricName: nodeMetricSpec(payload.metric).metricName, + }), + { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + bucketSeconds: payload.bucketSeconds ?? 60, + }, + ), + }), + + workloadInfraTimeseries: defineQuery({ + id: "workloadInfraTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: WorkloadInfraTimeseriesRequest, orgId: string) => + CH.compile( + CH.workloadGaugeTimeseriesQuery({ + kind: payload.kind, + workloadName: payload.workloadName, + namespace: payload.namespace, + metricName: workloadMetricSpec(payload.metric).metricName, + groupByPod: payload.groupByPod, + }), + { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + bucketSeconds: payload.bucketSeconds ?? 60, + }, + ), + }), + + cloudflareInfraZoneTimeseries: defineQuery({ + id: "cloudflareInfraZoneTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: CloudflareInfraZoneTimeseriesRequest, orgId: string) => + CH.compile( + Integrations.cloudflareZoneTimeseriesSQL(toCloudflareFilters(payload)), + { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + bucketSeconds: payload.bucketSeconds, + }, + { rowSchema: Integrations.cloudflareZoneTimeseriesRowSchema }, + ), + }), + + cloudflareInfraWorkerTimeseries: defineQuery({ + id: "cloudflareInfraWorkerTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: CloudflareInfraWorkerTimeseriesRequest, orgId: string) => + CH.compile( + Integrations.cloudflareWorkerTimeseriesSQL(), + { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + bucketSeconds: payload.bucketSeconds, + }, + { rowSchema: Integrations.cloudflareWorkerTimeseriesRowSchema }, + ), + }), +} as const diff --git a/apps/api/src/routes/query-helpers.ts b/apps/api/src/routes/query-helpers.ts new file mode 100644 index 000000000..c5278bde1 --- /dev/null +++ b/apps/api/src/routes/query-helpers.ts @@ -0,0 +1,113 @@ +import * as Integrations from "@maple/query-engine-integrations" +import { formatWarehouseDateTime, parseWarehouseDateTime } from "@maple/query-engine" +import type { + NodeInfraTimeseriesRequest, + PodInfraTimeseriesRequest, + WorkloadInfraTimeseriesRequest, +} from "@maple/domain/http" + +/** + * Helpers shared between the query-engine handlers and the app-side query + * registry (`./queries`). + * + * They live here rather than in either caller because both need them: the + * registry's `compile` needs the metric name, while the handler needs the unit + * for the response. Duplicating the switch would let those two drift, which is + * exactly the failure the registry exists to prevent. + */ + +export const toCloudflareFilters = (payload: { + readonly hosts?: ReadonlyArray | undefined + readonly cacheStatuses?: ReadonlyArray | undefined + readonly statusClasses?: ReadonlyArray | undefined + readonly paths?: ReadonlyArray | undefined + readonly pathContains?: string | undefined + readonly countries?: ReadonlyArray | undefined + readonly methods?: ReadonlyArray | undefined + readonly protocols?: ReadonlyArray | undefined + readonly deviceTypes?: ReadonlyArray | undefined + readonly firewallActions?: ReadonlyArray | undefined + readonly firewallSources?: ReadonlyArray | undefined + readonly firewallRuleIds?: ReadonlyArray | undefined + readonly dnsQueryNames?: ReadonlyArray | undefined + readonly dnsResponseCodes?: ReadonlyArray | undefined +}): Integrations.CloudflareFilterOpts => ({ + hosts: payload.hosts, + cacheStatuses: payload.cacheStatuses, + statusClasses: payload.statusClasses, + paths: payload.paths, + pathContains: payload.pathContains, + countries: payload.countries, + methods: payload.methods, + protocols: payload.protocols, + deviceTypes: payload.deviceTypes, + firewallActions: payload.firewallActions, + firewallSources: payload.firewallSources, + firewallRuleIds: payload.firewallRuleIds, + dnsQueryNames: payload.dnsQueryNames, + dnsResponseCodes: payload.dnsResponseCodes, +}) + +export const partitionWindowAround = (timestamp: string): { startTime: string; endTime: string } => { + const ms = parseWarehouseDateTime(timestamp) + return { + startTime: formatWarehouseDateTime(ms - 3_600_000), + endTime: formatWarehouseDateTime(ms + 3_600_000), + } +} + +/** Metric name + response unit for a pod infra metric. */ +export const podMetricSpec = (metric: PodInfraTimeseriesRequest["metric"]) => { + switch (metric) { + case "cpu_usage": + return { metricName: "k8s.pod.cpu.usage", unit: "cores" as const } + case "cpu_limit": + return { + metricName: "k8s.pod.cpu_limit_utilization", + unit: "percent" as const, + } + case "cpu_request": + return { + metricName: "k8s.pod.cpu_request_utilization", + unit: "percent" as const, + } + case "memory_limit": + return { + metricName: "k8s.pod.memory_limit_utilization", + unit: "percent" as const, + } + case "memory_request": + return { + metricName: "k8s.pod.memory_request_utilization", + unit: "percent" as const, + } + } +} + +/** Metric name + response unit for a node infra metric. */ +export const nodeMetricSpec = (metric: NodeInfraTimeseriesRequest["metric"]) => { + switch (metric) { + case "cpu_usage": + return { metricName: "k8s.node.cpu.usage", unit: "cores" as const } + case "uptime": + return { metricName: "k8s.node.uptime", unit: "seconds" as const } + } +} + +/** Metric name + response unit for a workload infra metric. */ +export const workloadMetricSpec = (metric: WorkloadInfraTimeseriesRequest["metric"]) => { + switch (metric) { + case "cpu_usage": + return { metricName: "k8s.pod.cpu.usage", unit: "cores" as const } + case "cpu_limit": + return { + metricName: "k8s.pod.cpu_limit_utilization", + unit: "percent" as const, + } + case "memory_limit": + return { + metricName: "k8s.pod.memory_limit_utilization", + unit: "percent" as const, + } + } +} diff --git a/apps/api/src/routes/query-runner.ts b/apps/api/src/routes/query-runner.ts new file mode 100644 index 000000000..db7e788dc --- /dev/null +++ b/apps/api/src/routes/query-runner.ts @@ -0,0 +1,120 @@ +import type { QueryDef } from "@maple/query-engine/registry" +import type { QueryEngineDirectError } from "@maple/query-engine/runtime" +import { Clock, Effect, Option } from "effect" +import type { TenantContext } from "@/services/auth/AuthService" +import { QueryEngineService } from "@/services/warehouse/QueryEngineService" +import { WarehouseQueryService } from "@/services/warehouse/WarehouseQueryService" + +/** + * Execute registry-declared warehouse queries. + * + * This is the single place that applies a `QueryDef`'s cost profile, settings, + * span context, error label and cache policy — the five things that used to be + * hand-repeated in each of 61 handlers, where the cache in particular was + * silently omitted 50 times. + * + * Handlers keep their own row-to-response mapping. What they lose is the + * plumbing, not the presentation. + */ + +/** + * Settings may be static or payload-dependent. Resolved to a spread so an + * absent/undefined result omits the key entirely rather than passing an + * explicit `settings: undefined`, which would read as "clear the profile + * defaults" downstream. + */ +const resolveSettings = (def: QueryDef, payload: Payload) => { + const settings = typeof def.settings === "function" ? def.settings(payload) : def.settings + return settings === undefined ? {} : { settings } +} + +/** + * Shared tail: annotate failures with the query id, then apply the declared + * cache policy (or don't). `cachedDirect` wraps the whole execution so a hit + * skips the warehouse entirely, and it takes the raw payload as key input — + * it snaps timestamps and sorts set-valued keys itself, which is why the + * payload goes in unnormalized. + */ +const withPolicy = ( + def: QueryDef, + tenant: TenantContext, + payload: Payload, + execute: Effect.Effect, +) => + Effect.gen(function* () { + const queryEngine = yield* QueryEngineService + // Only read the clock when a def actually needs it — a static policy must + // not pay for, or depend on, a Clock read. + const cache = + typeof def.cache === "function" + ? def.cache(payload, yield* Clock.currentTimeMillis) + : def.cache + const labelled = execute.pipe( + // Same annotation the old inline `mapExecError` produced, with the label + // derived from `def.id` rather than a hand-written string that could + // disagree with the span context beside it. + Effect.tapError(() => + Effect.annotateCurrentSpan({ + "maple.query_engine.failed_step": `${def.id} query failed`, + }), + ), + ) + if (cache === undefined) { + return yield* labelled + } + return yield* queryEngine.cachedDirect(tenant, def.id, payload, labelled, cache) + }) + +/** + * Run a `QueryDef` that returns many rows. + * + * Rows-vs-first-row is a call-site concern rather than a field on the def: the + * same compiled query legitimately supports both, and `compiledQueryFirst` + * takes the identical `CompiledQuery`. Putting it in the def would only let a + * caller disagree with it. + */ +export const runQuery = ( + def: QueryDef, + tenant: TenantContext, + payload: Payload, +) => + Effect.gen(function* () { + const warehouse = yield* WarehouseQueryService + return yield* withPolicy( + def, + tenant, + payload, + warehouse.compiledQuery(tenant, def.compile(payload, tenant.orgId), { + profile: def.profile, + ...resolveSettings(def, payload), + context: def.id, + }), + ) + }) + +/** + * Run a `QueryDef` that returns at most one row, as `Row | null`. + * + * Null rather than `Option` because every current caller immediately does + * `Option.getOrNull` to build a nullable response field. + */ +export const runQueryFirst = ( + def: QueryDef, + tenant: TenantContext, + payload: Payload, +) => + Effect.gen(function* () { + const warehouse = yield* WarehouseQueryService + return yield* withPolicy( + def, + tenant, + payload, + warehouse + .compiledQueryFirst(tenant, def.compile(payload, tenant.orgId), { + profile: def.profile, + ...resolveSettings(def, payload), + context: def.id, + }) + .pipe(Effect.map(Option.getOrNull)), + ) + }) diff --git a/apps/api/src/routes/v1/query-engine.http.ts b/apps/api/src/routes/v1/query-engine.http.ts index d738c6caa..6c430d6aa 100644 --- a/apps/api/src/routes/v1/query-engine.http.ts +++ b/apps/api/src/routes/v1/query-engine.http.ts @@ -83,6 +83,15 @@ import { parseWarehouseDateTime, } from "@maple/query-engine" import { LOGS_BODY_SEARCH_SETTINGS } from "@maple/query-engine/profiles" +import { + nodeMetricSpec, + partitionWindowAround, + podMetricSpec, + toCloudflareFilters, + workloadMetricSpec, +} from "@/routes/query-helpers" +import { Queries } from "@/routes/queries" +import { runQuery, runQueryFirst } from "@/routes/query-runner" import type { ExecutionTenant, WarehouseSqlError } from "@maple/query-engine/execution" import { buildBreakdownQuerySpec, buildTimeseriesQuerySpec } from "@maple/query-engine/query-builder" import * as Integrations from "@maple/query-engine-integrations" @@ -103,37 +112,6 @@ const mapExecError = (effect: Effect.Effect, context: string): * `Integrations.cloudflareIgnoredFiltersFor` answers that, and the answer ships in the response so the UI can * mark a panel zone-wide instead of pretending the filter applied. */ -const toCloudflareFilters = (payload: { - readonly hosts?: ReadonlyArray | undefined - readonly cacheStatuses?: ReadonlyArray | undefined - readonly statusClasses?: ReadonlyArray | undefined - readonly paths?: ReadonlyArray | undefined - readonly pathContains?: string | undefined - readonly countries?: ReadonlyArray | undefined - readonly methods?: ReadonlyArray | undefined - readonly protocols?: ReadonlyArray | undefined - readonly deviceTypes?: ReadonlyArray | undefined - readonly firewallActions?: ReadonlyArray | undefined - readonly firewallSources?: ReadonlyArray | undefined - readonly firewallRuleIds?: ReadonlyArray | undefined - readonly dnsQueryNames?: ReadonlyArray | undefined - readonly dnsResponseCodes?: ReadonlyArray | undefined -}): Integrations.CloudflareFilterOpts => ({ - hosts: payload.hosts, - cacheStatuses: payload.cacheStatuses, - statusClasses: payload.statusClasses, - paths: payload.paths, - pathContains: payload.pathContains, - countries: payload.countries, - methods: payload.methods, - protocols: payload.protocols, - deviceTypes: payload.deviceTypes, - firewallActions: payload.firewallActions, - firewallSources: payload.firewallSources, - firewallRuleIds: payload.firewallRuleIds, - dnsQueryNames: payload.dnsQueryNames, - dnsResponseCodes: payload.dnsResponseCodes, -}) const isMissingServiceOperationsRollup = (error: unknown): boolean => { if (typeof error !== "object" || error === null) return false @@ -163,15 +141,6 @@ const decodeStatusCodeOption = Schema.decodeUnknownOption(StatusCode) const coerceStatusCode = (value: string): StatusCode => Option.getOrElse(decodeStatusCodeOption(value), () => "Unset" as const) -// Build a ±1h partition-pruning window around a ClickHouse datetime string. -const partitionWindowAround = (timestamp: string): { startTime: string; endTime: string } => { - const ms = parseWarehouseDateTime(timestamp) - return { - startTime: formatWarehouseDateTime(ms - 3_600_000), - endTime: formatWarehouseDateTime(ms + 3_600_000), - } -} - // Most traces opened without a timestamp are still recent (list rows carry // `?t=`; it's direct/shared/AI links that don't, and those overwhelmingly // point at fresh traces). Probing the last 48h first prunes to ~2 daily @@ -288,33 +257,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("spanDetail", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const nowMs = yield* Clock.currentTimeMillis - const narrowByTime = payload.startTime != null && payload.endTime != null - const compiled = CH.compile( - CH.spanDetailQuery({ - traceId: payload.traceId, - spanId: payload.spanId, - narrowByTime, - }), - narrowByTime - ? { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime } - : { orgId: tenant.orgId }, - ) - const row = yield* queryEngine.cachedDirect( - tenant, - "spanDetail", - payload, - mapExecError( - warehouse - .compiledQueryFirst(tenant, compiled, { - profile: "discovery", - context: "spanDetail", - }) - .pipe(Effect.map(Option.getOrNull)), - "spanDetail query failed", - ), - traceCacheTtlSeconds(payload.endTime, nowMs), - ) + const row = yield* runQueryFirst(Queries.spanDetail, tenant, payload) return new SpanDetailResponse({ data: row ? { @@ -329,26 +272,9 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("errorsByType", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.errorsByTypeQuery({ - rootOnly: payload.rootOnly, - services: payload.services, - deploymentEnvs: payload.deploymentEnvs, - fingerprintHashes: payload.fingerprintHashes, - limit: payload.limit, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "errorsByType", - }), - "errorsByType query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.errorsByType, tenant, payload) return new ErrorsByTypeResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ fingerprintHash: decodeFingerprintHash(row.fingerprintHash), errorLabel: row.errorLabel, sampleMessage: row.sampleMessage, @@ -363,28 +289,9 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("errorsTimeseries", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.errorsTimeseriesQuery({ - fingerprintHash: payload.fingerprintHash, - services: payload.services, - }), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds: payload.bucketSeconds ?? 3600, - }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "errorsTimeseries", - }), - "errorsTimeseries query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.errorsTimeseries, tenant, payload) return new ErrorsTimeseriesResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ bucket: String(row.bucket), count: Number(row.count), })), @@ -394,24 +301,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("errorsSummary", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.errorsSummaryQuery({ - rootOnly: payload.rootOnly, - services: payload.services, - deploymentEnvs: payload.deploymentEnvs, - fingerprintHashes: payload.fingerprintHashes, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const row = yield* mapExecError( - warehouse - .compiledQueryFirst(tenant, compiled, { - profile: "aggregation", - context: "errorsSummary", - }) - .pipe(Effect.map(Option.getOrNull)), - "errorsSummary query failed", - ) + const row = yield* runQueryFirst(Queries.errorsSummary, tenant, payload) return new ErrorsSummaryResponse({ data: row ? { @@ -428,25 +318,9 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("errorDetailTraces", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.errorDetailTracesQuery({ - fingerprintHash: payload.fingerprintHash, - rootOnly: payload.rootOnly, - services: payload.services, - limit: payload.limit, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "list", - context: "errorDetailTraces", - }), - "errorDetailTraces query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.errorDetailTraces, tenant, payload) return new ErrorDetailTracesResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ traceId: decodeTraceId(row.traceId), startTime: String(row.startTime), durationMicros: Number(row.durationMicros), @@ -461,21 +335,9 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("errorRateByService", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile(CH.errorRateByServiceQuery(), { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - }) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "errorRateByService", - }), - "errorRateByService query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.errorRateByService, tenant, payload) return new ErrorRateByServiceResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ serviceName: decodeServiceName(row.serviceName), totalLogs: Number(row.totalLogs), errorLogs: Number(row.errorLogs), @@ -487,52 +349,14 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("serviceOverview", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.serviceOverviewQuery({ - environments: payload.environments, - namespaces: payload.namespaces, - commitShas: payload.commitShas, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* queryEngine.cachedDirect( - tenant, - "serviceOverview", - payload, - mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceOverview", - }), - "serviceOverview query failed", - ), - // v2: rows gained per-commit `firstSeen`; the version bump keeps - // pre-upgrade cached rows (missing the field) from being served. - makeDirectRouteCachePolicy({ ttlSeconds: 15, version: 2 }), - ) + const rows = yield* runQuery(Queries.serviceOverview, tenant, payload) return new ServiceOverviewResponse({ data: rows }) }), ) .handle("serviceHealthSnapshot", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.serviceHealthSnapshotQuery({ environments: payload.environments }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - { rowSchema: CH.serviceHealthSnapshotRowSchema }, - ) - const rows = yield* queryEngine.cachedDirect( - tenant, - "serviceHealthSnapshot", - payload, - mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceHealthSnapshot", - }), - "serviceHealthSnapshot query failed", - ), - ) + const rows = yield* runQuery(Queries.serviceHealthSnapshot, tenant, payload) return new ServiceHealthSnapshotResponse({ data: rows.map((row) => ({ serviceName: decodeServiceName(row.serviceName), @@ -547,31 +371,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("serviceHealthBaseline", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.serviceHealthBaselineQuery({ - environments: payload.environments, - namespaces: payload.namespaces, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* queryEngine.cachedDirect( - tenant, - "serviceHealthBaseline", - payload, - mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceHealthBaseline", - }), - "serviceHealthBaseline query failed", - ), - // The payload's start/end are floored to the hour upstream - // (`floorToHour`) and this is a trailing 7-day baseline that - // changes at most hourly, so the cache key already rotates once - // an hour — a 1h TTL yields ≤1 recompute/hour per (org, env, ns) - // instead of every 15s for an ~900ms query. - 3600, - ) + const rows = yield* runQuery(Queries.serviceHealthBaseline, tenant, payload) return new ServiceHealthBaselineResponse({ data: rows.map((row) => ({ serviceName: decodeServiceName(String(row.serviceName ?? "")), @@ -586,33 +386,9 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("serviceApdex", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.serviceApdexTimeseriesQuery({ - serviceName: payload.serviceName, - apdexThresholdMs: payload.apdexThresholdMs, - }), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds: payload.bucketSeconds ?? 60, - }, - ) - const rows = yield* queryEngine.cachedDirect( - tenant, - "serviceApdex", - payload, - mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceApdex", - }), - "serviceApdex query failed", - ), - ) - const typedRows = rows + const rows = yield* runQuery(Queries.serviceApdex, tenant, payload) return new ServiceApdexResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ bucket: String(row.bucket), totalCount: Number(row.totalCount), satisfiedCount: Number(row.satisfiedCount), @@ -625,82 +401,28 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("serviceDependencies", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.serviceDependenciesSQL( - { deploymentEnv: payload.deploymentEnv }, - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceDependencies", - }), - "serviceDependencies query failed", - ) + const rows = yield* runQuery(Queries.serviceDependencies, tenant, payload) return new ServiceDependenciesResponse({ data: rows.map((row) => ({ ...row })) }) }), ) .handle("serviceDependenciesForService", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.serviceDependenciesForServiceQuery({ - serviceName: payload.serviceName, - deploymentEnv: payload.deploymentEnv, - }), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceDependenciesForService", - }), - "serviceDependenciesForService query failed", - ) + const rows = yield* runQuery(Queries.serviceDependenciesForService, tenant, payload) return new ServiceDependenciesResponse({ data: rows }) }), ) .handle("serviceDbEdges", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.serviceDbEdgesSQL( - { deploymentEnv: payload.deploymentEnv }, - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceDbEdges", - }), - "serviceDbEdges query failed", - ) + const rows = yield* runQuery(Queries.serviceDbEdges, tenant, payload) return new ServiceDbEdgesResponse({ data: rows.map((row) => ({ ...row })) }) }), ) .handle("serviceDbEdgesForService", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.serviceDbEdgesForServiceQuery({ - serviceName: payload.serviceName, - deploymentEnv: payload.deploymentEnv, - }), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceDbEdgesForService", - }), - "serviceDbEdgesForService query failed", - ) + const rows = yield* runQuery(Queries.serviceDbEdgesForService, tenant, payload) return new ServiceDbEdgesResponse({ data: rows }) }), ) @@ -998,23 +720,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", Effect.gen(function* () { const tenant = yield* CurrentTenant.Context const filters = toCloudflareFilters(payload) - const compiled = CH.compile( - Integrations.cloudflareZoneTimeseriesSQL(filters), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds: payload.bucketSeconds, - }, - { rowSchema: Integrations.cloudflareZoneTimeseriesRowSchema }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "cloudflareInfraZoneTimeseries", - }), - "cloudflareInfraZoneTimeseries query failed", - ) + const rows = yield* runQuery(Queries.cloudflareInfraZoneTimeseries, tenant, payload) return new CloudflareInfraZoneTimeseriesResponse({ data: rows.map((row) => ({ ...row })), ignoredFilters: Integrations.cloudflareIgnoredFiltersFor(filters, [ @@ -1524,23 +1230,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("cloudflareInfraWorkerTimeseries", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - Integrations.cloudflareWorkerTimeseriesSQL(), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds: payload.bucketSeconds, - }, - { rowSchema: Integrations.cloudflareWorkerTimeseriesRowSchema }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "cloudflareInfraWorkerTimeseries", - }), - "cloudflareInfraWorkerTimeseries query failed", - ) + const rows = yield* runQuery(Queries.cloudflareInfraWorkerTimeseries, tenant, payload) return new CloudflareInfraWorkerTimeseriesResponse({ data: rows.map((row) => ({ ...row })), }) @@ -1549,21 +1239,6 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("serviceDetailOverview", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - - const releasesCompiled = CH.compile( - CH.serviceReleasesTimelineQuery({ serviceName: payload.serviceName }), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds: payload.releasesBucketSeconds ?? 300, - }, - ) - const environmentsCompiled = CH.compile( - CH.serviceEnvironmentsQuery({ serviceName: payload.serviceName }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - // One Worker invocation for the whole Overview tab: per-org config // resolves once (the first sub-query warms the in-isolate memo) and // the three queries run concurrently, replacing three separate @@ -1573,33 +1248,15 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", const [timeseries, releaseRows, environmentRows] = yield* Effect.all( [ queryEngine.execute(tenant, payload.timeseries), - mapExecError( - warehouse.compiledQuery(tenant, releasesCompiled, { - profile: "list", - context: "serviceReleases", - }), - "serviceReleases query failed", - ), - queryEngine.cachedDirect( - tenant, - "serviceEnvironments", - { - serviceName: payload.serviceName, - startTime: payload.startTime, - endTime: payload.endTime, - }, - mapExecError( - warehouse.compiledQuery(tenant, environmentsCompiled, { - profile: "discovery", - context: "serviceEnvironments", - }), - "serviceEnvironments query failed", - ), - ), + runQuery(Queries.serviceReleases, tenant, payload), + runQuery(Queries.serviceEnvironments, tenant, { + serviceName: payload.serviceName, + startTime: payload.startTime, + endTime: payload.endTime, + }), ], { concurrency: 3 }, ) - return new ServiceDetailOverviewResponse({ timeseries, releases: releaseRows.map((row) => ({ @@ -1617,56 +1274,17 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("serviceDependenciesBundle", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - - const dependenciesCompiled = CH.compile( - CH.serviceDependenciesForServiceQuery({ - serviceName: payload.serviceName, - deploymentEnv: payload.deploymentEnv, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const dbEdgesCompiled = CH.compile( - CH.serviceDbEdgesForServiceQuery({ - serviceName: payload.serviceName, - deploymentEnv: payload.deploymentEnv, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const externalEdgesCompiled = CH.serviceExternalEdgesSQL( - { deploymentEnv: payload.deploymentEnv, serviceName: payload.serviceName }, - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - // Dependencies tab in one Worker invocation: the three service-map // edge queries run concurrently and share a single config // resolution, replacing three independent round-trips. const [dependencyRows, dbEdgeRows, externalEdgeRows] = yield* Effect.all( [ - mapExecError( - warehouse.compiledQuery(tenant, dependenciesCompiled, { - profile: "aggregation", - context: "serviceDependenciesForService", - }), - "serviceDependenciesForService query failed", - ), - mapExecError( - warehouse.compiledQuery(tenant, dbEdgesCompiled, { - profile: "aggregation", - context: "serviceDbEdgesForService", - }), - "serviceDbEdgesForService query failed", - ), - mapExecError( - warehouse.compiledQuery(tenant, externalEdgesCompiled, { - profile: "aggregation", - context: "serviceExternalEdges", - }), - "serviceExternalEdges query failed", - ), + runQuery(Queries.serviceDependenciesForService, tenant, payload), + runQuery(Queries.serviceDbEdgesForService, tenant, payload), + runQuery(Queries.serviceExternalEdges, tenant, payload), ], { concurrency: 3 }, ) - return new ServiceDependenciesBundleResponse({ dependencies: dependencyRows.map((row) => ({ ...row })), dbEdges: dbEdgeRows.map((row) => ({ ...row })), @@ -1768,20 +1386,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("serviceExternalEdges", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.serviceExternalEdgesSQL( - { - deploymentEnv: payload.deploymentEnv, - serviceName: payload.serviceName, - }, - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceExternalEdges", - }), - "serviceExternalEdges query failed", - ) + const rows = yield* runQuery(Queries.serviceExternalEdges, tenant, payload) return new ServiceExternalEdgesResponse({ data: rows.map((row) => ({ ...row })) }) }), ) @@ -1844,17 +1449,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", if (payload.services.length === 0) { return new ServiceWorkloadsResponse({ data: [] }) } - const compiled = CH.serviceWorkloadsSQL( - { services: payload.services }, - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceWorkloads", - }), - "serviceWorkloads query failed", - ) + const rows = yield* runQuery(Queries.serviceWorkloads, tenant, payload) return new ServiceWorkloadsResponse({ data: rows.map((row) => ({ serviceName: decodeServiceName(String(row.serviceName ?? "")), @@ -1878,37 +1473,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("serviceUsage", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const prevStart = payload.previousStartTime - const prevEnd = payload.previousEndTime - const compiled = - prevStart != null && prevEnd != null - ? CH.compile(CH.serviceUsageWithPreviousQuery({ serviceName: payload.service }), { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - previousStartTime: prevStart, - previousEndTime: prevEnd, - }) - : CH.compile(CH.serviceUsageQuery({ serviceName: payload.service }), { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - }) - const rows = yield* queryEngine.cachedDirect( - tenant, - "serviceUsage", - payload, - mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceUsage", - }), - "serviceUsage query failed", - ), - // Usage totals (GB / session counts) tolerate a minute of - // staleness; a 60s TTL cuts repeat-load recomputes ~4× vs 15s. - 60, - ) + const rows = yield* runQuery(Queries.serviceUsage, tenant, payload) return new ServiceUsageResponse({ data: rows }) }), ) @@ -2076,116 +1641,30 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("listLogs", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.logsListQuery({ - serviceName: payload.service, - severity: payload.severity, - minSeverity: payload.minSeverity, - traceId: payload.traceId, - spanId: payload.spanId, - cursor: payload.cursor, - search: payload.search, - environments: payload.deploymentEnv ? [payload.deploymentEnv] : undefined, - namespaces: payload.namespace ? [payload.namespace] : undefined, - matchModes: Match.value([ - payload.deploymentEnvMatchMode, - payload.namespaceMatchMode, - ] as const).pipe( - Match.when([undefined, undefined], () => undefined), - Match.orElse(([deploymentEnv, serviceNamespace]) => ({ - deploymentEnv, - serviceNamespace, - })), - ), - limit: payload.limit, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* queryEngine.cachedDirect( - tenant, - "listLogs", - payload, - mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "list", - context: "listLogs", - // Body search reads the wide Body column for the ILIKE - // filter — cap the read block size (see - // WarehouseQuerySettings.maxBlockSize). - settings: payload.search ? LOGS_BODY_SEARCH_SETTINGS : undefined, - }), - "listLogs query failed", - ), - ) + const rows = yield* runQuery(Queries.listLogs, tenant, payload) return new ListLogsResponse({ data: rows }) }), ) .handle("getLog", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - // Bound the scan to a ±1h window around the requested log so - // ClickHouse can prune partitions instead of reading every - // retained daily partition for an exact-timestamp match. - const { startTime, endTime } = partitionWindowAround(payload.timestamp) - const compiled = CH.compile( - CH.getLogByKeyQuery({ - serviceName: payload.serviceName, - traceId: payload.traceId, - spanId: payload.spanId, - }), - { orgId: tenant.orgId, startTime, endTime, timestamp: payload.timestamp }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "list", - context: "getLog", - }), - "getLog query failed", - ) + const rows = yield* runQuery(Queries.getLog, tenant, payload) return new GetLogResponse({ data: rows }) }), ) .handle("listMetrics", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.listMetricsQuery({ - serviceName: payload.service, - metricType: payload.metricType, - search: payload.search, - limit: payload.limit, - offset: payload.offset, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "discovery", - context: "listMetrics", - }), - "listMetrics query failed", - ) + const rows = yield* runQuery(Queries.listMetrics, tenant, payload) return new ListMetricsResponse({ data: rows }) }), ) .handle("metricsSummary", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile(CH.metricsSummaryQuery({ serviceName: payload.service }), { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - }) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "discovery", - context: "metricsSummary", - }), - "metricsSummary query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.metricsSummary, tenant, payload) return new MetricsSummaryResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ metricType: row.metricType, metricCount: Number(row.metricCount), dataPointCount: Number(row.dataPointCount), @@ -2345,21 +1824,9 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("listHosts", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.listHostsQuery({ - search: payload.search, - limit: payload.limit, - offset: payload.offset, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { profile: "list", context: "listHosts" }), - "listHosts query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.listHosts, tenant, payload) return new ListHostsResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ hostName: row.hostName, osType: row.osType, hostArch: row.hostArch, @@ -2376,20 +1843,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("hostDetailSummary", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile(CH.hostDetailSummaryQuery({ hostName: payload.hostName }), { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - }) - const row = yield* mapExecError( - warehouse - .compiledQueryFirst(tenant, compiled, { - profile: "aggregation", - context: "hostDetailSummary", - }) - .pipe(Effect.map(Option.getOrNull)), - "hostDetailSummary query failed", - ) + const row = yield* runQueryFirst(Queries.hostDetailSummary, tenant, payload) return new HostDetailSummaryResponse({ data: row ? { @@ -2412,23 +1866,9 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("fleetUtilizationTimeseries", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const bucketSeconds = payload.bucketSeconds ?? 300 - const compiled = CH.compile(CH.fleetUtilizationTimeseriesQuery(), { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds, - }) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "fleetUtilizationTimeseries", - }), - "fleetUtilizationTimeseries query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.fleetUtilizationTimeseries, tenant, payload) return new FleetUtilizationTimeseriesResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ bucket: String(row.bucket), avgCpu: Number(row.avgCpu) || 0, avgMemory: Number(row.avgMemory) || 0, @@ -2659,27 +2099,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("podsSummary", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - // Scope only — no row filters. The browse band is what tells you how - // much of the fleet your filters just hid, so narrowing it too would - // defeat the point. - const compiled = CH.compile( - CH.listPodsSummaryQuery({ - namespaces: payload.namespaces, - clusters: payload.clusters, - environments: payload.environments, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - { rowSchema: CH.ListPodsSummaryOutputSchema }, - ) - const row = yield* mapExecError( - warehouse - .compiledQueryFirst(tenant, compiled, { - profile: "aggregation", - context: "podsSummary", - }) - .pipe(Effect.map(Option.getOrNull)), - "podsSummary query failed", - ) + const row = yield* runQueryFirst(Queries.podsSummary, tenant, payload) return new PodsSummaryResponse({ totalPods: Number(row?.totalPods) || 0, saturatedPods: Number(row?.saturatedPods) || 0, @@ -2692,19 +2112,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("podDetailSummary", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.podDetailSummaryQuery({ podName: payload.podName, namespace: payload.namespace }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const row = yield* mapExecError( - warehouse - .compiledQueryFirst(tenant, compiled, { - profile: "aggregation", - context: "podDetailSummary", - }) - .pipe(Effect.map(Option.getOrNull)), - "podDetailSummary query failed", - ) + const row = yield* runQueryFirst(Queries.podDetailSummary, tenant, payload) return new PodDetailSummaryResponse({ data: row ? { @@ -2733,58 +2141,10 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("podInfraTimeseries", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const bucketSeconds = payload.bucketSeconds ?? 60 - - const spec = (() => { - switch (payload.metric) { - case "cpu_usage": - return { metricName: "k8s.pod.cpu.usage", unit: "cores" as const } - case "cpu_limit": - return { - metricName: "k8s.pod.cpu_limit_utilization", - unit: "percent" as const, - } - case "cpu_request": - return { - metricName: "k8s.pod.cpu_request_utilization", - unit: "percent" as const, - } - case "memory_limit": - return { - metricName: "k8s.pod.memory_limit_utilization", - unit: "percent" as const, - } - case "memory_request": - return { - metricName: "k8s.pod.memory_request_utilization", - unit: "percent" as const, - } - } - })() - - const compiled = CH.compile( - CH.podGaugeTimeseriesQuery({ - podName: payload.podName, - namespace: payload.namespace, - metricName: spec.metricName, - }), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds, - }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "podInfraTimeseries", - }), - "podInfraTimeseries query failed", - ) - const typedRows = rows + const spec = podMetricSpec(payload.metric) + const rows = yield* runQuery(Queries.podInfraTimeseries, tenant, payload) return new PodInfraTimeseriesResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ bucket: String(row.bucket), attributeValue: String(row.attributeValue ?? ""), value: Number(row.avgValue) || 0, @@ -2796,24 +2156,9 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("listNodes", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.listNodesQuery({ - search: payload.search, - nodeNames: payload.nodeNames, - clusters: payload.clusters, - environments: payload.environments, - limit: payload.limit, - offset: payload.offset, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { profile: "list", context: "listNodes" }), - "listNodes query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.listNodes, tenant, payload) return new ListNodesResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ nodeName: row.nodeName, nodeUid: row.nodeUid, clusterName: row.clusterName, @@ -2829,20 +2174,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("nodeDetailSummary", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile(CH.nodeDetailSummaryQuery({ nodeName: payload.nodeName }), { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - }) - const row = yield* mapExecError( - warehouse - .compiledQueryFirst(tenant, compiled, { - profile: "aggregation", - context: "nodeDetailSummary", - }) - .pipe(Effect.map(Option.getOrNull)), - "nodeDetailSummary query failed", - ) + const row = yield* runQueryFirst(Queries.nodeDetailSummary, tenant, payload) return new NodeDetailSummaryResponse({ data: row ? { @@ -2862,39 +2194,10 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("nodeInfraTimeseries", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const bucketSeconds = payload.bucketSeconds ?? 60 - - const spec = (() => { - switch (payload.metric) { - case "cpu_usage": - return { metricName: "k8s.node.cpu.usage", unit: "cores" as const } - case "uptime": - return { metricName: "k8s.node.uptime", unit: "seconds" as const } - } - })() - - const compiled = CH.compile( - CH.nodeGaugeTimeseriesQuery({ - nodeName: payload.nodeName, - metricName: spec.metricName, - }), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds, - }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "nodeInfraTimeseries", - }), - "nodeInfraTimeseries query failed", - ) - const typedRows = rows + const spec = nodeMetricSpec(payload.metric) + const rows = yield* runQuery(Queries.nodeInfraTimeseries, tenant, payload) return new NodeInfraTimeseriesResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ bucket: String(row.bucket), attributeValue: String(row.attributeValue ?? ""), value: Number(row.avgValue) || 0, @@ -2906,30 +2209,9 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("listWorkloads", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.listWorkloadsQuery({ - kind: payload.kind, - search: payload.search, - workloadNames: payload.workloadNames, - namespaces: payload.namespaces, - clusters: payload.clusters, - environments: payload.environments, - computeTypes: payload.computeTypes, - limit: payload.limit, - offset: payload.offset, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "list", - context: "listWorkloads", - }), - "listWorkloads query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.listWorkloads, tenant, payload) return new ListWorkloadsResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ workloadName: row.workloadName, namespace: row.namespace, clusterName: row.clusterName, @@ -2946,23 +2228,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("workloadDetailSummary", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.workloadDetailSummaryQuery({ - kind: payload.kind, - workloadName: payload.workloadName, - namespace: payload.namespace, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const row = yield* mapExecError( - warehouse - .compiledQueryFirst(tenant, compiled, { - profile: "aggregation", - context: "workloadDetailSummary", - }) - .pipe(Effect.map(Option.getOrNull)), - "workloadDetailSummary query failed", - ) + const row = yield* runQueryFirst(Queries.workloadDetailSummary, tenant, payload) return new WorkloadDetailSummaryResponse({ data: row ? { @@ -2983,50 +2249,10 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("workloadInfraTimeseries", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const bucketSeconds = payload.bucketSeconds ?? 60 - - const spec = (() => { - switch (payload.metric) { - case "cpu_usage": - return { metricName: "k8s.pod.cpu.usage", unit: "cores" as const } - case "cpu_limit": - return { - metricName: "k8s.pod.cpu_limit_utilization", - unit: "percent" as const, - } - case "memory_limit": - return { - metricName: "k8s.pod.memory_limit_utilization", - unit: "percent" as const, - } - } - })() - - const compiled = CH.compile( - CH.workloadGaugeTimeseriesQuery({ - kind: payload.kind, - workloadName: payload.workloadName, - namespace: payload.namespace, - metricName: spec.metricName, - groupByPod: payload.groupByPod, - }), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds, - }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "workloadInfraTimeseries", - }), - "workloadInfraTimeseries query failed", - ) - const typedRows = rows + const spec = workloadMetricSpec(payload.metric) + const rows = yield* runQuery(Queries.workloadInfraTimeseries, tenant, payload) return new WorkloadInfraTimeseriesResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ bucket: String(row.bucket), attributeValue: String(row.attributeValue ?? ""), value: Number(row.avgValue) || 0, diff --git a/packages/query-engine/package.json b/packages/query-engine/package.json index 9e3cab148..a3b2d9e03 100644 --- a/packages/query-engine/package.json +++ b/packages/query-engine/package.json @@ -14,6 +14,7 @@ "./caching": "./src/caching/index.ts", "./execution": "./src/execution/index.ts", "./runtime": "./src/runtime/index.ts", + "./registry": "./src/registry/index.ts", "./observability": "./src/observability/index.ts", "./local": "./src/local.ts", "./sql-catalog": "./src/sql-catalog.ts", diff --git a/packages/query-engine/src/registry/index.ts b/packages/query-engine/src/registry/index.ts new file mode 100644 index 000000000..74a49658a --- /dev/null +++ b/packages/query-engine/src/registry/index.ts @@ -0,0 +1,9 @@ +/** + * Declarative warehouse query registry. + * + * Kept out of the root barrel deliberately: entries pull in `./runtime` for the + * cache-policy types, and the root barrel stays driver-free so web and cli can + * import it. Only `apps/api` should reach for this subpath. + */ +export { defineQuery, type QueryDef } from "./query-def" +export * as Queries from "./queries" diff --git a/packages/query-engine/src/registry/queries.ts b/packages/query-engine/src/registry/queries.ts new file mode 100644 index 000000000..ae46161c3 --- /dev/null +++ b/packages/query-engine/src/registry/queries.ts @@ -0,0 +1,572 @@ +import type { + ServiceDbEdgesRequest, + ServiceDependenciesRequest, + ServiceWorkloadsRequest, + ServiceUsageRequest, + ErrorDetailTracesRequest, + ErrorRateByServiceRequest, + ErrorsByTypeRequest, + ErrorsSummaryRequest, + ErrorsTimeseriesRequest, + HostDetailSummaryRequest, + ListHostsRequest, + ListLogsRequest, + ListMetricsRequest, + ListNodesRequest, + ListWorkloadsRequest, + MetricsSummaryRequest, + NodeDetailSummaryRequest, + PodDetailSummaryRequest, + PodsSummaryRequest, + ServiceApdexRequest, + ServiceDbEdgesForServiceRequest, + ServiceDependenciesForServiceRequest, + ServiceHealthBaselineRequest, + ServiceHealthSnapshotRequest, + ServiceOverviewRequest, + WorkloadDetailSummaryRequest, +} from "@maple/domain/http" +import { Match } from "effect" +import * as CH from "../ch" +import { LOGS_BODY_SEARCH_SETTINGS } from "../profiles" +import { makeDirectRouteCachePolicy } from "../runtime/query-engine" +import { defineQuery } from "./query-def" + +/** + * The declarative warehouse query registry. + * + * Each entry replaces the profile/context/error-label/cache wiring that used to + * be repeated inline in every handler in `apps/api/src/routes/v1/query-engine.http.ts`. + * Handlers keep their own row-to-response mapping; see `QueryDef` for why + * decoding is deliberately out of scope here. + * + * Migration is incremental and the two surfaces coexist: a handler either takes + * a `QueryDef` through `runQuery` or keeps its inline wiring. Nothing breaks + * while entries are added. + * + * Cache values below are carried over EXACTLY as the handlers had them, so this + * pilot changes no caching behaviour — `cache: undefined` here means the handler + * was uncached before, not that being uncached is correct. Turning any of those + * on is a separate, separately-reviewed change with a justified TTL. + */ + +export const errorsByType = defineQuery({ + id: "errorsByType", + profile: "aggregation", + // Was uncached inline. Preserved as-is: changing it belongs in its own commit. + cache: undefined, + compile: (payload: ErrorsByTypeRequest, orgId: string) => + CH.compile( + CH.errorsByTypeQuery({ + rootOnly: payload.rootOnly, + services: payload.services, + deploymentEnvs: payload.deploymentEnvs, + fingerprintHashes: payload.fingerprintHashes, + limit: payload.limit, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const errorsTimeseries = defineQuery({ + id: "errorsTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: ErrorsTimeseriesRequest, orgId: string) => + CH.compile( + CH.errorsTimeseriesQuery({ + fingerprintHash: payload.fingerprintHash, + services: payload.services, + }), + { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + // Matches the handler's previous inline default. The builder needs a + // bucket width and the request treats it as optional. + bucketSeconds: payload.bucketSeconds ?? 3600, + }, + ), +}) + +/** Single-row: the handler reads this through `runQueryFirst`. */ +export const errorsSummary = defineQuery({ + id: "errorsSummary", + profile: "aggregation", + cache: undefined, + compile: (payload: ErrorsSummaryRequest, orgId: string) => + CH.compile( + CH.errorsSummaryQuery({ + rootOnly: payload.rootOnly, + services: payload.services, + deploymentEnvs: payload.deploymentEnvs, + fingerprintHashes: payload.fingerprintHashes, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const errorRateByService = defineQuery({ + id: "errorRateByService", + profile: "aggregation", + cache: undefined, + // The builder takes no options — this query is scoped entirely by org and + // time range. The payload still carries the range. + compile: (payload: ErrorRateByServiceRequest, orgId: string) => + CH.compile(CH.errorRateByServiceQuery(), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + }), +}) + +export const serviceOverview = defineQuery({ + id: "serviceOverview", + profile: "aggregation", + // v2: rows gained per-commit `firstSeen`; the version bump keeps pre-upgrade + // cached rows (missing the field) from being served. Carried over verbatim + // from the handler — do not renumber without the same reasoning. + cache: makeDirectRouteCachePolicy({ ttlSeconds: 15, version: 2 }), + compile: (payload: ServiceOverviewRequest, orgId: string) => + CH.compile( + CH.serviceOverviewQuery({ + environments: payload.environments, + namespaces: payload.namespaces, + commitShas: payload.commitShas, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const errorDetailTraces = defineQuery({ + id: "errorDetailTraces", + profile: "list", + cache: undefined, + compile: (payload: ErrorDetailTracesRequest, orgId: string) => + CH.compile( + CH.errorDetailTracesQuery({ + fingerprintHash: payload.fingerprintHash, + rootOnly: payload.rootOnly, + services: payload.services, + limit: payload.limit, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const serviceHealthSnapshot = defineQuery({ + id: "serviceHealthSnapshot", + profile: "aggregation", + cache: 15, + compile: (payload: ServiceHealthSnapshotRequest, orgId: string) => + CH.compile( + CH.serviceHealthSnapshotQuery({ environments: payload.environments }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + { rowSchema: CH.serviceHealthSnapshotRowSchema }, + ), +}) + +export const serviceHealthBaseline = defineQuery({ + id: "serviceHealthBaseline", + profile: "aggregation", + cache: 3600, + compile: (payload: ServiceHealthBaselineRequest, orgId: string) => + CH.compile( + CH.serviceHealthBaselineQuery({ + environments: payload.environments, + namespaces: payload.namespaces, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const serviceApdex = defineQuery({ + id: "serviceApdex", + profile: "aggregation", + cache: 15, + compile: (payload: ServiceApdexRequest, orgId: string) => + CH.compile( + CH.serviceApdexTimeseriesQuery({ + serviceName: payload.serviceName, + apdexThresholdMs: payload.apdexThresholdMs, + }), + { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + bucketSeconds: payload.bucketSeconds ?? 60, + }, + ), +}) + +export const serviceDependenciesForService = defineQuery({ + id: "serviceDependenciesForService", + profile: "aggregation", + cache: undefined, + compile: (payload: ServiceDependenciesForServiceRequest, orgId: string) => + CH.compile( + CH.serviceDependenciesForServiceQuery({ + serviceName: payload.serviceName, + deploymentEnv: payload.deploymentEnv, + }), + { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + }, + ), +}) + +export const serviceDbEdgesForService = defineQuery({ + id: "serviceDbEdgesForService", + profile: "aggregation", + cache: undefined, + compile: (payload: ServiceDbEdgesForServiceRequest, orgId: string) => + CH.compile( + CH.serviceDbEdgesForServiceQuery({ + serviceName: payload.serviceName, + deploymentEnv: payload.deploymentEnv, + }), + { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + }, + ), +}) + +export const listLogs = defineQuery({ + id: "listLogs", + profile: "list", + settings: (payload) => (payload.search ? LOGS_BODY_SEARCH_SETTINGS : undefined), + cache: 15, + compile: (payload: ListLogsRequest, orgId: string) => + CH.compile( + CH.logsListQuery({ + serviceName: payload.service, + severity: payload.severity, + minSeverity: payload.minSeverity, + traceId: payload.traceId, + spanId: payload.spanId, + cursor: payload.cursor, + search: payload.search, + environments: payload.deploymentEnv ? [payload.deploymentEnv] : undefined, + namespaces: payload.namespace ? [payload.namespace] : undefined, + matchModes: Match.value([ + payload.deploymentEnvMatchMode, + payload.namespaceMatchMode, + ] as const).pipe( + Match.when([undefined, undefined], () => undefined), + Match.orElse(([deploymentEnv, serviceNamespace]) => ({ + deploymentEnv, + serviceNamespace, + })), + ), + limit: payload.limit, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const listMetrics = defineQuery({ + id: "listMetrics", + profile: "discovery", + cache: undefined, + compile: (payload: ListMetricsRequest, orgId: string) => + CH.compile( + CH.listMetricsQuery({ + serviceName: payload.service, + metricType: payload.metricType, + search: payload.search, + limit: payload.limit, + offset: payload.offset, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const metricsSummary = defineQuery({ + id: "metricsSummary", + profile: "discovery", + cache: undefined, + compile: (payload: MetricsSummaryRequest, orgId: string) => + CH.compile(CH.metricsSummaryQuery({ serviceName: payload.service }), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + }), +}) + +export const listHosts = defineQuery({ + id: "listHosts", + profile: "list", + cache: undefined, + compile: (payload: ListHostsRequest, orgId: string) => + CH.compile( + CH.listHostsQuery({ + search: payload.search, + limit: payload.limit, + offset: payload.offset, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const hostDetailSummary = defineQuery({ + id: "hostDetailSummary", + profile: "aggregation", + cache: undefined, + compile: (payload: HostDetailSummaryRequest, orgId: string) => + CH.compile(CH.hostDetailSummaryQuery({ hostName: payload.hostName }), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + }), +}) + +export const podsSummary = defineQuery({ + id: "podsSummary", + profile: "aggregation", + cache: undefined, + compile: (payload: PodsSummaryRequest, orgId: string) => + CH.compile( + CH.listPodsSummaryQuery({ + namespaces: payload.namespaces, + clusters: payload.clusters, + environments: payload.environments, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + { rowSchema: CH.ListPodsSummaryOutputSchema }, + ), +}) + +export const podDetailSummary = defineQuery({ + id: "podDetailSummary", + profile: "aggregation", + cache: undefined, + compile: (payload: PodDetailSummaryRequest, orgId: string) => + CH.compile(CH.podDetailSummaryQuery({ podName: payload.podName, namespace: payload.namespace }), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + }), +}) + +export const listNodes = defineQuery({ + id: "listNodes", + profile: "list", + cache: undefined, + compile: (payload: ListNodesRequest, orgId: string) => + CH.compile( + CH.listNodesQuery({ + search: payload.search, + nodeNames: payload.nodeNames, + clusters: payload.clusters, + environments: payload.environments, + limit: payload.limit, + offset: payload.offset, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const nodeDetailSummary = defineQuery({ + id: "nodeDetailSummary", + profile: "aggregation", + cache: undefined, + compile: (payload: NodeDetailSummaryRequest, orgId: string) => + CH.compile(CH.nodeDetailSummaryQuery({ nodeName: payload.nodeName }), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + }), +}) + +export const listWorkloads = defineQuery({ + id: "listWorkloads", + profile: "list", + cache: undefined, + compile: (payload: ListWorkloadsRequest, orgId: string) => + CH.compile( + CH.listWorkloadsQuery({ + kind: payload.kind, + search: payload.search, + workloadNames: payload.workloadNames, + namespaces: payload.namespaces, + clusters: payload.clusters, + environments: payload.environments, + computeTypes: payload.computeTypes, + limit: payload.limit, + offset: payload.offset, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const workloadDetailSummary = defineQuery({ + id: "workloadDetailSummary", + profile: "aggregation", + cache: undefined, + compile: (payload: WorkloadDetailSummaryRequest, orgId: string) => + CH.compile( + CH.workloadDetailSummaryQuery({ + kind: payload.kind, + workloadName: payload.workloadName, + namespace: payload.namespace, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +// --- Sub-queries of composite (bundle) handlers --------------------------- +// +// Bundle endpoints run several queries in one Worker invocation so per-org +// config resolves once and the browser makes one round-trip instead of three. +// Each sub-query keeps its own id, because that id is both its span context and +// its cache-key prefix -- collapsing them under the bundle's name would merge +// unrelated cache entries. +// +// Their payload types are the MINIMAL input each needs rather than the bundle's +// full payload. That is load-bearing: `runQuery` keys the cache on whatever +// payload it is handed, so typing these narrowly reproduces the exact key the +// hand-written `cachedDirect` calls used. + +/** Release markers for the service detail overview chart. Uncached, mirroring the standalone path. */ +export const serviceReleases = defineQuery({ + id: "serviceReleases", + profile: "list", + cache: undefined, + compile: ( + payload: { + readonly serviceName: string + readonly startTime: string + readonly endTime: string + readonly releasesBucketSeconds?: number | undefined + }, + orgId: string, + ) => + CH.compile(CH.serviceReleasesTimelineQuery({ serviceName: payload.serviceName }), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + bucketSeconds: payload.releasesBucketSeconds ?? 300, + }), +}) + +/** Environments a service reported in the window. Edge-cached on a service-scoped key. */ +export const serviceEnvironments = defineQuery({ + id: "serviceEnvironments", + profile: "discovery", + cache: 15, + compile: ( + payload: { readonly serviceName: string; readonly startTime: string; readonly endTime: string }, + orgId: string, + ) => + CH.compile(CH.serviceEnvironmentsQuery({ serviceName: payload.serviceName }), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + }), +}) + +/** + * External (non-service) edges for the dependencies tab. + * + * Built by `serviceExternalEdgesSQL`, which returns a CompiledQuery directly + * rather than going through `CH.compile`. + */ +export const serviceExternalEdges = defineQuery({ + id: "serviceExternalEdges", + profile: "aggregation", + cache: undefined, + compile: ( + payload: { + readonly serviceName: string + readonly deploymentEnv?: string | undefined + readonly startTime: string + readonly endTime: string + }, + orgId: string, + ) => + CH.serviceExternalEdgesSQL( + { deploymentEnv: payload.deploymentEnv, serviceName: payload.serviceName }, + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +/** + * Service usage totals. + * + * Two builders behind one id: with both previous-window bounds the query + * returns current-vs-previous in a single scan, otherwise just the current + * window. The branch lives in `compile` so the id, profile and TTL stay one + * decision. + */ +export const serviceUsage = defineQuery({ + id: "serviceUsage", + profile: "aggregation", + // Usage totals (GB / session counts) tolerate a minute of staleness; a 60s + // TTL cuts repeat-load recomputes ~4x vs 15s. + cache: 60, + compile: (payload: ServiceUsageRequest, orgId: string) => { + const prevStart = payload.previousStartTime + const prevEnd = payload.previousEndTime + return prevStart != null && prevEnd != null + ? CH.compile(CH.serviceUsageWithPreviousQuery({ serviceName: payload.service }), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + previousStartTime: prevStart, + previousEndTime: prevEnd, + }) + : CH.compile(CH.serviceUsageQuery({ serviceName: payload.service }), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + }) + }, +}) + +// --- Service-map edge queries --------------------------------------------- +// These use `*SQL(opts, params)` builders, which return a CompiledQuery +// directly instead of going through `CH.compile`. + +export const serviceDependencies = defineQuery({ + id: "serviceDependencies", + profile: "aggregation", + cache: undefined, + compile: (payload: ServiceDependenciesRequest, orgId: string) => + CH.serviceDependenciesSQL( + { deploymentEnv: payload.deploymentEnv }, + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const serviceDbEdges = defineQuery({ + id: "serviceDbEdges", + profile: "aggregation", + cache: undefined, + compile: (payload: ServiceDbEdgesRequest, orgId: string) => + CH.serviceDbEdgesSQL( + { deploymentEnv: payload.deploymentEnv }, + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +/** + * Workloads backing a set of services. + * + * The caller must skip this entirely for an empty service list — that guard + * stays in the handler because it avoids issuing a query at all, which a def + * cannot express. + */ +export const serviceWorkloads = defineQuery({ + id: "serviceWorkloads", + profile: "aggregation", + cache: undefined, + compile: (payload: ServiceWorkloadsRequest, orgId: string) => + CH.serviceWorkloadsSQL( + { services: payload.services }, + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) diff --git a/packages/query-engine/src/registry/query-def.ts b/packages/query-engine/src/registry/query-def.ts new file mode 100644 index 000000000..c6a5ddedf --- /dev/null +++ b/packages/query-engine/src/registry/query-def.ts @@ -0,0 +1,99 @@ +import type { CompiledQuery } from "@maple-dev/clickhouse-builder" +import type { QueryProfileName, WarehouseQuerySettings } from "../profiles/query-profile" +import type { DirectRouteCachePolicyInput } from "../runtime/query-engine" + +/** + * One warehouse query, declared in one place. + * + * The problem this solves: every handler in `query-engine.http.ts` wires its own + * cost profile, span context, error label and — if the author remembered — + * caching. Caching being opt-in per call site meant silence read as "off": 11 of + * 61 handlers called `cachedDirect`, the other 50 were uncached by omission + * rather than by decision. Nothing in the type system asked. + * + * A `QueryDef` makes those choices data instead of boilerplate, so `runQuery` + * can apply them uniformly and a reviewer can see all of them at once. The + * single most important field is `cache`, which is REQUIRED and nullable rather + * than optional: `cache: undefined` is a decision someone made, a missing + * `cache` is a compile error. + * + * Scope note: this deliberately does NOT own row decoding or response shaping. + * Handlers still map rows into their own response class. Folding hand-coercion + * (`Number(row.count)`, `decodeFingerprintHash(...)`) into a declared + * `rowSchema` is a real improvement but a separate, riskier change — it turns + * ClickHouse's UInt64-as-JSON-string behaviour into a hard failure where it is + * currently a silent `Number("123")`, and every migrated query has to go + * through the DESCRIBE sweep (`bun run ch:up && bun run ch:test`) to prove it. + * Keep the two changes apart so a decode regression is never tangled up with a + * caching change. + */ +export interface QueryDef { + /** + * Stable identifier. Used as the warehouse span's `query.context`, as the + * `cachedDirect` route name (and therefore part of the cache key), and as the + * error label. One id, so those three can never drift apart the way they can + * when each is passed as its own string literal. + * + * Where a pipe name already exists for the same logical query, use it — the + * goal is one vocabulary, not a sixth. + */ + readonly id: string + + /** + * Cost profile, applied as ClickHouse `SETTINGS`. Defaults are NOT inherited + * here: state it, because "which budget does this query get" is exactly the + * kind of decision that goes stale silently when it is implicit. + */ + readonly profile: QueryProfileName + + /** + * Extra settings merged over the profile — e.g. `LOGS_BODY_SEARCH_SETTINGS` + * for log body search, whose `maxBlockSize: 512` is the difference between a + * sub-2s query and an OOM on a large-body org. + * + * May be a function of the payload, because whether a query needs them can + * depend on the request: `listLogs` only wants the body-search settings when + * the caller actually passed a search term. + */ + readonly settings?: WarehouseQuerySettings | ((payload: Payload) => WarehouseQuerySettings | undefined) + + /** + * Edge cache policy, or `undefined` for deliberately uncached. + * + * Required-but-nullable on purpose. This is the field that turns ~50 silent + * opt-outs into ~50 reviewed decisions, and it only works if omitting it + * fails to compile. + * + * Be careful raising a TTL on anything live-tailing: a stale entry there is a + * user-visible correctness bug that presents as "the dashboard is frozen". + * + * May be a function of the payload and the current time. `spanDetail` needs + * that: a trace that finished long ago is immutable and can be cached hard, + * while one still receiving spans must not be — so its TTL is computed from + * the requested end time against now. `nowMs` is supplied by the runner from + * the Effect `Clock` rather than read here, so it stays testable. + */ + readonly cache: + | DirectRouteCachePolicyInput + | undefined + | ((payload: Payload, nowMs: number) => DirectRouteCachePolicyInput | undefined) + + /** + * Build the compiled query from the request payload. + * + * `orgId` is passed separately rather than read off the payload because it + * comes from the authenticated tenant, never from user input — every query + * must filter `OrgId`, and that guarantee should not depend on a client + * sending the right value. + */ + readonly compile: (payload: Payload, orgId: string) => CompiledQuery +} + +/** + * Identity helper that pins `Payload`/`Row` from the `compile` function while + * still checking the object against `QueryDef`. + * + * Without it, annotating an entry as `QueryDef` forces you + * to restate the row type that `CH.compile` already knows, and those two drift. + */ +export const defineQuery = (def: QueryDef): QueryDef => def