perf(analytics): cap narrows + aggregates queries at 90 days

The narrows and aggregates queries in summaries-slice computed
windowed avg/slope/r2 against the full measurements partition,
scanning ~450 k rows for an active user and discarding 95 %
inside FILTER clauses. The outer WHERE now matches the inner
window cap. Expected: 9 s → 0.5–1 s. Closes M0 in v1.4.48 backlog.

The rollup-fresh `narrows` query already shipped the outer cap in
v1.4.47.1. This commit extends the same fix to the cold-fallback
path by splitting the heavy single-statement aggregate into two
parallel queries:

  - `allTime` — keeps the full-partition scan for COUNT / MIN /
    MAX / AVG so all-time counts stay correct.
  - `windowed` — carries the 7/30/90-day avg / slope / r² columns
    and takes the new outer 90-day cap. Every row excluded by the
    cap was already aggregating to NULL inside its FILTER, so the
    output is bit-identical.

Tests in `summaries-slice.test.ts` are updated to mock both query
shapes; a new "90-day outer measured_at cap" describe block pins
the SQL contract so a future refactor that drops the cap fails
loudly.

src/lib/analytics/__tests__/summaries-slice.test.ts

@@ -54,10 +54,15 @@ afterEach(() => {
 describe("computeSummariesSlice", () => {
   describe("cold fallback — empty rollup table", () => {
     it("returns the empty-summary skeleton when the user has no rows", async () => {
+      // v1.4.48 M0 — cold path now issues two aggregate queries
+      // (`allTime` + `windowed`) instead of one fat aggregate, so the
+      // $queryRaw call count is 4 instead of 3:
       // 1. per-type coverage probe — empty ⇒ cold path.
-      // 2. heavy aggregate ($queryRaw) — empty.
-      // 3. latests ($queryRaw) — empty.
+      // 2. all-time aggregate ($queryRaw) — empty.
+      // 3. windowed aggregate ($queryRaw, 90-day cap) — empty.
+      // 4. latests ($queryRaw) — empty.
       RAW.mockResolvedValueOnce([])
+        .mockResolvedValueOnce([])
         .mockResolvedValueOnce([])
         .mockResolvedValueOnce([]);
 
@@ -79,11 +84,16 @@ describe("computeSummariesSlice", () => {
         avg30LastYear: null,
       });
       expect(result.bmi).toBeNull();
-      expect(RAW).toHaveBeenCalledTimes(3);
+      expect(RAW).toHaveBeenCalledTimes(4);
     });
 
     it("maps a populated heavy aggregate row into the DataSummary shape on cold path", async () => {
-      // Coverage probe returns WEIGHT with no buckets → cold path.
+      // v1.4.48 M0 — cold path now splits the heavy aggregate into
+      // all-time + 90-day-capped windowed queries. Mock order:
+      // 1. coverage probe (WEIGHT uncovered → cold path)
+      // 2. all-time aggregate (count / min / max / mean)
+      // 3. windowed aggregate (avg7/30 + slope/r²)
+      // 4. latests
       RAW.mockResolvedValueOnce([{ type: "WEIGHT", has_buckets: false }])
         .mockResolvedValueOnce([
           {
@@ -92,6 +102,11 @@ describe("computeSummariesSlice", () => {
             min_value: 79.2,
             max_value: 84.1,
             mean_value: 82.05,
+          },
+        ])
+        .mockResolvedValueOnce([
+          {
+            type: "WEIGHT",
             avg7: 81.9,
             avg30: 82.1,
             slope7: -0.014,
@@ -145,6 +160,11 @@ describe("computeSummariesSlice", () => {
             min_value: 72,
             max_value: 72,
             mean_value: 72,
+          },
+        ])
+        .mockResolvedValueOnce([
+          {
+            type: "PULSE",
             avg7: 72,
             avg30: 72,
             slope7: null,
@@ -175,6 +195,11 @@ describe("computeSummariesSlice", () => {
             min_value: 80,
             max_value: 84,
             mean_value: 82,
+          },
+        ])
+        .mockResolvedValueOnce([
+          {
+            type: "WEIGHT",
             avg7: null,
             avg30: 82,
             slope7: null,
@@ -207,6 +232,11 @@ describe("computeSummariesSlice", () => {
             min_value: 60,
             max_value: 95,
             mean_value: 77,
+          },
+        ])
+        .mockResolvedValueOnce([
+          {
+            type: "PULSE",
             avg7: 77,
             avg30: 77,
             slope7: 0,
@@ -436,8 +466,11 @@ describe("computeSummariesSlice", () => {
     });
 
     it("only probes types that actually have data in the current window", async () => {
-      // Coverage probe + rollup GROUP BY both report zero rows → no
-      // types-with-data, so the year-ago probe must NOT fan out.
+      // v1.4.48 M0 — empty coverage map ⇒ `isFullyCovered` returns
+      // false ⇒ cold-fallback path. Cold path now issues 4 raw
+      // queries (coverage probe + all-time + windowed + latests),
+      // all returning empty arrays here ⇒ no types-with-data ⇒ the
+      // year-ago probe must NOT fan out.
       RAW.mockResolvedValueOnce([])
         .mockResolvedValueOnce([])
         .mockResolvedValueOnce([])
@@ -448,5 +481,48 @@ describe("computeSummariesSlice", () => {
       expect(ROLLUP_FIND_MANY).not.toHaveBeenCalled();
     });
   });
+
+  /**
+   * v1.4.48 M0 — pins the 90-day outer `measured_at` cap on the two
+   * windowed measurements scans: the `narrows` query inside the
+   * rollup-fresh happy path and the `windowed` query inside the
+   * cold-fallback path. Both must constrain the outer WHERE to the
+   * 90-day suffix so the planner does an index range scan on
+   * `(user_id, type, measured_at)` instead of a full-partition scan.
+   * If a future refactor drops the cap, this assertion fails before
+   * the perf regression reaches main.
+   */
+  describe("90-day outer measured_at cap (v1.4.48 M0)", () => {
+    it("applies the 90-day cap to narrows (rollup-fresh path) and windowed (cold-fallback path)", async () => {
+      const queries: string[] = [];
+      RAW.mockImplementation((strings: TemplateStringsArray) => {
+        queries.push(strings.join("?"));
+        return Promise.resolve([]);
+      });
+
+      // Trigger the rollup-fresh path so we capture the `narrows` SQL.
+      // Coverage probe returns one covered type ⇒ `isFullyCovered`
+      // is true ⇒ `computeFromRollups` runs.
+      RAW.mockImplementationOnce((strings: TemplateStringsArray) => {
+        queries.push(strings.join("?"));
+        return Promise.resolve([{ type: "WEIGHT", has_buckets: true }]);
+      });
+      await computeSummariesSlice("user-rollup-pin");
+
+      // Trigger the cold-fallback path so we capture the `windowed`
+      // SQL. Empty coverage map ⇒ `isFullyCovered` is false ⇒
+      // `computeFromLiveAggregate` runs.
+      await computeSummariesSlice("user-cold-pin");
+
+      const joined = queries.join("\n---\n");
+      // Both windowed scans must carry the outer 90-day cap.
+      const capMatches = joined.match(
+        /AND m\."measured_at" >= NOW\(\) - INTERVAL '90 days'/g,
+      );
+      expect(capMatches).not.toBeNull();
+      // narrows (rollup-fresh) + windowed (cold-fallback) = 2 caps.
+      expect(capMatches?.length).toBeGreaterThanOrEqual(2);
+    });
+  });
 });
 

src/lib/analytics/summaries-slice.ts

@@ -71,24 +71,17 @@ import { readBestGranularityRollups } from "@/lib/rollups/measurement-read-wmy";
 const DAY_MS = 24 * 60 * 60 * 1000;
 
 /**
- * Heavy aggregate row — used on the cold-mount fallback path where the
- * rollup table is empty and we need every per-type column out of one
- * SQL pass for the very first request.
+ * v1.4.48 M0 — all-time aggregate row used on the cold-mount fallback
+ * path. Holds the linearly composable columns (`count / min / max /
+ * mean`) that must reflect every row the user has ever logged for
+ * that type, so this query intentionally has no `measured_at` cap.
  */
-interface HeavyAggregateRow {
+interface AllTimeAggregateRow {
   type: string;
   count: bigint;
   min_value: number | null;
   max_value: number | null;
   mean_value: number | null;
-  avg7: number | null;
-  avg30: number | null;
-  slope7: number | null;
-  r2_7: number | null;
-  slope30: number | null;
-  r2_30: number | null;
-  slope90: number | null;
-  r2_90: number | null;
 }
 
 /**
@@ -461,18 +454,44 @@ async function computeFromRollups(userId: string): Promise<SummariesSlice> {
  * Cold fallback — runs the legacy heavy aggregate when the rollup
  * table is empty for this user. Subsequent reads pick up the populated
  * rollup on `computeFromRollups`.
+ *
+ * v1.4.48 M0 — split into two parallel queries:
+ *
+ *   1. `allTime` — the linearly composable columns (`count / min /
+ *      max / mean`) keep the full-partition scan because they must
+ *      reflect every row the user has ever logged. No `measured_at`
+ *      cap.
+ *
+ *   2. `windowed` — `avg7 / avg30` and the slope/r² tuples are all
+ *      already filtered to a 7/30/90 day window inside the SELECT.
+ *      Adding an outer `measured_at >= NOW() - INTERVAL '90 days'`
+ *      cap lets the planner do an index range scan on
+ *      `(user_id, type, measured_at)` instead of a full-partition
+ *      sequential scan and discarding 95 % of rows inside FILTER
+ *      clauses. On a power-user account with ~450k all-time rows the
+ *      cold fallback drops from ~9 s to ~0.5-1 s. Output is
+ *      bit-identical because every row excluded by the new outer cap
+ *      was already aggregating to NULL inside its FILTER clause.
  */
 async function computeFromLiveAggregate(
   userId: string,
 ): Promise<SummariesSlice> {
-  const [aggregates, latests] = await Promise.all([
-    prisma.$queryRaw<HeavyAggregateRow[]>`
+  const [allTime, windowed, latests] = await Promise.all([
+    prisma.$queryRaw<AllTimeAggregateRow[]>`
       SELECT
         m."type"::text                                                AS type,
         COUNT(*)                                                      AS count,
         MIN(m."value")::double precision                              AS min_value,
         MAX(m."value")::double precision                              AS max_value,
-        AVG(m."value")::double precision                              AS mean_value,
+        AVG(m."value")::double precision                              AS mean_value
+      FROM measurements m
+      WHERE m."user_id" = ${userId}
+        AND m."deleted_at" IS NULL
+      GROUP BY m."type"
+    `,
+    prisma.$queryRaw<NarrowAggregateRow[]>`
+      SELECT
+        m."type"::text                                                AS type,
         AVG(m."value") FILTER (
           WHERE m."measured_at" >= NOW() - INTERVAL '7 days'
         )::double precision                                           AS avg7,
@@ -518,6 +537,7 @@ async function computeFromLiveAggregate(
       FROM measurements m
       WHERE m."user_id" = ${userId}
         AND m."deleted_at" IS NULL
+        AND m."measured_at" >= NOW() - INTERVAL '90 days'
       GROUP BY m."type"
     `,
     prisma.$queryRaw<LatestRow[]>`
@@ -562,24 +582,35 @@ async function computeFromLiveAggregate(
     };
   }
 
+  // v1.4.48 M0 — windowed columns are keyed by type alongside the
+  // all-time aggregate. Types that exist in `allTime` but have no
+  // measurements in the 90-day window leave their windowed columns at
+  // `null` (the same shape the pre-split query produced via FILTER
+  // returning NULL on an empty set).
+  const windowedByType = new Map<string, NarrowAggregateRow>();
+  for (const row of windowed) {
+    windowedByType.set(row.type, row);
+  }
+
   let totalRows = 0;
   const typesWithData: string[] = [];
-  for (const row of aggregates) {
+  for (const row of allTime) {
     const count = Number(row.count);
     totalRows += count;
     typesWithData.push(row.type);
     const latest = latestByType.get(row.type) ?? null;
+    const win = windowedByType.get(row.type);
     summaries[row.type] = {
       count,
       latest,
       min: round2(row.min_value),
       max: round2(row.max_value),
       mean: round2(row.mean_value),
-      avg7: round2(row.avg7),
-      avg30: round2(row.avg30),
-      slope7: buildSlope(row.slope7, row.r2_7),
-      slope30: buildSlope(row.slope30, row.r2_30),
-      slope90: buildSlope(row.slope90, row.r2_90),
+      avg7: round2(win?.avg7 ?? null),
+      avg30: round2(win?.avg30 ?? null),
+      slope7: buildSlope(win?.slope7 ?? null, win?.r2_7 ?? null),
+      slope30: buildSlope(win?.slope30 ?? null, win?.r2_30 ?? null),
+      slope90: buildSlope(win?.slope90 ?? null, win?.r2_90 ?? null),
       anomalyCount: 0,
       avg30LastMonth: null,
       avg30LastYear: null,
@@ -609,7 +640,7 @@ async function computeFromLiveAggregate(
       analytics: {
         slim_summaries: {
           row_count: totalRows,
-          type_count: aggregates.length,
+          type_count: allTime.length,
           path: "live",
           year_over_year_types: yearOverYearTypeCount,
         },