perf(analytics): fold the comprehensive aggregate's canonical self-join into one CTE

The warm narrow query and the cold heavy query each referenced the
canonical-source subquery twice — once in the window_stats CTE and once in
the outer body — so the DISTINCT-ON pick over the live 90-day rows ran twice
per rebuild. Extract canonicalMeasurementsCte (the SELECT body without the
alias wrapper), bind it once as WITH cm AS (...), and select FROM cm m from
both window_stats and the outer aggregate. Output is identical; the dominant
self-join runs once.

canonicalMeasurementsFrom now wraps the same body, byte-stable for the
single-reference call sites (pinned by a new parity test).

src/lib/analytics/__tests__/source-rank-sql.test.ts

@@ -0,0 +1,59 @@
+/**
+ * Unit pins for the source-rank SQL builders.
+ *
+ * `canonicalMeasurementsCte` (v1.18.11 perf#3a) was extracted so a statement
+ * that references the canonical-source set more than once can bind it as a
+ * single `WITH cm AS (…)` CTE and fold away the duplicate DISTINCT-ON
+ * self-join. These tests pin:
+ *   1. the alias-wrapped `canonicalMeasurementsFrom` is exactly the CTE body
+ *      wrapped in `( … ) m` — so the single-reference call site is unchanged;
+ *   2. the canonical body carries exactly ONE DISTINCT-ON pick — so a fold
+ *      that references the CTE twice runs the expensive pick once;
+ *   3. the interval whitelist still rejects an unsafe splice.
+ */
+import { describe, expect, it } from "vitest";
+
+import {
+  canonicalMeasurementsCte,
+  canonicalMeasurementsFrom,
+} from "../source-rank-sql";
+
+const RANK = "90";
+
+describe("canonicalMeasurementsCte / canonicalMeasurementsFrom", () => {
+  it("wraps the CTE body verbatim in the ( … ) m alias", () => {
+    const cte = canonicalMeasurementsCte(RANK, "90 days");
+    const from = canonicalMeasurementsFrom(RANK, "90 days");
+    // The alias form is the CTE body plus the closing `) m` wrapper. Pinning
+    // equality means the single-reference call sites keep byte-stable SQL
+    // while the CTE body can be reused for the folded double-reference case.
+    expect(from).toBe(`(${cte}\n      ) m`);
+  });
+
+  it("runs exactly one DISTINCT-ON canonical pick in the body", () => {
+    const cte = canonicalMeasurementsCte(RANK, "90 days");
+    const matches = cte.match(/DISTINCT ON/g) ?? [];
+    expect(matches).toHaveLength(1);
+  });
+
+  it("binds the user id as $1 and scopes both the inner and outer windows", () => {
+    const cte = canonicalMeasurementsCte(RANK, "90 days");
+    // user id is always a bound parameter, never spliced.
+    expect(cte).toContain('"user_id" = $1');
+    // The trailing-window bound appears on both the inner pick and the outer
+    // filter so the folded references stay scoped identically.
+    const intervalMatches = cte.match(/INTERVAL '90 days'/g) ?? [];
+    expect(intervalMatches).toHaveLength(2);
+  });
+
+  it("omits the window clause entirely when no interval is given", () => {
+    const cte = canonicalMeasurementsCte(RANK);
+    expect(cte).not.toContain("INTERVAL");
+  });
+
+  it("rejects an unsafe interval splice", () => {
+    expect(() => canonicalMeasurementsCte(RANK, "90 days; DROP TABLE")).toThrow(
+      /unsafe interval/,
+    );
+  });
+});

src/lib/analytics/source-rank-sql.ts

