fix(webapp): read queue metrics from the shared warehouse for dedicated-datastore orgs

Fix the Queues dashboard showing "No activity" on the queue-metrics
widgets for organizations whose data is stored in a dedicated
ClickHouse, even though their metrics were being collected. Queue
metrics are written to a single shared warehouse, but the dashboard was
reading them from the per-org ClickHouse; it now reads from the shared
warehouse, so the widgets display the collected data.

Mono-RevId: e5375536adcde57f1f1461228669bf6db46f2552

Author: ericallam

.server-changes/queue-metrics-dashboard-no-activity.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Fix the Queues page showing "No activity" on the queue-metrics charts for some organizations even though their metrics were being collected. Those charts now display the collected data.

apps/webapp/app/routes/api.v1.queues.$queueParam.metrics.ts

@@ -73,7 +73,7 @@ export const loader = createLoaderApiRoute(
     try {
       const clickhouse = await clickhouseFactory.getClickhouseForOrganization(
         authentication.environment.organizationId,
-        "query"
+        "queueMetrics"
       );
 
       const ids = {

apps/webapp/app/services/clickhouse/clickhouseFactory.server.test.ts

@@ -0,0 +1,69 @@
+import { describe, expect, it } from "vitest";
+import type { OrganizationDataStoresRegistry } from "~/services/dataStores/organizationDataStoresRegistry.server";
+import { ClickhouseFactory, getQueueMetricsClickhouseClient } from "./clickhouseFactory.server";
+
+const ORG_WITH_OVERRIDE = "org_with_dedicated_ch";
+const DEDICATED_CH_URL = "http://dedicated-ch.example:8123";
+
+function registryWithOverride(overrides: Record<string, string>): OrganizationDataStoresRegistry {
+  return {
+    isLoaded: true,
+    isReady: Promise.resolve(),
+    get(organizationId: string, kind: string) {
+      const url = overrides[organizationId];
+      if (kind === "CLICKHOUSE" && url) {
+        return { kind: "CLICKHOUSE", url };
+      }
+      return null;
+    },
+  } as unknown as OrganizationDataStoresRegistry;
+}
+
+describe("ClickhouseFactory queue-metrics routing", () => {
+  it("reads queue metrics from the shared warehouse even when the org has a dedicated CH override", () => {
+    const factory = new ClickhouseFactory(
+      registryWithOverride({ [ORG_WITH_OVERRIDE]: DEDICATED_CH_URL })
+    );
+
+    const client = factory.getClickhouseForOrganizationSync(ORG_WITH_OVERRIDE, "queueMetrics");
+
+    expect(client).toBe(getQueueMetricsClickhouseClient());
+  });
+
+  it("routes queue metrics to the shared client identically with or without an org override", () => {
+    const withOverride = new ClickhouseFactory(
+      registryWithOverride({ [ORG_WITH_OVERRIDE]: DEDICATED_CH_URL })
+    );
+    const withoutOverride = new ClickhouseFactory(registryWithOverride({}));
+
+    const overrideClient = withOverride.getClickhouseForOrganizationSync(
+      ORG_WITH_OVERRIDE,
+      "queueMetrics"
+    );
+    const defaultClient = withoutOverride.getClickhouseForOrganizationSync(
+      ORG_WITH_OVERRIDE,
+      "queueMetrics"
+    );
+
+    expect(overrideClient).toBe(defaultClient);
+    expect(overrideClient).toBe(getQueueMetricsClickhouseClient());
+  });
+
+  it("still honors the per-org override for non-shared client types (events)", () => {
+    const withOverride = new ClickhouseFactory(
+      registryWithOverride({ [ORG_WITH_OVERRIDE]: DEDICATED_CH_URL })
+    );
+    const withoutOverride = new ClickhouseFactory(registryWithOverride({}));
+
+    const overrideEvents = withOverride.getClickhouseForOrganizationSync(
+      ORG_WITH_OVERRIDE,
+      "events"
+    );
+    const defaultEvents = withoutOverride.getClickhouseForOrganizationSync(
+      ORG_WITH_OVERRIDE,
+      "events"
+    );
+
+    expect(overrideEvents).not.toBe(defaultEvents);
+  });
+});

apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts

@@ -489,6 +489,16 @@ export type ClientType =
   | "runsList"
   | "queueMetrics";
 
+/**
+ * Client types whose data is written to a single shared warehouse by an org-less writer, so their
+ * reads must always resolve to that same shared default client and must ignore any per-org
+ * ClickHouse datastore override (a dedicated CH holds the schema but no rows for these). Add a
+ * client type here to make it globally-routed. `queueMetrics` is written by the org-less ingestion
+ * consumer (`getQueueMetricsClickhouseClient`); the replication/engine writer types are NOT here
+ * because their consumers route writes per-org and so legitimately honor the override on both sides.
+ */
+const SHARED_WAREHOUSE_CLIENT_TYPES = new Set<ClientType>(["queueMetrics"]);
+
 function buildOrgClickhouseClient(url: string, clientType: ClientType): ClickHouse {
   const parsed = new URL(url);
   parsed.searchParams.delete("secure");
@@ -655,7 +665,9 @@ export class ClickhouseFactory {
   }
 
   getClickhouseForOrganizationSync(organizationId: string, clientType: ClientType): ClickHouse {
-    const dataStore = this._registry.get(organizationId, "CLICKHOUSE");
+    const dataStore = SHARED_WAREHOUSE_CLIENT_TYPES.has(clientType)
+      ? null
+      : this._registry.get(organizationId, "CLICKHOUSE");
 
     if (!dataStore) {
       switch (clientType) {

apps/webapp/app/services/dashboardAgentWatchChecks.server.ts

@@ -110,7 +110,7 @@ export async function readWatchQueueDepth(
 
   const clickhouse = await clickhouseFactory.getClickhouseForOrganization(
     environment.organizationId,
-    "query"
+    "queueMetrics"
   );
 
   const bucketMs = DEPTH_FALLBACK_BUCKET_SECONDS * 1000;

apps/webapp/vitest.config.ts

@@ -18,6 +18,7 @@ export default defineConfig({
       "app/runEngine/concerns/**/*.test.ts",
       "app/runEngine/services/**/*.test.ts",
       "app/services/realtime/**/*.test.ts",
+      "app/services/clickhouse/**/*.test.ts",
       "app/utils/**/*.test.ts",
       "app/components/code/**/*.test.ts",
       "app/components/runs/**/*.test.ts",