feat(query-engine): migrate spanHierarchy and the remaining tail (55/61)

apps/api/src/routes/queries.ts

@@ -3,6 +3,7 @@ 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 {
+	HostInfraTimeseriesRequest,
 	CloudflareInfraZoneFacetsRequest,
 	CloudflareInfraZoneDetailRequest,
 	ServicePlanetScaleStatsRequest,
@@ -24,6 +25,7 @@ import type {
 	WorkloadInfraTimeseriesRequest,
 } from "@maple/domain/http"
 import {
+	hostMetricSpec,
 	nodeMetricSpec,
 	partitionWindowAround,
 	podMetricSpec,
@@ -469,6 +471,49 @@ const cloudflareInfraZoneFacets = defineQuery({
 		}),
 })
 
+// --- hostInfraTimeseries: two query families behind one endpoint ----------
+//
+// Network reads a counter family, everything else a gauge family, so they are
+// separate defs rather than one def with a branch — the row shapes differ and
+// the handler maps them differently. Both keep the id "hostInfraTimeseries",
+// which is what their spans already report; renaming would break continuity of
+// existing telemetry for no gain.
+
+const hostInfraNetworkTimeseries = defineQuery({
+	id: "hostInfraTimeseries",
+	profile: "aggregation",
+	cache: undefined,
+	compile: (payload: HostInfraTimeseriesRequest, orgId: string) =>
+		CH.compile(CH.hostNetworkTimeseriesQuery({ hostName: payload.hostName }), {
+			orgId,
+			startTime: payload.startTime,
+			endTime: payload.endTime,
+			bucketSeconds: payload.bucketSeconds ?? 60,
+		}),
+})
+
+const hostInfraGaugeTimeseries = defineQuery({
+	id: "hostInfraTimeseries",
+	profile: "aggregation",
+	cache: undefined,
+	compile: (payload: HostInfraTimeseriesRequest, orgId: string) => {
+		const spec = hostMetricSpec(payload.metric)
+		return CH.compile(
+			CH.hostGaugeTimeseriesQuery({
+				hostName: payload.hostName,
+				metricName: spec.metricName,
+				groupByAttributeKey: spec.groupByAttributeKey,
+			}),
+			{
+				orgId,
+				startTime: payload.startTime,
+				endTime: payload.endTime,
+				bucketSeconds: payload.bucketSeconds ?? 60,
+			},
+		)
+	},
+})
+
 export const Queries = {
 	...Core,
 
@@ -651,4 +696,6 @@ export const Queries = {
 	planetscaleServiceConnections,
 	planetscaleServiceStorage,
 	cloudflareInfraZoneFacets,
+	hostInfraNetworkTimeseries,
+	hostInfraGaugeTimeseries,
 } as const

apps/api/src/routes/query-helpers.ts

@@ -1,6 +1,7 @@
 import * as Integrations from "@maple/query-engine-integrations"
 import { formatWarehouseDateTime, parseWarehouseDateTime } from "@maple/query-engine"
 import type {
+	HostInfraTimeseriesRequest,
 	NodeInfraTimeseriesRequest,
 	PodInfraTimeseriesRequest,
 	WorkloadInfraTimeseriesRequest,
@@ -111,3 +112,50 @@ export const workloadMetricSpec = (metric: WorkloadInfraTimeseriesRequest["metri
 			}
 	}
 }
+
+/**
+ * Metric name, grouping key, unit and query-family flag for a host metric.
+ *
+ * Shared like the pod/node/workload specs: the registry needs `metricName` and
+ * `isNetwork` to build the query, the handler needs `unit` and
+ * `groupByAttributeKey` for its response.
+ */
+export const hostMetricSpec = (metric: HostInfraTimeseriesRequest["metric"]) => {
+	switch (metric) {
+		case "cpu":
+			return {
+				metricName: "system.cpu.utilization",
+				groupByAttributeKey: "state",
+				unit: "percent" as const,
+				isNetwork: false,
+			}
+		case "memory":
+			return {
+				metricName: "system.memory.utilization",
+				groupByAttributeKey: "state",
+				unit: "percent" as const,
+				isNetwork: false,
+			}
+		case "filesystem":
+			return {
+				metricName: "system.filesystem.utilization",
+				groupByAttributeKey: "mountpoint",
+				unit: "percent" as const,
+				isNetwork: false,
+			}
+		case "load15":
+			return {
+				metricName: "system.cpu.load_average.15m",
+				groupByAttributeKey: undefined,
+				unit: "load" as const,
+				isNetwork: false,
+			}
+		case "network":
+			return {
+				metricName: "system.network.io",
+				groupByAttributeKey: "direction",
+				unit: "bytes_per_second" as const,
+				isNetwork: true,
+			}
+	}
+}

apps/api/src/routes/query-runner.ts

@@ -2,8 +2,8 @@ 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"
+import type { QueryEngineServiceShape } from "@/services/warehouse/QueryEngineService"
+import type { WarehouseQueryServiceShape } from "@/services/warehouse/WarehouseQueryService"
 
 /**
  * Execute registry-declared warehouse queries.
@@ -15,72 +15,81 @@ import { WarehouseQueryService } from "@/services/warehouse/WarehouseQueryServic
  *
  * Handlers keep their own row-to-response mapping. What they lose is the
  * plumbing, not the presentation.
+ *
+ * The services are taken as VALUES, not read from context, and the runners are
+ * built once per handler group. That is load-bearing rather than stylistic:
+ * `QueryEngineService.cachedDirect` accepts an `Effect` with no requirements, so
+ * a context-reading runner could never be composed inside a cache wrapper —
+ * which `spanHierarchy` needs, since its probe must only fire on a cache miss.
+ * Currying here keeps every call site written as `runQuery(def, tenant, payload)`
+ * while the effects it returns carry `R = never`.
  */
-
-/**
- * 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 = <Payload, Row>(def: QueryDef<Payload, Row>, payload: Payload) => {
-	const settings = typeof def.settings === "function" ? def.settings(payload) : def.settings
-	return settings === undefined ? {} : { settings }
+export interface QueryRunnerDeps {
+	readonly warehouse: WarehouseQueryServiceShape
+	readonly queryEngine: QueryEngineServiceShape
 }
 
-/**
- * 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 = <Payload, Row, A, E extends QueryEngineDirectError>(
-	def: QueryDef<Payload, Row>,
-	tenant: TenantContext,
-	payload: Payload,
-	execute: Effect.Effect<A, E>,
-) =>
-	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)
-	})
+export const makeQueryRunners = ({ warehouse, queryEngine }: QueryRunnerDeps) => {
+	/**
+	 * Settings may be static or payload-dependent. Resolved to a spread so an
+	 * absent result omits the key entirely rather than passing an explicit
+	 * `settings: undefined`, which would read as "clear the profile defaults".
+	 */
+	const resolveSettings = <Payload, Row>(def: QueryDef<Payload, Row>, payload: Payload) => {
+		const settings = typeof def.settings === "function" ? def.settings(payload) : def.settings
+		return settings === undefined ? {} : { settings }
+	}
 
-/**
- * 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 = <Payload, Row>(
-	def: QueryDef<Payload, Row>,
-	tenant: TenantContext,
-	payload: Payload,
-) =>
-	Effect.gen(function* () {
-		const warehouse = yield* WarehouseQueryService
-		return yield* withPolicy(
+	/**
+	 * 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 = <Payload, Row, A, E extends QueryEngineDirectError>(
+		def: QueryDef<Payload, Row>,
+		tenant: TenantContext,
+		payload: Payload,
+		execute: Effect.Effect<A, E>,
+	) =>
+		Effect.gen(function* () {
+			// 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.
+	 */
+	const runQuery = <Payload, Row>(
+		def: QueryDef<Payload, Row>,
+		tenant: TenantContext,
+		payload: Payload,
+	) =>
+		withPolicy(
 			def,
 			tenant,
 			payload,
@@ -90,22 +99,19 @@ export const runQuery = <Payload, Row>(
 				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 = <Payload, Row>(
-	def: QueryDef<Payload, Row>,
-	tenant: TenantContext,
-	payload: Payload,
-) =>
-	Effect.gen(function* () {
-		const warehouse = yield* WarehouseQueryService
-		return yield* withPolicy(
+	/**
+	 * 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.
+	 */
+	const runQueryFirst = <Payload, Row>(
+		def: QueryDef<Payload, Row>,
+		tenant: TenantContext,
+		payload: Payload,
+	) =>
+		withPolicy(
 			def,
 			tenant,
 			payload,
@@ -117,4 +123,6 @@ export const runQueryFirst = <Payload, Row>(
 				})
 				.pipe(Effect.map(Option.getOrNull)),
 		)
-	})
+
+	return { runQuery, runQueryFirst } as const
+}