diff --git a/src/lib/analytics/sleep-night.ts b/src/lib/analytics/sleep-night.ts index 71d82ee7a..50faedcdf 100644 --- a/src/lib/analytics/sleep-night.ts +++ b/src/lib/analytics/sleep-night.ts @@ -1324,9 +1324,16 @@ export function summarizeSleepNights( tz: string, priorityJson: unknown = null, ): SleepNightSummary { - const nights = reconstructSleepNights(rows, tz, priorityJson).filter( - (n) => n.asleepMinutes > 0, - ); + return summarizeNights(reconstructSleepNights(rows, tz, priorityJson)); +} + +/** + * `summarizeSleepNights` over nights already reconstructed, for a reader + * that gets them from the chunked Postgres walk (`readSleepNights`) instead + * of holding the stage rows itself. + */ +export function summarizeNights(all: readonly SleepNight[]): SleepNightSummary { + const nights = all.filter((n) => n.asleepMinutes > 0); const dataPoints = nights.map((n) => ({ date: n.measuredAt, value: n.asleepMinutes, diff --git a/src/lib/analytics/summaries-slice.ts b/src/lib/analytics/summaries-slice.ts index f489baf44..2d43ab91f 100644 --- a/src/lib/analytics/summaries-slice.ts +++ b/src/lib/analytics/summaries-slice.ts @@ -103,17 +103,20 @@ import { type RollupCoverageMap, } from "@/lib/rollups/measurement-coverage"; import { - summarizeSleepNights, + summarizeNights, type SleepSourceDiscrepancy, - type SleepStageRow, } from "@/lib/analytics/sleep-night"; import { resolveUserTimezone } from "@/lib/tz/resolver"; +import { readSleepNights } from "@/lib/measurements/list/sleep-nights"; import { userDayKey } from "@/lib/tz/format"; import { localDayWindow } from "@/lib/tz/local-day"; import { canonicalRowsOfDay } from "@/lib/mood/linked-context"; const DAY_MS = 24 * 60 * 60 * 1000; +/** No account holds this many nights; the walk is uncapped in practice. */ +const ALL_NIGHTS = 100_000; + /** * v1.18.10 — upper bound on the cold-fallback `allTime` aggregate's * scan window. Sits far beyond any real account's history (the earliest prod @@ -459,9 +462,10 @@ async function withLatestDayValues( * summary with one computed over per-night asleep totals (one DataPoint * per night). The unit stays MINUTES (canonical), so existing consumers * that already treat the summary as minutes keep working — they now - * average nights instead of stages, which is strictly more correct. The - * read is bounded to a year (~5 rows/night) and is skipped entirely when - * the user has logged no sleep. + * average nights instead of stages, which is strictly more correct. It + * covers the whole history, like every other type here (a year's cut left + * `avg30LastYear` and the all-time count and range short), and is skipped + * entirely when the user has logged no sleep. */ async function withSleepNightTotals( userId: string, @@ -471,32 +475,18 @@ async function withSleepNightTotals( // No sleep data at all → nothing to recompute (the empty summary stays). if (!existing || existing.count === 0) return slice; - const since = new Date(Date.now() - 365 * DAY_MS); - const rows = (await prisma.measurement.findMany({ - where: { - userId, - type: "SLEEP_DURATION", - deletedAt: null, - measuredAt: { gte: since }, - }, - orderBy: { measuredAt: "asc" }, - // `deviceType` feeds the writer-level collapse: two HealthKit apps - // behind one source (watch stages vs phone in-bed) must not blend. - select: { - value: true, - measuredAt: true, - sleepStage: true, - source: true, - deviceType: true, - }, - })) as SleepStageRow[]; - if (rows.length === 0) return slice; - const [tz, priorityJson] = await Promise.all([ resolveUserTimezone(userId), loadUserSourcePriority(userId), ]); - const { summary, latestNight } = summarizeSleepNights(rows, tz, priorityJson); + // The whole history, as every other type in the slice: the walk names the + // nights in Postgres and reconstructs them ninety at a time, so the heap + // holds a chunk's rows however long the history is. + const nights = await readSleepNights(userId, tz, priorityJson, { + maxNights: ALL_NIGHTS, + }); + if (nights.length === 0) return slice; + const { summary, latestNight } = summarizeNights(nights); // Preserve the slope tuples the night summary doesn't compute (the // dashboard tile reads slope30); recompute them off the night series is // out of scope here — the per-night `summarize()` already fills slope7 / diff --git a/src/lib/insights/__tests__/general-status.test.ts b/src/lib/insights/__tests__/general-status.test.ts index 1c3c74ada..bd44c3aec 100644 --- a/src/lib/insights/__tests__/general-status.test.ts +++ b/src/lib/insights/__tests__/general-status.test.ts @@ -14,6 +14,26 @@ vi.mock("@/lib/db", () => ({ }, })); +// The per-type latest-readings read is one LATERAL query in Postgres; here +// it is served from the `measurement.findMany` fixtures, newest first as the +// query orders them, so the suite keeps describing its data as rows. +vi.mock("@/lib/insights/recent-readings", async (importOriginal) => { + const actual = + await importOriginal(); + const { prisma: db } = await import("@/lib/db"); + return { + ...actual, + readLatestReadingsPerType: async () => + [ + ...((await db.measurement.findMany({} as never)) as Array<{ + type: string; + value: number; + measuredAt: Date; + }>), + ].reverse(), + }; +}); + vi.mock("@/lib/insights/status-provider", () => ({ runStatusCompletion: vi.fn(), })); diff --git a/src/lib/insights/__tests__/locale-pipeline.test.ts b/src/lib/insights/__tests__/locale-pipeline.test.ts index 4254c33be..5302d72e2 100644 --- a/src/lib/insights/__tests__/locale-pipeline.test.ts +++ b/src/lib/insights/__tests__/locale-pipeline.test.ts @@ -81,6 +81,26 @@ vi.mock("@/lib/db", () => ({ }, })); +// The per-type latest-readings read is one LATERAL query in Postgres; here +// it is served from the `measurement.findMany` fixtures, newest first as the +// query orders them, so the suite keeps describing its data as rows. +vi.mock("@/lib/insights/recent-readings", async (importOriginal) => { + const actual = + await importOriginal(); + const { prisma: db } = await import("@/lib/db"); + return { + ...actual, + readLatestReadingsPerType: async () => + [ + ...((await db.measurement.findMany({} as never)) as Array<{ + type: string; + value: number; + measuredAt: Date; + }>), + ].reverse(), + }; +}); + vi.mock("@/lib/insights/status-provider", () => ({ runStatusCompletion: vi.fn(), })); diff --git a/src/lib/insights/__tests__/prompt-soft-delete-filter.test.ts b/src/lib/insights/__tests__/prompt-soft-delete-filter.test.ts index c26ee7d95..5e7dcf05b 100644 --- a/src/lib/insights/__tests__/prompt-soft-delete-filter.test.ts +++ b/src/lib/insights/__tests__/prompt-soft-delete-filter.test.ts @@ -19,9 +19,10 @@ function read(rel: string): string { describe("prompt reads exclude tombstoned rows", () => { it("general-status filters its measurement read", () => { - const src = read("../general-status.ts"); - expect(src).toMatch( - /measurement\s*\n?\s*\.findMany\(\{\s*where:\s*\{[\s\S]{0,300}?\bdeletedAt:\s*null/, + // The overview reads through the per-type latest-readings query. + expect(read("../general-status.ts")).toMatch(/readLatestReadingsPerType\(/); + expect(read("../recent-readings.ts")).toMatch( + /readLatestReadingsPerType[\s\S]*?m\."deleted_at" IS NULL/, ); }); diff --git a/src/lib/insights/__tests__/pulse-status.test.ts b/src/lib/insights/__tests__/pulse-status.test.ts index 6965abddc..e593ba9b8 100644 --- a/src/lib/insights/__tests__/pulse-status.test.ts +++ b/src/lib/insights/__tests__/pulse-status.test.ts @@ -185,8 +185,11 @@ describe("generatePulseStatusForUser — A2 resting-target in-target %", () => { dateOfBirth: null, gender: null, } as never); - // pulse-status reads PULSE first, then RESTING_HEART_RATE. + // pulse-status reads PULSE first, then RESTING_HEART_RATE. The latest + // 365 PULSE readings reach back less than 30 days here, so the PULSE + // read takes the 30-day window as a second query. vi.mocked(prisma.measurement.findMany) + .mockResolvedValueOnce(pulseRecords as never) .mockResolvedValueOnce(pulseRecords as never) .mockResolvedValueOnce(restingRecords as never) .mockResolvedValue([] as never); diff --git a/src/lib/insights/__tests__/recent-readings.test.ts b/src/lib/insights/__tests__/recent-readings.test.ts new file mode 100644 index 000000000..4cb16aa6d --- /dev/null +++ b/src/lib/insights/__tests__/recent-readings.test.ts @@ -0,0 +1,61 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { findMany, annotate } = vi.hoisted(() => ({ + findMany: vi.fn(), + annotate: vi.fn(), +})); +vi.mock("@/lib/db", () => ({ prisma: { measurement: { findMany } } })); +vi.mock("@/lib/logging/context", () => ({ annotate })); + +import { readRecentReadings, RECENT_READINGS } from "../recent-readings"; + +const now = new Date("2026-10-20T12:00:00Z"); +const minutesAgo = (n: number) => new Date(now.getTime() - n * 60_000); + +beforeEach(() => { + findMany.mockReset(); + annotate.mockReset(); +}); + +describe("readRecentReadings", () => { + it("keeps the latest 365 when they reach back 30 days", async () => { + // One a day: 365 readings span a year. + const rows = Array.from({ length: RECENT_READINGS }, (_, i) => ({ + value: 70, + measuredAt: minutesAgo(i * 1440), + })); + findMany.mockResolvedValueOnce(rows); + const out = await readRecentReadings("u1", "PULSE", now); + expect(findMany).toHaveBeenCalledTimes(1); + expect(out[0].measuredAt < out[out.length - 1].measuredAt).toBe(true); + }); + + it("reads the last 30 days by time when 365 readings are a few hours", async () => { + const dense = Array.from({ length: RECENT_READINGS }, (_, i) => ({ + value: 70, + measuredAt: minutesAgo(i), + })); + const month = Array.from({ length: 30 * 24 }, (_, i) => ({ + value: 70, + measuredAt: minutesAgo(i * 60), + })); + findMany.mockResolvedValueOnce(dense).mockResolvedValueOnce(month); + const out = await readRecentReadings("u1", "PULSE", now); + expect(findMany).toHaveBeenCalledTimes(2); + const window = findMany.mock.calls[1][0] as { + where: { measuredAt: { gte: Date }; deletedAt: null }; + }; + expect(window.where.deletedAt).toBeNull(); + expect(window.where.measuredAt.gte.toISOString()).toBe( + "2026-09-20T12:00:00.000Z", + ); + expect(out).toHaveLength(month.length); + expect(annotate).not.toHaveBeenCalled(); + }); + + it("keeps fewer than 365 readings whole", async () => { + findMany.mockResolvedValueOnce([{ value: 1, measuredAt: minutesAgo(5) }]); + expect(await readRecentReadings("u1", "PULSE", now)).toHaveLength(1); + expect(findMany).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/lib/insights/general-status.ts b/src/lib/insights/general-status.ts index 0c69dfd90..1bb525a4e 100644 --- a/src/lib/insights/general-status.ts +++ b/src/lib/insights/general-status.ts @@ -1,5 +1,6 @@ import { prisma } from "@/lib/db"; import { readRecentMoodScores } from "@/lib/mood/recent-scores"; +import { readLatestReadingsPerType } from "@/lib/insights/recent-readings"; import { getGeneralStatusSystemPrompt, getGeneralStatusUserPrompt, @@ -53,7 +54,7 @@ import { TRACKED_INTAKE_EVENT_WHERE } from "@/lib/medications/intake-tracking"; const MEASUREMENT_TYPES = measurementTypeEnum.options; // The general overview folds its per-type graded series in memory from a -// single bounded read (take 5000, desc) rather than per-type rollup-tier +// single bounded read (the latest 365 per type) rather than per-type rollup-tier // reads: a many-metric account would otherwise fan out ~15 metrics × 3 // rollup round-trips per render. `degradeStatusSnapshotToBudget` already // caps the multi-metric snapshot, so the bounded in-memory fold is the @@ -169,28 +170,11 @@ export async function prepareGeneralStatusForUser( }, }); - // v1.4.28 — cap the snapshot input. General-status pulls - // every supported measurement type; without a cap the read scales - // linearly with account density. - const measurements = await prisma.measurement - .findMany({ - where: { - userId, - type: { - in: [...MEASUREMENT_TYPES], - }, - // Soft-deleted rows must never reach the prompt snapshot. - deletedAt: null, - }, - orderBy: { measuredAt: "desc" }, - take: 5000, - select: { - type: true, - value: true, - measuredAt: true, - }, - }) - .then((rows) => rows.reverse()); + // The latest 365 readings of each type, bounded per type so a dense + // stream cannot crowd the others out (`readLatestReadingsPerType`). + const measurements = await readLatestReadingsPerType(userId, [ + ...MEASUREMENT_TYPES, + ]); const now = new Date(); diff --git a/src/lib/insights/metric-status.ts b/src/lib/insights/metric-status.ts index 3d6168610..69c1aff79 100644 --- a/src/lib/insights/metric-status.ts +++ b/src/lib/insights/metric-status.ts @@ -22,6 +22,7 @@ * the card's insufficient-data state. */ import { prisma } from "@/lib/db"; +import { readRecentReadings } from "@/lib/insights/recent-readings"; import { getMetricArchetypeSystemPrompt, getMetricArchetypeUserPrompt, @@ -293,7 +294,7 @@ export async function generateMetricStatus(args: { let measurements: Array<{ measuredAt: Date }> = []; if (isSleep) { // The latest 365 nights, the sleep twin of the 365 latest readings the - // other types read. A cut on stage rows instead (2190 of them) reached + // other types read (`readRecentReadings`). A cut on stage rows instead (2190 of them) reached // back a few weeks on a stage-rich watch and began mid-night. const nights = ( await readSleepNights( @@ -308,18 +309,8 @@ export async function generateMetricStatus(args: { } measurements = nights.map((n) => ({ measuredAt: n.measuredAt })); } else { - const rows = await prisma.measurement - .findMany({ - where: { - userId: args.userId, - type: meta.measurementType, - deletedAt: null, - }, - orderBy: { measuredAt: "desc" }, - take: 365, - select: { value: true, measuredAt: true }, - }) - .then((r) => r.reverse()); + // The latest 365 readings, never less than the last 30 days. + const rows = await readRecentReadings(args.userId, meta.measurementType); for (const m of rows) { points.push({ measuredAt: m.measuredAt, diff --git a/src/lib/insights/pulse-status.ts b/src/lib/insights/pulse-status.ts index 2cd31d862..3f7dfee82 100644 --- a/src/lib/insights/pulse-status.ts +++ b/src/lib/insights/pulse-status.ts @@ -1,5 +1,6 @@ import { prisma } from "@/lib/db"; import { readRecentMoodScores } from "@/lib/mood/recent-scores"; +import { readRecentReadings } from "@/lib/insights/recent-readings"; import { getPulseSystemPrompt, getPulseUserPrompt, @@ -185,25 +186,9 @@ export async function preparePulseStatusForUser( // estimate for non-Berlin users. `userTz` was already resolved above for // the cache day-key. - // v1.4.28 — cap the snapshot input. The downstream - // `applyPayloadBudget` trims further but the unbounded findMany was - // pulling tens of thousands of rows for Apple-Health-rich accounts - // before the budget call even started. - const measurements = await prisma.measurement - .findMany({ - where: { - userId, - type: "PULSE", - deletedAt: null, - }, - orderBy: { measuredAt: "desc" }, - take: 365, - select: { - value: true, - measuredAt: true, - }, - }) - .then((rows) => rows.reverse()); + // The latest 365 readings, and never less than the last 30 days: a dense + // pulse stream holds 365 readings in a few hours (`readRecentReadings`). + const measurements = await readRecentReadings(userId, "PULSE"); // v1.15.12 — the RESTING band is judged against the RESTING series, // not the raw PULSE stream (Apple fills PULSE with workout HR). Pull diff --git a/src/lib/insights/recent-readings.ts b/src/lib/insights/recent-readings.ts new file mode 100644 index 000000000..f7761440e --- /dev/null +++ b/src/lib/insights/recent-readings.ts @@ -0,0 +1,102 @@ +/** + * The raw readings a per-metric status snapshot reads: the latest 365, and + * never less than the last 30 days. + * + * The snapshots used to read `take: 365` rows. For a type read a few times a + * day that reaches back months; for a dense stream (a watch's pulse every + * few minutes, hourly folds of an imported history) it reached back hours, + * and the "last 30 days" figures built on it (in-target share, the resting + * proxy, the latest day against the one before) were computed over a sliver + * of one day. When the latest 365 readings do not reach back 30 days, the + * last 30 days are read by time instead, bounded by {@link DENSE_READ_CAP} + * so a pathological stream cannot take the heap; hitting the bound is + * annotated, never silent. The longer tail of every snapshot comes from the + * rollup tier, not from here. + */ +import type { MeasurementType } from "@/generated/prisma/client"; +import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; + +export const RECENT_READINGS = 365; +export const RECENT_MIN_DAYS = 30; +/** A reading a minute for 30 days is 43 200; this is above any real stream. */ +export const DENSE_READ_CAP = 60_000; + +const DAY_MS = 24 * 60 * 60 * 1000; + +export interface RecentReading { + value: number; + measuredAt: Date; +} + +/** Oldest first. */ +export async function readRecentReadings( + userId: string, + type: MeasurementType, + now: Date = new Date(), +): Promise { + const latest = await prisma.measurement.findMany({ + where: { userId, type, deletedAt: null }, + orderBy: { measuredAt: "desc" }, + take: RECENT_READINGS, + select: { value: true, measuredAt: true }, + }); + const since = new Date(now.getTime() - RECENT_MIN_DAYS * DAY_MS); + const oldest = latest.at(-1); + if ( + latest.length < RECENT_READINGS || + !oldest || + oldest.measuredAt <= since + ) { + return latest.reverse(); + } + const window = await prisma.measurement.findMany({ + where: { userId, type, deletedAt: null, measuredAt: { gte: since } }, + orderBy: { measuredAt: "desc" }, + take: DENSE_READ_CAP, + select: { value: true, measuredAt: true }, + }); + if (window.length === DENSE_READ_CAP) { + annotate({ + action: { name: "insights.recent-readings.capped" }, + meta: { type, cap: DENSE_READ_CAP }, + }); + } + return window.reverse(); +} + +/** + * The latest `perType` readings of every type in `types`, oldest first, for + * the multi-metric overview. It read the newest 5 000 rows across all types + * at once, so a dense stream (pulse every few minutes) filled the whole read + * within days and every other type fell out of the overview. One index + * descent per type now bounds each type on its own. + */ +export async function readLatestReadingsPerType( + userId: string, + types: readonly MeasurementType[], + perType: number = RECENT_READINGS, +): Promise> { + if (types.length === 0) return []; + const rows = await prisma.$queryRaw< + Array<{ type: MeasurementType; value: number; measured_at: Date }> + >` + SELECT t."type" AS "type", x."value", x."measured_at" + FROM unnest(${[...types]}::text[]) AS t("type") + CROSS JOIN LATERAL ( + SELECT m."value", m."measured_at" + FROM measurements m + WHERE m."user_id" = ${userId} + AND m."type" = t."type"::measurement_type + AND m."deleted_at" IS NULL + ORDER BY m."measured_at" DESC + LIMIT ${perType} + ) x + ORDER BY x."measured_at" ASC + `; + return rows.map((r) => ({ + type: r.type, + value: Number(r.value), + measuredAt: new Date(r.measured_at), + })); +} diff --git a/tests/integration/analytics-summaries-slice.test.ts b/tests/integration/analytics-summaries-slice.test.ts index 244c2f98a..7ff10e366 100644 --- a/tests/integration/analytics-summaries-slice.test.ts +++ b/tests/integration/analytics-summaries-slice.test.ts @@ -185,6 +185,37 @@ describe("GET /api/analytics?slice=summaries (C1)", () => { expect(envelope.data.summaries.PULSE.count).toBe(0); }); + it("summarises sleep over the whole history, not the last year", async () => { + const prisma = getPrismaClient(); + const user = await seedSession("slim-sleep-history"); + const DAY_MS = 24 * 60 * 60 * 1000; + const morning = new Date(); + morning.setUTCHours(6, 0, 0, 0); + // One 7 h night every third day for two years. + const nights = Array.from({ length: 244 }, (_, i) => ({ + userId: user.id, + type: "SLEEP_DURATION" as const, + value: 420, + unit: "minutes", + source: "APPLE_HEALTH" as const, + measuredAt: new Date(morning.getTime() - i * 3 * DAY_MS), + externalId: `uuid-slice-night-${i}`, + sleepStage: "CORE" as const, + })); + await prisma.measurement.createMany({ data: nights }); + + const { GET } = await import("@/app/api/analytics/route"); + const response = await ( + GET as unknown as (req: Request) => Promise + )(new Request("http://localhost/api/analytics?slice=summaries")); + expect(response.status).toBe(200); + const sleep = ((await response.json()) as SlimEnvelope).data.summaries + .SLEEP_DURATION; + // Every night counts, and the window a year back has nights in it. + expect(sleep.count).toBe(244); + expect(sleep.avg30LastYear).toBe(420); + }); + it("default slice still carries the thick envelope (additive contract)", async () => { await seedSession("slim-default-still-thick"); diff --git a/tests/integration/insights-recent-readings.test.ts b/tests/integration/insights-recent-readings.test.ts new file mode 100644 index 000000000..3fa334a9d --- /dev/null +++ b/tests/integration/insights-recent-readings.test.ts @@ -0,0 +1,64 @@ +/** + * The overview's per-type latest-readings query against Postgres: a dense + * stream is bounded on its own and cannot crowd a sparse type out, deleted + * rows stay out, and the rows come back oldest first. + */ +import { beforeEach, describe, expect, it } from "vitest"; + +import { getPrismaClient, truncateAllTables } from "./setup"; +import { readLatestReadingsPerType } from "@/lib/insights/recent-readings"; + +beforeEach(async () => { + await truncateAllTables(getPrismaClient()); +}); + +describe("readLatestReadingsPerType", () => { + it("bounds each type on its own", async () => { + const prisma = getPrismaClient(); + const user = await prisma.user.create({ + data: { username: "recent-per-type", email: "rpt@example.test" }, + }); + const now = Date.now(); + const base = { + userId: user.id, + source: "APPLE_HEALTH" as const, + }; + await prisma.measurement.createMany({ + data: [ + // 6 000 pulse readings a minute apart: a few days of a watch. + ...Array.from({ length: 6000 }, (_, i) => ({ + ...base, + type: "PULSE" as const, + value: 60 + (i % 20), + unit: "bpm", + measuredAt: new Date(now - i * 60_000), + })), + // Weight once a week for a year, the newest one deleted. + ...Array.from({ length: 52 }, (_, i) => ({ + ...base, + source: "MANUAL" as const, + type: "WEIGHT" as const, + value: 80, + unit: "kg", + measuredAt: new Date(now - i * 7 * 86_400_000), + deletedAt: i === 0 ? new Date() : null, + })), + ], + }); + + const rows = await readLatestReadingsPerType( + user.id, + ["PULSE", "WEIGHT", "BODY_FAT"], + 365, + ); + const byType = (t: string) => rows.filter((r) => r.type === t); + expect(byType("PULSE")).toHaveLength(365); + expect(byType("WEIGHT")).toHaveLength(51); + expect(byType("BODY_FAT")).toHaveLength(0); + for (let i = 1; i < rows.length; i += 1) { + expect(rows[i].measuredAt >= rows[i - 1].measuredAt).toBe(true); + } + // The newest pulse reading is in, the oldest of the 6 000 is not. + expect(byType("PULSE").at(-1)!.measuredAt.getTime()).toBe(now); + }); +});