@@ -107,19 +107,25 @@ function assertInterval(value: string): string {
 }
 
 /**
- * Build a FROM-clause subquery (aliased `m`) that restricts raw `measurements`
- * to the canonical-source rows per (type, day): the inner `DISTINCT ON` picks
- * the ladder-winning source for each day, and the join keeps only that source's
- * readings. Use it in place of `FROM measurements m` so a live aggregate over
- * an overlapping vital never blends two devices — the same collapse the rollup
- * readers apply, kept in lockstep for live/rollup parity.
+ * Build the canonical-source SELECT body — the raw `measurements` rows
+ * restricted to the ladder-winning source per (type, day) — WITHOUT the
+ * outer `( … ) m` alias wrapper. The inner `DISTINCT ON` picks the
+ * canonical source for each day and the join keeps only that source's
+ * readings.
+ *
+ * Use this when the same canonical set must be referenced more than once in
+ * one statement (e.g. a `window_stats` CTE plus the outer aggregate): bind it
+ * once via `WITH cm AS (${canonicalMeasurementsCte(...)})` and select
+ * `FROM cm m` everywhere, so the expensive DISTINCT-ON self-join runs once
+ * rather than per reference. For the single-reference case use
+ * `canonicalMeasurementsFrom`, which wraps this body in the alias.
  *
  * The user id is bound as `$1`. `rankUnqualified` must be a CASE built with
  * unqualified `"type"`/`"source"` columns. `sinceInterval` (e.g. `"90 days"`)
  * is whitelisted and, when given, scopes both the inner pick and the outer
  * filter to a trailing window.
  */
-export function canonicalMeasurementsFrom(
+export function canonicalMeasurementsCte(
   rankUnqualified: string,
   sinceInterval?: string,
 ): string {
@@ -129,7 +135,7 @@ export function canonicalMeasurementsFrom(
   const sinceOuter = sinceInterval
     ? `AND mm."measured_at" >= NOW() - INTERVAL '${assertInterval(sinceInterval)}'`
     : "";
-  return `(
+  return `
         SELECT mm.*
         FROM measurements mm
         JOIN (
@@ -148,6 +154,26 @@ export function canonicalMeasurementsFrom(
           AND c.canon = mm."source"
         WHERE mm."user_id" = $1
           AND mm."deleted_at" IS NULL
-          ${sinceOuter}
+          ${sinceOuter}`;
+}
+
+/**
+ * Build a FROM-clause subquery (aliased `m`) that restricts raw `measurements`
+ * to the canonical-source rows per (type, day): the inner `DISTINCT ON` picks
+ * the ladder-winning source for each day, and the join keeps only that source's
+ * readings. Use it in place of `FROM measurements m` so a live aggregate over
+ * an overlapping vital never blends two devices — the same collapse the rollup
+ * readers apply, kept in lockstep for live/rollup parity.
+ *
+ * The user id is bound as `$1`. `rankUnqualified` must be a CASE built with
+ * unqualified `"type"`/`"source"` columns. `sinceInterval` (e.g. `"90 days"`)
+ * is whitelisted and, when given, scopes both the inner pick and the outer
+ * filter to a trailing window.
+ */
+export function canonicalMeasurementsFrom(
+  rankUnqualified: string,
+  sinceInterval?: string,
+): string {
+  return `(${canonicalMeasurementsCte(rankUnqualified, sinceInterval)}
       ) m`;
 }

src/lib/insights/__tests__/comprehensive-aggregator.test.ts

@@ -48,11 +48,9 @@ vi.mock("@/lib/logging/context", () => ({
 // aggregator's plumbing — path selection, bpRaw threading, dailyByType
 // composition — is exercised without coupling the unit test to the rank
 // builder's enum-whitelist internals.
-vi.mock("@/lib/analytics/source-rank-sql", () => ({
-  buildSourceRankCase: vi.fn(() => "90"),
-  canonicalMeasurementsFrom: vi.fn(
-    (_rank: string, sinceInterval?: string) =>
-      `(
+vi.mock("@/lib/analytics/source-rank-sql", () => {
+  const cte = (_rank: string, sinceInterval?: string) =>
+    `
         SELECT mm.*
         FROM measurements mm
         WHERE mm."user_id" = $1
@@ -61,10 +59,19 @@ vi.mock("@/lib/analytics/source-rank-sql", () => ({
             sinceInterval
               ? `AND mm."measured_at" >= NOW() - INTERVAL '${sinceInterval}'`
               : ""
-          }
+          }`;
+  return {
+    buildSourceRankCase: vi.fn(() => "90"),
+    // v1.18.11 perf#3a — the aggregator now folds the canonical-source
+    // self-join into a single `WITH cm AS (…)` CTE referenced twice, so the
+    // unit mock must expose both the CTE body and the alias-wrapped form.
+    canonicalMeasurementsCte: vi.fn(cte),
+    canonicalMeasurementsFrom: vi.fn(
+      (rank: string, sinceInterval?: string) => `(${cte(rank, sinceInterval)}
       ) m`,
-  ),
-}));
+    ),
+  };
+});
 
 import { prisma } from "@/lib/db";
 import { buildComprehensiveAggregate } from "../comprehensive-aggregator";

src/lib/insights/comprehensive-aggregator.ts

@@ -112,6 +112,7 @@ import {
 } from "@/lib/rollups/measurement-read";
 import {
   buildSourceRankCase,
+  canonicalMeasurementsCte,
   canonicalMeasurementsFrom,
 } from "@/lib/analytics/source-rank-sql";
 import {
@@ -398,12 +399,18 @@ async function buildFromRollups(
       timeSubquery(timings, "narrow", () =>
         prisma.$queryRawUnsafe<NarrowAggregateRow[]>(
           `
-      WITH window_stats AS (
+      -- v1.18.11 perf#3a — the canonical-source self-join is the dominant
+      -- cost; bind it ONCE as a CTE and reference it from both window_stats
+      -- and the outer aggregate so the DISTINCT-ON pick runs a single time
+      -- instead of twice per rebuild. Output is identical.
+      WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")}
+      ),
+      window_stats AS (
         SELECT
           m."type",
           AVG(m."value") AS mean_value,
           STDDEV_POP(m."value") AS stddev_value
-        FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")}
+        FROM cm m
         GROUP BY m."type"
       )
       SELECT
@@ -466,7 +473,7 @@ async function buildFromRollups(
         ) FILTER (
           WHERE m."measured_at" >= NOW() - INTERVAL '90 days'
         )::double precision                                           AS r2_90
-      FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")}
+      FROM cm m
       JOIN window_stats ws ON ws."type" = m."type"
       GROUP BY m."type", ws.stddev_value
     `,
@@ -617,12 +624,17 @@ async function buildFromLiveAggregate(
     timeSubquery(timings, "heavy", () =>
       prisma.$queryRawUnsafe<HeavyAggregateRow[]>(
         `
-      WITH window_stats AS (
+      -- v1.18.11 perf#3a — bind the canonical-source self-join once (see the
+      -- warm narrow query). window_stats + the outer aggregate both read cm,
+      -- so the DISTINCT-ON pick runs once. Output is identical.
+      WITH cm AS (${canonicalMeasurementsCte(rankUnqualified, "90 days")}
+      ),
+      window_stats AS (
         SELECT
           m."type",
           AVG(m."value") AS mean_value,
           STDDEV_POP(m."value") AS stddev_value
-        FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")}
+        FROM cm m
         GROUP BY m."type"
       )
       SELECT
@@ -685,7 +697,7 @@ async function buildFromLiveAggregate(
         ) FILTER (
           WHERE m."measured_at" >= NOW() - INTERVAL '90 days'
         )::double precision                                           AS r2_90
-      FROM ${canonicalMeasurementsFrom(rankUnqualified, "90 days")}
+      FROM cm m
       JOIN window_stats ws ON ws."type" = m."type"
       GROUP BY m."type", ws.stddev_value
     `,